🐛 Metadata objects should always preserve GroupVersionKind

Signed-off-by: Vince Prignano <vincepri@vmware.com>

pkg/builder/controller_test.go

@@ -408,8 +408,17 @@ var _ = Describe("application", func() {
 				Owns(&appsv1.ReplicaSet{}, OnlyMetadata).
 				Watches(&source.Kind{Type: &appsv1.StatefulSet{}},
 					handler.EnqueueRequestsFromMapFunc(func(o client.Object) []reconcile.Request {
+						defer GinkgoRecover()
+
 						ometa := o.(*metav1.PartialObjectMetadata)
 						statefulSetMaps <- ometa
+
+						// Validate that the GVK is not empty when dealing with PartialObjectMetadata objects.
+						Expect(o.GetObjectKind().GroupVersionKind()).To(Equal(schema.GroupVersionKind{
+							Group:   "apps",
+							Version: "v1",
+							Kind:    "StatefulSet",
+						}))
 						return nil
 					}),
 					OnlyMetadata)

pkg/cache/cache_test.go

@@ -1018,11 +1018,6 @@ func CacheTest(createCacheFunc func(config *rest.Config, opts cache.Options) (ca
 					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)))

pkg/cache/internal/informers_map.go

@@ -25,6 +25,7 @@ import (
 
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/meta"
+	metainternalversionscheme "k8s.io/apimachinery/pkg/apis/meta/internalversion/scheme"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
@@ -34,7 +35,6 @@ import (
 	"k8s.io/client-go/metadata"
 	"k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/cache"
-
 	"sigs.k8s.io/controller-runtime/pkg/client/apiutil"
 )
 
@@ -216,6 +216,13 @@ func (ip *specificInformersMap) addInformerToMap(gvk schema.GroupVersionKind, ob
 	if err != nil {
 		return nil, false, err
 	}
+
+	switch obj.(type) {
+	case *metav1.PartialObjectMetadata, *metav1.PartialObjectMetadataList:
+		ni = metadataSharedIndexInformerPreserveGVK(gvk, ni)
+	default:
+	}
+
 	i := &MapEntry{
 		Informer: ni,
 		Reader:   CacheReader{indexer: ni.GetIndexer(), groupVersionKind: gvk, scopeName: rm.Scope.Name()},
@@ -278,7 +285,12 @@ func createUnstructuredListWatch(gvk schema.GroupVersionKind, ip *specificInform
 	if err != nil {
 		return nil, err
 	}
-	dynamicClient, err := dynamic.NewForConfig(ip.config)
+
+	// If the rest configuration has a negotiated serializer passed in,
+	// we should remove it and use the one that the dynamic client sets for us.
+	cfg := rest.CopyConfig(ip.config)
+	cfg.NegotiatedSerializer = nil
+	dynamicClient, err := dynamic.NewForConfig(cfg)
 	if err != nil {
 		return nil, err
 	}
@@ -314,8 +326,16 @@ func createMetadataListWatch(gvk schema.GroupVersionKind, ip *specificInformersM
 		return nil, err
 	}
 
+	// WARNING! Our custom serializer is needed for metadata watches and clients
+	// otherwise the default one used in client-go removes the GVK on decoding
+	// which wouldn't allow us to know which object are we actually dealing with.
+	cfg := rest.CopyConfig(ip.config)
+	cfg.NegotiatedSerializer = apiutil.SerializerWithDecodedGVK{
+		WithoutConversionCodecFactory: serializer.WithoutConversionCodecFactory{CodecFactory: metainternalversionscheme.Codecs},
+	}
+
 	// grab the metadata client
-	client, err := metadata.NewForConfig(ip.config)
+	client, err := metadata.NewForConfig(cfg)
 	if err != nil {
 		return nil, err
 	}

pkg/cache/internal/metadata_infomer_wrapper.go

@@ -0,0 +1,71 @@
+/*
+Copyright 2021 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package internal
+
+import (
+	"time"
+
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/client-go/tools/cache"
+)
+
+func metadataSharedIndexInformerPreserveGVK(gvk schema.GroupVersionKind, si cache.SharedIndexInformer) cache.SharedIndexInformer {
+	return &sharedInformerWrapper{
+		gvk:                 gvk,
+		SharedIndexInformer: si,
+	}
+}
+
+type sharedInformerWrapper struct {
+	gvk schema.GroupVersionKind
+	cache.SharedIndexInformer
+}
+
+func (s *sharedInformerWrapper) AddEventHandler(handler cache.ResourceEventHandler) {
+	s.SharedIndexInformer.AddEventHandler(&handlerPreserveGVK{s.gvk, handler})
+}
+
+func (s *sharedInformerWrapper) AddEventHandlerWithResyncPeriod(handler cache.ResourceEventHandler, resyncPeriod time.Duration) {
+	s.SharedIndexInformer.AddEventHandlerWithResyncPeriod(&handlerPreserveGVK{s.gvk, handler}, resyncPeriod)
+}
+
+type handlerPreserveGVK struct {
+	gvk schema.GroupVersionKind
+	cache.ResourceEventHandler
+}
+
+func (h *handlerPreserveGVK) resetGroupVersionKind(obj interface{}) {
+	if v, ok := obj.(schema.ObjectKind); ok {
+		v.SetGroupVersionKind(h.gvk)
+	}
+}
+
+func (h *handlerPreserveGVK) OnAdd(obj interface{}) {
+	h.resetGroupVersionKind(obj)
+	h.ResourceEventHandler.OnAdd(obj)
+}
+
+func (h *handlerPreserveGVK) OnUpdate(oldObj, newObj interface{}) {
+	h.resetGroupVersionKind(oldObj)
+	h.resetGroupVersionKind(newObj)
+	h.ResourceEventHandler.OnUpdate(oldObj, newObj)
+}
+
+func (h *handlerPreserveGVK) OnDelete(obj interface{}) {
+	h.resetGroupVersionKind(obj)
+	h.ResourceEventHandler.OnDelete(obj)
+}

pkg/client/apiutil/apimachinery.go

@@ -118,15 +118,24 @@ func GVKForObject(obj runtime.Object, scheme *runtime.Scheme) (schema.GroupVersi
 // with the given GroupVersionKind. The REST client will be configured to use the negotiated serializer from
 // baseConfig, if set, otherwise a default serializer will be set.
 func RESTClientForGVK(gvk schema.GroupVersionKind, isUnstructured bool, baseConfig *rest.Config, codecs serializer.CodecFactory) (rest.Interface, error) {
-	cfg := createRestConfig(gvk, isUnstructured, baseConfig)
-	if cfg.NegotiatedSerializer == nil {
-		cfg.NegotiatedSerializer = serializer.WithoutConversionCodecFactory{CodecFactory: codecs}
-	}
-	return rest.RESTClientFor(cfg)
+	return rest.RESTClientFor(createRestConfig(gvk, isUnstructured, baseConfig, codecs))
+}
+
+// SerializerWithDecodedGVK is a CodecFactory that overrides the DecoderToVersion of a WithoutConversionCodecFactory
+// in order to avoid clearing the GVK from the decoded object.
+//
+// See https://github.com/kubernetes/kubernetes/issues/80609.
+type SerializerWithDecodedGVK struct {
+	serializer.WithoutConversionCodecFactory
+}
+
+// DecoderToVersion returns an decoder that does not do conversion.
+func (f SerializerWithDecodedGVK) DecoderToVersion(serializer runtime.Decoder, _ runtime.GroupVersioner) runtime.Decoder {
+	return serializer
 }
 
 //createRestConfig copies the base config and updates needed fields for a new rest config
-func createRestConfig(gvk schema.GroupVersionKind, isUnstructured bool, baseConfig *rest.Config) *rest.Config {
+func createRestConfig(gvk schema.GroupVersionKind, isUnstructured bool, baseConfig *rest.Config, codecs serializer.CodecFactory) *rest.Config {
 	gv := gvk.GroupVersion()
 
 	cfg := rest.CopyConfig(baseConfig)
@@ -147,5 +156,16 @@ func createRestConfig(gvk schema.GroupVersionKind, isUnstructured bool, baseConf
 		}
 		protobufSchemeLock.RUnlock()
 	}
+
+	if cfg.NegotiatedSerializer == nil {
+		if isUnstructured {
+			// If the object is unstructured, we need to preserve the GVK information.
+			// Use our own custom serializer.
+			cfg.NegotiatedSerializer = SerializerWithDecodedGVK{serializer.WithoutConversionCodecFactory{CodecFactory: codecs}}
+		} else {
+			cfg.NegotiatedSerializer = serializer.WithoutConversionCodecFactory{CodecFactory: codecs}
+		}
+	}
+
 	return cfg
 }

pkg/client/client.go

@@ -159,7 +159,6 @@ type client struct {
 }
 
 // resetGroupVersionKind is a helper function to restore and preserve GroupVersionKind on an object.
-// TODO(vincepri): Remove this function and its calls once controller-runtime dependencies are upgraded to 1.16?
 func (c *client) resetGroupVersionKind(obj runtime.Object, gvk schema.GroupVersionKind) {
 	if gvk != schema.EmptyObjectKind.GroupVersionKind() {
 		if v, ok := obj.(schema.ObjectKind); ok {
@@ -246,6 +245,8 @@ func (c *client) Get(ctx context.Context, key ObjectKey, obj Object) error {
 	case *unstructured.Unstructured:
 		return c.unstructuredClient.Get(ctx, key, obj)
 	case *metav1.PartialObjectMetadata:
+		// Metadata only object should always preserve the GVK coming in from the caller.
+		defer c.resetGroupVersionKind(obj, obj.GetObjectKind().GroupVersionKind())
 		return c.metadataClient.Get(ctx, key, obj)
 	default:
 		return c.typedClient.Get(ctx, key, obj)
@@ -258,6 +259,8 @@ func (c *client) List(ctx context.Context, obj ObjectList, opts ...ListOption) e
 	case *unstructured.UnstructuredList:
 		return c.unstructuredClient.List(ctx, obj, opts...)
 	case *metav1.PartialObjectMetadataList:
+		// Metadata only object should always preserve the GVK coming in from the caller.
+		defer c.resetGroupVersionKind(obj, obj.GetObjectKind().GroupVersionKind())
 		return c.metadataClient.List(ctx, obj, opts...)
 	default:
 		return c.typedClient.List(ctx, obj, opts...)

pkg/client/client_test.go

@@ -1558,16 +1558,20 @@ var _ = Describe("Client", func() {
 
 				By("fetching the created Deployment")
 				var actual metav1.PartialObjectMetadata
-				actual.SetGroupVersionKind(schema.GroupVersionKind{
+				gvk := schema.GroupVersionKind{
 					Group:   "apps",
 					Version: "v1",
 					Kind:    "Deployment",
-				})
+				}
+				actual.SetGroupVersionKind(gvk)
 				key := client.ObjectKey{Namespace: ns, Name: dep.Name}
 				err = cl.Get(context.TODO(), key, &actual)
 				Expect(err).NotTo(HaveOccurred())
 				Expect(actual).NotTo(BeNil())
 
+				By("validating that the GVK has been preserved")
+				Expect(actual.GroupVersionKind()).To(Equal(gvk))
+
 				By("validating the fetched deployment equals the created one")
 				Expect(metaOnlyFromObj(dep, scheme)).To(Equal(&actual))
 
@@ -2389,14 +2393,19 @@ var _ = Describe("Client", func() {
 				Expect(err).NotTo(HaveOccurred())
 
 				By("listing all objects of that type in the cluster")
-				metaList := &metav1.PartialObjectMetadataList{}
-				metaList.SetGroupVersionKind(schema.GroupVersionKind{
+				gvk := schema.GroupVersionKind{
 					Group:   "apps",
 					Version: "v1",
 					Kind:    "DeploymentList",
-				})
+				}
+				metaList := &metav1.PartialObjectMetadataList{}
+				metaList.SetGroupVersionKind(gvk)
 				Expect(cl.List(context.Background(), metaList)).NotTo(HaveOccurred())
 
+				By("validating that the list GVK has been preserved")
+				Expect(metaList.GroupVersionKind()).To(Equal(gvk))
+
+				By("validating that the list has the expected deployment")
 				Expect(metaList.Items).NotTo(BeEmpty())
 				hasDep := false
 				for _, item := range metaList.Items {