Merge pull request #248 from justinsb/test_for_backup_and_restore

Create test for backup and restore functionality

WORKSPACE

@@ -29,7 +29,7 @@ go_register_toolchains(
     go_version = "1.17.1",
 )
 
-load("@bazel_gazelle//:deps.bzl", "gazelle_dependencies", "go_repository")
+load("@bazel_gazelle//:deps.bzl", "gazelle_dependencies")
 
 gazelle_dependencies()
 

test/integration/backuprestore/BUILD.bazel

@@ -0,0 +1,34 @@
+load("@io_bazel_rules_go//go:def.bzl", "go_test")
+
+go_test(
+    name = "backuprestore_test",
+    size = "enormous",  # Takes more than 900 seconds to run on travis
+    srcs = ["backuprestore_test.go"],
+    data = [
+        "//:etcd-v3.1.12-linux-amd64_etcd",
+        "//:etcd-v3.1.12-linux-amd64_etcdctl",
+        "//:etcd-v3.2.18-linux-amd64_etcd",
+        "//:etcd-v3.2.18-linux-amd64_etcdctl",
+        "//:etcd-v3.2.24-linux-amd64_etcd",
+        "//:etcd-v3.2.24-linux-amd64_etcdctl",
+        "//:etcd-v3.3.10-linux-amd64_etcd",
+        "//:etcd-v3.3.10-linux-amd64_etcdctl",
+        "//:etcd-v3.3.13-linux-amd64_etcd",
+        "//:etcd-v3.3.13-linux-amd64_etcdctl",
+        "//:etcd-v3.3.17-linux-amd64_etcd",
+        "//:etcd-v3.3.17-linux-amd64_etcdctl",
+        "//:etcd-v3.4.13-linux-amd64_etcd",
+        "//:etcd-v3.4.13-linux-amd64_etcdctl",
+        "//:etcd-v3.4.3-linux-amd64_etcd",
+        "//:etcd-v3.4.3-linux-amd64_etcdctl",
+        "//:etcd-v3.5.0-linux-amd64_etcd",
+        "//:etcd-v3.5.0-linux-amd64_etcdctl",
+    ],
+    deps = [
+        "//pkg/apis/etcd",
+        "//pkg/backup",
+        "//pkg/etcdversions",
+        "//test/integration/harness",
+        "//vendor/k8s.io/klog/v2:klog",
+    ],
+)

test/integration/backuprestore/backuprestore_test.go

@@ -0,0 +1,137 @@
+/*
+Copyright 2021 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 integration
+
+import (
+	"context"
+	"os"
+	"testing"
+	"time"
+
+	"k8s.io/klog/v2"
+	protoetcd "sigs.k8s.io/etcdadm/etcd-manager/pkg/apis/etcd"
+	"sigs.k8s.io/etcdadm/etcd-manager/pkg/backup"
+	"sigs.k8s.io/etcdadm/etcd-manager/pkg/etcdversions"
+	"sigs.k8s.io/etcdadm/etcd-manager/test/integration/harness"
+)
+
+func TestBackupRestore(t *testing.T) {
+	for _, etcdVersion := range etcdversions.AllEtcdVersions {
+		t.Run("etcdVersion="+etcdVersion, func(t *testing.T) {
+			ctx := context.TODO()
+			ctx, cancel := context.WithCancel(ctx)
+			defer cancel()
+
+			clusterSpec := &protoetcd.ClusterSpec{MemberCount: 1, EtcdVersion: etcdVersion}
+
+			h1 := harness.NewTestHarness(t, ctx)
+			// Very short backup interval so we don't have to wait too long for a backup!
+			h1.BackupInterval = 5 * time.Second
+			h1.SeedNewCluster(clusterSpec)
+			defer h1.Close()
+
+			n1 := h1.NewNode("127.0.0.1")
+			go n1.Run()
+
+			n1.WaitForListMembers(20 * time.Second)
+
+			key := "/testing/backuprestore_" + etcdVersion
+			value := "world-" + etcdVersion
+
+			err := n1.Put(ctx, key, value)
+			if err != nil {
+				t.Fatalf("error writing key %q: %v", key, err)
+			}
+
+			{
+				actual, err := n1.GetQuorum(ctx, key)
+				if err != nil {
+					t.Fatalf("error reading key %q: %v", key, err)
+				}
+				if actual != value {
+					t.Fatalf("could not read back key %q: %q vs %q", key, actual, value)
+				}
+			}
+
+			h1.WaitForBackup(t, 5*time.Minute)
+
+			// We should be able to shut down the node, delete the local etcd data, and it should do a restore
+			if err := n1.Close(); err != nil {
+				t.Fatalf("failed to stop node 1: %v", err)
+			}
+
+			klog.Infof("removing directory %q", n1.NodeDir)
+			if err := os.RemoveAll(n1.NodeDir); err != nil {
+				t.Fatalf("failed to delete dir for node 1: %v", err)
+			}
+
+			// Create a new cluster, but reuse the backups
+			h2 := harness.NewTestHarness(t, ctx)
+			h2.BackupStorePath = h1.BackupStorePath
+			h2.SeedNewCluster(clusterSpec)
+			defer h2.Close()
+
+			backupStore, err := backup.NewStore(h2.BackupStorePath)
+			if err != nil {
+				t.Fatalf("error initializing backup store: %v", err)
+			}
+
+			backups, err := backupStore.ListBackups()
+			if err != nil {
+				t.Fatalf("error listing backups: %v", err)
+			}
+			if len(backups) == 0 {
+				t.Fatalf("no backups: %v", err)
+			}
+
+			// We also have to send a restore backup command for a full DR (no common data) scenario
+			backupName := backups[len(backups)-1]
+			klog.Infof("adding command to restore backup %q", backupName)
+			h2.AddCommand(&protoetcd.Command{
+				Timestamp: time.Now().UnixNano(),
+				RestoreBackup: &protoetcd.RestoreBackupCommand{
+					ClusterSpec: clusterSpec,
+					Backup:      backupName,
+				},
+			})
+
+			n2 := h2.NewNode("127.0.0.2")
+			defer n2.Close()
+
+			klog.Infof("starting node %v", n2.Address)
+			go n2.Run()
+
+			n2.WaitForListMembers(20 * time.Second)
+
+			klog.Infof("checking for key %q", key)
+			{
+				actual, err := n2.GetQuorum(ctx, key)
+				if err != nil {
+					t.Fatalf("error rereading key %q: %v", key, err)
+				}
+				if actual != value {
+					t.Fatalf("could not reread key %q: %q vs %q", key, actual, value)
+				}
+			}
+
+			klog.Infof("stopping node 2")
+			if err := n2.Close(); err != nil {
+				t.Fatalf("failed to stop node 2: %v", err)
+			}
+		})
+	}
+}

test/integration/harness/cluster.go

@@ -28,6 +28,7 @@ import (
 
 	"k8s.io/klog/v2"
 	protoetcd "sigs.k8s.io/etcdadm/etcd-manager/pkg/apis/etcd"
+	"sigs.k8s.io/etcdadm/etcd-manager/pkg/backup"
 	"sigs.k8s.io/etcdadm/etcd-manager/pkg/commands"
 	"sigs.k8s.io/etcdadm/etcd-manager/pkg/pki"
 )
@@ -53,6 +54,8 @@ type TestHarness struct {
 	Nodes map[string]*TestHarnessNode
 
 	Context context.Context
+
+	BackupInterval time.Duration
 }
 
 func NewTestHarness(t *testing.T, ctx context.Context) *TestHarness {
@@ -65,11 +68,12 @@ func NewTestHarness(t *testing.T, ctx context.Context) *TestHarness {
 
 	clusterName := "testharnesscluster"
 	h := &TestHarness{
-		T:           t,
-		ClusterName: clusterName,
-		WorkDir:     path.Join(tmpDir, clusterName),
-		Nodes:       make(map[string]*TestHarnessNode),
-		Context:     ctx,
+		T:              t,
+		ClusterName:    clusterName,
+		WorkDir:        path.Join(tmpDir, clusterName),
+		Nodes:          make(map[string]*TestHarnessNode),
+		Context:        ctx,
+		BackupInterval: 15 * time.Minute,
 	}
 
 	h.grpcCA, err = pki.NewCA(pki.NewFSStore(filepath.Join(h.WorkDir, "pki/grpc")))
@@ -211,6 +215,17 @@ func (h *TestHarness) SetClusterSpec(spec *protoetcd.ClusterSpec) {
 	}
 }
 
+func (h *TestHarness) AddCommand(cmd *protoetcd.Command) {
+	t := h.T
+	controlStore, err := commands.NewStore(h.BackupStorePath)
+	if err != nil {
+		t.Fatalf("error initializing control store: %v", err)
+	}
+	if err := controlStore.AddCommand(cmd); err != nil {
+		t.Fatalf("error adding command: %v", err)
+	}
+}
+
 func (h *TestHarness) InvalidateControlStore(nodes ...*TestHarnessNode) {
 	t := h.T
 
@@ -229,3 +244,28 @@ func (h *TestHarness) InvalidateControlStore(nodes ...*TestHarnessNode) {
 		}
 	}
 }
+
+func (h *TestHarness) WaitForBackup(t *testing.T, timeout time.Duration) {
+	backupStore, err := backup.NewStore(h.BackupStorePath)
+	if err != nil {
+		t.Fatalf("error initializing backup store: %v", err)
+	}
+
+	backups, err := backupStore.ListBackups()
+	if err != nil {
+		t.Fatalf("error listing backups: %v", err)
+	}
+	wantBackups := len(backups) + 1
+
+	description := fmt.Sprintf("wait for new backup")
+	h.WaitFor(timeout, description, func() error {
+		backups, err := backupStore.ListBackups()
+		if err != nil {
+			return fmt.Errorf("error listing backups: %w", err)
+		}
+		if len(backups) >= wantBackups {
+			return nil
+		}
+		return fmt.Errorf("insufficient backups; got %d, want %d", len(backups), wantBackups)
+	})
+}

test/integration/harness/node.go

@@ -190,7 +190,7 @@ func (n *TestHarnessNode) Run() {
 	if err != nil {
 		t.Fatalf("error initializing backup store: %v", err)
 	}
-	backupInterval := 15 * time.Minute
+	backupInterval := n.TestHarness.BackupInterval
 
 	commandStore, err := commands.NewStore(n.TestHarness.BackupStorePath)
 	if err != nil {