Scope controllers for a logical cluster

Author: vincepri

pkg/builder/controller.go

@@ -36,6 +36,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/predicate"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 	"sigs.k8s.io/controller-runtime/pkg/source"
+	"sigs.k8s.io/logical-cluster"
 )
 
 // Supporting mocking out functions for testing.
@@ -60,6 +61,7 @@ type Builder struct {
 	watchesInput     []WatchesInput
 	mgr              manager.Manager
 	cluster          cluster.Cluster
+	logicalName      logical.Name
 	globalPredicates []predicate.Predicate
 	ctrl             controller.Controller
 	ctrlOptions      controller.Options
@@ -69,6 +71,7 @@ type Builder struct {
 func (blder *Builder) clone() *Builder {
 	clone := *blder
 	clone.cluster = nil
+	clone.logicalName = ""
 	clone.ctrl = nil
 	return &clone
 }
@@ -244,9 +247,10 @@ func (blder *Builder) Build(r reconcile.Reconciler) (controller.Controller, erro
 		return nil, blder.forInput.err
 	}
 
-	if err := blder.mgr.AddRunnableBuilder(func(cl cluster.Cluster) (manager.Runnable, error) {
+	if err := blder.mgr.AddLogicalRunnableBuilder(func(name logical.Name, cl cluster.Cluster) (manager.Runnable, error) {
 		cloned := blder.clone()
 		cloned.cluster = cl
+		cloned.logicalName = name
 		if err := cloned.do(r); err != nil {
 			return nil, err
 		}
@@ -351,6 +355,14 @@ func (blder *Builder) doWatch() error {
 				return err
 			}
 			srckind.Type = typeForSrc
+		} else if !ok {
+			// If we're building a logical controller, raw watches are not allowed
+			// given that the cache cannot be validated to be coming from the same cluter.
+			// In the future, we could consider allowing this by satisfying a new interface
+			// that sets and uses the cluster.
+			if blder.logicalName != "" {
+				return fmt.Errorf("when using a logical adapter, custom raw watches %T are not allowed", w.src)
+			}
 		}
 
 		if err := blder.ctrl.Watch(w.src, w.eventhandler, allPredicates...); err != nil {
@@ -378,6 +390,10 @@ func (blder *Builder) doController(r reconcile.Reconciler) error {
 		ctrlOptions.Reconciler = r
 	}
 
+	if blder.logicalName != "" {
+		ctrlOptions.LogicalCluster = blder.logicalName
+	}
+
 	// Retrieve the GVK from the object we're reconciling
 	// to prepopulate logger information, and to optionally generate a default name.
 	var gvk schema.GroupVersionKind

pkg/controller/controller.go

@@ -33,6 +33,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/ratelimiter"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 	"sigs.k8s.io/controller-runtime/pkg/source"
+	"sigs.k8s.io/logical-cluster"
 )
 
 // Options are the arguments for creating a new Controller.
@@ -50,6 +51,10 @@ type Options struct {
 	// LogConstructor is used to construct a logger used for this controller and passed
 	// to each reconciliation via the context field.
 	LogConstructor func(request *reconcile.Request) logr.Logger
+
+	// LogicalCluster is populated when the controller was created for a logical cluster.
+	// This is used to determine if watch events without a logical.Name should be ignored.
+	LogicalCluster logical.Name
 }
 
 // Controller implements a Kubernetes API.  A Controller manages a work queue fed reconcile.Requests
@@ -121,7 +126,8 @@ func New(name string, mgr manager.Manager, options Options) (Controller, error)
 
 	// Create controller with dependencies set
 	return &controller.Controller{
-		Do: options.Reconciler,
+		Cluster: options.LogicalCluster,
+		Do:      options.Reconciler,
 		MakeQueue: func() workqueue.RateLimitingInterface {
 			return workqueue.NewNamedRateLimitingQueue(options.RateLimiter, name)
 		},

pkg/internal/controller/controller.go

@@ -42,6 +42,10 @@ type Controller struct {
 	// Name is used to uniquely identify a Controller in tracing, logging and monitoring.  Name is required.
 	Name string
 
+	// Cluster is the logical cluster that this controller is running against.
+	// +optional
+	Cluster logical.Name
+
 	// MaxConcurrentReconciles is the maximum number of concurrent Reconciles which can be run. Defaults to 1.
 	MaxConcurrentReconciles int
 
@@ -317,7 +321,10 @@ func (c *Controller) reconcileHandler(ctx context.Context, obj interface{}) {
 	ctx = addReconcileID(ctx, reconcileID)
 
 	// Set the Cluster on the request if it is set on the context.
-	req.Cluster = logical.FromContext(ctx)
+	if req.Cluster != c.Cluster {
+		panic(fmt.Sprintf("controller was setup for logical cluster %q, got a request for a cluster %q, not allowed!", c.Cluster, req.Cluster))
+	}
+	req.Cluster = c.Cluster
 
 	// RunInformersAndControllers the syncHandler, passing it the Namespace/Name string of the
 	// resource to be synced.

pkg/manager/internal.go

@@ -81,7 +81,7 @@ type controllerManager struct {
 	errChan              chan error
 	runnables            *runnables
 
-	runnableBuilders []func(cl cluster.Cluster) (Runnable, error)
+	runnableBuilders []func(logical.Name, cluster.Cluster) (Runnable, error)
 
 	// defaultCluster holds a variety of methods to interact with a defaultCluster. Required.
 	defaultCluster        cluster.Cluster
@@ -319,7 +319,7 @@ func (cm *controllerManager) syncClusterAwareRunnables() {
 			}
 
 			// Build the runnable.
-			runnable, err := build(cluster)
+			runnable, err := build(name, cluster)
 			if err != nil {
 				cluster.runnableBuilds = append(cluster.runnableBuilds, err)
 				cm.logger.Error(err, "failed to build cluster aware runnable, won't retry", "clusterName", name)
@@ -410,7 +410,7 @@ func (cm *controllerManager) removeLogicalCluster(name logical.Name) error {
 	return nil
 }
 
-func (cm *controllerManager) AddRunnableBuilder(fn func(cl cluster.Cluster) (Runnable, error)) error {
+func (cm *controllerManager) AddLogicalRunnableBuilder(fn func(name logical.Name, cl cluster.Cluster) (Runnable, error)) error {
 	cm.Lock()
 	defer cm.Unlock()
 	cm.runnableBuilders = append(cm.runnableBuilders, fn)

pkg/manager/manager.go

@@ -67,9 +67,9 @@ type Manager interface {
 	// election was configured.
 	Elected() <-chan struct{}
 
-	// AddRunnableBuilder adds a controller builder to the manager, which is used to build
+	// AddLogicalRunnableBuilder adds a controller builder to the manager, which is used to build
 	// controllers for a given cluster. This is useful when the Manager is running against many logical clusters.
-	AddRunnableBuilder(func(cluster.Cluster) (Runnable, error)) error
+	AddLogicalRunnableBuilder(func(logical.Name, cluster.Cluster) (Runnable, error)) error
 
 	// AddMetricsExtraHandler adds an extra handler served on path to the http server that serves metrics.
 	// Might be useful to register some diagnostic endpoints e.g. pprof. Note that these endpoints meant to be

pkg/manager/manager_test.go

@@ -1691,7 +1691,7 @@ var _ = Describe("manger.Manager", func() {
 
 			var built atomic.Bool
 			var started atomic.Bool
-			err = m.AddRunnableBuilder(func(cl cluster.Cluster) (Runnable, error) {
+			err = m.AddLogicalRunnableBuilder(func(name logical.Name, cl cluster.Cluster) (Runnable, error) {
 				built.Store(true)
 				return RunnableFunc(func(ctx context.Context) error {
 					Expect(logical.FromContext(ctx)).To(Equal(logical.Name("test-cluster")))
@@ -1735,7 +1735,7 @@ var _ = Describe("manger.Manager", func() {
 			var started atomic.Int64
 			var completed atomic.Int64
 			removedCh := make(chan struct{})
-			err = m.AddRunnableBuilder(func(cl cluster.Cluster) (Runnable, error) {
+			err = m.AddLogicalRunnableBuilder(func(name logical.Name, cl cluster.Cluster) (Runnable, error) {
 				built.Add(1)
 				return RunnableFunc(func(ctx context.Context) error {
 					defer completed.Add(1)