深入解读 KubeCube 多集群管理

// MultiClustersManager access to internal cluster
type MultiClustersManager interface {
	// Add runtime cache in memory
	Add(cluster string, internalCluster *InternalCluster) error
	Get(cluster string) (*InternalCluster, error)
	Del(cluster string) error

	// FuzzyCopy return fuzzy cluster of raw
	FuzzyCopy() map[string]*FuzzyCluster

	// ScoutFor scout heartbeat for warden
	ScoutFor(ctx context.Context, cluster string) error

	// GetClient get client for cluster
	GetClient(cluster string) (kubernetes.Client, error)
}

type Client interface {
	Cache() cache.Cache
	Direct() client.Client
	Metrics() versioned.Interface
	ClientSet() kubernetes.Interface
}

// Collect will scout a specified warden of cluster
func (s *Scout) Collect(ctx context.Context) {
	for {
		select {
		case info := <-s.Receiver:
			s.healthWarden(ctx, info)

		case <-time.Tick(time.Duration(s.WaitTimeoutSeconds) * time.Second):
			s.illWarden(ctx)

		case <-ctx.Done():
			clog.Warn("scout of %v warden stopped: %v", s.Cluster, ctx.Err())
			return
		}
	}
}

// waitForReady wait all components of warden ready
func (r *Reporter) waitForReady() error {
	counts := len(checkFuncs)
	if counts < 1 {
		return fmt.Errorf("less 1 components to check ready")
	}

	// wait all components ready in specified time
	ctx, cancel := context.WithTimeout(context.Background(), time.Duration(r.WaitSecond)*time.Second)
	defer cancel()

	readyzCh := make(chan struct{})

	for _, fn := range checkFuncs {
		go readyzCheck(ctx, readyzCh, fn)
	}

	for {
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-readyzCh:
			counts--
			if counts < 1 {
				return nil
			}
		}
	}
}

var checkFuncs []checkFunc

// RegisterCheckFunc should be used to register readyz check func
func RegisterCheckFunc(fn checkFunc) {
	checkFuncs = append(checkFuncs, fn)
}

func readyzCheck(ctx context.Context, ch chan struct{}, checkFn checkFunc) {
	for {
		select {
		case <-time.Tick(waitPeriod):
			if checkFn() {
				ch <- struct{}{}
				return
			}
		case <-ctx.Done():
			return
		}
	}
}

// report do real report loop
func (r *Reporter) report(stop <-chan struct{}) {
	for {
		select {
		case <-time.Tick(time.Duration(r.PeriodSecond) * time.Second):
			if !r.do() {
				r.illPivotCluster()
			} else {
				r.healPivotCluster()
			}
		case <-stop:
			return
		}
	}
}

		// generate internal cluster for current cluster and add
		// it to the cache of multi cluster manager
		skip, err := multiclustermgr.AddInternalCluster(currentCluster)
		if err != nil {
			log.Error(err.Error())
		}

		if err != nil && !skip {
			updateFn := func(cluster *clusterv1.Cluster) {
				initFailedState := clusterv1.ClusterInitFailed
				reason := fmt.Sprintf("cluster %s init failed", cluster.Name)
				cluster.Status.State = &initFailedState
				cluster.Status.Reason = reason
			}

			err := utils.UpdateStatus(ctx, r.Client, &currentCluster, updateFn)
			if err != nil {
				log.Error("update cluster %v status failed", currentCluster.Name)
				return ctrl.Result{}, err
			}

			r.enqueue(currentCluster)

			return ctrl.Result{}, nil
		}

// try to reconnect with cluster api server, requeue if every is ok
	go func() {
		log.Info("cluster %v init failed, keep retry background", cluster.Name)

		// pop from retry queue when reconnected or context exceed or context canceled
		defer r.retryQueue.Delete(cluster.Name)

		for {
			select {
			case <-time.Tick(retryInterval):
				_, err := client.New(config, client.Options{Scheme: r.Scheme})
				if err == nil {
					log.Info("enqueuing cluster %v for reconciliation", cluster.Name)
					r.Affected <- event.GenericEvent{Object: &cluster}
					return
				}
			case <-ctx.Done():
				log.Info("cluster %v retry task stopped: %v", cluster.Name, ctx.Err())

				// retrying timeout need update status
				// todo(weilaaa): to allow user reconnect cluster manually
				if ctx.Err().Error() == "context deadline exceeded" {
					updateFn := func(cluster *clusterv1.Cluster) {
						state := clusterv1.ClusterReconnectedFailed
						reason := fmt.Sprintf("cluster %s reconnect timeout: %v", cluster.Name, retryTimeout)
						cluster.Status.State = &state
						cluster.Status.Reason = reason
					}
					err := utils.UpdateStatus(ctx, r.Client, &cluster, updateFn)
					if err != nil {
						log.Warn("update cluster %v status failed: %v", cluster.Name, err)
					}
				}

				return
			}
		}
	}()

	// stop retry task if cluster in retry queue
	cancel, ok := r.retryQueue.Load(cluster.Name)
	if ok {
		cancel.(context.CancelFunc)()
		clog.Debug("stop retry task of cluster %v success", cluster.Name)
		return nil
	}

	if !isDisconnected(cluster, s.WaitTimeoutSeconds) {
		// going here means cluster heartbeat is normal

		if s.ClusterState != v1.ClusterNormal {
			clog.Info("cluster %v connected", cluster.Name)
		}

		s.LastHeartbeat = cluster.Status.LastHeartbeat.Time
		s.ClusterState = v1.ClusterNormal
		return
	}

	if s.ClusterState == v1.ClusterNormal {
		reason := fmt.Sprintf("cluster %s disconnected", s.Cluster)

		updateFn := func(obj *v1.Cluster) {
			state := v1.ClusterAbnormal
			obj.Status.State = &state
			obj.Status.Reason = reason
			obj.Status.LastHeartbeat = &metav1.Time{Time: s.LastHeartbeat}
		}

		clog.Warn("%v, last heartbeat: %v", reason, s.LastHeartbeat)

		err := utils.UpdateStatus(ctx, s.Client, cluster, updateFn)
		if err != nil {
			clog.Error(err.Error())
		}
	}

	s.ClusterState = v1.ClusterAbnormal

  if s.ClusterState != v1.ClusterNormal {
		clog.Info("cluster %v connected", cluster.Name)
	}

	s.LastHeartbeat = time.Now()

	updateFn := func(obj *v1.Cluster) {
		state := v1.ClusterNormal
		obj.Status.State = &state
		obj.Status.Reason = fmt.Sprintf("receive heartbeat from cluster %s", s.Cluster)
		obj.Status.LastHeartbeat = &metav1.Time{Time: s.LastHeartbeat}
	}

	err = utils.UpdateStatus(ctx, s.Client, cluster, updateFn)
	if err != nil {
		clog.Error(err.Error())
		return
	}

	s.ClusterState = v1.ClusterNormal

// PreCheck do cluster health check, early return
// if cluster if unhealthy
func PreCheck() gin.HandlerFunc {
	return func(c *gin.Context) {
		cluster := fetchCluster(c)

		clog.Debug("request path: %v, request cluster: %v", c.FullPath(), cluster)

		if len(cluster) > 0 {
			_, err := multicluster.Interface().Get(cluster)
			if err != nil {
				clog.Warn("cluster %v unhealthy, err: %v", cluster, err.Error())
				response.FailReturn(c, errcode.CustomReturn(http.StatusInternalServerError, "cluster %v unhealthy", cluster))
				return
			}
		}
	}
}