(ctx context.Context, clonedCluster *store.Cluster)
| 325 | } |
| 326 | |
| 327 | func (c *ClusterChecker) tryUpdateMigrationStatus(ctx context.Context, clonedCluster *store.Cluster) { |
| 328 | log := logger.Get().With( |
| 329 | zap.String("namespace", c.namespace), |
| 330 | zap.String("cluster", c.clusterName)) |
| 331 | |
| 332 | for i, shard := range clonedCluster.Shards { |
| 333 | if !shard.IsMigrating() { |
| 334 | continue |
| 335 | } |
| 336 | sourceNode := shard.GetMasterNode() |
| 337 | sourceNodeClusterInfo, err := sourceNode.GetClusterInfo(ctx) |
| 338 | if err != nil { |
| 339 | log.With( |
| 340 | zap.Int("shard_index", i), |
| 341 | zap.String("source_node", sourceNode.ID()), |
| 342 | ).Error("Failed to get the cluster info from the source node", zap.Error(err)) |
| 343 | continue |
| 344 | } |
| 345 | |
| 346 | // If there is no migration information on the source node or the source node migration slot is not equal to the shard, |
| 347 | // you need to clear the migration information on the controller. |
| 348 | if sourceNodeClusterInfo.MigratingSlot == nil || (sourceNodeClusterInfo.MigratingSlot != nil && |
| 349 | !sourceNodeClusterInfo.MigratingSlot.Equal(shard.MigratingSlot.SlotRange)) { |
| 350 | log.Error("Mismatch migrating slot", |
| 351 | zap.Int("shard_index", i), |
| 352 | zap.String("migrating_slot", shard.MigratingSlot.String()), |
| 353 | ) |
| 354 | clonedCluster.Shards[i].ClearMigrateState() |
| 355 | if err = c.clusterStore.UpdateCluster(ctx, c.namespace, clonedCluster); err != nil { |
| 356 | log.Error("Failed to update the migrate state by UpdateCluster method", zap.Error(err)) |
| 357 | return |
| 358 | } |
| 359 | c.updateCluster(clonedCluster) |
| 360 | continue |
| 361 | } |
| 362 | |
| 363 | if shard.TargetShardIndex < 0 || shard.TargetShardIndex >= len(clonedCluster.Shards) { |
| 364 | log.Error("Invalid target shard index", zap.Int("index", shard.TargetShardIndex)) |
| 365 | return |
| 366 | } |
| 367 | |
| 368 | migratingSlot := shard.MigratingSlot.String() |
| 369 | switch sourceNodeClusterInfo.MigratingState { |
| 370 | case "none", "start": |
| 371 | continue |
| 372 | case "fail": |
| 373 | clonedCluster.Shards[i].ClearMigrateState() |
| 374 | if err = c.clusterStore.UpdateCluster(ctx, c.namespace, clonedCluster); err != nil { |
| 375 | log.Error("Failed to update the cluster", zap.Error(err)) |
| 376 | return |
| 377 | } |
| 378 | c.updateCluster(clonedCluster) |
| 379 | log.Warn("Failed to migrate the slot", zap.String("slot", migratingSlot)) |
| 380 | case "success": |
| 381 | clonedCluster.Shards[i].SlotRanges = store.RemoveSlotFromSlotRanges(clonedCluster.Shards[i].SlotRanges, shard.MigratingSlot.SlotRange) |
| 382 | clonedCluster.Shards[shard.TargetShardIndex].SlotRanges = store.AddSlotToSlotRanges( |
| 383 | clonedCluster.Shards[shard.TargetShardIndex].SlotRanges, shard.MigratingSlot.SlotRange, |
| 384 | ) |
no test coverage detected