✨ Conditionally runnable controllers

Implement conditionally runnable controllers that wait and watch for
their respective resource to exist in the discovery doc before running
the controller.

This enables the controller manager to:
* Start without error when a CRD the controller watches does not exist.
* Begin watching the CRD and start its controller once it isinstalled.
* Stop the controller (and the respective informer) upon CRD uninstall.
* Restart the controller upon reinstallling the CRD.

Author: kevindelgado

go.mod

@@ -18,6 +18,7 @@ require (
 	github.com/prometheus/client_golang v1.0.0
 	github.com/prometheus/client_model v0.2.0
 	github.com/prometheus/procfs v0.0.11 // indirect
+	github.com/robfig/cron v1.2.0
 	github.com/spf13/pflag v1.0.5
 	go.uber.org/atomic v1.4.0 // indirect
 	go.uber.org/zap v1.10.0

go.sum

@@ -298,6 +298,8 @@ github.com/prometheus/procfs v0.0.2 h1:6LJUbpNm42llc4HRCuvApCSWB/WfhuNo9K98Q9sNG
 github.com/prometheus/procfs v0.0.2/go.mod h1:TjEm7ze935MbeOT/UhFTIMYKhuLP4wbCsTZCD3I8kEA=
 github.com/prometheus/procfs v0.0.11 h1:DhHlBtkHWPYi8O2y31JkK0TF+DGM+51OopZjH/Ia5qI=
 github.com/prometheus/procfs v0.0.11/go.mod h1:lV6e/gmhEcM9IjHGsFOCxxuZ+z1YqCvr4OA4YeYWdaU=
+github.com/robfig/cron v1.2.0 h1:ZjScXvvxeQ63Dbyxy76Fj3AT3Ut0aKsyd2/tl3DTMuQ=
+github.com/robfig/cron v1.2.0/go.mod h1:JGuDeoQd7Z6yL4zQhZ3OPEVHB7fL6Ka6skscFHfmt2k=
 github.com/rogpeppe/fastuuid v0.0.0-20150106093220-6724a57986af/go.mod h1:XWv6SoW27p1b0cqNHllgS5HIMJraePCO15w5zCzIWYg=
 github.com/russross/blackfriday v1.5.2 h1:HyvC0ARfnZBqnXwABFeSZHpKvJHJJfPz81GNueLj0oo=
 github.com/russross/blackfriday v1.5.2/go.mod h1:JO/DiYxRf+HjHt06OyowR9PTA263kcR/rfWxYHBV53g=

pkg/builder/controller.go

@@ -19,6 +19,7 @@ package builder
 import (
 	"fmt"
 	"strings"
+	"time"
 
 	"github.com/go-logr/logr"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -68,8 +69,10 @@ func (blder *Builder) ForType(apiType runtime.Object) *Builder {
 
 // ForInput represents the information set by For method.
 type ForInput struct {
-	object     runtime.Object
-	predicates []predicate.Predicate
+	object           runtime.Object
+	predicates       []predicate.Predicate
+	conditionallyRun bool
+	waitTime         time.Duration
 }
 
 // For defines the type of Object being *reconciled*, and configures the ControllerManagedBy to respond to create / delete /
@@ -263,6 +266,11 @@ func (blder *Builder) doController(r reconcile.Reconciler) error {
 	}
 	ctrlOptions.Log = ctrlOptions.Log.WithValues("reconcilerGroup", gvk.Group, "reconcilerKind", gvk.Kind)
 
+	if blder.forInput.conditionallyRun {
+		ctrlOptions.ConditionalOn = &blder.forInput.object
+		ctrlOptions.ConditionalWaitTime = blder.forInput.waitTime
+	}
+
 	// Build the controller and return.
 	blder.ctrl, err = newController(blder.getControllerName(gvk), blder.mgr, ctrlOptions)
 	return err

pkg/builder/options.go

@@ -17,6 +17,8 @@ limitations under the License.
 package builder
 
 import (
+	"time"
+
 	"sigs.k8s.io/controller-runtime/pkg/predicate"
 )
 
@@ -75,4 +77,21 @@ var _ ForOption = &Predicates{}
 var _ OwnsOption = &Predicates{}
 var _ WatchesOption = &Predicates{}
 
+// ConditionallyRun runs the controller
+// condtionally on the existence of the forInput object
+// in the cluster's discovery doc, letting you start a
+// controller manager for a CRD not yet installed on the cluster.
+type ConditionallyRun struct {
+	waitTime time.Duration
+}
+
+func (w ConditionallyRun) ApplyToFor(opts *ForInput) {
+	opts.conditionallyRun = true
+	if w.waitTime == time.Duration(0) {
+		opts.waitTime = 100 * time.Microsecond
+	} else {
+		opts.waitTime = w.waitTime
+	}
+}
+
 // }}}

pkg/cache/cache.go

@@ -58,6 +58,9 @@ type Informers interface {
 	// of the underlying object.
 	GetInformerForKind(ctx context.Context, gvk schema.GroupVersionKind) (Informer, error)
 
+	// Remove the informer for the given object.
+	Remove(obj runtime.Object) error
+
 	// Start runs all the informers known to this cache until the given channel is closed.
 	// It blocks.
 	Start(stopCh <-chan struct{}) error

pkg/cache/informertest/fake_cache.go

@@ -125,6 +125,11 @@ func (c *FakeInformers) Start(stopCh <-chan struct{}) error {
 	return c.Error
 }
 
+// Remove implements Cache
+func (c *FakeInformers) Remove(obj runtime.Object) error {
+	return c.Error
+}
+
 // IndexField implements Cache
 func (c *FakeInformers) IndexField(ctx context.Context, obj runtime.Object, field string, extractValue client.IndexerFunc) error {
 	return nil

pkg/cache/multi_namespace_cache.go

@@ -94,6 +94,15 @@ func (c *multiNamespaceCache) GetInformerForKind(ctx context.Context, gvk schema
 	return &multiNamespaceInformer{namespaceToInformer: informers}, nil
 }
 
+func (c *multiNamespaceCache) Remove(obj runtime.Object) error {
+	for _, cache := range c.namespaceToCache {
+		if err := cache.Remove(obj); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
 func (c *multiNamespaceCache) Start(stopCh <-chan struct{}) error {
 	for ns, cache := range c.namespaceToCache {
 		go func(ns string, cache Cache) {

pkg/controller/controller.go

@@ -18,8 +18,10 @@ package controller
 
 import (
 	"fmt"
+	"time"
 
 	"github.com/go-logr/logr"
+	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/client-go/util/workqueue"
 	"sigs.k8s.io/controller-runtime/pkg/handler"
 	"sigs.k8s.io/controller-runtime/pkg/internal/controller"
@@ -45,6 +47,20 @@ type Options struct {
 
 	// Log is the logger used for this controller.
 	Log logr.Logger
+
+	// Conditionally is a flag set during controller setup that indicates it should conditionally wait on the
+	// ForObject to appear in the discovery doc before starting the controller. If set,
+	// it will set ConditionalObject to the ForObject being reconciled for in the controller builder.
+	//Conditionally bool
+
+	// ConditionalObject is the Object that the manager should wait on to appear
+	// in the discovery document to indicate to begin running the controller
+	// (and should stop the informer for this object if it disappears from the discovery doc).
+	ConditionalOn *runtime.Object
+
+	// ConditionalWaitTime is the frequency at which the controller manager should check
+	// the discovery doc for the existence of the CondtionalOn object.
+	ConditionalWaitTime time.Duration
 }
 
 // Controller implements a Kubernetes API.  A Controller manages a work queue fed reconcile.Requests
@@ -117,6 +133,8 @@ func NewUnmanaged(name string, mgr manager.Manager, options Options) (Controller
 		MaxConcurrentReconciles: options.MaxConcurrentReconciles,
 		SetFields:               mgr.SetFields,
 		Name:                    name,
+		ConditionalOn:           options.ConditionalOn,
+		ConditionalWaitTime:     options.ConditionalWaitTime,
 		Log:                     options.Log.WithName("controller").WithValues("controller", name),
 	}, nil
 }

pkg/internal/controller/controller.go

@@ -22,6 +22,7 @@ import (
 	"time"
 
 	"github.com/go-logr/logr"
+	"k8s.io/apimachinery/pkg/runtime"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/util/workqueue"
@@ -43,6 +44,20 @@ type Controller struct {
 	// MaxConcurrentReconciles is the maximum number of concurrent Reconciles which can be run. Defaults to 1.
 	MaxConcurrentReconciles int
 
+	// Conditionally is a flag set during controller setup that indicates it should conditionally wait on the
+	// ForObject to appear in the discovery doc before starting the controller. If set,
+	// it will set ConditionalObject to the ForObject being reconciled for in the controller builder.
+	// Conditionally bool
+
+	// ConditionalOn is the Object that the manager should wait on to appear
+	// in the discovery document to indicate to begin running the controller
+	// (and should stop the informer for this object if it disappears from the discovery doc).
+	ConditionalOn *runtime.Object
+
+	// ConditionalWaitTime is the frequency at which the controller manager should check
+	// the discovery doc for the existence of the CondtionalOn object.
+	ConditionalWaitTime time.Duration
+
 	// Reconciler is a function that can be called at any time with the Name / Namespace of an object and
 	// ensures that the state of the system matches the state specified in the object.
 	// Defaults to the DefaultReconcileFunc.
@@ -182,6 +197,20 @@ func (c *Controller) Start(stop <-chan struct{}) error {
 	return nil
 }
 
+// GetConditionalOn returns the object that startCondtionalRunnables used
+// to determine whether a condtional runnable should be started/stopped (based
+// on the object's existence in the cluster's discovery doc).
+func (c *Controller) GetConditionalOn() *runtime.Object {
+	return c.ConditionalOn
+}
+
+// GetConditionalWaitTime returns the duration for which the
+// controller manager should wait on each iteration when checking the
+// discovery doc for the ConditionalOn object's existence.
+func (c *Controller) GetConditionalWaitTime() time.Duration {
+	return c.ConditionalWaitTime
+}
+
 // worker runs a worker thread that just dequeues items, processes them, and marks them done.
 // It enforces that the reconcileHandler is never invoked concurrently with the same object.
 func (c *Controller) worker() {