Merge pull request kubernetes-retired#58 from justinsb/standalone_backup

Build standalone etcd-backup tool

etcd-manager/Makefile

@@ -41,6 +41,17 @@ image-etcd-dump:
 push-etcd-dump: image-etcd-dump
 	docker push ${DOCKER_REGISTRY}/etcd-dump:${DOCKER_TAG}
 
+
+.PHONY: image-etcd-backup
+image-etcd-backup:
+	bazel build //images:*
+	bazel run //images:etcd-backup
+	docker tag bazel/images:etcd-backup ${DOCKER_REGISTRY}/etcd-backup:${DOCKER_TAG}
+
+.PHONY: push-etcd-backup
+push-etcd-backup: image-etcd-backup
+	docker push ${DOCKER_REGISTRY}/etcd-backup:${DOCKER_TAG}
+
 .PHONY: push
 push: push-etcd-manager push-etcd-dump
 	echo "pushed images"

etcd-manager/cmd/etcd-backup/BUILD.bazel

@@ -0,0 +1,20 @@
+load("@io_bazel_rules_go//go:def.bzl", "go_binary", "go_library")
+
+go_library(
+    name = "go_default_library",
+    srcs = ["main.go"],
+    importpath = "kope.io/etcd-manager/cmd/etcd-backup",
+    visibility = ["//visibility:private"],
+    deps = [
+        "//pkg/backup:go_default_library",
+        "//pkg/backupcontroller:go_default_library",
+        "//vendor/github.com/golang/glog:go_default_library",
+    ],
+)
+
+go_binary(
+    name = "etcd-backup",
+    embed = [":go_default_library"],
+    importpath = "kope.io/etcd-manager/cmd/etcd-backup",
+    visibility = ["//visibility:public"],
+)

etcd-manager/cmd/etcd-backup/main.go

