diff --git a/weed/pb/worker.proto b/weed/pb/worker.proto index a1ba6c5fa..51a84c007 100644 --- a/weed/pb/worker.proto +++ b/weed/pb/worker.proto @@ -374,6 +374,7 @@ message ErasureCodingTaskConfig { int32 min_volume_size_mb = 3; // Minimum volume size for EC string collection_filter = 4; // Only process volumes from specific collections repeated string preferred_tags = 5; // Disk tags to prioritize for EC shard placement + string replica_placement = 6; // EC shard replica placement (e.g. "020"); empty falls back to master default replication } // BalanceTaskConfig contains balance-specific configuration diff --git a/weed/pb/worker_pb/worker.pb.go b/weed/pb/worker_pb/worker.pb.go index e0284dee7..d3f0b3628 100644 --- a/weed/pb/worker_pb/worker.pb.go +++ b/weed/pb/worker_pb/worker.pb.go @@ -2960,6 +2960,7 @@ type ErasureCodingTaskConfig struct { MinVolumeSizeMb int32 `protobuf:"varint,3,opt,name=min_volume_size_mb,json=minVolumeSizeMb,proto3" json:"min_volume_size_mb,omitempty"` // Minimum volume size for EC CollectionFilter string `protobuf:"bytes,4,opt,name=collection_filter,json=collectionFilter,proto3" json:"collection_filter,omitempty"` // Only process volumes from specific collections PreferredTags []string `protobuf:"bytes,5,rep,name=preferred_tags,json=preferredTags,proto3" json:"preferred_tags,omitempty"` // Disk tags to prioritize for EC shard placement + ReplicaPlacement string `protobuf:"bytes,6,opt,name=replica_placement,json=replicaPlacement,proto3" json:"replica_placement,omitempty"` // EC shard replica placement (e.g. "020"); empty falls back to master default replication unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -3029,6 +3030,13 @@ func (x *ErasureCodingTaskConfig) GetPreferredTags() []string { return nil } +func (x *ErasureCodingTaskConfig) GetReplicaPlacement() string { + if x != nil { + return x.ReplicaPlacement + } + return "" +} + // BalanceTaskConfig contains balance-specific configuration type BalanceTaskConfig struct { state protoimpl.MessageState `protogen:"open.v1"` @@ -4218,13 +4226,14 @@ const file_worker_proto_rawDesc = "" + "\x10VacuumTaskConfig\x12+\n" + "\x11garbage_threshold\x18\x01 \x01(\x01R\x10garbageThreshold\x12/\n" + "\x14min_volume_age_hours\x18\x02 \x01(\x05R\x11minVolumeAgeHours\x120\n" + - "\x14min_interval_seconds\x18\x03 \x01(\x05R\x12minIntervalSeconds\"\xed\x01\n" + + "\x14min_interval_seconds\x18\x03 \x01(\x05R\x12minIntervalSeconds\"\x9a\x02\n" + "\x17ErasureCodingTaskConfig\x12%\n" + "\x0efullness_ratio\x18\x01 \x01(\x01R\rfullnessRatio\x12*\n" + "\x11quiet_for_seconds\x18\x02 \x01(\x05R\x0fquietForSeconds\x12+\n" + "\x12min_volume_size_mb\x18\x03 \x01(\x05R\x0fminVolumeSizeMb\x12+\n" + "\x11collection_filter\x18\x04 \x01(\tR\x10collectionFilter\x12%\n" + - "\x0epreferred_tags\x18\x05 \x03(\tR\rpreferredTags\"n\n" + + "\x0epreferred_tags\x18\x05 \x03(\tR\rpreferredTags\x12+\n" + + "\x11replica_placement\x18\x06 \x01(\tR\x10replicaPlacement\"n\n" + "\x11BalanceTaskConfig\x12/\n" + "\x13imbalance_threshold\x18\x01 \x01(\x01R\x12imbalanceThreshold\x12(\n" + "\x10min_server_count\x18\x02 \x01(\x05R\x0eminServerCount\"I\n" + diff --git a/weed/worker/tasks/erasure_coding/config.go b/weed/worker/tasks/erasure_coding/config.go index 9d03fdaf3..48371bb07 100644 --- a/weed/worker/tasks/erasure_coding/config.go +++ b/weed/worker/tasks/erasure_coding/config.go @@ -17,6 +17,7 @@ type Config struct { CollectionFilter string `json:"collection_filter"` MinSizeMB int `json:"min_size_mb"` PreferredTags []string `json:"preferred_tags"` + ReplicaPlacement string `json:"replica_placement"` // e.g. "020"; empty falls back to the master default replication } // NewDefaultConfig creates a new default erasure coding configuration @@ -157,6 +158,19 @@ func GetConfigSpec() base.ConfigSpec { InputType: "text", CSSClasses: "form-control", }, + { + Name: "replica_placement", + JSONName: "replica_placement", + Type: config.FieldTypeString, + DefaultValue: "", + Required: false, + DisplayName: "Replica Placement", + Description: "EC shard replica placement constraint (e.g. 020)", + HelpText: "Leave empty to use the master default replication. When set, the 2nd/3rd digits cap EC shards per rack and per node (best-effort during encode: relaxed rather than failing if the cluster can't satisfy them, then enforced by rebalancing). The 1st (data-center) digit is ignored for EC placement", + Placeholder: "020", + InputType: "text", + CSSClasses: "form-control", + }, }, } } @@ -177,6 +191,7 @@ func (c *Config) ToTaskPolicy() *worker_pb.TaskPolicy { MinVolumeSizeMb: int32(c.MinSizeMB), CollectionFilter: c.CollectionFilter, PreferredTags: preferredTagsCopy, + ReplicaPlacement: c.ReplicaPlacement, }, }, } @@ -200,6 +215,7 @@ func (c *Config) FromTaskPolicy(policy *worker_pb.TaskPolicy) error { c.MinSizeMB = int(ecConfig.MinVolumeSizeMb) c.CollectionFilter = ecConfig.CollectionFilter c.PreferredTags = append([]string(nil), ecConfig.PreferredTags...) + c.ReplicaPlacement = ecConfig.ReplicaPlacement } return nil diff --git a/weed/worker/tasks/erasure_coding/detection.go b/weed/worker/tasks/erasure_coding/detection.go index c333032b7..dd25129c4 100644 --- a/weed/worker/tasks/erasure_coding/detection.go +++ b/weed/worker/tasks/erasure_coding/detection.go @@ -14,7 +14,9 @@ import ( "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/ecbalancer" "github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding/placement" + "github.com/seaweedfs/seaweedfs/weed/storage/super_block" "github.com/seaweedfs/seaweedfs/weed/util" "github.com/seaweedfs/seaweedfs/weed/util/wildcard" "github.com/seaweedfs/seaweedfs/weed/worker/tasks/base" @@ -56,7 +58,17 @@ func Detection(ctx context.Context, metrics []*types.VolumeHealthMetrics, cluste skippedTooFewNodes := 0 consecutivePlanningFailures := 0 - var planner *ecPlacementPlanner + // EC shard replica placement: explicit config wins, else the master default. + var replicaPlacement *super_block.ReplicaPlacement + if clusterInfo != nil { + replicaPlacement = super_block.ResolveReplicaPlacement(ecConfig.ReplicaPlacement, clusterInfo.DefaultReplicaPlacement) + } + // EC placement honors only the rack/node digits; the data-center digit can't + // express a useful per-DC EC shard cap (it maxes at 2). Warn once per cycle so a + // 1xx/2xx setting isn't silently ineffective. + if replicaPlacement != nil && replicaPlacement.DiffDataCenterCount > 0 { + glog.Warningf("EC Detection: replica placement data-center digit (%d) is ignored for EC; only rack/node digits are honored", replicaPlacement.DiffDataCenterCount) + } allowedCollections := wildcard.CompileWildcardMatchers(ecConfig.CollectionFilter) @@ -219,12 +231,9 @@ func Detection(ctx context.Context, metrics []*types.VolumeHealthMetrics, cluste } glog.Infof("EC Detection: ActiveTopology available, planning destinations for volume %d", metric.VolumeID) - if planner == nil { - planner = newECPlacementPlanner(clusterInfo.ActiveTopology, ecConfig.PreferredTags) - } dataShards := erasure_coding.DataShardsCount parityShards := erasure_coding.ParityShardsCount - multiPlan, err := planECDestinations(planner, metric, ecConfig, dataShards, parityShards) + multiPlan, shardsPerPlan, err := planECDestinations(clusterInfo.ActiveTopology, metric, ecConfig, replicaPlacement, dataShards, parityShards) if err != nil { glog.V(2).Infof("Failed to plan EC destinations for volume %d: %v", metric.VolumeID, err) consecutivePlanningFailures++ @@ -304,14 +313,13 @@ func Detection(ctx context.Context, metrics []*types.VolumeHealthMetrics, cluste glog.V(2).Infof("Found %d volume replicas and %d existing EC shards for volume %d (total %d cleanup sources)", len(replicaLocations), len(existingECShards), metric.VolumeID, len(sources)) - // Convert shard destinations to TaskDestinationSpec. With fewer - // disks than shards a destination holds several shards, so reserve - // capacity for the actual per-disk shard count (round-robin matches - // createECTargets) rather than assuming one shard each. + // Convert shard destinations to TaskDestinationSpec. A destination may + // hold several shards (small clusters), so reserve capacity for the + // actual per-disk shard count that Place assigned (shardsPerPlan), + // which is exactly what createECTargets writes. destinations := make([]topology.TaskDestinationSpec, len(shardDestinations)) - shardsPerDest := distributeECShards(dataShards+parityShards, len(shardDestinations)) for i, dest := range shardDestinations { - shardCount := len(shardsPerDest[i]) + shardCount := len(shardsPerPlan[i]) shardImpact := topology.CalculateECShardStorageImpact(int32(shardCount), int64(expectedShardSize)) destSize := int64(expectedShardSize) * int64(shardCount) destinations[i] = topology.TaskDestinationSpec{ @@ -342,9 +350,9 @@ func Detection(ctx context.Context, metrics []*types.VolumeHealthMetrics, cluste continue // Skip this volume if topology task addition fails } - if planner != nil { - planner.applyTaskReservations(int64(metric.Size), sources, destinations) - } + // Cross-volume in-cycle capacity is tracked by ActiveTopology via the + // pending task above, which the next volume's FromActiveTopology snapshot + // reflects; no separate planner reservation is needed. glog.V(2).Infof("Added pending EC shard task %s to ActiveTopology for volume %d with %d cleanup sources and %d shard destinations", taskID, metric.VolumeID, len(sources), len(multiPlan.Plans)) @@ -360,7 +368,7 @@ func Detection(ctx context.Context, metrics []*types.VolumeHealthMetrics, cluste Sources: sourcesProto, // Unified targets - all EC shard destinations - Targets: createECTargets(multiPlan, dataShards, parityShards), + Targets: createECTargets(multiPlan, shardsPerPlan), TaskParams: &worker_pb.TaskParams_ErasureCodingParams{ ErasureCodingParams: createECTaskParams(dataShards, parityShards, metric.DiskType), @@ -699,12 +707,20 @@ func countTopologyNodes(at *topology.ActiveTopology) int { return n } -func planECDestinations(planner *ecPlacementPlanner, metric *types.VolumeHealthMetrics, ecConfig *Config, dataShards, parityShards int) (*topology.MultiDestinationPlan, error) { - if planner == nil || planner.activeTopology == nil { - return nil, fmt.Errorf("active topology not available for EC placement") +// planECDestinations places all shards of the volume via the shared ecbalancer +// policy and returns the per-disk destination plans plus, parallel to them, the +// shard ids ecbalancer.Place assigned to each disk (so createECTargets and the +// capacity reservations use the real assignment, not a round-robin guess). +// +// Encode is lenient (PlaceDurabilityFirst): it relaxes caps/anti-affinity/RP as +// needed rather than fail, and prefers the source disk type but spills if that +// type can't hold every shard. rp is the resolved replica placement (may be nil). +func planECDestinations(at *topology.ActiveTopology, metric *types.VolumeHealthMetrics, ecConfig *Config, rp *super_block.ReplicaPlacement, dataShards, parityShards int) (*topology.MultiDestinationPlan, [][]uint32, error) { + if at == nil { + return nil, nil, fmt.Errorf("active topology not available for EC placement") } if dataShards <= 0 || parityShards <= 0 { - return nil, fmt.Errorf("invalid EC ratio: dataShards=%d parityShards=%d", dataShards, parityShards) + return nil, nil, fmt.Errorf("invalid EC ratio: dataShards=%d parityShards=%d", dataShards, parityShards) } totalShards := dataShards + parityShards // Survive losing one disk: each disk holds at most parityShards shards, @@ -712,105 +728,103 @@ func planECDestinations(planner *ecPlacementPlanner, metric *types.VolumeHealthM minTotalDisks := (totalShards + parityShards - 1) / parityShards expectedShardSize := uint64(metric.Size) / uint64(dataShards) - // Get source node information from topology - var sourceRack, sourceDC string - - // Extract rack and DC from topology info - topologyInfo := planner.activeTopology.GetTopologyInfo() - if topologyInfo != nil { - for _, dc := range topologyInfo.DataCenterInfos { - for _, rack := range dc.RackInfos { - for _, dataNodeInfo := range rack.DataNodeInfos { - if dataNodeInfo.Id == metric.Server { - sourceDC = dc.Id - sourceRack = rack.Id - break - } - } - if sourceRack != "" { - break - } - } - if sourceDC != "" { - break - } - } + snap := ecbalancer.FromActiveTopology(at, dataShards) + // Encode is greenfield: any EC shards already present for this volume are stale + // leftovers from a prior failed attempt, which the task deletes + // (cleanupStaleEcShards) before distributing the new shards. Release them so they + // don't occupy capacity or skew anti-affinity / per-disk caps during planning. + snap.ReleaseVolumeShards(metric.Collection, metric.VolumeID) + need := make([]int, totalShards) + for i := range need { + need[i] = i } - - // Select best disks for EC placement with rack/DC diversity using the cached planner. - // Pass source disk type so placement prefers matching-type disks (#9423). - selectedDisks, err := planner.selectDestinations(sourceRack, sourceDC, metric.DiskType, totalShards) + res, err := snap.Place(metric.VolumeID, metric.Collection, need, ecbalancer.Constraints{ + DiskType: metric.DiskType, + DiskTypePolicy: ecbalancer.DiskTypePrefer, + PreferredTags: ecConfig.PreferredTags, + ReplicaPlacement: rp, + Ratio: func(string) (int, int) { return dataShards, parityShards }, + }, ecbalancer.PlaceDurabilityFirst) if err != nil { - return nil, err + return nil, nil, err } - if len(selectedDisks) < minTotalDisks { - return nil, fmt.Errorf("found %d disks, but EC %d+%d needs at least %d disks so no disk holds more than %d shards", - len(selectedDisks), dataShards, parityShards, minTotalDisks, parityShards) + if res.SpilledToOtherDiskType { + glog.Warningf("EC volume %d: placed shards outside preferred disk type %q", metric.VolumeID, metric.DiskType) } - // Fewer than totalShards disks is fine: createECTargets round-robins the - // shards across the available disks, packing several distinct shards onto a - // disk when needed (matching ec.encode's "spread as 4,4,3,3" fallback for - // small clusters). A disk holding several shards of one volume is safe — - // each is a separate .ecNN file and ReceiveFile keys by that extension. The - // minTotalDisks floor above keeps any single disk under parityShards shards, - // so the volume still survives losing any one disk. - if len(selectedDisks) < totalShards { - glog.V(1).Infof("EC volume %d: only %d disks for %d shards, packing up to %d shards per disk", - metric.VolumeID, len(selectedDisks), totalShards, (totalShards+len(selectedDisks)-1)/len(selectedDisks)) + if res.SpilledOutsidePreferredTags { + glog.Warningf("EC volume %d: placed shards outside preferred tags %v", metric.VolumeID, ecConfig.PreferredTags) + } + if len(res.Relaxed) > 0 { + // Encode is best-effort (PlaceDurabilityFirst): it relaxes these constraints + // rather than defer when the cluster can't satisfy them. Surface it so a tight + // replica placement isn't silently weakened; rebalancing tightens the spread. + glog.Warningf("EC volume %d: placed with relaxed constraints %v; replica placement not fully satisfied (rebalancing will adjust)", metric.VolumeID, res.Relaxed) + } + + // Group the per-shard destinations into one plan per (node,disk), iterating + // shard ids in order for determinism. + type diskGroup struct { + node, rack, dc string + diskID uint32 + shards []uint32 + } + type diskKey struct { + node string + diskID uint32 + } + groups := make(map[diskKey]*diskGroup, totalShards) + order := make([]diskKey, 0, totalShards) + for sid := 0; sid < totalShards; sid++ { + d, ok := res.Destinations[sid] + if !ok { + return nil, nil, fmt.Errorf("EC volume %d: shard %d was not placed", metric.VolumeID, sid) + } + key := diskKey{node: d.Node, diskID: d.DiskID} + g := groups[key] + if g == nil { + g = &diskGroup{node: d.Node, rack: d.Rack, dc: d.DataCenter, diskID: d.DiskID} + groups[key] = g + order = append(order, key) + } + g.shards = append(g.shards, uint32(sid)) + } + if len(order) < minTotalDisks { + return nil, nil, fmt.Errorf("placed onto %d disks, but EC %d+%d needs at least %d so no disk holds more than %d shards", + len(order), dataShards, parityShards, minTotalDisks, parityShards) } var plans []*topology.DestinationPlan + shardsPerPlan := make([][]uint32, 0, len(order)) rackCount := make(map[string]int) dcCount := make(map[string]int) - - for _, disk := range selectedDisks { - // Get the target server address - targetAddress, err := workerutil.ResolveServerAddress(disk.NodeID, planner.activeTopology) + for _, key := range order { + g := groups[key] + targetAddress, err := workerutil.ResolveServerAddress(g.node, at) if err != nil { - return nil, fmt.Errorf("failed to resolve address for target server %s: %v", disk.NodeID, err) + return nil, nil, fmt.Errorf("failed to resolve address for target server %s: %v", g.node, err) } - - plan := &topology.DestinationPlan{ - TargetNode: disk.NodeID, - TargetAddress: targetAddress, - TargetDisk: disk.DiskID, - TargetRack: disk.Rack, - TargetDC: disk.DataCenter, - ExpectedSize: expectedShardSize, // Set calculated EC shard size - PlacementScore: calculateECScoreCandidate(disk, sourceRack, sourceDC), - } - plans = append(plans, plan) - - // Count rack and DC diversity - rackKey := fmt.Sprintf("%s:%s", disk.DataCenter, disk.Rack) - rackCount[rackKey]++ - dcCount[disk.DataCenter]++ + plans = append(plans, &topology.DestinationPlan{ + TargetNode: g.node, + TargetAddress: targetAddress, + TargetDisk: g.diskID, + TargetRack: g.rack, + TargetDC: g.dc, + ExpectedSize: expectedShardSize, + }) + shardsPerPlan = append(shardsPerPlan, g.shards) + rackCount[fmt.Sprintf("%s:%s", g.dc, g.rack)]++ + dcCount[g.dc]++ } - // Log capacity utilization information using ActiveTopology's encapsulated logic - totalEffectiveCapacity := int64(0) - for _, plan := range plans { - key := ecDiskKey(plan.TargetNode, plan.TargetDisk) - if candidate, ok := planner.candidateByKey[key]; ok { - totalEffectiveCapacity += int64(candidate.FreeSlots) - } - } - - glog.V(1).Infof("Planned EC destinations for volume %d (size=%d bytes): expected shard size=%d bytes, %d shards across %d racks, %d DCs, total effective capacity=%d slots", - metric.VolumeID, metric.Size, expectedShardSize, len(plans), len(rackCount), len(dcCount), totalEffectiveCapacity) - - // Log storage impact for EC task (source only - EC has multiple targets handled individually) - sourceChange, _ := topology.CalculateTaskStorageImpact(topology.TaskTypeErasureCoding, int64(metric.Size)) - glog.V(2).Infof("EC task capacity management: source_reserves_with_zero_impact={VolumeSlots:%d, ShardSlots:%d}, %d_targets_will_receive_shards, estimated_size=%d", - sourceChange.VolumeSlots, sourceChange.ShardSlots, len(plans), metric.Size) - glog.V(2).Infof("EC source reserves capacity but with zero StorageSlotChange impact") + glog.V(1).Infof("Planned EC destinations for volume %d (size=%d bytes): expected shard size=%d bytes, %d shards across %d disks, %d racks, %d DCs", + metric.VolumeID, metric.Size, expectedShardSize, totalShards, len(plans), len(rackCount), len(dcCount)) return &topology.MultiDestinationPlan{ Plans: plans, - TotalShards: len(plans), + TotalShards: totalShards, SuccessfulRack: len(rackCount), SuccessfulDCs: len(dcCount), - }, nil + }, shardsPerPlan, nil } // distributeECShards assigns shard ids 0..totalShards-1 across numTargets @@ -830,41 +844,22 @@ func distributeECShards(totalShards, numTargets int) [][]uint32 { return targetShards } -// createECTargets builds TaskTargets, round-robining shards across the plan -// entries. With fewer disks than shards a target receives several shard ids. -func createECTargets(multiPlan *topology.MultiDestinationPlan, dataShards, parityShards int) []*worker_pb.TaskTarget { - var targets []*worker_pb.TaskTarget - numTargets := len(multiPlan.Plans) - totalShards := dataShards + parityShards - - targetShards := distributeECShards(totalShards, numTargets) - +// createECTargets builds TaskTargets from the per-disk plans and the shard ids +// ecbalancer.Place assigned to each (shardsPerPlan is parallel to multiPlan.Plans). +func createECTargets(multiPlan *topology.MultiDestinationPlan, shardsPerPlan [][]uint32) []*worker_pb.TaskTarget { + targets := make([]*worker_pb.TaskTarget, 0, len(multiPlan.Plans)) for i, plan := range multiPlan.Plans { - target := &worker_pb.TaskTarget{ + shardIDs := shardsPerPlan[i] + targets = append(targets, &worker_pb.TaskTarget{ Node: plan.TargetAddress, DiskId: plan.TargetDisk, Rack: plan.TargetRack, DataCenter: plan.TargetDC, - ShardIds: targetShards[i], + ShardIds: shardIDs, EstimatedSize: plan.ExpectedSize, - } - targets = append(targets, target) - - assignedData := make([]uint32, 0) - assignedParity := make([]uint32, 0) - for _, shardId := range targetShards[i] { - if int(shardId) < dataShards { - assignedData = append(assignedData, shardId) - } else { - assignedParity = append(assignedParity, shardId) - } - } - glog.V(2).Infof("EC planning: target %s assigned shards %v (data: %v, parity: %v)", - plan.TargetNode, targetShards[i], assignedData, assignedParity) + }) + glog.V(2).Infof("EC planning: target %s disk %d assigned shards %v", plan.TargetNode, plan.TargetDisk, shardIDs) } - - glog.V(1).Infof("EC planning: distributed %d shards across %d targets using round-robin (data shards 0-%d, parity shards %d-%d)", - totalShards, numTargets, dataShards-1, dataShards, totalShards-1) return targets } diff --git a/weed/worker/tasks/erasure_coding/detection_disk_type_test.go b/weed/worker/tasks/erasure_coding/detection_disk_type_test.go index efee93aca..45b79bae6 100644 --- a/weed/worker/tasks/erasure_coding/detection_disk_type_test.go +++ b/weed/worker/tasks/erasure_coding/detection_disk_type_test.go @@ -19,9 +19,6 @@ func TestPlanECDestinationsPrefersSourceDiskType_FullCluster(t *testing.T) { // for a 10+4 layout with one-shard-per-(server,disk) diversity. activeTopology := buildActiveTopology(t, erasure_coding.TotalShardsCount, []string{"hdd", "ssd"}, 100, 0) - planner := newECPlacementPlanner(activeTopology, nil) - require.NotNil(t, planner) - metric := &types.VolumeHealthMetrics{ VolumeID: 1, Server: "10.0.0.1:8080", @@ -30,7 +27,7 @@ func TestPlanECDestinationsPrefersSourceDiskType_FullCluster(t *testing.T) { DiskType: "ssd", // the property being plumbed end-to-end } - plan, err := planECDestinations(planner, metric, NewDefaultConfig(), erasure_coding.DataShardsCount, erasure_coding.ParityShardsCount) + plan, _, err := planECDestinations(activeTopology, metric, NewDefaultConfig(), nil, erasure_coding.DataShardsCount, erasure_coding.ParityShardsCount) require.NoError(t, err) require.Len(t, plan.Plans, erasure_coding.TotalShardsCount) @@ -67,9 +64,6 @@ func TestPlanECDestinationsSpillsToOtherDiskType_WhenPreferredScarce(t *testing. } require.NoError(t, activeTopology.UpdateTopology(topo)) - planner := newECPlacementPlanner(activeTopology, nil) - require.NotNil(t, planner) - metric := &types.VolumeHealthMetrics{ VolumeID: 2, Server: "10.0.0.1:8080", @@ -78,7 +72,7 @@ func TestPlanECDestinationsSpillsToOtherDiskType_WhenPreferredScarce(t *testing. DiskType: "ssd", } - plan, err := planECDestinations(planner, metric, NewDefaultConfig(), erasure_coding.DataShardsCount, erasure_coding.ParityShardsCount) + plan, _, err := planECDestinations(activeTopology, metric, NewDefaultConfig(), nil, erasure_coding.DataShardsCount, erasure_coding.ParityShardsCount) require.NoError(t, err) require.Len(t, plan.Plans, erasure_coding.TotalShardsCount) diff --git a/weed/worker/tasks/erasure_coding/detection_test.go b/weed/worker/tasks/erasure_coding/detection_test.go index df2e2edaa..118f498a7 100644 --- a/weed/worker/tasks/erasure_coding/detection_test.go +++ b/weed/worker/tasks/erasure_coding/detection_test.go @@ -47,8 +47,6 @@ func TestECPlacementPlannerApplyReservations(t *testing.T) { func TestPlanECDestinationsUsesPlanner(t *testing.T) { activeTopology := buildActiveTopology(t, 7, []string{"hdd", "ssd"}, 100, 0) - planner := newECPlacementPlanner(activeTopology, nil) - require.NotNil(t, planner) metric := &types.VolumeHealthMetrics{ VolumeID: 1, @@ -57,10 +55,10 @@ func TestPlanECDestinationsUsesPlanner(t *testing.T) { Collection: "", } - plan, err := planECDestinations(planner, metric, NewDefaultConfig(), erasure_coding.DataShardsCount, erasure_coding.ParityShardsCount) + plan, shardsPerPlan, err := planECDestinations(activeTopology, metric, NewDefaultConfig(), nil, erasure_coding.DataShardsCount, erasure_coding.ParityShardsCount) require.NoError(t, err) require.NotNil(t, plan) - assert.Equal(t, erasure_coding.TotalShardsCount, len(plan.Plans)) + requireAllShardsPlaced(t, plan, shardsPerPlan) } func TestECPlacementPlannerPrefersTaggedDisks(t *testing.T) { @@ -363,9 +361,6 @@ func TestPlanECDestinationsSpreadsAcrossPhysicalDisks(t *testing.T) { }}, })) - planner := newECPlacementPlanner(activeTopology, nil) - require.NotNil(t, planner) - metric := &types.VolumeHealthMetrics{ VolumeID: 42, Server: "127.0.0.1:8081", @@ -373,23 +368,14 @@ func TestPlanECDestinationsSpreadsAcrossPhysicalDisks(t *testing.T) { Collection: "", } - plan, err := planECDestinations(planner, metric, NewDefaultConfig(), erasure_coding.DataShardsCount, erasure_coding.ParityShardsCount) + plan, shardsPerPlan, err := planECDestinations(activeTopology, metric, NewDefaultConfig(), nil, erasure_coding.DataShardsCount, erasure_coding.ParityShardsCount) require.NoError(t, err) require.NotNil(t, plan) - require.Equal(t, erasure_coding.TotalShardsCount, len(plan.Plans)) - - seen := make(map[string]bool, len(plan.Plans)) - for _, p := range plan.Plans { - key := fmt.Sprintf("%s:%d", p.TargetNode, p.TargetDisk) - assert.False(t, seen[key], "duplicate (server,disk_id) target %s", key) - seen[key] = true - } + requireAllShardsPlaced(t, plan, shardsPerPlan) } func TestPlanECDestinationsFailsWithInsufficientCapacity(t *testing.T) { activeTopology := buildActiveTopology(t, 1, []string{"hdd"}, 1, 1) - planner := newECPlacementPlanner(activeTopology, nil) - require.NotNil(t, planner) metric := &types.VolumeHealthMetrics{ VolumeID: 2, @@ -398,7 +384,7 @@ func TestPlanECDestinationsFailsWithInsufficientCapacity(t *testing.T) { Collection: "", } - _, err := planECDestinations(planner, metric, NewDefaultConfig(), erasure_coding.DataShardsCount, erasure_coding.ParityShardsCount) + _, _, err := planECDestinations(activeTopology, metric, NewDefaultConfig(), nil, erasure_coding.DataShardsCount, erasure_coding.ParityShardsCount) require.Error(t, err) } @@ -440,9 +426,6 @@ func TestPlanECDestinationsPacksWhenFewerDisksThanShards(t *testing.T) { DataCenterInfos: []*master_pb.DataCenterInfo{{Id: "dc1", RackInfos: rackInfos}}, })) - planner := newECPlacementPlanner(activeTopology, nil) - require.NotNil(t, planner) - metric := &types.VolumeHealthMetrics{ VolumeID: 4569, Server: "192.168.1.145:8081", @@ -450,16 +433,18 @@ func TestPlanECDestinationsPacksWhenFewerDisksThanShards(t *testing.T) { Collection: "", } - plan, err := planECDestinations(planner, metric, NewDefaultConfig(), erasure_coding.DataShardsCount, erasure_coding.ParityShardsCount) + plan, shardsPerPlan, err := planECDestinations(activeTopology, metric, NewDefaultConfig(), nil, erasure_coding.DataShardsCount, erasure_coding.ParityShardsCount) require.NoError(t, err) require.NotNil(t, plan) - // One plan entry per available disk; fewer than the 14 shards. - require.Equal(t, numServers, len(plan.Plans)) + // Packed onto the available disks: more than one shard per disk but never more + // than the 8 disks, and at least the durability floor of distinct disks. + require.LessOrEqual(t, len(plan.Plans), numServers) + require.GreaterOrEqual(t, len(plan.Plans), (erasure_coding.TotalShardsCount+erasure_coding.ParityShardsCount-1)/erasure_coding.ParityShardsCount) // createECTargets must cover all 14 shards exactly once, packing onto the // available disks without any disk exceeding parityShards shards. - targets := createECTargets(plan, erasure_coding.DataShardsCount, erasure_coding.ParityShardsCount) - require.Equal(t, numServers, len(targets)) + targets := createECTargets(plan, shardsPerPlan) + require.Equal(t, len(plan.Plans), len(targets)) seenShards := make(map[uint32]bool) for _, target := range targets { @@ -473,6 +458,28 @@ func TestPlanECDestinationsPacksWhenFewerDisksThanShards(t *testing.T) { require.Len(t, seenShards, erasure_coding.TotalShardsCount, "every shard must be placed exactly once") } +// requireAllShardsPlaced asserts every EC shard landed exactly once, on a distinct +// (node,disk) target, with no disk holding more than parityShards shards (so losing +// any one disk cannot lose the volume). shardsPerPlan is parallel to plan.Plans. +func requireAllShardsPlaced(t *testing.T, plan *topology.MultiDestinationPlan, shardsPerPlan [][]uint32) { + t.Helper() + require.Equal(t, len(plan.Plans), len(shardsPerPlan), "one shard list per plan entry") + keys := make(map[string]bool, len(plan.Plans)) + seen := make(map[uint32]bool) + for i, p := range plan.Plans { + key := fmt.Sprintf("%s:%d", p.TargetNode, p.TargetDisk) + require.False(t, keys[key], "duplicate (node,disk) target %s", key) + keys[key] = true + require.LessOrEqual(t, len(shardsPerPlan[i]), erasure_coding.ParityShardsCount, + "disk %s holds %d shards, over parityShards", key, len(shardsPerPlan[i])) + for _, s := range shardsPerPlan[i] { + require.False(t, seen[s], "shard %d placed more than once", s) + seen[s] = true + } + } + require.Len(t, seen, erasure_coding.TotalShardsCount, "every shard must be placed exactly once") +} + func buildVolumeMetricsForIDs(count int) []*types.VolumeHealthMetrics { metrics := make([]*types.VolumeHealthMetrics, 0, count) now := time.Now() diff --git a/weed/worker/tasks/erasure_coding/ec_task.go b/weed/worker/tasks/erasure_coding/ec_task.go index 89ca6830c..cb2304bef 100644 --- a/weed/worker/tasks/erasure_coding/ec_task.go +++ b/weed/worker/tasks/erasure_coding/ec_task.go @@ -271,8 +271,17 @@ func (t *ErasureCodingTask) Validate(params *worker_pb.TaskParams) error { return fmt.Errorf("invalid parity shards: %d (must be >= 1)", ecParams.ParityShards) } - if len(params.Targets) < int(ecParams.DataShards+ecParams.ParityShards) { - return fmt.Errorf("insufficient targets: got %d, need %d", len(params.Targets), ecParams.DataShards+ecParams.ParityShards) + // Count distinct shard ids across targets, not target rows: Place packs several + // shards onto one (node,disk) target when there are fewer disks than shards, so + // a valid plan can have fewer target rows than total shards. + distinctShards := make(map[uint32]struct{}) + for _, target := range params.Targets { + for _, sid := range target.ShardIds { + distinctShards[sid] = struct{}{} + } + } + if total := int(ecParams.DataShards + ecParams.ParityShards); len(distinctShards) < total { + return fmt.Errorf("insufficient shard targets: got %d distinct shards across %d targets, need %d", len(distinctShards), len(params.Targets), total) } return nil diff --git a/weed/worker/tasks/erasure_coding/plugin_handler.go b/weed/worker/tasks/erasure_coding/plugin_handler.go index ef1991bb5..e0fed1fa4 100644 --- a/weed/worker/tasks/erasure_coding/plugin_handler.go +++ b/weed/worker/tasks/erasure_coding/plugin_handler.go @@ -138,6 +138,14 @@ func (h *ErasureCodingHandler) Descriptor() *plugin_pb.JobTypeDescriptor { FieldType: plugin_pb.ConfigFieldType_CONFIG_FIELD_TYPE_STRING, Widget: plugin_pb.ConfigWidget_CONFIG_WIDGET_TEXT, }, + { + Name: "replica_placement", + Label: "Replica Placement", + Description: "EC shard placement (e.g. 020): 2nd/3rd digits cap shards per rack/node (best-effort during encode, enforced by rebalancing); the data-center digit is ignored. Empty uses the master default.", + Placeholder: "020", + FieldType: plugin_pb.ConfigFieldType_CONFIG_FIELD_TYPE_STRING, + Widget: plugin_pb.ConfigWidget_CONFIG_WIDGET_TEXT, + }, }, }, }, @@ -154,6 +162,9 @@ func (h *ErasureCodingHandler) Descriptor() *plugin_pb.JobTypeDescriptor { "preferred_tags": { Kind: &plugin_pb.ConfigValue_StringValue{StringValue: ""}, }, + "replica_placement": { + Kind: &plugin_pb.ConfigValue_StringValue{StringValue: ""}, + }, }, }, AdminRuntimeDefaults: &plugin_pb.AdminRuntimeDefaults{ @@ -217,7 +228,11 @@ func (h *ErasureCodingHandler) Detect( return err } - clusterInfo := &workertypes.ClusterInfo{ActiveTopology: activeTopology, GrpcDialOption: h.grpcDialOption} + clusterInfo := &workertypes.ClusterInfo{ + ActiveTopology: activeTopology, + GrpcDialOption: h.grpcDialOption, + DefaultReplicaPlacement: pluginworker.FetchDefaultReplicaPlacement(ctx, masters, h.grpcDialOption), + } maxResults := int(request.MaxResults) if maxResults < 0 { maxResults = 0 @@ -592,6 +607,8 @@ func deriveErasureCodingWorkerConfig(values map[string]*plugin_pb.ConfigValue) * taskConfig.PreferredTags = util.NormalizeTagList(pluginworker.ReadStringListConfig(values, "preferred_tags")) + taskConfig.ReplicaPlacement = strings.TrimSpace(pluginworker.ReadStringConfig(values, "replica_placement", taskConfig.ReplicaPlacement)) + return &erasureCodingWorkerConfig{ TaskConfig: taskConfig, }