feat: reconcile redis sentinel only on master changed (#1122)

* feat: redissentinel reconcile only on master changed

Signed-off-by: drivebyer <[email protected]>

* fix Lint

Signed-off-by: drivebyer <[email protected]>

* e2e cleanup

Signed-off-by: drivebyer <[email protected]>

---------

Signed-off-by: drivebyer <[email protected]>

example/v1beta2/redis-replication.yaml

@@ -20,9 +20,9 @@ spec:
       limits:
         cpu: 101m
         memory: 128Mi
-    redisSecret:
-      name: redis-secret
-      key: password
+#    redisSecret:
+#      name: redis-secret
+#      key: password
       # imagePullSecrets:
       #   - name: regcred
   redisExporter:

main.go

@@ -27,6 +27,7 @@ import (
 	"github.com/OT-CONTAINER-KIT/redis-operator/pkg/controllers/rediscluster"
 	"github.com/OT-CONTAINER-KIT/redis-operator/pkg/controllers/redisreplication"
 	"github.com/OT-CONTAINER-KIT/redis-operator/pkg/controllers/redissentinel"
+	intctrlutil "github.com/OT-CONTAINER-KIT/redis-operator/pkg/controllerutil"
 	"github.com/OT-CONTAINER-KIT/redis-operator/pkg/k8sutils"
 	"k8s.io/apimachinery/pkg/runtime"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
@@ -153,11 +154,12 @@ func main() {
 		os.Exit(1)
 	}
 	if err = (&redissentinel.RedisSentinelReconciler{
-		Client:     mgr.GetClient(),
-		K8sClient:  k8sclient,
-		Dk8sClient: dk8sClient,
-		Log:        ctrl.Log.WithName("controllers").WithName("RedisSentinel"),
-		Scheme:     mgr.GetScheme(),
+		Client:             mgr.GetClient(),
+		K8sClient:          k8sclient,
+		Dk8sClient:         dk8sClient,
+		Log:                ctrl.Log.WithName("controllers").WithName("RedisSentinel"),
+		Scheme:             mgr.GetScheme(),
+		ReplicationWatcher: intctrlutil.NewResourceWatcher(),
 	}).SetupWithManager(mgr); err != nil {
 		setupLog.Error(err, "unable to create controller", "controller", "RedisSentinel")
 		os.Exit(1)

pkg/controllers/redissentinel/redissentinel_controller.go

@@ -9,6 +9,7 @@ import (
 	"github.com/OT-CONTAINER-KIT/redis-operator/pkg/k8sutils"
 	"github.com/go-logr/logr"
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/dynamic"
 	"k8s.io/client-go/kubernetes"
 	ctrl "sigs.k8s.io/controller-runtime"
@@ -22,6 +23,8 @@ type RedisSentinelReconciler struct {
 	Dk8sClient dynamic.Interface
 	Log        logr.Logger
 	Scheme     *runtime.Scheme
+
+	ReplicationWatcher *intctrlutil.ResourceWatcher
 }
 
 func (r *RedisSentinelReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
@@ -55,6 +58,17 @@ func (r *RedisSentinelReconciler) Reconcile(ctx context.Context, req ctrl.Reques
 		return intctrlutil.RequeueAfter(reqLogger, time.Second*10, "Redis Replication is specified but not ready")
 	}
 
+	if instance.Spec.RedisSentinelConfig != nil {
+		r.ReplicationWatcher.Watch(
+			ctx,
+			types.NamespacedName{
+				Namespace: req.Namespace,
+				Name:      instance.Spec.RedisSentinelConfig.RedisReplicationName,
+			},
+			req.NamespacedName,
+		)
+	}
+
 	// Create Redis Sentinel
 	err = k8sutils.CreateRedisSentinel(ctx, r.K8sClient, r.Log, instance, r.K8sClient, r.Dk8sClient)
 	if err != nil {
@@ -71,12 +85,13 @@ func (r *RedisSentinelReconciler) Reconcile(ctx context.Context, req ctrl.Reques
 	if err != nil {
 		return intctrlutil.RequeueWithError(err, reqLogger, "")
 	}
-	return intctrlutil.RequeueAfter(reqLogger, time.Second*10, "")
+	return intctrlutil.Reconciled()
 }
 
 // SetupWithManager sets up the controller with the Manager.
 func (r *RedisSentinelReconciler) SetupWithManager(mgr ctrl.Manager) error {
 	return ctrl.NewControllerManagedBy(mgr).
 		For(&redisv1beta2.RedisSentinel{}).
+		Watches(&redisv1beta2.RedisReplication{}, r.ReplicationWatcher).
 		Complete(r)
 }

pkg/controllers/redissentinel/redissentinel_controller_suite_test.go

@@ -23,6 +23,7 @@ import (
 	"time"
 
 	redisv1beta2 "github.com/OT-CONTAINER-KIT/redis-operator/api/v1beta2"
+	intctrlutil "github.com/OT-CONTAINER-KIT/redis-operator/pkg/controllerutil"
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
 	"github.com/onsi/gomega/gexec"
@@ -100,10 +101,11 @@ var _ = BeforeSuite(func() {
 	Expect(err).ToNot(HaveOccurred())
 
 	err = (&RedisSentinelReconciler{
-		Client:     k8sManager.GetClient(),
-		K8sClient:  k8sClient,
-		Dk8sClient: dk8sClient,
-		Scheme:     k8sManager.GetScheme(),
+		Client:             k8sManager.GetClient(),
+		K8sClient:          k8sClient,
+		Dk8sClient:         dk8sClient,
+		Scheme:             k8sManager.GetScheme(),
+		ReplicationWatcher: intctrlutil.NewResourceWatcher(),
 	}).SetupWithManager(k8sManager)
 	Expect(err).ToNot(HaveOccurred())
 

pkg/controllerutil/resource_watcher.go

@@ -0,0 +1,76 @@
+package controllerutil
+
+import (
+	"context"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/client-go/util/workqueue"
+	"sigs.k8s.io/controller-runtime/pkg/event"
+	"sigs.k8s.io/controller-runtime/pkg/handler"
+	"sigs.k8s.io/controller-runtime/pkg/reconcile"
+)
+
+// ResourceWatcher implements handler.EventHandler and is used to trigger reconciliation when
+// a watched object changes. It's designed to only be used for a single type of object.
+// If multiple types should be watched, one ResourceWatcher for each type should be used.
+type ResourceWatcher struct {
+	watched map[types.NamespacedName][]types.NamespacedName
+}
+
+var _ handler.EventHandler = &ResourceWatcher{}
+
+// NewResourceWatcher will create a new ResourceWatcher with no watched objects.
+func NewResourceWatcher() *ResourceWatcher {
+	return &ResourceWatcher{
+		watched: make(map[types.NamespacedName][]types.NamespacedName),
+	}
+}
+
+// Watch will add a new object to watch.
+func (w ResourceWatcher) Watch(ctx context.Context, watchedName, dependentName types.NamespacedName) {
+	existing, hasExisting := w.watched[watchedName]
+	if !hasExisting {
+		existing = []types.NamespacedName{}
+	}
+
+	for _, dependent := range existing {
+		if dependent == dependentName {
+			return
+		}
+	}
+	w.watched[watchedName] = append(existing, dependentName)
+}
+
+func (w ResourceWatcher) Create(ctx context.Context, event event.CreateEvent, queue workqueue.RateLimitingInterface) {
+	w.handleEvent(event.Object, queue)
+}
+
+func (w ResourceWatcher) Update(ctx context.Context, event event.UpdateEvent, queue workqueue.RateLimitingInterface) {
+	w.handleEvent(event.ObjectOld, queue)
+}
+
+func (w ResourceWatcher) Delete(ctx context.Context, event event.DeleteEvent, queue workqueue.RateLimitingInterface) {
+	w.handleEvent(event.Object, queue)
+}
+
+func (w ResourceWatcher) Generic(ctx context.Context, event event.GenericEvent, queue workqueue.RateLimitingInterface) {
+	w.handleEvent(event.Object, queue)
+}
+
+// handleEvent is called when an event is received for an object.
+// It will check if the object is being watched and trigger a reconciliation for
+// the dependent object.
+func (w ResourceWatcher) handleEvent(meta metav1.Object, queue workqueue.RateLimitingInterface) {
+	changedObjectName := types.NamespacedName{
+		Name:      meta.GetName(),
+		Namespace: meta.GetNamespace(),
+	}
+
+	// Enqueue reconciliation for each dependent object.
+	for _, dep := range w.watched[changedObjectName] {
+		queue.Add(reconcile.Request{
+			NamespacedName: dep,
+		})
+	}
+}

tests/e2e-chainsaw/v1beta2/ha-failover/chainsaw-test.yaml → tests/e2e-chainsaw/v1beta2/setup/ha/chainsaw-test.yaml

@@ -2,7 +2,7 @@
 apiVersion: chainsaw.kyverno.io/v1alpha1
 kind: Test
 metadata:
-  name: ha-failover
+  name: setup-ha
 spec:
   steps:
     - try: