apiserver: forward request with prefix path or header

Signed-off-by: Iceber Gu <[email protected]>

Author: Iceber

pkg/kubeapiserver/apiserver.go

@@ -130,30 +130,15 @@ func (c completedConfig) New(delegationTarget genericapiserver.DelegationTarget)
 
 	delegate := delegationTarget.UnprotectedHandler()
 	if delegate == nil {
+		// To prevent anomalous requests from being sent to member clusters due to improper usage,
+		// we do not allow proxying requests with unmatched prefixes to member clusters.
 		delegate = http.NotFoundHandler()
 	}
 
 	restManager := NewRESTManager(c.GenericConfig.Serializer, runtime.ContentTypeJSON, c.StorageFactory, c.InitialAPIGroupResources)
 	discoveryManager := discovery.NewDiscoveryManager(c.GenericConfig.Serializer, restManager, delegate)
 
-	// handle root discovery request
-	genericserver.Handler.NonGoRestfulMux.Handle("/api", discoveryManager)
-	genericserver.Handler.NonGoRestfulMux.Handle("/apis", discoveryManager)
-
-	resourceHandler := &ResourceHandler{
-		minRequestTimeout: time.Duration(c.GenericConfig.MinRequestTimeout) * time.Second,
-
-		delegate:      delegate,
-		rest:          restManager,
-		discovery:     discoveryManager,
-		clusterLister: c.InformerFactory.Cluster().V1alpha2().PediaClusters().Lister(),
-	}
-	genericserver.Handler.NonGoRestfulMux.HandlePrefix("/api/", resourceHandler)
-	genericserver.Handler.NonGoRestfulMux.HandlePrefix("/apis/", resourceHandler)
-
 	clusterInformer := c.InformerFactory.Cluster().V1alpha2().PediaClusters()
-	_ = NewClusterResourceController(restManager, discoveryManager, clusterInformer)
-
 	connector := proxyrest.NewProxyConnector(clusterInformer.Lister(), c.ExtraConfig.AllowPediaClusterConfigReuse, c.ExtraConfig.ExtraProxyRequestHeaderPrefixes)
 
 	methodSet := sets.New("GET")
@@ -181,8 +166,30 @@ func (c completedConfig) New(delegationTarget genericapiserver.DelegationTarget)
 			methods = append(methods, m)
 		}
 	}
+	proxy := proxyrest.NewRemoteProxyREST(c.GenericConfig.Serializer, connector)
+
+	// forward request
+	genericserver.Handler.NonGoRestfulMux.HandlePrefix("/proxy/", http.StripPrefix("/proxy", proxy))
+
+	// handle root discovery request
+	discoveryHandler := WrapForwardRequestHandler(discoveryManager, proxy)
+	genericserver.Handler.NonGoRestfulMux.Handle("/api", discoveryHandler)
+	genericserver.Handler.NonGoRestfulMux.Handle("/apis", discoveryHandler)
 
-	resourceHandler.proxy = proxyrest.NewRemoteProxyREST(c.GenericConfig.Serializer, connector)
+	resourceHandler := &ResourceHandler{
+		minRequestTimeout: time.Duration(c.GenericConfig.MinRequestTimeout) * time.Second,
+
+		delegate:      delegate,
+		proxy:         proxy,
+		rest:          restManager,
+		discovery:     discoveryManager,
+		clusterLister: c.InformerFactory.Cluster().V1alpha2().PediaClusters().Lister(),
+	}
+
+	genericserver.Handler.NonGoRestfulMux.HandlePrefix("/api/", resourceHandler)
+	genericserver.Handler.NonGoRestfulMux.HandlePrefix("/apis/", resourceHandler)
+
+	_ = NewClusterResourceController(restManager, discoveryManager, clusterInformer)
 	return genericserver, methods, nil
 }
 
@@ -228,3 +235,13 @@ func (r wrapRequestInfoResolverForNamespace) NewRequestInfo(req *http.Request) (
 	}
 	return info, nil
 }
+
+func WrapForwardRequestHandler(handler http.Handler, proxy http.Handler) http.Handler {
+	return http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) {
+		if HasForwardRequestHeader(req) {
+			proxy.ServeHTTP(rw, req)
+			return
+		}
+		handler.ServeHTTP(rw, req)
+	})
+}

