Repository: scylladb/scylla-operator
Pull request #634 (2021-06-09 15:05:57)

Add e2e test for automatic orphaned volume cleanup
**Description of your changes:**
Adds e2e coverage for automatic orphaned volume cleanup.




Patch:
diff --git a/go.mod b/go.mod
index f5df5279c..8591906ec 100644
--- a/go.mod
+++ b/go.mod
@@ -8,6 +8,7 @@ require (
 	github.com/cenkalti/backoff v2.2.1+incompatible
 	github.com/docker/spdystream v0.0.0-20181023171402-6480d4af844c // indirect
 	github.com/elazarl/goproxy v0.0.0-20191011121108-aa519ddbe484 // indirect
+	github.com/evanphx/json-patch v4.9.0+incompatible
 	github.com/ghodss/yaml v1.0.0
 	github.com/go-git/go-git/v5 v5.3.0
 	github.com/go-logr/logr v0.4.0 // indirect
diff --git a/pkg/controller/helpers/mergepatch.go b/pkg/controller/helpers/mergepatch.go
new file mode 100644
index 000000000..38ab75192
--- /dev/null
+++ b/pkg/controller/helpers/mergepatch.go
@@ -0,0 +1,28 @@
+package helpers
+
+import (
+	"fmt"
+
+	jsonpatch "github.com/evanphx/json-patch"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/runtime"
+)
+
+func GenerateMergePatch(original runtime.Object, modified runtime.Object) ([]byte, error) {
+	originalJSON, err := runtime.Encode(unstructured.UnstructuredJSONScheme, original)
+	if err != nil {
+		return nil, fmt.Errorf("unable to decode original to JSON: %v", err)
+	}
+
+	modifiedJSON, err := runtime.Encode(unstructured.UnstructuredJSONScheme, modified)
+	if err != nil {
+		return nil, fmt.Errorf("unable to decode modified to JSON: %v", err)
+	}
+
+	patchBytes, err := jsonpatch.CreateMergePatch(originalJSON, modifiedJSON)
+	if err != nil {
+		return nil, fmt.Errorf("unable to create JSON patch: %v", err)
+	}
+
+	return patchBytes, nil
+}
diff --git a/vendor/modules.txt b/vendor/modules.txt
index 491eb1288..c49677545 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -45,6 +45,7 @@ github.com/emirpasic/gods/trees
 github.com/emirpasic/gods/trees/binaryheap
 github.com/emirpasic/gods/utils
 # github.com/evanphx/json-patch v4.9.0+incompatible
+## explicit
 github.com/evanphx/json-patch
 # github.com/fatih/color v1.7.0
 github.com/fatih/color


Test patch:
diff --git a/pkg/controller/helpers/mergepatch_test.go b/pkg/controller/helpers/mergepatch_test.go
new file mode 100644
index 000000000..54f88eb82
--- /dev/null
+++ b/pkg/controller/helpers/mergepatch_test.go
@@ -0,0 +1,54 @@
+package helpers
+
+import (
+	"testing"
+
+	corev1 "k8s.io/api/core/v1"
+	apiequality "k8s.io/apimachinery/pkg/api/equality"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/utils/pointer"
+)
+
+func TestGenerateMergePatch(t *testing.T) {
+	tt := []struct {
+		name     string
+		original runtime.Object
+		modifyFn func(runtime.Object)
+		expected []byte
+	}{
+		{
+			name: "pod replica change",
+			original: &corev1.Pod{
+				ObjectMeta: metav1.ObjectMeta{
+					Namespace: "foo",
+					Name:      "bar",
+				},
+				Spec: corev1.PodSpec{
+					ActiveDeadlineSeconds: pointer.Int64Ptr(30),
+					NodeName:              "node",
+				},
+			},
+			modifyFn: func(obj runtime.Object) {
+				pod := obj.(*corev1.Pod)
+				pod.Spec.ActiveDeadlineSeconds = pointer.Int64Ptr(10)
+			},
+			expected: []byte(`{"spec":{"activeDeadlineSeconds":10}}`),
+		},
+	}
+
+	for _, tc := range tt {
+		t.Run(tc.name, func(t *testing.T) {
+			modified := tc.original.DeepCopyObject()
+			tc.modifyFn(modified)
+			got, err := GenerateMergePatch(tc.original, modified)
+			if err != nil {
+				t.Error(err)
+			}
+
+			if !apiequality.Semantic.DeepEqual(got, tc.expected) {
+				t.Errorf("expected %s, got %s", tc.expected, got)
+			}
+		})
+	}
+}
diff --git a/test/e2e/set/scyllacluster/helpers.go b/test/e2e/set/scyllacluster/helpers.go
index 87904eb06..d575365d5 100644
--- a/test/e2e/set/scyllacluster/helpers.go
+++ b/test/e2e/set/scyllacluster/helpers.go
@@ -178,6 +178,37 @@ func waitForPodState(ctx context.Context, client corev1client.CoreV1Interface, n
 	return event.Object.(*corev1.Pod), nil
 }
 
+func waitForPVCState(ctx context.Context, client corev1client.CoreV1Interface, namespace string, name string, condition func(sc *corev1.PersistentVolumeClaim) (bool, error), o waitForStateOptions) (*corev1.PersistentVolumeClaim, error) {
+	fieldSelector := fields.OneTermEqualSelector("metadata.name", name).String()
+	lw := &cache.ListWatch{
+		ListFunc: func(options metav1.ListOptions) (runtime.Object, error) {
+			options.FieldSelector = fieldSelector
+			return client.PersistentVolumeClaims(namespace).List(ctx, options)
+		},
+		WatchFunc: func(options metav1.ListOptions) (i watch.Interface, e error) {
+			options.FieldSelector = fieldSelector
+			return client.PersistentVolumeClaims(namespace).Watch(ctx, options)
+		},
+	}
+	event, err := watchtools.UntilWithSync(ctx, lw, &corev1.PersistentVolumeClaim{}, nil, func(event watch.Event) (bool, error) {
+		switch event.Type {
+		case watch.Added, watch.Modified:
+			return condition(event.Object.(*corev1.PersistentVolumeClaim))
+		case watch.Deleted:
+			if o.tolerateDelete {
+				return condition(event.Object.(*corev1.PersistentVolumeClaim))
+			}
+			fallthrough
+		default:
+			return true, fmt.Errorf("unexpected event: %#v", event)
+		}
+	})
+	if err != nil {
+		return nil, err
+	}
+	return event.Object.(*corev1.PersistentVolumeClaim), nil
+}
+
 func getStatefulSetsForScyllaCluster(ctx context.Context, client appv1client.AppsV1Interface, sc *scyllav1.ScyllaCluster) (map[string]*appsv1.StatefulSet, error) {
 	statefulsetList, err := client.StatefulSets(sc.Namespace).List(ctx, metav1.ListOptions{
 		LabelSelector: labels.Set{
diff --git a/test/e2e/set/scyllacluster/scyllacluster_pv.go b/test/e2e/set/scyllacluster/scyllacluster_pv.go
new file mode 100644
index 000000000..b66a19180
--- /dev/null
+++ b/test/e2e/set/scyllacluster/scyllacluster_pv.go
@@ -0,0 +1,112 @@
+// Copyright (C) 2021 ScyllaDB
+
+package scyllacluster
+
+import (
+	"context"
+	"fmt"
+
+	g "github.com/onsi/ginkgo"
+	o "github.com/onsi/gomega"
+	"github.com/scylladb/scylla-operator/pkg/controller/helpers"
+	"github.com/scylladb/scylla-operator/pkg/naming"
+	scyllaclusterfixture "github.com/scylladb/scylla-operator/test/e2e/fixture/scyllacluster"
+	"github.com/scylladb/scylla-operator/test/e2e/framework"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
+)
+
+var _ = g.Describe("ScyllaCluster Orphaned PV", func() {
+	defer g.GinkgoRecover()
+
+	f := framework.NewFramework("scyllacluster")
+
+	g.It("should replace a node with orphaned PV", func() {
+		ctx, cancel := context.WithTimeout(context.Background(), testTimout)
+		defer cancel()
+
+		sc := scyllaclusterfixture.BasicScyllaCluster.ReadOrFail()
+		sc.Spec.Datacenter.Racks[0].Members = 3
+		sc.Spec.AutomaticOrphanedNodeCleanup = true
+
+		framework.By("Creating a ScyllaCluster")
+		err := framework.SetupScyllaClusterSA(ctx, f.KubeClient().CoreV1(), f.KubeClient().RbacV1(), f.Namespace(), sc.Name)
+		o.Expect(err).NotTo(o.HaveOccurred())
+
+		sc, err = f.ScyllaClient().ScyllaV1().ScyllaClusters(f.Namespace()).Create(ctx, sc, metav1.CreateOptions{})
+		o.Expect(err).NotTo(o.HaveOccurred())
+
+		framework.By("Waiting for the ScyllaCluster to deploy")
+		waitCtx1, waitCtx1Cancel := contextForRollout(ctx, sc)
+		defer waitCtx1Cancel()
+		sc, err = waitForScyllaClusterState(waitCtx1, f.ScyllaClient().ScyllaV1(), sc.Namespace, sc.Name, scyllaClusterRolledOut)
+		o.Expect(err).NotTo(o.HaveOccurred())
+
+		verifyScyllaCluster(ctx, f.KubeClient(), sc)
+
+		framework.By("Simulating a PV on node that's gone")
+		stsName := naming.StatefulSetNameForRack(sc.Spec.Datacenter.Racks[0], sc)
+		podName := fmt.Sprintf("%s-%d", stsName, sc.Spec.Datacenter.Racks[0].Members-1)
+		pvcName := naming.PVCNameForPod(podName)
+
+		pvc, err := f.KubeClient().CoreV1().PersistentVolumeClaims(f.Namespace()).Get(ctx, pvcName, metav1.GetOptions{})
+		o.Expect(err).NotTo(o.HaveOccurred())
+		o.Expect(pvc.Spec.VolumeName).NotTo(o.BeEmpty())
+
+		pv, err := f.KubeAdminClient().CoreV1().PersistentVolumes().Get(ctx, pvc.Spec.VolumeName, metav1.GetOptions{})
+		o.Expect(err).NotTo(o.HaveOccurred())
+
+		pvCopy := pv.DeepCopy()
+		pvCopy.Spec.NodeAffinity = &corev1.VolumeNodeAffinity{
+			Required: &corev1.NodeSelector{
+				NodeSelectorTerms: []corev1.NodeSelectorTerm{
+					{
+						MatchExpressions: []corev1.NodeSelectorRequirement{
+							{
+								Key:      corev1.LabelHostname,
+								Operator: corev1.NodeSelectorOpIn,
+								Values:   []string{"this-node-does-not-exist-42"},
+							},
+						},
+					},
+				},
+			},
+		}
+
+		patchData, err := helpers.GenerateMergePatch(pv, pvCopy)
+		o.Expect(err).NotTo(o.HaveOccurred())
+
+		pv, err = f.KubeAdminClient().CoreV1().PersistentVolumes().Patch(ctx, pv.Name, types.MergePatchType, patchData, metav1.PatchOptions{})
+		o.Expect(err).NotTo(o.HaveOccurred())
+		o.Expect(pv.Spec.NodeAffinity).NotTo(o.BeNil())
+
+		// We are not listening to PV changes, so we will make a dummy edit on the ScyllaCluster.
+		sc, err = f.ScyllaClient().ScyllaV1().ScyllaClusters(f.Namespace()).Patch(
+			ctx,
+			sc.Name,
+			types.MergePatchType,
+			[]byte(`{"metadata": {"annotations": {"foo": "bar"} } }`),
+			metav1.PatchOptions{},
+		)
+		o.Expect(err).NotTo(o.HaveOccurred())
+
+		framework.By("Waiting for the PVC to be replaced")
+		waitCtx2, waitCtx2Cancel := contextForRollout(ctx, sc)
+		defer waitCtx2Cancel()
+		pvc, err = waitForPVCState(waitCtx2, f.KubeClient().CoreV1(), pvc.Namespace, pvc.Name, func(freshPVC *corev1.PersistentVolumeClaim) (bool, error) {
+			return freshPVC.UID != pvc.UID, nil
+		}, waitForStateOptions{
+			tolerateDelete: true,
+		})
+		o.Expect(err).NotTo(o.HaveOccurred())
+
+		framework.By("Waiting for the ScyllaCluster to deploy")
+		waitCtx3, waitCtx3Cancel := contextForRollout(ctx, sc)
+		defer waitCtx3Cancel()
+		sc, err = waitForScyllaClusterState(waitCtx3, f.ScyllaClient().ScyllaV1(), sc.Namespace, sc.Name, scyllaClusterRolledOut)
+		o.Expect(err).NotTo(o.HaveOccurred())
+
+		verifyScyllaCluster(ctx, f.KubeClient(), sc)
+	})
+})