Repository: clusternet/clusternet
Pull request #18 (2021-07-08 03:52:41)

optimize proxy handler
<!--  Thanks for sending a pull request!

-->

#### What type of PR is this?
kind/change

#### What this PR does / why we need it:
Use subresource `proxy` instead.


#### Which issue(s) this PR fixes:
<!--
*Automatically closes linked issue when PR is merged.
Usage: `Fixes #<issue number>`, or `Fixes (paste link of issue)`.
-->
Supercede part of #3 .

#### Special notes for your reviewer:



Patch:
diff --git a/README.md b/README.md
index e66d467..67063d0 100644
--- a/README.md
+++ b/README.md
@@ -274,14 +274,14 @@ users:
     client-key-data: REDACTED
 # suppose your child cluster running at http://demo1.cluster.net
 $ kubectl config set-cluster `kubectl config get-clusters | grep -v NAME` \
-  --server=https://10.0.0.10:6443/apis/proxies.clusternet.io/v1alpha1/sockets/dc91021d-2361-4f6d-a404-7c33b9e01118/http/demo1.cluster.net
-# or just use the direct path
+  --server=https://10.0.0.10:6443/apis/proxies.clusternet.io/v1alpha1/sockets/dc91021d-2361-4f6d-a404-7c33b9e01118/proxy/http/demo1.cluster.net
+# or just use the proxy path
 $ kubectl config set-cluster `kubectl config get-clusters | grep -v NAME` \
