yurthub
├── cachemanager  # cache 管理器，
├── certificate # 证书token管理
├── filter 
├── gc # GCManager
├── healthchecker # cloud apiserver 探火机制
├── kubernetes # 
├── metrics
├── network # 网络iptables配置 
├── otaupdate
├── poolcoordinator
├── proxy # 核心代码，反向代理机制，包括remote proxy和local proxy
├── server # yurthub server
├── storage # 本地存储的实现
├── tenant
├── transport
└── util

func main() {
    newRand := rand.New(rand.NewSource(time.Now().UnixNano()))
    newRand.Seed(time.Now().UnixNano())

    cmd := app.NewCmdStartYurtHub(server.SetupSignalContext())
    cmd.Flags().AddGoFlagSet(flag.CommandLine)
    if err := cmd.Execute(); err != nil {
        panic(err)
    }
}

// NewCmdStartYurtHub creates a *cobra.Command object with default parameters
func NewCmdStartYurtHub(ctx context.Context) *cobra.Command {
    yurtHubOptions := options.NewYurtHubOptions()

    cmd := &cobra.Command{
        Use:   projectinfo.GetHubName(),
        Short: "Launch " + projectinfo.GetHubName(),
        Long:  "Launch " + projectinfo.GetHubName(),
        Run: func(cmd *cobra.Command, args []string) {
            if yurtHubOptions.Version {
                fmt.Printf("%s: %#v\n", projectinfo.GetHubName(), projectinfo.Get())
                return
            }
            fmt.Printf("%s version: %#v\n", projectinfo.GetHubName(), projectinfo.Get())

            cmd.Flags().VisitAll(func(flag *pflag.Flag) {
                klog.V(1).Infof("FLAG: --%s=%q", flag.Name, flag.Value)
            })
            if err := yurtHubOptions.Validate(); err != nil {
                klog.Fatalf("validate options: %v", err)
            }

            yurtHubCfg, err := config.Complete(yurtHubOptions)
            if err != nil {
                klog.Fatalf("complete %s configuration error, %v", projectinfo.GetHubName(), err)
            }
            klog.Infof("%s cfg: %#+v", projectinfo.GetHubName(), yurtHubCfg)

            if err := Run(ctx, yurtHubCfg); err != nil {
                klog.Fatalf("run %s failed, %v", projectinfo.GetHubName(), err)
            }
        },
    }

    yurtHubOptions.AddFlags(cmd.Flags())
    return cmd
}

