feat: improve config and start process of yurthub component

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

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

@@ -58,8 +58,8 @@ spec:
               host: {{ .Values.yurthubBindingAddr }}
               path: /v1/readyz
               port: 10267
-            initialDelaySeconds: 120
-            periodSeconds: 20
+            initialDelaySeconds: 30
+            periodSeconds: 15
             failureThreshold: 3
           resources:
             requests:

charts/yurthub/templates/yurthub-yurtstaticset.yaml

@@ -58,8 +58,8 @@ spec:
               host: {{ .Values.yurthubBindingAddr }}
               path: /v1/readyz
               port: 10267
-            initialDelaySeconds: 120
-            periodSeconds: 20
+            initialDelaySeconds: 30
+            periodSeconds: 15
             failureThreshold: 3
           resources:
             requests:

cmd/yurthub/app/config/config_test.go

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

cmd/yurthub/app/options/options.go

@@ -136,7 +136,16 @@ func (options *YurtHubOptions) Validate() error {
 		return fmt.Errorf("server-address is empty")
 	}
 
-	if options.WorkingMode != string(util.WorkingModeLocal) {
+	if !util.IsSupportedWorkingMode(util.WorkingMode(options.WorkingMode)) {
+		return fmt.Errorf("working mode %s is not supported", options.WorkingMode)
+	}
+
+	switch options.WorkingMode {
+	case string(util.WorkingModeLocal):
+		if len(options.HostControlPlaneAddr) == 0 {
+			return fmt.Errorf("host-control-plane-address is empty")
+		}
+	default:
 		if options.BootstrapMode != certificate.KubeletCertificateBootstrapMode {
 			if len(options.JoinToken) == 0 && len(options.BootstrapFile) == 0 {
 				return fmt.Errorf("bootstrap token and bootstrap file are empty, one of them must be set")
@@ -147,10 +156,6 @@ func (options *YurtHubOptions) Validate() error {
 			return fmt.Errorf("lb mode(%s) is not supported", options.LBMode)
 		}
 
-		if !util.IsSupportedWorkingMode(util.WorkingMode(options.WorkingMode)) {
-			return fmt.Errorf("working mode %s is not supported", options.WorkingMode)
-		}
-
 		if err := options.verifyDummyIP(); err != nil {
 			return fmt.Errorf("dummy ip %s is not invalid, %w", options.HubAgentDummyIfIP, err)
 		}
@@ -162,10 +167,6 @@ 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")
 		}
-	} else {
-		if len(options.HostControlPlaneAddr) == 0 {
-			return fmt.Errorf("host-control-plane-address is empty")
-		}
 	}
 
 	return nil

cmd/yurthub/app/start.go

@@ -19,13 +19,9 @@ package app
 import (
 	"context"
 	"fmt"
-	"net/url"
-	"time"
 
 	"github.com/spf13/cobra"
 	"github.com/spf13/pflag"
-	"k8s.io/client-go/kubernetes"
-	"k8s.io/client-go/rest"
 	"k8s.io/component-base/cli/globalflag"
 	"k8s.io/klog/v2"
 
@@ -35,12 +31,10 @@ import (
 	"github.com/openyurtio/openyurt/pkg/yurthub/cachemanager"
 	"github.com/openyurtio/openyurt/pkg/yurthub/gc"
 	"github.com/openyurtio/openyurt/pkg/yurthub/healthchecker"
-	"github.com/openyurtio/openyurt/pkg/yurthub/kubernetes/directclient"
 	"github.com/openyurtio/openyurt/pkg/yurthub/locallb"
 	"github.com/openyurtio/openyurt/pkg/yurthub/proxy"
 	"github.com/openyurtio/openyurt/pkg/yurthub/server"
-	"github.com/openyurtio/openyurt/pkg/yurthub/tenant"
-	"github.com/openyurtio/openyurt/pkg/yurthub/transport"
+	"github.com/openyurtio/openyurt/pkg/yurthub/storage/disk"
 	"github.com/openyurtio/openyurt/pkg/yurthub/util"
 )
 
@@ -67,7 +61,7 @@ func NewCmdStartYurtHub(ctx context.Context) *cobra.Command {
 				klog.Fatalf("validate options: %v", err)
 			}
 
-			yurtHubCfg, err := config.Complete(yurtHubOptions)
+			yurtHubCfg, err := config.Complete(yurtHubOptions, ctx.Done())
 			if err != nil {
 				klog.Fatalf("complete %s configuration error, %v", projectinfo.GetHubName(), err)
 			}
@@ -89,127 +83,89 @@ func NewCmdStartYurtHub(ctx context.Context) *cobra.Command {
 
 // Run runs the YurtHubConfiguration. This should never exit
 func Run(ctx context.Context, cfg *config.YurtHubConfiguration) error {
-	if cfg.WorkingMode != util.WorkingModeLocal {
-		defer cfg.CertManager.Stop()
-		trace := 1
-		klog.Infof("%d. new transport manager", trace)
-		transportManager, err := transport.NewTransportManager(cfg.CertManager, ctx.Done())
-		if err != nil {
-			return fmt.Errorf("could not new transport manager, %w", err)
-		}
-		trace++
+	klog.Infof("%s works in %s mode", projectinfo.GetHubName(), string(cfg.WorkingMode))
 
-		klog.Infof("%d. prepare cloud kube clients", trace)
-		cloudClients, err := createClients(cfg.HeartbeatTimeoutSeconds, cfg.RemoteServers, transportManager)
+	switch cfg.WorkingMode {
+	case util.WorkingModeLocal:
+		klog.Infof("new locallb manager for node %s ", cfg.NodeName)
+		locallbMgr, err := locallb.NewLocalLBManager(cfg.TenantKasService, cfg.SharedFactory)
+		// when local mode yurthub exits, we need to clean configured iptables
+		defer locallbMgr.CleanIptables()
 		if err != nil {
-			return fmt.Errorf("could not create cloud clients, %w", err)
+			return fmt.Errorf("could not new locallb manager, %w", err)
 		}
-		trace++
+		// Start the informer factory if all informers have been registered
+		cfg.SharedFactory.Start(ctx.Done())
 
+	case util.WorkingModeCloud, util.WorkingModeEdge:
+		defer cfg.CertManager.Stop()
+		trace := 1
+		// compare cloud working mode, edge working mode need following preparations:
+		// 1. health checker: periodically check the health status of cloud kube-apiserver
+		// 1. cache manager: used for caching response on local disk.
+		// 2. gc: used for garbaging collect unused cache on local disk.
 		var cloudHealthChecker healthchecker.MultipleBackendsHealthChecker
+		var storageWrapper cachemanager.StorageWrapper
+		var cacheManager cachemanager.CacheManager
+		var err error
 		if cfg.WorkingMode == util.WorkingModeEdge {
 			klog.Infof("%d. create health checkers for remote servers", trace)
-			cloudHealthChecker, err = healthchecker.NewCloudAPIServerHealthChecker(cfg, cloudClients, ctx.Done())
+			cloudHealthChecker, err = healthchecker.NewCloudAPIServerHealthChecker(cfg, ctx.Done())
 			if err != nil {
-				return fmt.Errorf("could not new cloud health checker, %w", err)
+				return fmt.Errorf("could not new health checker for cloud kube-apiserver, %w", err)
 			}
-		} else {
-			klog.Infof("%d. disable health checker for node %s because it is a cloud node", trace, cfg.NodeName)
-			// In cloud mode, cloud health checker is not needed.
-			// This fake checker will always report that the cloud is healthy is unhealthy.
-			cloudHealthChecker = healthchecker.NewFakeChecker(true, make(map[string]int))
-		}
-		trace++
-
-		klog.Infof("%d. new direct client manager", trace)
-		directClientManager, err := directclient.NewRestClientManager(cfg.RemoteServers, transportManager, cloudHealthChecker)
-		if err != nil {
-			return fmt.Errorf("could not new restConfig manager, %w", err)
-		}
-		trace++
+			trace++
 
-		var cacheMgr cachemanager.CacheManager
-		if cfg.WorkingMode == util.WorkingModeEdge {
 			klog.Infof("%d. new cache manager with storage wrapper and serializer manager", trace)
-			cacheMgr = cachemanager.NewCacheManager(cfg.StorageWrapper, cfg.SerializerManager, cfg.RESTMapperManager, cfg.ConfigManager)
-		} else {
-			klog.Infof("%d. disable cache manager for node %s because it is a cloud node", trace, cfg.NodeName)
-		}
-		trace++
+			storageManager, err := disk.NewDiskStorage(cfg.DiskCachePath)
+			if err != nil {
+				klog.Errorf("could not create storage manager, %v", err)
+				return err
+			}
+			storageWrapper = cachemanager.NewStorageWrapper(storageManager)
+			cacheManager = cachemanager.NewCacheManager(storageWrapper, cfg.SerializerManager, cfg.RESTMapperManager, cfg.ConfigManager)
+			cfg.StorageWrapper = storageWrapper
+			trace++
 
-		if cfg.WorkingMode == util.WorkingModeEdge {
 			klog.Infof("%d. new gc manager for node %s, and gc frequency is a random time between %d min and %d min", trace, cfg.NodeName, cfg.GCFrequency, 3*cfg.GCFrequency)
-			gcMgr, err := gc.NewGCManager(cfg, directClientManager, ctx.Done())
+			gcMgr, err := gc.NewGCManager(cfg, cloudHealthChecker, ctx.Done())
 			if err != nil {
 				return fmt.Errorf("could not new gc manager, %w", err)
 			}
 			gcMgr.Run()
-		} else {
-			klog.Infof("%d. disable gc manager for node %s because it is a cloud node", trace, cfg.NodeName)
+			trace++
 		}
-		trace++
 
-		klog.Infof("%d. new tenant sa manager", trace)
-		tenantMgr := tenant.New(cfg.TenantNs, cfg.SharedFactory, ctx.Done())
-		trace++
+		if cfg.NetworkMgr != nil {
+			klog.Infof("%d. start network manager for ensuing dummy interface", trace)
+			cfg.NetworkMgr.Run(ctx.Done())
+			trace++
+		}
 
 		// Start the informer factory if all informers have been registered
 		cfg.SharedFactory.Start(ctx.Done())
-		cfg.NodePoolInformerFactory.Start(ctx.Done())
+		cfg.DynamicSharedFactory.Start(ctx.Done())
 
-		klog.Infof("%d. new reverse proxy handler for remote servers", trace)
+		// Start to prepare proxy handler and start server serving.
+		klog.Infof("%d. new reverse proxy handler for forwarding requests", trace)
 		yurtProxyHandler, err := proxy.NewYurtReverseProxyHandler(
 			cfg,
-			cacheMgr,
-			directClientManager,
-			transportManager,
+			cacheManager,
 			cloudHealthChecker,
-			tenantMgr,
 			ctx.Done())
 		if err != nil {
 			return fmt.Errorf("could not create reverse proxy handler, %w", err)
 		}
 		trace++
 
-		if cfg.NetworkMgr != nil {
-			cfg.NetworkMgr.Run(ctx.Done())
-		}
-
 		klog.Infof("%d. new %s server and begin to serve", trace, projectinfo.GetHubName())
-		if err := server.RunYurtHubServers(cfg, yurtProxyHandler, directClientManager, ctx.Done()); err != nil {
+		if err := server.RunYurtHubServers(cfg, yurtProxyHandler, cloudHealthChecker, ctx.Done()); err != nil {
 			return fmt.Errorf("could not run hub servers, %w", err)
 		}
-	} else {
-		klog.Infof("new locallb manager for node %s ", cfg.NodeName)
-		locallbMgr, err := locallb.NewLocalLBManager(cfg, cfg.SharedFactory)
-		// when local mode yurthub exits, we need to clean configured iptables
-		defer locallbMgr.CleanIptables()
-		if err != nil {
-			return fmt.Errorf("could not new locallb manager, %w", err)
-		}
-		// Start the informer factory if all informers have been registered
-		cfg.SharedFactory.Start(ctx.Done())
+	default:
+
 	}
 	<-ctx.Done()
 	klog.Info("hub agent exited")
 	return nil
 }
-
-// createClients will create clients for all cloud APIServer
-// It will return a map, mapping cloud APIServer URL to its client
-func createClients(heartbeatTimeoutSeconds int, remoteServers []*url.URL, tp transport.Interface) (map[string]kubernetes.Interface, error) {
-	cloudClients := make(map[string]kubernetes.Interface)
-	for i := range remoteServers {
-		restConf := &rest.Config{
-			Host:      remoteServers[i].String(),
-			Transport: tp.CurrentTransport(),
-			Timeout:   time.Duration(heartbeatTimeoutSeconds) * time.Second,
-		}
-		c, err := kubernetes.NewForConfig(restConf)
-		if err != nil {
-			return cloudClients, err
-		}
-		cloudClients[remoteServers[i].String()] = c
-	}
-	return cloudClients, nil
-}

go.mod

@@ -128,7 +128,6 @@ require (
 	github.com/rivo/uniseg v0.2.0 // indirect
 	github.com/russross/blackfriday/v2 v2.1.0 // indirect
 	github.com/stoewer/go-strcase v1.3.0 // indirect
-	github.com/stretchr/objx v0.5.2 // indirect
 	github.com/vishvananda/netns v0.0.4 // indirect
 	github.com/x448/float16 v0.8.4 // indirect
 	github.com/xlab/treeprint v1.2.0 // indirect

pkg/yurthub/cachemanager/cache_manager.go

@@ -168,6 +168,7 @@ func (cm *cacheManager) queryListObject(req *http.Request) (runtime.Object, erro
 	var listGvk schema.GroupVersionKind
 	convertGVK, ok := util.ConvertGVKFrom(ctx)
 	if ok && convertGVK != nil {
+		// partial object metadata request
 		listGvk = schema.GroupVersionKind{
 			Group:   convertGVK.Group,
 			Version: convertGVK.Version,
@@ -204,6 +205,7 @@ func (cm *cacheManager) queryListObject(req *http.Request) (runtime.Object, erro
 	if err != nil {
 		return nil, err
 	}
+
 	objs, err := cm.storage.List(key)
 	if err == storage.ErrStorageNotFound && isListRequestWithNameFieldSelector(req) {
 		// When the request is a list request with FieldSelector "metadata.name", we should not return error

pkg/yurthub/gc/gc.go

@@ -31,8 +31,9 @@ import (
 
 	"github.com/openyurtio/openyurt/cmd/yurthub/app/config"
 	"github.com/openyurtio/openyurt/pkg/yurthub/cachemanager"
-	"github.com/openyurtio/openyurt/pkg/yurthub/kubernetes/directclient"
+	"github.com/openyurtio/openyurt/pkg/yurthub/healthchecker"
 	"github.com/openyurtio/openyurt/pkg/yurthub/storage"
+	"github.com/openyurtio/openyurt/pkg/yurthub/transport"
 	"github.com/openyurtio/openyurt/pkg/yurthub/util"
 )
 
@@ -43,15 +44,16 @@ var (
 // GCManager is responsible for cleanup garbage of yurthub
 type GCManager struct {
 	store             cachemanager.StorageWrapper
-	manager           *directclient.DirectClientManager
+	healthChecker     healthchecker.MultipleBackendsHealthChecker
+	clientManager     transport.Interface
 	nodeName          string
 	eventsGCFrequency time.Duration
 	lastTime          time.Time
 	stopCh            <-chan struct{}
 }
 
 // NewGCManager creates a *GCManager object
-func NewGCManager(cfg *config.YurtHubConfiguration, directClientManager *directclient.DirectClientManager, stopCh <-chan struct{}) (*GCManager, error) {
+func NewGCManager(cfg *config.YurtHubConfiguration, healthChecker healthchecker.MultipleBackendsHealthChecker, stopCh <-chan struct{}) (*GCManager, error) {
 	gcFrequency := cfg.GCFrequency
 	if gcFrequency == 0 {
 		gcFrequency = defaultEventGcInterval
@@ -60,7 +62,8 @@ func NewGCManager(cfg *config.YurtHubConfiguration, directClientManager *directc
 		// TODO: use disk storage directly
 		store:             cfg.StorageWrapper,
 		nodeName:          cfg.NodeName,
-		manager:           directClientManager,
+		healthChecker:     healthChecker,
+		clientManager:     cfg.TransportAndDirectClientManager,
 		eventsGCFrequency: time.Duration(gcFrequency) * time.Minute,
 		stopCh:            stopCh,
 	}
@@ -75,11 +78,16 @@ func (m *GCManager) Run() {
 	go wait.JitterUntil(func() {
 		klog.V(2).Infof("start gc events after waiting %v from previous gc", time.Since(m.lastTime))
 		m.lastTime = time.Now()
-		kubeClient := m.manager.GetDirectClientset(true)
-		if kubeClient == nil {
+		u, err := m.healthChecker.PickHealthyServer()
+		if err != nil || u == nil {
 			klog.Warningf("all remote servers are unhealthy, skip gc events")
 			return
 		}
+		kubeClient := m.clientManager.GetDirectClientset(u)
+		if kubeClient == nil {
+			klog.Warningf("couldn't get direct clientset for server %s, skip gc events", u.String())
+			return
+		}
 
 		m.gcEvents(kubeClient, "kubelet")
 		m.gcEvents(kubeClient, "kube-proxy")
@@ -105,11 +113,17 @@ func (m *GCManager) gcPodsWhenRestart() {
 		return
 	}
 
-	kubeClient := m.manager.GetDirectClientset(true)
-	if kubeClient == nil {
+	// get a clientset of a healthy kube-apiserver
+	u, err := m.healthChecker.PickHealthyServer()
+	if err != nil || u == nil {
 		klog.Warningf("all remote servers are unhealthy, skip gc pods")
 		return
 	}
+	kubeClient := m.clientManager.GetDirectClientset(u)
+	if kubeClient == nil {
+		klog.Warningf("couldn't get direct clientset for server %s, skip gc pods", u.String())
+		return
+	}
 
 	listOpts := metav1.ListOptions{FieldSelector: fields.OneTermEqualSelector("spec.nodeName", m.nodeName).String()}
 	podList, err := kubeClient.CoreV1().Pods(v1.NamespaceAll).List(context.Background(), listOpts)

pkg/yurthub/healthchecker/fake_checker.go

@@ -49,7 +49,6 @@ func (fc *fakeChecker) IsHealthy() bool {
 }
 
 func (fc *fakeChecker) RenewKubeletLeaseTime() {
-	return
 }
 
 func (fc *fakeChecker) PickHealthyServer() (*url.URL, error) {