feat(crdb): use crdb native backup and s3 backup added (#1915)

* fix(zitadelctl): implement takedown command * fix(zitadelctl): correct destroy flow * fix(zitadelctl): correct backup commands to read crds beforehand * fix: add of destroyfile * fix: clean for userlist * fix: change backup and restore to crdb native * fix: timeout for delete pvc for cockroachdb * fix: corrected unit tests * fix: add ignored file for scale * fix: correct handling of gitops in backup command * feat: add s3 backup kind * fix: backuplist for s3 and timeout for pv deletion * fix(database): fix nil pointer with binary version * fix(database): cleanup of errors which cam with merging of the s3 logic * fix: correct unit tests * fix: cleanup monitor output Co-authored-by: Elio Bischof <[email protected]> * fix: backup imagepullpolixy to ifnotpresent Co-authored-by: Elio Bischof <[email protected]>
2026-11-02 04:12:16 +00:00 · 2021-10-13 14:34:03 +02:00
parent 591a460450
commit 425a8b5fd5
67 changed files with 3867 additions and 626 deletions
@@ -33,6 +33,8 @@ const (
 	privateServiceName = SfsName
 	cockroachPort      = int32(26257)
 	cockroachHTTPPort  = int32(8080)
+	Clean              = "clean"
+	DBReady            = "dbready"
 )

 func Adapter(
@@ -89,14 +91,14 @@ func Adapter(

 		var (
 			isFeatureDatabase bool
-			isFeatureRestore  bool
+			isFeatureClean    bool
 		)
 		for _, feature := range features {
 			switch feature {
 			case "database":
 				isFeatureDatabase = true
-			case "restore":
-				isFeatureRestore = true
+			case Clean:
+				isFeatureClean = true
 			}
 		}

@@ -179,9 +181,6 @@ func Adapter(
 				queryS,
 				operator.EnsureFuncToQueryFunc(ensureInit),
 			)
-		}
-
-		if isFeatureDatabase {
 			destroyers = append(destroyers,
 				destroyS,
 				operator.ResourceDestroyToZitadelDestroy(destroySFS),
@@ -191,6 +190,21 @@ func Adapter(
 			)
 		}

+		if isFeatureClean {
+			queriers = append(queriers,
+				operator.ResourceQueryToZitadelQuery(
+					statefulset.CleanPVCs(
+						monitor,
+						namespace,
+						cockroachSelectabel,
+						desiredKind.Spec.ReplicaCount,
+					),
+				),
+				operator.EnsureFuncToQueryFunc(ensureInit),
+				operator.EnsureFuncToQueryFunc(checkDBReady),
+			)
+		}
+
 		if desiredKind.Spec.Backups != nil {

 			oneBackup := false
@@ -215,6 +229,8 @@ func Adapter(
 						nodeselector,
 						tolerations,
 						version,
+						PublicServiceName,
+						cockroachPort,
 						features,
 						customImageRegistry,
 					)
@@ -233,21 +249,19 @@ func Adapter(
 		}

 		return func(k8sClient kubernetes.ClientInt, queried map[string]interface{}) (operator.EnsureFunc, error) {
-				if !isFeatureRestore {
-					queriedCurrentDB, err := core.ParseQueriedForDatabase(queried)
-					if err != nil || queriedCurrentDB == nil {
-						// TODO: query system state
-						currentDB.Current.Port = strconv.Itoa(int(cockroachPort))
-						currentDB.Current.URL = PublicServiceName
-						currentDB.Current.ReadyFunc = checkDBReady
-						currentDB.Current.AddUserFunc = addUser
-						currentDB.Current.DeleteUserFunc = deleteUser
-						currentDB.Current.ListUsersFunc = listUsers
-						currentDB.Current.ListDatabasesFunc = listDatabases
+				queriedCurrentDB, err := core.ParseQueriedForDatabase(queried)
+				if err != nil || queriedCurrentDB == nil {
+					// TODO: query system state
+					currentDB.Current.Port = strconv.Itoa(int(cockroachPort))
+					currentDB.Current.URL = PublicServiceName
+					currentDB.Current.ReadyFunc = checkDBReady
+					currentDB.Current.AddUserFunc = addUser
+					currentDB.Current.DeleteUserFunc = deleteUser
+					currentDB.Current.ListUsersFunc = listUsers
+					currentDB.Current.ListDatabasesFunc = listDatabases

-						core.SetQueriedForDatabase(queried, current)
-						internalMonitor.Info("set current state of managed database")
-					}
+					core.SetQueriedForDatabase(queried, current)
+					internalMonitor.Info("set current state of managed database")
 				}

 				ensure, err := operator.QueriersToEnsureFunc(internalMonitor, true, queriers, k8sClient, queried)
@@ -11,7 +11,6 @@ import (
 	"github.com/caos/orbos/pkg/tree"
 	"github.com/caos/zitadel/operator/database/kinds/backups/bucket"
 	"github.com/caos/zitadel/operator/database/kinds/backups/bucket/backup"
-	"github.com/caos/zitadel/operator/database/kinds/backups/bucket/clean"
 	"github.com/caos/zitadel/operator/database/kinds/backups/bucket/restore"
 	"github.com/golang/mock/gomock"
 	"github.com/stretchr/testify/assert"
@@ -66,11 +65,11 @@ func TestManaged_AdaptBucketBackup(t *testing.T) {
 	timestamp := "testTs"
 	nodeselector := map[string]string{"test": "test"}
 	tolerations := []corev1.Toleration{}
-	version := "testVersion"
 	k8sClient := kubernetesmock.NewMockClientInt(gomock.NewController(t))
 	backupName := "testBucket"
 	saJson := "testSA"
 	masterkey := "testMk"
+	version := "test"

 	desired := getTreeWithDBAndBackup(t, masterkey, saJson, backupName)

@@ -106,11 +105,11 @@ func TestManaged_AdaptBucketInstantBackup(t *testing.T) {
 	timestamp := "testTs"
 	nodeselector := map[string]string{"test": "test"}
 	tolerations := []corev1.Toleration{}
-	version := "testVersion"
 	masterkey := "testMk"
 	k8sClient := kubernetesmock.NewMockClientInt(gomock.NewController(t))
 	saJson := "testSA"
 	backupName := "testBucket"
+	version := "test"

 	features := []string{backup.Instant}
 	bucket.SetInstantBackup(k8sClient, namespace, backupName, labels, saJson)
@@ -152,10 +151,10 @@ func TestManaged_AdaptBucketCleanAndRestore(t *testing.T) {
 	saJson := "testSA"
 	backupName := "testBucket"

-	features := []string{restore.Instant, clean.Instant}
+	features := []string{restore.Instant}
 	bucket.SetRestore(k8sClient, namespace, backupName, labels, saJson)
-	bucket.SetClean(k8sClient, namespace, backupName, labels, saJson)
-	k8sClient.EXPECT().WaitUntilStatefulsetIsReady(namespace, SfsName, true, true, 60*time.Second).Times(2)
+	//SetClean(k8sClient, namespace, 1)
+	k8sClient.EXPECT().WaitUntilStatefulsetIsReady(namespace, SfsName, true, true, 60*time.Second).Times(1)

 	desired := getTreeWithDBAndBackup(t, masterkey, saJson, backupName)

@@ -66,12 +66,12 @@ func TestManaged_Adapt1(t *testing.T) {
 	timestamp := "testTs"
 	nodeselector := map[string]string{"test": "test"}
 	tolerations := []corev1.Toleration{}
-	version := "testVersion"
 	features := []string{"database"}
 	masterkey := "testMk"
 	k8sClient := kubernetesmock.NewMockClientInt(gomock.NewController(t))
 	dbCurrent := coremock.NewMockDatabaseCurrent(gomock.NewController(t))
 	queried := map[string]interface{}{}
+	version := "test"

 	desired := getDesiredTree(t, masterkey, &DesiredV0{
 		Common: tree.NewCommon("databases.caos.ch/CockroachDB", "v0", false),
@@ -130,16 +130,7 @@ func TestManaged_Adapt1(t *testing.T) {
 	dbCurrent.EXPECT().SetCertificateKey(gomock.Any()).MinTimes(1).MaxTimes(1)
 	k8sClient.EXPECT().ApplySecret(gomock.Any()).MinTimes(1).MaxTimes(1)

-	query, _, _, _, _, _, err := Adapter(
-		componentLabels,
-		namespace,
-		timestamp,
-		nodeselector,
-		tolerations,
-		version,
-		features,
-		"",
-	)(monitor, desired, &tree.Tree{})
+	query, _, _, _, _, _, err := Adapter(componentLabels, namespace, timestamp, nodeselector, tolerations, version, features, "")(monitor, desired, &tree.Tree{})
 	assert.NoError(t, err)

 	ensure, err := query(k8sClient, queried)
@@ -184,12 +175,12 @@ func TestManaged_Adapt2(t *testing.T) {

 	nodeselector := map[string]string{"test2": "test2"}
 	var tolerations []corev1.Toleration
-	version := "testVersion2"
 	features := []string{"database"}
 	masterkey := "testMk2"
 	k8sClient := kubernetesmock.NewMockClientInt(gomock.NewController(t))
 	dbCurrent := coremock.NewMockDatabaseCurrent(gomock.NewController(t))
 	queried := map[string]interface{}{}
+	version := "test"

 	desired := getDesiredTree(t, masterkey, &DesiredV0{
 		Common: tree.NewCommon("databases.caos.ch/CockroachDB", "v0", false),
@@ -248,16 +239,7 @@ func TestManaged_Adapt2(t *testing.T) {
 	dbCurrent.EXPECT().SetCertificateKey(gomock.Any()).MinTimes(1).MaxTimes(1)
 	k8sClient.EXPECT().ApplySecret(gomock.Any()).MinTimes(1).MaxTimes(1)

-	query, _, _, _, _, _, err := Adapter(
-		componentLabels,
-		namespace,
-		timestamp,
-		nodeselector,
-		tolerations,
-		version,
-		features,
-		"",
-	)(monitor, desired, &tree.Tree{})
+	query, _, _, _, _, _, err := Adapter(componentLabels, namespace, timestamp, nodeselector, tolerations, version, features, "")(monitor, desired, &tree.Tree{})
 	assert.NoError(t, err)

 	ensure, err := query(k8sClient, queried)
@@ -0,0 +1,35 @@
+package managed
+
+import (
+	kubernetesmock "github.com/caos/orbos/pkg/kubernetes/mock"
+	"github.com/golang/mock/gomock"
+	core "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"time"
+)
+
+func SetClean(
+	k8sClient *kubernetesmock.MockClientInt,
+	namespace string,
+	replicas int,
+) {
+	k8sClient.EXPECT().ScaleStatefulset(namespace, gomock.Any(), 0).Return(nil)
+	k8sClient.EXPECT().ListPersistentVolumeClaims(namespace).Return(&core.PersistentVolumeClaimList{
+		Items: []core.PersistentVolumeClaim{
+			{ObjectMeta: metav1.ObjectMeta{
+				Name: "datadir-cockroachdb-0",
+			}},
+		},
+	}, nil)
+	k8sClient.EXPECT().ScaleStatefulset(namespace, gomock.Any(), 1).Return(nil)
+	k8sClient.EXPECT().DeletePersistentVolumeClaim(namespace, gomock.Any(), gomock.Any()).Times(replicas).Return(nil)
+	k8sClient.EXPECT().WaitUntilStatefulsetIsReady(namespace, gomock.Any(), true, false, gomock.Any())
+	k8sClient.EXPECT().WaitUntilStatefulsetIsReady(namespace, gomock.Any(), true, true, time.Second*1)
+	k8sClient.EXPECT().WaitUntilStatefulsetIsReady(namespace, gomock.Any(), true, true, gomock.Any())
+
+	/*
+		k8sClient.EXPECT().ApplyJob(gomock.Any()).Times(1).Return(nil)
+		k8sClient.EXPECT().GetJob(namespace, clean.GetJobName(backupName)).Times(1).Return(nil, macherrs.NewNotFound(schema.GroupResource{"batch", "jobs"}, clean.GetJobName(backupName)))
+		k8sClient.EXPECT().WaitUntilJobCompleted(namespace, clean.GetJobName(backupName), gomock.Any()).Times(1).Return(nil)
+		k8sClient.EXPECT().DeleteJob(namespace, clean.GetJobName(backupName)).Times(1).Return(nil)*/
+}
@@ -34,6 +34,7 @@ const (
 	defaultMode         = int32(256)
 	nodeSecret          = "cockroachdb.node"
 	rootSecret          = "cockroachdb.client.root"
+	cleanTimeout        = time.Minute * 5
 )

 type Affinity struct {
@@ -0,0 +1,60 @@
+package statefulset
+
+import (
+	"github.com/caos/orbos/mntr"
+	"github.com/caos/orbos/pkg/kubernetes"
+	"github.com/caos/orbos/pkg/kubernetes/resources"
+	"github.com/caos/orbos/pkg/labels"
+	macherrs "k8s.io/apimachinery/pkg/api/errors"
+	"strings"
+	"time"
+)
+
+func CleanPVCs(
+	monitor mntr.Monitor,
+	namespace string,
+	sfsSelectable *labels.Selectable,
+	replicaCount int,
+) resources.QueryFunc {
+	name := sfsSelectable.Name()
+	return func(k8sClient kubernetes.ClientInt) (resources.EnsureFunc, error) {
+		pvcs, err := k8sClient.ListPersistentVolumeClaims(namespace)
+		if err != nil {
+			return nil, err
+		}
+		internalPvcs := []string{}
+		for _, pvc := range pvcs.Items {
+			if strings.HasPrefix(pvc.Name, datadirInternal+"-"+name) {
+				internalPvcs = append(internalPvcs, pvc.Name)
+			}
+		}
+		return func(k8sClient kubernetes.ClientInt) error {
+			noSFS := false
+			monitor.Info("Scale down statefulset")
+			if err := k8sClient.ScaleStatefulset(namespace, name, 0); err != nil {
+				if macherrs.IsNotFound(err) {
+					noSFS = true
+				} else {
+					return err
+				}
+			}
+			time.Sleep(2 * time.Second)
+
+			monitor.Info("Delete persistent volume claims")
+			for _, pvcName := range internalPvcs {
+				if err := k8sClient.DeletePersistentVolumeClaim(namespace, pvcName, cleanTimeout); err != nil {
+					return err
+				}
+			}
+			time.Sleep(2 * time.Second)
+
+			if !noSFS {
+				monitor.Info("Scale up statefulset")
+				if err := k8sClient.ScaleStatefulset(namespace, name, replicaCount); err != nil {
+					return err
+				}
+			}
+			return nil
+		}, nil
+	}
+}