-  --server=https://10.0.0.10:6443/apis/proxies.clusternet.io/v1alpha1/sockets/dc91021d-2361-4f6d-a404-7c33b9e01118/direct
+  --server=https://10.0.0.10:6443/apis/proxies.clusternet.io/v1alpha1/sockets/dc91021d-2361-4f6d-a404-7c33b9e01118/proxy
 ```
 
-What you need is to append `/apis/proxies.clusternet.io/v1alpha1/sockets/<CLUSTER-ID>/http/<SERVER-URL>`
-or `/apis/proxies.clusternet.io/v1alpha1/sockets/<CLUSTER-ID>/direct` at the end of original parent cluster server address.
+What you need is to append `/apis/proxies.clusternet.io/v1alpha1/sockets/<CLUSTER-ID>/proxy/http/<SERVER-URL>`
+or `/apis/proxies.clusternet.io/v1alpha1/sockets/<CLUSTER-ID>/proxy` at the end of original parent cluster server address.
 - `CLUSTER-ID` is a UUID for your child cluster, which is auto-populated by `clusternet-agent`, such as
   dc91021d-2361-4f6d-a404-7c33b9e01118. You could get this UUID from objects `ClusterRegistrationRequest`,
   `ManagedCluster`, etc. Also this UUID is labeled with key `clusters.clusternet.io/cluster-id`.
diff --git a/go.mod b/go.mod
index aec455e..78b2b4e 100644
--- a/go.mod
+++ b/go.mod
@@ -6,7 +6,7 @@ require (
 	github.com/coreos/go-systemd v0.0.0-20190620071333-e64a0ec8b42a // indirect
 	github.com/go-openapi/spec v0.19.5
 	github.com/google/go-cmp v0.5.4 // indirect
-	github.com/gorilla/mux v1.8.0
+	github.com/gorilla/mux v1.8.0 // indirect
 	github.com/gorilla/websocket v1.4.2
 	github.com/grpc-ecosystem/go-grpc-middleware v1.2.2 // indirect
 	github.com/grpc-ecosystem/grpc-gateway v1.12.1 // indirect
diff --git a/pkg/apis/proxies/types.go b/pkg/apis/proxies/types.go
index 4943e66..938d183 100644
--- a/pkg/apis/proxies/types.go
+++ b/pkg/apis/proxies/types.go
@@ -26,5 +26,13 @@ import (
 type Socket struct {
 	metav1.TypeMeta
 
-	// TODO
+	// Path is the part of URLs that include service endpoints, suffixes,
+	// and parameters to use for the current proxy request to child cluster.
+	// For example, the whole request URL is
+	// http://localhost:8001/apis/proxies.clusternet.io/v1alpha1/sockets/cae6feb5-a23f-4354-8ee7-66527aa03f54/proxy/https/demo.com:6443.
+	// Path is /https/demo.com:6443.
+	//
+	// +optional
+	// Path is the URL path to use for the current proxy request
+	Path string
 }
diff --git a/pkg/apis/proxies/v1alpha1/types.go b/pkg/apis/proxies/v1alpha1/types.go
index 62b243c..e84af17 100644
--- a/pkg/apis/proxies/v1alpha1/types.go
+++ b/pkg/apis/proxies/v1alpha1/types.go
@@ -27,5 +27,13 @@ import (
 type Socket struct {
 	metav1.TypeMeta `json:",inline"`
 
-	// TODO
+	// Path is the part of URLs that include service endpoints, suffixes,
+	// and parameters to use for the current proxy request to child cluster.
+	// For example, the whole request URL is
+	// http://localhost:8001/apis/proxies.clusternet.io/v1alpha1/sockets/cae6feb5-a23f-4354-8ee7-66527aa03f54/proxy/https/demo.com:6443.
+	// Path is /https/demo.com:6443.
+	//
+	// +optional
+	// Path is the URL path to use for the current proxy request
+	Path string `json:"path,omitempty"`
 }
diff --git a/pkg/apis/proxies/v1alpha1/zz_generated.conversion.go b/pkg/apis/proxies/v1alpha1/zz_generated.conversion.go
index e998797..6a49240 100644
--- a/pkg/apis/proxies/v1alpha1/zz_generated.conversion.go
+++ b/pkg/apis/proxies/v1alpha1/zz_generated.conversion.go
@@ -53,6 +53,7 @@ func RegisterConversions(s *runtime.Scheme) error {
 }
 
 func autoConvert_v1alpha1_Socket_To_proxies_Socket(in *Socket, out *proxies.Socket, s conversion.Scope) error {
+	out.Path = in.Path
 	return nil
 }
 
@@ -62,6 +63,7 @@ func Convert_v1alpha1_Socket_To_proxies_Socket(in *Socket, out *proxies.Socket,
 }
 
 func autoConvert_proxies_Socket_To_v1alpha1_Socket(in *proxies.Socket, out *Socket, s conversion.Scope) error {
+	out.Path = in.Path
 	return nil
 }
 
@@ -73,6 +75,13 @@ func Convert_proxies_Socket_To_v1alpha1_Socket(in *proxies.Socket, out *Socket,
 func autoConvert_url_Values_To_v1alpha1_Socket(in *url.Values, out *Socket, s conversion.Scope) error {
 	// WARNING: Field TypeMeta does not have json tag, skipping.
 
+	if values, ok := map[string][]string(*in)["path"]; ok && len(values) > 0 {
+		if err := runtime.Convert_Slice_string_To_string(&values, &out.Path, s); err != nil {
+			return err
+		}
+	} else {
+		out.Path = ""
+	}
 	return nil
 }
 
diff --git a/pkg/exchanger/exchanger.go b/pkg/exchanger/exchanger.go
index 5553d58..6481426 100644
--- a/pkg/exchanger/exchanger.go
+++ b/pkg/exchanger/exchanger.go
@@ -18,14 +18,15 @@ package exchanger
 
 import (
 	"context"
+	"crypto/tls"
 	"fmt"
 	"net/http"
 	"net/url"
 	"path"
+	"strings"
 	"sync"
 	"time"
 
-	"github.com/gorilla/mux"
 	"github.com/rancher/remotedialer"
 	"github.com/sirupsen/logrus"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
@@ -61,8 +62,7 @@ var (
 )
 
 func authorizer(req *http.Request) (string, bool, error) {
-	vars := mux.Vars(req)
-	clusterID := vars["cluster"]
+	clusterID := strings.TrimPrefix(strings.TrimRight(req.URL.Path, "/"), urlPrefix)
 	return clusterID, clusterID != "", nil
 }
 
@@ -93,6 +93,9 @@ func (e *Exchanger) getClonedTransport(clusterID string) *http.Transport {
 
 	dialer := e.dialerServer.Dialer(clusterID)
 	transport = &http.Transport{
+		TLSClientConfig: &tls.Config{
+			InsecureSkipVerify: true,
+		},
 		DialContext: dialer,
 		// apply default settings from http.DefaultTransport
 		MaxIdleConns:          100,
@@ -106,54 +109,46 @@ func (e *Exchanger) getClonedTransport(clusterID string) *http.Transport {
 }
 
 func (e *Exchanger) Connect(ctx context.Context, id string, opts *proxies.Socket, responder rest.Responder) (http.Handler, error) {
-	// serve websocket connection
-	router := mux.NewRouter().PathPrefix(fmt.Sprintf("%s{cluster}", urlPrefix)).Subrouter()
-	router.Handle("", e.dialerServer)
-	router.Handle("/", e.dialerServer)
+	return e.dialerServer, nil
+}
+
+func (e *Exchanger) ProxyConnect(ctx context.Context, id string, opts *proxies.Socket, responder rest.Responder) (http.Handler, error) {
+	// Wait for the caches to be synced before starting workers
+	if !cache.WaitForCacheSync(ctx.Done(), e.mcSynced) {
+		err := apierrors.NewServiceUnavailable("cache for ManagedCluster is not ready yet, please retry later")
+		responder.Error(err)
+		return nil, err
+	}
+
+	location, transport, err := e.ClusterLocation(true, id, opts)
+	if err != nil {
+		responder.Error(err)
+		return nil, err
+	}
 
 	// TODO: add metrics
 
-	// proxy and re-dial through websocket connection
-	router.HandleFunc("/direct{path:.*}",
-		func(writer http.ResponseWriter, request *http.Request) {
-			// Wait for the caches to be synced before starting workers
-			if !cache.WaitForCacheSync(ctx.Done(), e.mcSynced) {
-				responder.Error(apierrors.NewServiceUnavailable("cache for ManagedCluster is not ready yet, please retry later"))
-				return
-			}
-
-			location, transport, err := e.ClusterLocation(true, true, request, id)
-			if err != nil {
-				responder.Error(err)
-				return
-			}
-
-			klog.V(4).Infof("Request to %q will be redialed from cluster %q", location.String(), id)
-			handler := newThrottledUpgradeAwareProxyHandler(location, transport, false, false, true, true, responder)
-			handler.ServeHTTP(writer, request)
-		})
-	router.HandleFunc("/{scheme}/{host}{path:.*}",
-		func(writer http.ResponseWriter, request *http.Request) {
-			location, transport, err := e.ClusterLocation(false, true, request, id)
-			if err != nil {
-				responder.Error(err)
-				return
-			}
-
-			klog.V(4).Infof("Request to %q will be redialed from cluster %q", location.String(), id)
-			handler := newThrottledUpgradeAwareProxyHandler(location, transport, false, false, true, true, responder)
-			handler.ServeHTTP(writer, request)
-		})
-	return router, nil
+	handler := http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
+		location.RawQuery = request.URL.RawQuery
+		// proxy and re-dial through websocket connection
+		klog.V(4).Infof("Request to %q will be redialed from cluster %q", location.String(), id)
+		handler := newThrottledUpgradeAwareProxyHandler(location, transport, false, false, true, true, responder)
+		handler.ServeHTTP(writer, request)
+	})
+
+	return handler, nil
 }
 
-func (e *Exchanger) ClusterLocation(isShortPath, useSocket bool, request *http.Request, id string) (*url.URL, http.RoundTripper, error) {
-	location := new(url.URL)
+func (e *Exchanger) ClusterLocation(useSocket bool, id string, opts *proxies.Socket) (*url.URL, http.RoundTripper, error) {
 	var transport *http.Transport
 
-	vars := mux.Vars(request)
-	location.Path = vars["path"]
-	location.RawQuery = request.URL.RawQuery
+	location := new(url.URL)
+
+	reqPath := strings.TrimLeft(opts.Path, "/")
+	var isShortPath bool
+	if len(reqPath) == 0 {
+		isShortPath = true
+	}
 
 	if isShortPath {
 		mcls, err := e.mcLister.List(labels.SelectorFromSet(labels.Set{
@@ -178,10 +173,19 @@ func (e *Exchanger) ClusterLocation(isShortPath, useSocket bool, request *http.R
 
 		location.Scheme = loc.Scheme
 		location.Host = loc.Host
-		location.Path = path.Join(loc.Path, location.Path)
+		location.Path = loc.Path
 	} else {
-		location.Scheme = vars["scheme"]
-		location.Host = vars["host"]
+		parts := strings.Split(reqPath, "/")
+		if len(parts) == 0 {
+			return nil, nil, fmt.Errorf("unexpected error: invalid request path %s", reqPath)
+		}
+		location.Scheme = parts[0]
+		if len(parts) > 1 {
+			location.Host = parts[1]
+		}
+		if len(parts) > 2 {
+			location.Path = path.Join(parts[2:]...)
+		}
 	}
 
 	// TODO: support https as well
diff --git a/pkg/generated/openapi/zz_generated.openapi.go b/pkg/generated/openapi/zz_generated.openapi.go
index aaa5443..6d73676 100644
--- a/pkg/generated/openapi/zz_generated.openapi.go
+++ b/pkg/generated/openapi/zz_generated.openapi.go
@@ -106,6 +106,13 @@ func schema_pkg_apis_proxies_v1alpha1_Socket(ref common.ReferenceCallback) commo
 							Format:      "",
 						},
 					},
+					"path": {
+						SchemaProps: spec.SchemaProps{
+							Description: "Path is the part of URLs that include service endpoints, suffixes, and parameters to use for the current proxy request to child cluster. For example, the whole request URL is http://localhost:8001/apis/proxies.clusternet.io/v1alpha1/sockets/cae6feb5-a23f-4354-8ee7-66527aa03f54/proxy/https/demo.com:6443. Path is /https/demo.com:6443.\n\nPath is the URL path to use for the current proxy request",
+							Type:        []string{"string"},
+							Format:      "",
+						},
+					},
 				},
 			},
 		},
diff --git a/pkg/hub/apiserver/apiserver.go b/pkg/hub/apiserver/apiserver.go
index 7dd8596..5daca0a 100644
--- a/pkg/hub/apiserver/apiserver.go
+++ b/pkg/hub/apiserver/apiserver.go
@@ -27,8 +27,10 @@ import (
 
 	"github.com/clusternet/clusternet/pkg/apis/proxies"
 	"github.com/clusternet/clusternet/pkg/apis/proxies/install"
+	"github.com/clusternet/clusternet/pkg/exchanger"
 	clusterInformers "github.com/clusternet/clusternet/pkg/generated/informers/externalversions/clusters/v1beta1"
 	socketstorage "github.com/clusternet/clusternet/pkg/registry/proxies/socket"
+	"github.com/clusternet/clusternet/pkg/registry/proxies/socket/subresources"
 )
 
 var (
@@ -113,8 +115,14 @@ func (c completedConfig) New(tunnelLogging, socketConnection bool, mclsInformer
 
 	apiGroupInfo := genericapiserver.NewDefaultAPIGroupInfo(proxies.GroupName, Scheme, ParameterCodec, Codecs)
 
+	var ec *exchanger.Exchanger
+	if socketConnection {
+		ec = exchanger.NewExchanger(tunnelLogging, mclsInformer)
+	}
+
 	v1alpha1storage := map[string]rest.Storage{}
-	v1alpha1storage["sockets"] = socketstorage.NewREST(tunnelLogging, socketConnection, mclsInformer)
+	v1alpha1storage["sockets"] = socketstorage.NewREST(socketConnection, ec)
+	v1alpha1storage["sockets/proxy"] = subresources.NewProxyREST(socketConnection, ec)
 	apiGroupInfo.VersionedResourcesStorageMap["v1alpha1"] = v1alpha1storage
 
 	if err := s.GenericAPIServer.InstallAPIGroup(&apiGroupInfo); err != nil {
diff --git a/pkg/registry/proxies/socket/rest.go b/pkg/registry/proxies/socket/rest.go
index 514ad0a..a50ba40 100644
--- a/pkg/registry/proxies/socket/rest.go
+++ b/pkg/registry/proxies/socket/rest.go
@@ -28,7 +28,6 @@ import (
 	proxies "github.com/clusternet/clusternet/pkg/apis/proxies/v1alpha1"
 	"github.com/clusternet/clusternet/pkg/exchanger"
 	"github.com/clusternet/clusternet/pkg/features"
-	clusterInformers "github.com/clusternet/clusternet/pkg/generated/informers/externalversions/clusters/v1beta1"
 )
 
 const (
@@ -37,7 +36,7 @@ const (
 
 // REST implements a RESTStorage for Proxies API
 type REST struct {
-	exchanger        *exchanger.Exchanger
+	Exchanger        *exchanger.Exchanger
 	socketConnection bool
 }
 
@@ -80,17 +79,13 @@ func (r *REST) Connect(ctx context.Context, id string, opts runtime.Object, resp
 		return nil, fmt.Errorf("invalid options object: %#v", opts)
 	}
 
-	return r.exchanger.Connect(ctx, id, socket, responder)
+	return r.Exchanger.Connect(ctx, id, socket, responder)
 }
 
 // NewREST returns a RESTStorage object that will work against API services.
-func NewREST(tunnelLogging, socketConnection bool, mclsInformer clusterInformers.ManagedClusterInformer) *REST {
-	var ec *exchanger.Exchanger
-	if socketConnection {
-		ec = exchanger.NewExchanger(tunnelLogging, mclsInformer)
-	}
+func NewREST(socketConnection bool, ec *exchanger.Exchanger) *REST {
 	return &REST{
-		exchanger:        ec,
+		Exchanger:        ec,
 		socketConnection: socketConnection,
 	}
 }
diff --git a/pkg/registry/proxies/socket/subresources/proxy.go b/pkg/registry/proxies/socket/subresources/proxy.go
new file mode 100644
index 0000000..3dc8fc4
--- /dev/null
+++ b/pkg/registry/proxies/socket/subresources/proxy.go
@@ -0,0 +1,79 @@
+/*
+Copyright 2021 The Clusternet 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 subresources
+
+import (
+	"context"
+	"fmt"
+	"github.com/clusternet/clusternet/pkg/features"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	"net/http"
+
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apiserver/pkg/registry/rest"
+
+	proxiesapi "github.com/clusternet/clusternet/pkg/apis/proxies/v1alpha1"
+	"github.com/clusternet/clusternet/pkg/exchanger"
+)
+
+// ProxyREST implements the proxy subresource for a Socket
+type ProxyREST struct {
+	Exchanger        *exchanger.Exchanger
+	socketConnection bool
+}
+
+// Implement Connecter
+var _ = rest.Connecter(&ProxyREST{})
+
+var proxyMethods = []string{"GET", "POST", "PUT", "PATCH", "DELETE", "HEAD", "OPTIONS"}
+
+// New returns an empty podProxyOptions object.
+func (r *ProxyREST) New() runtime.Object {
+	return &proxiesapi.Socket{}
+}
+
+// ConnectMethods returns the list of HTTP methods that can be proxied
+func (r *ProxyREST) ConnectMethods() []string {
+	return proxyMethods
+}
+
+// NewConnectOptions returns versioned resource that represents proxy parameters
+func (r *ProxyREST) NewConnectOptions() (runtime.Object, bool, string) {
+	return &proxiesapi.Socket{}, true, "path"
+}
+
+// Connect returns a handler for the pod proxy
+func (r *ProxyREST) Connect(ctx context.Context, id string, opts runtime.Object, responder rest.Responder) (http.Handler, error) {
+	if !r.socketConnection {
+		return nil, apierrors.NewServiceUnavailable(fmt.Sprintf("featuregate %s has not been enabled on the server side", features.SocketConnection))
+	}
+
+	proxyOpts, ok := opts.(*proxiesapi.Socket)
+	if !ok {
+		return nil, fmt.Errorf("invalid options object: %#v", opts)
+	}
+
+	return r.Exchanger.ProxyConnect(ctx, id, proxyOpts, responder)
+}
+
+// NewProxyREST returns a RESTStorage object that will work against API services.
+func NewProxyREST(socketConnection bool, ec *exchanger.Exchanger) *ProxyREST {
+	return &ProxyREST{
+		Exchanger:        ec,
+		socketConnection: socketConnection,
+	}
+}


Test patch:
diff --git a/pkg/exchanger/exchanger_test.go b/pkg/exchanger/exchanger_test.go
new file mode 100644
index 0000000..ee79127
--- /dev/null
+++ b/pkg/exchanger/exchanger_test.go
@@ -0,0 +1,90 @@
+/*
+Copyright 2021 The Clusternet 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 exchanger
+
+import (
+	"net/http"
+	"net/url"
+	"testing"
+)
+
+func TestAuthorizer(t *testing.T) {
+	tests := []struct {
+		name      string
+		req       *http.Request
+		clusterID string
+		authed    bool
+		wantErr   bool
+	}{
+		{
+			name: "normal url",
+			req: &http.Request{
+				URL: &url.URL{
+					Path: "/apis/proxies.clusternet.io/v1alpha1/sockets/9bb775e3-b177-4e77-8685-1aadcb03b0a8",
+				},
+			},
+			clusterID: "9bb775e3-b177-4e77-8685-1aadcb03b0a8",
+			authed:    true,
+		},
+		{
+			name: "normal url with single trailing slash",
+			req: &http.Request{
+				URL: &url.URL{
+					Path: "/apis/proxies.clusternet.io/v1alpha1/sockets/9bb775e3-b177-4e77-8685-1aadcb03b0a8/",
+				},
+			},
+			clusterID: "9bb775e3-b177-4e77-8685-1aadcb03b0a8",
+			authed:    true,
+		},
+		{
+			name: "normal url with multiple trailing slashes",
+			req: &http.Request{
+				URL: &url.URL{
+					Path: "/apis/proxies.clusternet.io/v1alpha1/sockets/9bb775e3-b177-4e77-8685-1aadcb03b0a8////",
+				},
+			},
+			clusterID: "9bb775e3-b177-4e77-8685-1aadcb03b0a8",
+			authed:    true,
+		},
+		{
+			name: "normal url with multiple trailing slashes and query params",
+			req: &http.Request{
+				URL: &url.URL{
+					Path:     "/apis/proxies.clusternet.io/v1alpha1/sockets/9bb775e3-b177-4e77-8685-1aadcb03b0a8///",
+					RawQuery: "abc=def",
+				},
+			},
+			clusterID: "9bb775e3-b177-4e77-8685-1aadcb03b0a8",
+			authed:    true,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got, got1, err := authorizer(tt.req)
+			if (err != nil) != tt.wantErr {
+				t.Errorf("authorizer() error = %v, wantErr %v", err, tt.wantErr)
+				return
+			}
+			if got != tt.clusterID {
+				t.Errorf("authorizer() got = %v, clusterID %v", got, tt.clusterID)
+			}
+			if got1 != tt.authed {
+				t.Errorf("authorizer() got1 = %v, clusterID %v", got1, tt.authed)
+			}
+		})
+	}
+}