// Run runs the YurtHubConfiguration. This should never exit
func Run(ctx context.Context, cfg *config.YurtHubConfiguration) error {
    defer cfg.CertManager.Stop()
    trace := 1
    klog.Infof("%d. new transport manager", trace)
    // 构造NewTransportManager
    transportManager, err := transport.NewTransportManager(cfg.CertManager, ctx.Done())
    if err != nil {
        return fmt.Errorf("could not new transport manager, %w", err)
    }
    trace++

    klog.Infof("%d. prepare cloud kube clients", trace)
    cloudClients, err := createClients(cfg.HeartbeatTimeoutSeconds, cfg.RemoteServers, cfg.CoordinatorServerURL, transportManager)
    if err != nil {
        return fmt.Errorf("failed to create cloud clients, %w", err)
    }
    trace++

    var cloudHealthChecker healthchecker.MultipleBackendsHealthChecker
    if cfg.WorkingMode == util.WorkingModeEdge {
        klog.Infof("%d. create health checkers for remote servers and pool coordinator", trace)
        cloudHealthChecker, err = healthchecker.NewCloudAPIServerHealthChecker(cfg, cloudClients, ctx.Done())
        if err != nil {
            return fmt.Errorf("could not new cloud health checker, %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 and pool coordinator is unhealthy.
        cloudHealthChecker = healthchecker.NewFakeChecker(true, make(map[string]int))
    }
    trace++

    klog.Infof("%d. new restConfig manager", trace)
    restConfigMgr, err := hubrest.NewRestConfigManager(cfg.CertManager, cloudHealthChecker)
    if err != nil {
        return fmt.Errorf("could not new restConfig manager, %w", err)
    }
    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.SharedFactory)
    } else {
        klog.Infof("%d. disable cache manager for node %s because it is a cloud node", trace, cfg.NodeName)
    }
    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, restConfigMgr, ctx.Done())
        if err != nil {
            return fmt.Errorf("could not new gc manager, %w", err)
        }
        // 直接运行manager
        gcMgr.Run()
    } else {
        klog.Infof("%d. disable gc manager for node %s because it is a cloud node", trace, cfg.NodeName)
    }
    trace++

    klog.Infof("%d. new tenant sa manager", trace)
    tenantMgr := tenant.New(cfg.TenantNs, cfg.SharedFactory, ctx.Done())
    trace++

    var coordinatorHealthCheckerGetter func() healthchecker.HealthChecker = getFakeCoordinatorHealthChecker
    var coordinatorTransportManagerGetter func() transport.Interface = getFakeCoordinatorTransportManager
    var coordinatorGetter func() poolcoordinator.Coordinator = getFakeCoordinator

    if cfg.EnableCoordinator {
        klog.Infof("%d. start to run coordinator", trace)
        trace++

        coordinatorInformerRegistryChan := make(chan struct{})
        // coordinatorRun will register secret informer into sharedInformerFactory, and start a new goroutine to periodically check
        // if certs has been got from cloud APIServer. It will close the coordinatorInformerRegistryChan if the secret channel has
        // been registered into informer factory.
        coordinatorHealthCheckerGetter, coordinatorTransportManagerGetter, coordinatorGetter = coordinatorRun(ctx, cfg, restConfigMgr, cloudHealthChecker, coordinatorInformerRegistryChan)
        // wait for coordinator informer registry
        klog.Infof("waiting for coordinator informer registry")
        <-coordinatorInformerRegistryChan
        klog.Infof("coordinator informer registry finished")
    }

    // Start the informer factory if all informers have been registered
    cfg.SharedFactory.Start(ctx.Done())
    cfg.YurtSharedFactory.Start(ctx.Done())

    klog.Infof("%d. new reverse proxy handler for remote servers", trace)
    // 将之前构造的manager作为参数构建yurtProxyHandler
    yurtProxyHandler, err := proxy.NewYurtReverseProxyHandler(
        cfg,
        cacheMgr,
        transportManager,
        cloudHealthChecker,
        tenantMgr,
        coordinatorGetter,
        coordinatorTransportManagerGetter,
        coordinatorHealthCheckerGetter,
        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())
    // 基于yurtProxyHandler运行一个http server.
    if err := server.RunYurtHubServers(cfg, yurtProxyHandler, restConfigMgr, ctx.Done()); err != nil {
        return fmt.Errorf("could not run hub servers, %w", err)
    }
    <-ctx.Done()
    klog.Infof("hub agent exited")
    return nil
}

    hubServerHandler := mux.NewRouter()
    registerHandlers(hubServerHandler, cfg, rest)

    // start yurthub http server for serving metrics, pprof.
    if cfg.YurtHubServerServing != nil {
        if err := cfg.YurtHubServerServing.Serve(hubServerHandler, 0, stopCh); err != nil {
            return err
        }
    }

