implement the dbinstance status watcher

api/v1/condition.go

@@ -2,6 +2,7 @@ package v1
 
 import (
 	"fmt"
+
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
@@ -100,6 +101,19 @@ func ReconcileErrorCondition(err error) metav1.Condition {
 	)
 }
 
+func ReconcileSyncErrorCondition(err error) metav1.Condition {
+	message := "Reconciliation encountered an issue"
+	if err != nil {
+		message = fmt.Sprintf("%s: %v", message, err)
+	}
+	return CreateCondition(
+		ConditionSync,
+		metav1.ConditionFalse,
+		ReasonUnavailable,
+		message,
+	)
+}
+
 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"
@@ -75,6 +76,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 +89,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
@@ -237,7 +241,6 @@ func main() {
 		setupLog.Error(err, "unable to create controller", "controller", "DbRoleClaim")
 		os.Exit(1)
 	}
-
 	if err = webhookpersistancev1.SetupDatabaseClaimWebhookWithManager(mgr); err != nil {
 		setupLog.Error(err, "unable to create webhook", "webhook", "DatabaseClaim")
 		os.Exit(1)
@@ -247,6 +250,13 @@ func main() {
 		setupLog.Error(err, "unable to create webhook", "webhook", "DbRoleClaim")
 		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
 
@@ -289,9 +299,22 @@ func main() {
 	ctx := ctrl.SetupSignalHandler()
 	go metrics.StartUpdater(ctx, mgr.GetClient())
 
+	// Start the manager.
 	setupLog.Info("starting manager")
-	if err := mgr.Start(ctx); err != nil {
+	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/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/databaseclaim_controller_test.go

@@ -291,9 +291,8 @@ var _ = Describe("DatabaseClaim Controller", func() {
 			Expect(k8sClient.Get(ctx, typeNamespacedName, resource)).NotTo(HaveOccurred())
 
 			resource.Labels = map[string]string{
-				"app.kubernetes.io/component": resource.Labels["app.kubernetes.io/component"],
-				"app.kubernetes.io/instance":  resource.Labels["app.kubernetes.io/instance"],
-				"app.kubernetes.io/name":      resource.Labels["app.kubernetes.io/name"],
+				"app.kubernetes.io/dbclaim-name":      resource.Name,
+				"app.kubernetes.io/dbclaim-namespace": resource.Namespace,
 			}
 			Expect(k8sClient.Update(ctx, resource)).To(Succeed())
 

internal/controller/dbinstance_labels.go

@@ -0,0 +1,98 @@
+package controller
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/go-logr/logr"
+	v1 "github.com/infobloxopen/db-controller/api/v1"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	"github.com/crossplane-contrib/provider-aws/apis/rds/v1alpha1"
+	crossplaneaws "github.com/crossplane-contrib/provider-aws/apis/rds/v1alpha1"
+	"github.com/spf13/viper"
+)
+
+// SyncDBInstances ensures that each DBInstance has the appropriate dbclaim-name and dbclaim-namespace labels.
+func SyncDBInstances(ctx context.Context, viper *viper.Viper, kubeClient client.Client, logger logr.Logger) error {
+	logger.Info("starting synchronization of DBInstance labels")
+
+	// List all DBInstances.
+	var dbInstances crossplaneaws.DBInstanceList
+	if err := kubeClient.List(ctx, &dbInstances); err != nil {
+		return fmt.Errorf("error listing DBInstances: %w", err)
+	}
+
+	// Process each DBInstance.
+	for _, dbInstance := range dbInstances.Items {
+		instanceLogger := logger.WithValues("DBInstanceName", dbInstance.Name)
+		instanceLogger.Info("processing DBInstance")
+
+		// Extract dbclaim-name and dbclaim-namespace from existing DBInstance labels.
+		dbClaimName := dbInstance.Labels["app.kubernetes.io/dbclaim-name"]
+		dbClaimNamespace := dbInstance.Labels["app.kubernetes.io/dbclaim-namespace"]
+
+		// Update labels if any are missing.
+		if dbClaimName == "" || dbClaimNamespace == "" {
+			instanceLogger.Info("dbclaim-name or dbclaim-namespace labels are missing; proceeding to update")
+
+			newLabels := map[string]string{
+				"app.kubernetes.io/dbclaim-name":      dbInstance.Name,
+				"app.kubernetes.io/dbclaim-namespace": "default",
+			}
+			if err := updateDBInstanceLabels(ctx, kubeClient, &dbInstance, newLabels, instanceLogger); err != nil {
+				instanceLogger.Error(err, "failed to update labels for DBInstance")
+				continue
+			}
+		} else {
+			instanceLogger.Info("DBInstance has valid dbclaim-name and dbclaim-namespace labels; no update necessary")
+		}
+
+		// Check if the DatabaseClaim exists.
+		var dbClaim v1.DatabaseClaim
+		if err := kubeClient.Get(ctx, client.ObjectKey{Name: dbClaimName, Namespace: dbClaimNamespace}, &dbClaim); err != nil {
+			instanceLogger.Error(err, "DatabaseClaim not found for DBInstance")
+			return fmt.Errorf("error updating DBInstance %s: %w", dbInstance.Name, err)
+		}
+
+		instanceLogger.Info("labels updated successfully")
+	}
+
+	logger.Info("synchronization of DBInstance labels completed successfully")
+	return nil
+}
+
+// updateDBInstanceLabels updates the labels of a DBInstance while preserving existing ones.
+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 {
+		logger.Info("DBInstance has no labels; initializing")
+		dbInstance.Labels = make(map[string]string)
+	}
+
+	updated := false
+
+	// Update or add new labels.
+	for key, value := range newLabels {
+		if oldValue, exists := dbInstance.Labels[key]; exists && oldValue == 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
+	}
+
+	// Apply the updated labels to 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
+}