Merge pull request kubernetes#128497 from benluddy/cbor-request-contenttype-circuit-breaker

KEP-4222: Fall back to JSON request encoding after CBOR 415.

Author: k8s-ci-robot

staging/src/k8s.io/client-go/rest/client.go

@@ -24,6 +24,7 @@ import (
 	"os"
 	"strconv"
 	"strings"
+	"sync/atomic"
 	"time"
 
 	"github.com/munnerz/goautoneg"
@@ -89,7 +90,7 @@ type RESTClient struct {
 	versionedAPIPath string
 
 	// content describes how a RESTClient encodes and decodes responses.
-	content ClientContentConfig
+	content requestClientContentConfigProvider
 
 	// creates BackoffManager that is passed to requests.
 	createBackoffMgr func() BackoffManager
@@ -119,11 +120,10 @@ func NewRESTClient(baseURL *url.URL, versionedAPIPath string, config ClientConte
 	return &RESTClient{
 		base:             &base,
 		versionedAPIPath: versionedAPIPath,
-		content:          scrubCBORContentConfigIfDisabled(config),
+		content:          requestClientContentConfigProvider{base: scrubCBORContentConfigIfDisabled(config)},
 		createBackoffMgr: readExpBackoffConfig,
 		rateLimiter:      rateLimiter,
-
-		Client: client,
+		Client:           client,
 	}, nil
 }
 
@@ -237,5 +237,60 @@ func (c *RESTClient) Delete() *Request {
 
 // APIVersion returns the APIVersion this RESTClient is expected to use.
 func (c *RESTClient) APIVersion() schema.GroupVersion {
-	return c.content.GroupVersion
+	return c.content.GetClientContentConfig().GroupVersion
+}
+
+// requestClientContentConfigProvider observes HTTP 415 (Unsupported Media Type) responses to detect
+// that the server does not understand CBOR. Once this has happened, future requests are forced to
+// use JSON so they can succeed. This is convenient for client users that want to prefer CBOR, but
+// also need to interoperate with older servers so requests do not permanently fail. The clients
+// will not default to using CBOR until at least all supported kube-apiservers have enable-CBOR
+// locked to true, so this path will be rarely taken. Additionally, all generated clients accessing
+// built-in kube resources are forced to protobuf, so those will not degrade to JSON.
+type requestClientContentConfigProvider struct {
+	base ClientContentConfig
+
+	// Becomes permanently true if a server responds with HTTP 415 (Unsupported Media Type) to a
+	// request with "Content-Type" header containing the CBOR media type.
+	sawUnsupportedMediaTypeForCBOR atomic.Bool
+}
+
+// GetClientContentConfig returns the ClientContentConfig that should be used for new requests by
+// this client.
+func (p *requestClientContentConfigProvider) GetClientContentConfig() ClientContentConfig {
+	if !clientfeatures.TestOnlyFeatureGates.Enabled(clientfeatures.TestOnlyClientAllowsCBOR) {
+		return p.base
+	}
+
+	if sawUnsupportedMediaTypeForCBOR := p.sawUnsupportedMediaTypeForCBOR.Load(); !sawUnsupportedMediaTypeForCBOR {
+		return p.base
+	}
+
+	if mediaType, _, _ := mime.ParseMediaType(p.base.ContentType); mediaType != runtime.ContentTypeCBOR {
+		return p.base
+	}
+
+	config := p.base
+	// The default ClientContentConfig sets ContentType to CBOR and the client has previously
+	// received an HTTP 415 in response to a CBOR request. Override ContentType to JSON.
+	config.ContentType = runtime.ContentTypeJSON
+	return config
+}
+
+// UnsupportedMediaType reports that the server has responded to a request with HTTP 415 Unsupported
+// Media Type.
+func (p *requestClientContentConfigProvider) UnsupportedMediaType(requestContentType string) {
+	if !clientfeatures.TestOnlyFeatureGates.Enabled(clientfeatures.TestOnlyClientAllowsCBOR) {
+		return
+	}
+
+	// This could be extended to consider the Content-Encoding request header, the Accept and
+	// Accept-Encoding response headers, the request method, and URI (as mentioned in
+	// https://www.rfc-editor.org/rfc/rfc9110.html#section-15.5.16). The request Content-Type
+	// header is sufficient to implement a blanket CBOR fallback mechanism.
+	requestContentType, _, _ = mime.ParseMediaType(requestContentType)
+	switch requestContentType {
+	case runtime.ContentTypeCBOR, string(types.ApplyCBORPatchType):
+		p.sawUnsupportedMediaTypeForCBOR.Store(true)
+	}
 }

staging/src/k8s.io/client-go/rest/request.go

@@ -156,7 +156,7 @@ func NewRequest(c *RESTClient) *Request {
 		timeout = c.Client.Timeout
 	}
 
-	contentConfig := c.content
+	contentConfig := c.content.GetClientContentConfig()
 	contentTypeNotSet := len(contentConfig.ContentType) == 0
 	if contentTypeNotSet {
 		contentConfig.ContentType = "application/json"
@@ -188,7 +188,7 @@ func NewRequestWithClient(base *url.URL, versionedAPIPath string, content Client
 	return NewRequest(&RESTClient{
 		base:             base,
 		versionedAPIPath: versionedAPIPath,
-		content:          content,
+		content:          requestClientContentConfigProvider{base: content},
 		Client:           client,
 	})
 }
@@ -1235,6 +1235,9 @@ func (r *Request) request(ctx context.Context, fn func(*http.Request, *http.Resp
 		if req.ContentLength >= 0 && !(req.Body != nil && req.ContentLength == 0) {
 			metrics.RequestSize.Observe(ctx, r.verb, r.URL().Host, float64(req.ContentLength))
 		}
+		if resp != nil && resp.StatusCode == http.StatusUnsupportedMediaType {
+			r.c.content.UnsupportedMediaType(resp.Request.Header.Get("Content-Type"))
+		}
 		retry.After(ctx, r, resp, err)
 
 		done := func() bool {

test/integration/client/client_test.go

@@ -1984,3 +1984,51 @@ func TestCBORWithTypedClient(t *testing.T) {
 		t.Fatal(err)
 	}
 }
+
+func TestUnsupportedMediaTypeCircuitBreaker(t *testing.T) {
+	framework.SetTestOnlyCBORClientFeatureGatesForTest(t, true, true)
+
+	server := kubeapiservertesting.StartTestServerOrDie(t, nil, framework.DefaultTestServerFlags(), framework.SharedEtcd())
+	t.Cleanup(server.TearDownFn)
+
+	config := rest.CopyConfig(server.ClientConfig)
+	config.ContentType = "application/cbor"
+	config.AcceptContentTypes = "application/json"
+
+	client, err := corev1client.NewForConfig(config)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if _, err := client.Namespaces().Create(
+		context.TODO(),
+		&v1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: "test-client-415"}},
+		metav1.CreateOptions{DryRun: []string{metav1.DryRunAll}},
+	); !apierrors.IsUnsupportedMediaType(err) {
+		t.Errorf("expected to receive unsupported media type on first cbor request, got: %v", err)
+	}
+
+	// Requests from this client should fall back from application/cbor to application/json.
+	if _, err := client.Namespaces().Create(
+		context.TODO(),
+		&v1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: "test-client-415"}},
+		metav1.CreateOptions{DryRun: []string{metav1.DryRunAll}},
+	); err != nil {
+		t.Errorf("expected to receive nil error on subsequent cbor request, got: %v", err)
+	}
+
+	// The circuit breaker trips on a per-client basis, so it should not begin tripped for a
+	// fresh client with identical config.
+	client, err = corev1client.NewForConfig(config)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if _, err := client.Namespaces().Create(
+		context.TODO(),
+		&v1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: "test-client-415"}},
+		metav1.CreateOptions{DryRun: []string{metav1.DryRunAll}},
+	); !apierrors.IsUnsupportedMediaType(err) {
+		t.Errorf("expected to receive unsupported media type on cbor request with fresh client, got: %v", err)
+	}
+}

test/integration/client/dynamic_client_test.go

@@ -589,3 +589,49 @@ func TestDynamicClientCBOREnablement(t *testing.T) {
 		})
 	}
 }
+
+func TestUnsupportedMediaTypeCircuitBreakerDynamicClient(t *testing.T) {
+	framework.SetTestOnlyCBORClientFeatureGatesForTest(t, true, true)
+
+	server := kubeapiservertesting.StartTestServerOrDie(t, nil, framework.DefaultTestServerFlags(), framework.SharedEtcd())
+	t.Cleanup(server.TearDownFn)
+
+	config := rest.CopyConfig(server.ClientConfig)
+
+	client, err := dynamic.NewForConfig(config)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if _, err := client.Resource(corev1.SchemeGroupVersion.WithResource("namespaces")).Create(
+		context.TODO(),
+		&unstructured.Unstructured{Object: map[string]interface{}{"metadata": map[string]interface{}{"name": "test-dynamic-client-415"}}},
+		metav1.CreateOptions{DryRun: []string{metav1.DryRunAll}},
+	); !errors.IsUnsupportedMediaType(err) {
+		t.Errorf("expected to receive unsupported media type on first cbor request, got: %v", err)
+	}
+
+	// Requests from this client should fall back from application/cbor to application/json.
+	if _, err := client.Resource(corev1.SchemeGroupVersion.WithResource("namespaces")).Create(
+		context.TODO(),
+		&unstructured.Unstructured{Object: map[string]interface{}{"metadata": map[string]interface{}{"name": "test-dynamic-client-415"}}},
+		metav1.CreateOptions{DryRun: []string{metav1.DryRunAll}},
+	); err != nil {
+		t.Errorf("expected to receive nil error on subsequent cbor request, got: %v", err)
+	}
+
+	// The circuit breaker trips on a per-client basis, so it should not begin tripped for a
+	// fresh client with identical config.
+	client, err = dynamic.NewForConfig(config)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if _, err := client.Resource(corev1.SchemeGroupVersion.WithResource("namespaces")).Create(
+		context.TODO(),
+		&unstructured.Unstructured{Object: map[string]interface{}{"metadata": map[string]interface{}{"name": "test-dynamic-client-415"}}},
+		metav1.CreateOptions{DryRun: []string{metav1.DryRunAll}},
+	); !errors.IsUnsupportedMediaType(err) {
+		t.Errorf("expected to receive unsupported media type on cbor request with fresh client, got: %v", err)
+	}
+}