diff --git a/weed/admin/dash/admin_server.go b/weed/admin/dash/admin_server.go index b02e869f8..492705d4c 100644 --- a/weed/admin/dash/admin_server.go +++ b/weed/admin/dash/admin_server.go @@ -1778,7 +1778,7 @@ func collectCollectionStats(topologyInfo *master_pb.TopologyInfo) map[string]col shards := erasure_coding.ShardsInfoFromVolumeEcShardInformationMessage(ecShardInfo) data := collectionMap[collection] data.PhysicalSize += int64(shards.TotalSize()) - data.LogicalSize += int64(shards.MinusParityShards().TotalSize()) + data.LogicalSize += int64(shards.MinusParityShards(erasure_coding.DataShardsCount).TotalSize()) collectionMap[collection] = data // fileCount is volume-wide (same .ecx on every shard diff --git a/weed/shell/command_ec_decode.go b/weed/shell/command_ec_decode.go index 62b201b05..edb1120b4 100644 --- a/weed/shell/command_ec_decode.go +++ b/weed/shell/command_ec_decode.go @@ -115,7 +115,7 @@ func doEcDecode(commandEnv *CommandEnv, topoInfo *master_pb.TopologyInfo, collec } // find volume location - nodeToEcShardsInfo := collectEcNodeShardsInfo(topoInfo, vid, diskType) + nodeToEcShardsInfo, dataShards := collectEcNodeShardsInfo(topoInfo, vid, diskType) fmt.Printf("ec volume %d shard locations: %+v\n", vid, nodeToEcShardsInfo) @@ -148,7 +148,7 @@ func doEcDecode(commandEnv *CommandEnv, topoInfo *master_pb.TopologyInfo, collec } // collect ec shards to the server with most space - targetNodeLocation, err := collectEcShards(commandEnv, nodeToEcShardsInfo, collection, vid, eligibleTargets) + targetNodeLocation, err := collectEcShards(commandEnv, nodeToEcShardsInfo, collection, vid, eligibleTargets, dataShards) if err != nil { return fmt.Errorf("collectEcShards for volume %d: %v", vid, err) } @@ -281,7 +281,7 @@ func generateNormalVolume(grpcDialOption grpc.DialOption, vid needle.VolumeId, c } -func collectEcShards(commandEnv *CommandEnv, nodeToShardsInfo map[pb.ServerAddress]*erasure_coding.ShardsInfo, collection string, vid needle.VolumeId, eligibleTargets map[pb.ServerAddress]struct{}) (targetNodeLocation pb.ServerAddress, err error) { +func collectEcShards(commandEnv *CommandEnv, nodeToShardsInfo map[pb.ServerAddress]*erasure_coding.ShardsInfo, collection string, vid needle.VolumeId, eligibleTargets map[pb.ServerAddress]struct{}, dataShards int) (targetNodeLocation pb.ServerAddress, err error) { maxShardCount := -1 existingShardsInfo := erasure_coding.NewShardsInfo() @@ -291,7 +291,7 @@ func collectEcShards(commandEnv *CommandEnv, nodeToShardsInfo map[pb.ServerAddre continue } } - toBeCopiedShardCount := si.MinusParityShards().Count() + toBeCopiedShardCount := si.MinusParityShards(dataShards).Count() if toBeCopiedShardCount > maxShardCount { maxShardCount = toBeCopiedShardCount targetNodeLocation = loc @@ -310,7 +310,7 @@ func collectEcShards(commandEnv *CommandEnv, nodeToShardsInfo map[pb.ServerAddre continue } - needToCopyShardsInfo := si.Minus(existingShardsInfo).MinusParityShards() + needToCopyShardsInfo := si.Minus(existingShardsInfo).MinusParityShards(dataShards) err = operation.WithVolumeServerClient(false, targetNodeLocation, commandEnv.option.GrpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error { @@ -401,7 +401,7 @@ func collectEcShardIds(topoInfo *master_pb.TopologyInfo, collectionPattern strin return } -func collectEcNodeShardsInfo(topoInfo *master_pb.TopologyInfo, vid needle.VolumeId, diskType types.DiskType) map[pb.ServerAddress]*erasure_coding.ShardsInfo { +func collectEcNodeShardsInfo(topoInfo *master_pb.TopologyInfo, vid needle.VolumeId, diskType types.DiskType) (map[pb.ServerAddress]*erasure_coding.ShardsInfo, int) { res := make(map[pb.ServerAddress]*erasure_coding.ShardsInfo) eachDataNode(topoInfo, func(dc DataCenterId, rack RackId, dn *master_pb.DataNodeInfo) { if diskInfo, found := dn.DiskInfos[string(diskType)]; found { @@ -423,7 +423,8 @@ func collectEcNodeShardsInfo(topoInfo *master_pb.TopologyInfo, vid needle.Volume } }) - return res + // OSS is always 10+4; the per-volume ratio override lives in the enterprise build. + return res, erasure_coding.DataShardsCount } type decodeDiskUsageState struct { diff --git a/weed/shell/command_ec_encode.go b/weed/shell/command_ec_encode.go index a81ef2650..6fff4de5a 100644 --- a/weed/shell/command_ec_encode.go +++ b/weed/shell/command_ec_encode.go @@ -359,7 +359,7 @@ func verifyEcShardsBeforeDelete(commandEnv *CommandEnv, volumeIds []needle.Volum lastErr = nil for _, vid := range volumeIds { - nodeShards := collectEcNodeShardsInfo(topoInfo, vid, diskType) + nodeShards, _ := collectEcNodeShardsInfo(topoInfo, vid, diskType) var union erasure_coding.ShardBits for _, info := range nodeShards { diff --git a/weed/storage/erasure_coding/ec_parity_shards_ratio_test.go b/weed/storage/erasure_coding/ec_parity_shards_ratio_test.go new file mode 100644 index 000000000..f1995b65a --- /dev/null +++ b/weed/storage/erasure_coding/ec_parity_shards_ratio_test.go @@ -0,0 +1,49 @@ +package erasure_coding + +import ( + "testing" + + "github.com/stretchr/testify/assert" +) + +// TestMinusParityShardsRatio covers the data/parity split for the default 10+4 +// layout and other ratios. The pre-fix code hardcoded ids 10..13 as parity, +// which for a wide ratio (12+4, 16+6) DROPPED real data ids >= 10 and for a +// narrow ratio (9+3) kept parity id 9. The method now takes the data-shard +// count (<= 0 falls back to DataShardsCount). +func TestMinusParityShardsRatio(t *testing.T) { + all := func(n int) []ShardId { + ids := make([]ShardId, n) + for i := range ids { + ids[i] = ShardId(i) + } + return ids + } + tests := []struct { + name string + dataShards int + inputShards []ShardId + expectedCount int + expectedIds []ShardId + }{ + {"default 10+4 all present", DataShardsCount, all(14), 10, all(10)}, + {"zero ratio falls back to default", 0, all(14), 10, all(10)}, + {"9+3 all present", 9, all(12), 9, all(9)}, + {"9+3 drops parity id 9", 9, []ShardId{0, 1, 2, 9, 10, 11}, 3, []ShardId{0, 1, 2}}, + {"12+4 keeps data ids >= 10", 12, all(16), 12, all(12)}, + {"16+6 keeps data ids >= 10", 16, all(22), 16, all(16)}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + si := NewShardsInfo() + for _, id := range tt.inputShards { + si.Set(NewShardInfo(id, ShardSize(1000))) + } + result := si.MinusParityShards(tt.dataShards) + assert.Equal(t, tt.expectedCount, result.Count(), "count for ratio %d", tt.dataShards) + for _, id := range tt.expectedIds { + assert.True(t, result.Has(id), "data shard %d must remain for ratio %d", id, tt.dataShards) + } + }) + } +} diff --git a/weed/storage/erasure_coding/ec_shards_info.go b/weed/storage/erasure_coding/ec_shards_info.go index 87a62a249..8d6496c76 100644 --- a/weed/storage/erasure_coding/ec_shards_info.go +++ b/weed/storage/erasure_coding/ec_shards_info.go @@ -338,17 +338,37 @@ func (si *ShardsInfo) Copy() *ShardsInfo { } } -// DeleteParityShards removes parity shards from a ShardInfo. -func (si *ShardsInfo) DeleteParityShards() { - for id := DataShardsCount; id < TotalShardsCount; id++ { - si.Delete(ShardId(id)) +// DeleteParityShards removes parity shards (those with id >= dataShards) from +// a ShardInfo. dataShards is the volume's data-shard count; passing <= 0 falls +// back to DataShardsCount (the fixed 10+4 layout). The upper bound is +// MaxShardCount, not TotalShardsCount, so a custom ratio's high parity ids +// (e.g. 16+6 reaches id 21) are cleared too; Delete no-ops on absent ids. +func (si *ShardsInfo) DeleteParityShards(dataShards int) { + if dataShards <= 0 { + dataShards = DataShardsCount } + if dataShards >= MaxShardCount { + return // every id is a data shard; nothing to remove + } + si.mu.Lock() + defer si.mu.Unlock() + + // Parity ids are >= dataShards. shards stays sorted by Id and shardBits is + // a bitmap, so clear them in one locked pass instead of a per-id Delete: + // mask off the high bits, then truncate the sorted slice at the first + // parity id via binary search. + si.shardBits &= ShardBits((uint32(1) << uint(dataShards)) - 1) + idx := sort.Search(len(si.shards), func(i int) bool { + return si.shards[i].Id >= ShardId(dataShards) + }) + si.shards = si.shards[:idx] } -// MinusParityShards creates a ShardInfo copy, but with parity shards removed. -func (si *ShardsInfo) MinusParityShards() *ShardsInfo { +// MinusParityShards creates a ShardInfo copy with parity shards removed for the +// given data-shard count (<= 0 falls back to DataShardsCount). +func (si *ShardsInfo) MinusParityShards(dataShards int) *ShardsInfo { result := si.Copy() - result.DeleteParityShards() + result.DeleteParityShards(dataShards) return result } diff --git a/weed/storage/erasure_coding/ec_shards_info_test.go b/weed/storage/erasure_coding/ec_shards_info_test.go index 5604447d4..bd3055c43 100644 --- a/weed/storage/erasure_coding/ec_shards_info_test.go +++ b/weed/storage/erasure_coding/ec_shards_info_test.go @@ -243,7 +243,7 @@ func TestShardsInfo_DeleteParityShards(t *testing.T) { si.Set(ShardInfo{Id: ShardId(i), Size: ShardSize((i + 1) * 1000)}) } - si.DeleteParityShards() + si.DeleteParityShards(DataShardsCount) // Verify only data shards remain for i := 0; i < DataShardsCount; i++ { diff --git a/weed/storage/erasure_coding/ec_volume_info_test.go b/weed/storage/erasure_coding/ec_volume_info_test.go index 61fff54ce..a7bb3abcd 100644 --- a/weed/storage/erasure_coding/ec_volume_info_test.go +++ b/weed/storage/erasure_coding/ec_volume_info_test.go @@ -13,7 +13,7 @@ func TestShardsInfoDeleteParityShards(t *testing.T) { for _, id := range erasure_coding.AllShardIds() { si.Set(erasure_coding.ShardInfo{Id: id, Size: 123}) } - si.DeleteParityShards() + si.DeleteParityShards(erasure_coding.DataShardsCount) if got, want := si.String(), "0:123 B 1:123 B 2:123 B 3:123 B 4:123 B 5:123 B 6:123 B 7:123 B 8:123 B 9:123 B"; got != want { t.Errorf("expected %q, got %q", want, got) diff --git a/weed/worker/tasks/erasure_coding/detection.go b/weed/worker/tasks/erasure_coding/detection.go index c22197625..b9e9b75c8 100644 --- a/weed/worker/tasks/erasure_coding/detection.go +++ b/weed/worker/tasks/erasure_coding/detection.go @@ -307,13 +307,19 @@ func Detection(ctx context.Context, metrics []*types.VolumeHealthMetrics, cluste for _, shard := range existingECShards { key := fmt.Sprintf("%s:%d", shard.ServerID, shard.DiskID) if !duplicateCheck[key] { // Avoid duplicates if EC shards are on same disk as volume replicas + shardIds := append([]uint32(nil), shard.ShardIds...) + // Free exactly the shards on this disk. Without an explicit + // impact the cleanup falls back to CalculateECShardCleanupImpact, + // which credits TotalShardsCount (14) regardless of ratio. + cleanupImpact := topology.StorageSlotChange{ShardSlots: -int32(len(shardIds))} sources = append(sources, topology.TaskSourceSpec{ - ServerID: shard.ServerID, - DiskID: shard.DiskID, - DataCenter: shard.DataCenter, - Rack: shard.Rack, - CleanupType: topology.CleanupECShards, - ShardIds: append([]uint32(nil), shard.ShardIds...), + ServerID: shard.ServerID, + DiskID: shard.DiskID, + DataCenter: shard.DataCenter, + Rack: shard.Rack, + CleanupType: topology.CleanupECShards, + ShardIds: shardIds, + StorageImpact: &cleanupImpact, }) duplicateCheck[key] = true }