@@ -0,0 +1,74 @@
+/*
+Copyright 2018 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package main
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"os"
+
+	"github.com/golang/glog"
+
+	"kope.io/etcd-manager/pkg/backup"
+	"kope.io/etcd-manager/pkg/backupcontroller"
+)
+
+func main() {
+	flag.Set("logtostderr", "true")
+
+	clusterName := ""
+	flag.StringVar(&clusterName, "cluster-name", clusterName, "name of cluster")
+	backupStorePath := "/backups"
+	flag.StringVar(&backupStorePath, "backup-store", backupStorePath, "backup store location")
+	dataDir := "/data"
+	flag.StringVar(&dataDir, "data-dir", dataDir, "directory for storing etcd data")
+	clientURL := "http://127.0.0.1:4001"
+	flag.StringVar(&clientURL, "client-url", clientURL, "URL on which to connect to etcd")
+	etcdVersion := "2.2.1"
+	flag.StringVar(&etcdVersion, "etcd-version", etcdVersion, "etcd version in use")
+
+	flag.Parse()
+
+	fmt.Printf("etcd-backup agent\n")
+
+	if clusterName == "" {
+		fmt.Fprintf(os.Stderr, "cluster-name is required\n")
+		os.Exit(1)
+	}
+
+	if backupStorePath == "" {
+		fmt.Fprintf(os.Stderr, "backup-store is required\n")
+		os.Exit(1)
+	}
+
+	ctx := context.TODO()
+
+	backupStore, err := backup.NewStore(backupStorePath)
+	if err != nil {
+		glog.Fatalf("error initializing backup store: %v", err)
+	}
+	clientURLs := []string{clientURL}
+	c, err := backupcontroller.NewBackupController(backupStore, clusterName, clientURLs, etcdVersion, dataDir)
+	if err != nil {
+		glog.Fatalf("error building backup controller: %v", err)
+	}
+
+	c.Run(ctx)
+
+	os.Exit(0)
+}

etcd-manager/images/BUILD

@@ -31,3 +31,12 @@ container_image(
         "//cmd/etcd-dump",
     ],
 )
+
+container_image(
+    name = "etcd-backup",
+    base = "etcd-manager-base",
+    entrypoint = ["/etcd-backup"],
+    files = [
+        "//cmd/etcd-backup",
+    ],
+)

etcd-manager/pkg/backupcontroller/BUILD.bazel

@@ -0,0 +1,19 @@
+load("@io_bazel_rules_go//go:def.bzl", "go_library")
+
+go_library(
+    name = "go_default_library",
+    srcs = [
+        "cleanup.go",
+        "controller.go",
+    ],
+    importpath = "kope.io/etcd-manager/pkg/backupcontroller",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//pkg/apis/etcd:go_default_library",
+        "//pkg/backup:go_default_library",
+        "//pkg/contextutil:go_default_library",
+        "//pkg/etcd:go_default_library",
+        "//pkg/etcdclient:go_default_library",
+        "//vendor/github.com/golang/glog:go_default_library",
+    ],
+)

etcd-manager/pkg/backupcontroller/cleanup.go

@@ -0,0 +1,115 @@
+package backupcontroller
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/golang/glog"
+
+	"kope.io/etcd-manager/pkg/backup"
+)
+
+// BackupCleanup encapsulates the logic around periodically removing old backups
+type BackupCleanup struct {
+	backupStore backup.Store
+
+	// lastBackupCleanup is the time at which we last performed a backup store cleanup (as leader)
+	lastBackupCleanup time.Time
+
+	backupCleanupInterval time.Duration
+}
+
+// NewBackupCleanup constructs a BackupCleanup
+func NewBackupCleanup(backupStore backup.Store) *BackupCleanup {
+	return &BackupCleanup{
+		backupStore:           backupStore,
+		backupCleanupInterval: time.Hour,
+	}
+}
+
+// MaybeDoBackupMaintenance removes old backups, if a suitable interval has passed.
+// It should be called periodically, after every backup for example.
+func (m *BackupCleanup) MaybeDoBackupMaintenance(ctx context.Context) error {
+	now := time.Now()
+
+	if now.Sub(m.lastBackupCleanup) < m.backupCleanupInterval {
+		return nil
+	}
+
+	backupNames, err := m.backupStore.ListBackups()
+	if err != nil {
+		return fmt.Errorf("error listing backups: %v", err)
+	}
+
+	minRetention := time.Hour
+	hourly := time.Hour * 24 * 7
+	daily := time.Hour * 24 * 7 * 365
+
+	backups := make(map[time.Time]string)
+	retain := make(map[string]bool)
+	buckets := make(map[time.Time]time.Time)
+
+	for _, backup := range backupNames {
+		// Time parsing uses the same layout values as `Format`.
+		t, err := time.Parse(time.RFC3339, backup)
+		if err != nil {
+			glog.Warningf("ignoring unparseable backup %q", backup)
+			continue
+		}
+
+		backups[t] = backup
+
+		age := now.Sub(t)
+
+		if age < minRetention {
+			retain[backup] = true
+			continue
+		}
+
+		if age < hourly {
+			bucketed := t.Truncate(time.Hour)
+			existing := buckets[bucketed]
+			if existing.IsZero() || existing.After(t) {
+				buckets[bucketed] = t
+			}
+			continue
+		}
+
+		if age < daily {
+			bucketed := t.Truncate(time.Hour * 24)
+			existing := buckets[bucketed]
+			if existing.IsZero() || existing.After(t) {
+				buckets[bucketed] = t
+			}
+			continue
+		}
+	}
+
+	for _, t := range buckets {
+		retain[backups[t]] = true
+	}
+
+	removedCount := 0
+	for _, backup := range backupNames {
+		if retain[backup] {
+			glog.V(4).Infof("retaining backup %q", backup)
+			continue
+		}
+		glog.V(4).Infof("removing backup %q", backup)
+		if err := m.backupStore.RemoveBackup(backup); err != nil {
+			glog.Warningf("failed to remove backup %q: %v", backup, err)
+		} else {
+			glog.V(2).Infof("removed backup %q", backup)
+			removedCount++
+		}
+	}
+
+	if removedCount != 0 {
+		glog.Infof("Removed %d old backups", removedCount)
+	}
+
+	m.lastBackupCleanup = now
+
+	return nil
+}

etcd-manager/pkg/backupcontroller/controller.go

@@ -0,0 +1,128 @@
+package backupcontroller
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/golang/glog"
+
+	protoetcd "kope.io/etcd-manager/pkg/apis/etcd"
+	"kope.io/etcd-manager/pkg/backup"
+	"kope.io/etcd-manager/pkg/contextutil"
+	"kope.io/etcd-manager/pkg/etcd"
+	"kope.io/etcd-manager/pkg/etcdclient"
+)
+
+const loopInterval = time.Minute
+
+type BackupController struct {
+	clusterName string
+	backupStore backup.Store
+
+	dataDir string
+
+	clientUrls  []string
+	etcdVersion string
+
+	// lastBackup is the time at which we last performed a backup (as leader)
+	lastBackup time.Time
+
+	backupInterval time.Duration
+
+	backupCleanup *BackupCleanup
+}
+
+func NewBackupController(backupStore backup.Store, clusterName string, clientUrls []string, etcdVersion string, dataDir string) (*BackupController, error) {
+	if clusterName == "" {
+		return nil, fmt.Errorf("ClusterName is required")
+	}
+
+	if etcdclient.IsV2(etcdVersion) && dataDir == "" {
+		return nil, fmt.Errorf("DataDir is required for etcd v2")
+	}
+
+	m := &BackupController{
+		clusterName:    clusterName,
+		backupStore:    backupStore,
+		dataDir:        dataDir,
+		clientUrls:     clientUrls,
+		etcdVersion:    etcdVersion,
+		backupInterval: 5 * time.Minute,
+		backupCleanup:  NewBackupCleanup(backupStore),
+	}
+	return m, nil
+}
+
+func (m *BackupController) Run(ctx context.Context) {
+	contextutil.Forever(ctx,
+		loopInterval, // We do our own sleeping
+		func() {
+			err := m.run(ctx)
+			if err != nil {
+				glog.Warningf("unexpected error running backup controller loop: %v", err)
+			}
+		})
+}
+
+func (m *BackupController) run(ctx context.Context) error {
+	glog.V(2).Infof("starting backup controller iteration")
+
+	etcdClient, err := etcdclient.NewClient(m.etcdVersion, m.clientUrls)
+	if err != nil {
+		return fmt.Errorf("unable to reach etcd on %s: %v", m.clientUrls, err)
+	}
+	members, err := etcdClient.ListMembers(ctx)
+	if err != nil {
+		etcdClient.Close()
+		return fmt.Errorf("unable to list members on %s: %v", m.clientUrls, err)
+	}
+
+	self, err := etcdClient.LocalNodeInfo(ctx)
+	etcdClient.Close()
+	if err != nil {
+		return fmt.Errorf("unable to get node state on %s: %v", m.clientUrls, err)
+	}
+
+	if !self.IsLeader {
+		glog.V(2).Infof("Not leader, won't backup")
+		return nil
+	}
+
+	return m.maybeBackup(ctx, members)
+}
+
+func (m *BackupController) maybeBackup(ctx context.Context, members []*etcdclient.EtcdProcessMember) error {
+	now := time.Now()
+
+	shouldBackup := now.Sub(m.lastBackup) > m.backupInterval
+	if !shouldBackup {
+		return nil
+	}
+
+	backup, err := m.doClusterBackup(ctx, members)
+	if err != nil {
+		return err
+	}
+
+	glog.Infof("took backup: %v", backup)
+	m.lastBackup = now
+
+	if err := m.backupCleanup.MaybeDoBackupMaintenance(ctx); err != nil {
+		glog.Warningf("error during backup cleanup: %v", err)
+	}
+
+	return nil
+}
+
+func (m *BackupController) doClusterBackup(ctx context.Context, members []*etcdclient.EtcdProcessMember) (*protoetcd.DoBackupResponse, error) {
+	info := &protoetcd.BackupInfo{
+		ClusterSpec: &protoetcd.ClusterSpec{
+			MemberCount: int32(len(members)),
+			EtcdVersion: m.etcdVersion,
+		},
+		EtcdVersion: m.etcdVersion,
+	}
+
+	return etcd.DoBackup(m.backupStore, info, m.dataDir, m.clientUrls)
+}

etcd-manager/pkg/controller/BUILD.bazel

@@ -14,6 +14,7 @@ go_library(
     deps = [
         "//pkg/apis/etcd:go_default_library",
         "//pkg/backup:go_default_library",
+        "//pkg/backupcontroller:go_default_library",
         "//pkg/contextutil:go_default_library",
         "//pkg/etcdclient:go_default_library",
         "//pkg/locking:go_default_library",