Merge pull request #132928 from p0lyn0mial/upstream-watchlist-reflector-table-unsupported

reflector: detects unsupported meta.Table gvks for watchlist

staging/src/k8s.io/apiserver/pkg/endpoints/handlers/get.go

@@ -45,7 +45,6 @@ import (
 	utilfeature "k8s.io/apiserver/pkg/util/feature"
 	"k8s.io/component-base/tracing"
 	"k8s.io/klog/v2"
-	"k8s.io/utils/ptr"
 )
 
 // getterFunc performs a get request with the given context and object name. The request
@@ -204,12 +203,7 @@ func ListResource(r rest.Lister, rw rest.Watcher, scope *RequestScope, forceWatc
 			return
 		}
 
-		var restrictions negotiation.EndpointRestrictions
-		restrictions = scope
-		if isListWatchRequest(opts) {
-			restrictions = &watchListEndpointRestrictions{scope}
-		}
-		outputMediaType, _, err := negotiation.NegotiateOutputMediaType(req, scope.Serializer, restrictions)
+		outputMediaType, _, err := negotiation.NegotiateOutputMediaType(req, scope.Serializer, scope)
 		if err != nil {
 			scope.err(err, w, req)
 			return
@@ -315,18 +309,3 @@ func ListResource(r rest.Lister, rw rest.Watcher, scope *RequestScope, forceWatc
 		transformResponseObject(ctx, scope, req, w, http.StatusOK, outputMediaType, result)
 	}
 }
-
-type watchListEndpointRestrictions struct {
-	negotiation.EndpointRestrictions
-}
-
-func (e *watchListEndpointRestrictions) AllowsMediaTypeTransform(mimeType, mimeSubType string, target *schema.GroupVersionKind) bool {
-	if target != nil && target.Kind == "Table" {
-		return false
-	}
-	return e.EndpointRestrictions.AllowsMediaTypeTransform(mimeType, mimeSubType, target)
-}
-
-func isListWatchRequest(opts metainternalversion.ListOptions) bool {
-	return utilfeature.DefaultFeatureGate.Enabled(features.WatchList) && ptr.Deref(opts.SendInitialEvents, false) && opts.AllowWatchBookmarks
-}

staging/src/k8s.io/client-go/tools/cache/reflector.go

@@ -31,6 +31,7 @@ import (
 	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	metav1beta1 "k8s.io/apimachinery/pkg/apis/meta/v1beta1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/util/naming"
@@ -912,6 +913,15 @@ loop:
 					continue
 				}
 			}
+			// For now, let’s block unsupported Table
+			// resources for watchlist only
+			// see #132926 for more info
+			if exitOnWatchListBookmarkReceived {
+				if unsupportedGVK := isUnsupportedTableObject(event.Object); unsupportedGVK {
+					utilruntime.HandleErrorWithContext(ctx, nil, "Unsupported watch event object gvk", "reflector", name, "actualGVK", event.Object.GetObjectKind().GroupVersionKind())
+					continue
+				}
+			}
 			meta, err := meta.Accessor(event.Object)
 			if err != nil {
 				utilruntime.HandleErrorWithContext(ctx, err, "Unable to understand watch event", "reflector", name, "event", event)
@@ -1179,3 +1189,22 @@ func (e *VeryShortWatchError) Error() string {
 	return fmt.Sprintf("very short watch: %s: Unexpected watch close - "+
 		"watch lasted less than a second and no items received", e.Name)
 }
+
+var unsupportedTableGVK = map[schema.GroupVersionKind]bool{
+	metav1beta1.SchemeGroupVersion.WithKind("Table"): true,
+	metav1.SchemeGroupVersion.WithKind("Table"):      true,
+}
+
+// isUnsupportedTableObject checks whether the given runtime.Object
+// is a "Table" object that belongs to a set of well-known unsupported GroupVersionKinds.
+func isUnsupportedTableObject(rawObject runtime.Object) bool {
+	unstructuredObj, ok := rawObject.(*unstructured.Unstructured)
+	if !ok {
+		return false
+	}
+	if unstructuredObj.GetKind() != "Table" {
+		return false
+	}
+
+	return unsupportedTableGVK[rawObject.GetObjectKind().GroupVersionKind()]
+}

staging/src/k8s.io/client-go/tools/cache/reflector_test.go

@@ -2361,3 +2361,56 @@ func BenchmarkReflectorList(b *testing.B) {
 		})
 	}
 }
+
+func TestIsUnsupportedTableObject(t *testing.T) {
+	tests := []struct {
+		name     string
+		obj      runtime.Object
+		expected bool
+	}{
+		{
+			name: "Unsupported Table object in meta.k8s.io/v1beta1",
+			obj: &unstructured.Unstructured{
+				Object: map[string]interface{}{
+					"apiVersion": "meta.k8s.io/v1beta1",
+					"kind":       "Table",
+				},
+			},
+			expected: true,
+		},
+		{
+			name: "Unsupported Table object in meta.k8s.io/v1",
+			obj: &unstructured.Unstructured{
+				Object: map[string]interface{}{
+					"apiVersion": "meta.k8s.io/v1",
+					"kind":       "Table",
+				},
+			},
+			expected: true,
+		},
+		{
+			name:     "Pod obj is not a Table",
+			obj:      &v1.Pod{},
+			expected: false,
+		},
+		{
+			name: "Table object with unrecognised API group",
+			obj: &unstructured.Unstructured{
+				Object: map[string]interface{}{
+					"apiVersion": "custom.group/v1",
+					"kind":       "Table",
+				},
+			},
+			expected: false,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			result := isUnsupportedTableObject(tt.obj)
+			if result != tt.expected {
+				t.Errorf("Expected %v, got %v", tt.expected, result)
+			}
+		})
+	}
+}

