Refactor

Signed-off-by: SK Ali Arman <[email protected]>
kubedb · May 8, 2024 · 3e5996d · 3e5996d
1 parent f71f184
commit 3e5996d
Show file tree

Hide file tree

Showing 13 changed files with 346 additions and 225 deletions.
diff --git a/pkg/cmds/pause.go b/pkg/cmds/pause.go
@@ -183,6 +183,7 @@ func (o *PauseOptions) Run() error {
 			}
 			allErrs = append(allErrs, err)
 			errs.Insert(err.Error())
+			continue
 		}
 		pauseAll := !(o.onlyBackup || o.onlyDb || o.onlyArchiver)
 

diff --git a/pkg/pauser/archiver.go b/pkg/pauser/archiver.go
@@ -0,0 +1,157 @@
+package pauser
+
+import (
+	"context"
+
+	coreapi "kubedb.dev/apimachinery/apis/archiver/v1alpha1"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	kmapi "kmodules.xyz/client-go/api/v1"
+	kmc "kmodules.xyz/client-go/client"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+)
+
+func PauseOrResumeMySQLArchiver(klient client.Client, value bool, reference kmapi.ObjectReference) error {
+	name := reference.Name
+	namespace := reference.Namespace
+	archiver, err := getMysqlArchiver(klient, kmapi.ObjectReference{
+		Name:      name,
+		Namespace: namespace,
+	})
+	if err != nil {
+		return err
+	}
+	_, err = kmc.CreateOrPatch(
+		context.Background(),
+		klient,
+		archiver,
+		func(obj client.Object, createOp bool) client.Object {
+			in := obj.(*coreapi.MySQLArchiver)
+			in.Spec.Pause = value
+			return in
+		},
+	)
+	return err
+}
+
+func getMysqlArchiver(klient client.Client, ref kmapi.ObjectReference) (*coreapi.MySQLArchiver, error) {
+	archiver := &coreapi.MySQLArchiver{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      ref.Name,
+			Namespace: ref.Namespace,
+		},
+	}
+	if err := klient.Get(context.Background(), client.ObjectKeyFromObject(archiver), archiver); err != nil {
+		return nil, err
+	}
+	return archiver, nil
+}
+
+func PauseOrResumeMariaDBArchiver(klient client.Client, value bool, reference kmapi.ObjectReference) error {
+	name := reference.Name
+	namespace := reference.Namespace
+	archiver, err := getMariaDBArchiver(klient, kmapi.ObjectReference{
+		Name:      name,
+		Namespace: namespace,
+	})
+	if err != nil {
+		return err
+	}
+	_, err = kmc.CreateOrPatch(
+		context.Background(),
+		klient,
+		archiver,
+		func(obj client.Object, createOp bool) client.Object {
+			in := obj.(*coreapi.MariaDBArchiver)
+			in.Spec.Pause = value
+			return in
+		},
+	)
+	return err
+}
+
+func getMariaDBArchiver(klient client.Client, ref kmapi.ObjectReference) (*coreapi.MariaDBArchiver, error) {
+	archiver := &coreapi.MariaDBArchiver{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      ref.Name,
+			Namespace: ref.Namespace,
+		},
+	}
+	if err := klient.Get(context.Background(), client.ObjectKeyFromObject(archiver), archiver); err != nil {
+		return nil, err
+	}
+
+	return archiver, nil
+}
+
+func PauseOrResumePostgresArchiver(klient client.Client, value bool, reference kmapi.ObjectReference) error {
+	name := reference.Name
+	namespace := reference.Namespace
+	archiver, err := getPostgresArchiver(klient, kmapi.ObjectReference{
+		Name:      name,
+		Namespace: namespace,
+	})
+	if err != nil {
+		return err
+	}
+	_, err = kmc.CreateOrPatch(
+		context.Background(),
+		klient,
+		archiver,
+		func(obj client.Object, createOp bool) client.Object {
+			in := obj.(*coreapi.PostgresArchiver)
+			in.Spec.Pause = value
+			return in
+		},
+	)
+	return err
+}
+
+func getPostgresArchiver(klient client.Client, ref kmapi.ObjectReference) (*coreapi.PostgresArchiver, error) {
+	archiver := &coreapi.PostgresArchiver{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      ref.Name,
+			Namespace: ref.Namespace,
+		},
+	}
+	if err := klient.Get(context.Background(), client.ObjectKeyFromObject(archiver), archiver); err != nil {
+		return nil, err
+	}
+	return archiver, nil
+}
+
+func PauseOrResumeMongoDBArchiver(klient client.Client, value bool, reference kmapi.ObjectReference) error {
+	name := reference.Name
+	namespace := reference.Namespace
+	archiver, err := getMongoDBArchiver(klient, kmapi.ObjectReference{
+		Name:      name,
+		Namespace: namespace,
+	})
+	if err != nil {
+		return err
+	}
+	_, err = kmc.CreateOrPatch(
+		context.Background(),
+		klient,
+		archiver,
+		func(obj client.Object, createOp bool) client.Object {
+			in := obj.(*coreapi.MongoDBArchiver)
+			in.Spec.Pause = value
+			return in
+		},
+	)
+	return err
+}
+
+func getMongoDBArchiver(klient client.Client, ref kmapi.ObjectReference) (*coreapi.MongoDBArchiver, error) {
+	archiver := &coreapi.MongoDBArchiver{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      ref.Name,
+			Namespace: ref.Namespace,
+		},
+	}
+	if err := klient.Get(context.Background(), client.ObjectKeyFromObject(archiver), archiver); err != nil {
+		return nil, err
+	}
+	return archiver, nil
+}
diff --git a/pkg/pauser/mariadb.go b/pkg/pauser/mariadb.go
@@ -19,25 +19,30 @@ package pauser
 import (
 	"context"
 
+	coreapi "kubedb.dev/apimachinery/apis/archiver/v1alpha1"
 	api "kubedb.dev/apimachinery/apis/kubedb/v1alpha2"
 	cs "kubedb.dev/apimachinery/client/clientset/versioned/typed/kubedb/v1alpha2"
 	dbutil "kubedb.dev/apimachinery/client/clientset/versioned/typed/kubedb/v1alpha2/util"
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/rest"
+	kmc "kmodules.xyz/client-go/client"
 	condutil "kmodules.xyz/client-go/conditions"
+	"sigs.k8s.io/controller-runtime/pkg/client"
 	scs "stash.appscode.dev/apimachinery/client/clientset/versioned/typed/stash/v1beta1"
 )
 
 type MariaDBPauser struct {
-	dbClient    cs.KubedbV1alpha2Interface
-	stashClient scs.StashV1beta1Interface
-	onlyDb      bool
-	onlyBackup  bool
+	dbClient       cs.KubedbV1alpha2Interface
+	stashClient    scs.StashV1beta1Interface
+	uncachedClient client.Client
+	onlyDb         bool
+	onlyBackup     bool
+	onlyArchiver   bool
 }
 
-func NewMariaDBPauser(clientConfig *rest.Config, onlyDb, onlyBackup bool) (*MariaDBPauser, error) {
+func NewMariaDBPauser(clientConfig *rest.Config, onlyDb, onlyBackup, onlyArchiver bool) (*MariaDBPauser, error) {
 	dbClient, err := cs.NewForConfig(clientConfig)
 	if err != nil {
 		return nil, err
@@ -48,11 +53,18 @@ func NewMariaDBPauser(clientConfig *rest.Config, onlyDb, onlyBackup bool) (*Mari
 		return nil, err
 	}
 
+	uncachedClient, err := kmc.NewUncachedClient(clientConfig, coreapi.AddToScheme)
+	if err != nil {
+		return nil, err
+	}
+
 	return &MariaDBPauser{
-		dbClient:    dbClient,
-		stashClient: stashClient,
-		onlyDb:      onlyDb,
-		onlyBackup:  onlyBackup,
+		dbClient:       dbClient,
+		stashClient:    stashClient,
+		uncachedClient: uncachedClient,
+		onlyDb:         onlyDb,
+		onlyBackup:     onlyBackup,
+		onlyArchiver:   onlyArchiver,
 	}, nil
 }
 
@@ -62,7 +74,15 @@ func (e *MariaDBPauser) Pause(name, namespace string) (bool, error) {
 		return false, nil
 	}
 
-	pauseAll := !(e.onlyBackup || e.onlyDb)
+	pauseAll := !(e.onlyBackup || e.onlyDb || e.onlyArchiver)
+	if e.onlyArchiver || pauseAll {
+		if err := PauseOrResumeMariaDBArchiver(e.uncachedClient, true, db.Spec.Archiver.Ref); err != nil {
+			return false, err
+		}
+		if e.onlyArchiver {
+			return false, nil
+		}
+	}
 
 	if e.onlyDb || pauseAll {
 		_, err = dbutil.UpdateMariaDBStatus(context.TODO(), e.dbClient, db.ObjectMeta, func(status *api.MariaDBStatus) (types.UID, *api.MariaDBStatus) {

diff --git a/pkg/pauser/mongodb.go b/pkg/pauser/mongodb.go
@@ -19,23 +19,27 @@ package pauser
 import (
 	"context"
 
+	coreapi "kubedb.dev/apimachinery/apis/archiver/v1alpha1"
 	api "kubedb.dev/apimachinery/apis/kubedb/v1alpha2"
 	cs "kubedb.dev/apimachinery/client/clientset/versioned/typed/kubedb/v1alpha2"
 	dbutil "kubedb.dev/apimachinery/client/clientset/versioned/typed/kubedb/v1alpha2/util"
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/rest"
+	kmc "kmodules.xyz/client-go/client"
 	condutil "kmodules.xyz/client-go/conditions"
+	"sigs.k8s.io/controller-runtime/pkg/client"
 	scs "stash.appscode.dev/apimachinery/client/clientset/versioned/typed/stash/v1beta1"
 )
 
 type MongoDBPauser struct {
-	dbClient     cs.KubedbV1alpha2Interface
-	stashClient  scs.StashV1beta1Interface
-	onlyDb       bool
-	onlyBackup   bool
-	onlyArchiver bool
+	dbClient       cs.KubedbV1alpha2Interface
+	stashClient    scs.StashV1beta1Interface
+	uncachedClient client.Client
+	onlyDb         bool
+	onlyBackup     bool
+	onlyArchiver   bool
 }
 
 func NewMongoDBPauser(clientConfig *rest.Config, onlyDb, onlyBackup, onlyArchiver bool) (*MongoDBPauser, error) {
@@ -49,12 +53,18 @@ func NewMongoDBPauser(clientConfig *rest.Config, onlyDb, onlyBackup, onlyArchive
 		return nil, err
 	}
 
+	uncachedClient, err := kmc.NewUncachedClient(clientConfig, coreapi.AddToScheme)
+	if err != nil {
+		return nil, err
+	}
+
 	return &MongoDBPauser{
-		dbClient:     dbClient,
-		stashClient:  stashClient,
-		onlyDb:       onlyDb,
-		onlyBackup:   onlyBackup,
-		onlyArchiver: onlyArchiver,
+		dbClient:       dbClient,
+		stashClient:    stashClient,
+		uncachedClient: uncachedClient,
+		onlyDb:         onlyDb,
+		onlyBackup:     onlyBackup,
+		onlyArchiver:   onlyArchiver,
 	}, nil
 }
 
@@ -66,7 +76,7 @@ func (e *MongoDBPauser) Pause(name, namespace string) (bool, error) {
 
 	pauseAll := !(e.onlyBackup || e.onlyDb || e.onlyArchiver)
 	if e.onlyArchiver || pauseAll {
-		if err := PauseMongoDBArchiver(true, db.Spec.Archiver.Ref.Name, db.Spec.Archiver.Ref.Namespace); err != nil {
+		if err := PauseOrResumeMongoDBArchiver(e.uncachedClient, true, db.Spec.Archiver.Ref); err != nil {
 			return false, err
 		}
 		if e.onlyArchiver {

diff --git a/pkg/pauser/mysql.go b/pkg/pauser/mysql.go
@@ -19,23 +19,27 @@ package pauser
 import (
 	"context"
 
+	coreapi "kubedb.dev/apimachinery/apis/archiver/v1alpha1"
 	api "kubedb.dev/apimachinery/apis/kubedb/v1alpha2"
 	cs "kubedb.dev/apimachinery/client/clientset/versioned/typed/kubedb/v1alpha2"
 	dbutil "kubedb.dev/apimachinery/client/clientset/versioned/typed/kubedb/v1alpha2/util"
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/rest"
+	kmc "kmodules.xyz/client-go/client"
 	condutil "kmodules.xyz/client-go/conditions"
+	"sigs.k8s.io/controller-runtime/pkg/client"
 	scs "stash.appscode.dev/apimachinery/client/clientset/versioned/typed/stash/v1beta1"
 )
 
 type MySQLPauser struct {
-	dbClient     cs.KubedbV1alpha2Interface
-	stashClient  scs.StashV1beta1Interface
-	onlyDb       bool
-	onlyBackup   bool
-	onlyArchiver bool
+	dbClient       cs.KubedbV1alpha2Interface
+	stashClient    scs.StashV1beta1Interface
+	uncachedClient client.Client
+	onlyDb         bool
+	onlyBackup     bool
+	onlyArchiver   bool
 }
 
 func NewMySQLPauser(clientConfig *rest.Config, onlyDb, onlyBackup, onlyArchiver bool) (*MySQLPauser, error) {
@@ -49,12 +53,18 @@ func NewMySQLPauser(clientConfig *rest.Config, onlyDb, onlyBackup, onlyArchiver
 		return nil, err
 	}
 
+	uncachedClient, err := kmc.NewUncachedClient(clientConfig, coreapi.AddToScheme)
+	if err != nil {
+		return nil, err
+	}
+
 	return &MySQLPauser{
-		dbClient:     dbClient,
-		stashClient:  stashClient,
-		onlyDb:       onlyDb,
-		onlyBackup:   onlyBackup,
-		onlyArchiver: onlyArchiver,
+		dbClient:       dbClient,
+		stashClient:    stashClient,
+		uncachedClient: uncachedClient,
+		onlyDb:         onlyDb,
+		onlyBackup:     onlyBackup,
+		onlyArchiver:   onlyArchiver,
 	}, nil
 }
 
@@ -65,7 +75,7 @@ func (e *MySQLPauser) Pause(name string, namespace string) (bool, error) {
 	}
 	pauseAll := !(e.onlyBackup || e.onlyDb || e.onlyArchiver)
 	if e.onlyArchiver || pauseAll {
-		if err := PauseMySQLArchiver(true, db.Spec.Archiver.Ref.Name, db.Spec.Archiver.Ref.Namespace); err != nil {
+		if err := PauseOrResumeMySQLArchiver(e.uncachedClient, true, db.Spec.Archiver.Ref); err != nil {
 			return false, err
 		}
 		if e.onlyArchiver {

diff --git a/pkg/pauser/pauser.go b/pkg/pauser/pauser.go
@@ -46,7 +46,7 @@ func NewPauser(restClientGetter genericclioptions.RESTClientGetter, mapping *met
 	case api.ResourceKindMySQL:
 		return NewMySQLPauser(clientConfig, onlyDb, onlyBackup, onlyArchiver)
 	case api.ResourceKindMariaDB:
-		return NewMariaDBPauser(clientConfig, onlyDb, onlyBackup)
+		return NewMariaDBPauser(clientConfig, onlyDb, onlyBackup, onlyArchiver)
 	case api.ResourceKindPostgres:
 		return NewPostgresPauser(clientConfig, onlyDb, onlyBackup, onlyArchiver)
 	case api.ResourceKindRedis: