PTEUDO-2201: Job for syncing DBClaim labels into DBInstance

api/v1/condition.go

@@ -2,6 +2,7 @@ package v1
 
 import (
 	"fmt"
+
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
@@ -100,6 +101,15 @@ func ReconcileErrorCondition(err error) metav1.Condition {
 	)
 }
 
+func ReconcileSyncErrorCondition(err error) metav1.Condition {
+	return CreateCondition(
+		ConditionSync,
+		metav1.ConditionFalse,
+		ReasonUnavailable,
+		fmt.Sprintf("Reconciliation encountered an issue: %v", err),
+	)
+}
+
 func ReconcileSuccessCondition() metav1.Condition {
 	return CreateCondition(
 		ConditionReady,

cmd/main.go

@@ -17,6 +17,7 @@ limitations under the License.
 package main
 
 import (
+	"context"
 	"crypto/tls"
 	"flag"
 	"fmt"
@@ -40,12 +41,12 @@ import (
 	"github.com/infobloxopen/db-controller/internal/controller"
 	"github.com/infobloxopen/db-controller/internal/metrics"
 	mutating "github.com/infobloxopen/db-controller/internal/webhook"
+	webhookpersistancev1 "github.com/infobloxopen/db-controller/internal/webhook/v1"
 	"github.com/infobloxopen/db-controller/pkg/config"
 	"github.com/infobloxopen/db-controller/pkg/databaseclaim"
 	"github.com/infobloxopen/db-controller/pkg/rdsauth"
 	"github.com/infobloxopen/db-controller/pkg/roleclaim"
 
-	webhookpersistancev1 "github.com/infobloxopen/db-controller/internal/webhook/v1"
 	// +kubebuilder:scaffold:imports
 	crossplanerdsv1alpha1 "github.com/crossplane-contrib/provider-aws/apis/rds/v1alpha1"
 	crossplanegcpv1beta2 "github.com/upbound/provider-gcp/apis/alloydb/v1beta2"
@@ -67,6 +68,7 @@ func init() {
 	utilruntime.Must(crossplanerdsv1alpha1.SchemeBuilder.AddToScheme(scheme))
 
 	utilruntime.Must(crossplanegcpv1beta2.SchemeBuilder.AddToScheme(scheme))
+
 }
 
 func main() {
@@ -75,6 +77,7 @@ func main() {
 	var probeAddr string
 	var secureMetrics bool
 	var enableHTTP2 bool
+	var enableLabelPropagation bool
 	var tlsOpts []func(*tls.Config)
 	flag.StringVar(&metricsAddr, "metrics-bind-address", "0", "The address the metrics endpoint binds to. "+
 		"Use :8443 for HTTPS or :8080 for HTTP, or leave as 0 to disable the metrics service.")
@@ -87,6 +90,8 @@ func main() {
 		"If set, the metrics endpoint is served securely via HTTPS. Use --metrics-secure=false to use HTTP instead.")
 	flag.BoolVar(&enableHTTP2, "enable-http2", false,
 		"If set, HTTP/2 will be enabled for the metrics and webhook servers")
+	flag.BoolVar(&enableLabelPropagation, "enable-label-propagation", false,
+		"Enable the propagation of DatabaseClaim labels to DBInstance objects")
 
 	var class string
 	var configFile string
@@ -248,6 +253,14 @@ func main() {
 		os.Exit(1)
 	}
 
+	if err := (&controller.DBInstanceStatusReconciler{
+		Client: mgr.GetClient(),
+		Scheme: mgr.GetScheme(),
+	}).SetupWithManager(mgr); err != nil {
+		setupLog.Error(err, "unable to create controller", "controller", "DBInstanceStatus")
+		os.Exit(1)
+	}
+
 	// +kubebuilder:scaffold:builder
 
 	if err := mgr.AddHealthzCheck("healthz", healthz.Ping); err != nil {
@@ -294,4 +307,23 @@ func main() {
 		setupLog.Error(err, "problem running manager")
 		os.Exit(1)
 	}
+
+	// Start the manager
+	setupLog.Info("Starting manager")
+	if err := mgr.Start(ctrl.SetupSignalHandler()); err != nil {
+		setupLog.Error(err, "problem running manager")
+		os.Exit(1)
+	}
+
+	// Start label propagation logic if enabled.
+	if enableLabelPropagation {
+		setupLog.Info("Starting label propagation for DBInstances")
+		go func() {
+			if err := controller.SyncDBInstances(context.Background(), ctlConfig, mgr.GetClient(), setupLog); err != nil {
+				setupLog.Error(err, "failed to propagate labels for dbinstances")
+			} else {
+				setupLog.Info("label propagation completed successfully")
+			}
+		}()
+	}
 }

config/crd/bases/_.yaml

@@ -0,0 +1,13 @@
+---
+apiVersion: apiextensions.k8s.io/v1
+kind: CustomResourceDefinition
+metadata:
+  annotations:
+    controller-gen.kubebuilder.io/version: v0.15.0
+spec:
+  group: ""
+  names:
+    kind: ""
+    plural: ""
+  scope: ""
+  versions: null

config/rbac/role.yaml

@@ -11,6 +11,14 @@ rules:
   verbs:
   - create
   - patch
+- apiGroups:
+  - database.aws.crossplane.io
+  resources:
+  - dbinstances
+  verbs:
+  - get
+  - list
+  - watch
 - apiGroups:
   - persistance.atlas.infoblox.com
   resources:

internal/controller/dbinstance_labels.go

@@ -0,0 +1,126 @@
+package controller
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"strings"
+
+	"github.com/go-logr/logr"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	"github.com/crossplane-contrib/provider-aws/apis/rds/v1alpha1"
+	v1 "github.com/infobloxopen/db-controller/api/v1"
+	"github.com/infobloxopen/db-controller/pkg/databaseclaim"
+	"github.com/spf13/viper"
+)
+
+// SyncDBInstances synchronizes the labels of DBInstances with their associated DatabaseClaims.
+func SyncDBInstances(ctx context.Context, viper *viper.Viper, kubeClient client.Client, logger logr.Logger) error {
+	logger.Info("starting synchronization of dbinstance labels")
+
+	// Dynamically retrieve the prefix from the environment configuration.
+	prefix := viper.GetString("env")
+	logger.Info("using dynamic prefix from environment", "prefix", prefix)
+
+	// List all DBInstances.
+	var dbInstances v1alpha1.DBInstanceList
+	if err := kubeClient.List(ctx, &dbInstances); err != nil {
+		return fmt.Errorf("error listing DBInstances: %w", err)
+	}
+
+	logger.Info("dbinstances fetched successfully", "total_dbinstances", len(dbInstances.Items))
+
+	for _, dbInstance := range dbInstances.Items {
+		instanceLogger := logger.WithValues("DBInstanceName", dbInstance.Name)
+		instanceLogger.Info("processing DBInstance")
+
+		// Remove the prefix from the DBInstance name to derive the DatabaseClaim name.
+		nameWithoutPrefix := strings.TrimPrefix(dbInstance.Name, prefix+"-")
+		nameWithoutSuffix := nameWithoutPrefix
+
+		if len(nameWithoutSuffix) > 9 {
+			nameWithoutSuffix = nameWithoutSuffix[:len(nameWithoutSuffix)-9]
+		}
+		dbClaimName := strings.TrimSpace(nameWithoutSuffix)
+
+		instanceLogger.Info("derived databaseclaim name", "dbclaim_ref", dbClaimName)
+
+		if dbClaimName == "" {
+			instanceLogger.Error(errors.New("empty DatabaseClaim name"), "skipping DBInstance due to invalid naming format")
+			continue
+		}
+
+		// Attempt to fetch the associated DatabaseClaim by field selector on name.
+		dbClaimsTest := v1.DatabaseClaimList{}
+		fieldSelectorOptions := []client.ListOption{
+			client.MatchingFields{
+				"metadata.name": dbClaimName,
+			},
+		}
+		if err := kubeClient.List(ctx, &dbClaimsTest, fieldSelectorOptions...); err != nil {
+			instanceLogger.Error(err, "failed to fetch DatabaseClaim", "DatabaseClaimName", dbClaimName)
+			continue
+		}
+		if len(dbClaimsTest.Items) == 0 {
+			instanceLogger.Info("no DatabaseClaim found for DBInstance", "DatabaseClaimName", dbClaimName)
+			continue
+		}
+		dbClaim := dbClaimsTest.Items[0]
+		instanceLogger.Info("DatabaseClaim fetched successfully", "DatabaseClaimLabels", dbClaim.Labels)
+
+		// Propagate labels from the DatabaseClaim.
+		newLabels := databaseclaim.PropagateLabels(dbClaim.Labels)
+		if len(newLabels) == 0 {
+			instanceLogger.Info("no labels to propagate from DatabaseClaim", "DatabaseClaimName", dbClaimName)
+			continue
+		}
+
+		// Update the DBInstance labels.
+		instanceLogger.Info("updating DBInstance labels", "currentlabels", dbInstance.Labels, "newlabels", newLabels)
+
+		if err := updateDBInstanceLabels(ctx, kubeClient, &dbInstance, newLabels, instanceLogger); err != nil {
+			instanceLogger.Error(err, "failed to update labels for DBInstance")
+			continue
+		}
+
+		instanceLogger.Info("labels updated successfully for DBInstance", "updatedlabels", dbInstance.Labels)
+	}
+
+	logger.Info("synchronization of DBInstance labels completed successfully")
+	return nil
+}
+
+// updateDBInstanceLabels updates the labels of a DBInstance while preserving existing labels.
+func updateDBInstanceLabels(ctx context.Context, kubeClient client.Client, dbInstance *v1alpha1.DBInstance, newLabels map[string]string, logger logr.Logger) error {
+	logger.Info("starting update of DBInstance labels")
+
+	if dbInstance.Labels == nil {
+		dbInstance.Labels = make(map[string]string)
+	}
+
+	updated := false
+
+	for key, value := range newLabels {
+		if oldValue, exists := dbInstance.Labels[key]; exists && oldValue == value {
+			logger.Info("label already exists and is unchanged", "key", key, "value", value)
+			continue
+		}
+		dbInstance.Labels[key] = value
+		updated = true
+		logger.Info("label added or updated", "key", key, "value", value)
+	}
+
+	if !updated {
+		logger.Info("no label updates required for DBInstance")
+		return nil
+	}
+
+	// Attempt to update the DBInstance.
+	logger.Info("applying updated labels to DBInstance", "UpdatedLabels", dbInstance.Labels)
+	if err := kubeClient.Update(ctx, dbInstance); err != nil {
+		return fmt.Errorf("error updating DBInstance labels: %w", err)
+	}
+
+	return nil
+}

internal/controller/dbinstance_labels_test.go

@@ -0,0 +1,123 @@
+package controller
+
+import (
+	"context"
+	"path/filepath"
+
+	"github.com/crossplane-contrib/provider-aws/apis/rds/v1alpha1"
+	persistencev1 "github.com/infobloxopen/db-controller/api/v1"
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+	"github.com/spf13/viper"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes/scheme"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/envtest"
+	"sigs.k8s.io/controller-runtime/pkg/log/zap"
+)
+
+// Helper function to create string pointers
+func strPtr(s string) *string {
+	return &s
+}
+
+var _ = Describe("DBInstance Labels Management with envtest", func() {
+	var (
+		testEnv    *envtest.Environment
+		k8sClient  client.Client
+		testLogger = zap.New(zap.UseDevMode(true))
+		ctx        context.Context
+		cancel     context.CancelFunc
+	)
+
+	BeforeEach(func() {
+		// Set up a new context and environment for each test
+		ctx, cancel = context.WithCancel(context.Background())
+		testEnv = &envtest.Environment{
+			CRDDirectoryPaths: []string{
+				filepath.Join("..", "..", "config", "crd", "bases"),
+				filepath.Join("..", "..", "test", "crd"),
+			},
+			ErrorIfCRDPathMissing: true,
+		}
+
+		// Start the test environment
+		cfg, err := testEnv.Start()
+		Expect(err).NotTo(HaveOccurred())
+		Expect(cfg).NotTo(BeNil())
+
+		// Initialize the Kubernetes client
+		k8sClient, err = client.New(cfg, client.Options{})
+		Expect(err).NotTo(HaveOccurred())
+		Expect(k8sClient).NotTo(BeNil())
+
+		// Register custom resource schemas
+		err = v1alpha1.SchemeBuilder.AddToScheme(scheme.Scheme)
+		Expect(err).NotTo(HaveOccurred())
+
+		err = persistencev1.AddToScheme(scheme.Scheme)
+		Expect(err).NotTo(HaveOccurred())
+	})
+
+	AfterEach(func() {
+		cancel()
+		Expect(testEnv.Stop()).To(Succeed())
+	})
+
+	It("Job should replicate labels from DatabaseClaim to DBInstance", func() {
+		dbClaim := &persistencev1.DatabaseClaim{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      "redirect-dbapi",
+				Namespace: "default",
+				Labels: map[string]string{
+					"app.kubernetes.io/component": "database",
+					"app.kubernetes.io/instance":  "redirect",
+					"app.kubernetes.io/name":      "redirect",
+				},
+			},
+		}
+		Expect(k8sClient.Create(ctx, dbClaim)).To(Succeed())
+
+		dbInstance := &v1alpha1.DBInstance{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: "box-3-redirect-dbapi-2f2d3cd1",
+				Labels: map[string]string{
+					"existing.label": "unchanged",
+				},
+			},
+			Spec: v1alpha1.DBInstanceSpec{
+				ForProvider: v1alpha1.DBInstanceParameters{
+					DBInstanceClass: strPtr("db.t3.micro"),
+					Engine:          strPtr("postgres"),
+				},
+			},
+		}
+		Expect(k8sClient.Create(ctx, dbInstance)).To(Succeed())
+
+		// Execute the SyncDBInstances function.
+		viperInstance := viper.New()
+		viperInstance.Set("env", "box-3")
+		err := SyncDBInstances(ctx, viperInstance, k8sClient, testLogger)
+		Expect(err).NotTo(HaveOccurred())
+
+		// Verify that labels were propagated to the DBInstance
+		updatedDBInstance := &v1alpha1.DBInstance{}
+		Eventually(func() map[string]string {
+			_ = k8sClient.Get(ctx, client.ObjectKey{Name: "box-3-redirect-dbapi-2f2d3cd1"}, updatedDBInstance)
+			return updatedDBInstance.Labels
+		}, "10s", "1s").Should(HaveKeyWithValue("app.kubernetes.io/component", "database"))
+		Eventually(func() map[string]string {
+			_ = k8sClient.Get(ctx, client.ObjectKey{Name: "box-3-redirect-dbapi-2f2d3cd1"}, updatedDBInstance)
+			return updatedDBInstance.Labels
+		}, "10s", "1s").Should(HaveKeyWithValue("app.kubernetes.io/instance", "redirect"))
+		Eventually(func() map[string]string {
+			_ = k8sClient.Get(ctx, client.ObjectKey{Name: "box-3-redirect-dbapi-2f2d3cd1"}, updatedDBInstance)
+			return updatedDBInstance.Labels
+		}, "10s", "1s").Should(HaveKeyWithValue("app.kubernetes.io/name", "redirect"))
+		Eventually(func() map[string]string {
+			_ = k8sClient.Get(ctx, client.ObjectKey{Name: "box-3-redirect-dbapi-2f2d3cd1"}, updatedDBInstance)
+			return updatedDBInstance.Labels
+		}, "10s", "1s").Should(HaveKeyWithValue("existing.label", "unchanged"))
+	})
+
+})