✨ 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

@@ -3,27 +3,38 @@ module sigs.k8s.io/controller-runtime
 go 1.15
 
 require (
+	github.com/coreos/go-etcd v2.0.0+incompatible // indirect
+	github.com/cpuguy83/go-md2man v1.0.10 // indirect
+	github.com/docker/docker v0.7.3-0.20190327010347-be7ac8be2ae0 // indirect
 	github.com/evanphx/json-patch v4.9.0+incompatible
 	github.com/fsnotify/fsnotify v1.4.9
 	github.com/go-logr/logr v0.2.1
 	github.com/go-logr/zapr v0.2.0
 	github.com/google/go-cmp v0.5.2 // indirect
 	github.com/googleapis/gnostic v0.5.1 // indirect
+	github.com/gophercloud/gophercloud v0.1.0 // indirect
 	github.com/hashicorp/golang-lru v0.5.4 // indirect
 	github.com/imdario/mergo v0.3.10 // indirect
 	github.com/onsi/ginkgo v1.14.1
 	github.com/onsi/gomega v1.10.2
 	github.com/prometheus/client_golang v1.7.1
 	github.com/prometheus/client_model v0.2.0
+	github.com/robfig/cron v1.2.0
+	github.com/spf13/pflag v1.0.5
+	github.com/ugorji/go/codec v0.0.0-20181204163529-d75b2dcb6bc8 // indirect
 	go.uber.org/goleak v1.1.10
 	go.uber.org/zap v1.15.0
+	golang.org/x/text v0.3.3 // indirect
 	golang.org/x/time v0.0.0-20200630173020-3af7569d3a1e
 	gomodules.xyz/jsonpatch/v2 v2.1.0
 	google.golang.org/appengine v1.6.6 // indirect
+	gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15 // indirect
 	k8s.io/api v0.19.2
 	k8s.io/apiextensions-apiserver v0.19.2
 	k8s.io/apimachinery v0.19.2
 	k8s.io/client-go v0.19.2
+	k8s.io/klog v1.0.0 // indirect
 	k8s.io/utils v0.0.0-20200729134348-d5654de09c73
+	sigs.k8s.io/structured-merge-diff/v3 v3.0.0 // indirect
 	sigs.k8s.io/yaml v1.2.0
 )

go.sum

@@ -19,6 +19,7 @@ package builder
 import (
 	"fmt"
 	"strings"
+	"time"
 
 	"github.com/go-logr/logr"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -57,9 +58,11 @@ func ControllerManagedBy(m manager.Manager) *Builder {
 
 // ForInput represents the information set by For method.
 type ForInput struct {
-	object     runtime.Object
-	predicates []predicate.Predicate
-	err        error
+	object           runtime.Object
+	predicates       []predicate.Predicate
+	err              error
+	conditionallyRun bool
+	waitTime         time.Duration
 }
 
 // For defines the type of Object being *reconciled*, and configures the ControllerManagedBy to respond to create / delete /
@@ -252,6 +255,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/controller.go

@@ -17,9 +17,13 @@ limitations under the License.
 package builder
 
 import (
+	"time"
+
 	"sigs.k8s.io/controller-runtime/pkg/predicate"
 )
 
+const defaultWaitTime = time.Minute
+
 // {{{ "Functional" Option Interfaces
 
 // ForOption is some configuration that modifies options for a For request.
@@ -75,4 +79,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 = defaultWaitTime
+	} else {
+		opts.waitTime = w.WaitTime
+	}
+}
+
 // }}}

pkg/builder/options.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/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/informertest/fake_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/cache/multi_namespace_cache.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"
@@ -46,6 +48,15 @@ type Options struct {
 	// Log is the logger used for this controller and passed to each reconciliation
 	// request via the context field.
 	Log logr.Logger
+
+	// 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
@@ -123,5 +134,7 @@ func NewUnmanaged(name string, mgr manager.Manager, options Options) (Controller
 		SetFields:               mgr.SetFields,
 		Name:                    name,
 		Log:                     options.Log.WithName("controller").WithName(name),
+		ConditionalOn:           options.ConditionalOn,
+		ConditionalWaitTime:     options.ConditionalWaitTime,
 	}, nil
 }

pkg/controller/controller.go

@@ -24,6 +24,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"
@@ -46,6 +47,15 @@ type Controller struct {
 	// MaxConcurrentReconciles is the maximum number of concurrent Reconciles which can be run. Defaults to 1.
 	MaxConcurrentReconciles int
 
+	// 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.
@@ -172,9 +182,14 @@ func (c *Controller) Start(stop <-chan struct{}) error {
 
 		// All the watches have been started, we can reset the local slice.
 		//
-		// We should never hold watches more than necessary, each watch source can hold a backing cache,
+		// We should usually hold watches more than necessary, each watch source can hold a backing cache,
 		// which won't be garbage collected if we hold a reference to it.
-		c.startWatches = nil
+
+		// The exception to this is when the controller is configured as a conditional runnable, in which
+		// case it needs to knowledge of the watches in the event that the controller is restarted.
+		if c.ConditionalOn == nil {
+			c.startWatches = nil
+		}
 
 		if c.JitterPeriod == 0 {
 			c.JitterPeriod = 1 * time.Second
@@ -200,6 +215,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() {