Make reconcilers multicluster aware (#1215)

* Make reconcilers multicluster aware

* Fix linter issues

* Re-run generate for license header

Author: andrewstucki

operator/cmd/run/run.go

@@ -397,14 +397,12 @@ func Run(
 	if v2Controllers {
 		// Redpanda Reconciler
 		if err := (&redpandacontrollers.RedpandaReconciler{
-			KubeConfig:           mgr.GetConfig(),
-			Client:               mgr.GetClient(),
-			EventRecorder:        mgr.GetEventRecorderFor("RedpandaReconciler"),
-			LifecycleClient:      lifecycle.NewResourceClient(mgr, lifecycle.V2ResourceManagers(redpandaImage, sidecarImage, cloudSecrets)),
+			Manager:              mcmanager,
+			LifecycleClient:      lifecycle.NewResourceClient(mcmanager, lifecycle.V2ResourceManagers(redpandaImage, sidecarImage, cloudSecrets)),
 			ClientFactory:        factory,
 			CloudSecretsExpander: cloudExpander,
 			UseNodePools:         opts.enableV2NodepoolController,
-		}).SetupWithManager(ctx, mgr); err != nil {
+		}).SetupWithManager(ctx, mcmanager); err != nil {
 			setupLog.Error(err, "unable to create controller", "controller", "Redpanda")
 			return err
 		}
@@ -415,8 +413,8 @@ func Run(
 		// NodePool Reconciler
 		if opts.enableV2NodepoolController {
 			if err := (&redpandacontrollers.NodePoolReconciler{
-				Client: mgr.GetClient(),
-			}).SetupWithManager(ctx, mgr); err != nil {
+				Manager: mcmanager,
+			}).SetupWithManager(ctx, mcmanager); err != nil {
 				setupLog.Error(err, "unable to create controller", "controller", "NodePool")
 				return err
 			}
@@ -438,7 +436,7 @@ func Run(
 				return err
 			}
 
-			if err := (&consolecontroller.Controller{Ctl: ctl}).SetupWithManager(ctx, mgr); err != nil {
+			if err := (&consolecontroller.Controller{Ctl: ctl}).SetupWithManager(ctx, mcmanager); err != nil {
 				setupLog.Error(err, "unable to create controller", "controller", "Console")
 				return err
 			}

operator/internal/controller/console/controller.go

@@ -26,13 +26,16 @@ import (
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
+	mcbuilder "sigs.k8s.io/multicluster-runtime/pkg/builder"
+	mcreconcile "sigs.k8s.io/multicluster-runtime/pkg/reconcile"
 
 	"github.com/redpanda-data/redpanda-operator/charts/console/v3"
 	redpandav1alpha2 "github.com/redpanda-data/redpanda-operator/operator/api/redpanda/v1alpha2"
 	"github.com/redpanda-data/redpanda-operator/operator/api/redpanda/v1alpha2/conversion"
 	"github.com/redpanda-data/redpanda-operator/operator/internal/controller"
 	"github.com/redpanda-data/redpanda-operator/operator/pkg/functional"
 	"github.com/redpanda-data/redpanda-operator/pkg/kube"
+	"github.com/redpanda-data/redpanda-operator/pkg/multicluster"
 	"github.com/redpanda-data/redpanda-operator/pkg/otelutil/log"
 )
 
@@ -58,35 +61,37 @@ type Controller struct {
 	rng *rand.Rand
 }
 
-func (c *Controller) SetupWithManager(ctx context.Context, mgr ctrl.Manager) error {
+func (c *Controller) SetupWithManager(ctx context.Context, mgr multicluster.Manager) error {
 	// If rng is not set for testing, create and seed a new one.
 	if c.rng == nil {
 		// TODO: Weak RNG is probably acceptable here but best to doublecheck
 		c.rng = rand.New(rand.NewSource(time.Now().UnixMicro())) //nolint:gosec
 	}
 
-	builder := ctrl.NewControllerManagedBy(mgr)
+	builder := mcbuilder.ControllerManagedBy(mgr).
+		For(&redpandav1alpha2.Console{}, mcbuilder.WithEngageWithLocalCluster(true), mcbuilder.WithEngageWithProviderClusters(true))
 
 	// NB: As of writing, all console types are namespace scoped.
 	for _, t := range console.Types() {
-		builder = builder.Owns(t)
+		builder = builder.Owns(t, mcbuilder.WithEngageWithLocalCluster(true), mcbuilder.WithEngageWithProviderClusters(true))
 	}
 
-	eventHandler, err := controller.RegisterClusterSourceIndex(ctx, mgr, "console", &redpandav1alpha2.Console{}, &redpandav1alpha2.ConsoleList{})
-	if err != nil {
-		return err
-	}
+	for _, clusterName := range mgr.GetClusterNames() {
+		eventHandler, err := controller.RegisterClusterSourceIndex(ctx, mgr, "console", clusterName, &redpandav1alpha2.Console{}, &redpandav1alpha2.ConsoleList{})
+		if err != nil {
+			return err
+		}
 
-	return builder.
-		For(&redpandav1alpha2.Console{}).
 		// Configure a watch on redpandas using controller-runtime's indexing.
 		// If a redpanda is updated, any console's referring to it will be
 		// re-reconciled.
-		Watches(&redpandav1alpha2.Redpanda{}, eventHandler).
-		Complete(c)
+		builder.Watches(&redpandav1alpha2.Redpanda{}, eventHandler, controller.WatchOptions(clusterName)...)
+	}
+
+	return builder.Complete(c)
 }
 
-func (c *Controller) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
+func (c *Controller) Reconcile(ctx context.Context, req mcreconcile.Request) (ctrl.Result, error) {
 	cr, err := kube.Get[redpandav1alpha2.Console](ctx, c.Ctl, req.NamespacedName)
 	if err != nil {
 		if apierrors.IsNotFound(err) {

operator/internal/controller/console/controller_test.go

@@ -26,6 +26,8 @@ import (
 	"k8s.io/utils/ptr"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+	mcmanager "sigs.k8s.io/multicluster-runtime/pkg/manager"
+	mcreconcile "sigs.k8s.io/multicluster-runtime/pkg/reconcile"
 	"sigs.k8s.io/yaml"
 
 	consolechart "github.com/redpanda-data/redpanda-operator/charts/console/v3"
@@ -168,7 +170,9 @@ func TestController(t *testing.T) {
 
 			// Reconcile the console a few times to ensure determinism.
 			for range 3 {
-				_, err = consoleCtrl.Reconcile(t.Context(), ctrl.Request{NamespacedName: kube.AsKey(console)})
+				req := mcreconcile.Request{Request: ctrl.Request{NamespacedName: kube.AsKey(console)}, ClusterName: mcmanager.LocalCluster}
+
+				_, err = consoleCtrl.Reconcile(t.Context(), req)
 				require.NoError(t, err)
 
 				// Get updated console status
@@ -203,7 +207,9 @@ func TestController(t *testing.T) {
 
 			// Reconcile the deletion a few times.
 			for range 3 {
-				_, err = consoleCtrl.Reconcile(t.Context(), ctrl.Request{NamespacedName: kube.AsKey(console)})
+				req := mcreconcile.Request{Request: ctrl.Request{NamespacedName: kube.AsKey(console)}, ClusterName: mcmanager.LocalCluster}
+
+				_, err = consoleCtrl.Reconcile(t.Context(), req)
 				require.NoError(t, err)
 			}
 

operator/internal/controller/index.go

@@ -16,12 +16,12 @@ import (
 
 	"k8s.io/apimachinery/pkg/fields"
 	"k8s.io/apimachinery/pkg/types"
-	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
-	"sigs.k8s.io/controller-runtime/pkg/handler"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
+	mchandler "sigs.k8s.io/multicluster-runtime/pkg/handler"
 
 	redpandav1alpha2 "github.com/redpanda-data/redpanda-operator/operator/api/redpanda/v1alpha2"
+	"github.com/redpanda-data/redpanda-operator/pkg/multicluster"
 )
 
 type clientList[T client.Object] interface {
@@ -33,24 +33,32 @@ func clusterReferenceIndexName(name string) string {
 	return fmt.Sprintf("__%s_referencing_cluster", name)
 }
 
-func RegisterClusterSourceIndex[T redpandav1alpha2.ClusterReferencingObject, U clientList[T]](ctx context.Context, mgr ctrl.Manager, name string, o T, l U) (handler.EventHandler, error) {
+func RegisterClusterSourceIndex[T redpandav1alpha2.ClusterReferencingObject, U clientList[T]](ctx context.Context, mgr multicluster.Manager, name, clusterName string, o T, l U) (mchandler.EventHandlerFunc, error) {
 	indexName := clusterReferenceIndexName(name)
-	if err := mgr.GetFieldIndexer().IndexField(ctx, o, indexName, indexByClusterSource(func(cr *redpandav1alpha2.ClusterRef) bool {
+	cluster, err := mgr.GetCluster(ctx, clusterName)
+	if err != nil {
+		return nil, err
+	}
+	if err := cluster.GetFieldIndexer().IndexField(ctx, o, indexName, indexByClusterSource(func(cr *redpandav1alpha2.ClusterRef) bool {
 		return cr.IsV2()
 	})); err != nil {
 		return nil, err
 	}
-	return enqueueFromSourceCluster(mgr, name, l), nil
+	return enqueueFromSourceCluster(mgr, name, clusterName, l), nil
 }
 
-func RegisterV1ClusterSourceIndex[T redpandav1alpha2.ClusterReferencingObject, U clientList[T]](ctx context.Context, mgr ctrl.Manager, name string, o T, l U) (handler.EventHandler, error) {
+func RegisterV1ClusterSourceIndex[T redpandav1alpha2.ClusterReferencingObject, U clientList[T]](ctx context.Context, mgr multicluster.Manager, name, clusterName string, o T, l U) (mchandler.EventHandlerFunc, error) {
 	indexName := clusterReferenceIndexName(name)
-	if err := mgr.GetFieldIndexer().IndexField(ctx, o, indexName, indexByClusterSource(func(cr *redpandav1alpha2.ClusterRef) bool {
+	cluster, err := mgr.GetCluster(ctx, clusterName)
+	if err != nil {
+		return nil, err
+	}
+	if err := cluster.GetFieldIndexer().IndexField(ctx, o, indexName, indexByClusterSource(func(cr *redpandav1alpha2.ClusterRef) bool {
 		return cr.IsV1()
 	})); err != nil {
 		return nil, err
 	}
-	return enqueueFromSourceCluster(mgr, name, l), nil
+	return enqueueFromSourceCluster(mgr, name, clusterName, l), nil
 }
 
 func indexByClusterSource(checkRef func(*redpandav1alpha2.ClusterRef) bool) func(o client.Object) []string {
@@ -97,10 +105,15 @@ func sourceClusters[T client.Object, U clientList[T]](ctx context.Context, c cli
 	return requests, nil
 }
 
-func enqueueFromSourceCluster[T client.Object, U clientList[T]](mgr ctrl.Manager, name string, l U) handler.EventHandler {
-	return handler.EnqueueRequestsFromMapFunc(func(ctx context.Context, o client.Object) []reconcile.Request {
+func enqueueFromSourceCluster[T client.Object, U clientList[T]](mgr multicluster.Manager, name string, clusterName string, l U) mchandler.EventHandlerFunc {
+	return mchandler.EnqueueRequestsFromMapFunc(func(ctx context.Context, o client.Object) []reconcile.Request {
+		cluster, err := mgr.GetCluster(ctx, clusterName)
+		if err != nil {
+			mgr.GetLogger().V(1).Info(fmt.Sprintf("possibly skipping %s reconciliation due to failure to fetch %s associated with cluster", name, name), "error", err)
+			return nil
+		}
 		list := reflect.New(reflect.TypeOf(l).Elem()).Interface().(U)
-		requests, err := sourceClusters(ctx, mgr.GetClient(), list, name, client.ObjectKeyFromObject(o))
+		requests, err := sourceClusters(ctx, cluster.GetClient(), list, name, client.ObjectKeyFromObject(o))
 		if err != nil {
 			mgr.GetLogger().V(1).Info(fmt.Sprintf("possibly skipping %s reconciliation due to failure to fetch %s associated with cluster", name, name), "error", err)
 			return nil

operator/internal/controller/redpanda/nodepool_controller.go

@@ -13,6 +13,7 @@ import (
 	"context"
 	"reflect"
 	"strconv"
+	"time"
 
 	"go.opentelemetry.io/otel/attribute"
 	appsv1 "k8s.io/api/apps/v1"
@@ -22,15 +23,18 @@ import (
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
-	"sigs.k8s.io/controller-runtime/pkg/handler"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
+	mcbuilder "sigs.k8s.io/multicluster-runtime/pkg/builder"
+	mchandler "sigs.k8s.io/multicluster-runtime/pkg/handler"
+	mcreconcile "sigs.k8s.io/multicluster-runtime/pkg/reconcile"
 
 	"github.com/redpanda-data/redpanda-operator/charts/redpanda/v25"
 	redpandav1alpha2 "github.com/redpanda-data/redpanda-operator/operator/api/redpanda/v1alpha2"
 	"github.com/redpanda-data/redpanda-operator/operator/internal/controller"
 	"github.com/redpanda-data/redpanda-operator/operator/internal/lifecycle"
 	"github.com/redpanda-data/redpanda-operator/operator/internal/statuses"
 	"github.com/redpanda-data/redpanda-operator/operator/pkg/feature"
+	"github.com/redpanda-data/redpanda-operator/pkg/multicluster"
 	"github.com/redpanda-data/redpanda-operator/pkg/otelutil/log"
 	"github.com/redpanda-data/redpanda-operator/pkg/otelutil/otelkube"
 	"github.com/redpanda-data/redpanda-operator/pkg/otelutil/trace"
@@ -45,20 +49,14 @@ import (
 // NodePoolReconciler reconciles a NodePool object. This reconciler in particular should only update status
 // fields and finalizers on the NodePool objects, rendering of NodePools takes place within the RedpandaReconciler.
 type NodePoolReconciler struct {
-	Client client.Client
+	Manager multicluster.Manager
 }
 
 // SetupWithManager sets up the controller with the Manager.
-func (r *NodePoolReconciler) SetupWithManager(ctx context.Context, mgr ctrl.Manager) error {
-	enqueueNodePoolFromCluster, err := controller.RegisterClusterSourceIndex(ctx, mgr, "pool", &redpandav1alpha2.NodePool{}, &redpandav1alpha2.NodePoolList{})
-	if err != nil {
-		return err
-	}
-
-	return ctrl.NewControllerManagedBy(mgr).
-		For(&redpandav1alpha2.NodePool{}).
-		Watches(&redpandav1alpha2.Redpanda{}, enqueueNodePoolFromCluster).
-		Watches(&appsv1.StatefulSet{}, handler.EnqueueRequestsFromMapFunc(func(ctx context.Context, o client.Object) []reconcile.Request {
+func (r *NodePoolReconciler) SetupWithManager(ctx context.Context, mgr multicluster.Manager) error {
+	builder := mcbuilder.ControllerManagedBy(mgr).
+		For(&redpandav1alpha2.NodePool{}, mcbuilder.WithEngageWithLocalCluster(true), mcbuilder.WithEngageWithProviderClusters(true)).
+		Watches(&appsv1.StatefulSet{}, mchandler.EnqueueRequestsFromMapFunc(func(ctx context.Context, o client.Object) []reconcile.Request {
 			labels := o.GetLabels()
 			if labels == nil {
 				return nil
@@ -77,14 +75,40 @@ func (r *NodePoolReconciler) SetupWithManager(ctx context.Context, mgr ctrl.Mana
 					Name:      name,
 				},
 			}}
-		})).
-		Complete(r)
+		}))
+
+	for _, clusterName := range mgr.GetClusterNames() {
+		enqueueNodePoolFromCluster, err := controller.RegisterClusterSourceIndex(ctx, mgr, "pool", clusterName, &redpandav1alpha2.NodePool{}, &redpandav1alpha2.NodePoolList{})
+		if err != nil {
+			return err
+		}
+
+		builder.Watches(&redpandav1alpha2.Redpanda{}, enqueueNodePoolFromCluster, controller.WatchOptions(clusterName)...)
+	}
+
+	return builder.Complete(r)
 }
 
 // Reconcile reconciles NodePool objects
-func (r *NodePoolReconciler) Reconcile(ctx context.Context, req ctrl.Request) (result ctrl.Result, err error) {
+func (r *NodePoolReconciler) Reconcile(ctx context.Context, req mcreconcile.Request) (result ctrl.Result, err error) {
+	l := log.FromContext(ctx).WithName("NodePoolReconciler.Reconcile")
+	l.V(1).Info("Starting reconcile loop")
+	start := time.Now()
+	defer func() {
+		l.V(1).Info("Finished reconciling", "elapsed", time.Since(start))
+	}()
+
+	k8sCluster, err := r.Manager.GetCluster(ctx, req.ClusterName)
+	if err != nil {
+		l.Error(err, "unable to fetch cluster, skipping reconciliation")
+		return ctrl.Result{}, nil
+	}
+
+	k8sClient := k8sCluster.GetClient()
+
 	pool := &redpandav1alpha2.NodePool{}
-	if err := r.Client.Get(ctx, req.NamespacedName, pool); err != nil {
+
+	if err := k8sClient.Get(ctx, req.NamespacedName, pool); err != nil {
 		return ctrl.Result{}, client.IgnoreNotFound(err)
 	}
 
@@ -98,7 +122,7 @@ func (r *NodePoolReconciler) Reconcile(ctx context.Context, req ctrl.Request) (r
 
 	if !feature.V2Managed.Get(ctx, pool) {
 		if controllerutil.RemoveFinalizer(pool, FinalizerKey) {
-			if err := r.Client.Update(ctx, pool); err != nil {
+			if err := k8sClient.Update(ctx, pool); err != nil {
 				logger.Error(err, "updating cluster finalizer")
 				// no need to update the status at this point since the
 				// previous update failed
@@ -111,7 +135,7 @@ func (r *NodePoolReconciler) Reconcile(ctx context.Context, req ctrl.Request) (r
 	// Examine if the object is under deletion
 	if !pool.ObjectMeta.DeletionTimestamp.IsZero() {
 		if controllerutil.RemoveFinalizer(pool, FinalizerKey) {
-			if err := r.Client.Update(ctx, pool); err != nil {
+			if err := k8sClient.Update(ctx, pool); err != nil {
 				logger.Error(err, "updating cluster finalizer")
 				// no need to update the status at this point since the
 				// previous update failed
@@ -125,7 +149,7 @@ func (r *NodePoolReconciler) Reconcile(ctx context.Context, req ctrl.Request) (r
 	// If any changes are made, persist the changes and immediately requeue to
 	// prevent any cache / resource version synchronization issues.
 	if controllerutil.AddFinalizer(pool, FinalizerKey) || feature.SetDefaults(ctx, feature.V2Flags, pool) {
-		if err := r.Client.Update(ctx, pool); err != nil {
+		if err := k8sClient.Update(ctx, pool); err != nil {
 			logger.Error(err, "updating cluster finalizer or Annotation")
 			return ignoreConflict(err)
 		}
@@ -134,7 +158,7 @@ func (r *NodePoolReconciler) Reconcile(ctx context.Context, req ctrl.Request) (r
 
 	var status statuses.NodePoolStatus
 	var statefulSets appsv1.StatefulSetList
-	if err := r.Client.List(ctx, &statefulSets, client.MatchingLabels{
+	if err := k8sClient.List(ctx, &statefulSets, client.MatchingLabels{
 		lifecycle.DefaultNamespaceLabel: pool.Namespace,
 		redpanda.NodePoolLabelName:      pool.Name,
 	}); err != nil {
@@ -191,7 +215,7 @@ func (r *NodePoolReconciler) Reconcile(ctx context.Context, req ctrl.Request) (r
 	}
 
 	cluster := &redpandav1alpha2.Redpanda{}
-	if err := r.Client.Get(ctx, types.NamespacedName{Name: pool.Spec.ClusterRef.Name, Namespace: req.Namespace}, cluster); err != nil {
+	if err := k8sClient.Get(ctx, types.NamespacedName{Name: pool.Spec.ClusterRef.Name, Namespace: req.Namespace}, cluster); err != nil {
 		if apierrors.IsNotFound(err) {
 			status.SetBound(statuses.NodePoolBoundReasonNotBound)
 		} else {
@@ -204,7 +228,7 @@ func (r *NodePoolReconciler) Reconcile(ctx context.Context, req ctrl.Request) (r
 	if status.UpdateConditions(pool) ||
 		!reflect.DeepEqual(originalPoolStatus, pool.Status.EmbeddedNodePoolStatus) ||
 		(pool.Status.DeployedGeneration != originalPoolGeneration) {
-		return ignoreConflict(r.Client.Status().Update(ctx, pool))
+		return ignoreConflict(k8sClient.Status().Update(ctx, pool))
 	}
 
 	return ctrl.Result{}, nil