feat: support forward requests to leader hub for sharing pool scope m…

…etadata in nodepool.

Signed-off-by: rambohe-ch <[email protected]>

charts/yurthub/templates/yurthub-cfg.yaml

@@ -102,6 +102,7 @@ metadata:
   name: yurt-hub-cfg
   namespace: {{ .Release.Namespace }}
   labels:
+    openyurt.io/configmap-name: yurt-hub-cfg
     {{- include "yurthub.labels" . | nindent 4 }}
 data:
   cache_agents: {{ .Values.cacheAgents | quote }}

charts/yurthub/templates/yurthub-cloud-yurtstaticset.yaml

@@ -36,6 +36,7 @@ spec:
             - --bind-address={{ .Values.yurthubBindingAddr }}
             - --server-addr={{ .Values.kubernetesServerAddr }}
             - --node-name=$(NODE_NAME)
+            - --nodepool-name={{ .Values.nodePoolName }}
             - --bootstrap-file={{ .Values.bootstrapFile }}
             - --working-mode=cloud
             - --namespace={{ .Release.Namespace }}

charts/yurthub/templates/yurthub-yurtstaticset.yaml

@@ -36,6 +36,7 @@ spec:
             - --bind-address={{ .Values.yurthubBindingAddr }}
             - --server-addr={{ .Values.kubernetesServerAddr }}
             - --node-name=$(NODE_NAME)
+            - --nodepool-name={{ .Values.nodePoolName }}
             - --bootstrap-file={{ .Values.bootstrapFile }}
             - --working-mode=edge
             - --namespace={{ .Release.Namespace }}

charts/yurthub/values.yaml

@@ -10,6 +10,7 @@ yurthubBindingAddr: 127.0.0.1
 kubernetesServerAddr: https://127.0.0.1:6443
 bootstrapFile: "/var/lib/yurthub/bootstrap-hub.conf"
 organizations: ""
+nodePoolName: ""
 
 image:
   registry: openyurt

cmd/yurthub/app/config/config.go

@@ -28,6 +28,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/fields"
 	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/util/wait"
 	apiserver "k8s.io/apiserver/pkg/server"
 	"k8s.io/apiserver/pkg/server/dynamiccertificates"
