Add experimental support for pluggable cluster providers to manager

On-behalf-of: SAP [email protected]
Co-authored-by: Vince Prignano <[email protected]>
Co-authored-by: Dr. Stefan Schimanski <[email protected]>
Signed-off-by: Marvin Beckers <[email protected]>

Author: 3 people

pkg/cluster/multicluster.go

@@ -42,14 +42,17 @@ type Aware interface {
 	Disengage(context.Context, Cluster) error
 }
 
-// Provider defines methods to retrieve clusters by name. The provider is
-// responsible for discovering and managing the lifecycle of each cluster.
+// Provider allows to retrieve clusters by name. The provider is responsible for discovering
+// and managing the lifecycle of each cluster, calling `Engage` and `Disengage` on the manager
+// it is run against whenever a new cluster is discovered or a cluster is unregistered.
 //
 // Example: A Cluster API provider would be responsible for discovering and
 // managing clusters that are backed by Cluster API resources, which can live
 // in multiple namespaces in a single management cluster.
 type Provider interface {
 	// Get returns a cluster for the given identifying cluster name. Get
 	// returns an existing cluster if it has been created before.
+	// If no cluster is known to the provider under the given cluster name,
+	// an error should be returned.
 	Get(ctx context.Context, clusterName string) (Cluster, error)
 }

pkg/config/controller.go

@@ -59,4 +59,16 @@ type Controller struct {
 	//
 	// Note: This flag is disabled by default until a future version. It's currently in beta.
 	UsePriorityQueue bool
+
+	// EngageWithDefaultCluster indicates whether the controller should engage
+	// with the default cluster. This default to false if a cluster provider
+	// is configured, and to true otherwise.
+	//
+	// This is an experimental feature and is subject to change.
+	EngageWithDefaultCluster *bool
+
+	// EngageWithProvidedClusters indicates whether the controller should engage
+	// with the provided clusters of the manager. This defaults to true if a
+	// cluster provider is set, and to false otherwise.
+	EngageWithProviderClusters *bool
 }

pkg/manager/internal.go

@@ -59,6 +59,8 @@ const (
 )
 
 var _ Runnable = &controllerManager{}
+var _ cluster.Aware = &controllerManager{}
+var _ Manager = &controllerManager{}
 
 type controllerManager struct {
 	sync.Mutex
@@ -68,8 +70,14 @@ type controllerManager struct {
 	errChan              chan error
 	runnables            *runnables
 
-	// cluster holds a variety of methods to interact with a cluster. Required.
-	cluster cluster.Cluster
+	// defaultCluster holds a variety of methods to interact with a defaultCluster. Required.
+	defaultCluster        cluster.Cluster
+	defaultClusterOptions cluster.Option
+
+	// engagedCluster is a map of engaged clusters. The can come and go as the manager is running.
+	engagedClustersLock   sync.RWMutex
+	engagedClusters       map[string]cluster.Cluster
+	clusterAwareRunnables []cluster.Aware
 
 	// recorderProvider is used to generate event recorders that will be injected into Controllers
 	// (and EventHandlers, Sources and Predicates).
@@ -161,6 +169,9 @@ type controllerManager struct {
 	// internalProceduresStop channel is used internally to the manager when coordinating
 	// the proper shutdown of servers. This channel is also used for dependency injection.
 	internalProceduresStop chan struct{}
+
+	// clusterProvider is used to get clusters by name, beyond the default cluster.
+	clusterProvider cluster.Provider
 }
 
 type hasCache interface {
@@ -176,7 +187,40 @@ func (cm *controllerManager) Add(r Runnable) error {
 }
 
 func (cm *controllerManager) add(r Runnable) error {
-	return cm.runnables.Add(r)
+	var engaged []cluster.Aware
+	var errs []error
+	disengage := func() {
+		for _, aware := range engaged {
+			if err := aware.Disengage(cm.internalCtx, cm.defaultCluster); err != nil {
+				errs = append(errs, err)
+			}
+		}
+	}
+
+	// engage with existing clusters (this is reversible)
+	if aware, ok := r.(cluster.Aware); ok {
+		cm.engagedClustersLock.RLock()
+		defer cm.engagedClustersLock.RUnlock()
+		for _, cl := range cm.engagedClusters {
+			if err := aware.Engage(cm.internalCtx, cl); err != nil {
+				errs = append(errs, err)
+				break
+			}
+			engaged = append(engaged, aware)
+		}
+		if len(errs) > 0 {
+			disengage()
+			return kerrors.NewAggregate(errs)
+		}
+		cm.clusterAwareRunnables = append(cm.clusterAwareRunnables, aware)
+	} else {
+		if err := cm.runnables.Add(r); err != nil {
+			disengage()
+			return err
+		}
+	}
+
+	return nil
 }
 
 // AddMetricsServerExtraHandler adds extra handler served on path to the http server that serves metrics.
@@ -231,40 +275,58 @@ func (cm *controllerManager) AddReadyzCheck(name string, check healthz.Checker)
 	return nil
 }
 
+func (cm *controllerManager) Name() string {
+	return cm.defaultCluster.Name()
+}
+
+func (cm *controllerManager) GetCluster(ctx context.Context, clusterName string) (cluster.Cluster, error) {
+	if clusterName == "" || clusterName == cm.defaultCluster.Name() {
+		return cm.defaultCluster, nil
+	}
+
+	if cm.clusterProvider == nil {
+		return nil, fmt.Errorf("cluster %q not found, cluster provider is not set", clusterName)
+	}
+
+	// intentionally not returning from engaged clusters. This can be used
+	// without engaging clusters.
+	return cm.clusterProvider.Get(ctx, clusterName)
+}
+
 func (cm *controllerManager) GetHTTPClient() *http.Client {
-	return cm.cluster.GetHTTPClient()
+	return cm.defaultCluster.GetHTTPClient()
 }
 
 func (cm *controllerManager) GetConfig() *rest.Config {
-	return cm.cluster.GetConfig()
+	return cm.defaultCluster.GetConfig()
 }
 
 func (cm *controllerManager) GetClient() client.Client {
-	return cm.cluster.GetClient()
+	return cm.defaultCluster.GetClient()
 }
 
 func (cm *controllerManager) GetScheme() *runtime.Scheme {
-	return cm.cluster.GetScheme()
+	return cm.defaultCluster.GetScheme()
 }
 
 func (cm *controllerManager) GetFieldIndexer() client.FieldIndexer {
-	return cm.cluster.GetFieldIndexer()
+	return cm.defaultCluster.GetFieldIndexer()
 }
 
 func (cm *controllerManager) GetCache() cache.Cache {
-	return cm.cluster.GetCache()
+	return cm.defaultCluster.GetCache()
 }
 
 func (cm *controllerManager) GetEventRecorderFor(name string) record.EventRecorder {
-	return cm.cluster.GetEventRecorderFor(name)
+	return cm.defaultCluster.GetEventRecorderFor(name)
 }
 
 func (cm *controllerManager) GetRESTMapper() meta.RESTMapper {
-	return cm.cluster.GetRESTMapper()
+	return cm.defaultCluster.GetRESTMapper()
 }
 
 func (cm *controllerManager) GetAPIReader() client.Reader {
-	return cm.cluster.GetAPIReader()
+	return cm.defaultCluster.GetAPIReader()
 }
 
 func (cm *controllerManager) GetWebhookServer() webhook.Server {
@@ -381,7 +443,7 @@ func (cm *controllerManager) Start(ctx context.Context) (err error) {
 	}()
 
 	// Add the cluster runnable.
-	if err := cm.add(cm.cluster); err != nil {
+	if err := cm.add(cm.defaultCluster); err != nil {
 		return fmt.Errorf("failed to add cluster to runnables: %w", err)
 	}
 
@@ -614,6 +676,70 @@ func (cm *controllerManager) initLeaderElector() (*leaderelection.LeaderElector,
 	return leaderElector, nil
 }
 
+func (cm *controllerManager) Engage(ctx context.Context, cl cluster.Cluster) error {
+	cm.Lock()
+	defer cm.Unlock()
+
+	// be reentrant via noop
+	cm.engagedClustersLock.RLock()
+	if _, ok := cm.engagedClusters[cl.Name()]; ok {
+		cm.engagedClustersLock.RUnlock()
+		return nil
+	}
+	cm.engagedClustersLock.RUnlock()
+
+	// add early because any engaged runnable could access it
+	cm.engagedClustersLock.Lock()
+	cm.engagedClusters[cl.Name()] = cl
+	cm.engagedClustersLock.Unlock()
+
+	// engage known runnables
+	var errs []error
+	engaged := []cluster.Aware{}
+	for _, r := range cm.clusterAwareRunnables {
+		if err := r.Engage(ctx, cl); err != nil {
+			errs = append(errs, err)
+			break
+		}
+		engaged = append(engaged, r)
+	}
+
+	// clean-up
+	if len(errs) > 0 {
+		for _, aware := range engaged {
+			if err := aware.Disengage(ctx, cl); err != nil {
+				errs = append(errs, err)
+			}
+		}
+
+		cm.engagedClustersLock.Lock()
+		delete(cm.engagedClusters, cl.Name())
+		cm.engagedClustersLock.Unlock()
+
+		return kerrors.NewAggregate(errs)
+	}
+
+	return nil
+}
+
+func (cm *controllerManager) Disengage(ctx context.Context, cl cluster.Cluster) error {
+	cm.Lock()
+	defer cm.Unlock()
+
+	var errs []error
+	for _, r := range cm.clusterAwareRunnables {
+		if err := r.Disengage(ctx, cl); err != nil {
+			errs = append(errs, err)
+		}
+	}
+
+	cm.engagedClustersLock.Lock()
+	delete(cm.engagedClusters, cl.Name())
+	cm.engagedClustersLock.Unlock()
+
+	return kerrors.NewAggregate(errs)
+}
+
 func (cm *controllerManager) startLeaderElectionRunnables() error {
 	return cm.runnables.LeaderElection.Start(cm.internalCtx)
 }

pkg/manager/manager.go

@@ -53,6 +53,11 @@ type Manager interface {
 	// Cluster holds a variety of methods to interact with a cluster.
 	cluster.Cluster
 
+	// Aware is an interface for dynamic cluster addition and removal. The
+	// Manager will call Engage and Disengage on cluster-aware runnables like
+	// controllers to e.g. watch multiple clusters.
+	cluster.Aware
+
 	// Add will set requested dependencies on the component, and cause the component to be
 	// started when Start is called.
 	// Depending on if a Runnable implements LeaderElectionRunnable interface, a Runnable can be run in either
@@ -87,6 +92,10 @@ type Manager interface {
 	// lock was lost.
 	Start(ctx context.Context) error
 
+	// GetCluster retrieves a Cluster from a given identifying cluster name. An
+	// empty string will return the default cluster of the manager.
+	GetCluster(ctx context.Context, clusterName string) (cluster.Cluster, error)
+
 	// GetWebhookServer returns a webhook.Server
 	GetWebhookServer() webhook.Server
 
@@ -281,6 +290,11 @@ type Options struct {
 	newMetricsServer       func(options metricsserver.Options, config *rest.Config, httpClient *http.Client) (metricsserver.Server, error)
 	newHealthProbeListener func(addr string) (net.Listener, error)
 	newPprofListener       func(addr string) (net.Listener, error)
+
+	// ExperimentalClusterProvider is an EXPERIMENTAL feature that allows the manager to
+	// operate against many Kubernetes clusters at once. Individual clusters can
+	// be accessed by calling GetCluster on the Manager.
+	ExperimentalClusterProvider cluster.Provider
 }
 
 // BaseContextFunc is a function used to provide a base Context to Runnables
@@ -325,7 +339,7 @@ func New(config *rest.Config, options Options) (Manager, error) {
 	// Set default values for options fields
 	options = setOptionsDefaults(options)
 
-	cluster, err := cluster.New(config, func(clusterOptions *cluster.Options) {
+	clusterOptions := func(clusterOptions *cluster.Options) {
 		clusterOptions.Scheme = options.Scheme
 		clusterOptions.MapperProvider = options.MapperProvider
 		clusterOptions.Logger = options.Logger
@@ -334,7 +348,9 @@ func New(config *rest.Config, options Options) (Manager, error) {
 		clusterOptions.Cache = options.Cache
 		clusterOptions.Client = options.Client
 		clusterOptions.EventBroadcaster = options.EventBroadcaster //nolint:staticcheck
-	})
+	}
+
+	cl, err := cluster.New(config, clusterOptions)
 	if err != nil {
 		return nil, err
 	}
@@ -347,7 +363,7 @@ func New(config *rest.Config, options Options) (Manager, error) {
 	// Create the recorder provider to inject event recorders for the components.
 	// TODO(directxman12): the log for the event provider should have a context (name, tags, etc) specific
 	// to the particular controller that it's being injected into, rather than a generic one like is here.
-	recorderProvider, err := options.newRecorderProvider(config, cluster.GetHTTPClient(), cluster.GetScheme(), options.Logger.WithName("events"), options.makeBroadcaster)
+	recorderProvider, err := options.newRecorderProvider(config, cl.GetHTTPClient(), cl.GetScheme(), options.Logger.WithName("events"), options.makeBroadcaster)
 	if err != nil {
 		return nil, err
 	}
@@ -361,7 +377,7 @@ func New(config *rest.Config, options Options) (Manager, error) {
 		leaderRecorderProvider = recorderProvider
 	} else {
 		leaderConfig = rest.CopyConfig(options.LeaderElectionConfig)
-		scheme := cluster.GetScheme()
+		scheme := cl.GetScheme()
 		err := corev1.AddToScheme(scheme)
 		if err != nil {
 			return nil, err
@@ -397,7 +413,7 @@ func New(config *rest.Config, options Options) (Manager, error) {
 	}
 
 	// Create the metrics server.
-	metricsServer, err := options.newMetricsServer(options.Metrics, config, cluster.GetHTTPClient())
+	metricsServer, err := options.newMetricsServer(options.Metrics, config, cl.GetHTTPClient())
 	if err != nil {
 		return nil, err
 	}
@@ -417,11 +433,13 @@ func New(config *rest.Config, options Options) (Manager, error) {
 	}
 
 	errChan := make(chan error, 1)
-	runnables := newRunnables(options.BaseContext, errChan)
 	return &controllerManager{
 		stopProcedureEngaged:          ptr.To(int64(0)),
-		cluster:                       cluster,
-		runnables:                     runnables,
+		defaultCluster:                cl,
+		defaultClusterOptions:         clusterOptions,
+		clusterProvider:               options.ExperimentalClusterProvider,
+		engagedClusters:               make(map[string]cluster.Cluster),
+		runnables:                     newRunnables(options.BaseContext, errChan),
 		errChan:                       errChan,
 		recorderProvider:              recorderProvider,
 		resourceLock:                  resourceLock,
@@ -552,5 +570,10 @@ func setOptionsDefaults(options Options) Options {
 		options.WebhookServer = webhook.NewServer(webhook.Options{})
 	}
 
+	if options.Controller.EngageWithDefaultCluster == nil {
+		options.Controller.EngageWithDefaultCluster = ptr.To(options.ExperimentalClusterProvider == nil)
+		options.Controller.EngageWithProviderClusters = ptr.To(options.ExperimentalClusterProvider != nil)
+	}
+
 	return options
 }