ec.encode: require shard placement to meet -shardReplicaPlacement (#11947)

* ec: measure shard placement against replica-placement caps

ecShardPlacementViolations reports where a volume's newest-generation
shard placement exceeds the per-node and per-rack caps a
-shardReplicaPlacement requests, using the same digit mapping the
balance planner enforces (SameRackCount = shards/node, DiffRackCount =
shards/rack, dc-qualified rack keys).

* ec.encode: -strictPlacement keeps originals when placement breaches caps

-shardReplicaPlacement tells the balancer the requested per-node and
per-rack shard limits, but nothing verifies the outcome. With
-strictPlacement the pre-delete shard verification also measures the
final placement against those caps: a volume still over the node or
rack limit aborts before its source is deleted, so the encode only
completes when the requested failure-domain spread is actually met.

* ec.encode: cover the strict-placement delete gate with a test

The gate in verifyEcShardsBeforeDelete now runs against a stubbed
topology: a compliant spread passes while an over-cap placement is
refused before source deletion. The retry knobs become package vars so
the test does not wait out the heartbeat-propagation backoff.

* ec.encode: enforce -shardReplicaPlacement without a flag

Drop the -strictPlacement option: the requested per-node and per-rack
caps are always requirements at the pre-delete verification, so the
encode only completes when the placement actually honors them.

weed/ec/ec_encode.go

@@ -121,7 +121,7 @@ func ProcessEcEncodeBatch(env *Env, writer io.Writer, volumeIds []needle.VolumeI
 	if err := EcBalance(env, balanceCollections, "", rp, diskType, maxParallelization, 0, applyBalancing, skippedNodes, nil, volumeIds); err != nil {
 		return fmt.Errorf("re-balance ec shards for collection(s) %v: %w", balanceCollections, err)
 	}
-	if err := verifyEcShardsBeforeDelete(env, volumeIds, diskType, applyBalancing); err != nil {
+	if err := verifyEcShardsBeforeDelete(env, volumeIds, diskType, applyBalancing, rp); err != nil {
 		return fmt.Errorf("verify EC shards before deleting originals: %w", err)
 	}
 	// Past verify the EC copy is recoverable; a delete failure below must not
@@ -727,6 +727,54 @@ func ecShardsClumpedOnOneNode(topoInfo *master_pb.TopologyInfo, vid needle.Volum
 	return "", false
 }
 
+// verifyEcShardAttempts and verifyEcShardRetryInterval bound how long
+// verifyEcShardsBeforeDelete waits for post-rebalance shard locations to
+// propagate through volume-server heartbeats before giving up. They are
+// package vars so tests can shrink the wait.
+var verifyEcShardAttempts = 10
+var verifyEcShardRetryInterval = 2 * time.Second
+
+// ecShardRackByNode maps every volume server's address to its "dc:rack"
+// failure-domain key, matching how the balance planner keys racks.
+func ecShardRackByNode(topoInfo *master_pb.TopologyInfo) map[pb.ServerAddress]string {
+	rackOf := make(map[pb.ServerAddress]string)
+	EachDataNode(topoInfo, func(dc DataCenterId, rack RackId, dn *master_pb.DataNodeInfo) {
+		rackOf[pb.NewServerAddressFromDataNode(dn)] = fmt.Sprintf("%s:%s", dc, rack)
+	})
+	return rackOf
+}
+
+// ecShardPlacementViolations reports where the newest-generation placement of
+// vid breaches the caps the encode asked for: the replica placement's node
+// digit (SameRackCount) bounds the shards of the volume one node may hold and
+// its rack digit (DiffRackCount) the shards one rack may hold. A zero digit
+// leaves that level unbounded; a nil placement binds nothing. byNode counts
+// only the newest encode generation -- a stale orphan's shards do not weaken
+// the fresh set.
+func ecShardPlacementViolations(vid needle.VolumeId, byNode map[pb.ServerAddress]erasure_coding.ShardBits, rackOf map[pb.ServerAddress]string, rp *super_block.ReplicaPlacement) []string {
+	if rp == nil || (rp.SameRackCount <= 0 && rp.DiffRackCount <= 0) {
+		return nil
+	}
+	rackCount := make(map[string]int)
+	var violations []string
+	for addr, bits := range byNode {
+		count := bits.Count()
+		if rp.SameRackCount > 0 && count > rp.SameRackCount {
+			violations = append(violations, fmt.Sprintf("volume %d: %d shards on node %s, over the requested %d per node", vid, count, addr, rp.SameRackCount))
+		}
+		rackCount[rackOf[addr]] += count
+	}
+	if rp.DiffRackCount > 0 {
+		for rack, count := range rackCount {
+			if count > rp.DiffRackCount {
+				violations = append(violations, fmt.Sprintf("volume %d: %d shards on rack %s, over the requested %d per rack", vid, count, rack, rp.DiffRackCount))
+			}
+		}
+	}
+	sort.Strings(violations)
+	return violations
+}
+
 // ecShardSummaryByNode says where a volume's shards are, one entry per node,
 // sorted so the message is stable. It names the ids and not just the count: a
 // set holding shards 0-9 and one holding 4-13 are both "10 shards", and which
@@ -789,7 +837,7 @@ func ecShardSummaryByNode(byNode map[pb.ServerAddress]erasure_coding.ShardBits)
 	return summary
 }
 
-func verifyEcShardsBeforeDelete(env *Env, volumeIds []needle.VolumeId, diskType types.DiskType, expectSpread bool) error {
+func verifyEcShardsBeforeDelete(env *Env, volumeIds []needle.VolumeId, diskType types.DiskType, expectSpread bool, rp *super_block.ReplicaPlacement) error {
 	// Shard relocations from the preceding EC balance reach the master via
 	// volume-server heartbeats, so freshly distributed shards may not all be
 	// visible in the master topology immediately. Poll a few times before
@@ -802,12 +850,16 @@ func verifyEcShardsBeforeDelete(env *Env, volumeIds []needle.VolumeId, diskType
 	// rebalance ran in apply mode), a volume whose shards all still sit on one
 	// node while another node has free slots also aborts the deletion: losing
 	// that node would lose the volume, so the original is the safer copy.
-	const maxAttempts = 10
-	const retryInterval = 2 * time.Second
+	// The replica-placement caps the encode asked for are requirements, not
+	// preferences: a volume still over the node or rack limit aborts the
+	// deletion the same way.
+	maxAttempts := verifyEcShardAttempts
+	retryInterval := verifyEcShardRetryInterval
 
 	var lastErr error
 	var lastDegraded []string
 	var lastClumped []string
+	var lastPlacement []string
 	for attempt := 0; attempt < maxAttempts; attempt++ {
 		topoInfo, _, err := env.FetchTopology(0)
 		if err != nil {
@@ -817,6 +869,11 @@ func verifyEcShardsBeforeDelete(env *Env, volumeIds []needle.VolumeId, diskType
 		lastErr = nil
 		lastDegraded = lastDegraded[:0]
 		lastClumped = lastClumped[:0]
+		lastPlacement = lastPlacement[:0]
+		var rackOf map[pb.ServerAddress]string
+		if rp != nil && (rp.SameRackCount > 0 || rp.DiffRackCount > 0) {
+			rackOf = ecShardRackByNode(topoInfo)
+		}
 		for _, vid := range volumeIds {
 			// Count the shards wherever they landed, as waitForEcShardsToRegister
 			// above already does. generateEcShards writes them beside the source
@@ -852,6 +909,10 @@ func verifyEcShardsBeforeDelete(env *Env, volumeIds []needle.VolumeId, diskType
 					continue
 				}
 			}
+			if violations := ecShardPlacementViolations(vid, byNode, rackOf, rp); len(violations) > 0 {
+				lastPlacement = append(lastPlacement, violations...)
+				continue
+			}
 			if degraded {
 				lastDegraded = append(lastDegraded, fmt.Sprintf("volume %d: %d/%d shards", vid, union.Count(), totalShards))
 				continue
@@ -861,12 +922,12 @@ func verifyEcShardsBeforeDelete(env *Env, volumeIds []needle.VolumeId, diskType
 				vid, union.Count(), totalShards, len(byNode))
 		}
 
-		if lastErr == nil && len(lastDegraded) == 0 && len(lastClumped) == 0 {
+		if lastErr == nil && len(lastDegraded) == 0 && len(lastClumped) == 0 && len(lastPlacement) == 0 {
 			return nil
 		}
 		if attempt < maxAttempts-1 {
-			glog.V(0).Infof("EC shard verification incomplete (attempt %d/%d), waiting for shard locations to propagate: %v %v %v",
-				attempt+1, maxAttempts, lastErr, lastDegraded, lastClumped)
+			glog.V(0).Infof("EC shard verification incomplete (attempt %d/%d), waiting for shard locations to propagate: %v %v %v %v",
+				attempt+1, maxAttempts, lastErr, lastDegraded, lastClumped, lastPlacement)
 			time.Sleep(retryInterval)
 		}
 	}
@@ -878,6 +939,9 @@ func verifyEcShardsBeforeDelete(env *Env, volumeIds []needle.VolumeId, diskType
 	if len(lastClumped) > 0 {
 		return fmt.Errorf("EC shards still sit on a single node after rebalance even though other nodes have free slots (%v); keeping the original volumes. Run ec.balance -apply, verify the spread, then delete the originals or re-run ec.encode", lastClumped)
 	}
+	if len(lastPlacement) > 0 {
+		return fmt.Errorf("EC shard placement does not meet the requested replica placement limits (%v); keeping the original volumes. Loosen -shardReplicaPlacement or add capacity, then re-run ec.encode", lastPlacement)
+	}
 	glog.Warningf("EC shard set incomplete but recoverable after %d attempts, proceeding with source deletion (rebuild missing shards with ec.rebuild): %v",
 		maxAttempts, lastDegraded)
 	return nil

weed/ec/ec_encode_test.go

@@ -1,6 +1,7 @@
 package ec
 
 import (
+	"fmt"
 	"testing"
 	"time"
 
@@ -8,6 +9,7 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
 	"github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding"
 	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
+	"github.com/seaweedfs/seaweedfs/weed/storage/super_block"
 	"github.com/seaweedfs/seaweedfs/weed/storage/types"
 	"github.com/seaweedfs/seaweedfs/weed/util/wildcard"
 	"github.com/stretchr/testify/assert"
@@ -529,3 +531,125 @@ func TestCollectSourceFreeVolumeCountsRefundsClearedShards(t *testing.T) {
 	assert.Equal(t, 1, counts["3-127.0.0.1:8110"],
 		"no refund on a node the sweep skipped as unreachable")
 }
+
+func TestEcShardPlacementViolations(t *testing.T) {
+	node := func(id string, shardBits uint32) *master_pb.DataNodeInfo {
+		return &master_pb.DataNodeInfo{
+			Id: id,
+			DiskInfos: map[string]*master_pb.DiskInfo{
+				"": {
+					MaxVolumeCount: 10,
+					EcShardInfos: []*master_pb.VolumeEcShardInformationMessage{
+						{Id: 1, EcIndexBits: shardBits, DiskId: 0},
+					},
+				},
+			},
+		}
+	}
+	topo := &master_pb.TopologyInfo{
+		DataCenterInfos: []*master_pb.DataCenterInfo{{
+			Id: "dc1",
+			RackInfos: []*master_pb.RackInfo{
+				{Id: "rack1", DataNodeInfos: []*master_pb.DataNodeInfo{
+					node("n1:8080", 0b0011), // shards 0,1
+					node("n2:8080", 0b0100), // shard 2
+				}},
+				{Id: "rack2", DataNodeInfos: []*master_pb.DataNodeInfo{
+					node("n3:8080", 0b1000), // shard 3
+				}},
+			},
+		}},
+	}
+	vid := needle.VolumeId(1)
+	violationsOf := func(topo *master_pb.TopologyInfo, rp *super_block.ReplicaPlacement) []string {
+		return ecShardPlacementViolations(vid, CollectEcShardBitsByNode(topo, vid), ecShardRackByNode(topo), rp)
+	}
+
+	// No caps requested: nothing to violate.
+	assert.Empty(t, violationsOf(topo, nil))
+	assert.Empty(t, violationsOf(topo, &super_block.ReplicaPlacement{}))
+
+	// Node cap: n1 holds 2 shards, over the requested 1 per node.
+	violations := violationsOf(topo, &super_block.ReplicaPlacement{SameRackCount: 1})
+	assert.Len(t, violations, 1)
+	assert.Contains(t, violations[0], "per node")
+
+	// Rack cap: rack1 holds 3 shards, over the requested 2 per rack; the
+	// same-named rack in another dc is a different failure domain.
+	topo2 := &master_pb.TopologyInfo{
+		DataCenterInfos: []*master_pb.DataCenterInfo{
+			{Id: "dc1", RackInfos: []*master_pb.RackInfo{
+				{Id: "rack1", DataNodeInfos: []*master_pb.DataNodeInfo{
+					node("n1:8080", 0b0111), // shards 0,1,2
+				}},
+			}},
+			{Id: "dc2", RackInfos: []*master_pb.RackInfo{
+				{Id: "rack1", DataNodeInfos: []*master_pb.DataNodeInfo{
+					node("n9:8080", 0b1000), // shard 3
+				}},
+			}},
+		},
+	}
+	violations = violationsOf(topo2, &super_block.ReplicaPlacement{DiffRackCount: 2})
+	assert.Len(t, violations, 1)
+	assert.Contains(t, violations[0], "dc1:rack1")
+
+	// Caps that fit the layout: no violations.
+	assert.Empty(t, violationsOf(topo, &super_block.ReplicaPlacement{SameRackCount: 2, DiffRackCount: 3}))
+}
+
+func TestVerifyEcShardsBeforeDeletePlacement(t *testing.T) {
+	oldAttempts, oldInterval := verifyEcShardAttempts, verifyEcShardRetryInterval
+	verifyEcShardAttempts, verifyEcShardRetryInterval = 2, time.Millisecond
+	t.Cleanup(func() {
+		verifyEcShardAttempts, verifyEcShardRetryInterval = oldAttempts, oldInterval
+	})
+
+	vid := needle.VolumeId(1)
+	topologyWith := func(nodeShards map[string]uint32) *master_pb.TopologyInfo {
+		var nodes []*master_pb.DataNodeInfo
+		for id, bits := range nodeShards {
+			si := erasure_coding.NewShardsInfo()
+			for sid := 0; sid < erasure_coding.TotalShardsCount; sid++ {
+				if bits&(1<<sid) != 0 {
+					si.Set(erasure_coding.NewShardInfo(erasure_coding.ShardId(sid), 1))
+				}
+			}
+			nodes = append(nodes, &master_pb.DataNodeInfo{
+				Id: id,
+				DiskInfos: map[string]*master_pb.DiskInfo{"": {
+					MaxVolumeCount: 100,
+					EcShardInfos: []*master_pb.VolumeEcShardInformationMessage{{
+						Id:          uint32(vid),
+						EcIndexBits: si.Bitmap(),
+						ShardSizes:  si.SizesInt64(),
+						EncodeTsNs:  1,
+					}},
+				}},
+			})
+		}
+		return &master_pb.TopologyInfo{DataCenterInfos: []*master_pb.DataCenterInfo{{
+			Id:        "dc1",
+			RackInfos: []*master_pb.RackInfo{{Id: "rack1", DataNodeInfos: nodes}},
+		}}}
+	}
+	envWith := func(topo *master_pb.TopologyInfo) *Env {
+		return &Env{FetchTopology: func(delay time.Duration) (*master_pb.TopologyInfo, uint64, error) {
+			return topo, 0, nil
+		}}
+	}
+	rp := &super_block.ReplicaPlacement{SameRackCount: 1}
+
+	// One shard per node: the requested cap is met, deletion is allowed.
+	onePerNode := map[string]uint32{}
+	for i := 0; i < erasure_coding.TotalShardsCount; i++ {
+		onePerNode[fmt.Sprintf("n%d:8080", i)] = 1 << i
+	}
+	assert.NoError(t, verifyEcShardsBeforeDelete(envWith(topologyWith(onePerNode)), []needle.VolumeId{vid}, types.HardDriveType, false, rp))
+
+	// All 14 shards on one node: over the cap, so the gate refuses to let the
+	// source volume be deleted.
+	err := verifyEcShardsBeforeDelete(envWith(topologyWith(map[string]uint32{"n1:8080": 0x3FFF})), []needle.VolumeId{vid}, types.HardDriveType, false, rp)
+	assert.Error(t, err)
+	assert.Contains(t, err.Error(), "does not meet the requested replica placement limits")
+}

weed/shell/command_ec_encode.go

@@ -54,7 +54,8 @@ func (c *commandEcEncode) Help() string {
 	per node (the data-center digit is not used for EC). For example
 	-shardReplicaPlacement=021 requests at most 1 shard per node and 2 per rack.
 	Only when the final placement meets these limits does losing one rack cost
-	at most 2 shards; the command does not guarantee that the limits are met.
+	at most 2 shards, so the encode fails (and keeps the original volumes)
+	when the cluster cannot honor them.
 
 	The -collection parameter is a comma-separated list of collection names, with
 	"*" and "?" wildcards, and regex patterns: