one cluster should have only one transport to reduce the number of TC…

…P connections

Signed-off-by: mengying <[email protected]>

go.mod

@@ -27,6 +27,7 @@ require (
 	github.com/yuin/gopher-lua v0.0.0-20220504180219-658193537a64
 	go.uber.org/mock v0.4.0
 	golang.org/x/net v0.24.0
+	golang.org/x/sync v0.7.0
 	golang.org/x/term v0.19.0
 	golang.org/x/text v0.14.0
 	golang.org/x/time v0.5.0
@@ -174,7 +175,6 @@ require (
 	golang.org/x/exp v0.0.0-20231226003508-02704c960a9b // indirect
 	golang.org/x/mod v0.17.0 // indirect
 	golang.org/x/oauth2 v0.18.0 // indirect
-	golang.org/x/sync v0.7.0 // indirect
 	golang.org/x/sys v0.19.0 // indirect
 	google.golang.org/appengine v1.6.8 // indirect
 	google.golang.org/genproto v0.0.0-20231212172506-995d672761c0 // indirect

pkg/util/proxy/proxy.go

@@ -18,16 +18,21 @@ package proxy
 
 import (
 	"context"
+	"crypto/md5"
 	"crypto/tls"
 	"crypto/x509"
+	"encoding/hex"
 	"errors"
 	"fmt"
 	"net/http"
 	"net/url"
 	"path"
+	"sort"
 	"strings"
+	"sync"
 	"time"
 
+	"golang.org/x/sync/singleflight"
 	authenticationv1 "k8s.io/api/authentication/v1"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/util/httpstream"
@@ -44,6 +49,13 @@ import (
 // SecretGetterFunc is a function to get secret.
 type SecretGetterFunc func(context.Context, string, string) (*corev1.Secret, error)
 
+type clusterEndpointInfo struct {
+	Transport *http.Transport
+}
+
+var clusterEndpointInfoStore sync.Map
+var singleExecution singleflight.Group
+
 // ConnectCluster returns a handler for proxy cluster.
 func ConnectCluster(ctx context.Context, cluster *clusterapis.Cluster, proxyPath string, secretGetter SecretGetterFunc, responder registryrest.Responder) (http.Handler, error) {
 	tlsConfig, err := GetTLSConfigForCluster(ctx, cluster, secretGetter)
@@ -159,12 +171,33 @@ func Location(cluster *clusterapis.Cluster, tlsConfig *tls.Config) (*url.URL, ht
 		return nil, nil, err
 	}
 
-	proxyTransport, err := createProxyTransport(cluster, tlsConfig)
+	clusterHash := generateMd5HashForCluster(cluster)
+	if v, ok := clusterEndpointInfoStore.Load(clusterHash); ok {
+		clusterEndpointsInfo := v.(clusterEndpointInfo)
+		return location, clusterEndpointsInfo.Transport, nil
+	}
+
+	endpointInfo, err, _ := singleExecution.Do(clusterHash, func() (interface{}, error) {
+		if value, ok := clusterEndpointInfoStore.Load(clusterHash); ok {
+			return value, nil
+		}
+		proxyTransport, err := createProxyTransport(cluster, tlsConfig)
+		if err != nil {
+			return nil, err
+		}
+		clusterEndpoint := clusterEndpointInfo{
+			Transport: proxyTransport,
+		}
+		clusterEndpointInfoStore.Store(clusterHash, clusterEndpointInfo{
+			Transport: proxyTransport,
+		})
+		return clusterEndpoint, nil
+	})
 	if err != nil {
 		return nil, nil, err
 	}
-
-	return location, proxyTransport, nil
+	transport := endpointInfo.(clusterEndpointInfo).Transport
+	return location, transport, nil
 }
 
 func constructLocation(cluster *clusterapis.Cluster) (*url.URL, error) {
@@ -238,3 +271,18 @@ func SkipGroup(group string) bool {
 		return false
 	}
 }
+
+func generateMd5HashForCluster(cluster *clusterapis.Cluster) string {
+	usedFields := make([]string, 0)
+	proxyHeaderKeys := make([]string, 0, len(cluster.Spec.ProxyHeader))
+	for key, _ := range cluster.Spec.ProxyHeader {
+		proxyHeaderKeys = append(proxyHeaderKeys, key)
+	}
+	sort.Strings(proxyHeaderKeys)
+	for _, key := range proxyHeaderKeys {
+		usedFields = append(usedFields, key, cluster.Spec.ProxyHeader[key])
+	}
+	usedFields = append(usedFields, string(cluster.UID), cluster.Spec.ProxyURL, cluster.Spec.APIEndpoint)
+	hash := md5.Sum([]byte(strings.Join(usedFields, "")))
+	return hex.EncodeToString(hash[:])
+}

pkg/util/proxy/proxy_test.go

@@ -27,6 +27,7 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/uuid"
 	"k8s.io/apiserver/pkg/authentication/user"
 	"k8s.io/apiserver/pkg/endpoints/request"
 
@@ -68,7 +69,7 @@ func TestConnectCluster(t *testing.T) {
 			name: "apiEndpoint is empty",
 			args: args{
 				cluster: &clusterapis.Cluster{
-					ObjectMeta: metav1.ObjectMeta{Namespace: "ns", Name: "cluster"},
+					ObjectMeta: metav1.ObjectMeta{Namespace: "ns", Name: "cluster", UID: uuid.NewUUID()},
 					Spec:       clusterapis.ClusterSpec{},
 				},
 				secretGetter: nil,
@@ -80,7 +81,7 @@ func TestConnectCluster(t *testing.T) {
 			name: "apiEndpoint is invalid",
 			args: args{
 				cluster: &clusterapis.Cluster{
-					ObjectMeta: metav1.ObjectMeta{Namespace: "ns", Name: "cluster"},
+					ObjectMeta: metav1.ObjectMeta{Namespace: "ns", Name: "cluster", UID: uuid.NewUUID()},
 					Spec:       clusterapis.ClusterSpec{APIEndpoint: "h :/ invalid"},
 				},
 				secretGetter: nil,
@@ -92,7 +93,7 @@ func TestConnectCluster(t *testing.T) {
 			name: "ProxyURL is invalid",
 			args: args{
 				cluster: &clusterapis.Cluster{
-					ObjectMeta: metav1.ObjectMeta{Namespace: "ns", Name: "cluster"},
+					ObjectMeta: metav1.ObjectMeta{Namespace: "ns", Name: "cluster", UID: uuid.NewUUID()},
 					Spec: clusterapis.ClusterSpec{
 						APIEndpoint: s.URL,
 						ProxyURL:    "h :/ invalid",
@@ -107,7 +108,7 @@ func TestConnectCluster(t *testing.T) {
 			name: "ImpersonatorSecretRef is nil",
 			args: args{
 				cluster: &clusterapis.Cluster{
-					ObjectMeta: metav1.ObjectMeta{Namespace: "ns", Name: "cluster"},
+					ObjectMeta: metav1.ObjectMeta{Namespace: "ns", Name: "cluster", UID: uuid.NewUUID()},
 					Spec: clusterapis.ClusterSpec{
 						APIEndpoint: s.URL,
 						ProxyURL:    "http://proxy",
@@ -122,7 +123,7 @@ func TestConnectCluster(t *testing.T) {
 			name: "secret not found",
 			args: args{
 				cluster: &clusterapis.Cluster{
-					ObjectMeta: metav1.ObjectMeta{Namespace: "ns", Name: "cluster"},
+					ObjectMeta: metav1.ObjectMeta{Namespace: "ns", Name: "cluster", UID: uuid.NewUUID()},
 					Spec: clusterapis.ClusterSpec{
 						APIEndpoint:           s.URL,
 						ImpersonatorSecretRef: &clusterapis.LocalSecretReference{Namespace: "ns", Name: "secret"},
@@ -139,7 +140,7 @@ func TestConnectCluster(t *testing.T) {
 			name: "SecretTokenKey not found",
 			args: args{
 				cluster: &clusterapis.Cluster{
-					ObjectMeta: metav1.ObjectMeta{Namespace: "ns", Name: "cluster"},
+					ObjectMeta: metav1.ObjectMeta{Namespace: "ns", Name: "cluster", UID: uuid.NewUUID()},
 					Spec: clusterapis.ClusterSpec{
 						APIEndpoint:           s.URL,
 						ImpersonatorSecretRef: &clusterapis.LocalSecretReference{Namespace: "ns", Name: "secret"},
@@ -160,7 +161,7 @@ func TestConnectCluster(t *testing.T) {
 			args: args{
 				ctx: context.TODO(),
 				cluster: &clusterapis.Cluster{
-					ObjectMeta: metav1.ObjectMeta{Namespace: "ns", Name: "cluster"},
+					ObjectMeta: metav1.ObjectMeta{Namespace: "ns", Name: "cluster", UID: uuid.NewUUID()},
 					Spec: clusterapis.ClusterSpec{
 						APIEndpoint:           s.URL,
 						ImpersonatorSecretRef: &clusterapis.LocalSecretReference{Namespace: "ns", Name: "secret"},
@@ -181,7 +182,7 @@ func TestConnectCluster(t *testing.T) {
 			args: args{
 				ctx: request.WithUser(request.NewContext(), &user.DefaultInfo{Name: testUser, Groups: []string{testGroup, user.AllAuthenticated, user.AllUnauthenticated}}),
 				cluster: &clusterapis.Cluster{
-					ObjectMeta: metav1.ObjectMeta{Namespace: "ns", Name: "cluster"},
+					ObjectMeta: metav1.ObjectMeta{Namespace: "ns", Name: "cluster", UID: uuid.NewUUID()},
 					Spec: clusterapis.ClusterSpec{
 						APIEndpoint:                 s.URL,
 						ImpersonatorSecretRef:       &clusterapis.LocalSecretReference{Namespace: "ns", Name: "secret"},