Extend interface with runner

Author: mjudeikis

examples/cluster-api/main.go

@@ -129,7 +129,7 @@ func main() {
 		return ignoreCanceled(localMgr.Start(ctx))
 	})
 	g.Go(func() error {
-		return ignoreCanceled(provider.Run(ctx, mcMgr))
+		return ignoreCanceled(provider.Start(ctx, mcMgr))
 	})
 	g.Go(func() error {
 		return ignoreCanceled(mcMgr.Start(ctx))

examples/file/main.go

@@ -114,7 +114,7 @@ func main() {
 		return ignoreCanceled(mgr.Start(ctx))
 	})
 	g.Go(func() error {
-		return ignoreCanceled(provider.Run(ctx, mgr))
+		return ignoreCanceled(provider.Start(ctx, mgr))
 	})
 	if err := g.Wait(); err != nil {
 		entryLog.Info("error in errgroup: %w", err)

examples/kind/main.go

@@ -92,7 +92,7 @@ func main() {
 	// Starting everything.
 	g, ctx := errgroup.WithContext(ctx)
 	g.Go(func() error {
-		return ignoreCanceled(provider.Run(ctx, mgr))
+		return ignoreCanceled(provider.Start(ctx, mgr))
 	})
 	g.Go(func() error {
 		return ignoreCanceled(mgr.Start(ctx))

examples/namespace/main.go

@@ -154,7 +154,7 @@ func run(ctx context.Context, log logr.Logger, kubeconfig string) error {
 	// Starting everything.
 	g, ctx := errgroup.WithContext(ctx)
 	g.Go(func() error {
-		return ignoreCanceled(provider.Run(ctx, mgr))
+		return ignoreCanceled(provider.Start(ctx, mgr))
 	})
 	g.Go(func() error {
 		return ignoreCanceled(cl.Start(ctx))

pkg/multicluster/multicluster.go

@@ -60,4 +60,9 @@ type Provider interface {
 	// IndexField indexes the given object by the given field on all engaged
 	// clusters, current and future.
 	IndexField(ctx context.Context, obj client.Object, field string, extractValue client.IndexerFunc) error
+
+	// Start runs the provider. Implementation of this method should block.
+	// If you need to pass in manager, it is recommended to implement SetupWithManager(mgr mcmanager.Manager) error method on individual providers.
+	// Even if a provider gets a manager through e.g. `SetupWithManager` the `Aware` passed to this method must be used to engage clusters.
+	Start(context.Context, Aware) error
 }

providers/cluster-api/provider.go

@@ -39,7 +39,6 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/manager"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 
-	mcmanager "sigs.k8s.io/multicluster-runtime/pkg/manager"
 	"sigs.k8s.io/multicluster-runtime/pkg/multicluster"
 )
 
@@ -109,7 +108,7 @@ type Provider struct {
 	client client.Client
 
 	lock      sync.Mutex
-	mcMgr     mcmanager.Manager
+	mcAware   multicluster.Aware
 	clusters  map[string]cluster.Cluster
 	cancelFns map[string]context.CancelFunc
 	indexers  []index
@@ -126,13 +125,14 @@ func (p *Provider) Get(_ context.Context, clusterName string) (cluster.Cluster,
 	return nil, multicluster.ErrClusterNotFound
 }
 
-// Run starts the provider and blocks.
-func (p *Provider) Run(ctx context.Context, mgr mcmanager.Manager) error {
-	p.log.Info("Starting Cluster-API cluster provider")
-
+// Start starts the provider and blocks.
+func (p *Provider) Start(ctx context.Context, mcAware multicluster.Aware) error {
 	p.lock.Lock()
-	p.mcMgr = mgr
-	p.lock.Unlock()
+	defer p.lock.Unlock()
+
+	p.mcAware = mcAware
+
+	p.log.Info("Starting Cluster-API cluster provider")
 
 	<-ctx.Done()
 
@@ -171,7 +171,7 @@ func (p *Provider) Reconcile(ctx context.Context, req reconcile.Request) (reconc
 	// TODO(sttts): do tighter logging.
 
 	// provider already started?
-	if p.mcMgr == nil {
+	if p.mcAware == nil {
 		return reconcile.Result{RequeueAfter: time.Second * 2}, nil
 	}
 
@@ -222,7 +222,7 @@ func (p *Provider) Reconcile(ctx context.Context, req reconcile.Request) (reconc
 	p.log.Info("Added new cluster")
 
 	// engage manager.
-	if err := p.mcMgr.Engage(clusterCtx, key, cl); err != nil {
+	if err := p.mcAware.Engage(clusterCtx, key, cl); err != nil {
 		log.Error(err, "failed to engage manager")
 		delete(p.clusters, key)
 		delete(p.cancelFns, key)

providers/cluster-inventory-api/provider.go

@@ -78,7 +78,7 @@ type Provider struct {
 	strategy kubeconfigstrategy.Interface
 
 	lock       sync.RWMutex
-	mcMgr      mcmanager.Manager
+	manager    mcmanager.Manager
 	clusters   map[string]cluster.Cluster
 	cancelFns  map[string]context.CancelFunc
 	kubeconfig map[string]*rest.Config
@@ -125,7 +125,7 @@ func (p *Provider) SetupWithManager(mgr mcmanager.Manager) error {
 	if mgr == nil {
 		return fmt.Errorf("manager is nil")
 	}
-	p.mcMgr = mgr
+	p.manager = mgr
 
 	// Get the local manager from the multi-cluster manager.
 	localMgr := mgr.GetLocalManager()
@@ -199,7 +199,7 @@ func (p *Provider) Reconcile(ctx context.Context, req reconcile.Request) (reconc
 	defer p.lock.Unlock()
 
 	// provider already started?
-	if p.mcMgr == nil {
+	if p.manager == nil {
 		log.V(3).Info("Provider not started yet, requeuing")
 		return reconcile.Result{RequeueAfter: time.Second * 2}, nil
 	}
@@ -267,7 +267,7 @@ func (p *Provider) Reconcile(ctx context.Context, req reconcile.Request) (reconc
 	log.Info("Added new cluster for ClusterProfile")
 
 	// engage manager.
-	if err := p.mcMgr.Engage(clusterCtx, key, cl); err != nil {
+	if err := p.manager.Engage(clusterCtx, key, cl); err != nil {
 		log.Error(err, "failed to engage manager for ClusterProfile")
 		delete(p.clusters, key)
 		delete(p.cancelFns, key)
@@ -301,3 +301,9 @@ func (p *Provider) IndexField(ctx context.Context, obj client.Object, field stri
 
 	return nil
 }
+
+// Start runs the provider and blocks.
+func (p *Provider) Start(ctx context.Context, _ multicluster.Aware) error {
+	<-ctx.Done()
+	return ctx.Err()
+}

providers/file/provider.go

@@ -33,7 +33,6 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/cluster"
 	"sigs.k8s.io/controller-runtime/pkg/log"
 
-	mcmanager "sigs.k8s.io/multicluster-runtime/pkg/manager"
 	"sigs.k8s.io/multicluster-runtime/pkg/multicluster"
 )
 
@@ -137,14 +136,19 @@ type Provider struct {
 
 	log logr.Logger
 
-	clustersLock  sync.RWMutex
+	lock          sync.RWMutex
+	mcAware       multicluster.Aware
 	clusters      map[string]cluster.Cluster
 	clusterCancel map[string]func()
 }
 
-// Run starts the provider and updates the clusters and is blocking.
-func (p *Provider) Run(ctx context.Context, mgr mcmanager.Manager) error {
-	if err := p.run(ctx, mgr); err != nil {
+// Start starts the provider and updates the clusters and is blocking.
+func (p *Provider) Start(ctx context.Context, mcAware multicluster.Aware) error {
+	p.lock.Lock()
+	p.mcAware = mcAware
+	p.lock.Unlock()
+
+	if err := p.run(ctx); err != nil {
 		return fmt.Errorf("initial update failed: %w", err)
 	}
 
@@ -181,7 +185,7 @@ func (p *Provider) Run(ctx context.Context, mgr mcmanager.Manager) error {
 			// would also require to track which cluster belongs to
 			// which file.
 			// Instead clusters are just updated from all files.
-			if err := p.run(ctx, mgr); err != nil {
+			if err := p.run(ctx); err != nil {
 				p.log.Error(err, "failed to update clusters after file change")
 			}
 		case err, ok := <-watcher.Errors:
@@ -194,17 +198,17 @@ func (p *Provider) Run(ctx context.Context, mgr mcmanager.Manager) error {
 }
 
 // RunOnce performs a single update of the clusters.
-func (p *Provider) RunOnce(ctx context.Context, mgr mcmanager.Manager) error {
-	return p.run(ctx, mgr)
+func (p *Provider) RunOnce(ctx context.Context) error {
+	return p.run(ctx)
 }
 
-func (p *Provider) addCluster(ctx context.Context, mgr mcmanager.Manager, name string, cl cluster.Cluster) {
+func (p *Provider) addCluster(ctx context.Context, name string, cl cluster.Cluster) {
 	ctx, cancel := context.WithCancel(ctx)
 
-	p.clustersLock.Lock()
+	p.lock.Lock()
 	p.clusters[name] = cl
 	p.clusterCancel[name] = cancel
-	p.clustersLock.Unlock()
+	p.lock.Unlock()
 
 	go func() {
 		if err := cl.Start(ctx); err != nil {
@@ -213,17 +217,17 @@ func (p *Provider) addCluster(ctx context.Context, mgr mcmanager.Manager, name s
 		p.removeCluster(name)
 	}()
 
-	if mgr != nil {
-		if err := mgr.Engage(ctx, name, cl); err != nil {
+	if p.mcAware != nil {
+		if err := p.mcAware.Engage(ctx, name, cl); err != nil {
 			cancel()
 			p.log.Error(err, "failed to engage cluster", "name", name)
 		}
 	}
 }
 
 func (p *Provider) removeCluster(name string) {
-	p.clustersLock.Lock()
-	defer p.clustersLock.Unlock()
+	p.lock.Lock()
+	defer p.lock.Unlock()
 
 	if cancel, ok := p.clusterCancel[name]; ok {
 		cancel()
@@ -232,7 +236,7 @@ func (p *Provider) removeCluster(name string) {
 	}
 }
 
-func (p *Provider) run(ctx context.Context, mgr mcmanager.Manager) error {
+func (p *Provider) run(ctx context.Context) error {
 	loadedClusters, err := p.loadClusters()
 	if err != nil {
 		return fmt.Errorf("failed to load clusters: %w", err)
@@ -247,12 +251,12 @@ func (p *Provider) run(ctx context.Context, mgr mcmanager.Manager) error {
 			if !cmp.Equal(existingCluster.GetConfig(), cl.GetConfig()) {
 				p.log.Info("updating cluster", "name", name)
 				p.removeCluster(name)
-				p.addCluster(ctx, mgr, name, cl)
+				p.addCluster(ctx, name, cl)
 			}
 			continue
 		}
 		p.log.Info("adding cluster", "name", name)
-		p.addCluster(ctx, mgr, name, cl)
+		p.addCluster(ctx, name, cl)
 	}
 
 	// delete clusters that are no longer present
@@ -271,8 +275,8 @@ func (p *Provider) run(ctx context.Context, mgr mcmanager.Manager) error {
 // If the cluster name is empty (""), it returns the first cluster
 // found.
 func (p *Provider) Get(_ context.Context, clusterName string) (cluster.Cluster, error) {
-	p.clustersLock.RLock()
-	defer p.clustersLock.RUnlock()
+	p.lock.RLock()
+	defer p.lock.RUnlock()
 
 	if clusterName == "" {
 		for _, cl := range p.clusters {
@@ -289,8 +293,8 @@ func (p *Provider) Get(_ context.Context, clusterName string) (cluster.Cluster,
 
 // IndexField indexes a field on all clusters.
 func (p *Provider) IndexField(ctx context.Context, obj client.Object, field string, extractValue client.IndexerFunc) error {
-	p.clustersLock.RLock()
-	defer p.clustersLock.RUnlock()
+	p.lock.RLock()
+	defer p.lock.RUnlock()
 
 	for name, cl := range p.clusters {
 		if err := cl.GetCache().IndexField(ctx, obj, field, extractValue); err != nil {
@@ -302,7 +306,7 @@ func (p *Provider) IndexField(ctx context.Context, obj client.Object, field stri
 
 // ClusterNames returns the names of all clusters known to the provider.
 func (p *Provider) ClusterNames() []string {
-	p.clustersLock.RLock()
-	defer p.clustersLock.RUnlock()
+	p.lock.RLock()
+	defer p.lock.RUnlock()
 	return slices.Sorted(maps.Keys(p.clusters))
 }

providers/file/provider_test.go

@@ -54,11 +54,12 @@ var _ = Describe("Provider File", Ordered, func() {
 				KubeconfigDirs:  []string{discoverDir},
 			})
 			Expect(err).NotTo(HaveOccurred())
+			Expect(err).NotTo(HaveOccurred())
 		})
 
 		By("Starting the provider", func() {
 			g.Go(func() error {
-				return ignoreCanceled(provider.Run(ctx, nil))
+				return ignoreCanceled(provider.Start(ctx, nil)) // Pass nil for aware since we are not using a manager in this test
 			})
 		})
 

providers/kind/go.mod

@@ -16,12 +16,9 @@ require (
 require (
 	github.com/BurntSushi/toml v1.4.0 // indirect
 	github.com/alessio/shellescape v1.4.2 // indirect
-	github.com/beorn7/perks v1.0.1 // indirect
-	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
 	github.com/emicklei/go-restful/v3 v3.11.0 // indirect
 	github.com/evanphx/json-patch/v5 v5.9.11 // indirect
-	github.com/fsnotify/fsnotify v1.7.0 // indirect
 	github.com/fxamacker/cbor/v2 v2.7.0 // indirect
 	github.com/go-openapi/jsonpointer v0.21.0 // indirect
 	github.com/go-openapi/jsonreference v0.20.2 // indirect
@@ -41,10 +38,6 @@ require (
 	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
 	github.com/pelletier/go-toml v1.9.5 // indirect
 	github.com/pkg/errors v0.9.1 // indirect
-	github.com/prometheus/client_golang v1.22.0 // indirect
-	github.com/prometheus/client_model v0.6.1 // indirect
-	github.com/prometheus/common v0.62.0 // indirect
-	github.com/prometheus/procfs v0.15.1 // indirect
 	github.com/spf13/cobra v1.8.1 // indirect
 	github.com/spf13/pflag v1.0.5 // indirect
 	github.com/x448/float16 v0.8.4 // indirect
@@ -54,7 +47,6 @@ require (
 	golang.org/x/term v0.30.0 // indirect
 	golang.org/x/text v0.23.0 // indirect
 	golang.org/x/time v0.9.0 // indirect
-	gomodules.xyz/jsonpatch/v2 v2.4.0 // indirect
 	google.golang.org/protobuf v1.36.5 // indirect
 	gopkg.in/evanphx/json-patch.v4 v4.12.0 // indirect
 	gopkg.in/inf.v0 v0.9.1 // indirect