diff --git a/weed/worker/tasks/erasure_coding/detection.go b/weed/worker/tasks/erasure_coding/detection.go index a2c045ff8..0d6ac752c 100644 --- a/weed/worker/tasks/erasure_coding/detection.go +++ b/weed/worker/tasks/erasure_coding/detection.go @@ -4,10 +4,14 @@ import ( "context" "fmt" "sort" + "strings" "time" "github.com/seaweedfs/seaweedfs/weed/admin/topology" "github.com/seaweedfs/seaweedfs/weed/glog" + "github.com/seaweedfs/seaweedfs/weed/operation" + "github.com/seaweedfs/seaweedfs/weed/pb" + "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb" "github.com/seaweedfs/seaweedfs/weed/pb/worker_pb" "github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding" "github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding/placement" @@ -102,6 +106,51 @@ func Detection(ctx context.Context, metrics []*types.VolumeHealthMetrics, cluste continue } + // Handle the "stuck source" state from #9448: a previous encode + // succeeded but the post-encode source-delete left a regular replica + // behind, so the master heartbeats BOTH the replica AND its EC shards. + // metric.IsECVolume above is set only for the EC-side metric path, so + // the canonical metric we picked is the regular replica with + // IsECVolume=false. Re-proposing an encode in that state collides with + // the mounted shards on the targets ("ec volume %d is mounted; refusing + // overwrite") and the detector re-queues forever. + // + // We only act when the EC shard set is COMPLETE — fewer than + // totalShards present means the existing recovery branch below + // (around the `existingECShards` block) should keep its chance to + // fold the partial shards into the new task. Counting walks + // EcIndexBits to handle a single info entry carrying multiple shards. + if clusterInfo.ActiveTopology != nil { + shardCount := countExistingEcShardsForVolume(clusterInfo.ActiveTopology, metric.VolumeID, metric.Collection) + totalShards := erasure_coding.DataShardsCount + erasure_coding.ParityShardsCount + if shardCount >= totalShards { + glog.Warningf("EC Detection: Volume %d has all %d EC shards in topology; "+ + "source replica on %s is orphaned (#9448).", + metric.VolumeID, totalShards, metric.Server) + if clusterInfo.GrpcDialOption != nil { + deleted, cleanupErr := cleanupOrphanSourceReplicas(ctx, clusterInfo, metric, totalShards) + switch { + case cleanupErr != nil: + // Don't fall through to a re-encode — that would just + // collide with the mounted shards again. Surface the + // failure and wait for the next cycle; the source is + // still safe. + glog.Warningf("EC Detection: failed to auto-clean orphaned source for volume %d: %v", metric.VolumeID, cleanupErr) + case deleted > 0: + glog.Infof("EC Detection: auto-cleaned %d orphaned source replica(s) for volume %d after verifying all %d EC shards present", deleted, metric.VolumeID, totalShards) + default: + glog.V(1).Infof("EC Detection: no orphaned regular replicas found in topology for volume %d (collection %q)", metric.VolumeID, metric.Collection) + } + } else { + glog.Warningf("EC Detection: no gRPC dial option available to auto-clean orphaned source for volume %d; "+ + "to clean up by hand, send a targeted VolumeDelete RPC to %s only — DO NOT use the cluster-wide `volume.delete` shell command, which would also delete the EC shards.", + metric.VolumeID, metric.Server) + } + skippedAlreadyEC++ + continue + } + } + // Check minimum size requirement if metric.Size < minSizeBytes { skippedTooSmall++ @@ -885,3 +934,94 @@ func findExistingECShards(activeTopology *topology.ActiveTopology, volumeID uint } return activeTopology.GetECShardLocations(volumeID, collection) } + +// cleanupOrphanSourceReplicas deletes any regular volume replicas still +// present in the topology for (volumeID, collection) after re-verifying that +// the full EC shard set is intact. Caller must hold expectedShards equal to +// the configured totalShards count. Issues VolumeDelete RPC to each replica +// server's address — that endpoint only touches the regular volume on the +// targeted server, never EC shards (those live in a separate store path). +// The cluster-wide `volume.delete` shell command is what would have nuked +// the EC shards too; the targeted RPC used here is safe by construction. +// Returns the count of replicas successfully deleted plus any error. +func cleanupOrphanSourceReplicas(ctx context.Context, clusterInfo *types.ClusterInfo, metric *types.VolumeHealthMetrics, expectedShards int) (int, error) { + if clusterInfo == nil || clusterInfo.ActiveTopology == nil { + return 0, fmt.Errorf("active topology unavailable") + } + if clusterInfo.GrpcDialOption == nil { + return 0, fmt.Errorf("grpc dial option unavailable") + } + + // Re-verify shard completeness right before acting. Defensive: detection + // processes many volumes sequentially and the topology snapshot we built + // at start-of-detection could have lost shards in between (a volume + // server going down between iterations). Refusing to delete the source + // when we can no longer prove the shards are complete is the safer + // failure mode — the source replica is the only complete copy. + actualShards := countExistingEcShardsForVolume(clusterInfo.ActiveTopology, metric.VolumeID, metric.Collection) + if actualShards < expectedShards { + return 0, fmt.Errorf("EC shard set shrank between detection and cleanup (%d < %d); refusing to delete source replica", actualShards, expectedShards) + } + + replicas := findVolumeReplicaLocations(clusterInfo.ActiveTopology, metric.VolumeID, metric.Collection) + if len(replicas) == 0 { + return 0, nil + } + + deleted := 0 + var deleteErrors []string + for _, replica := range replicas { + serverAddress := replica.ServerID + err := operation.WithVolumeServerClient(false, pb.ServerAddress(serverAddress), clusterInfo.GrpcDialOption, + func(client volume_server_pb.VolumeServerClient) error { + _, deleteErr := client.VolumeDelete(ctx, &volume_server_pb.VolumeDeleteRequest{ + VolumeId: metric.VolumeID, + OnlyEmpty: false, + }) + return deleteErr + }) + if err != nil { + deleteErrors = append(deleteErrors, fmt.Sprintf("server %s: %v", serverAddress, err)) + continue + } + deleted++ + glog.V(1).Infof("EC Detection: deleted orphan regular replica for volume %d on %s", metric.VolumeID, serverAddress) + } + + if len(deleteErrors) > 0 { + return deleted, fmt.Errorf("%d of %d replica delete(s) failed: %s", len(deleteErrors), len(replicas), strings.Join(deleteErrors, "; ")) + } + return deleted, nil +} + +// countExistingEcShardsForVolume returns the number of distinct EC shard IDs +// for (volumeID, collection) present in the topology. Walks every disk's +// EcIndexBits bitmap rather than trusting len(EcShardInfos), because a single +// info entry can carry multiple shards. Used by the #9448 guard to decide +// whether the EC shard set is complete enough that the orphaned regular +// replica is safe to delete. +func countExistingEcShardsForVolume(activeTopology *topology.ActiveTopology, volumeID uint32, collection string) int { + if activeTopology == nil { + return 0 + } + topologyInfo := activeTopology.GetTopologyInfo() + if topologyInfo == nil { + return 0 + } + var seen erasure_coding.ShardBits + for _, dc := range topologyInfo.DataCenterInfos { + for _, rack := range dc.RackInfos { + for _, node := range rack.DataNodeInfos { + for _, diskInfo := range node.DiskInfos { + for _, ecShardInfo := range diskInfo.EcShardInfos { + if ecShardInfo.Id != volumeID || ecShardInfo.Collection != collection { + continue + } + seen |= erasure_coding.ShardBits(ecShardInfo.EcIndexBits) + } + } + } + } + } + return seen.Count() +} diff --git a/weed/worker/tasks/erasure_coding/detection_test.go b/weed/worker/tasks/erasure_coding/detection_test.go index 4ded8e2bc..91ca8aab5 100644 --- a/weed/worker/tasks/erasure_coding/detection_test.go +++ b/weed/worker/tasks/erasure_coding/detection_test.go @@ -127,6 +127,179 @@ func TestECPlacementPlannerFallsBackWhenTagsInsufficient(t *testing.T) { assert.Less(t, taggedCount, len(selected)) } +// TestDetectionSkipsWhenECShardsAlreadyExist guards against issue #9448: a +// regular replica that survived a previous successful EC encode (source +// delete didn't clean it up for some reason) gets re-proposed for encoding, +// the new encode collides with the already-mounted shards on the targets +// ("ec volume %d is mounted; refusing overwrite"), and detection loops +// forever on the same volume. Detection must see the existing shards and +// skip the volume so an admin can clean it up out-of-band. +// +// The guard fires ONLY when the EC shard set is complete (count >= +// totalShards), so a partially-distributed previous attempt still falls +// through to the existing recovery branch in the encode path. +func TestDetectionSkipsWhenECShardsAlreadyExist(t *testing.T) { + const volumeID uint32 = 42 + activeTopology := buildStuckSourceTopology(t, volumeID, erasure_coding.TotalShardsCount) + + clusterInfo := &types.ClusterInfo{ActiveTopology: activeTopology} + metrics := buildStuckSourceMetrics(volumeID, "127.0.0.1:8080") + + results, hasMore, err := Detection(context.Background(), metrics, clusterInfo, NewDefaultConfig(), 0) + require.NoError(t, err) + require.False(t, hasMore) + require.Empty(t, results, "stuck source replica with all EC shards present must not yield a new encoding proposal") +} + +// TestDetectionAllowsRegularReplicaWhenShardsPartial covers the partial-EC +// branch of the #9448 guard: when fewer than totalShards exist, the volume +// is allowed to flow through to the normal encoding path so the existing +// recovery branch (the `existingECShards` block in the encode arm) can fold +// the partial shards into the new task. A bug here would either (a) skip +// the volume entirely or (b) emit a proposal that later collides on the +// mounted shards. +func TestDetectionAllowsRegularReplicaWhenShardsPartial(t *testing.T) { + const volumeID uint32 = 43 + activeTopology := buildStuckSourceTopology(t, volumeID, erasure_coding.DataShardsCount-1) + + clusterInfo := &types.ClusterInfo{ActiveTopology: activeTopology} + metrics := buildStuckSourceMetrics(volumeID, "127.0.0.1:8080") + + results, _, err := Detection(context.Background(), metrics, clusterInfo, NewDefaultConfig(), 0) + require.NoError(t, err) + // Partial shards are not a "stuck source" — the encode arm must keep + // its chance to either propose a fresh task that folds the partial + // shards into cleanup, or fail planning on the constrained topology. + // We don't require len(results) > 0 because the constrained topology + // (one disk per node, the orphaned shards already taking slots) can + // legitimately fail destination planning. The assertion that matters + // is: the #9448 guard did NOT silently swallow the volume into a + // skippedAlreadyEC counter, and any emitted result is still an EC + // task and not a no-op. + for _, r := range results { + require.Equal(t, types.TaskTypeErasureCoding, r.TaskType, "any emitted result should still be an EC task, not a no-op") + } +} + +// buildStuckSourceTopology constructs a topology that mimics the #9448 stuck +// state: a regular volume replica on node 0 plus `presentShardCount` EC +// shards distributed across nodes 0..presentShardCount-1. +func buildStuckSourceTopology(t *testing.T, volumeID uint32, presentShardCount int) *topology.ActiveTopology { + t.Helper() + require.LessOrEqual(t, presentShardCount, erasure_coding.TotalShardsCount) + activeTopology := topology.NewActiveTopology(10) + nodes := make([]*master_pb.DataNodeInfo, 0, erasure_coding.TotalShardsCount) + for i := 0; i < erasure_coding.TotalShardsCount; i++ { + nodeID := fmt.Sprintf("127.0.0.1:%d", 8080+i) + diskInfo := &master_pb.DiskInfo{ + DiskId: 0, + VolumeCount: 1, + MaxVolumeCount: 100, + } + if i < presentShardCount { + diskInfo.EcShardInfos = []*master_pb.VolumeEcShardInformationMessage{{ + Id: volumeID, + Collection: "", + EcIndexBits: uint32(1) << uint(i), + DiskId: 0, + }} + } + if i == 0 { + diskInfo.VolumeInfos = []*master_pb.VolumeInformationMessage{{ + Id: volumeID, + DiskId: 0, + DiskType: "hdd", + Size: 200 * 1024 * 1024, + }} + } + nodes = append(nodes, &master_pb.DataNodeInfo{ + Id: nodeID, + DiskInfos: map[string]*master_pb.DiskInfo{"hdd": diskInfo}, + }) + } + require.NoError(t, activeTopology.UpdateTopology(&master_pb.TopologyInfo{ + DataCenterInfos: []*master_pb.DataCenterInfo{{ + Id: "dc1", + RackInfos: []*master_pb.RackInfo{{ + Id: "rack1", + DataNodeInfos: nodes, + }}, + }}, + })) + return activeTopology +} + +// buildStuckSourceMetrics returns a metric that already satisfies the EC +// criteria (Age, FullnessRatio, Size), with `Age` derived from `LastModified` +// so the two fields stay consistent for any reader. +func buildStuckSourceMetrics(volumeID uint32, server string) []*types.VolumeHealthMetrics { + lastModified := time.Now().Add(-time.Hour) + return []*types.VolumeHealthMetrics{{ + VolumeID: volumeID, + Server: server, + Size: 200 * 1024 * 1024, + Collection: "", + FullnessRatio: 0.9, + LastModified: lastModified, + Age: time.Since(lastModified), + }} +} + +// TestCountExistingEcShardsForVolume verifies that the helper walks the +// EcIndexBits bitmap (not just len(EcShardInfos)) so it correctly counts +// distinct shard ids even when a single info entry on one disk carries +// multiple shards. +func TestCountExistingEcShardsForVolume(t *testing.T) { + const volumeID uint32 = 99 + activeTopology := topology.NewActiveTopology(10) + require.NoError(t, activeTopology.UpdateTopology(&master_pb.TopologyInfo{ + DataCenterInfos: []*master_pb.DataCenterInfo{{ + Id: "dc1", + RackInfos: []*master_pb.RackInfo{{ + Id: "rack1", + DataNodeInfos: []*master_pb.DataNodeInfo{ + { + Id: "127.0.0.1:8080", + DiskInfos: map[string]*master_pb.DiskInfo{ + "hdd": { + DiskId: 0, + MaxVolumeCount: 100, + // One info entry, three shards present (ids 0, 2, 5). + EcShardInfos: []*master_pb.VolumeEcShardInformationMessage{{ + Id: volumeID, + Collection: "", + EcIndexBits: (uint32(1) << 0) | (uint32(1) << 2) | (uint32(1) << 5), + DiskId: 0, + }}, + }, + }, + }, + { + Id: "127.0.0.1:8081", + DiskInfos: map[string]*master_pb.DiskInfo{ + "hdd": { + DiskId: 0, + MaxVolumeCount: 100, + // One info entry, one shard (id 3) — overlaps with neither. + EcShardInfos: []*master_pb.VolumeEcShardInformationMessage{{ + Id: volumeID, + Collection: "", + EcIndexBits: uint32(1) << 3, + DiskId: 0, + }}, + }, + }, + }, + }, + }}, + }}, + })) + + assert.Equal(t, 4, countExistingEcShardsForVolume(activeTopology, volumeID, "")) + assert.Equal(t, 0, countExistingEcShardsForVolume(activeTopology, volumeID, "other-collection")) + assert.Equal(t, 0, countExistingEcShardsForVolume(nil, volumeID, "")) +} + func TestDetectionContextCancellation(t *testing.T) { activeTopology := buildActiveTopology(t, 5, []string{"hdd", "ssd"}, 50, 0) clusterInfo := &types.ClusterInfo{ActiveTopology: activeTopology} diff --git a/weed/worker/tasks/erasure_coding/plugin_handler.go b/weed/worker/tasks/erasure_coding/plugin_handler.go index ba22b010a..f29026bf1 100644 --- a/weed/worker/tasks/erasure_coding/plugin_handler.go +++ b/weed/worker/tasks/erasure_coding/plugin_handler.go @@ -217,7 +217,7 @@ func (h *ErasureCodingHandler) Detect( return err } - clusterInfo := &workertypes.ClusterInfo{ActiveTopology: activeTopology} + clusterInfo := &workertypes.ClusterInfo{ActiveTopology: activeTopology, GrpcDialOption: h.grpcDialOption} maxResults := int(request.MaxResults) if maxResults < 0 { maxResults = 0 diff --git a/weed/worker/types/data_types.go b/weed/worker/types/data_types.go index dcb69cf7e..cc6de8feb 100644 --- a/weed/worker/types/data_types.go +++ b/weed/worker/types/data_types.go @@ -4,6 +4,7 @@ import ( "time" "github.com/seaweedfs/seaweedfs/weed/admin/topology" + "google.golang.org/grpc" ) // ReplicaLocation identifies where a volume replica lives. @@ -21,6 +22,11 @@ type ClusterInfo struct { LastUpdated time.Time ActiveTopology *topology.ActiveTopology // Added for destination planning in detection VolumeReplicaMap map[uint32][]ReplicaLocation + // GrpcDialOption is set when a detector needs to make targeted gRPC calls + // during detection (e.g., the EC detector auto-cleans up an orphaned + // regular replica that survived a previous encode; see #9448). Optional: + // detectors that don't need RPC access ignore this. + GrpcDialOption grpc.DialOption `json:"-"` } // VolumeHealthMetrics contains health information about a volume (simplified)