Conditional Controller with SporadicRunnable

pkg/builder/controller.go

@@ -72,6 +72,7 @@ type ForInput struct {
 	predicates       []predicate.Predicate
 	objectProjection objectProjection
 	err              error
+	sporadic         bool
 }
 
 // For defines the type of Object being *reconciled*, and configures the ControllerManagedBy to respond to create / delete /
@@ -97,6 +98,7 @@ type OwnsInput struct {
 	object           client.Object
 	predicates       []predicate.Predicate
 	objectProjection objectProjection
+	sporadic         bool
 }
 
 // Owns defines types of Objects being *generated* by the ControllerManagedBy, and configures the ControllerManagedBy to respond to
@@ -118,6 +120,7 @@ type WatchesInput struct {
 	eventhandler     handler.EventHandler
 	predicates       []predicate.Predicate
 	objectProjection objectProjection
+	sporadic         bool
 }
 
 // Watches exposes the lower-level ControllerManagedBy Watches functions through the builder.  Consider using
@@ -222,7 +225,23 @@ func (blder *Builder) doWatch() error {
 	if err != nil {
 		return err
 	}
-	src := &source.Kind{Type: typeForSrc}
+	var src source.Source
+	if blder.forInput.sporadic {
+		gvk, err := getGvk(blder.forInput.object, blder.mgr.GetScheme())
+		if err != nil {
+			return err
+		}
+		existsInDiscovery := func() bool {
+			if _, err := blder.mgr.GetRESTMapper().RESTMapping(gvk.GroupKind(), gvk.Version); err != nil {
+				return false
+			}
+			return true
+		}
+
+		src = &source.SporadicKind{Kind: source.Kind{Type: typeForSrc}, DiscoveryCheck: existsInDiscovery}
+	} else {
+		src = &source.Kind{Type: typeForSrc}
+	}
 	hdler := &handler.EnqueueRequestForObject{}
 	allPredicates := append(blder.globalPredicates, blder.forInput.predicates...)
 	if err := blder.ctrl.Watch(src, hdler, allPredicates...); err != nil {

pkg/internal/controller/controller.go

@@ -81,7 +81,8 @@ type Controller struct {
 	CacheSyncTimeout time.Duration
 
 	// startWatches maintains a list of sources, handlers, and predicates to start when the controller is started.
-	startWatches []watchDescription
+	startWatches    []watchDescription
+	sporadicWatches []sporadicWatchDescription
 
 	// Log is used to log messages to users during reconciliation, or for example when a watch is started.
 	Log logr.Logger
@@ -94,6 +95,12 @@ type watchDescription struct {
 	predicates []predicate.Predicate
 }
 
+type sporadicWatchDescription struct {
+	src        source.SporadicSource
+	handler    handler.EventHandler
+	predicates []predicate.Predicate
+}
+
 // Reconcile implements reconcile.Reconciler
 func (c *Controller) Reconcile(ctx context.Context, req reconcile.Request) (reconcile.Result, error) {
 	log := c.Log.WithValues("name", req.Name, "namespace", req.Namespace)
@@ -119,6 +126,11 @@ func (c *Controller) Watch(src source.Source, evthdler handler.EventHandler, prc
 		}
 	}
 
+	if sporadicSource, ok := src.(source.SporadicSource); ok && !c.Started {
+		c.sporadicWatches = append(c.sporadicWatches, sporadicWatchDescription{src: sporadicSource, handler: evthdler, predicates: prct})
+		return nil
+	}
+
 	// Controller hasn't started yet, store the watches locally and return.
 	//
 	// These watches are going to be held on the controller struct until the manager or user calls Start(...).
@@ -131,6 +143,27 @@ func (c *Controller) Watch(src source.Source, evthdler handler.EventHandler, prc
 	return src.Start(c.ctx, evthdler, c.Queue, prct...)
 }
 
+// Rea
+func (c *Controller) ReadyToStart(ctx context.Context) <-chan struct{} {
+	ready := make(chan struct{})
+	if len(c.sporadicWatches) == 0 {
+		close(ready)
+		return ready
+	}
+
+	var wg sync.WaitGroup
+	for _, w := range c.sporadicWatches {
+		wg.Add(1)
+		go w.src.Ready(ctx, &wg)
+	}
+
+	go func() {
+		wg.Wait()
+		close(ready)
+	}()
+	return ready
+}
+
 // Start implements controller.Controller
 func (c *Controller) Start(ctx context.Context) error {
 	// use an IIFE to get proper lock handling

pkg/manager/internal.go

@@ -71,6 +71,8 @@ type controllerManager struct {
 	// These Runnables will not be blocked by lead election.
 	nonLeaderElectionRunnables []Runnable
 
+	sporadicRunnables []SporadicRunnable
+
 	// recorderProvider is used to generate event recorders that will be injected into Controllers
 	// (and EventHandlers, Sources and Predicates).
 	recorderProvider *intrec.Provider
@@ -207,6 +209,10 @@ func (cm *controllerManager) Add(r Runnable) error {
 		cm.nonLeaderElectionRunnables = append(cm.nonLeaderElectionRunnables, r)
 	} else if hasCache, ok := r.(hasCache); ok {
 		cm.caches = append(cm.caches, hasCache)
+
+	} else if sporadicRunnable, ok := r.(SporadicRunnable); ok {
+		cm.sporadicRunnables = append(cm.sporadicRunnables, sporadicRunnable)
+
 	} else {
 		shouldStart = cm.startedLeader
 		cm.leaderElectionRunnables = append(cm.leaderElectionRunnables, r)
@@ -583,6 +589,25 @@ func (cm *controllerManager) waitForRunnableToEnd(shutdownCancel context.CancelF
 	return nil
 }
 
+func (cm *controllerManager) startSporadicRunnables() {
+	cm.mu.Lock()
+	cm.waitForCache(cm.internalCtx)
+	cm.mu.Unlock()
+
+	for _, sr := range cm.sporadicRunnables {
+		go func(sr SporadicRunnable) {
+			for {
+				select {
+				case <-cm.internalCtx.Done():
+					return
+				case <-sr.Ready(cm.internalCtx):
+					cm.startRunnable(sr)
+				}
+			}
+		}(sr)
+	}
+}
+
 func (cm *controllerManager) startNonLeaderElectionRunnables() {
 	cm.mu.Lock()
 	defer cm.mu.Unlock()

pkg/manager/manager.go

@@ -290,6 +290,12 @@ func (r RunnableFunc) Start(ctx context.Context) error {
 	return r(ctx)
 }
 
+type SporadicRunnable interface {
+	Runnable
+	Ready(ctx context.Context) ReadySignal
+}
+type ReadySignal <-chan struct{}
+
 // LeaderElectionRunnable knows if a Runnable needs to be run in the leader election mode.
 type LeaderElectionRunnable interface {
 	// NeedLeaderElection returns true if the Runnable needs to be run in the leader election mode.

pkg/source/source.go

@@ -21,6 +21,7 @@ import (
 	"errors"
 	"fmt"
 	"sync"
+	"time"
 
 	"k8s.io/apimachinery/pkg/api/meta"
 	"k8s.io/client-go/util/workqueue"
@@ -57,6 +58,11 @@ type Source interface {
 	Start(context.Context, handler.EventHandler, workqueue.RateLimitingInterface, ...predicate.Predicate) error
 }
 
+type SporadicSource interface {
+	Source
+	Ready(ctx context.Context, wg *sync.WaitGroup) <-chan struct{}
+}
+
 // SyncingSource is a source that needs syncing prior to being usable. The controller
 // will call its WaitForSync prior to starting workers.
 type SyncingSource interface {
@@ -98,6 +104,34 @@ type Kind struct {
 	startCancel func()
 }
 
+type SporadicKind struct {
+	Kind
+	DiscoveryCheck func() bool
+}
+
+func (sk *SporadicKind) Ready(ctx context.Context, wg *sync.WaitGroup) <-chan struct{} {
+	defer wg.Done()
+	ready := make(chan struct{})
+	go func() {
+		for {
+			select {
+			case <-ctx.Done():
+				close(ready)
+				return
+			default:
+				if sk.DiscoveryCheck() {
+					close(ready)
+					return
+				}
+				//TODO: parameterize this
+				time.Sleep(5 * time.Second)
+			}
+		}
+	}()
+
+	return ready
+}
+
 var _ SyncingSource = &Kind{}
 
 // Start is internal and should be called only by the Controller to register an EventHandler with the Informer