// registerHandler registers handlers for yurtHubServer, and yurtHubServer can handle requests like profiling, healthz, update token.
func registerHandlers(c *mux.Router, cfg *config.YurtHubConfiguration, rest *rest.RestConfigManager) {
    // register handlers for update join token
    c.Handle("/v1/token", updateTokenHandler(cfg.CertManager)).Methods("POST", "PUT")

    // register handler for health check
    c.HandleFunc("/v1/healthz", healthz).Methods("GET")
    c.Handle("/v1/readyz", readyz(cfg.CertManager)).Methods("GET")

    // register handler for profile
    if cfg.EnableProfiling {
        profile.Install(c)
    }

    // register handler for metrics
    c.Handle("/metrics", promhttp.Handler())

    // register handler for ota upgrade
    c.Handle("/pods", ota.GetPods(cfg.StorageWrapper)).Methods("GET")
    c.Handle("/openyurt.io/v1/namespaces/{ns}/pods/{podname}/upgrade",
        ota.HealthyCheck(rest, cfg.NodeName, ota.UpdatePod)).Methods("POST")
}

    // start yurthub proxy servers for forwarding requests to cloud kube-apiserver
    if cfg.WorkingMode == util.WorkingModeEdge {
        proxyHandler = wrapNonResourceHandler(proxyHandler, cfg, rest)
    }
    if cfg.YurtHubProxyServerServing != nil {
        if err := cfg.YurtHubProxyServerServing.Serve(proxyHandler, 0, stopCh); err != nil {
            return err
        }
    }

    lb, err := remote.NewLoadBalancer(
        yurtHubCfg.LBMode,
        yurtHubCfg.RemoteServers,
        localCacheMgr,
        transportMgr,
        coordinatorGetter,
        cloudHealthChecker,
        yurtHubCfg.FilterManager,
        yurtHubCfg.WorkingMode,
        stopCh)

        // When yurthub works in Edge mode, we may use local proxy or pool proxy to handle
        // the request when offline.
        localProxy = local.NewLocalProxy(localCacheMgr,
            cloudHealthChecker.IsHealthy,
            isCoordinatorHealthy,
            yurtHubCfg.MinRequestTimeout,
        )
        localProxy = local.WithFakeTokenInject(localProxy, yurtHubCfg.SerializerManager)

    yurtProxy := &yurtReverseProxy{
        resolver:                      resolver,
        loadBalancer:                  lb,
        cloudHealthChecker:            cloudHealthChecker,
        coordinatorHealtCheckerGetter: coordinatorHealthCheckerGetter,
        localProxy:                    localProxy,
        poolProxy:                     poolProxy,
        maxRequestsInFlight:           yurtHubCfg.MaxRequestInFlight,
        isCoordinatorReady:            isCoordinatorReady,
        enablePoolCoordinator:         yurtHubCfg.EnableCoordinator,
        tenantMgr:                     tenantMgr,
        workingMode:                   yurtHubCfg.WorkingMode,
    }

    return yurtProxy.buildHandlerChain(yurtProxy), nil

type yurtReverseProxy struct {
    resolver                      apirequest.RequestInfoResolver
    loadBalancer                  remote.LoadBalancer
    cloudHealthChecker            healthchecker.MultipleBackendsHealthChecker
    coordinatorHealtCheckerGetter func() healthchecker.HealthChecker
    localProxy                    http.Handler
    poolProxy                     http.Handler
    maxRequestsInFlight           int
    tenantMgr                     tenant.Interface
    isCoordinatorReady            func() bool
    workingMode                   hubutil.WorkingMode
    enablePoolCoordinator         bool
}

func (p *yurtReverseProxy) ServeHTTP(rw http.ResponseWriter, req *http.Request) {
    if p.workingMode == hubutil.WorkingModeCloud {
        p.loadBalancer.ServeHTTP(rw, req)
        return
    }

    switch {
    case util.IsKubeletLeaseReq(req):
        p.handleKubeletLease(rw, req)
    case util.IsEventCreateRequest(req):
        p.eventHandler(rw, req)
    case util.IsPoolScopedResouceListWatchRequest(req):
        p.poolScopedResouceHandler(rw, req)
    case util.IsSubjectAccessReviewCreateGetRequest(req):
        p.subjectAccessReviewHandler(rw, req)
    default:
        // For resource request that do not need to be handled by pool-coordinator,
        // handling the request with cloud apiserver or local cache.
        if p.cloudHealthChecker.IsHealthy() {
            p.loadBalancer.ServeHTTP(rw, req)
        } else {
            p.localProxy.ServeHTTP(rw, req)
        }
    }
}

