YDBOPS-9168 Storage finalizer waiting for existing Databases (#211)

Author: kobzonega

deploy/ydb-operator/Chart.yaml

@@ -15,10 +15,10 @@ type: application
 # This is the chart version. This version number should be incremented each time you make changes
 # to the chart and its templates, including the app version.
 # Versions are expected to follow Semantic Versioning (https://semver.org/)
-version: 0.5.19
+version: 0.5.20
 
 # This is the version number of the application being deployed. This version number should be
 # incremented each time you make changes to the application. Versions are not expected to
 # follow Semantic Versioning. They should reflect the version the application is using.
 # It is recommended to use it with quotes.
-appVersion: "0.5.19"
+appVersion: "0.5.20"

e2e/tests/smoke_test.go

@@ -22,6 +22,7 @@ import (
 	. "github.com/onsi/gomega"
 	v1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
@@ -31,6 +32,7 @@ import (
 	testobjects "github.com/ydb-platform/ydb-kubernetes-operator/e2e/tests/test-objects"
 	. "github.com/ydb-platform/ydb-kubernetes-operator/internal/controllers/constants"
 	"github.com/ydb-platform/ydb-kubernetes-operator/internal/resources"
+	"github.com/ydb-platform/ydb-kubernetes-operator/internal/test"
 )
 
 const (
@@ -158,7 +160,7 @@ func bringYdbCliToPod(podName, podNamespace string) {
 	}, Timeout, Interval).Should(BeNil())
 }
 
-func executeSimpleQuery(ctx context.Context, podName, podNamespace, storageEndpoint string) {
+func executeSimpleQuery(podName, podNamespace, storageEndpoint string) {
 	Eventually(func(g Gomega) string {
 		args := []string{
 			"-n",
@@ -345,7 +347,7 @@ var _ = Describe("Operator smoke test", func() {
 		bringYdbCliToPod(podName, testobjects.YdbNamespace)
 
 		By("execute simple query inside ydb database pod...")
-		executeSimpleQuery(ctx, podName, testobjects.YdbNamespace, storageEndpoint)
+		executeSimpleQuery(podName, testobjects.YdbNamespace, storageEndpoint)
 	})
 
 	It("pause and un-pause Storage, should destroy and bring up Pods", func() {
@@ -492,7 +494,6 @@ var _ = Describe("Operator smoke test", func() {
 	It("create storage and database with nodeSets", func() {
 		By("issuing create commands...")
 		storageSample = testobjects.DefaultStorage(filepath.Join(".", "data", "storage-block-4-2-config-nodeSets.yaml"))
-		databaseSample = testobjects.DefaultDatabase()
 		testNodeSetName := "nodeset"
 		for idx := 1; idx <= 2; idx++ {
 			storageSample.Spec.NodeSets = append(storageSample.Spec.NodeSets, v1alpha1.StorageNodeSetSpecInline{
@@ -572,7 +573,7 @@ var _ = Describe("Operator smoke test", func() {
 		bringYdbCliToPod(podName, testobjects.YdbNamespace)
 
 		By("execute simple query inside ydb database pod...")
-		executeSimpleQuery(ctx, podName, testobjects.YdbNamespace, storageEndpoint)
+		executeSimpleQuery(podName, testobjects.YdbNamespace, storageEndpoint)
 	})
 
 	It("operatorConnection check, create storage with default staticCredentials", func() {
@@ -652,28 +653,115 @@ var _ = Describe("Operator smoke test", func() {
 			LocalObjectReference: corev1.LocalObjectReference{Name: testobjects.CertificateSecretName},
 			Key:                  "ca.crt",
 		}
-
 		Expect(k8sClient.Create(ctx, storageSample)).Should(Succeed())
 		defer func() {
 			Expect(k8sClient.Delete(ctx, storageSample)).Should(Succeed())
 		}()
+
+		By("waiting until Storage is ready...")
+		waitUntilStorageReady(ctx, storageSample.Name, testobjects.YdbNamespace)
+
+		By("checking that all the storage pods are running and ready...")
+		checkPodsRunningAndReady(ctx, "ydb-cluster", "kind-storage", storageSample.Spec.Nodes)
+
 		By("create database...")
+		databaseSample.Spec.Service.GRPC.TLSConfiguration.Enabled = true
+		databaseSample.Spec.Service.GRPC.TLSConfiguration.Certificate = corev1.SecretKeySelector{
+			LocalObjectReference: corev1.LocalObjectReference{Name: testobjects.CertificateSecretName},
+			Key:                  "tls.crt",
+		}
+		databaseSample.Spec.Service.GRPC.TLSConfiguration.Key = corev1.SecretKeySelector{
+			LocalObjectReference: corev1.LocalObjectReference{Name: testobjects.CertificateSecretName},
+			Key:                  "tls.key",
+		}
+		databaseSample.Spec.Service.GRPC.TLSConfiguration.CertificateAuthority = corev1.SecretKeySelector{
+			LocalObjectReference: corev1.LocalObjectReference{Name: testobjects.CertificateSecretName},
+			Key:                  "ca.crt",
+		}
 		Expect(k8sClient.Create(ctx, databaseSample)).Should(Succeed())
 		defer func() {
 			Expect(k8sClient.Delete(ctx, databaseSample)).Should(Succeed())
 		}()
 
+		By("waiting until database is ready...")
+		waitUntilDatabaseReady(ctx, databaseSample.Name, testobjects.YdbNamespace)
+
+		By("checking that all the database pods are running and ready...")
+		checkPodsRunningAndReady(ctx, "ydb-cluster", "kind-database", databaseSample.Spec.Nodes)
+
+		storagePods := corev1.PodList{}
+		Expect(k8sClient.List(ctx, &storagePods,
+			client.InNamespace(testobjects.YdbNamespace),
+			client.MatchingLabels{
+				"ydb-cluster": "kind-database",
+			})).Should(Succeed())
+		podName := storagePods.Items[0].Name
+
+		By("bring YDB CLI inside ydb storage pod...")
+		bringYdbCliToPod(podName, testobjects.YdbNamespace)
+
+		By("execute simple query inside ydb storage pod...")
+		storageEndpoint := fmt.Sprintf("grpcs://%s:%d", testobjects.StorageGRPCService, testobjects.StorageGRPCPort)
+		executeSimpleQuery(podName, testobjects.YdbNamespace, storageEndpoint)
+	})
+
+	It("Check that Storage deleted after Database...", func() {
+		By("create storage...")
+		Expect(k8sClient.Create(ctx, storageSample)).Should(Succeed())
+
+		By("create database...")
+		Expect(k8sClient.Create(ctx, databaseSample)).Should(Succeed())
+
 		By("waiting until Storage is ready...")
 		waitUntilStorageReady(ctx, storageSample.Name, testobjects.YdbNamespace)
 
 		By("checking that all the storage pods are running and ready...")
 		checkPodsRunningAndReady(ctx, "ydb-cluster", "kind-storage", storageSample.Spec.Nodes)
 
-		By("waiting until database is ready...")
+		By("waiting until Database is ready...")
 		waitUntilDatabaseReady(ctx, databaseSample.Name, testobjects.YdbNamespace)
 
 		By("checking that all the database pods are running and ready...")
 		checkPodsRunningAndReady(ctx, "ydb-cluster", "kind-database", databaseSample.Spec.Nodes)
+
+		By("delete Storage...")
+		Expect(k8sClient.Delete(ctx, storageSample)).Should(Succeed())
+
+		By("checking that Storage deletionTimestamp is not nil...")
+		Eventually(func() bool {
+			foundStorage := v1alpha1.Storage{}
+			err := k8sClient.Get(ctx, types.NamespacedName{
+				Name:      storageSample.Name,
+				Namespace: testobjects.YdbNamespace,
+			}, &foundStorage)
+			if err != nil {
+				return false
+			}
+			return !foundStorage.DeletionTimestamp.IsZero()
+		}, test.Timeout, test.Interval).Should(BeTrue())
+
+		By("checking that Storage is present in cluster...")
+		Consistently(func() error {
+			foundStorage := v1alpha1.Storage{}
+			err := k8sClient.Get(ctx, types.NamespacedName{
+				Name:      storageSample.Name,
+				Namespace: testobjects.YdbNamespace,
+			}, &foundStorage)
+			return err
+		}, test.Timeout, test.Interval).ShouldNot(HaveOccurred())
+
+		By("delete Database...")
+		Expect(k8sClient.Delete(ctx, databaseSample)).Should(Succeed())
+
+		By("checking that Storage deleted from cluster...")
+		Eventually(func() bool {
+			foundStorage := v1alpha1.Storage{}
+			err := k8sClient.Get(ctx, types.NamespacedName{
+				Name:      storageSample.Name,
+				Namespace: testobjects.YdbNamespace,
+			}, &foundStorage)
+			return apierrors.IsNotFound(err)
+		}, test.Timeout, test.Interval).Should(BeTrue())
 	})
 
 	It("TLS for status service", func() {
@@ -740,13 +828,18 @@ var _ = Describe("Operator smoke test", func() {
 				Key:                  "ca.crt",
 			},
 		}
-
 		Expect(k8sClient.Create(ctx, storageSample)).Should(Succeed())
+		defer func() {
+			Expect(k8sClient.Delete(ctx, storageSample)).Should(Succeed())
+		}()
 
 		By("create database...")
 		databaseSample.Spec.Nodes = 1
 		databaseSample.Spec.Service.Status = *storageSample.Spec.Service.Status.DeepCopy()
 		Expect(k8sClient.Create(ctx, databaseSample)).Should(Succeed())
+		defer func() {
+			Expect(k8sClient.Delete(ctx, databaseSample)).Should(Succeed())
+		}()
 
 		By("waiting until Storage is ready...")
 		waitUntilStorageReady(ctx, storageSample.Name, testobjects.YdbNamespace)

e2e/tests/test-objects/objects.go

@@ -18,6 +18,8 @@ const (
 	CertificateSecretName = "storage-crt"
 	DefaultDomain         = "Root"
 	ReadyStatus           = "Ready"
+	StorageGRPCService    = "storage-grpc.ydb.svc.cluster.local"
+	StorageGRPCPort       = 2135
 )
 
 func constructAntiAffinityFor(key, value string) *corev1.Affinity {

internal/annotations/annotations.go

@@ -5,6 +5,7 @@ const (
 	PrimaryResourceDatabaseAnnotation = "ydb.tech/primary-resource-database"
 	RemoteResourceVersionAnnotation   = "ydb.tech/remote-resource-version"
 	ConfigurationChecksum             = "ydb.tech/configuration-checksum"
+	StorageFinalizerKey               = "ydb.tech/storage-finalizer"
 	RemoteFinalizerKey                = "ydb.tech/remote-finalizer"
 	LastAppliedAnnotation             = "ydb.tech/last-applied"
 )

internal/controllers/remotedatabasenodeset/controller_test.go

@@ -347,6 +347,8 @@ var _ = Describe("RemoteDatabaseNodeSet controller tests", func() {
 	})
 
 	AfterEach(func() {
+		Expect(localClient.Delete(ctx, databaseSample)).Should(Succeed())
+		Expect(localClient.Delete(ctx, storageSample)).Should(Succeed())
 		deleteAll(localEnv, localClient, &localNamespace)
 		deleteAll(remoteEnv, remoteClient, &localNamespace)
 	})

internal/controllers/remotestoragenodeset/controller_test.go

@@ -279,6 +279,7 @@ var _ = Describe("RemoteStorageNodeSet controller tests", func() {
 	})
 
 	AfterEach(func() {
+		Expect(localClient.Delete(ctx, storageSample)).Should(Succeed())
 		deleteAll(localEnv, localClient, &localNamespace)
 		deleteAll(remoteEnv, remoteClient, &localNamespace)
 	})

internal/controllers/storage/controller.go

@@ -2,6 +2,8 @@ package storage
 
 import (
 	"context"
+	"errors"
+	"fmt"
 
 	"github.com/go-logr/logr"
 	monitoringv1 "github.com/prometheus-operator/prometheus-operator/pkg/apis/monitoring/v1"
@@ -16,13 +18,15 @@ import (
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/builder"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
 	"sigs.k8s.io/controller-runtime/pkg/handler"
 	"sigs.k8s.io/controller-runtime/pkg/log"
 	"sigs.k8s.io/controller-runtime/pkg/predicate"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 	"sigs.k8s.io/controller-runtime/pkg/source"
 
 	"github.com/ydb-platform/ydb-kubernetes-operator/api/v1alpha1"
+	ydbannotations "github.com/ydb-platform/ydb-kubernetes-operator/internal/annotations"
 	. "github.com/ydb-platform/ydb-kubernetes-operator/internal/controllers/constants" //nolint:revive,stylecheck
 	"github.com/ydb-platform/ydb-kubernetes-operator/internal/resources"
 )
@@ -77,6 +81,40 @@ func (r *Reconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Resu
 		r.Log.Error(err, "unexpected Get error")
 		return ctrl.Result{RequeueAfter: DefaultRequeueDelay}, err
 	}
+
+	//nolint:nestif
+	// examine DeletionTimestamp to determine if object is under deletion
+	if resource.ObjectMeta.DeletionTimestamp.IsZero() {
+		// The object is not being deleted, so if it does not have our finalizer,
+		// then lets add the finalizer and update the object. This is equivalent
+		// to registering our finalizer.
+		if !controllerutil.ContainsFinalizer(resource, ydbannotations.StorageFinalizerKey) {
+			controllerutil.AddFinalizer(resource, ydbannotations.StorageFinalizerKey)
+			if err := r.Client.Update(ctx, resource); err != nil {
+				return ctrl.Result{RequeueAfter: DefaultRequeueDelay}, err
+			}
+		}
+	} else {
+		// The object is being deleted
+		if controllerutil.ContainsFinalizer(resource, ydbannotations.StorageFinalizerKey) {
+			// our finalizer is present, so lets handle any external dependency
+			if err := r.checkExistingDatabases(ctx, resource); err != nil {
+				// if fail to check dependency existence, return with error
+				// so that it can be retried.
+				return ctrl.Result{RequeueAfter: DefaultRequeueDelay}, err
+			}
+
+			// remove our finalizer from the list and update it.
+			controllerutil.RemoveFinalizer(resource, ydbannotations.StorageFinalizerKey)
+			if err := r.Client.Update(ctx, resource); err != nil {
+				return ctrl.Result{RequeueAfter: DefaultRequeueDelay}, err
+			}
+		}
+
+		// Stop reconciliation as the item is being deleted
+		return ctrl.Result{Requeue: false}, nil
+	}
+
 	result, err := r.Sync(ctx, resource)
 	if err != nil {
 		r.Log.Error(err, "unexpected Sync error")
@@ -130,6 +168,18 @@ func createFieldIndexers(mgr ctrl.Manager) error {
 		return err
 	}
 
+	if err := mgr.GetFieldIndexer().IndexField(
+		context.Background(),
+		&v1alpha1.Database{},
+		StorageRefField,
+		func(obj client.Object) []string {
+			// grab the Database object, extract the .spec.storageRef.name...
+			database := obj.(*v1alpha1.Database)
+			return []string{database.Spec.StorageClusterRef.Name}
+		}); err != nil {
+		return err
+	}
+
 	return mgr.GetFieldIndexer().IndexField(
 		context.Background(),
 		&v1alpha1.Storage{},
@@ -212,3 +262,46 @@ func (r *Reconciler) findStoragesForSecret(secret client.Object) []reconcile.Req
 	}
 	return requests
 }
+
+func (r *Reconciler) checkExistingDatabases(
+	ctx context.Context,
+	storage *v1alpha1.Storage,
+) error {
+	databaseList := &v1alpha1.DatabaseList{}
+	err := r.Client.List(
+		ctx,
+		databaseList,
+		client.InNamespace(storage.Namespace),
+		client.MatchingFields{
+			StorageRefField: storage.Name,
+		},
+	)
+	if err != nil {
+		r.Log.Error(err, "failed to list Databases")
+		r.Recorder.Event(
+			storage,
+			corev1.EventTypeWarning,
+			"ControllerError",
+			fmt.Sprintf("Failed to list Databases: %s", err),
+		)
+		return err
+	}
+
+	if len(databaseList.Items) > 0 {
+		var databases []string
+		for _, database := range databaseList.Items {
+			databases = append(databases, database.Name)
+		}
+		errMessage := fmt.Sprintf("Waiting for existing Databases to be deleted: %v", databases)
+		r.Log.Info(errMessage)
+		r.Recorder.Event(
+			storage,
+			corev1.EventTypeNormal,
+			string(StorageProvisioning),
+			fmt.Sprintf(errMessage, databases),
+		)
+		return errors.New(errMessage)
+	}
+
+	return nil
+}