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.

Author: DirectXMan12

pkg/cache/cache_test.go

@@ -153,7 +153,7 @@ func CacheTest(createCacheFunc func(config *rest.Config, opts cache.Options) (ca
 					Expect(listObj.Items).NotTo(BeEmpty())
 					hasKubeService := false
 					for _, svc := range listObj.Items {
-						if svc.Namespace == "default" && svc.Name == "kubernetes" {
+						if isKubeService(&svc) {
 							hasKubeService = true
 							break
 						}
@@ -299,7 +299,7 @@ func CacheTest(createCacheFunc func(config *rest.Config, opts cache.Options) (ca
 					Expect(listObj.Items).NotTo(BeEmpty())
 					hasKubeService := false
 					for _, svc := range listObj.Items {
-						if svc.GetNamespace() == "default" && svc.GetName() == "kubernetes" {
+						if isKubeService(&svc) {
 							hasKubeService = true
 							break
 						}
@@ -471,6 +471,204 @@ func CacheTest(createCacheFunc func(config *rest.Config, opts cache.Options) (ca
 					svc := &kcorev1.Service{}
 					svcKey := client.ObjectKey{Namespace: "unknown", Name: "unknown"}
 
+					By("verifying that an error is returned")
+					err := informerCache.Get(context.Background(), svcKey, svc)
+					Expect(err).To(HaveOccurred())
+				})
+			})
+			Context("with metadata-only objects", func() {
+				It("should be able to list objects that haven't been watched previously", func() {
+					By("listing all services in the cluster")
+					listObj := &kmetav1.PartialObjectMetadataList{}
+					listObj.SetGroupVersionKind(schema.GroupVersionKind{
+						Group:   "",
+						Version: "v1",
+						Kind:    "ServiceList",
+					})
+					err := informerCache.List(context.Background(), listObj)
+					Expect(err).To(Succeed())
+
+					By("verifying that the returned list contains the Kubernetes service")
+					// NB: kubernetes default service is automatically created in testenv.
+					Expect(listObj.Items).NotTo(BeEmpty())
+					hasKubeService := false
+					for _, svc := range listObj.Items {
+						if isKubeService(&svc) {
+							hasKubeService = true
+							break
+						}
+					}
+					Expect(hasKubeService).To(BeTrue())
+				})
+				It("should be able to get objects that haven't been watched previously", func() {
+					By("getting the Kubernetes service")
+					svc := &kmetav1.PartialObjectMetadata{}
+					svc.SetGroupVersionKind(schema.GroupVersionKind{
+						Group:   "",
+						Version: "v1",
+						Kind:    "Service",
+					})
+					svcKey := client.ObjectKey{Namespace: "default", Name: "kubernetes"}
+					Expect(informerCache.Get(context.Background(), svcKey, svc)).To(Succeed())
+
+					By("verifying that the returned service looks reasonable")
+					Expect(svc.GetName()).To(Equal("kubernetes"))
+					Expect(svc.GetNamespace()).To(Equal("default"))
+				})
+
+				It("should support filtering by labels in a single namespace", func() {
+					By("listing pods with a particular label")
+					// NB: each pod has a "test-label": <pod-name>
+					out := kmetav1.PartialObjectMetadataList{}
+					out.SetGroupVersionKind(schema.GroupVersionKind{
+						Group:   "",
+						Version: "v1",
+						Kind:    "PodList",
+					})
+					err := informerCache.List(context.Background(), &out,
+						client.InNamespace(testNamespaceTwo),
+						client.MatchingLabels(map[string]string{"test-label": "test-pod-2"}))
+					Expect(err).To(Succeed())
+
+					By("verifying the returned pods have the correct label")
+					Expect(out.Items).NotTo(BeEmpty())
+					Expect(out.Items).Should(HaveLen(1))
+					actual := out.Items[0]
+					Expect(actual.GetLabels()["test-label"]).To(Equal("test-pod-2"))
+				})
+
+				It("should support filtering by labels from multiple namespaces", func() {
+					By("creating another pod with the same label but different namespace")
+					anotherPod := createPod("test-pod-2", testNamespaceOne, kcorev1.RestartPolicyAlways)
+					defer deletePod(anotherPod)
+
+					By("listing pods with a particular label")
+					// NB: each pod has a "test-label": <pod-name>
+					out := kmetav1.PartialObjectMetadataList{}
+					out.SetGroupVersionKind(schema.GroupVersionKind{
+						Group:   "",
+						Version: "v1",
+						Kind:    "PodList",
+					})
+					labels := map[string]string{"test-label": "test-pod-2"}
+					err := informerCache.List(context.Background(), &out, client.MatchingLabels(labels))
+					Expect(err).To(Succeed())
+
+					By("verifying multiple pods with the same label in different namespaces are returned")
+					Expect(out.Items).NotTo(BeEmpty())
+					Expect(out.Items).Should(HaveLen(2))
+					for _, actual := range out.Items {
+						Expect(actual.GetLabels()["test-label"]).To(Equal("test-pod-2"))
+					}
+
+				})
+
+				It("should be able to list objects by namespace", func() {
+					By("listing pods in test-namespace-1")
+					listObj := &kmetav1.PartialObjectMetadataList{}
+					listObj.SetGroupVersionKind(schema.GroupVersionKind{
+						Group:   "",
+						Version: "v1",
+						Kind:    "PodList",
+					})
+					err := informerCache.List(context.Background(), listObj, client.InNamespace(testNamespaceOne))
+					Expect(err).To(Succeed())
+
+					By("verifying that the returned pods are in test-namespace-1")
+					Expect(listObj.Items).NotTo(BeEmpty())
+					Expect(listObj.Items).Should(HaveLen(1))
+					actual := listObj.Items[0]
+					Expect(actual.GetNamespace()).To(Equal(testNamespaceOne))
+				})
+
+				It("should be able to restrict cache to a namespace", func() {
+					By("creating a namespaced cache")
+					namespacedCache, err := cache.New(cfg, cache.Options{Namespace: testNamespaceOne})
+					Expect(err).NotTo(HaveOccurred())
+
+					By("running the cache and waiting for it to sync")
+					go func() {
+						defer GinkgoRecover()
+						Expect(namespacedCache.Start(informerCacheCtx)).To(Succeed())
+					}()
+					Expect(namespacedCache.WaitForCacheSync(informerCacheCtx)).NotTo(BeFalse())
+
+					By("listing pods in all namespaces")
+					out := &kmetav1.PartialObjectMetadataList{}
+					out.SetGroupVersionKind(schema.GroupVersionKind{
+						Group:   "",
+						Version: "v1",
+						Kind:    "PodList",
+					})
+					Expect(namespacedCache.List(context.Background(), out)).To(Succeed())
+
+					By("verifying the returned pod is from the watched namespace")
+					Expect(out.Items).NotTo(BeEmpty())
+					Expect(out.Items).Should(HaveLen(1))
+					Expect(out.Items[0].GetNamespace()).To(Equal(testNamespaceOne))
+
+					By("listing all namespaces - should still be able to get a cluster-scoped resource")
+					namespaceList := &kmetav1.PartialObjectMetadataList{}
+					namespaceList.SetGroupVersionKind(schema.GroupVersionKind{
+						Group:   "",
+						Version: "v1",
+						Kind:    "NamespaceList",
+					})
+					Expect(namespacedCache.List(context.Background(), namespaceList)).To(Succeed())
+
+					By("verifying the namespace list is not empty")
+					Expect(namespaceList.Items).NotTo(BeEmpty())
+				})
+
+				It("should deep copy the object unless told otherwise", func() {
+					By("retrieving a specific pod from the cache")
+					out := &kmetav1.PartialObjectMetadata{}
+					out.SetGroupVersionKind(schema.GroupVersionKind{
+						Group:   "",
+						Version: "v1",
+						Kind:    "Pod",
+					})
+					uKnownPod2 := &kmetav1.PartialObjectMetadata{}
+					knownPod2.(*kcorev1.Pod).ObjectMeta.DeepCopyInto(&uKnownPod2.ObjectMeta)
+					uKnownPod2.SetGroupVersionKind(schema.GroupVersionKind{
+						Group:   "",
+						Version: "v1",
+						Kind:    "Pod",
+					})
+
+					podKey := client.ObjectKey{Name: "test-pod-2", Namespace: testNamespaceTwo}
+					Expect(informerCache.Get(context.Background(), podKey, out)).To(Succeed())
+
+					By("verifying the retrieved pod is equal to a known pod")
+					Expect(out).To(Equal(uKnownPod2))
+
+					By("altering a field in the retrieved pod")
+					out.Labels["foo"] = "bar"
+
+					By("verifying the pods are no longer equal")
+					Expect(out).NotTo(Equal(knownPod2))
+				})
+
+				It("should return an error if the object is not found", func() {
+					By("getting a service that does not exists")
+					svc := &kmetav1.PartialObjectMetadata{}
+					svc.SetGroupVersionKind(schema.GroupVersionKind{
+						Group:   "",
+						Version: "v1",
+						Kind:    "Service",
+					})
+					svcKey := client.ObjectKey{Namespace: testNamespaceOne, Name: "unknown"}
+
+					By("verifying that an error is returned")
+					err := informerCache.Get(context.Background(), svcKey, svc)
+					Expect(err).To(HaveOccurred())
+					Expect(errors.IsNotFound(err)).To(BeTrue())
+				})
+				It("should return an error if getting object in unwatched namespace", func() {
+					By("getting a service that does not exists")
+					svc := &kcorev1.Service{}
+					svcKey := client.ObjectKey{Namespace: "unknown", Name: "unknown"}
+
 					By("verifying that an error is returned")
 					err := informerCache.Get(context.Background(), svcKey, svc)
 					Expect(err).To(HaveOccurred())
@@ -517,7 +715,6 @@ func CacheTest(createCacheFunc func(config *rest.Config, opts cache.Options) (ca
 					Eventually(out).Should(Receive(Equal(pod)))
 					close(done)
 				})
-				// TODO: Add a test for when GVK is not in Scheme. Does code support informer for unstructured object?
 				It("should be able to get an informer by group/version/kind", func(done Done) {
 					By("getting an shared index informer for gvk = core/v1/pod")
 					gvk := schema.GroupVersionKind{Group: "", Version: "v1", Kind: "Pod"}
@@ -740,6 +937,126 @@ func CacheTest(createCacheFunc func(config *rest.Config, opts cache.Options) (ca
 					Expect(errors.IsTimeout(err)).To(BeTrue())
 				})
 			})
+			Context("with metadata-only objects", func() {
+				It("should be able to get informer for the object", func(done Done) {
+					By("getting a shared index informer for a pod")
+
+					pod := &kcorev1.Pod{
+						ObjectMeta: kmetav1.ObjectMeta{
+							Name:      "informer-obj",
+							Namespace: "default",
+						},
+						Spec: kcorev1.PodSpec{
+							Containers: []kcorev1.Container{
+								{
+									Name:  "nginx",
+									Image: "nginx",
+								},
+							},
+						},
+					}
+
+					podMeta := &kmetav1.PartialObjectMetadata{}
+					pod.ObjectMeta.DeepCopyInto(&podMeta.ObjectMeta)
+					podMeta.SetGroupVersionKind(schema.GroupVersionKind{
+						Group:   "",
+						Version: "v1",
+						Kind:    "Pod",
+					})
+
+					sii, err := informerCache.GetInformer(context.TODO(), podMeta)
+					Expect(err).NotTo(HaveOccurred())
+					Expect(sii).NotTo(BeNil())
+					Expect(sii.HasSynced()).To(BeTrue())
+
+					By("adding an event handler listening for object creation which sends the object to a channel")
+					out := make(chan interface{})
+					addFunc := func(obj interface{}) {
+						out <- obj
+					}
+					sii.AddEventHandler(kcache.ResourceEventHandlerFuncs{AddFunc: addFunc})
+
+					By("adding an object")
+					cl, err := client.New(cfg, client.Options{})
+					Expect(err).NotTo(HaveOccurred())
+					Expect(cl.Create(context.Background(), pod)).To(Succeed())
+					defer deletePod(pod)
+					// re-copy the result in so that we can match on it properly
+					pod.ObjectMeta.DeepCopyInto(&podMeta.ObjectMeta)
+					// NB(directxman12): proto doesn't care typemeta, and
+					// partialobjectmetadata is proto, so no typemeta
+					// TODO(directxman12): we should paper over this in controller-runtime
+					podMeta.APIVersion = ""
+					podMeta.Kind = ""
+
+					By("verifying the object's metadata is received on the channel")
+					Eventually(out).Should(Receive(Equal(podMeta)))
+					close(done)
+				}, 3)
+
+				It("should be able to index an object field then retrieve objects by that field", func() {
+					By("creating the cache")
+					informer, err := cache.New(cfg, cache.Options{})
+					Expect(err).NotTo(HaveOccurred())
+
+					By("indexing the restartPolicy field of the Pod object before starting")
+					pod := &kmetav1.PartialObjectMetadata{}
+					pod.SetGroupVersionKind(schema.GroupVersionKind{
+						Group:   "",
+						Version: "v1",
+						Kind:    "Pod",
+					})
+					indexFunc := func(obj client.Object) []string {
+						metadata := obj.(*kmetav1.PartialObjectMetadata)
+						return []string{metadata.Labels["test-label"]}
+					}
+					Expect(informer.IndexField(context.TODO(), pod, "metadata.labels.test-label", indexFunc)).To(Succeed())
+
+					By("running the cache and waiting for it to sync")
+					go func() {
+						defer GinkgoRecover()
+						Expect(informer.Start(informerCacheCtx)).To(Succeed())
+					}()
+					Expect(informer.WaitForCacheSync(informerCacheCtx)).NotTo(BeFalse())
+
+					By("listing Pods with restartPolicyOnFailure")
+					listObj := &kmetav1.PartialObjectMetadataList{}
+					listObj.SetGroupVersionKind(schema.GroupVersionKind{
+						Group:   "",
+						Version: "v1",
+						Kind:    "PodList",
+					})
+					err = informer.List(context.Background(), listObj,
+						client.MatchingFields{"metadata.labels.test-label": "test-pod-3"})
+					Expect(err).To(Succeed())
+
+					By("verifying that the returned pods have correct restart policy")
+					Expect(listObj.Items).NotTo(BeEmpty())
+					Expect(listObj.Items).Should(HaveLen(1))
+					actual := listObj.Items[0]
+					Expect(actual.GetName()).To(Equal("test-pod-3"))
+				}, 3)
+
+				It("should allow for get informer to be cancelled", func() {
+					By("creating a context and cancelling it")
+					ctx, cancel := context.WithCancel(context.Background())
+					cancel()
+
+					By("getting a shared index informer for a pod with a cancelled context")
+					pod := &kmetav1.PartialObjectMetadata{}
+					pod.SetName("informer-obj2")
+					pod.SetNamespace("default")
+					pod.SetGroupVersionKind(schema.GroupVersionKind{
+						Group:   "",
+						Version: "v1",
+						Kind:    "Pod",
+					})
+					sii, err := informerCache.GetInformer(ctx, pod)
+					Expect(err).To(HaveOccurred())
+					Expect(sii).To(BeNil())
+					Expect(errors.IsTimeout(err)).To(BeTrue())
+				})
+			})
 		})
 	})
 }
@@ -761,3 +1078,9 @@ func ensureNamespace(namespace string, client client.Client) error {
 	}
 	return err
 }
+
+//nolint:interfacer
+func isKubeService(svc kmetav1.Object) bool {
+	// grumble grumble linters grumble grumble
+	return svc.GetNamespace() == "default" && svc.GetName() == "kubernetes"
+}