Merge pull request #128243 from benluddy/cbor-dynamic-integration

KEP-4222: Add CBOR variant of admission webhook integration test.

test/featuregates_linter/test_data/versioned_feature_list.yaml

@@ -1252,6 +1252,12 @@
     lockToDefault: false
     preRelease: Beta
     version: "1.32"
+- name: TestOnlyCBORServingAndStorage
+  versionedSpecs:
+  - default: false
+    lockToDefault: false
+    preRelease: Alpha
+    version: "1.32"
 - name: TopologyAwareHints
   versionedSpecs:
   - default: false

test/integration/apiserver/admissionwebhook/admission_test.go

@@ -20,7 +20,6 @@ import (
 	"context"
 	"crypto/tls"
 	"crypto/x509"
-	"encoding/json"
 	"fmt"
 	"io"
 	"net/http"
@@ -49,6 +48,7 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/json"
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/apimachinery/pkg/util/wait"
 	genericapirequest "k8s.io/apiserver/pkg/endpoints/request"
@@ -446,16 +446,25 @@ func (w *warningHandler) HandleWarningHeader(code int, agent string, message str
 
 // TestWebhookAdmissionWithWatchCache tests communication between API server and webhook process.
 func TestWebhookAdmissionWithWatchCache(t *testing.T) {
-	testWebhookAdmission(t, true)
+	testWebhookAdmission(t, true, func(testing.TB, *rest.Config) {})
 }
 
 // TestWebhookAdmissionWithoutWatchCache tests communication between API server and webhook process.
 func TestWebhookAdmissionWithoutWatchCache(t *testing.T) {
-	testWebhookAdmission(t, false)
+	testWebhookAdmission(t, false, func(testing.TB, *rest.Config) {})
+}
+
+func TestWebhookAdmissionWithCBOR(t *testing.T) {
+	framework.EnableCBORServingAndStorageForTest(t)
+	framework.SetTestOnlyCBORClientFeatureGatesForTest(t, true, true)
+	testWebhookAdmission(t, false, func(t testing.TB, config *rest.Config) {
+		config.Wrap(framework.AssertRequestResponseAsCBOR(t))
+	})
 }
 
 // testWebhookAdmission tests communication between API server and webhook process.
-func testWebhookAdmission(t *testing.T, watchCache bool) {
+func testWebhookAdmission(t *testing.T, watchCache bool, reconfigureClient func(testing.TB, *rest.Config)) {
+
 	// holder communicates expectations to webhooks, and results from webhooks
 	holder := &holder{
 		t:                 t,
@@ -528,10 +537,6 @@ func testWebhookAdmission(t *testing.T, watchCache bool) {
 	}
 
 	// gather resources to test
-	dynamicClient, err := dynamic.NewForConfig(clientConfig)
-	if err != nil {
-		t.Fatal(err)
-	}
 	_, resources, err := client.Discovery().ServerGroupsAndResources()
 	if err != nil {
 		t.Fatalf("Failed to get ServerGroupsAndResources with error: %+v", err)
@@ -640,6 +645,13 @@ func testWebhookAdmission(t *testing.T, watchCache bool) {
 			for _, verb := range []string{"create", "update", "patch", "connect", "delete", "deletecollection"} {
 				if shouldTestResourceVerb(gvr, resource, verb) {
 					t.Run(verb, func(t *testing.T) {
+						clientConfig := rest.CopyConfig(clientConfig)
+						reconfigureClient(t, clientConfig)
+						dynamicClient, err := dynamic.NewForConfig(clientConfig)
+						if err != nil {
+							t.Fatal(err)
+						}
+
 						count++
 						holder.reset(t)
 						testFunc := getTestFunc(gvr, verb)

test/integration/client/dynamic_client_test.go

@@ -20,11 +20,13 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"net/http"
 	"reflect"
 	"testing"
 	"time"
 
-	v1 "k8s.io/api/core/v1"
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/fields"
@@ -38,6 +40,7 @@ import (
 	"k8s.io/client-go/dynamic"
 	clientset "k8s.io/client-go/kubernetes"
 	clientscheme "k8s.io/client-go/kubernetes/scheme"
+	"k8s.io/client-go/rest"
 	kubeapiservertesting "k8s.io/kubernetes/cmd/kube-apiserver/app/testing"
 	"k8s.io/kubernetes/test/integration/framework"
 )
@@ -55,12 +58,12 @@ func TestDynamicClient(t *testing.T) {
 	resource := schema.GroupVersionResource{Group: "", Version: "v1", Resource: "pods"}
 
 	// Create a Pod with the normal client
-	pod := &v1.Pod{
+	pod := &corev1.Pod{
 		ObjectMeta: metav1.ObjectMeta{
 			GenerateName: "test",
 		},
-		Spec: v1.PodSpec{
-			Containers: []v1.Container{
+		Spec: corev1.PodSpec{
+			Containers: []corev1.Container{
 				{
 					Name:  "test",
 					Image: "test-image",
@@ -134,16 +137,16 @@ func TestDynamicClientWatch(t *testing.T) {
 		t.Fatalf("unexpected error creating dynamic client: %v", err)
 	}
 
-	resource := v1.SchemeGroupVersion.WithResource("events")
+	resource := corev1.SchemeGroupVersion.WithResource("events")
 
-	mkEvent := func(i int) *v1.Event {
+	mkEvent := func(i int) *corev1.Event {
 		name := fmt.Sprintf("event-%v", i)
-		return &v1.Event{
+		return &corev1.Event{
 			ObjectMeta: metav1.ObjectMeta{
 				Namespace: "default",
 				Name:      name,
 			},
-			InvolvedObject: v1.ObjectReference{
+			InvolvedObject: corev1.ObjectReference{
 				Namespace: "default",
 				Name:      name,
 			},
@@ -276,24 +279,171 @@ func TestUnstructuredExtract(t *testing.T) {
 
 }
 
-func unstructuredToPod(obj *unstructured.Unstructured) (*v1.Pod, error) {
+func unstructuredToPod(obj *unstructured.Unstructured) (*corev1.Pod, error) {
 	json, err := runtime.Encode(unstructured.UnstructuredJSONScheme, obj)
 	if err != nil {
 		return nil, err
 	}
-	pod := new(v1.Pod)
-	err = runtime.DecodeInto(clientscheme.Codecs.LegacyCodec(v1.SchemeGroupVersion), json, pod)
+	pod := new(corev1.Pod)
+	err = runtime.DecodeInto(clientscheme.Codecs.LegacyCodec(corev1.SchemeGroupVersion), json, pod)
 	pod.Kind = ""
 	pod.APIVersion = ""
 	return pod, err
 }
 
-func unstructuredToEvent(obj *unstructured.Unstructured) (*v1.Event, error) {
+func unstructuredToEvent(obj *unstructured.Unstructured) (*corev1.Event, error) {
 	json, err := runtime.Encode(unstructured.UnstructuredJSONScheme, obj)
 	if err != nil {
 		return nil, err
 	}
-	event := new(v1.Event)
-	err = runtime.DecodeInto(clientscheme.Codecs.LegacyCodec(v1.SchemeGroupVersion), json, event)
+	event := new(corev1.Event)
+	err = runtime.DecodeInto(clientscheme.Codecs.LegacyCodec(corev1.SchemeGroupVersion), json, event)
 	return event, err
 }
+
+func TestDynamicClientCBOREnablement(t *testing.T) {
+	for _, tc := range []struct {
+		name                    string
+		serving                 bool
+		allowed                 bool
+		preferred               bool
+		wantRequestContentType  string
+		wantRequestAccept       string
+		wantResponseContentType string
+		wantResponseStatus      int
+		wantStatusError         bool
+	}{
+		{
+			name:                    "sends cbor accepts both gets cbor",
+			serving:                 true,
+			allowed:                 true,
+			preferred:               true,
+			wantRequestContentType:  "application/cbor",
+			wantRequestAccept:       "application/json;q=0.9,application/cbor;q=1",
+			wantResponseContentType: "application/cbor",
+			wantResponseStatus:      http.StatusCreated,
+			wantStatusError:         false,
+		},
+		{
+			name:                    "sends cbor accepts both gets 415",
+			serving:                 false,
+			allowed:                 true,
+			preferred:               true,
+			wantRequestContentType:  "application/cbor",
+			wantRequestAccept:       "application/json;q=0.9,application/cbor;q=1",
+			wantResponseContentType: "application/json",
+			wantResponseStatus:      http.StatusUnsupportedMediaType,
+			wantStatusError:         true,
+		},
+		{
+			name:                    "sends json accepts both gets cbor",
+			serving:                 true,
+			allowed:                 true,
+			preferred:               false,
+			wantRequestContentType:  "application/json",
+			wantRequestAccept:       "application/json;q=0.9,application/cbor;q=1",
+			wantResponseContentType: "application/cbor",
+			wantResponseStatus:      http.StatusCreated,
+			wantStatusError:         false,
+		},
+		{
+			name:                    "sends json accepts both gets json",
+			serving:                 false,
+			allowed:                 true,
+			preferred:               false,
+			wantRequestContentType:  "application/json",
+			wantRequestAccept:       "application/json;q=0.9,application/cbor;q=1",
+			wantResponseContentType: "application/json",
+			wantResponseStatus:      http.StatusCreated,
+			wantStatusError:         false,
+		},
+		{
+			name:                    "sends json accepts json gets json with serving enabled",
+			serving:                 true,
+			allowed:                 false,
+			preferred:               false,
+			wantRequestContentType:  "application/json",
+			wantRequestAccept:       "application/json",
+			wantResponseContentType: "application/json",
+			wantResponseStatus:      http.StatusCreated,
+			wantStatusError:         false,
+		},
+		{
+			name:                    "sends json accepts json gets json with serving disabled",
+			serving:                 false,
+			allowed:                 false,
+			preferred:               false,
+			wantRequestContentType:  "application/json",
+			wantRequestAccept:       "application/json",
+			wantResponseContentType: "application/json",
+			wantResponseStatus:      http.StatusCreated,
+			wantStatusError:         false,
+		},
+		{
+			name:                    "sends json without both gates enabled",
+			serving:                 true,
+			allowed:                 false,
+			preferred:               true,
+			wantRequestContentType:  "application/json",
+			wantRequestAccept:       "application/json",
+			wantResponseContentType: "application/json",
+			wantResponseStatus:      http.StatusCreated,
+			wantStatusError:         false,
+		},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			if tc.serving {
+				framework.EnableCBORServingAndStorageForTest(t)
+			}
+
+			server := kubeapiservertesting.StartTestServerOrDie(t, nil, framework.DefaultTestServerFlags(), framework.SharedEtcd())
+			defer server.TearDownFn()
+
+			framework.SetTestOnlyCBORClientFeatureGatesForTest(t, tc.allowed, tc.preferred)
+
+			config := rest.CopyConfig(server.ClientConfig)
+			config.Wrap(func(rt http.RoundTripper) http.RoundTripper {
+				return roundTripperFunc(func(request *http.Request) (*http.Response, error) {
+					response, err := rt.RoundTrip(request)
+					if got := response.Request.Header.Get("Content-Type"); got != tc.wantRequestContentType {
+						t.Errorf("want request content type %q, got %q", tc.wantRequestContentType, got)
+					}
+					if got := response.Request.Header.Get("Accept"); got != tc.wantRequestAccept {
+						t.Errorf("want request accept %q, got %q", tc.wantRequestAccept, got)
+					}
+					if got := response.Header.Get("Content-Type"); got != tc.wantResponseContentType {
+						t.Errorf("want response content type %q, got %q", tc.wantResponseContentType, got)
+					}
+					if got := response.StatusCode; got != tc.wantResponseStatus {
+						t.Errorf("want response status %d, got %d", tc.wantResponseStatus, got)
+					}
+					return response, err
+				})
+			})
+			client, err := dynamic.NewForConfig(config)
+			if err != nil {
+				t.Fatal(err)
+			}
+
+			_, err = client.Resource(corev1.SchemeGroupVersion.WithResource("namespaces")).Create(
+				context.TODO(),
+				&unstructured.Unstructured{
+					Object: map[string]interface{}{
+						"metadata": map[string]interface{}{
+							"name": "test-dynamic-client-cbor-enablement",
+						},
+					},
+				},
+				metav1.CreateOptions{DryRun: []string{metav1.DryRunAll}},
+			)
+			switch {
+			case tc.wantStatusError && errors.IsUnsupportedMediaType(err):
+				// ok
+			case !tc.wantStatusError && err == nil:
+				// ok
+			default:
+				t.Errorf("unexpected error: %v", err)
+			}
+		})
+	}
+}

test/integration/framework/cbor.go

@@ -0,0 +1,173 @@
+/*
+Copyright 2024 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 framework
+
+import (
+	"bytes"
+	"io"
+	"net/http"
+	"testing"
+
+	apiextensionsapiserver "k8s.io/apiextensions-apiserver/pkg/apiserver"
+	metainternalscheme "k8s.io/apimachinery/pkg/apis/meta/internalversion/scheme"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/serializer"
+	"k8s.io/apimachinery/pkg/runtime/serializer/cbor"
+	"k8s.io/apimachinery/pkg/util/sets"
+	"k8s.io/apiserver/pkg/features"
+	utilfeature "k8s.io/apiserver/pkg/util/feature"
+	clientfeatures "k8s.io/client-go/features"
+	"k8s.io/client-go/transport"
+	featuregatetesting "k8s.io/component-base/featuregate/testing"
+	aggregatorscheme "k8s.io/kube-aggregator/pkg/apiserver/scheme"
+	"k8s.io/kubernetes/pkg/api/legacyscheme"
+)
+
+// SetTestOnlyCBORClientFeatureGatesForTest overrides the CBOR client feature gates in the test-only
+// client feature gate instance for the duration of a test. The CBOR client feature gates are
+// temporarily registered in their own feature gate instance that does not include runtime wiring to
+// command-line flags or environment variables in order to mitigate the risk of enabling a new
+// encoding before all integration tests have been demonstrated to pass.
+//
+// This will be removed as an alpha requirement. The client feature gates will be registered with
+// the existing feature gate instance and tests will use
+// k8s.io/client-go/features/testing.SetFeatureDuringTest (which unlike
+// k8s.io/component-base/featuregate/testing.SetFeatureGateDuringTest does not accept a feature gate
+// instance as a parameter).
+func SetTestOnlyCBORClientFeatureGatesForTest(tb testing.TB, allowed, preferred bool) {
+	originalAllowed := clientfeatures.TestOnlyFeatureGates.Enabled(clientfeatures.TestOnlyClientAllowsCBOR)
+	tb.Cleanup(func() {
+		if err := clientfeatures.TestOnlyFeatureGates.Set(clientfeatures.TestOnlyClientAllowsCBOR, originalAllowed); err != nil {
+			tb.Fatal(err)
+		}
+	})
+	if err := clientfeatures.TestOnlyFeatureGates.Set(clientfeatures.TestOnlyClientAllowsCBOR, allowed); err != nil {
+		tb.Fatal(err)
+	}
+
+	originalPreferred := clientfeatures.TestOnlyFeatureGates.Enabled(clientfeatures.TestOnlyClientPrefersCBOR)
+	tb.Cleanup(func() {
+		if err := clientfeatures.TestOnlyFeatureGates.Set(clientfeatures.TestOnlyClientPrefersCBOR, originalPreferred); err != nil {
+			tb.Fatal(err)
+		}
+	})
+	if err := clientfeatures.TestOnlyFeatureGates.Set(clientfeatures.TestOnlyClientPrefersCBOR, preferred); err != nil {
+		tb.Fatal(err)
+	}
+}
+
+// EnableCBORForTest patches global state to enable the CBOR serializer and reverses those changes
+// at the end of the test. As a risk mitigation, integration tests are initially written this way so
+// that integration tests can be implemented fully and incrementally before exposing options
+// (including feature gates) that can enable CBOR at runtime. After integration test coverage is
+// complete, feature gates will be introduced to completely supersede this mechanism.
+func EnableCBORServingAndStorageForTest(tb testing.TB) {
+	featuregatetesting.SetFeatureGateDuringTest(tb, utilfeature.TestOnlyFeatureGate, features.TestOnlyCBORServingAndStorage, true)
+
+	newCBORSerializerInfo := func(creater runtime.ObjectCreater, typer runtime.ObjectTyper) runtime.SerializerInfo {
+		return runtime.SerializerInfo{
+			MediaType:        "application/cbor",
+			MediaTypeType:    "application",
+			MediaTypeSubType: "cbor",
+			Serializer:       cbor.NewSerializer(creater, typer),
+			StrictSerializer: cbor.NewSerializer(creater, typer, cbor.Strict(true)),
+			StreamSerializer: &runtime.StreamSerializerInfo{
+				Framer:     cbor.NewFramer(),
+				Serializer: cbor.NewSerializer(creater, typer, cbor.Transcode(false)),
+			},
+		}
+	}
+
+	// Codecs for built-in types are constructed at package initialization time and read by
+	// value from REST storage providers.
+	codecs := map[*runtime.Scheme]*serializer.CodecFactory{
+		legacyscheme.Scheme:           &legacyscheme.Codecs,
+		metainternalscheme.Scheme:     &metainternalscheme.Codecs,
+		aggregatorscheme.Scheme:       &aggregatorscheme.Codecs,
+		apiextensionsapiserver.Scheme: &apiextensionsapiserver.Codecs,
+	}
+
+	for scheme, factory := range codecs {
+		original := *factory // shallow copy of original value
+		tb.Cleanup(func() { *codecs[scheme] = original })
+		*codecs[scheme] = serializer.NewCodecFactory(scheme, serializer.WithSerializer(newCBORSerializerInfo))
+	}
+}
+
+// AssertRequestResponseAsCBOR returns a transport.WrapperFunc that will report a test error if a
+// non-empty request or response body contains data that does not appear to be CBOR-encoded.
+func AssertRequestResponseAsCBOR(t testing.TB) transport.WrapperFunc {
+	recognizer := cbor.NewSerializer(runtime.NewScheme(), runtime.NewScheme())
+
+	unsupportedPatchContentTypes := sets.New(
+		"application/json-patch+json",
+		"application/merge-patch+json",
+		"application/strategic-merge-patch+json",
+	)
+
+	return func(rt http.RoundTripper) http.RoundTripper {
+		return roundTripperFunc(func(request *http.Request) (*http.Response, error) {
+			if request.Body != nil && !unsupportedPatchContentTypes.Has(request.Header.Get("Content-Type")) {
+				requestbody, err := io.ReadAll(request.Body)
+				if err != nil {
+					t.Error(err)
+				}
+				recognized, _, err := recognizer.RecognizesData(requestbody)
+				if err != nil {
+					t.Error(err)
+				}
+				if len(requestbody) > 0 && !recognized {
+					t.Errorf("non-cbor request: 0x%x", requestbody)
+				}
+				request.Body = io.NopCloser(bytes.NewReader(requestbody))
+			}
+
+			response, rterr := rt.RoundTrip(request)
+			if rterr != nil {
+				return response, rterr
+			}
+
+			// We can't synchronously inspect streaming responses, so tee to a buffer
+			// and inspect it at the end of the test.
+			var buf bytes.Buffer
+			response.Body = struct {
+				io.Reader
+				io.Closer
+			}{
+				Reader: io.TeeReader(response.Body, &buf),
+				Closer: response.Body,
+			}
+			t.Cleanup(func() {
+				recognized, _, err := recognizer.RecognizesData(buf.Bytes())
+				if err != nil {
+					t.Error(err)
+				}
+				if buf.Len() > 0 && !recognized {
+					t.Errorf("non-cbor response: 0x%x", buf.Bytes())
+				}
+			})
+
+			return response, rterr
+		})
+	}
+}
+
+type roundTripperFunc func(*http.Request) (*http.Response, error)
+
+func (f roundTripperFunc) RoundTrip(r *http.Request) (*http.Response, error) {
+	return f(r)
+}