(ctx context.Context, cluster *store.Cluster)
| 208 | } |
| 209 | |
| 210 | func (c *ClusterChecker) parallelProbeNodes(ctx context.Context, cluster *store.Cluster) { |
| 211 | var mu sync.Mutex |
| 212 | var latestNodeVersion int64 = 0 |
| 213 | var latestClusterNodesStr string |
| 214 | var wg sync.WaitGroup |
| 215 | |
| 216 | for i, shard := range cluster.Shards { |
| 217 | for _, node := range shard.Nodes { |
| 218 | wg.Add(1) |
| 219 | go func(shardIdx int, n store.Node) { |
| 220 | defer wg.Done() |
| 221 | log := logger.Get().With( |
| 222 | zap.String("cluster_name", c.clusterName), |
| 223 | zap.String("id", n.ID()), |
| 224 | zap.Bool("is_master", n.IsMaster()), |
| 225 | zap.String("addr", n.Addr()), |
| 226 | ) |
| 227 | version, err := c.probeNode(ctx, n) |
| 228 | // Don't sync the cluster info to the node if it is restoring the db from backup |
| 229 | if errors.Is(err, ErrRestoringBackUp) { |
| 230 | log.Error("The node is restoring the db from backup") |
| 231 | return |
| 232 | } |
| 233 | if err != nil && !errors.Is(err, ErrClusterNotInitialized) { |
| 234 | failureCount := c.increaseFailureCount(shardIdx, n) |
| 235 | log.With(zap.Error(err), |
| 236 | zap.Int64("failure_count", failureCount), |
| 237 | ).Warn("Failed to probe the node") |
| 238 | return |
| 239 | } |
| 240 | log.Debug("Probe the clusterName node") |
| 241 | |
| 242 | clusterVersion := cluster.Version.Load() |
| 243 | if version < clusterVersion { |
| 244 | // sync the clusterName to the latest version |
| 245 | if err := n.SyncClusterInfo(ctx, cluster); err != nil { |
| 246 | log.With(zap.Error(err)).Error("Failed to sync the clusterName info") |
| 247 | } |
| 248 | } else if version > clusterVersion { |
| 249 | log.With( |
| 250 | zap.Int64("node.version", version), |
| 251 | zap.Int64("clusterName.version", clusterVersion), |
| 252 | ).Warn("The node is in a higher version") |
| 253 | mu.Lock() |
| 254 | if version > latestNodeVersion { |
| 255 | latestNodeVersion = version |
| 256 | clusterNodesStr, errX := n.GetClusterNodesString(ctx) |
| 257 | if errX != nil { |
| 258 | log.With(zap.String("node", n.ID()), zap.Error(errX)).Error("Failed to get the cluster nodes info from node") |
| 259 | // set empty explicitly |
| 260 | latestClusterNodesStr = "" |
| 261 | } else { |
| 262 | latestClusterNodesStr = clusterNodesStr |
| 263 | } |
| 264 | } |
| 265 | mu.Unlock() |
| 266 | } |
| 267 | c.resetFailureCount(n.ID()) |
no test coverage detected