Support metadata-only watches

Add support for metadata-only watches. This backports a series of
commits from the main branch:

Add low-level metadata-only informer support

This adds support for informers that communicate with the API server in
metadata-only form.  They are *completely* separate from normal
informers -- that is: just like unstructured, if you ask for both a
"normal" informer & a metadata-only informer, you'll get two copies of
the cache.

Support metadata-only client operations

This adds support for a metadata-only client.  It only implements
the operations supported by metadata (delete, deleteallof, patch,
get, list, status.patch).  The higher-level client will now delegate
to this for when a PartialObjectMetadata object is passed in.

Support "projections" in the controller builder

This adds options to "project" watches as only metadata to the builder,
making it more convienient to use these forms.  For instance:

```go
.Owns(&corev1.Pod{}, builder.OnlyMetadata)
```

is equivalent to

```go
.Owns(&metav1.PartialObjectMetadata{
        TypeMeta: metav1.TypeMeta{
                APIVersion: "v1",
                Kind: "Pod",
        },
})
```

Co-authored-by: Solly Ross <[email protected]>

Author: ncdc

pkg/builder/controller.go

@@ -21,6 +21,7 @@ import (
 	"strings"
 
 	"github.com/go-logr/logr"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/client-go/rest"
@@ -37,6 +38,17 @@ import (
 var newController = controller.New
 var getGvk = apiutil.GVKForObject
 
+// project represents other forms that the we can use to
+// send/receive a given resource (metadata-only, unstructured, etc)
+type objectProjection int
+
+const (
+	// projectAsNormal doesn't change the object from the form given
+	projectAsNormal objectProjection = iota
+	// projectAsMetadata turns this into an metadata-only watch
+	projectAsMetadata
+)
+
 // Builder builds a Controller.
 type Builder struct {
 	forInput         ForInput
@@ -68,8 +80,9 @@ 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
+	objectProjection objectProjection
 }
 
 // For defines the type of Object being *reconciled*, and configures the ControllerManagedBy to respond to create / delete /
@@ -88,8 +101,9 @@ func (blder *Builder) For(object runtime.Object, opts ...ForOption) *Builder {
 
 // OwnsInput represents the information set by Owns method.
 type OwnsInput struct {
-	object     runtime.Object
-	predicates []predicate.Predicate
+	object           runtime.Object
+	predicates       []predicate.Predicate
+	objectProjection objectProjection
 }
 
 // Owns defines types of Objects being *generated* by the ControllerManagedBy, and configures the ControllerManagedBy to respond to
@@ -195,19 +209,43 @@ func (blder *Builder) Build(r reconcile.Reconciler) (controller.Controller, erro
 	return blder.ctrl, nil
 }
 
+func (blder *Builder) project(obj runtime.Object, proj objectProjection) (runtime.Object, error) {
+	switch proj {
+	case projectAsNormal:
+		return obj, nil
+	case projectAsMetadata:
+		metaObj := &metav1.PartialObjectMetadata{}
+		gvk, err := getGvk(obj, blder.mgr.GetScheme())
+		if err != nil {
+			return nil, fmt.Errorf("unable to determine GVK of %T for a metadata-only watch: %w", obj, err)
+		}
+		metaObj.SetGroupVersionKind(gvk)
+		return metaObj, nil
+	default:
+		panic(fmt.Sprintf("unexpected projection type %v on type %T, should not be possible since this is an internal field", proj, obj))
+	}
+}
+
 func (blder *Builder) doWatch() error {
 	// Reconcile type
-	src := &source.Kind{Type: blder.forInput.object}
+	typeForSrc, err := blder.project(blder.forInput.object, blder.forInput.objectProjection)
+	if err != nil {
+		return err
+	}
+	src := &source.Kind{Type: typeForSrc}
 	hdler := &handler.EnqueueRequestForObject{}
 	allPredicates := append(blder.globalPredicates, blder.forInput.predicates...)
-	err := blder.ctrl.Watch(src, hdler, allPredicates...)
-	if err != nil {
+	if err := blder.ctrl.Watch(src, hdler, allPredicates...); err != nil {
 		return err
 	}
 
 	// Watches the managed types
 	for _, own := range blder.ownsInput {
-		src := &source.Kind{Type: own.object}
+		typeForSrc, err := blder.project(own.object, own.objectProjection)
+		if err != nil {
+			return err
+		}
+		src := &source.Kind{Type: typeForSrc}
 		hdler := &handler.EnqueueRequestForOwner{
 			OwnerType:    blder.forInput.object,
 			IsController: true,

pkg/builder/controller_test.go

@@ -24,6 +24,8 @@ import (
 
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
+	"k8s.io/client-go/rest"
+	"sigs.k8s.io/controller-runtime/pkg/cache"
 
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
@@ -294,8 +296,58 @@ var _ = Describe("application", func() {
 		})
 	})
 
+	Describe("watching with projections", func() {
+		var mgr manager.Manager
+		BeforeEach(func() {
+			// use a cache that intercepts requests for fully typed objects to
+			// ensure we use the projected versions
+			var err error
+			mgr, err = manager.New(cfg, manager.Options{NewCache: newNonTypedOnlyCache})
+			Expect(err).NotTo(HaveOccurred())
+		})
+
+		It("should support watching For & Owns as metadata", func() {
+			bldr := ControllerManagedBy(mgr).
+				For(&appsv1.Deployment{}, OnlyMetadata).
+				Owns(&appsv1.ReplicaSet{}, OnlyMetadata)
+
+			doReconcileTest("8", stop, bldr, mgr, true)
+		})
+	})
 })
 
+// newNonTypedOnlyCache returns a new cache that wraps the normal cache,
+// returning an error if normal, typed objects have informers requested.
+func newNonTypedOnlyCache(config *rest.Config, opts cache.Options) (cache.Cache, error) {
+	normalCache, err := cache.New(config, opts)
+	if err != nil {
+		return nil, err
+	}
+	return &nonTypedOnlyCache{
+		Cache: normalCache,
+	}, nil
+}
+
+// nonTypedOnlyCache is a cache.Cache that only provides metadata &
+// unstructured informers.
+type nonTypedOnlyCache struct {
+	cache.Cache
+}
+
+func (c *nonTypedOnlyCache) GetInformer(ctx context.Context, obj runtime.Object) (cache.Informer, error) {
+	switch obj.(type) {
+	case (*metav1.PartialObjectMetadata):
+		return c.Cache.GetInformer(ctx, obj)
+	default:
+		return nil, fmt.Errorf("did not want to provide an informer for normal type %T", obj)
+	}
+}
+func (c *nonTypedOnlyCache) GetInformerForKind(ctx context.Context, gvk schema.GroupVersionKind) (cache.Informer, error) {
+	return nil, fmt.Errorf("don't try to sidestep the restriction on informer types by calling GetInformerForKind")
+}
+
+// TODO(directxman12): this function has too many arguments, and the whole
+// "nameSuffix" think is a bit of a hack It should be cleaned up significantly by someone with a bit of time
 func doReconcileTest(nameSuffix string, stop chan struct{}, blder *Builder, mgr manager.Manager, complete bool) {
 	deployName := "deploy-name-" + nameSuffix
 	rsName := "rs-name-" + nameSuffix
@@ -358,8 +410,8 @@ func doReconcileTest(nameSuffix string, stop chan struct{}, blder *Builder, mgr
 	Expect(err).NotTo(HaveOccurred())
 
 	By("Waiting for the Deployment Reconcile")
-	Expect(<-ch).To(Equal(reconcile.Request{
-		NamespacedName: types.NamespacedName{Namespace: "default", Name: deployName}}))
+	Eventually(ch).Should(Receive(Equal(reconcile.Request{
+		NamespacedName: types.NamespacedName{Namespace: "default", Name: deployName}})))
 
 	By("Creating a ReplicaSet")
 	// Expect a Reconcile when an Owned object is managedObjects.
@@ -388,8 +440,8 @@ func doReconcileTest(nameSuffix string, stop chan struct{}, blder *Builder, mgr
 	Expect(err).NotTo(HaveOccurred())
 
 	By("Waiting for the ReplicaSet Reconcile")
-	Expect(<-ch).To(Equal(reconcile.Request{
-		NamespacedName: types.NamespacedName{Namespace: "default", Name: deployName}}))
+	Eventually(ch).Should(Receive(Equal(reconcile.Request{
+		NamespacedName: types.NamespacedName{Namespace: "default", Name: deployName}})))
 
 }
 

pkg/builder/example_test.go

@@ -21,6 +21,7 @@ import (
 	"fmt"
 	"os"
 
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	logf "sigs.k8s.io/controller-runtime/pkg/log"
 
 	appsv1 "k8s.io/api/apps/v1"
@@ -34,6 +35,60 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 )
 
+func ExampleBuilder_metadata_only() {
+	logf.SetLogger(zap.New())
+
+	var log = logf.Log.WithName("builder-examples")
+
+	mgr, err := manager.New(config.GetConfigOrDie(), manager.Options{})
+	if err != nil {
+		log.Error(err, "could not create manager")
+		os.Exit(1)
+	}
+
+	cl := mgr.GetClient()
+	err = builder.
+		ControllerManagedBy(mgr).                  // Create the ControllerManagedBy
+		For(&appsv1.ReplicaSet{}).                 // ReplicaSet is the Application API
+		Owns(&corev1.Pod{}, builder.OnlyMetadata). // ReplicaSet owns Pods created by it, and caches them as metadata only
+		Complete(reconcile.Func(func(req reconcile.Request) (reconcile.Result, error) {
+			ctx, cancel := context.WithCancel(context.Background())
+			defer cancel()
+
+			// Read the ReplicaSet
+			rs := &appsv1.ReplicaSet{}
+			err := cl.Get(ctx, req.NamespacedName, rs)
+			if err != nil {
+				return reconcile.Result{}, client.IgnoreNotFound(err)
+			}
+
+			// List the Pods matching the PodTemplate Labels, but only their metadata
+			var podsMeta metav1.PartialObjectMetadataList
+			err = cl.List(ctx, &podsMeta, client.InNamespace(req.Namespace), client.MatchingLabels(rs.Spec.Template.Labels))
+			if err != nil {
+				return reconcile.Result{}, client.IgnoreNotFound(err)
+			}
+
+			// Update the ReplicaSet
+			rs.Labels["pod-count"] = fmt.Sprintf("%v", len(podsMeta.Items))
+			err = cl.Update(ctx, rs)
+			if err != nil {
+				return reconcile.Result{}, err
+			}
+
+			return reconcile.Result{}, nil
+		}))
+	if err != nil {
+		log.Error(err, "could not create controller")
+		os.Exit(1)
+	}
+
+	if err := mgr.Start(signals.SetupSignalHandler()); err != nil {
+		log.Error(err, "could not start manager")
+		os.Exit(1)
+	}
+}
+
 // This example creates a simple application ControllerManagedBy that is configured for ReplicaSets and Pods.
 //
 // * Create a new application for ReplicaSets that manages Pods owned by the ReplicaSet and calls into

pkg/builder/options.go

@@ -76,3 +76,36 @@ var _ OwnsOption = &Predicates{}
 var _ WatchesOption = &Predicates{}
 
 // }}}
+
+// {{{ For & Owns Dual-Type options
+
+// asProjection configures the projection (currently only metadata) on the input.
+// Currently only metadata is supported.  We might want to expand
+// this to arbitrary non-special local projections in the future.
+type projectAs objectProjection
+
+// ApplyToFor applies this configuration to the given ForInput options.
+func (p projectAs) ApplyToFor(opts *ForInput) {
+	opts.objectProjection = objectProjection(p)
+}
+
+// ApplyToOwns applies this configuration to the given OwnsInput options.
+func (p projectAs) ApplyToOwns(opts *OwnsInput) {
+	opts.objectProjection = objectProjection(p)
+}
+
+var (
+	// OnlyMetadata tells the controller to *only* cache metadata, and to watch
+	// the the API server in metadata-only form.  This is useful when watching
+	// lots of objects, really big objects, or objects for which you only know
+	// the the GVK, but not the structure.  You'll need to pass
+	// metav1.PartialObjectMetadata to the client when fetching objects in your
+	// reconciler, otherwise you'll end up with a duplicate structured or
+	// unstructured cache.
+	OnlyMetadata = projectAs(projectAsMetadata)
+
+	_ ForOption  = OnlyMetadata
+	_ OwnsOption = OnlyMetadata
+)
+
+// }}}