Feature: support multi-cluster proxy

.golangci.yaml

@@ -4,6 +4,7 @@ run:
   - ".*.gen.go"
   - pkg/utils/http/reverse_proxy.go
   - pkg/proxy/apiserver/options.go
+  - pkg/proxy/apiserver/common/options.go
   skip-dirs:
   - vendor/
 

pkg/apis/ctrlmesh/constants/constants.go

@@ -19,13 +19,14 @@ package constants
 const (
 	ProxyUserID = 1359
 
-	ProxyMetricsHealthPort = 5441
-	ProxyApiserverPort     = 5443
-	ProxyWebhookPort       = 5445
-	ProxyGRPCPort          = 5447
-	ProxyMetricsPort       = 5449
-	ProxyManagerHealthPort = 5451
-	ProxyGRPCServerPort    = 5453
+	ProxyMetricsHealthPort   = 5441
+	ProxyApiServerPort       = 5443
+	ProxyWebhookPort         = 5445
+	ProxyGRPCPort            = 5447
+	ProxyMetricsPort         = 5449
+	ProxyManagerHealthPort   = 5451
+	ProxyGRPCServerPort      = 5453
+	ProxyRemoteApiServerPort = 5455
 
 	ProxyIptablesPort = 15002
 	PprofListenPort   = 5050
@@ -34,7 +35,7 @@ const (
 
 	ProxyIptablesPortFlag           = "proxy-metrics"
 	ProxyMetricsHealthPortFlag      = "metrics-health-port"
-	ProxyApiserverPortFlag          = "proxy-apiserver-port"
+	ProxyApiServerPortFlag          = "proxy-apiserver-port"
 	ProxyWebhookPortFlag            = "proxy-webhook-port"
 	ProxyGRPCPortFlag               = "grpc-port"
 	ProxyIptablesFlag               = "tport"

pkg/apis/ctrlmesh/http/http.go

@@ -20,4 +20,5 @@ const (
 	HeaderMeshRealEndpoint    = "Mesh-Real-Endpoint"
 	HeaderHttpApiServerPreUrl = "Mesh-Pre-Url-Added"
 	HeaderEscapeMesh          = "Mesh-Escape"
+	HeaderRemoteApiServerHost = "Remote-Api-Server-Host"
 )

pkg/apis/ctrlmesh/rest/types.go

@@ -0,0 +1,32 @@
+/*
+Copyright 2024 The KusionStack 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 rest
+
+const RemoteRegisterPath = "/remote-register"
+
+type ConfigRequest struct {
+	Action      Action   `json:"action"`
+	Kubeconfigs [][]byte `json:"kubeconfig"`
+}
+
+type Action string
+
+const (
+	Add    Action = "add"
+	Delete Action = "delete"
+	Update Action = "update"
+)

pkg/cmd/proxy/main.go

@@ -36,6 +36,7 @@ import (
 	"github.com/KusionStack/controller-mesh/pkg/apis/ctrlmesh/constants"
 	"github.com/KusionStack/controller-mesh/pkg/client"
 	proxyapiserver "github.com/KusionStack/controller-mesh/pkg/proxy/apiserver"
+	"github.com/KusionStack/controller-mesh/pkg/proxy/apiserver/common"
 	proxycache "github.com/KusionStack/controller-mesh/pkg/proxy/cache"
 	"github.com/KusionStack/controller-mesh/pkg/proxy/circuitbreaker"
 	"github.com/KusionStack/controller-mesh/pkg/proxy/faultinjection"
@@ -47,7 +48,7 @@ import (
 
 var (
 	metricsHealthPort  = flag.Int(constants.ProxyMetricsHealthPortFlag, constants.ProxyMetricsHealthPort, "Port to bind 0.0.0.0 and serve metric endpoint/healthz/pprof.")
-	proxyApiserverPort = flag.Int(constants.ProxyApiserverPortFlag, constants.ProxyApiserverPort, "Port to bind localhost and proxy the requests to apiserver.")
+	proxyApiServerPort = flag.Int(constants.ProxyApiServerPortFlag, constants.ProxyApiServerPort, "Port to bind localhost and proxy the requests to apiserver.")
 	proxyWebhookPort   = flag.Int(constants.ProxyWebhookPortFlag, constants.ProxyWebhookPort, "Port to bind 0.0.0.0 and proxy the requests to webhook.")
 
 	leaderElectionName = flag.String(constants.ProxyLeaderElectionNameFlag, "", "The name of leader election.")
@@ -103,19 +104,19 @@ func main() {
 		klog.Fatalf("Failed to start proxy client: %v", err)
 	}
 
-	var stoppedApiserver, stoppedWebhook <-chan struct{}
+	var stoppedApiserver, stoppedRemoteApiserver <-chan struct{}
 
 	// TODO: webhook proxy
 
 	// ApiServer proxy
 	{
-		opts := proxyapiserver.NewOptions()
+		opts := common.NewOptions()
 		opts.Config = rest.CopyConfig(cfg)
 		// Certs generated by proxy-init.sh
 		opts.SecureServingOptions.ServerCert.CertKey.KeyFile = "/var/run/secrets/kubernetes.io/serviceaccount/ctrlmesh/tls.key"
 		opts.SecureServingOptions.ServerCert.CertKey.CertFile = "/var/run/secrets/kubernetes.io/serviceaccount/ctrlmesh/tls.crt"
 		opts.SecureServingOptions.BindAddress = net.ParseIP("127.0.0.1")
-		opts.SecureServingOptions.BindPort = *proxyApiserverPort
+		opts.SecureServingOptions.BindPort = *proxyApiServerPort
 		opts.LeaderElectionName = *leaderElectionName
 		opts.SpecManager = proxyClient.GetSpecManager()
 		opts.BreakerWrapperFunc = breakerMgr.HandlerWrapper()
@@ -136,16 +137,27 @@ func main() {
 		}
 	}
 
+	// Remote ApiServer Proxy
+	{
+		opts := common.NewOptions()
+		opts.SpecManager = proxyClient.GetSpecManager()
+		proxy, err := proxyapiserver.NewRemoteProxy(opts)
+		if err != nil {
+			klog.Fatalf("Failed to new remote apiserver proxy: %v", err)
+		}
+		stoppedRemoteApiserver, err = proxy.Start()
+		if err != nil {
+			klog.Fatalf("Failed to start remote apiserver proxy: %v", err)
+		}
+	}
+
 	{
 		go tproxy.NewTProxy(*proxyIptablePort, faultInjectionMgr, breakerMgr).Start()
 	}
 
 	serveHTTP(ctx, readyHandler)
-	if stoppedWebhook != nil {
-		<-stoppedWebhook
-		klog.Infof("Webhook proxy stopped")
-	}
 	<-stoppedApiserver
+	<-stoppedRemoteApiserver
 	klog.Infof("Apiserver proxy stopped")
 }
 

pkg/proxy/apiserver/injector.go → pkg/proxy/apiserver/common/injector.go

@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package apiserver
+package common
 
 import (
 	"fmt"
@@ -43,7 +43,7 @@ type injector struct {
 	specManager *protomanager.SpecManager
 }
 
-func New(m *protomanager.SpecManager) Injector {
+func NewInjector(m *protomanager.SpecManager) Injector {
 	return &injector{specManager: m}
 }
 

pkg/proxy/apiserver/options.go → pkg/proxy/apiserver/common/options.go

@@ -15,7 +15,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package apiserver
+package common
 
 import (
 	"fmt"

pkg/proxy/apiserver/request.go → pkg/proxy/apiserver/common/request.go

@@ -15,7 +15,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package apiserver
+package common
 
 import (
 	"fmt"

pkg/proxy/apiserver/handler.go → pkg/proxy/apiserver/in_cluster_handler.go

@@ -41,6 +41,7 @@ import (
 	"k8s.io/klog/v2"
 
 	"github.com/KusionStack/controller-mesh/pkg/apis/ctrlmesh/constants"
+	"github.com/KusionStack/controller-mesh/pkg/proxy/apiserver/common"
 	proxyfilters "github.com/KusionStack/controller-mesh/pkg/proxy/filters"
 	"github.com/KusionStack/controller-mesh/pkg/proxy/leaderelection"
 	"github.com/KusionStack/controller-mesh/pkg/utils"
@@ -49,21 +50,20 @@ import (
 )
 
 var (
-	upgradeSubresources = sets.NewString("exec", "attach")
-	enableIpTable       = os.Getenv(constants.EnvIPTable) == "true"
-
+	upgradeSubresources   = sets.NewString("exec", "attach")
+	enableIpTable         = os.Getenv(constants.EnvIPTable) == "true"
 	disableCircuitBreaker = os.Getenv(constants.EnvDisableCircuitBreaker) == "true"
 	enableFaultInjection  = os.Getenv(constants.EnvEnableFaultInjection) == "true"
 )
 
 type Proxy struct {
-	opts           *Options
+	opts           *common.Options
 	inSecureServer *http.Server
 	servingInfo    *server.SecureServingInfo
 	handler        http.Handler
 }
 
-func NewProxy(opts *Options) (*Proxy, error) {
+func NewProxy(opts *common.Options) (*Proxy, error) {
 	var servingInfo *server.SecureServingInfo
 	if enableIpTable {
 		if err := opts.ApplyTo(&servingInfo); err != nil {
@@ -79,7 +79,7 @@ func NewProxy(opts *Options) (*Proxy, error) {
 	inHandler := &handler{
 		cfg:       opts.Config,
 		transport: tp,
-		injector:  New(opts.SpecManager),
+		injector:  common.NewInjector(opts.SpecManager),
 	}
 	if opts.LeaderElectionName != "" {
 		inHandler.electionHandler = leaderelection.New(opts.SpecManager, opts.LeaderElectionName)
@@ -95,7 +95,7 @@ func NewProxy(opts *Options) (*Proxy, error) {
 		handler = opts.FaultInjectionWrapperFunc(handler)
 	}
 	handler = genericfilters.WithWaitGroup(handler, opts.LongRunningFunc, opts.HandlerChainWaitGroup)
-	handler = WithRequestInfo(handler, opts.RequestInfoResolver)
+	handler = common.WithRequestInfo(handler, opts.RequestInfoResolver)
 	handler = proxyfilters.WithPanicRecovery(handler, opts.RequestInfoResolver)
 
 	inSecureServer := &http.Server{
@@ -136,7 +136,7 @@ func (p *Proxy) Start(ctx context.Context) (<-chan struct{}, error) {
 type handler struct {
 	cfg             *rest.Config
 	transport       http.RoundTripper
-	injector        Injector
+	injector        common.Injector
 	electionHandler leaderelection.Handler
 }
 

pkg/proxy/apiserver/remote.go

@@ -0,0 +1,115 @@
+/*
+Copyright 2024 The KusionStack 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 apiserver
+
+import (
+	"bytes"
+	"crypto/sha256"
+	"errors"
+	"fmt"
+	"net/http"
+	"sync"
+
+	"k8s.io/client-go/rest"
+	"k8s.io/client-go/tools/clientcmd"
+)
+
+func NewClusterStore() *ClusterStore {
+	return &ClusterStore{
+		remoteClusters: make(map[string]*Cluster),
+	}
+}
+
+type ClusterStore struct {
+	sync.RWMutex
+	// keyed by api server host
+	remoteClusters map[string]*Cluster
+}
+
+// Cluster defines cluster struct
+type Cluster struct {
+	// Client for accessing the cluster.
+	sync.RWMutex
+	Config        *rest.Config
+	Transport     http.RoundTripper
+	KubeConfigSha [sha256.Size]byte
+}
+
+func (c *ClusterStore) Get(host string) *Cluster {
+	c.RLock()
+	defer c.RUnlock()
+	return c.remoteClusters[host]
+}
+
+func (c *ClusterStore) StoreListOf(kubeConfigs ...[]byte) (err error) {
+	for _, kubeConfig := range kubeConfigs {
+		if localErr := c.Store(kubeConfig); localErr != nil {
+			err = errors.Join(err, localErr)
+		}
+	}
+	return err
+}
+
+func (c *ClusterStore) Store(kubeConfig []byte) error {
+	sha := sha256.Sum256(kubeConfig)
+	cfg, err := DefaultBuildRestConfig(kubeConfig)
+	if err != nil {
+		return err
+	}
+	c.Lock()
+	defer c.Unlock()
+	cluster, ok := c.remoteClusters[cfg.Host]
+	if ok && bytes.Equal(sha[:], cluster.KubeConfigSha[:]) {
+		return nil
+	}
+	tp, err := rest.TransportFor(cfg)
+	if err != nil {
+		return err
+	}
+	c.remoteClusters[cfg.Host] = &Cluster{
+		Config:        cfg,
+		KubeConfigSha: sha,
+		Transport:     tp,
+	}
+	return nil
+}
+
+func (c *ClusterStore) Delete(kubeConfig []byte) error {
+	cfg, err := DefaultBuildRestConfig(kubeConfig)
+	if err != nil {
+		return err
+	}
+	c.Lock()
+	defer c.Unlock()
+	delete(c.remoteClusters, cfg.Host)
+	return nil
+}
+
+func DefaultBuildRestConfig(kubeConfig []byte) (*rest.Config, error) {
+	if len(kubeConfig) == 0 {
+		return nil, fmt.Errorf("kubeconfig is empty")
+	}
+	rawConfig, err := clientcmd.Load(kubeConfig)
+	if err != nil {
+		return nil, fmt.Errorf("kubeconfig cannot be loaded: %v", err)
+	}
+	if err = clientcmd.Validate(*rawConfig); err != nil {
+		return nil, fmt.Errorf("kubeconfig is not valid: %v", err)
+	}
+	clientConfig := clientcmd.NewDefaultClientConfig(*rawConfig, &clientcmd.ConfigOverrides{})
+	return clientConfig.ClientConfig()
+}