type loadBalancer struct {
    backends          []*util.RemoteProxy
    algo              loadBalancerAlgo
    localCacheMgr     cachemanager.CacheManager
    filterManager     *manager.Manager
    coordinatorGetter func() poolcoordinator.Coordinator
    workingMode       hubutil.WorkingMode
    stopCh            <-chan struct{}
}

    backends := make([]*util.RemoteProxy, 0, len(remoteServers))
    for i := range remoteServers {
        b, err := util.NewRemoteProxy(remoteServers[i], lb.modifyResponse, lb.errorHandler, transportMgr, stopCh)
        if err != nil {
            klog.Errorf("could not new proxy backend(%s), %v", remoteServers[i].String(), err)
            continue
        }
        backends = append(backends, b)
    }

    var algo loadBalancerAlgo
    switch lbMode {
    case "rr":
        algo = &rrLoadBalancerAlgo{backends: backends, checker: healthChecker}
    case "priority":
        algo = &priorityLoadBalancerAlgo{backends: backends, checker: healthChecker}
    default:
        algo = &rrLoadBalancerAlgo{backends: backends, checker: healthChecker}
    }

    // pick a remote proxy based on the load balancing algorithm.
    rp := lb.algo.PickOne()
    rp.ServeHTTP(rw, req)

func (lb *loadBalancer) errorHandler(rw http.ResponseWriter, req *http.Request, err error) {
    klog.Errorf("remote proxy error handler: %s, %v", hubutil.ReqString(req), err)
    if lb.localCacheMgr == nil || !lb.localCacheMgr.CanCacheFor(req) {
        rw.WriteHeader(http.StatusBadGateway)
        return
    }

    ctx := req.Context()
    if info, ok := apirequest.RequestInfoFrom(ctx); ok {
        if info.Verb == "get" || info.Verb == "list" {
            # 读取cache内容
            if obj, err := lb.localCacheMgr.QueryCache(req); err == nil {
                hubutil.WriteObject(http.StatusOK, obj, rw, req)
                return
            }
        }
    }
    rw.WriteHeader(http.StatusBadGateway)
}

// QueryCache get runtime object from backend storage for request
func (cm *cacheManager) QueryCache(req *http.Request) (runtime.Object, error) {
    ctx := req.Context()
    info, ok := apirequest.RequestInfoFrom(ctx)
    if !ok || info == nil || info.Resource == "" {
        return nil, fmt.Errorf("failed to get request info for request %s", util.ReqString(req))
    }
    if !info.IsResourceRequest {
        return nil, fmt.Errorf("failed to QueryCache for getting non-resource request %s", util.ReqString(req))
    }

    # 根据verb查询storage中的数据
    switch info.Verb {
    case "list":
        return cm.queryListObject(req)
    case "get", "patch", "update":
        return cm.queryOneObject(req)
    default:
        return nil, fmt.Errorf("failed to QueryCache, unsupported verb %s of request %s", info.Verb, util.ReqString(req))
    }
}

func (cm *cacheManager) queryOneObject(req *http.Request) (runtime.Object, error) {
    ...
    klog.V(4).Infof("component: %s try to get key: %s", comp, key.Key())
    obj, err := cm.storage.Get(key)
    if err != nil {
        klog.Errorf("failed to get obj %s from storage, %v", key.Key(), err)
        return nil, err
    }
    ...
}

// Get will get content from the regular file that specified by key.
// If key points to a dir, return ErrKeyHasNoContent.
func (ds *diskStorage) Get(key storage.Key) ([]byte, error) {
    if err := utils.ValidateKey(key, storageKey{}); err != nil {
        return []byte{}, storage.ErrKeyIsEmpty
    }
    storageKey := key.(storageKey)

    if !ds.lockKey(storageKey) {
        return nil, storage.ErrStorageAccessConflict
    }
    defer ds.unLockKey(storageKey)

    path := filepath.Join(ds.baseDir, storageKey.Key())
    buf, err := ds.fsOperator.Read(path)
    switch err {
    case nil:
        return buf, nil
    case fs.ErrNotExists:
        return nil, storage.ErrStorageNotFound
    case fs.ErrIsNotFile:
        return nil, storage.ErrKeyHasNoContent
    default:
        return buf, fmt.Errorf("failed to read file at %s, %v", path, err)
    }
}

// RemoteProxy is an reverse proxy for remote server
type RemoteProxy struct {
    reverseProxy         *httputil.ReverseProxy
    remoteServer         *url.URL
    currentTransport     http.RoundTripper
    bearerTransport      http.RoundTripper
    upgradeHandler       *proxy.UpgradeAwareHandler
    bearerUpgradeHandler *proxy.UpgradeAwareHandler
    stopCh               <-chan struct{}
}

func (rp *RemoteProxy) ServeHTTP(rw http.ResponseWriter, req *http.Request) {
    if httpstream.IsUpgradeRequest(req) {
        klog.V(5).Infof("get upgrade request %s", req.URL)
        if isBearerRequest(req) {
            rp.bearerUpgradeHandler.ServeHTTP(rw, req)
        } else {
            rp.upgradeHandler.ServeHTTP(rw, req)
        }
        return
    }

    rp.reverseProxy.ServeHTTP(rw, req)
}

func (r *responder) Error(w http.ResponseWriter, req *http.Request, err error) {
    klog.Errorf("failed while proxying request %s, %v", req.URL, err)
    http.Error(w, err.Error(), http.StatusInternalServerError)
}

// LocalProxy is responsible for handling requests when remote servers are unhealthy
type LocalProxy struct {
    cacheMgr           manager.CacheManager
    isCloudHealthy     IsHealthy
    isCoordinatorReady IsHealthy
    minRequestTimeout  time.Duration
}

// ServeHTTP implements http.Handler for LocalProxy
func (lp *LocalProxy) ServeHTTP(w http.ResponseWriter, req *http.Request) {
    var err error
    ctx := req.Context()
    if reqInfo, ok := apirequest.RequestInfoFrom(ctx); ok && reqInfo != nil && reqInfo.IsResourceRequest {
        klog.V(3).Infof("go into local proxy for request %s", hubutil.ReqString(req))
        switch reqInfo.Verb {
        case "watch":
            err = lp.localWatch(w, req)
        case "create":
            err = lp.localPost(w, req)
        case "delete", "deletecollection":
            err = localDelete(w, req)
        default: // list, get, update
            err = lp.localReqCache(w, req)
        }

        if err != nil {
            klog.Errorf("could not proxy local for %s, %v", hubutil.ReqString(req), err)
            util.Err(err, w, req)
        }
    } else {
        klog.Errorf("local proxy does not support request(%s), requestInfo: %s", hubutil.ReqString(req), hubutil.ReqInfoString(reqInfo))
        util.Err(apierrors.NewBadRequest(fmt.Sprintf("local proxy does not support request(%s)", hubutil.ReqString(req))), w, req)
    }
}

// localReqCache handles Get/List/Update requests when remote servers are unhealthy
func (lp *LocalProxy) localReqCache(w http.ResponseWriter, req *http.Request) error {
    if !lp.cacheMgr.CanCacheFor(req) {
        klog.Errorf("can not cache for %s", hubutil.ReqString(req))
        return apierrors.NewBadRequest(fmt.Sprintf("can not cache for %s", hubutil.ReqString(req)))
    }

    obj, err := lp.cacheMgr.QueryCache(req)
    if errors.Is(err, storage.ErrStorageNotFound) || errors.Is(err, hubmeta.ErrGVRNotRecognized) {
        klog.Errorf("object not found for %s", hubutil.ReqString(req))
        reqInfo, _ := apirequest.RequestInfoFrom(req.Context())
        return apierrors.NewNotFound(schema.GroupResource{Group: reqInfo.APIGroup, Resource: reqInfo.Resource}, reqInfo.Name)
    } else if err != nil {
        klog.Errorf("failed to query cache for %s, %v", hubutil.ReqString(req), err)
        return apierrors.NewInternalError(err)
    } else if obj == nil {
        klog.Errorf("no cache object for %s", hubutil.ReqString(req))
        return apierrors.NewInternalError(fmt.Errorf("no cache object for %s", hubutil.ReqString(req)))
    }

    return util.WriteObject(http.StatusOK, obj, w, req)
}

obj, err := lp.cacheMgr.QueryCache(req)
return util.WriteObject(http.StatusOK, obj, w, req)

- proxy: 反向代理的实现
- cachemanager：cache的实现
- storage：本地存储的实现