test/e2e/apimachinery/watchlist.go

@@ -30,7 +30,6 @@ import (
 	"github.com/onsi/gomega"
 
 	v1 "k8s.io/api/core/v1"
-	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
@@ -214,9 +213,7 @@ var _ = SIGDescribe("API Streaming (aka. WatchList)", framework.WithFeatureGate(
 		gomega.Expect(rt.actualRequests).To(gomega.Equal(expectedRequestsMadeByMetaClient))
 	})
 
-	// Validates unsupported Accept headers in WatchList.
-	// Sets AcceptContentType to "application/json;as=Table", which the API doesn't support, returning a 406 error.
-	ginkgo.It("doesn't support receiving resources as Tables", func(ctx context.Context) {
+	ginkgo.It("server supports sending resources in Table format", func(ctx context.Context) {
 		featuregatetesting.SetFeatureGateDuringTest(ginkgo.GinkgoTB(), utilfeature.DefaultFeatureGate, featuregate.Feature(clientfeatures.WatchListClient), true)
 
 		modifiedClientConfig := dynamic.ConfigFor(f.ClientConfig())
@@ -232,20 +229,17 @@ var _ = SIGDescribe("API Streaming (aka. WatchList)", framework.WithFeatureGate(
 		framework.ExpectNoError(err)
 		gomega.Expect(hasPreparedOptions).To(gomega.BeTrueBecause("it should be possible to prepare watchlist opts from an empty ListOptions"))
 
-		_, err = dynamicClient.Resource(v1.SchemeGroupVersion.WithResource("secrets")).Namespace("default").Watch(ctx, opts)
-		gomega.Expect(err).To(gomega.HaveOccurred())
-		gomega.Expect(err.(apierrors.APIStatus)).To(gomega.HaveField("Status().Code", gomega.Equal(int32(406))))
+		w, err := dynamicClient.Resource(v1.SchemeGroupVersion.WithResource("secrets")).Namespace("default").Watch(ctx, opts)
+		framework.ExpectNoError(err)
+		defer w.Stop()
 	})
 
-	// Sets AcceptContentType to both "application/json;as=Table" and "application/json".
-	// Unlike the previous test, no 406 error occurs, as the API falls back to "application/json" and returns a valid response.
-	ginkgo.It("falls backs to supported content type when when receiving resources as Tables was requested", func(ctx context.Context) {
+	ginkgo.It("reflector doesn't support receiving resources as Tables", func(ctx context.Context) {
 		featuregatetesting.SetFeatureGateDuringTest(ginkgo.GinkgoTB(), utilfeature.DefaultFeatureGate, featuregate.Feature(clientfeatures.WatchListClient), true)
 
 		modifiedClientConfig := dynamic.ConfigFor(f.ClientConfig())
 		modifiedClientConfig.AcceptContentTypes = strings.Join([]string{
 			fmt.Sprintf("application/json;as=Table;v=%s;g=%s", metav1.SchemeGroupVersion.Version, metav1.GroupName),
-			"application/json",
 		}, ",")
 		modifiedClientConfig.GroupVersion = &v1.SchemeGroupVersion
 		restClient, err := rest.RESTClientFor(modifiedClientConfig)
@@ -270,19 +264,17 @@ var _ = SIGDescribe("API Streaming (aka. WatchList)", framework.WithFeatureGate(
 			nil,
 		)
 
-		expectedSecrets := addWellKnownUnstructuredSecrets(ctx, f)
+		_ = addWellKnownUnstructuredSecrets(ctx, f)
 
 		ginkgo.By("Starting the secret informer")
 		go secretInformer.Run(stopCh)
 
-		ginkgo.By("Waiting until the secret informer is fully synchronised")
-		err = wait.PollUntilContextTimeout(ctx, 100*time.Millisecond, 30*time.Second, false, func(context.Context) (done bool, err error) {
+		ginkgo.By("Checking if the secret informer hasn't been synced")
+		err = wait.PollUntilContextTimeout(ctx, 100*time.Millisecond, 10*time.Second, false, func(context.Context) (done bool, err error) {
 			return secretInformer.HasSynced(), nil
 		})
-		framework.ExpectNoError(err, "Failed waiting for the secret informer in %s namespace to be synced", f.Namespace.Namespace)
-
-		ginkgo.By("Verifying if the secret informer was properly synchronised")
-		verifyStoreFor(ctx, verifyStoreForMetaObject[unstructured.Unstructured](expectedSecrets, secretInformer.GetStore()))
+		gomega.Expect(err).To(gomega.HaveOccurred())
+		gomega.Expect(secretInformer.GetStore().List()).To(gomega.BeEmpty(), "unsupported resources should not have been added to the store")
 	})
 })