pkg/kubeapiserver/resource_handler.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"net/http"
+	"strings"
 	"time"
 
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
@@ -44,8 +45,13 @@ func (r *ResourceHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
 		return
 	}
 
+	shouldForwardRequest := HasForwardRequestHeader(req)
 	// handle discovery request
 	if !requestInfo.IsResourceRequest {
+		if shouldForwardRequest {
+			r.proxy.ServeHTTP(w, req)
+			return
+		}
 		r.discovery.ServeHTTP(w, req)
 		return
 	}
@@ -80,11 +86,16 @@ func (r *ResourceHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
 		return
 	}
 
+	if shouldForwardRequest {
+		r.proxy.ServeHTTP(w, req)
+		return
+	}
+
 	resource, reqScope, storage, existed := r.rest.GetResourceREST(gvr, requestInfo.Subresource)
 	if !existed {
 		// TODO(iceber): Add the specialized error for subresources
 		err := fmt.Errorf("not found request scope or resource storage")
-		klog.ErrorS(err, "Failed to handle resource request", "resource", gvr)
+		klog.ErrorS(err, "Failed to handle resource request", "resource", gvr, "subresource", requestInfo.Subresource)
 		responsewriters.ErrorNegotiated(
 			apierrors.NewInternalError(err),
 			Codecs, gvr.GroupVersion(), w, req,
@@ -168,3 +179,8 @@ func checkClusterAndWarning(ctx context.Context, cluster *clusterv1alpha2.PediaC
 		warning.AddWarning(ctx, "", msg)
 	}
 }
+
+func HasForwardRequestHeader(req *http.Request) bool {
+	value := req.Header.Get("x-clusterpedia-forward")
+	return strings.ToLower(value) == "true"
+}

pkg/kubeapiserver/resourcerest/proxy/proxy.go

@@ -2,18 +2,17 @@ package proxy
 
 import (
 	"context"
-	"errors"
 	"net/http"
 	"net/url"
 
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	utilnet "k8s.io/apimachinery/pkg/util/net"
 	"k8s.io/apimachinery/pkg/util/proxy"
 	auditinternal "k8s.io/apiserver/pkg/apis/audit"
 	"k8s.io/apiserver/pkg/audit"
 	"k8s.io/apiserver/pkg/endpoints/handlers/responsewriters"
-	genericrequest "k8s.io/apiserver/pkg/endpoints/request"
 
 	"github.com/clusterpedia-io/clusterpedia/pkg/utils/request"
 )
@@ -46,12 +45,7 @@ func (r *RemoteProxyREST) Error(w http.ResponseWriter, req *http.Request, err er
 func proxyConn(ctx context.Context, connGetter ClusterConnectionGetter, upgradeRequired bool, responder proxy.ErrorResponder, wrapProxy func(*proxy.UpgradeAwareHandler) http.Handler) (http.Handler, error) {
 	clusterName := request.ClusterNameValue(ctx)
 	if clusterName == "" {
-		return nil, errors.New("missing cluster")
-	}
-
-	requestInfo, ok := genericrequest.RequestInfoFrom(ctx)
-	if !ok {
-		return nil, errors.New("missing RequestInfo")
+		return nil, apierrors.NewBadRequest("please specify the cluster name when using proxy model.")
 	}
 
 	return http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) {
@@ -62,12 +56,18 @@ func proxyConn(ctx context.Context, connGetter ClusterConnectionGetter, upgradeR
 			return
 		}
 
-		target, err := url.ParseRequestURI(endpoint + requestInfo.Path)
+		target, err := url.ParseRequestURI(endpoint + req.URL.Path)
 		if err != nil {
 			responder.Error(rw, req, err)
 			return
 		}
-		target.RawQuery = request.RequestQueryFrom(ctx).Encode()
+
+		// First get URL Query from context.Context
+		if query := request.RequestQueryFrom(ctx); query != nil {
+			target.RawQuery = query.Encode()
+		} else {
+			target.RawQuery = req.URL.RawQuery
+		}
 
 		proxy := proxy.NewUpgradeAwareHandler(target, transport, false, upgradeRequired, responder)
 		proxy.UseLocationHost = true