Merge pull request #6326 from ikaven1024/fix-proxy-pb

fix proxy request error with pb content type

Author: karmada-bot

pkg/search/proxy/controller.go

@@ -291,6 +291,7 @@ func (ctl *Controller) Connect(ctx context.Context, proxyPath string, responder
 		}
 
 		h, err := ctl.proxy.Connect(newCtx, framework.ProxyRequest{
+			RestMapper:           ctl.restMapper,
 			RequestInfo:          requestInfo,
 			GroupVersionResource: gvr,
 			ProxyPath:            proxyPath,

pkg/search/proxy/framework/interface.go

@@ -20,6 +20,7 @@ import (
 	"context"
 	"net/http"
 
+	"k8s.io/apimachinery/pkg/api/meta"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apiserver/pkg/endpoints/request"
 	"k8s.io/apiserver/pkg/registry/rest"
@@ -74,6 +75,7 @@ type Plugin interface {
 
 // ProxyRequest holds parameter for Proxy.Connect()
 type ProxyRequest struct {
+	RestMapper           meta.RESTMapper
 	RequestInfo          *request.RequestInfo
 	GroupVersionResource schema.GroupVersionResource
 	ProxyPath            string

pkg/search/proxy/framework/plugins/cache/cache.go

@@ -25,6 +25,7 @@ import (
 	metainternalversion "k8s.io/apimachinery/pkg/apis/meta/internalversion"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured/unstructuredscheme"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/watch"
@@ -109,7 +110,7 @@ func (c *Cache) Connect(_ context.Context, request framework.ProxyRequest) (http
 			Namer:         meta.NewAccessor(),
 			ClusterScoped: mapping.Scope.Name() == meta.RESTScopeNameRoot,
 		},
-		Serializer:       scheme.Codecs.WithoutConversion(),
+		Serializer:       cacheSerializer,
 		Convertor:        cacheScheme,
 		Subresource:      requestInfo.Subresource,
 		MetaGroupVersion: metav1.SchemeGroupVersion,
@@ -126,6 +127,22 @@ func (c *Cache) Connect(_ context.Context, request framework.ProxyRequest) (http
 	return h, nil
 }
 
+var (
+	cacheSerializer = &customSerializer{
+		NegotiatedSerializer: scheme.Codecs.WithoutConversion(),
+		supportedMediaTypes:  unstructuredscheme.NewUnstructuredNegotiatedSerializer().SupportedMediaTypes(),
+	}
+)
+
+type customSerializer struct {
+	runtime.NegotiatedSerializer
+	supportedMediaTypes []runtime.SerializerInfo
+}
+
+func (u *customSerializer) SupportedMediaTypes() []runtime.SerializerInfo {
+	return u.supportedMediaTypes
+}
+
 type rester struct {
 	store          store.Store
 	gvr            schema.GroupVersionResource

pkg/search/proxy/framework/plugins/cluster/cluster.go

@@ -25,7 +25,10 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/client-go/kubernetes/scheme"
 	listcorev1 "k8s.io/client-go/listers/core/v1"
+	"k8s.io/klog/v2"
 
 	clusterapis "github.com/karmada-io/karmada/pkg/apis/cluster"
 	clusterv1alpha1 "github.com/karmada-io/karmada/pkg/apis/cluster/v1alpha1"
@@ -118,15 +121,15 @@ func (c *Cluster) Connect(ctx context.Context, request framework.ProxyRequest) (
 	// Objects get by client via proxy are edited some fields, different from objects in member clusters.
 	// So before update, we shall recover these fields.
 	return http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) {
-		if err = modifyRequest(req, clusterName); err != nil {
+		if err = modifyRequest(req, request, clusterName); err != nil {
 			request.Responder.Error(err)
 			return
 		}
 		h.ServeHTTP(rw, req)
 	}), nil
 }
 
-func modifyRequest(req *http.Request, cluster string) error {
+func modifyRequest(req *http.Request, requestInfo framework.ProxyRequest, cluster string) error {
 	if req.ContentLength == 0 {
 		return nil
 	}
@@ -143,9 +146,16 @@ func modifyRequest(req *http.Request, cluster string) error {
 		req.ContentLength = int64(body.Len())
 	}()
 
-	obj := &unstructured.Unstructured{}
-	_, _, err = unstructured.UnstructuredJSONScheme.Decode(body.Bytes(), nil, obj)
+	encoder, decoder, obj, err := getSerializer(req, requestInfo)
 	if err != nil {
+		klog.Warningf("modifyRequest getSerializer error for request %s: %v", req.RequestURI, err)
+		// ignore error
+		return nil
+	}
+
+	_, _, err = decoder.Decode(body.Bytes(), nil, obj)
+	if err != nil {
+		klog.Warningf("modifyRequest decode error for request %s: %v", req.RequestURI, err)
 		// ignore error
 		return nil
 	}
@@ -156,8 +166,23 @@ func modifyRequest(req *http.Request, cluster string) error {
 	if changed {
 		// write changed object into body
 		body.Reset()
-		return unstructured.UnstructuredJSONScheme.Encode(obj, body)
+		return encoder.Encode(obj, body)
 	}
 
 	return nil
 }
+
+func getSerializer(req *http.Request, requestInfo framework.ProxyRequest) (runtime.Encoder, runtime.Decoder, runtime.Object, error) {
+	gvk, _ := requestInfo.RestMapper.KindFor(requestInfo.GroupVersionResource)
+	if scheme.Scheme.Recognizes(gvk) {
+		negotiator := runtime.NewClientNegotiator(scheme.Codecs.WithoutConversion(), requestInfo.GroupVersionResource.GroupVersion())
+		contentType := req.Header.Get("Content-Type")
+		encoder, _ := negotiator.Encoder(contentType, nil)
+		decoder, _ := negotiator.Decoder(contentType, nil)
+
+		obj, err := scheme.Scheme.New(gvk)
+		return encoder, decoder, obj, err
+	}
+
+	return unstructured.UnstructuredJSONScheme, unstructured.UnstructuredJSONScheme, &unstructured.Unstructured{}, nil
+}