@@ -37,7 +38,6 @@ import (
 	"k8s.io/client-go/informers"
 	coreinformers "k8s.io/client-go/informers/core/v1"
 	"k8s.io/client-go/kubernetes"
-	"k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/cache"
 	"k8s.io/client-go/tools/clientcmd"
 	"k8s.io/klog/v2"
@@ -55,9 +55,8 @@ import (
 	"github.com/openyurtio/openyurt/pkg/yurthub/filter/manager"
 	"github.com/openyurtio/openyurt/pkg/yurthub/kubernetes/meta"
 	"github.com/openyurtio/openyurt/pkg/yurthub/kubernetes/serializer"
-	"github.com/openyurtio/openyurt/pkg/yurthub/multiplexer"
-	"github.com/openyurtio/openyurt/pkg/yurthub/multiplexer/storage"
 	"github.com/openyurtio/openyurt/pkg/yurthub/network"
+	"github.com/openyurtio/openyurt/pkg/yurthub/proxy/remote"
 	"github.com/openyurtio/openyurt/pkg/yurthub/tenant"
 	"github.com/openyurtio/openyurt/pkg/yurthub/transport"
 	"github.com/openyurtio/openyurt/pkg/yurthub/util"
@@ -90,11 +89,15 @@ type YurtHubConfiguration struct {
 	YurtHubProxyServerServing       *apiserver.DeprecatedInsecureServingInfo
 	YurtHubDummyProxyServerServing  *apiserver.DeprecatedInsecureServingInfo
 	YurtHubSecureProxyServerServing *apiserver.SecureServingInfo
+	YurtHubMultiplexerServerServing *apiserver.SecureServingInfo
 	DiskCachePath                   string
-	RequestMultiplexerManager       *multiplexer.MultiplexerManager
 	ConfigManager                   *configuration.Manager
 	TenantManager                   tenant.Interface
 	TransportAndDirectClientManager transport.Interface
+	LoadBalancerForLeaderHub        remote.LoadBalancer
+	PoolScopeResources              []schema.GroupVersionResource
+	PortForMultiplexer              int
+	NodePoolName                    string
 }
 
 // Complete converts *options.YurtHubOptions to *YurtHubConfiguration
@@ -116,7 +119,9 @@ func Complete(options *options.YurtHubOptions, stopCh <-chan struct{}) (*YurtHub
 
 		// list/watch endpoints from host cluster in order to resolve tenant cluster address.
 		newEndpointsInformer := func(client kubernetes.Interface, resyncPeriod time.Duration) cache.SharedIndexInformer {
-			return coreinformers.NewFilteredEndpointsInformer(client, "kube-public", resyncPeriod, nil, nil)
+			informer := coreinformers.NewFilteredEndpointsInformer(client, "kube-public", resyncPeriod, nil, nil)
+			informer.SetTransform(pkgutil.TransformStripManagedFields())
+			return informer
 		}
 		sharedFactory.InformerFor(&corev1.Endpoints{}, newEndpointsInformer)
 		cfg.SharedFactory = sharedFactory
@@ -128,6 +133,9 @@ func Complete(options *options.YurtHubOptions, stopCh <-chan struct{}) (*YurtHub
 		cfg.RemoteServers = us
 		cfg.LBMode = options.LBMode
 		tenantNamespce := util.ParseTenantNsFromOrgs(options.YurtHubCertOrganizations)
+		cfg.PoolScopeResources = options.PoolScopeResources
+		cfg.PortForMultiplexer = options.PortForMultiplexer
+		cfg.NodePoolName = options.NodePoolName
 
 		// prepare some basic configurations as following:
 		// - serializer manager: used for managing serializer for encoding or decoding response from kube-apiserver.
@@ -145,7 +153,7 @@ func Complete(options *options.YurtHubOptions, stopCh <-chan struct{}) (*YurtHub
 		if err != nil {
 			return nil, err
 		}
-		registerInformers(sharedFactory, options.YurtHubNamespace, options.NodeName, (cfg.WorkingMode == util.WorkingModeCloud), tenantNamespce)
+		registerInformers(sharedFactory, options.YurtHubNamespace, options.NodePoolName, options.NodeName, (cfg.WorkingMode == util.WorkingModeCloud), tenantNamespce)
 
 		certMgr, err := certificatemgr.NewYurtHubCertManager(options, us)
 		if err != nil {
@@ -191,7 +199,6 @@ func Complete(options *options.YurtHubOptions, stopCh <-chan struct{}) (*YurtHub
 
 		cfg.ConfigManager = configManager
 		cfg.FilterFinder = filterFinder
-		cfg.RequestMultiplexerManager = newRequestMultiplexerManager(options, restMapperManager)
 
 		if options.EnableDummyIf {
 			klog.V(2).Infof("create dummy network interface %s(%s)", options.HubAgentDummyIfName, options.HubAgentDummyIfIP)
@@ -284,15 +291,17 @@ func createClientAndSharedInformerFactories(serverAddr, nodePoolName string) (ku
 func registerInformers(
 	informerFactory informers.SharedInformerFactory,
 	namespace string,
+	poolName string,
 	nodeName string,
 	enablePodInformer bool,
 	tenantNs string) {
 
-	// configmap informer is used for list/watching yurt-hub-cfg configmap which includes configurations about cache agents and filters.
-	// and is used by approver in filter and cache manager on cloud and edge working mode.
+	// configmap informer is used for list/watching yurt-hub-cfg configmap and leader-hub-{poolName} configmap.
+	// yurt-hub-cfg configmap includes configurations about cache agents and filters which are needed by approver in filter and cache manager on cloud and edge working mode.
+	// leader-hub-{nodePoolName} configmap includes leader election configurations which are used by multiplexer manager.
 	newConfigmapInformer := func(client kubernetes.Interface, resyncPeriod time.Duration) cache.SharedIndexInformer {
 		tweakListOptions := func(options *metav1.ListOptions) {
-			options.FieldSelector = fields.Set{"metadata.name": util.YurthubConfigMapName}.String()
+			options.LabelSelector = fmt.Sprintf("openyurt.io/configmap-name in (%s, %s)", util.YurthubConfigMapName, "leader-hub-"+poolName)
 		}
 		informer := coreinformers.NewFilteredConfigMapInformer(client, namespace, resyncPeriod, nil, tweakListOptions)
 		informer.SetTransform(pkgutil.TransformStripManagedFields())
@@ -385,17 +394,22 @@ func prepareServerServing(options *options.YurtHubOptions, certMgr certificate.Y
 	cfg.YurtHubSecureProxyServerServing.ClientCA = caBundleProvider
 	cfg.YurtHubSecureProxyServerServing.DisableHTTP2 = true
 
-	return nil
-}
-
-func newRequestMultiplexerManager(options *options.YurtHubOptions, restMapperManager *meta.RESTMapperManager) *multiplexer.MultiplexerManager {
-	config := &rest.Config{
-		Host:      fmt.Sprintf("http://%s:%d", options.YurtHubProxyHost, options.YurtHubProxyPort),
-		UserAgent: util.MultiplexerProxyClientUserAgentPrefix + options.NodeName,
+	if err := (&apiserveroptions.SecureServingOptions{
+		BindAddress: net.ParseIP(options.NodeIP),
+		BindPort:    options.PortForMultiplexer,
+		BindNetwork: "tcp",
+		ServerCert: apiserveroptions.GeneratableKeyCert{
+			CertKey: apiserveroptions.CertKey{
+				CertFile: serverCertPath,
+				KeyFile:  serverCertPath,
+			},
+		},
+	}).ApplyTo(&cfg.YurtHubMultiplexerServerServing); err != nil {
+		return err
 	}
-	storageProvider := storage.NewStorageProvider(config)
-
-	return multiplexer.NewRequestMultiplexerManager(storageProvider, restMapperManager, options.PoolScopeResources)
+	cfg.YurtHubMultiplexerServerServing.ClientCA = caBundleProvider
+	cfg.YurtHubMultiplexerServerServing.DisableHTTP2 = true
+	return nil
 }
 
 func ReadinessCheck(cfg *YurtHubConfiguration) error {

cmd/yurthub/app/config/config_test.go

@@ -38,6 +38,7 @@ func TestComplete(t *testing.T) {
 	options.NodeName = "foo"
 	options.EnableDummyIf = false
 	options.HubAgentDummyIfIP = "169.254.2.1"
+	options.NodeIP = "127.0.0.1"
 	cfg, err := Complete(options, nil)
 	if err != nil {
 		t.Errorf("expect no err, but got %v", err)

cmd/yurthub/app/options/options.go

@@ -17,13 +17,15 @@ limitations under the License.
 package options
 
 import (
+	"errors"
 	"fmt"
 	"net"
 	"path/filepath"
 	"time"
 
 	"github.com/spf13/pflag"
 	"k8s.io/apimachinery/pkg/runtime/schema"
+	apinet "k8s.io/apimachinery/pkg/util/net"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/klog/v2"
 	utilnet "k8s.io/utils/net"
@@ -83,6 +85,8 @@ type YurtHubOptions struct {
 	ClientForTest             kubernetes.Interface
 	EnablePoolServiceTopology bool
 	PoolScopeResources        PoolScopeMetadatas
+	PortForMultiplexer        int
+	NodeIP                    string
 }
 
 // NewYurtHubOptions creates a new YurtHubOptions with a default config.
@@ -93,6 +97,7 @@ func NewYurtHubOptions() *YurtHubOptions {
 		YurtHubProxyPort:          util.YurtHubProxyPort,
 		YurtHubPort:               util.YurtHubPort,
 		YurtHubProxySecurePort:    util.YurtHubProxySecurePort,
+		PortForMultiplexer:        util.YurtHubMultiplexerPort,
 		YurtHubNamespace:          util.YurtHubNamespace,
 		GCFrequency:               120,
 		YurtHubCertOrganizations:  make([]string, 0),
@@ -167,6 +172,19 @@ func (options *YurtHubOptions) Validate() error {
 		if len(options.CACertHashes) == 0 && !options.UnsafeSkipCAVerification {
 			return fmt.Errorf("set --discovery-token-unsafe-skip-ca-verification flag as true or pass CACertHashes to continue")
 		}
+
+		if len(options.NodePoolName) == 0 {
+			return errors.New("node-pool-name is empty")
+		}
+
+		if len(options.NodeIP) == 0 {
+			ipAddr, err := apinet.ResolveBindAddress(nil)
+			if err != nil {
+				return fmt.Errorf("couldn't get the node ip, %v", err)
+			}
+			options.NodeIP = ipAddr.String()
+			klog.Infof("node ip is configured as %s", options.NodeIP)
+		}
 	}
 
 	return nil
@@ -179,11 +197,13 @@ func (o *YurtHubOptions) AddFlags(fs *pflag.FlagSet) {
 	fs.StringVar(&o.YurtHubProxyHost, "bind-proxy-address", o.YurtHubProxyHost, "the IP address of YurtHub Proxy Server")
 	fs.IntVar(&o.YurtHubProxyPort, "proxy-port", o.YurtHubProxyPort, "the port on which to proxy HTTP requests to kube-apiserver")
 	fs.IntVar(&o.YurtHubProxySecurePort, "proxy-secure-port", o.YurtHubProxySecurePort, "the port on which to proxy HTTPS requests to kube-apiserver")
+	fs.IntVar(&o.PortForMultiplexer, "multiplexer-port", o.PortForMultiplexer, "the port on which to proxy HTTPS requests to multiplexer in yurthub")
 	fs.StringVar(&o.YurtHubNamespace, "namespace", o.YurtHubNamespace, "the namespace of YurtHub Server")
 	fs.StringVar(&o.ServerAddr, "server-addr", o.ServerAddr, "the address of Kubernetes kube-apiserver, the format is: \"server1,server2,...\"; when yurthub is in local mode, server-addr represents the service address of apiservers, the format is: \"ip:port\".")
 	fs.StringSliceVar(&o.YurtHubCertOrganizations, "hub-cert-organizations", o.YurtHubCertOrganizations, "Organizations that will be added into hub's apiserver client certificate, the format is: certOrg1,certOrg2,...")
 	fs.IntVar(&o.GCFrequency, "gc-frequency", o.GCFrequency, "the frequency to gc cache in storage(unit: minute).")
 	fs.StringVar(&o.NodeName, "node-name", o.NodeName, "the name of node that runs hub agent")
+	fs.StringVar(&o.NodeIP, "node-ip", o.NodeIP, "the same IP address of the node which used by kubelet. if unset, node's default IPv4 address will be used.")
 	fs.StringVar(&o.LBMode, "lb-mode", o.LBMode, "the mode of load balancer to connect remote servers(round-robin, priority)")
 	fs.IntVar(&o.HeartbeatFailedRetry, "heartbeat-failed-retry", o.HeartbeatFailedRetry, "number of heartbeat request retry after having failed.")
 	fs.IntVar(&o.HeartbeatHealthyThreshold, "heartbeat-healthy-threshold", o.HeartbeatHealthyThreshold, "minimum consecutive successes for the heartbeat to be considered healthy after having failed.")

cmd/yurthub/app/options/options_test.go

@@ -38,6 +38,7 @@ func TestNewYurtHubOptions(t *testing.T) {
 		YurtHubProxyPort:          util.YurtHubProxyPort,
 		YurtHubPort:               util.YurtHubPort,
 		YurtHubProxySecurePort:    util.YurtHubProxySecurePort,
+		PortForMultiplexer:        util.YurtHubMultiplexerPort,
 		YurtHubNamespace:          util.YurtHubNamespace,
 		GCFrequency:               120,
 		YurtHubCertOrganizations:  make([]string, 0),
@@ -179,10 +180,11 @@ func TestValidate(t *testing.T) {
 				LBMode:                   "rr",
 				WorkingMode:              "cloud",
 				UnsafeSkipCAVerification: true,
+				NodePoolName:             "foo",
 			},
 			isErr: false,
 		},
-		"normal options with ipv4": {
+		"normal options with ipv6": {
 			options: &YurtHubOptions{
 				NodeName:                 "foo",
 				ServerAddr:               "1.2.3.4:56",
@@ -191,10 +193,11 @@ func TestValidate(t *testing.T) {
 				WorkingMode:              "cloud",
 				UnsafeSkipCAVerification: true,
 				HubAgentDummyIfIP:        "fd00::2:1",
+				NodePoolName:             "foo",
 			},
 			isErr: false,
 		},
-		"normal options with ipv6": {
+		"normal options with ipv4": {
 			options: &YurtHubOptions{
 				NodeName:                 "foo",
 				ServerAddr:               "1.2.3.4:56",
@@ -203,6 +206,7 @@ func TestValidate(t *testing.T) {
 				WorkingMode:              "cloud",
 				UnsafeSkipCAVerification: true,
 				HubAgentDummyIfIP:        "169.254.2.1",
+				NodePoolName:             "foo",
 			},
 			isErr: false,
 		},

cmd/yurthub/app/start.go

@@ -19,9 +19,12 @@ package app
 import (
 	"context"
 	"fmt"
+	"net/url"
+	"time"
 
 	"github.com/spf13/cobra"
 	"github.com/spf13/pflag"
+	"k8s.io/client-go/rest"
 	"k8s.io/component-base/cli/globalflag"
 	"k8s.io/klog/v2"
 
@@ -32,10 +35,15 @@ import (
 	"github.com/openyurtio/openyurt/pkg/yurthub/gc"
 	"github.com/openyurtio/openyurt/pkg/yurthub/healthchecker"
 	"github.com/openyurtio/openyurt/pkg/yurthub/healthchecker/cloudapiserver"
+	"github.com/openyurtio/openyurt/pkg/yurthub/healthchecker/leaderhub"
 	"github.com/openyurtio/openyurt/pkg/yurthub/locallb"
+	"github.com/openyurtio/openyurt/pkg/yurthub/multiplexer"
+	"github.com/openyurtio/openyurt/pkg/yurthub/multiplexer/storage"
 	"github.com/openyurtio/openyurt/pkg/yurthub/proxy"
+	"github.com/openyurtio/openyurt/pkg/yurthub/proxy/remote"
 	"github.com/openyurtio/openyurt/pkg/yurthub/server"
 	"github.com/openyurtio/openyurt/pkg/yurthub/storage/disk"
+	"github.com/openyurtio/openyurt/pkg/yurthub/transport"
 	"github.com/openyurtio/openyurt/pkg/yurthub/util"
 )
 
@@ -137,6 +145,18 @@ func Run(ctx context.Context, cfg *config.YurtHubConfiguration) error {
 			trace++
 		}
 
+		// no leader hub servers for transport manager at startup time.
+		// and don't filter response of request for pool scope metadata from leader hub.
+		transportManagerForLeaderHub, err := transport.NewTransportAndClientManager([]*url.URL{}, 2, cfg.CertManager, ctx.Done())
+		if err != nil {
+			return fmt.Errorf("could not new transport manager for leader hub, %w", err)
+		}
+		healthCheckerForLeaderHub := leaderhub.NewLeaderHubHealthChecker(20*time.Second, nil, ctx.Done())
+		loadBalancerForLeaderHub := remote.NewLoadBalancer("round-robin", []*url.URL{}, cacheManager, transportManagerForLeaderHub, healthCheckerForLeaderHub, nil, ctx.Done())
+
+		cfg.LoadBalancerForLeaderHub = loadBalancerForLeaderHub
+		requestMultiplexerManager := newRequestMultiplexerManager(cfg, healthCheckerForLeaderHub)
+
 		if cfg.NetworkMgr != nil {
 			klog.Infof("%d. start network manager for ensuing dummy interface", trace)
 			cfg.NetworkMgr.Run(ctx.Done())
@@ -153,6 +173,7 @@ func Run(ctx context.Context, cfg *config.YurtHubConfiguration) error {
 			cfg,
 			cacheManager,
 			cloudHealthChecker,
+			requestMultiplexerManager,
 			ctx.Done())
 		if err != nil {
 			return fmt.Errorf("could not create reverse proxy handler, %w", err)
@@ -170,3 +191,15 @@ func Run(ctx context.Context, cfg *config.YurtHubConfiguration) error {
 	klog.Info("hub agent exited")
 	return nil
 }
+
+func newRequestMultiplexerManager(cfg *config.YurtHubConfiguration, healthCheckerForLeaderHub healthchecker.Interface) *multiplexer.MultiplexerManager {
+	insecureHubProxyAddress := cfg.YurtHubProxyServerServing.Listener.Addr().String()
+	klog.Infof("hub insecure proxy address: %s", insecureHubProxyAddress)
+	config := &rest.Config{
+		Host:      fmt.Sprintf("http://%s", insecureHubProxyAddress),
+		UserAgent: util.MultiplexerProxyClientUserAgentPrefix + cfg.NodeName,
+	}
+	storageProvider := storage.NewStorageProvider(config)
+
+	return multiplexer.NewRequestMultiplexerManager(cfg, storageProvider, healthCheckerForLeaderHub)
+}

pkg/node-servant/components/yurthub.go

@@ -51,14 +51,16 @@ const (
 type yurtHubOperator struct {
 	apiServerAddr             string
 	joinToken                 string
+	nodePoolName              string
 	yurthubHealthCheckTimeout time.Duration
 }
 
 // NewYurthubOperator new yurtHubOperator struct
-func NewYurthubOperator(apiServerAddr string, joinToken string, yurthubHealthCheckTimeout time.Duration) *yurtHubOperator {
+func NewYurthubOperator(apiServerAddr string, joinToken, nodePoolName string, yurthubHealthCheckTimeout time.Duration) *yurtHubOperator {
 	return &yurtHubOperator{
 		apiServerAddr:             apiServerAddr,
 		joinToken:                 joinToken,
+		nodePoolName:              nodePoolName,
 		yurthubHealthCheckTimeout: yurthubHealthCheckTimeout,
 	}
 }
@@ -91,6 +93,8 @@ func (op *yurtHubOperator) Install() error {
 	}
 	klog.Infof("yurt-hub.yaml apiServerAddr: %+v", op.apiServerAddr)
 	yssYurtHub := strings.ReplaceAll(string(content), "KUBERNETES_SERVER_ADDRESS", op.apiServerAddr)
+	klog.Infof("yurt-hub.yaml nodePoolName: %s", op.nodePoolName)
+	yssYurtHub = strings.ReplaceAll(string(yssYurtHub), "NODE_POOL_NAME", op.nodePoolName)
 	if err = os.WriteFile(getYurthubYaml(podManifestPath), []byte(yssYurtHub), fileMode); err != nil {
 		return err
 	}

pkg/node-servant/constant.go

@@ -54,7 +54,7 @@ spec:
         - /bin/sh
         - -c
         args:
-        - "/usr/local/bin/entry.sh convert {{if .yurthub_healthcheck_timeout}}--yurthub-healthcheck-timeout={{.yurthub_healthcheck_timeout}} {{end}}--join-token={{.joinToken}}"
+        - "/usr/local/bin/entry.sh convert {{if .yurthub_healthcheck_timeout}}--yurthub-healthcheck-timeout={{.yurthub_healthcheck_timeout}} {{end}}--join-token={{.joinToken}} --nodepool-name={{.nodePoolName}}"
         securityContext:
           privileged: true
         volumeMounts: