Method syncClusterToNodes

controller/cluster.go:183–208

(ctx context.Context)

}

func (c *ClusterChecker) syncClusterToNodes(ctx context.Context) error {
	clusterInfo, err := c.clusterStore.GetCluster(ctx, c.namespace, c.clusterName)
	if err != nil {
		return err
	}
	version := clusterInfo.Version.Load()
	for _, shard := range clusterInfo.Shards {
		for _, node := range shard.Nodes {
			go func(n store.Node) {
				log := logger.Get().With(
					zap.String("namespace", c.namespace),
					zap.String("cluster", c.clusterName),
					zap.Int64("version", version),
					zap.String("node_id", n.ID()),
					zap.String("addr", n.Addr()))
				// sync the clusterName to the latest version
				if err := n.SyncClusterInfo(ctx, clusterInfo); err != nil {
					log.Error("Failed to sync the cluster topology to the node", zap.Error(err))
				} else {
					log.Info("Succeed to sync the cluster topology to the node")
				}
			}(node)
		}
	}
	return nil
}

func (c *ClusterChecker) parallelProbeNodes(ctx context.Context, cluster *store.Cluster) {
	var mu sync.Mutex