From 8d1cbc83d8c4fdd242a7827ce296a97dd5a154cb Mon Sep 17 00:00:00 2001 From: Lukasz Szaszkiewicz Date: Fri, 4 Jul 2025 13:39:15 +0200 Subject: [PATCH 1/3] endpoints/handlers/get: remove watchListEndpointRestrictions --- .../apiserver/pkg/endpoints/handlers/get.go | 23 +------------------ 1 file changed, 1 insertion(+), 22 deletions(-) diff --git a/staging/src/k8s.io/apiserver/pkg/endpoints/handlers/get.go b/staging/src/k8s.io/apiserver/pkg/endpoints/handlers/get.go index 423925ea982..eb52baf1946 100644 --- a/staging/src/k8s.io/apiserver/pkg/endpoints/handlers/get.go +++ b/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 -} From da7c55e0d2f962c8493864a069be5acb8902579f Mon Sep 17 00:00:00 2001 From: Lukasz Szaszkiewicz Date: Mon, 14 Jul 2025 13:13:44 +0200 Subject: [PATCH 2/3] reflector: detects unsupported meta.Table gvks for watchlist --- .../k8s.io/client-go/tools/cache/reflector.go | 29 ++++++++++ .../client-go/tools/cache/reflector_test.go | 53 +++++++++++++++++++ 2 files changed, 82 insertions(+) diff --git a/staging/src/k8s.io/client-go/tools/cache/reflector.go b/staging/src/k8s.io/client-go/tools/cache/reflector.go index 84aa4275999..ee9be77278a 100644 --- a/staging/src/k8s.io/client-go/tools/cache/reflector.go +++ b/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()] +} diff --git a/staging/src/k8s.io/client-go/tools/cache/reflector_test.go b/staging/src/k8s.io/client-go/tools/cache/reflector_test.go index 558cdceeaef..8c20ae0b02a 100644 --- a/staging/src/k8s.io/client-go/tools/cache/reflector_test.go +++ b/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) + } + }) + } +} From 7d9eb2b3d0dc032796be2614c81132a65ffeeab7 Mon Sep 17 00:00:00 2001 From: Lukasz Szaszkiewicz Date: Mon, 14 Jul 2025 13:28:03 +0200 Subject: [PATCH 3/3] test/apimachinery/watchlist: update tests that receive resources in Table format --- test/e2e/apimachinery/watchlist.go | 28 ++++++++++------------------ 1 file changed, 10 insertions(+), 18 deletions(-) diff --git a/test/e2e/apimachinery/watchlist.go b/test/e2e/apimachinery/watchlist.go index 9e5cd7f0e59..2b5ead8b981 100644 --- a/test/e2e/apimachinery/watchlist.go +++ b/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") }) })