Add cluster.Provider

Signed-off-by: Vince Prignano <[email protected]>

Author: vincepri

examples/fleet/main.go

@@ -15,6 +15,7 @@ import (
 	"k8s.io/klog/v2"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/builder"
+	"sigs.k8s.io/controller-runtime/pkg/cluster"
 	"sigs.k8s.io/controller-runtime/pkg/envtest"
 	"sigs.k8s.io/controller-runtime/pkg/log"
 	"sigs.k8s.io/controller-runtime/pkg/manager"
@@ -124,14 +125,14 @@ func (k *KindAdapter) List() ([]logical.Name, error) {
 	return res, nil
 }
 
-func (k *KindAdapter) Watch() (logical.Watcher, error) {
-	return &KindWatcher{ch: make(chan logical.Event)}, nil
+func (k *KindAdapter) Watch() (cluster.Watcher, error) {
+	return &KindWatcher{ch: make(chan cluster.WatchEvent)}, nil
 }
 
 type KindWatcher struct {
 	init   sync.Once
 	wg     sync.WaitGroup
-	ch     chan logical.Event
+	ch     chan cluster.WatchEvent
 	cancel context.CancelFunc
 }
 
@@ -142,7 +143,7 @@ func (k *KindWatcher) Stop() {
 	k.wg.Wait()
 	close(k.ch)
 }
-func (k *KindWatcher) ResultChan() <-chan logical.Event {
+func (k *KindWatcher) ResultChan() <-chan cluster.WatchEvent {
 	k.init.Do(func() {
 		ctx, cancel := context.WithCancel(context.Background())
 		k.cancel = cancel
@@ -161,23 +162,23 @@ func (k *KindWatcher) ResultChan() <-chan logical.Event {
 					}
 					newSet := sets.New(list...)
 					// Check for new clusters.
-					for _, cluster := range newSet.Difference(set).UnsortedList() {
-						if !strings.HasPrefix(cluster, "fleet-") {
+					for _, cl := range newSet.Difference(set).UnsortedList() {
+						if !strings.HasPrefix(cl, "fleet-") {
 							continue
 						}
-						k.ch <- logical.Event{
+						k.ch <- cl.WatchEvent{
 							Type: watch.Added,
-							Name: logical.Name(cluster),
+							Name: logical.Name(cl),
 						}
 					}
 					// Check for deleted clusters.
-					for _, cluster := range set.Difference(newSet).UnsortedList() {
-						if !strings.HasPrefix(cluster, "fleet-") {
+					for _, cl := range set.Difference(newSet).UnsortedList() {
+						if !strings.HasPrefix(cl, "fleet-") {
 							continue
 						}
-						k.ch <- logical.Event{
+						k.ch <- cluster.WatchEvent{
 							Type: watch.Deleted,
-							Name: logical.Name(cluster),
+							Name: logical.Name(cl),
 						}
 					}
 					set = newSet

go.mod

@@ -24,7 +24,7 @@ require (
 	k8s.io/component-base v0.26.1
 	k8s.io/klog/v2 v2.90.0
 	k8s.io/utils v0.0.0-20221128185143-99ec85e7a448
-	sigs.k8s.io/logical-cluster v0.0.0-00010101000000-000000000000
+	sigs.k8s.io/logical-cluster v0.0.1-alpha.0
 	sigs.k8s.io/yaml v1.3.0
 )
 
@@ -74,5 +74,3 @@ require (
 	sigs.k8s.io/json v0.0.0-20220713155537-f223a00ba0e2 // indirect
 	sigs.k8s.io/structured-merge-diff/v4 v4.2.3 // indirect
 )
-
-replace sigs.k8s.io/logical-cluster => github.com/vincepri/logical-cluster v0.0.0-20230221175249-b5d99f705d5b

go.sum

@@ -268,8 +268,6 @@ github.com/stretchr/testify v1.5.1/go.mod h1:5W2xD1RspED5o8YsWQXVCued0rvSQ+mT+I5
 github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 github.com/stretchr/testify v1.8.0 h1:pSgiaMZlXftHpm5L7V1+rVB+AZJydKsMxsQBIJw4PKk=
-github.com/vincepri/logical-cluster v0.0.0-20230221175249-b5d99f705d5b h1:eGYbpQr7r86O5wqbYMgLIp+wUZ5DSJpICK6UKYXTwPk=
-github.com/vincepri/logical-cluster v0.0.0-20230221175249-b5d99f705d5b/go.mod h1:5keWHzDm2ppkz5PTL3upRzPeVeci8JLApFv1u+Q6uYE=
 github.com/yuin/goldmark v1.1.25/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
 github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
 github.com/yuin/goldmark v1.1.32/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
@@ -619,6 +617,8 @@ rsc.io/quote/v3 v3.1.0/go.mod h1:yEA65RcK8LyAZtP9Kv3t0HmxON59tX3rD+tICJqUlj0=
 rsc.io/sampler v1.3.0/go.mod h1:T1hPZKmBbMNahiBKFy5HrXp6adAjACjK9JXDnKaTXpA=
 sigs.k8s.io/json v0.0.0-20220713155537-f223a00ba0e2 h1:iXTIw73aPyC+oRdyqqvVJuloN1p0AC/kzH07hu3NE+k=
 sigs.k8s.io/json v0.0.0-20220713155537-f223a00ba0e2/go.mod h1:B8JuhiUyNFVKdsE8h686QcCxMaH6HrOAZj4vswFpcB0=
+sigs.k8s.io/logical-cluster v0.0.1-alpha.0 h1:vigMG0I1fgDVn0hsTOeZB55AmplXC7D4iLa60qeyX70=
+sigs.k8s.io/logical-cluster v0.0.1-alpha.0/go.mod h1:7YymTkuUFI+tkwCRPMsk+TiyBQiPDKRArxVAAGpezZI=
 sigs.k8s.io/structured-merge-diff/v4 v4.2.3 h1:PRbqxJClWWYMNV1dhaG4NsibJbArud9kFxnAMREiWFE=
 sigs.k8s.io/structured-merge-diff/v4 v4.2.3/go.mod h1:qjx8mGObPmV2aSZepjQjbmb2ihdVs8cGKBraizNC69E=
 sigs.k8s.io/yaml v1.3.0 h1:a2VclLzOGrwOHDiV8EfBGhvjHvP46CtW5j6POvhYGGo=

pkg/builder/controller_test.go

@@ -37,6 +37,7 @@ import (
 
 	"sigs.k8s.io/controller-runtime/pkg/cache"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/cluster"
 	"sigs.k8s.io/controller-runtime/pkg/config"
 	"sigs.k8s.io/controller-runtime/pkg/controller"
 	"sigs.k8s.io/controller-runtime/pkg/event"
@@ -560,14 +561,14 @@ var _ = Describe("application", func() {
 
 	Context("with logical adapter", func() {
 		It("should support watching across clusters", func() {
-			adapter := &fakeLogicalAdapter{
+			adapter := &fakeClusterProvider{
 				list: []logical.Name{
 					"cluster1",
 					"cluster2",
 				},
-				watch: make(chan logical.Event),
+				watch: make(chan cluster.WatchEvent),
 			}
-			mgr, err := manager.New(cfg, manager.Options{}.WithExperimentalLogicalAdapter(adapter))
+			mgr, err := manager.New(cfg, manager.Options{}.WithExperimentalClusterProvider(adapter))
 			Expect(err).NotTo(HaveOccurred())
 
 			ctx, cancel := context.WithCancel(context.Background())
@@ -836,32 +837,32 @@ type fakeType struct {
 func (*fakeType) GetObjectKind() schema.ObjectKind { return nil }
 func (*fakeType) DeepCopyObject() runtime.Object   { return nil }
 
-type fakeLogicalAdapter struct {
+type fakeClusterProvider struct {
 	list    []logical.Name
 	listErr error
 
-	watch chan logical.Event
+	watch chan cluster.WatchEvent
 }
 
-func (f *fakeLogicalAdapter) RESTConfig(name logical.Name) (*rest.Config, error) {
-	return testenv.Config, nil
+func (f *fakeClusterProvider) Get(ctx context.Context, name logical.Name, opts ...cluster.Option) (cluster.Cluster, error) {
+	return cluster.New(testenv.Config, opts...)
 }
 
-func (f *fakeLogicalAdapter) List() ([]logical.Name, error) {
+func (f *fakeClusterProvider) List() ([]logical.Name, error) {
 	return f.list, f.listErr
 }
 
-func (f *fakeLogicalAdapter) Watch() (logical.Watcher, error) {
+func (f *fakeClusterProvider) Watch() (cluster.Watcher, error) {
 	return &fakeLogicalWatcher{ch: f.watch}, nil
 }
 
 type fakeLogicalWatcher struct {
-	ch chan logical.Event
+	ch chan cluster.WatchEvent
 }
 
 func (f *fakeLogicalWatcher) Stop() {
 }
 
-func (f *fakeLogicalWatcher) ResultChan() <-chan logical.Event {
+func (f *fakeLogicalWatcher) ResultChan() <-chan cluster.WatchEvent {
 	return f.ch
 }

pkg/cluster/provider.go

@@ -0,0 +1,59 @@
+package cluster
+
+import (
+	"context"
+
+	"k8s.io/apimachinery/pkg/watch"
+	"sigs.k8s.io/logical-cluster"
+)
+
+// Provider defines methods to retrieve, list, and watch fleet of clusters.
+// The provider is responsible for discovering and managing the lifecycle of each
+// cluster.
+//
+// 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(ctx context.Context, name logical.Name, opts ...Option) (Cluster, error)
+
+	// List returns a list of logical clusters.
+	// This method is used to discover the initial set of logical clusters
+	// and to refresh the list of logical clusters periodically.
+	List() ([]logical.Name, error)
+
+	// Watch returns a Watcher that watches for changes to a list of logical clusters
+	// and react to potential changes.
+	Watch() (Watcher, error)
+}
+
+// Watcher watches for changes to clusters and provides events to a channel
+// for the Manager to react to.
+type Watcher interface {
+	// Stop stops watching. Will close the channel returned by ResultChan(). Releases
+	// any resources used by the watch.
+	Stop()
+
+	// ResultChan returns a chan which will receive all the events. If an error occurs
+	// or Stop() is called, the implementation will close this channel and
+	// release any resources used by the watch.
+	ResultChan() <-chan WatchEvent
+}
+
+// WatchEvent is an event that is sent when a cluster is added, modified, or deleted.
+type WatchEvent struct {
+	// Type is the type of event that occurred.
+	//
+	// - ADDED or MODIFIED
+	//	 	The logical cluster was added or updated: a new RESTConfig is available, or needs to be refreshed.
+	// - DELETED
+	// 		The logical cluster was deleted: the cluster is removed.
+	// - ERROR
+	// 		An error occurred while watching the logical cluster: the cluster is removed.
+	// - BOOKMARK
+	// 		A periodic event is sent that contains no new data: ignored.
+	Type watch.EventType
+
+	// Name is the name of the logical cluster related to the event.
+	Name logical.Name
+}

pkg/manager/internal.go

@@ -85,7 +85,7 @@ type controllerManager struct {
 	defaultCluster        cluster.Cluster
 	defaultClusterOptions cluster.Option
 
-	logicalAdapter logical.Adapter
+	logicalProvider cluster.Provider
 
 	logicalLock                  sync.RWMutex // protects logicalClusters
 	logicalClusters              map[logical.Name]*logicalCluster
@@ -307,7 +307,7 @@ func (cm *controllerManager) engageClusterAwareRunnables() {
 	defer cm.Unlock()
 
 	// If we don't have a logical adapter, we cannot sync the runnables.
-	if cm.logicalAdapter == nil {
+	if cm.logicalProvider == nil {
 		return
 	}
 
@@ -326,7 +326,7 @@ func (cm *controllerManager) engageClusterAwareRunnables() {
 func (cm *controllerManager) getLogicalCluster(ctx context.Context, name logical.Name) (c *logicalCluster, err error) {
 	// Check if the manager was configured with a logical adapter,
 	// otherwise we cannot retrieve the cluster.
-	if cm.logicalAdapter == nil {
+	if cm.logicalProvider == nil {
 		return nil, fmt.Errorf("manager was not configured with a logical adapter, cannot retrieve %q", name)
 	}
 
@@ -349,28 +349,23 @@ func (cm *controllerManager) getLogicalCluster(ctx context.Context, name logical
 	}
 
 	// Create a new cluster.
-	var cfg *rest.Config
+	var cl cluster.Cluster
 	{
 		// TODO(vincepri): Make this timeout configurable.
 		ctx, cancel := context.WithTimeout(ctx, 10*time.Second)
 		defer cancel()
 		var watchErr error
 		if err := wait.PollImmediateUntilWithContext(ctx, 1*time.Second, func(ctx context.Context) (done bool, err error) {
-			cfg, watchErr = cm.logicalAdapter.RESTConfig(name)
+			cl, watchErr = cm.logicalProvider.Get(ctx, name, cm.defaultClusterOptions, cluster.WithName(name))
 			if watchErr != nil {
 				return false, nil //nolint:nilerr // We want to keep trying.
 			}
 			return true, nil
 		}); err != nil {
-			return nil, fmt.Errorf("failed to retrieve RESTConfig: %w", kerrors.NewAggregate([]error{err, watchErr}))
+			return nil, fmt.Errorf("failed  create logical cluster %q: %w", name, kerrors.NewAggregate([]error{err, watchErr}))
 		}
 	}
 
-	cl, err := cluster.New(cfg, cm.defaultClusterOptions, cluster.WithName(name))
-	if err != nil {
-		return nil, fmt.Errorf("cannot create logical cluster %q: %w", name, err)
-	}
-
 	// We add the Cluster to the manager as a Runnable, which is going to be categorized
 	// as a Cache-backed Runnable.
 	//
@@ -380,7 +375,7 @@ func (cm *controllerManager) getLogicalCluster(ctx context.Context, name logical
 		return nil, fmt.Errorf("cannot add logical cluster %q to manager: %w", name, err)
 	}
 	// Create a new context for the Cluster, so that it can be stopped independently.
-	ctx, cancel := context.WithCancel(logical.IntoContext(context.Background(), name))
+	ctx, cancel := context.WithCancel(context.Background())
 	c = &logicalCluster{
 		Cluster: cl,
 		ctx:     ctx,
@@ -393,7 +388,7 @@ func (cm *controllerManager) getLogicalCluster(ctx context.Context, name logical
 func (cm *controllerManager) removeLogicalCluster(name logical.Name) error {
 	// Check if the manager was configured with a logical adapter,
 	// otherwise we cannot retrieve the cluster.
-	if cm.logicalAdapter == nil {
+	if cm.logicalProvider == nil {
 		return fmt.Errorf("manager was not configured with a logical adapter, cannot retrieve %q", name)
 	}
 
@@ -613,10 +608,10 @@ func (cm *controllerManager) Start(ctx context.Context) (err error) { //nolint:g
 	}
 
 	// If the manager has been configured with a logical adapter, start it.
-	if cm.logicalAdapter != nil {
+	if cm.logicalProvider != nil {
 		if err := cm.add(RunnableFunc(func(ctx context.Context) error {
 			resync := func() error {
-				clusterList, err := cm.logicalAdapter.List()
+				clusterList, err := cm.logicalProvider.List()
 				if err != nil {
 					return err
 				}
@@ -644,7 +639,7 @@ func (cm *controllerManager) Start(ctx context.Context) (err error) { //nolint:g
 			}
 
 			// Create a watcher and start watching for changes.
-			watcher, err := cm.logicalAdapter.Watch()
+			watcher, err := cm.logicalProvider.Watch()
 			if err != nil {
 				return err
 			}

pkg/manager/manager.go

@@ -351,10 +351,10 @@ type Options struct {
 	newMetricsListener     func(addr string) (net.Listener, error)
 	newHealthProbeListener func(addr string) (net.Listener, error)
 
-	// logicalAdapter is an EXPERIMENTAL feature that allows the manager to
+	// logicalClusterProvider is an EXPERIMENTAL feature that allows the manager to
 	// operate against many Kubernetes clusters at once.
 	// It can be used by invoking WithExperimentalLogicalAdapter(adapter) on Options.
-	logicalAdapter logical.Adapter
+	logicalClusterProvider cluster.Provider
 }
 
 // BaseContextFunc is a function used to provide a base Context to Runnables
@@ -473,7 +473,7 @@ func New(config *rest.Config, options Options) (Manager, error) {
 		stopProcedureEngaged:          pointer.Int64(0),
 		defaultCluster:                cluster,
 		defaultClusterOptions:         clusterOptions,
-		logicalAdapter:                options.logicalAdapter,
+		logicalProvider:               options.logicalClusterProvider,
 		logicalClusters:               make(map[logical.Name]*logicalCluster),
 		runnables:                     newRunnables(options.BaseContext, errChan),
 		errChan:                       errChan,
@@ -587,13 +587,13 @@ func (o Options) AndFromOrDie(loader config.ControllerManagerConfiguration) Opti
 	return o
 }
 
-// WithExperimentalLogicalAdapter sets the logical adapter to use for the manager.
+// WithExperimentalClusterProvider sets the logical adapter to use for the manager.
 // This is an EXPERIMENTAL feature that allows a Manager to be used against a fleet
 // of Kubernetes clusters.
 //
 // NOTE: The method signature may change or be removed in the future.
-func (o Options) WithExperimentalLogicalAdapter(adapter logical.Adapter) Options {
-	o.logicalAdapter = adapter
+func (o Options) WithExperimentalClusterProvider(provider cluster.Provider) Options {
+	o.logicalClusterProvider = provider
 	return o
 }