From 4f50c5b0d425dd5627f11272d85f63fa5e7180a9 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Thu, 13 Aug 2026 13:22:58 -0700 Subject: [PATCH] feat: throughput limits for replicate, EC shard, and worker-driven moves (#10749) * feat: throughput limits for replicate, EC shard, and worker-driven moves VolumeCopy was the only rate-limitable transfer; EC shard copies, replica creation, and worker-driven moves all ran at whatever the receiving server's maintenance rate allowed, with no per-operation control. - proto: VolumeEcShardsCopyRequest and the balance / ec_balance task params and configs gain io_byte_per_second; 0 keeps today's behavior (the volume server's own maintenance rate governs). - volume server: VolumeEcShardsCopy throttles with one WriteThrottler per request, shared across the shard, .ecx, .ecj, .vif, and .ecsum copies so the limit caps the transfer as a whole - the same shape as VolumeCopy. - volume_move: ReplicateVolume accepts the limit; EcMoveOptions carries it through MoveEcShards/CopyAndMountEcShards into the copy request, with fake-client tests asserting propagation. - shell: ec.balance gains -ioBytePerSecond; volume.tier.move's replication top-up honors the command's existing -ioBytePerSecond instead of running unthrottled. - worker: balance and ec_balance configs gain io_byte_per_second (surfaced in the admin config schema), carried through detection and plugin job parameters into task params and handed to the shared mover; batch balance jobs inherit the limit from their detection results. The limit is per copy stream, so maxParallelization multiplies the aggregate ceiling. * worker plugins: expose io_byte_per_second in the plugin config and derive it The plugin-driven detection path derives its task Config from the plugin configuration values, and both balance and ec_balance left IoBytePerSecond at zero there - a configured limit silently reverted to the server maintenance rate. Both derive functions now read the field (clamped at zero), and the plugin descriptors expose it with defaults so the configuration form carries it. --- weed/operation/volume_move/ec_move.go | 26 +++-- weed/operation/volume_move/ec_move_test.go | 6 +- weed/operation/volume_move/volume_move.go | 12 ++- .../operation/volume_move/volume_move_test.go | 16 ++- weed/pb/volume_server.proto | 1 + weed/pb/volume_server_pb/volume_server.pb.go | 38 ++++--- .../volume_server_pb/volume_server_grpc.pb.go | 102 +++++++++--------- weed/pb/worker.proto | 4 + weed/pb/worker_pb/worker.pb.go | 58 ++++++++-- weed/pb/worker_pb/worker_grpc.pb.go | 8 +- weed/server/volume_grpc_erasure_coding.go | 18 +++- weed/shell/command_ec_balance.go | 3 +- weed/shell/command_ec_common.go | 8 +- weed/shell/command_ec_encode.go | 2 +- weed/shell/command_fs_distribute_chunks.go | 3 +- weed/shell/command_s3_lifecycle_run_shard.go | 2 +- weed/shell/command_volume_fix_replication.go | 2 +- weed/shell/command_volume_move.go | 4 +- weed/shell/command_volume_tier_move.go | 6 +- weed/shell/command_volume_tier_upload.go | 2 +- weed/worker/tasks/balance/balance_task.go | 3 +- weed/worker/tasks/balance/config.go | 18 ++++ weed/worker/tasks/balance/detection.go | 9 +- weed/worker/tasks/balance/plugin_handler.go | 41 ++++++- .../tasks/balance/replica_placement_test.go | 2 +- weed/worker/tasks/ec_balance/config.go | 20 +++- weed/worker/tasks/ec_balance/detection.go | 7 +- .../tasks/ec_balance/ec_balance_task.go | 3 +- .../worker/tasks/ec_balance/plugin_handler.go | 20 +++- .../ec_task_encode_files_e2e_test.go | 4 +- .../ec_task_julor_layout_e2e_test.go | 6 +- weed/worker/tasks/iceberg/handler.go | 2 +- 32 files changed, 312 insertions(+), 144 deletions(-) diff --git a/weed/operation/volume_move/ec_move.go b/weed/operation/volume_move/ec_move.go index 3650909e1..4c0929702 100644 --- a/weed/operation/volume_move/ec_move.go +++ b/weed/operation/volume_move/ec_move.go @@ -24,6 +24,9 @@ type EcShardMove struct { // EcMoveOptions control MoveEcShards. type EcMoveOptions struct { + // IoBytePerSecond limits the shard copy rate; 0 falls back to the volume + // server's maintenance rate. + IoBytePerSecond int64 // Writer receives human-readable progress lines (nil discards them). Writer io.Writer // Progress, when set, receives percent/stage callbacks as the move advances. @@ -54,7 +57,7 @@ func (m *Mover) MoveEcShards(ctx context.Context, move EcShardMove, opts EcMoveO } progress(10, fmt.Sprintf("copying EC shard(s) %d.%v from %s to %s", move.VolumeId, move.ShardIds, move.Source, move.Target)) - if err := m.CopyAndMountEcShards(ctx, move.VolumeId, move.Collection, move.ShardIds, move.Source, move.Target, move.TargetDisk, writer); err != nil { + if err := m.CopyAndMountEcShards(ctx, move.VolumeId, move.Collection, move.ShardIds, move.Source, move.Target, move.TargetDisk, opts.IoBytePerSecond, writer); err != nil { return err } @@ -83,7 +86,7 @@ func (m *Mover) MoveEcShards(ctx context.Context, move EcShardMove, opts EcMoveO // .vif/.ecsum sidecars) from source and mount them. A same-address call skips // the copy and just mounts — ec.encode uses that to bring freshly generated // shards online in place. -func (m *Mover) CopyAndMountEcShards(ctx context.Context, volumeId needle.VolumeId, collection string, shardIds []erasure_coding.ShardId, source, target pb.ServerAddress, targetDisk uint32, writer io.Writer) error { +func (m *Mover) CopyAndMountEcShards(ctx context.Context, volumeId needle.VolumeId, collection string, shardIds []erasure_coding.ShardId, source, target pb.ServerAddress, targetDisk uint32, ioBytePerSecond int64, writer io.Writer) error { if writer == nil { writer = io.Discard } @@ -98,15 +101,16 @@ func (m *Mover) CopyAndMountEcShards(ctx context.Context, volumeId needle.Volume if !SameServer(target, source) { fmt.Fprintf(writer, "copy %d.%v %s => %s\n", volumeId, shardIds, source, target) _, copyErr := client.VolumeEcShardsCopy(ctx, &volume_server_pb.VolumeEcShardsCopyRequest{ - VolumeId: uint32(volumeId), - Collection: collection, - ShardIds: erasure_coding.ShardIdsToUint32(shardIds), - CopyEcxFile: true, - CopyEcjFile: true, - CopyVifFile: true, - CopyEcsumFile: true, // propagate the bitrot sidecar with the shards (no-op if the source has none) - SourceDataNode: string(source), - DiskId: targetDisk, + VolumeId: uint32(volumeId), + Collection: collection, + ShardIds: erasure_coding.ShardIdsToUint32(shardIds), + CopyEcxFile: true, + CopyEcjFile: true, + CopyVifFile: true, + CopyEcsumFile: true, // propagate the bitrot sidecar with the shards (no-op if the source has none) + SourceDataNode: string(source), + DiskId: targetDisk, + IoBytePerSecond: ioBytePerSecond, }) if copyErr != nil { return fmt.Errorf("copy %d.%v %s => %s: %v", volumeId, shardIds, source, target, copyErr) diff --git a/weed/operation/volume_move/ec_move_test.go b/weed/operation/volume_move/ec_move_test.go index 796f9214e..c71bd01fa 100644 --- a/weed/operation/volume_move/ec_move_test.go +++ b/weed/operation/volume_move/ec_move_test.go @@ -33,7 +33,7 @@ func TestMoveEcShardsSequence(t *testing.T) { cluster := newFakeCluster() cluster.ecShards[string(dstAddr)] = dstShards(3, 4) - err := cluster.mover().MoveEcShards(context.Background(), ecMove(3, 4), EcMoveOptions{}) + err := cluster.mover().MoveEcShards(context.Background(), ecMove(3, 4), EcMoveOptions{IoBytePerSecond: 77}) if err != nil { t.Fatalf("MoveEcShards: %v", err) } @@ -50,7 +50,7 @@ func TestMoveEcShardsSequence(t *testing.T) { if !copyReq.CopyEcxFile || !copyReq.CopyEcjFile || !copyReq.CopyVifFile || !copyReq.CopyEcsumFile { t.Errorf("shard sidecars not all copied: %+v", copyReq) } - if copyReq.DiskId != 2 || copyReq.SourceDataNode != string(srcAddr) || copyReq.Collection != "c1" { + if copyReq.DiskId != 2 || copyReq.SourceDataNode != string(srcAddr) || copyReq.Collection != "c1" || copyReq.IoBytePerSecond != 77 { t.Errorf("copy request not propagated: %+v", copyReq) } } @@ -106,7 +106,7 @@ func TestRemoveEcShards(t *testing.T) { func TestCopyAndMountEcShardsSameAddressMountsOnly(t *testing.T) { cluster := newFakeCluster() - err := cluster.mover().CopyAndMountEcShards(context.Background(), 7, "c1", []erasure_coding.ShardId{3}, srcAddr, srcAddr, 0, nil) + err := cluster.mover().CopyAndMountEcShards(context.Background(), 7, "c1", []erasure_coding.ShardId{3}, srcAddr, srcAddr, 0, 0, nil) if err != nil { t.Fatalf("CopyAndMountEcShards: %v", err) } diff --git a/weed/operation/volume_move/volume_move.go b/weed/operation/volume_move/volume_move.go index 9232699f4..c7c4754f0 100644 --- a/weed/operation/volume_move/volume_move.go +++ b/weed/operation/volume_move/volume_move.go @@ -445,8 +445,9 @@ func (m *Mover) MarkVolumeWritable(ctx context.Context, volumeId needle.VolumeId } // ReplicateVolume copies a volume from source to target without touching the -// source — the replica-creation half of a move. -func (m *Mover) ReplicateVolume(ctx context.Context, volumeId needle.VolumeId, source, target pb.ServerAddress, diskType string, writer io.Writer) error { +// source — the replica-creation half of a move. ioBytePerSecond limits the +// copy rate; 0 falls back to the volume server's maintenance rate. +func (m *Mover) ReplicateVolume(ctx context.Context, volumeId needle.VolumeId, source, target pb.ServerAddress, diskType string, ioBytePerSecond int64, writer io.Writer) error { if writer == nil { writer = io.Discard } @@ -460,9 +461,10 @@ func (m *Mover) ReplicateVolume(ctx context.Context, volumeId needle.VolumeId, s } return m.withClient(false, target, func(client volume_server_pb.VolumeServerClient) error { stream, replicateErr := client.VolumeCopy(ctx, &volume_server_pb.VolumeCopyRequest{ - VolumeId: uint32(volumeId), - SourceDataNode: string(source), - DiskType: diskType, + VolumeId: uint32(volumeId), + SourceDataNode: string(source), + DiskType: diskType, + IoBytePerSecond: ioBytePerSecond, }) if replicateErr != nil { return replicateErr diff --git a/weed/operation/volume_move/volume_move_test.go b/weed/operation/volume_move/volume_move_test.go index c500c18fb..0b41fcbe7 100644 --- a/weed/operation/volume_move/volume_move_test.go +++ b/weed/operation/volume_move/volume_move_test.go @@ -80,13 +80,13 @@ func TestEmbeddedSourceAddressValidated(t *testing.T) { ops := map[string]func(m *Mover) error{ "ReplicateVolume": func(m *Mover) error { - return m.ReplicateVolume(context.Background(), 7, bad, dstAddr, "", nil) + return m.ReplicateVolume(context.Background(), 7, bad, dstAddr, "", 0, nil) }, "TailVolume": func(m *Mover) error { return m.TailVolume(context.Background(), 7, bad, dstAddr, 0, time.Second) }, "CopyAndMountEcShards": func(m *Mover) error { - return m.CopyAndMountEcShards(context.Background(), 7, "c1", []erasure_coding.ShardId{3}, bad, dstAddr, 0, nil) + return m.CopyAndMountEcShards(context.Background(), 7, "c1", []erasure_coding.ShardId{3}, bad, dstAddr, 0, 0, nil) }, "MoveEcShards": func(m *Mover) error { move := ecMove(3) @@ -601,6 +601,18 @@ func TestLiveMoveVolumeReadonlyMarkFailureRestores(t *testing.T) { } } +func TestReplicateVolumePropagatesThroughput(t *testing.T) { + cluster := newFakeCluster() + + if err := cluster.mover().ReplicateVolume(context.Background(), 7, srcAddr, dstAddr, "ssd", 55, nil); err != nil { + t.Fatalf("ReplicateVolume: %v", err) + } + copyReq := cluster.copyReqs[0] + if copyReq.IoBytePerSecond != 55 || copyReq.DiskType != "ssd" { + t.Errorf("replicate request not propagated: %+v", copyReq) + } +} + func TestCopyVolumeRestoreWritable(t *testing.T) { for _, restoreWritable := range []bool{true, false} { cluster := newFakeCluster() diff --git a/weed/pb/volume_server.proto b/weed/pb/volume_server.proto index 5abaa06f9..93215af0a 100644 --- a/weed/pb/volume_server.proto +++ b/weed/pb/volume_server.proto @@ -461,6 +461,7 @@ message VolumeEcShardsCopyRequest { bool copy_vif_file = 7; uint32 disk_id = 8; // Target disk ID for storing EC shards bool copy_ecsum_file = 9; // copy the bitrot checksum sidecar (.ecsum) when present; tolerant of a missing source (no-op), since this non-2PC path has no Prepare backstop + int64 io_byte_per_second = 10; // limit the copy rate; 0 falls back to the server's maintenance rate } message VolumeEcShardsCopyResponse { } diff --git a/weed/pb/volume_server_pb/volume_server.pb.go b/weed/pb/volume_server_pb/volume_server.pb.go index 368d16e4b..8159e8eaa 100644 --- a/weed/pb/volume_server_pb/volume_server.pb.go +++ b/weed/pb/volume_server_pb/volume_server.pb.go @@ -3559,18 +3559,19 @@ func (x *VolumeEcShardsRebuildResponse) GetRebuiltShardIds() []uint32 { } type VolumeEcShardsCopyRequest struct { - state protoimpl.MessageState `protogen:"open.v1"` - VolumeId uint32 `protobuf:"varint,1,opt,name=volume_id,json=volumeId,proto3" json:"volume_id,omitempty"` - Collection string `protobuf:"bytes,2,opt,name=collection,proto3" json:"collection,omitempty"` - ShardIds []uint32 `protobuf:"varint,3,rep,packed,name=shard_ids,json=shardIds,proto3" json:"shard_ids,omitempty"` - CopyEcxFile bool `protobuf:"varint,4,opt,name=copy_ecx_file,json=copyEcxFile,proto3" json:"copy_ecx_file,omitempty"` - SourceDataNode string `protobuf:"bytes,5,opt,name=source_data_node,json=sourceDataNode,proto3" json:"source_data_node,omitempty"` - CopyEcjFile bool `protobuf:"varint,6,opt,name=copy_ecj_file,json=copyEcjFile,proto3" json:"copy_ecj_file,omitempty"` - CopyVifFile bool `protobuf:"varint,7,opt,name=copy_vif_file,json=copyVifFile,proto3" json:"copy_vif_file,omitempty"` - DiskId uint32 `protobuf:"varint,8,opt,name=disk_id,json=diskId,proto3" json:"disk_id,omitempty"` // Target disk ID for storing EC shards - CopyEcsumFile bool `protobuf:"varint,9,opt,name=copy_ecsum_file,json=copyEcsumFile,proto3" json:"copy_ecsum_file,omitempty"` // copy the bitrot checksum sidecar (.ecsum) when present; tolerant of a missing source (no-op), since this non-2PC path has no Prepare backstop - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + state protoimpl.MessageState `protogen:"open.v1"` + VolumeId uint32 `protobuf:"varint,1,opt,name=volume_id,json=volumeId,proto3" json:"volume_id,omitempty"` + Collection string `protobuf:"bytes,2,opt,name=collection,proto3" json:"collection,omitempty"` + ShardIds []uint32 `protobuf:"varint,3,rep,packed,name=shard_ids,json=shardIds,proto3" json:"shard_ids,omitempty"` + CopyEcxFile bool `protobuf:"varint,4,opt,name=copy_ecx_file,json=copyEcxFile,proto3" json:"copy_ecx_file,omitempty"` + SourceDataNode string `protobuf:"bytes,5,opt,name=source_data_node,json=sourceDataNode,proto3" json:"source_data_node,omitempty"` + CopyEcjFile bool `protobuf:"varint,6,opt,name=copy_ecj_file,json=copyEcjFile,proto3" json:"copy_ecj_file,omitempty"` + CopyVifFile bool `protobuf:"varint,7,opt,name=copy_vif_file,json=copyVifFile,proto3" json:"copy_vif_file,omitempty"` + DiskId uint32 `protobuf:"varint,8,opt,name=disk_id,json=diskId,proto3" json:"disk_id,omitempty"` // Target disk ID for storing EC shards + CopyEcsumFile bool `protobuf:"varint,9,opt,name=copy_ecsum_file,json=copyEcsumFile,proto3" json:"copy_ecsum_file,omitempty"` // copy the bitrot checksum sidecar (.ecsum) when present; tolerant of a missing source (no-op), since this non-2PC path has no Prepare backstop + IoBytePerSecond int64 `protobuf:"varint,10,opt,name=io_byte_per_second,json=ioBytePerSecond,proto3" json:"io_byte_per_second,omitempty"` // limit the copy rate; 0 falls back to the server's maintenance rate + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *VolumeEcShardsCopyRequest) Reset() { @@ -3666,6 +3667,13 @@ func (x *VolumeEcShardsCopyRequest) GetCopyEcsumFile() bool { return false } +func (x *VolumeEcShardsCopyRequest) GetIoBytePerSecond() int64 { + if x != nil { + return x.IoBytePerSecond + } + return 0 +} + type VolumeEcShardsCopyResponse struct { state protoimpl.MessageState `protogen:"open.v1"` unknownFields protoimpl.UnknownFields @@ -7424,7 +7432,7 @@ const file_volume_server_proto_rawDesc = "" + "collection\x122\n" + "\x15unsafe_ignore_sidecar\x18\x03 \x01(\bR\x13unsafeIgnoreSidecar\"K\n" + "\x1dVolumeEcShardsRebuildResponse\x12*\n" + - "\x11rebuilt_shard_ids\x18\x01 \x03(\rR\x0frebuiltShardIds\"\xcc\x02\n" + + "\x11rebuilt_shard_ids\x18\x01 \x03(\rR\x0frebuiltShardIds\"\xf9\x02\n" + "\x19VolumeEcShardsCopyRequest\x12\x1b\n" + "\tvolume_id\x18\x01 \x01(\rR\bvolumeId\x12\x1e\n" + "\n" + @@ -7436,7 +7444,9 @@ const file_volume_server_proto_rawDesc = "" + "\rcopy_ecj_file\x18\x06 \x01(\bR\vcopyEcjFile\x12\"\n" + "\rcopy_vif_file\x18\a \x01(\bR\vcopyVifFile\x12\x17\n" + "\adisk_id\x18\b \x01(\rR\x06diskId\x12&\n" + - "\x0fcopy_ecsum_file\x18\t \x01(\bR\rcopyEcsumFile\"\x1c\n" + + "\x0fcopy_ecsum_file\x18\t \x01(\bR\rcopyEcsumFile\x12+\n" + + "\x12io_byte_per_second\x18\n" + + " \x01(\x03R\x0fioBytePerSecond\"\x1c\n" + "\x1aVolumeEcShardsCopyResponse\"\xbe\x01\n" + "\x1bVolumeEcShardsDeleteRequest\x12\x1b\n" + "\tvolume_id\x18\x01 \x01(\rR\bvolumeId\x12\x1e\n" + diff --git a/weed/pb/volume_server_pb/volume_server_grpc.pb.go b/weed/pb/volume_server_pb/volume_server_grpc.pb.go index 692b9702c..eb659eab0 100644 --- a/weed/pb/volume_server_pb/volume_server_grpc.pb.go +++ b/weed/pb/volume_server_pb/volume_server_grpc.pb.go @@ -1,6 +1,6 @@ // Code generated by protoc-gen-go-grpc. DO NOT EDIT. // versions: -// - protoc-gen-go-grpc v1.5.1 +// - protoc-gen-go-grpc v1.6.2 // - protoc v7.35.0 // source: volume_server.proto @@ -794,151 +794,151 @@ type VolumeServerServer interface { type UnimplementedVolumeServerServer struct{} func (UnimplementedVolumeServerServer) BatchDelete(context.Context, *BatchDeleteRequest) (*BatchDeleteResponse, error) { - return nil, status.Errorf(codes.Unimplemented, "method BatchDelete not implemented") + return nil, status.Error(codes.Unimplemented, "method BatchDelete not implemented") } func (UnimplementedVolumeServerServer) VacuumVolumeCheck(context.Context, *VacuumVolumeCheckRequest) (*VacuumVolumeCheckResponse, error) { - return nil, status.Errorf(codes.Unimplemented, "method VacuumVolumeCheck not implemented") + return nil, status.Error(codes.Unimplemented, "method VacuumVolumeCheck not implemented") } func (UnimplementedVolumeServerServer) VacuumVolumeCompact(*VacuumVolumeCompactRequest, grpc.ServerStreamingServer[VacuumVolumeCompactResponse]) error { - return status.Errorf(codes.Unimplemented, "method VacuumVolumeCompact not implemented") + return status.Error(codes.Unimplemented, "method VacuumVolumeCompact not implemented") } func (UnimplementedVolumeServerServer) VacuumVolumeCommit(context.Context, *VacuumVolumeCommitRequest) (*VacuumVolumeCommitResponse, error) { - return nil, status.Errorf(codes.Unimplemented, "method VacuumVolumeCommit not implemented") + return nil, status.Error(codes.Unimplemented, "method VacuumVolumeCommit not implemented") } func (UnimplementedVolumeServerServer) VacuumVolumeCleanup(context.Context, *VacuumVolumeCleanupRequest) (*VacuumVolumeCleanupResponse, error) { - return nil, status.Errorf(codes.Unimplemented, "method VacuumVolumeCleanup not implemented") + return nil, status.Error(codes.Unimplemented, "method VacuumVolumeCleanup not implemented") } func (UnimplementedVolumeServerServer) DeleteCollection(context.Context, *DeleteCollectionRequest) (*DeleteCollectionResponse, error) { - return nil, status.Errorf(codes.Unimplemented, "method DeleteCollection not implemented") + return nil, status.Error(codes.Unimplemented, "method DeleteCollection not implemented") } func (UnimplementedVolumeServerServer) AllocateVolume(context.Context, *AllocateVolumeRequest) (*AllocateVolumeResponse, error) { - return nil, status.Errorf(codes.Unimplemented, "method AllocateVolume not implemented") + return nil, status.Error(codes.Unimplemented, "method AllocateVolume not implemented") } func (UnimplementedVolumeServerServer) VolumeSyncStatus(context.Context, *VolumeSyncStatusRequest) (*VolumeSyncStatusResponse, error) { - return nil, status.Errorf(codes.Unimplemented, "method VolumeSyncStatus not implemented") + return nil, status.Error(codes.Unimplemented, "method VolumeSyncStatus not implemented") } func (UnimplementedVolumeServerServer) VolumeIncrementalCopy(*VolumeIncrementalCopyRequest, grpc.ServerStreamingServer[VolumeIncrementalCopyResponse]) error { - return status.Errorf(codes.Unimplemented, "method VolumeIncrementalCopy not implemented") + return status.Error(codes.Unimplemented, "method VolumeIncrementalCopy not implemented") } func (UnimplementedVolumeServerServer) VolumeMount(context.Context, *VolumeMountRequest) (*VolumeMountResponse, error) { - return nil, status.Errorf(codes.Unimplemented, "method VolumeMount not implemented") + return nil, status.Error(codes.Unimplemented, "method VolumeMount not implemented") } func (UnimplementedVolumeServerServer) VolumeUnmount(context.Context, *VolumeUnmountRequest) (*VolumeUnmountResponse, error) { - return nil, status.Errorf(codes.Unimplemented, "method VolumeUnmount not implemented") + return nil, status.Error(codes.Unimplemented, "method VolumeUnmount not implemented") } func (UnimplementedVolumeServerServer) VolumeConsolidateIndex(context.Context, *VolumeConsolidateIndexRequest) (*VolumeConsolidateIndexResponse, error) { - return nil, status.Errorf(codes.Unimplemented, "method VolumeConsolidateIndex not implemented") + return nil, status.Error(codes.Unimplemented, "method VolumeConsolidateIndex not implemented") } func (UnimplementedVolumeServerServer) VolumeDelete(context.Context, *VolumeDeleteRequest) (*VolumeDeleteResponse, error) { - return nil, status.Errorf(codes.Unimplemented, "method VolumeDelete not implemented") + return nil, status.Error(codes.Unimplemented, "method VolumeDelete not implemented") } func (UnimplementedVolumeServerServer) VolumeMarkReadonly(context.Context, *VolumeMarkReadonlyRequest) (*VolumeMarkReadonlyResponse, error) { - return nil, status.Errorf(codes.Unimplemented, "method VolumeMarkReadonly not implemented") + return nil, status.Error(codes.Unimplemented, "method VolumeMarkReadonly not implemented") } func (UnimplementedVolumeServerServer) VolumeMarkWritable(context.Context, *VolumeMarkWritableRequest) (*VolumeMarkWritableResponse, error) { - return nil, status.Errorf(codes.Unimplemented, "method VolumeMarkWritable not implemented") + return nil, status.Error(codes.Unimplemented, "method VolumeMarkWritable not implemented") } func (UnimplementedVolumeServerServer) VolumeConfigure(context.Context, *VolumeConfigureRequest) (*VolumeConfigureResponse, error) { - return nil, status.Errorf(codes.Unimplemented, "method VolumeConfigure not implemented") + return nil, status.Error(codes.Unimplemented, "method VolumeConfigure not implemented") } func (UnimplementedVolumeServerServer) VolumeStatus(context.Context, *VolumeStatusRequest) (*VolumeStatusResponse, error) { - return nil, status.Errorf(codes.Unimplemented, "method VolumeStatus not implemented") + return nil, status.Error(codes.Unimplemented, "method VolumeStatus not implemented") } func (UnimplementedVolumeServerServer) GetState(context.Context, *GetStateRequest) (*GetStateResponse, error) { - return nil, status.Errorf(codes.Unimplemented, "method GetState not implemented") + return nil, status.Error(codes.Unimplemented, "method GetState not implemented") } func (UnimplementedVolumeServerServer) SetState(context.Context, *SetStateRequest) (*SetStateResponse, error) { - return nil, status.Errorf(codes.Unimplemented, "method SetState not implemented") + return nil, status.Error(codes.Unimplemented, "method SetState not implemented") } func (UnimplementedVolumeServerServer) VolumeCopy(*VolumeCopyRequest, grpc.ServerStreamingServer[VolumeCopyResponse]) error { - return status.Errorf(codes.Unimplemented, "method VolumeCopy not implemented") + return status.Error(codes.Unimplemented, "method VolumeCopy not implemented") } func (UnimplementedVolumeServerServer) ReadVolumeFileStatus(context.Context, *ReadVolumeFileStatusRequest) (*ReadVolumeFileStatusResponse, error) { - return nil, status.Errorf(codes.Unimplemented, "method ReadVolumeFileStatus not implemented") + return nil, status.Error(codes.Unimplemented, "method ReadVolumeFileStatus not implemented") } func (UnimplementedVolumeServerServer) CopyFile(*CopyFileRequest, grpc.ServerStreamingServer[CopyFileResponse]) error { - return status.Errorf(codes.Unimplemented, "method CopyFile not implemented") + return status.Error(codes.Unimplemented, "method CopyFile not implemented") } func (UnimplementedVolumeServerServer) ReceiveFile(grpc.ClientStreamingServer[ReceiveFileRequest, ReceiveFileResponse]) error { - return status.Errorf(codes.Unimplemented, "method ReceiveFile not implemented") + return status.Error(codes.Unimplemented, "method ReceiveFile not implemented") } func (UnimplementedVolumeServerServer) ReadNeedleBlob(context.Context, *ReadNeedleBlobRequest) (*ReadNeedleBlobResponse, error) { - return nil, status.Errorf(codes.Unimplemented, "method ReadNeedleBlob not implemented") + return nil, status.Error(codes.Unimplemented, "method ReadNeedleBlob not implemented") } func (UnimplementedVolumeServerServer) ReadNeedleMeta(context.Context, *ReadNeedleMetaRequest) (*ReadNeedleMetaResponse, error) { - return nil, status.Errorf(codes.Unimplemented, "method ReadNeedleMeta not implemented") + return nil, status.Error(codes.Unimplemented, "method ReadNeedleMeta not implemented") } func (UnimplementedVolumeServerServer) WriteNeedleBlob(context.Context, *WriteNeedleBlobRequest) (*WriteNeedleBlobResponse, error) { - return nil, status.Errorf(codes.Unimplemented, "method WriteNeedleBlob not implemented") + return nil, status.Error(codes.Unimplemented, "method WriteNeedleBlob not implemented") } func (UnimplementedVolumeServerServer) ReadAllNeedles(*ReadAllNeedlesRequest, grpc.ServerStreamingServer[ReadAllNeedlesResponse]) error { - return status.Errorf(codes.Unimplemented, "method ReadAllNeedles not implemented") + return status.Error(codes.Unimplemented, "method ReadAllNeedles not implemented") } func (UnimplementedVolumeServerServer) VolumeTailSender(*VolumeTailSenderRequest, grpc.ServerStreamingServer[VolumeTailSenderResponse]) error { - return status.Errorf(codes.Unimplemented, "method VolumeTailSender not implemented") + return status.Error(codes.Unimplemented, "method VolumeTailSender not implemented") } func (UnimplementedVolumeServerServer) VolumeTailReceiver(context.Context, *VolumeTailReceiverRequest) (*VolumeTailReceiverResponse, error) { - return nil, status.Errorf(codes.Unimplemented, "method VolumeTailReceiver not implemented") + return nil, status.Error(codes.Unimplemented, "method VolumeTailReceiver not implemented") } func (UnimplementedVolumeServerServer) VolumeEcShardsGenerate(context.Context, *VolumeEcShardsGenerateRequest) (*VolumeEcShardsGenerateResponse, error) { - return nil, status.Errorf(codes.Unimplemented, "method VolumeEcShardsGenerate not implemented") + return nil, status.Error(codes.Unimplemented, "method VolumeEcShardsGenerate not implemented") } func (UnimplementedVolumeServerServer) VolumeEcShardsRebuild(context.Context, *VolumeEcShardsRebuildRequest) (*VolumeEcShardsRebuildResponse, error) { - return nil, status.Errorf(codes.Unimplemented, "method VolumeEcShardsRebuild not implemented") + return nil, status.Error(codes.Unimplemented, "method VolumeEcShardsRebuild not implemented") } func (UnimplementedVolumeServerServer) VolumeEcShardsCopy(context.Context, *VolumeEcShardsCopyRequest) (*VolumeEcShardsCopyResponse, error) { - return nil, status.Errorf(codes.Unimplemented, "method VolumeEcShardsCopy not implemented") + return nil, status.Error(codes.Unimplemented, "method VolumeEcShardsCopy not implemented") } func (UnimplementedVolumeServerServer) VolumeEcShardsDelete(context.Context, *VolumeEcShardsDeleteRequest) (*VolumeEcShardsDeleteResponse, error) { - return nil, status.Errorf(codes.Unimplemented, "method VolumeEcShardsDelete not implemented") + return nil, status.Error(codes.Unimplemented, "method VolumeEcShardsDelete not implemented") } func (UnimplementedVolumeServerServer) VolumeEcShardsMount(context.Context, *VolumeEcShardsMountRequest) (*VolumeEcShardsMountResponse, error) { - return nil, status.Errorf(codes.Unimplemented, "method VolumeEcShardsMount not implemented") + return nil, status.Error(codes.Unimplemented, "method VolumeEcShardsMount not implemented") } func (UnimplementedVolumeServerServer) VolumeEcShardsUnmount(context.Context, *VolumeEcShardsUnmountRequest) (*VolumeEcShardsUnmountResponse, error) { - return nil, status.Errorf(codes.Unimplemented, "method VolumeEcShardsUnmount not implemented") + return nil, status.Error(codes.Unimplemented, "method VolumeEcShardsUnmount not implemented") } func (UnimplementedVolumeServerServer) VolumeEcShardRead(*VolumeEcShardReadRequest, grpc.ServerStreamingServer[VolumeEcShardReadResponse]) error { - return status.Errorf(codes.Unimplemented, "method VolumeEcShardRead not implemented") + return status.Error(codes.Unimplemented, "method VolumeEcShardRead not implemented") } func (UnimplementedVolumeServerServer) VolumeEcBlobDelete(context.Context, *VolumeEcBlobDeleteRequest) (*VolumeEcBlobDeleteResponse, error) { - return nil, status.Errorf(codes.Unimplemented, "method VolumeEcBlobDelete not implemented") + return nil, status.Error(codes.Unimplemented, "method VolumeEcBlobDelete not implemented") } func (UnimplementedVolumeServerServer) VolumeEcShardsToVolume(context.Context, *VolumeEcShardsToVolumeRequest) (*VolumeEcShardsToVolumeResponse, error) { - return nil, status.Errorf(codes.Unimplemented, "method VolumeEcShardsToVolume not implemented") + return nil, status.Error(codes.Unimplemented, "method VolumeEcShardsToVolume not implemented") } func (UnimplementedVolumeServerServer) VolumeEcShardsInfo(context.Context, *VolumeEcShardsInfoRequest) (*VolumeEcShardsInfoResponse, error) { - return nil, status.Errorf(codes.Unimplemented, "method VolumeEcShardsInfo not implemented") + return nil, status.Error(codes.Unimplemented, "method VolumeEcShardsInfo not implemented") } func (UnimplementedVolumeServerServer) VolumeTierMoveDatToRemote(*VolumeTierMoveDatToRemoteRequest, grpc.ServerStreamingServer[VolumeTierMoveDatToRemoteResponse]) error { - return status.Errorf(codes.Unimplemented, "method VolumeTierMoveDatToRemote not implemented") + return status.Error(codes.Unimplemented, "method VolumeTierMoveDatToRemote not implemented") } func (UnimplementedVolumeServerServer) VolumeTierMoveDatFromRemote(*VolumeTierMoveDatFromRemoteRequest, grpc.ServerStreamingServer[VolumeTierMoveDatFromRemoteResponse]) error { - return status.Errorf(codes.Unimplemented, "method VolumeTierMoveDatFromRemote not implemented") + return status.Error(codes.Unimplemented, "method VolumeTierMoveDatFromRemote not implemented") } func (UnimplementedVolumeServerServer) VolumeServerStatus(context.Context, *VolumeServerStatusRequest) (*VolumeServerStatusResponse, error) { - return nil, status.Errorf(codes.Unimplemented, "method VolumeServerStatus not implemented") + return nil, status.Error(codes.Unimplemented, "method VolumeServerStatus not implemented") } func (UnimplementedVolumeServerServer) VolumeServerLeave(context.Context, *VolumeServerLeaveRequest) (*VolumeServerLeaveResponse, error) { - return nil, status.Errorf(codes.Unimplemented, "method VolumeServerLeave not implemented") + return nil, status.Error(codes.Unimplemented, "method VolumeServerLeave not implemented") } func (UnimplementedVolumeServerServer) FetchAndWriteNeedle(context.Context, *FetchAndWriteNeedleRequest) (*FetchAndWriteNeedleResponse, error) { - return nil, status.Errorf(codes.Unimplemented, "method FetchAndWriteNeedle not implemented") + return nil, status.Error(codes.Unimplemented, "method FetchAndWriteNeedle not implemented") } func (UnimplementedVolumeServerServer) ScrubVolume(context.Context, *ScrubVolumeRequest) (*ScrubVolumeResponse, error) { - return nil, status.Errorf(codes.Unimplemented, "method ScrubVolume not implemented") + return nil, status.Error(codes.Unimplemented, "method ScrubVolume not implemented") } func (UnimplementedVolumeServerServer) ScrubEcVolume(context.Context, *ScrubEcVolumeRequest) (*ScrubEcVolumeResponse, error) { - return nil, status.Errorf(codes.Unimplemented, "method ScrubEcVolume not implemented") + return nil, status.Error(codes.Unimplemented, "method ScrubEcVolume not implemented") } func (UnimplementedVolumeServerServer) Query(*QueryRequest, grpc.ServerStreamingServer[QueriedStripe]) error { - return status.Errorf(codes.Unimplemented, "method Query not implemented") + return status.Error(codes.Unimplemented, "method Query not implemented") } func (UnimplementedVolumeServerServer) VolumeNeedleStatus(context.Context, *VolumeNeedleStatusRequest) (*VolumeNeedleStatusResponse, error) { - return nil, status.Errorf(codes.Unimplemented, "method VolumeNeedleStatus not implemented") + return nil, status.Error(codes.Unimplemented, "method VolumeNeedleStatus not implemented") } func (UnimplementedVolumeServerServer) Ping(context.Context, *PingRequest) (*PingResponse, error) { - return nil, status.Errorf(codes.Unimplemented, "method Ping not implemented") + return nil, status.Error(codes.Unimplemented, "method Ping not implemented") } func (UnimplementedVolumeServerServer) mustEmbedUnimplementedVolumeServerServer() {} func (UnimplementedVolumeServerServer) testEmbeddedByValue() {} @@ -951,7 +951,7 @@ type UnsafeVolumeServerServer interface { } func RegisterVolumeServerServer(s grpc.ServiceRegistrar, srv VolumeServerServer) { - // If the following call pancis, it indicates UnimplementedVolumeServerServer was + // If the following call panics, it indicates UnimplementedVolumeServerServer was // embedded by pointer and is nil. This will cause panics if an // unimplemented method is ever invoked, so we test this at initialization // time to prevent it from happening at runtime later due to I/O. diff --git a/weed/pb/worker.proto b/weed/pb/worker.proto index 1505b128c..d32f2efb3 100644 --- a/weed/pb/worker.proto +++ b/weed/pb/worker.proto @@ -220,6 +220,7 @@ message BalanceTaskParams { int32 timeout_seconds = 2; // Operation timeout int32 max_concurrent_moves = 3; // Max concurrent moves in a batch job (0 = default 5) repeated BalanceMoveSpec moves = 4; // Batch: multiple volume moves in one job + int64 io_byte_per_second = 5; // limit each move's copy rate; 0 falls back to the volume server's maintenance rate } // ReplicationTaskParams for adding replicas @@ -382,6 +383,7 @@ message ErasureCodingTaskConfig { message BalanceTaskConfig { double imbalance_threshold = 1; // Threshold for triggering rebalancing (0.0-1.0) int32 min_server_count = 2; // Minimum number of servers required for balancing + int64 io_byte_per_second = 3; // limit each move's copy rate; 0 falls back to the volume server's maintenance rate } // ReplicationTaskConfig contains replication-specific configuration @@ -399,6 +401,7 @@ message EcBalanceTaskParams { // node really holds it before deleting the copy, so a topology entry naming a // location that holds nothing cannot cause the last copy to be removed. string dedup_keep_node = 5; + int64 io_byte_per_second = 6; // limit each shard copy's rate; 0 falls back to the volume server's maintenance rate } // EcShardMoveSpec describes a single EC shard move within a batch @@ -420,6 +423,7 @@ message EcBalanceTaskConfig { string disk_type = 4; // Disk type filter repeated string preferred_tags = 5; // Preferred disk tags for placement string replica_placement = 6; // EC shard replica placement (e.g. "020"); empty falls back to master default replication + int64 io_byte_per_second = 7; // limit each shard copy's rate; 0 falls back to the volume server's maintenance rate } // ========== Task Persistence Messages ========== diff --git a/weed/pb/worker_pb/worker.pb.go b/weed/pb/worker_pb/worker.pb.go index a4048c51d..501e9bd7d 100644 --- a/weed/pb/worker_pb/worker.pb.go +++ b/weed/pb/worker_pb/worker.pb.go @@ -1684,6 +1684,7 @@ type BalanceTaskParams struct { TimeoutSeconds int32 `protobuf:"varint,2,opt,name=timeout_seconds,json=timeoutSeconds,proto3" json:"timeout_seconds,omitempty"` // Operation timeout MaxConcurrentMoves int32 `protobuf:"varint,3,opt,name=max_concurrent_moves,json=maxConcurrentMoves,proto3" json:"max_concurrent_moves,omitempty"` // Max concurrent moves in a batch job (0 = default 5) Moves []*BalanceMoveSpec `protobuf:"bytes,4,rep,name=moves,proto3" json:"moves,omitempty"` // Batch: multiple volume moves in one job + IoBytePerSecond int64 `protobuf:"varint,5,opt,name=io_byte_per_second,json=ioBytePerSecond,proto3" json:"io_byte_per_second,omitempty"` // limit each move's copy rate; 0 falls back to the volume server's maintenance rate unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -1746,6 +1747,13 @@ func (x *BalanceTaskParams) GetMoves() []*BalanceMoveSpec { return nil } +func (x *BalanceTaskParams) GetIoBytePerSecond() int64 { + if x != nil { + return x.IoBytePerSecond + } + return 0 +} + // ReplicationTaskParams for adding replicas type ReplicationTaskParams struct { state protoimpl.MessageState `protogen:"open.v1"` @@ -3050,6 +3058,7 @@ type BalanceTaskConfig struct { state protoimpl.MessageState `protogen:"open.v1"` ImbalanceThreshold float64 `protobuf:"fixed64,1,opt,name=imbalance_threshold,json=imbalanceThreshold,proto3" json:"imbalance_threshold,omitempty"` // Threshold for triggering rebalancing (0.0-1.0) MinServerCount int32 `protobuf:"varint,2,opt,name=min_server_count,json=minServerCount,proto3" json:"min_server_count,omitempty"` // Minimum number of servers required for balancing + IoBytePerSecond int64 `protobuf:"varint,3,opt,name=io_byte_per_second,json=ioBytePerSecond,proto3" json:"io_byte_per_second,omitempty"` // limit each move's copy rate; 0 falls back to the volume server's maintenance rate unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -3098,6 +3107,13 @@ func (x *BalanceTaskConfig) GetMinServerCount() int32 { return 0 } +func (x *BalanceTaskConfig) GetIoBytePerSecond() int64 { + if x != nil { + return x.IoBytePerSecond + } + return 0 +} + // ReplicationTaskConfig contains replication-specific configuration type ReplicationTaskConfig struct { state protoimpl.MessageState `protogen:"open.v1"` @@ -3153,9 +3169,10 @@ type EcBalanceTaskParams struct { // For a dedup move, the node that keeps the shard. The worker confirms this // node really holds it before deleting the copy, so a topology entry naming a // location that holds nothing cannot cause the last copy to be removed. - DedupKeepNode string `protobuf:"bytes,5,opt,name=dedup_keep_node,json=dedupKeepNode,proto3" json:"dedup_keep_node,omitempty"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + DedupKeepNode string `protobuf:"bytes,5,opt,name=dedup_keep_node,json=dedupKeepNode,proto3" json:"dedup_keep_node,omitempty"` + IoBytePerSecond int64 `protobuf:"varint,6,opt,name=io_byte_per_second,json=ioBytePerSecond,proto3" json:"io_byte_per_second,omitempty"` // limit each shard copy's rate; 0 falls back to the volume server's maintenance rate + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *EcBalanceTaskParams) Reset() { @@ -3223,6 +3240,13 @@ func (x *EcBalanceTaskParams) GetDedupKeepNode() string { return "" } +func (x *EcBalanceTaskParams) GetIoBytePerSecond() int64 { + if x != nil { + return x.IoBytePerSecond + } + return 0 +} + // EcShardMoveSpec describes a single EC shard move within a batch type EcShardMoveSpec struct { state protoimpl.MessageState `protogen:"open.v1"` @@ -3325,6 +3349,7 @@ type EcBalanceTaskConfig struct { DiskType string `protobuf:"bytes,4,opt,name=disk_type,json=diskType,proto3" json:"disk_type,omitempty"` // Disk type filter PreferredTags []string `protobuf:"bytes,5,rep,name=preferred_tags,json=preferredTags,proto3" json:"preferred_tags,omitempty"` // Preferred disk tags for 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 + IoBytePerSecond int64 `protobuf:"varint,7,opt,name=io_byte_per_second,json=ioBytePerSecond,proto3" json:"io_byte_per_second,omitempty"` // limit each shard copy's rate; 0 falls back to the volume server's maintenance rate unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -3401,6 +3426,13 @@ func (x *EcBalanceTaskConfig) GetReplicaPlacement() string { return "" } +func (x *EcBalanceTaskConfig) GetIoBytePerSecond() int64 { + if x != nil { + return x.IoBytePerSecond + } + return 0 +} + // MaintenanceTaskData represents complete task state for persistence type MaintenanceTaskData struct { state protoimpl.MessageState `protogen:"open.v1"` @@ -4118,13 +4150,14 @@ const file_worker_proto_rawDesc = "" + "collection\x18\x04 \x01(\tR\n" + "collection\x12\x1f\n" + "\vvolume_size\x18\x05 \x01(\x04R\n" + - "volumeSize\"\xbf\x01\n" + + "volumeSize\"\xec\x01\n" + "\x11BalanceTaskParams\x12\x1d\n" + "\n" + "force_move\x18\x01 \x01(\bR\tforceMove\x12'\n" + "\x0ftimeout_seconds\x18\x02 \x01(\x05R\x0etimeoutSeconds\x120\n" + "\x14max_concurrent_moves\x18\x03 \x01(\x05R\x12maxConcurrentMoves\x120\n" + - "\x05moves\x18\x04 \x03(\v2\x1a.worker_pb.BalanceMoveSpecR\x05moves\"k\n" + + "\x05moves\x18\x04 \x03(\v2\x1a.worker_pb.BalanceMoveSpecR\x05moves\x12+\n" + + "\x12io_byte_per_second\x18\x05 \x01(\x03R\x0fioBytePerSecond\"k\n" + "\x15ReplicationTaskParams\x12#\n" + "\rreplica_count\x18\x01 \x01(\x05R\freplicaCount\x12-\n" + "\x12verify_consistency\x18\x02 \x01(\bR\x11verifyConsistency\"\x8e\x02\n" + @@ -4254,18 +4287,20 @@ const file_worker_proto_rawDesc = "" + "\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\x12+\n" + - "\x11replica_placement\x18\x06 \x01(\tR\x10replicaPlacement\"n\n" + + "\x11replica_placement\x18\x06 \x01(\tR\x10replicaPlacement\"\x9b\x01\n" + "\x11BalanceTaskConfig\x12/\n" + "\x13imbalance_threshold\x18\x01 \x01(\x01R\x12imbalanceThreshold\x12(\n" + - "\x10min_server_count\x18\x02 \x01(\x05R\x0eminServerCount\"I\n" + + "\x10min_server_count\x18\x02 \x01(\x05R\x0eminServerCount\x12+\n" + + "\x12io_byte_per_second\x18\x03 \x01(\x03R\x0fioBytePerSecond\"I\n" + "\x15ReplicationTaskConfig\x120\n" + - "\x14target_replica_count\x18\x01 \x01(\x05R\x12targetReplicaCount\"\xe6\x01\n" + + "\x14target_replica_count\x18\x01 \x01(\x05R\x12targetReplicaCount\"\x93\x02\n" + "\x13EcBalanceTaskParams\x12\x1b\n" + "\tdisk_type\x18\x01 \x01(\tR\bdiskType\x12/\n" + "\x13max_parallelization\x18\x02 \x01(\x05R\x12maxParallelization\x12'\n" + "\x0ftimeout_seconds\x18\x03 \x01(\x05R\x0etimeoutSeconds\x120\n" + "\x05moves\x18\x04 \x03(\v2\x1a.worker_pb.EcShardMoveSpecR\x05moves\x12&\n" + - "\x0fdedup_keep_node\x18\x05 \x01(\tR\rdedupKeepNode\"\xf7\x01\n" + + "\x0fdedup_keep_node\x18\x05 \x01(\tR\rdedupKeepNode\x12+\n" + + "\x12io_byte_per_second\x18\x06 \x01(\x03R\x0fioBytePerSecond\"\xf7\x01\n" + "\x0fEcShardMoveSpec\x12\x1b\n" + "\tvolume_id\x18\x01 \x01(\rR\bvolumeId\x12\x19\n" + "\bshard_id\x18\x02 \x01(\rR\ashardId\x12\x1e\n" + @@ -4277,14 +4312,15 @@ const file_worker_proto_rawDesc = "" + "\x0esource_disk_id\x18\x05 \x01(\rR\fsourceDiskId\x12\x1f\n" + "\vtarget_node\x18\x06 \x01(\tR\n" + "targetNode\x12$\n" + - "\x0etarget_disk_id\x18\a \x01(\rR\ftargetDiskId\"\x8e\x02\n" + + "\x0etarget_disk_id\x18\a \x01(\rR\ftargetDiskId\"\xbb\x02\n" + "\x13EcBalanceTaskConfig\x12/\n" + "\x13imbalance_threshold\x18\x01 \x01(\x01R\x12imbalanceThreshold\x12(\n" + "\x10min_server_count\x18\x02 \x01(\x05R\x0eminServerCount\x12+\n" + "\x11collection_filter\x18\x03 \x01(\tR\x10collectionFilter\x12\x1b\n" + "\tdisk_type\x18\x04 \x01(\tR\bdiskType\x12%\n" + "\x0epreferred_tags\x18\x05 \x03(\tR\rpreferredTags\x12+\n" + - "\x11replica_placement\x18\x06 \x01(\tR\x10replicaPlacement\"\xae\a\n" + + "\x11replica_placement\x18\x06 \x01(\tR\x10replicaPlacement\x12+\n" + + "\x12io_byte_per_second\x18\a \x01(\x03R\x0fioBytePerSecond\"\xae\a\n" + "\x13MaintenanceTaskData\x12\x0e\n" + "\x02id\x18\x01 \x01(\tR\x02id\x12\x12\n" + "\x04type\x18\x02 \x01(\tR\x04type\x12\x1a\n" + diff --git a/weed/pb/worker_pb/worker_grpc.pb.go b/weed/pb/worker_pb/worker_grpc.pb.go index 7e7aba748..649ab6165 100644 --- a/weed/pb/worker_pb/worker_grpc.pb.go +++ b/weed/pb/worker_pb/worker_grpc.pb.go @@ -1,7 +1,7 @@ // Code generated by protoc-gen-go-grpc. DO NOT EDIT. // versions: -// - protoc-gen-go-grpc v1.5.1 -// - protoc v6.33.4 +// - protoc-gen-go-grpc v1.6.2 +// - protoc v7.35.0 // source: worker.proto package worker_pb @@ -72,7 +72,7 @@ type WorkerServiceServer interface { type UnimplementedWorkerServiceServer struct{} func (UnimplementedWorkerServiceServer) WorkerStream(grpc.BidiStreamingServer[WorkerMessage, AdminMessage]) error { - return status.Errorf(codes.Unimplemented, "method WorkerStream not implemented") + return status.Error(codes.Unimplemented, "method WorkerStream not implemented") } func (UnimplementedWorkerServiceServer) mustEmbedUnimplementedWorkerServiceServer() {} func (UnimplementedWorkerServiceServer) testEmbeddedByValue() {} @@ -85,7 +85,7 @@ type UnsafeWorkerServiceServer interface { } func RegisterWorkerServiceServer(s grpc.ServiceRegistrar, srv WorkerServiceServer) { - // If the following call pancis, it indicates UnimplementedWorkerServiceServer was + // If the following call panics, it indicates UnimplementedWorkerServiceServer was // embedded by pointer and is nil. This will cause panics if an // unimplemented method is ever invoked, so we test this at initialization // time to prevent it from happening at runtime later due to I/O. diff --git a/weed/server/volume_grpc_erasure_coding.go b/weed/server/volume_grpc_erasure_coding.go index 086808d01..9004bf964 100644 --- a/weed/server/volume_grpc_erasure_coding.go +++ b/weed/server/volume_grpc_erasure_coding.go @@ -349,11 +349,19 @@ func (vs *VolumeServer) VolumeEcShardsCopy(ctx context.Context, req *volume_serv dataBaseFileName := storage.VolumeFileName(location.Directory, req.Collection, int(req.VolumeId)) indexBaseFileName := storage.VolumeFileName(location.IdxDirectory, req.Collection, int(req.VolumeId)) + // One throttler for the whole request, so the limit caps the transfer as a + // whole rather than each file separately — same shape as VolumeCopy. + ioBytePerSecond := vs.maintenanceBytePerSecond + if req.IoBytePerSecond > 0 { + ioBytePerSecond = req.IoBytePerSecond + } + throttler := util.NewWriteThrottler(ioBytePerSecond) + err := operation.WithVolumeServerClient(true, pb.ServerAddress(req.SourceDataNode), vs.grpcDialOption, func(client volume_server_pb.VolumeServerClient) error { // copy ec data slices for _, shardId := range req.ShardIds { - if _, err := vs.doCopyFile(client, true, req.Collection, req.VolumeId, math.MaxUint32, math.MaxInt64, dataBaseFileName, erasure_coding.ToExt(int(shardId)), false, false, nil); err != nil { + if _, err := vs.doCopyFileWithThrottler(client, true, req.Collection, req.VolumeId, math.MaxUint32, math.MaxInt64, dataBaseFileName, erasure_coding.ToExt(int(shardId)), false, false, nil, throttler); err != nil { return err } } @@ -361,7 +369,7 @@ func (vs *VolumeServer) VolumeEcShardsCopy(ctx context.Context, req *volume_serv if req.CopyEcxFile { // copy ecx file - if _, err := vs.doCopyFile(client, true, req.Collection, req.VolumeId, math.MaxUint32, math.MaxInt64, indexBaseFileName, ".ecx", false, false, nil); err != nil { + if _, err := vs.doCopyFileWithThrottler(client, true, req.Collection, req.VolumeId, math.MaxUint32, math.MaxInt64, indexBaseFileName, ".ecx", false, false, nil, throttler); err != nil { return err } // Defense in depth: writeToFile now removes partial files on @@ -394,14 +402,14 @@ func (vs *VolumeServer) VolumeEcShardsCopy(ctx context.Context, req *volume_serv if req.CopyEcjFile { // copy ecj file - if _, err := vs.doCopyFile(client, true, req.Collection, req.VolumeId, math.MaxUint32, math.MaxInt64, indexBaseFileName, ".ecj", true, true, nil); err != nil { + if _, err := vs.doCopyFileWithThrottler(client, true, req.Collection, req.VolumeId, math.MaxUint32, math.MaxInt64, indexBaseFileName, ".ecj", true, true, nil, throttler); err != nil { return err } } if req.CopyVifFile { // copy vif file - if _, err := vs.doCopyFile(client, true, req.Collection, req.VolumeId, math.MaxUint32, math.MaxInt64, dataBaseFileName, ".vif", false, true, nil); err != nil { + if _, err := vs.doCopyFileWithThrottler(client, true, req.Collection, req.VolumeId, math.MaxUint32, math.MaxInt64, dataBaseFileName, ".vif", false, true, nil, throttler); err != nil { return err } } @@ -412,7 +420,7 @@ func (vs *VolumeServer) VolumeEcShardsCopy(ctx context.Context, req *volume_serv // distribution) has no Prepare backstop, and fresh-encode sidecar // writes are best-effort, so a missing source sidecar is a no-op // (ignore-not-found): the holder is simply unprotected. - if _, err := vs.doCopyFile(client, true, req.Collection, req.VolumeId, math.MaxUint32, math.MaxInt64, dataBaseFileName, erasure_coding.BitrotSidecarExt, false, true, nil); err != nil { + if _, err := vs.doCopyFileWithThrottler(client, true, req.Collection, req.VolumeId, math.MaxUint32, math.MaxInt64, dataBaseFileName, erasure_coding.BitrotSidecarExt, false, true, nil, throttler); err != nil { return fmt.Errorf("VolumeEcShardsCopy volume %d: copy %s sidecar: %w", req.VolumeId, erasure_coding.BitrotSidecarExt, err) } } diff --git a/weed/shell/command_ec_balance.go b/weed/shell/command_ec_balance.go index 695ec676b..0ce9127a0 100644 --- a/weed/shell/command_ec_balance.go +++ b/weed/shell/command_ec_balance.go @@ -48,6 +48,7 @@ func (c *commandEcBalance) Do(args []string, commandEnv *CommandEnv, writer io.W diskTypeStr := balanceCommand.String("diskType", "", "the disk type for EC shards (hdd, ssd, or empty for default hdd)") volumeIdsStr := balanceCommand.String("volumeIds", "", "optional comma-separated list of ec volume ids to balance; defaults to all") maxParallelization := balanceCommand.Int("maxParallelization", DefaultMaxParallelization, "run up to X tasks in parallel, whenever possible") + ioBytePerSecond := balanceCommand.Int64("ioBytePerSecond", 0, "limit the speed of each shard copy; 0 falls back to the volume server's maintenance rate") applyBalancing := balanceCommand.Bool("apply", false, "apply the balancing plan") // TODO: remove this alias applyBalancingAlias := balanceCommand.Bool("force", false, "apply the balancing plan (alias for -apply)") @@ -88,5 +89,5 @@ func (c *commandEcBalance) Do(args []string, commandEnv *CommandEnv, writer io.W diskType := types.ToDiskType(*diskTypeStr) - return EcBalance(commandEnv, collections, *dc, rp, diskType, *maxParallelization, *applyBalancing, nil, volumeIds) + return EcBalance(commandEnv, collections, *dc, rp, diskType, *maxParallelization, *ioBytePerSecond, *applyBalancing, nil, volumeIds) } diff --git a/weed/shell/command_ec_common.go b/weed/shell/command_ec_common.go index 54fbbf479..82843c32f 100644 --- a/weed/shell/command_ec_common.go +++ b/weed/shell/command_ec_common.go @@ -390,7 +390,7 @@ func oneServerCopyAndMountEcShardsFromSource(grpcDialOption grpc.DialOption, fmt.Printf("allocate %d.%v %s => %s\n", volumeId, shardIdsToCopy, existingLocation, targetServer.info.Id) targetAddress := pb.NewServerAddressFromDataNode(targetServer.info) - err = volume_move.NewMover(grpcDialOption).CopyAndMountEcShards(context.Background(), volumeId, collection, shardIdsToCopy, existingLocation, targetAddress, destDiskId, os.Stdout) + err = volume_move.NewMover(grpcDialOption).CopyAndMountEcShards(context.Background(), volumeId, collection, shardIdsToCopy, existingLocation, targetAddress, destDiskId, 0, os.Stdout) if err != nil { return } @@ -781,6 +781,7 @@ type ecBalancer struct { replicaPlacement *super_block.ReplicaPlacement applyBalancing bool maxParallelization int + ioBytePerSecond int64 diskType types.DiskType // volumeIds narrows the plan to these ec volume ids; nil balances every volume // of the selected collections. @@ -795,7 +796,7 @@ type ecBalancer struct { // // volumeIds, when non-empty, restricts the plan to those ec volume ids; empty // balances every volume of the given collections. -func EcBalance(commandEnv *CommandEnv, collections []string, dc string, ecReplicaPlacement *super_block.ReplicaPlacement, diskType types.DiskType, maxParallelization int, applyBalancing bool, excludeNodes map[pb.ServerAddress]struct{}, volumeIds []needle.VolumeId) (err error) { +func EcBalance(commandEnv *CommandEnv, collections []string, dc string, ecReplicaPlacement *super_block.ReplicaPlacement, diskType types.DiskType, maxParallelization int, ioBytePerSecond int64, applyBalancing bool, excludeNodes map[pb.ServerAddress]struct{}, volumeIds []needle.VolumeId) (err error) { // collect all ec nodes allEcNodes, totalFreeEcSlots, err := collectEcNodesForDC(commandEnv, dc, diskType) if err != nil { @@ -837,6 +838,7 @@ func EcBalance(commandEnv *CommandEnv, collections []string, dc string, ecReplic replicaPlacement: ecReplicaPlacement, applyBalancing: applyBalancing, maxParallelization: maxParallelization, + ioBytePerSecond: ioBytePerSecond, diskType: diskType, volumeIds: volumeIdFilter, } @@ -1108,7 +1110,7 @@ func (ecb *ecBalancer) applyShardMoveRPC(src, dst *EcNode, collection string, vi Source: srcAddr, Target: dstAddr, TargetDisk: destDiskId, - }, volume_move.EcMoveOptions{Writer: os.Stdout}) + }, volume_move.EcMoveOptions{IoBytePerSecond: ecb.ioBytePerSecond, Writer: os.Stdout}) } // parseVolumeIdsFlag parses a comma-separated -volumeIds flag value, dropping diff --git a/weed/shell/command_ec_encode.go b/weed/shell/command_ec_encode.go index fef67fb6c..a4bcc53f0 100644 --- a/weed/shell/command_ec_encode.go +++ b/weed/shell/command_ec_encode.go @@ -274,7 +274,7 @@ func processEcEncodeBatch(commandEnv *CommandEnv, writer io.Writer, volumeIds [] // safely verified and deleted without waiting for all batches to finish. // skippedNodes are excluded so a recovered node's stale orphan is never // paired with a new-generation shard. - if err := EcBalance(commandEnv, balanceCollections, "", rp, diskType, maxParallelization, applyBalancing, skippedNodes, nil); err != nil { + if err := EcBalance(commandEnv, balanceCollections, "", rp, diskType, maxParallelization, 0, applyBalancing, skippedNodes, nil); err != nil { return fmt.Errorf("re-balance ec shards for collection(s) %v: %w", balanceCollections, err) } if err := verifyEcShardsBeforeDelete(commandEnv, volumeIds, diskType, applyBalancing); err != nil { diff --git a/weed/shell/command_fs_distribute_chunks.go b/weed/shell/command_fs_distribute_chunks.go index 2ff2eedd4..b66df41dc 100644 --- a/weed/shell/command_fs_distribute_chunks.go +++ b/weed/shell/command_fs_distribute_chunks.go @@ -6,9 +6,9 @@ import ( "flag" "fmt" "io" + "maps" "math" "net/http" - "maps" "slices" "sort" "strings" @@ -87,7 +87,6 @@ type chunkMove struct { toNode string } - func (c *commandFsDistributeChunks) Do(args []string, commandEnv *CommandEnv, writer io.Writer) (err error) { fsDistributeCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError) diff --git a/weed/shell/command_s3_lifecycle_run_shard.go b/weed/shell/command_s3_lifecycle_run_shard.go index e53b56c53..9ec168298 100644 --- a/weed/shell/command_s3_lifecycle_run_shard.go +++ b/weed/shell/command_s3_lifecycle_run_shard.go @@ -180,7 +180,7 @@ func (c *commandS3LifecycleRunShard) Do(args []string, env *CommandEnv, writer i // integration tests and CI. Fan out across the selected shards // so recovery walks do not serialize 16 shard scans into a 10s // timeout budget. - Workers: len(shards), + Workers: len(shards), Walker: walker, EventBudget: *eventBudget, ClientName: fmt.Sprintf("shell-lifecycle-%s", formatShardLabel(shards)), diff --git a/weed/shell/command_volume_fix_replication.go b/weed/shell/command_volume_fix_replication.go index 21e05d583..c2d9f8cba 100644 --- a/weed/shell/command_volume_fix_replication.go +++ b/weed/shell/command_volume_fix_replication.go @@ -542,7 +542,7 @@ func (c *commandVolumeFixReplication) fixOneUnderReplicatedVolume(commandEnv *Co err := replicateVolumeToServer(context.Background(), commandEnv.option.GrpcDialOption, writer, needle.VolumeId(replica.info.Id), pb.NewServerAddressFromDataNode(replica.location.dataNode), pb.NewServerAddressFromDataNode(dst.dataNode), - replica.info.DiskType) + replica.info.DiskType, 0) scheduler.releaseTarget(dst, replica.info.DiskType, err == nil) if err != nil { return false, err diff --git a/weed/shell/command_volume_move.go b/weed/shell/command_volume_move.go index c19f0cd52..41d0c5b40 100644 --- a/weed/shell/command_volume_move.go +++ b/weed/shell/command_volume_move.go @@ -159,8 +159,8 @@ func markVolumeReplicasWritable(ctx context.Context, grpcDialOption grpc.DialOpt } // replicateVolumeToServer copies a volume from sourceAddress to targetAddress via the VolumeCopy gRPC stream. -func replicateVolumeToServer(ctx context.Context, grpcDialOption grpc.DialOption, writer io.Writer, volumeId needle.VolumeId, sourceAddress, targetAddress pb.ServerAddress, diskType string) error { - return volume_move.NewMover(grpcDialOption).ReplicateVolume(ctx, volumeId, sourceAddress, targetAddress, diskType, writer) +func replicateVolumeToServer(ctx context.Context, grpcDialOption grpc.DialOption, writer io.Writer, volumeId needle.VolumeId, sourceAddress, targetAddress pb.ServerAddress, diskType string, ioBytePerSecond int64) error { + return volume_move.NewMover(grpcDialOption).ReplicateVolume(ctx, volumeId, sourceAddress, targetAddress, diskType, ioBytePerSecond, writer) } // configureVolumeReplication sets the replication setting on a volume at the given server. diff --git a/weed/shell/command_volume_tier_move.go b/weed/shell/command_volume_tier_move.go index eef50210e..8e5efb0db 100644 --- a/weed/shell/command_volume_tier_move.go +++ b/weed/shell/command_volume_tier_move.go @@ -358,7 +358,7 @@ func (c *commandVolumeTierMove) doMoveOneVolume(commandEnv *CommandEnv, writer i // deleting old replicas to avoid data-loss risk. // Use the explicit -toReplication if given, otherwise preserve the volume's // existing replication from the source tier. - preserveServers, replicateErr := c.ensureReplicationFulfilled(commandEnv, writer, vid, toDiskType, toDataCenter, dst, *replicationString) + preserveServers, replicateErr := c.ensureReplicationFulfilled(commandEnv, writer, vid, toDiskType, toDataCenter, dst, *replicationString, ioBytePerSecond) if replicateErr != nil { // Replication not fully achieved — do NOT delete old replicas. restoreSurvivingReplicasWritable(commandEnv, vid, locations, deletedSource) @@ -413,7 +413,7 @@ func restoreSurvivingReplicasWritable(commandEnv *CommandEnv, vid needle.VolumeI // move so it can see the newly placed volume and find suitable destinations for additional copies. // It returns a set of server URLs (from the original locations) that host target-tier replicas // counted toward fulfillment, so the caller can avoid deleting them during cleanup. -func (c *commandVolumeTierMove) ensureReplicationFulfilled(commandEnv *CommandEnv, writer io.Writer, vid needle.VolumeId, toDiskType types.DiskType, toDataCenter string, movedDst location, replicationString string) (preserveServers map[string]bool, err error) { +func (c *commandVolumeTierMove) ensureReplicationFulfilled(commandEnv *CommandEnv, writer io.Writer, vid needle.VolumeId, toDiskType types.DiskType, toDataCenter string, movedDst location, replicationString string, ioBytePerSecond int64) (preserveServers map[string]bool, err error) { preserveServers = make(map[string]bool) // keeps an anchored pre-existing replica writable on the early-return paths preserveServers[movedDst.dataNode.Id] = true @@ -529,7 +529,7 @@ func (c *commandVolumeTierMove) ensureReplicationFulfilled(commandEnv *CommandEn candidateDst := placeOf[dst.Id] candidateAddress := pb.NewServerAddressFromDataNode(dst) - if copyErr := replicateVolumeToServer(context.Background(), commandEnv.option.GrpcDialOption, writer, vid, sourceAddress, candidateAddress, toDiskType.ReadableString()); copyErr != nil { + if copyErr := replicateVolumeToServer(context.Background(), commandEnv.option.GrpcDialOption, writer, vid, sourceAddress, candidateAddress, toDiskType.ReadableString(), ioBytePerSecond); copyErr != nil { return nil, fmt.Errorf("replicate volume %d to %s: %v", vid, candidateDst.dataNode.Id, copyErr) } diff --git a/weed/shell/command_volume_tier_upload.go b/weed/shell/command_volume_tier_upload.go index a48607435..4bf83e124 100644 --- a/weed/shell/command_volume_tier_upload.go +++ b/weed/shell/command_volume_tier_upload.go @@ -153,7 +153,7 @@ func doVolumeTierUpload(commandEnv *CommandEnv, writer io.Writer, collection str continue } fmt.Fprintf(writer, "replicate remote volume %d metadata from %s to %s\n", vid, existingLocations[0].Url, location.Url) - err = replicateVolumeToServer(context.Background(), commandEnv.option.GrpcDialOption, writer, vid, existingLocations[0].ServerAddress(), location.ServerAddress(), "") + err = replicateVolumeToServer(context.Background(), commandEnv.option.GrpcDialOption, writer, vid, existingLocations[0].ServerAddress(), location.ServerAddress(), "", 0) if err != nil { return fmt.Errorf("replicate volume %d from %s to %s: %v", vid, existingLocations[0].Url, location.Url, err) } diff --git a/weed/worker/tasks/balance/balance_task.go b/weed/worker/tasks/balance/balance_task.go index 84e62ef16..7bf1a1b93 100644 --- a/weed/worker/tasks/balance/balance_task.go +++ b/weed/worker/tasks/balance/balance_task.go @@ -77,7 +77,8 @@ func (t *BalanceTask) Execute(ctx context.Context, params *worker_pb.TaskParams) // shell's volume.move/volume.balance commands. mover := volume_move.NewMover(t.grpcDialOption) err := mover.LiveMoveVolume(ctx, needle.VolumeId(t.volumeID), pb.ServerAddress(sourceNode), pb.ServerAddress(destNode), volume_move.VolumeMoveOptions{ - IdleTimeout: 60 * time.Second, + IdleTimeout: 60 * time.Second, + IoBytePerSecond: balanceParams.IoBytePerSecond, Progress: func(percent float64, stage string) { t.ReportProgress(percent) t.GetLogger().Info(stage) diff --git a/weed/worker/tasks/balance/config.go b/weed/worker/tasks/balance/config.go index e9d2af637..1616788ee 100644 --- a/weed/worker/tasks/balance/config.go +++ b/weed/worker/tasks/balance/config.go @@ -14,6 +14,7 @@ type Config struct { base.BaseConfig ImbalanceThreshold float64 `json:"imbalance_threshold"` MinServerCount int `json:"min_server_count"` + IoBytePerSecond int64 `json:"io_byte_per_second"` DataCenterFilter string `json:"-"` // per-detection-run, not persisted RackFilter string `json:"-"` // per-detection-run, not persisted NodeFilter string `json:"-"` // per-detection-run, not persisted @@ -96,6 +97,21 @@ func GetConfigSpec() base.ConfigSpec { InputType: "number", CSSClasses: "form-control", }, + { + Name: "io_byte_per_second", + JSONName: "io_byte_per_second", + Type: config.FieldTypeInt, + DefaultValue: 0, + MinValue: 0, + Required: false, + DisplayName: "Move IO Limit (bytes/sec)", + Description: "Limit each volume move's copy rate", + HelpText: "0 falls back to each volume server's own maintenance rate (-maintenanceBytePerSecond)", + Placeholder: "0 (server maintenance rate)", + Unit: config.UnitNone, + InputType: "number", + CSSClasses: "form-control", + }, { Name: "min_server_count", JSONName: "min_server_count", @@ -127,6 +143,7 @@ func (c *Config) ToTaskPolicy() *worker_pb.TaskPolicy { BalanceConfig: &worker_pb.BalanceTaskConfig{ ImbalanceThreshold: float64(c.ImbalanceThreshold), MinServerCount: int32(c.MinServerCount), + IoBytePerSecond: c.IoBytePerSecond, }, }, } @@ -147,6 +164,7 @@ func (c *Config) FromTaskPolicy(policy *worker_pb.TaskPolicy) error { if balanceConfig := policy.GetBalanceConfig(); balanceConfig != nil { c.ImbalanceThreshold = float64(balanceConfig.ImbalanceThreshold) c.MinServerCount = int(balanceConfig.MinServerCount) + c.IoBytePerSecond = balanceConfig.IoBytePerSecond } return nil diff --git a/weed/worker/tasks/balance/detection.go b/weed/worker/tasks/balance/detection.go index 4ff865b4c..03268f047 100644 --- a/weed/worker/tasks/balance/detection.go +++ b/weed/worker/tasks/balance/detection.go @@ -400,7 +400,7 @@ func detectForDiskType(diskType string, diskMetrics []*types.VolumeHealthMetrics eligibleTargets[s] = c } } - task, destServerID := createBalanceTask(diskType, selectedVolume, clusterInfo, minServer, eligibleTargets) + task, destServerID := createBalanceTask(diskType, selectedVolume, clusterInfo, minServer, eligibleTargets, balanceConfig.IoBytePerSecond) if task == nil { glog.V(1).Infof("BALANCE [%s]: Cannot plan task for volume %d on server %s, trying next volume", diskType, selectedVolume.VolumeID, maxServer) continue @@ -438,7 +438,7 @@ func detectForDiskType(diskType string, diskMetrics []*types.VolumeHealthMetrics // allowedServers is the set of servers that passed DC/rack/node filtering in // the detection loop. When non-empty, the fallback destination planner is // checked against this set so that filter scope cannot leak. -func createBalanceTask(diskType string, selectedVolume *types.VolumeHealthMetrics, clusterInfo *types.ClusterInfo, targetServer string, allowedServers map[string]int) (*types.TaskDetectionResult, string) { +func createBalanceTask(diskType string, selectedVolume *types.VolumeHealthMetrics, clusterInfo *types.ClusterInfo, targetServer string, allowedServers map[string]int, ioBytePerSecond int64) (*types.TaskDetectionResult, string) { taskID := fmt.Sprintf("balance_vol_%d_%d", selectedVolume.VolumeID, time.Now().UnixNano()) task := &types.TaskDetectionResult{ @@ -566,8 +566,9 @@ func createBalanceTask(diskType string, selectedVolume *types.VolumeHealthMetric TaskParams: &worker_pb.TaskParams_BalanceParams{ BalanceParams: &worker_pb.BalanceTaskParams{ - ForceMove: false, - TimeoutSeconds: 600, // 10 minutes default + ForceMove: false, + TimeoutSeconds: 600, // 10 minutes default + IoBytePerSecond: ioBytePerSecond, }, }, } diff --git a/weed/worker/tasks/balance/plugin_handler.go b/weed/worker/tasks/balance/plugin_handler.go index d271b0bd2..f9cedab55 100644 --- a/weed/worker/tasks/balance/plugin_handler.go +++ b/weed/worker/tasks/balance/plugin_handler.go @@ -176,6 +176,14 @@ func (h *VolumeBalanceHandler) Descriptor() *plugin_pb.JobTypeDescriptor { Required: true, MinValue: &plugin_pb.ConfigValue{Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 2}}, }, + { + Name: "io_byte_per_second", + Label: "Move IO Limit (bytes/sec)", + Description: "Limit each volume move's copy's rate in bytes per second. 0 falls back to each volume server's own maintenance rate.", + FieldType: plugin_pb.ConfigFieldType_CONFIG_FIELD_TYPE_INT64, + Widget: plugin_pb.ConfigWidget_CONFIG_WIDGET_NUMBER, + MinValue: &plugin_pb.ConfigValue{Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 0}}, + }, }, }, { @@ -211,6 +219,9 @@ func (h *VolumeBalanceHandler) Descriptor() *plugin_pb.JobTypeDescriptor { "min_server_count": { Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 2}, }, + "io_byte_per_second": { + Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 0}, + }, "max_concurrent_moves": { Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: int64(defaultMaxConcurrentMoves)}, }, @@ -238,6 +249,9 @@ func (h *VolumeBalanceHandler) Descriptor() *plugin_pb.JobTypeDescriptor { "min_server_count": { Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 2}, }, + "io_byte_per_second": { + Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 0}, + }, "max_concurrent_moves": { Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: int64(defaultMaxConcurrentMoves)}, }, @@ -1017,11 +1031,13 @@ func (h *VolumeBalanceHandler) executeBatchMoves( func buildMoveTaskParams(move *worker_pb.BalanceMoveSpec, outerParams *worker_pb.BalanceTaskParams) *worker_pb.TaskParams { timeoutSeconds := defaultBalanceTimeoutSeconds forceMove := false + var ioBytePerSecond int64 if outerParams != nil { if outerParams.TimeoutSeconds > 0 { timeoutSeconds = outerParams.TimeoutSeconds } forceMove = outerParams.ForceMove + ioBytePerSecond = outerParams.IoBytePerSecond } return &worker_pb.TaskParams{ VolumeId: move.VolumeId, @@ -1035,8 +1051,9 @@ func buildMoveTaskParams(move *worker_pb.BalanceMoveSpec, outerParams *worker_pb }, TaskParams: &worker_pb.TaskParams_BalanceParams{ BalanceParams: &worker_pb.BalanceTaskParams{ - ForceMove: forceMove, - TimeoutSeconds: timeoutSeconds, + ForceMove: forceMove, + TimeoutSeconds: timeoutSeconds, + IoBytePerSecond: ioBytePerSecond, }, }, } @@ -1068,6 +1085,12 @@ func deriveBalanceWorkerConfig(values map[string]*plugin_pb.ConfigValue) *volume } taskConfig.MinServerCount = minServerCount + ioBytePerSecond := pluginworker.ReadInt64Config(values, "io_byte_per_second", taskConfig.IoBytePerSecond) + if ioBytePerSecond < 0 { + ioBytePerSecond = 0 + } + taskConfig.IoBytePerSecond = ioBytePerSecond + maxConcurrentMoves := pluginworker.ReadIntConfig(values, "max_concurrent_moves", defaultMaxConcurrentMoves) if maxConcurrentMoves < 1 { maxConcurrentMoves = 1 @@ -1277,13 +1300,19 @@ func buildBatchVolumeBalanceProposals( continue } - // Serialize batch params + // Serialize batch params. The io limit rides along from the detection + // results, which carry it from the task configuration. + var ioBytePerSecond int64 + if p := batch[0].TypedParams.GetBalanceParams(); p != nil { + ioBytePerSecond = p.IoBytePerSecond + } taskParams := &worker_pb.TaskParams{ TaskParams: &worker_pb.TaskParams_BalanceParams{ BalanceParams: &worker_pb.BalanceTaskParams{ TimeoutSeconds: defaultBalanceTimeoutSeconds, MaxConcurrentMoves: int32(maxConcurrentMoves), Moves: moves, + IoBytePerSecond: ioBytePerSecond, }, }, } @@ -1399,6 +1428,7 @@ func decodeVolumeBalanceTaskParams(job *plugin_pb.JobSpec) (*worker_pb.TaskParam timeoutSeconds = defaultBalanceTimeoutSeconds } forceMove := readBoolConfig(job.Parameters, "force_move", false) + ioBytePerSecond := pluginworker.ReadInt64Config(job.Parameters, "io_byte_per_second", 0) if volumeID == 0 { return nil, fmt.Errorf("missing volume_id in job parameters") @@ -1428,8 +1458,9 @@ func decodeVolumeBalanceTaskParams(job *plugin_pb.JobSpec) (*worker_pb.TaskParam }, TaskParams: &worker_pb.TaskParams_BalanceParams{ BalanceParams: &worker_pb.BalanceTaskParams{ - ForceMove: forceMove, - TimeoutSeconds: timeoutSeconds, + ForceMove: forceMove, + TimeoutSeconds: timeoutSeconds, + IoBytePerSecond: ioBytePerSecond, }, }, }, nil diff --git a/weed/worker/tasks/balance/replica_placement_test.go b/weed/worker/tasks/balance/replica_placement_test.go index baa8f1eb7..396ea3b30 100644 --- a/weed/worker/tasks/balance/replica_placement_test.go +++ b/weed/worker/tasks/balance/replica_placement_test.go @@ -240,7 +240,7 @@ func TestCreateBalanceTask_FallbackSelectsValidCompositeDestination(t *testing.T task, destination := createBalanceTask("hdd", volumes[0], clusterInfo, "node-b", map[string]int{ "node-b": 0, "node-c": 0, - }) + }, 0) if task == nil { t.Fatal("expected a balance task") } diff --git a/weed/worker/tasks/ec_balance/config.go b/weed/worker/tasks/ec_balance/config.go index 16515bae4..cc3397228 100644 --- a/weed/worker/tasks/ec_balance/config.go +++ b/weed/worker/tasks/ec_balance/config.go @@ -17,7 +17,8 @@ type Config struct { CollectionFilter string `json:"collection_filter"` DiskType string `json:"disk_type"` PreferredTags []string `json:"preferred_tags"` - ReplicaPlacement string `json:"replica_placement"` // e.g. "020"; empty falls back to the master default replication (even spread only when that default is empty or zero) + ReplicaPlacement string `json:"replica_placement"` // e.g. "020"; empty falls back to the master default replication (even spread only when that default is empty or zero) + IoBytePerSecond int64 `json:"io_byte_per_second"` // limit each shard copy's rate; 0 falls back to the volume server's maintenance rate DataCenterFilter string `json:"-"` // per-detection-run, not persisted } @@ -169,6 +170,21 @@ func GetConfigSpec() base.ConfigSpec { InputType: "text", CSSClasses: "form-control", }, + { + Name: "io_byte_per_second", + JSONName: "io_byte_per_second", + Type: config.FieldTypeInt, + DefaultValue: 0, + MinValue: 0, + Required: false, + DisplayName: "Shard Copy IO Limit (bytes/sec)", + Description: "Limit each EC shard copy's rate", + HelpText: "0 falls back to each volume server's own maintenance rate (-maintenanceBytePerSecond)", + Placeholder: "0 (server maintenance rate)", + Unit: config.UnitNone, + InputType: "number", + CSSClasses: "form-control", + }, }, } } @@ -189,6 +205,7 @@ func (c *Config) ToTaskPolicy() *worker_pb.TaskPolicy { DiskType: c.DiskType, PreferredTags: preferredTagsCopy, ReplicaPlacement: c.ReplicaPlacement, + IoBytePerSecond: c.IoBytePerSecond, }, }, } @@ -211,6 +228,7 @@ func (c *Config) FromTaskPolicy(policy *worker_pb.TaskPolicy) error { c.DiskType = ecbConfig.DiskType c.PreferredTags = append([]string(nil), ecbConfig.PreferredTags...) c.ReplicaPlacement = ecbConfig.ReplicaPlacement + c.IoBytePerSecond = ecbConfig.IoBytePerSecond } return nil diff --git a/weed/worker/tasks/ec_balance/detection.go b/weed/worker/tasks/ec_balance/detection.go index 3999f6e00..192ed2788 100644 --- a/weed/worker/tasks/ec_balance/detection.go +++ b/weed/worker/tasks/ec_balance/detection.go @@ -126,9 +126,10 @@ func Detection( }}, TaskParams: &worker_pb.TaskParams_EcBalanceParams{ EcBalanceParams: &worker_pb.EcBalanceTaskParams{ - DiskType: normalizedDiskType, - DedupKeepNode: m.KeepNode, - TimeoutSeconds: 600, + DiskType: normalizedDiskType, + DedupKeepNode: m.KeepNode, + TimeoutSeconds: 600, + IoBytePerSecond: ecConfig.IoBytePerSecond, }, }, }, diff --git a/weed/worker/tasks/ec_balance/ec_balance_task.go b/weed/worker/tasks/ec_balance/ec_balance_task.go index 5d2bf462c..b7d4222b8 100644 --- a/weed/worker/tasks/ec_balance/ec_balance_task.go +++ b/weed/worker/tasks/ec_balance/ec_balance_task.go @@ -124,7 +124,8 @@ func (t *ECBalanceTask) Execute(ctx context.Context, params *worker_pb.TaskParam Target: targetAddr, TargetDisk: target.DiskId, }, volume_move.EcMoveOptions{ - Progress: t.reportProgress, + IoBytePerSecond: ecParams.GetIoBytePerSecond(), + Progress: t.reportProgress, }) if err != nil { return err diff --git a/weed/worker/tasks/ec_balance/plugin_handler.go b/weed/worker/tasks/ec_balance/plugin_handler.go index 68d97191c..dc3b89468 100644 --- a/weed/worker/tasks/ec_balance/plugin_handler.go +++ b/weed/worker/tasks/ec_balance/plugin_handler.go @@ -149,6 +149,14 @@ func (h *ECBalanceHandler) Descriptor() *plugin_pb.JobTypeDescriptor { Required: true, MinValue: &plugin_pb.ConfigValue{Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: ecBalanceMinServerCount}}, }, + { + Name: "io_byte_per_second", + Label: "Shard Copy IO Limit (bytes/sec)", + Description: "Limit each EC shard copy's rate in bytes per second. 0 falls back to each volume server's own maintenance rate.", + FieldType: plugin_pb.ConfigFieldType_CONFIG_FIELD_TYPE_INT64, + Widget: plugin_pb.ConfigWidget_CONFIG_WIDGET_NUMBER, + MinValue: &plugin_pb.ConfigValue{Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 0}}, + }, { Name: "preferred_tags", Label: "Preferred Tags", @@ -163,6 +171,7 @@ func (h *ECBalanceHandler) Descriptor() *plugin_pb.JobTypeDescriptor { DefaultValues: map[string]*plugin_pb.ConfigValue{ "imbalance_threshold": {Kind: &plugin_pb.ConfigValue_DoubleValue{DoubleValue: 0.2}}, "min_server_count": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 3}}, + "io_byte_per_second": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 0}}, "preferred_tags": {Kind: &plugin_pb.ConfigValue_StringValue{StringValue: ""}}, }, }, @@ -181,6 +190,7 @@ func (h *ECBalanceHandler) Descriptor() *plugin_pb.JobTypeDescriptor { WorkerDefaultValues: map[string]*plugin_pb.ConfigValue{ "imbalance_threshold": {Kind: &plugin_pb.ConfigValue_DoubleValue{DoubleValue: 0.2}}, "min_server_count": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 3}}, + "io_byte_per_second": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 0}}, "preferred_tags": {Kind: &plugin_pb.ConfigValue_StringValue{StringValue: ""}}, }, } @@ -418,6 +428,12 @@ func deriveECBalanceWorkerConfig(values map[string]*plugin_pb.ConfigValue) *ecBa } taskConfig.MinServerCount = minServerCount + ioBytePerSecond := pluginworker.ReadInt64Config(values, "io_byte_per_second", taskConfig.IoBytePerSecond) + if ioBytePerSecond < 0 { + ioBytePerSecond = 0 + } + taskConfig.IoBytePerSecond = ioBytePerSecond + taskConfig.PreferredTags = util.NormalizeTagList(pluginworker.ReadStringListConfig(values, "preferred_tags")) return &ecBalanceWorkerConfig{ @@ -577,6 +593,7 @@ func decodeECBalanceTaskParams(job *plugin_pb.JobSpec) (*worker_pb.TaskParams, e if targetDiskID < 0 || targetDiskID > math.MaxUint32 { return nil, fmt.Errorf("decodeECBalanceTaskParams: invalid target_disk_id: %d", targetDiskID) } + ioBytePerSecond := pluginworker.ReadInt64Config(job.Parameters, "io_byte_per_second", 0) return &worker_pb.TaskParams{ TaskId: job.JobId, @@ -594,7 +611,8 @@ func decodeECBalanceTaskParams(job *plugin_pb.JobSpec) (*worker_pb.TaskParams, e }}, TaskParams: &worker_pb.TaskParams_EcBalanceParams{ EcBalanceParams: &worker_pb.EcBalanceTaskParams{ - TimeoutSeconds: 600, + TimeoutSeconds: 600, + IoBytePerSecond: ioBytePerSecond, }, }, }, nil diff --git a/weed/worker/tasks/erasure_coding/ec_task_encode_files_e2e_test.go b/weed/worker/tasks/erasure_coding/ec_task_encode_files_e2e_test.go index 66632a09b..baf6b42a0 100644 --- a/weed/worker/tasks/erasure_coding/ec_task_encode_files_e2e_test.go +++ b/weed/worker/tasks/erasure_coding/ec_task_encode_files_e2e_test.go @@ -66,8 +66,8 @@ func TestEcEncodeLeavesRightFilesAndRemovesStubAndSource(t *testing.T) { dataShards := int(erasure_coding.DataShardsCount) totalShards := int(erasure_coding.DataShardsCount + erasure_coding.ParityShardsCount) - aShards := shardRange(0, dataShards) // 0..DataShardsCount-1 on A - bShards := shardRange(dataShards, totalShards) // parity range on B + aShards := shardRange(0, dataShards) // 0..DataShardsCount-1 on A + bShards := shardRange(dataShards, totalShards) // parity range on B task := NewErasureCodingTask("ec-e2e", addrA, volumeID, collection, dialOption) params := &worker_pb.TaskParams{ diff --git a/weed/worker/tasks/erasure_coding/ec_task_julor_layout_e2e_test.go b/weed/worker/tasks/erasure_coding/ec_task_julor_layout_e2e_test.go index f543f0134..5961f2e67 100644 --- a/weed/worker/tasks/erasure_coding/ec_task_julor_layout_e2e_test.go +++ b/weed/worker/tasks/erasure_coding/ec_task_julor_layout_e2e_test.go @@ -108,9 +108,9 @@ func TestEcEncodeJulorLayoutConverges(t *testing.T) { VolumeId: volumeID, Collection: collection, Sources: []*worker_pb.TaskSource{ - {Node: addr(srcServer), VolumeId: volumeID}, // real source - {Node: addr(stubServer), VolumeId: volumeID}, // 0-byte stub - {Node: addr(srcServer), VolumeId: volumeID, ShardIds: staleShards}, // stale EC shards to clear + {Node: addr(srcServer), VolumeId: volumeID}, // real source + {Node: addr(stubServer), VolumeId: volumeID}, // 0-byte stub + {Node: addr(srcServer), VolumeId: volumeID, ShardIds: staleShards}, // stale EC shards to clear }, Targets: targets, TaskParams: &worker_pb.TaskParams_ErasureCodingParams{ diff --git a/weed/worker/tasks/iceberg/handler.go b/weed/worker/tasks/iceberg/handler.go index d2bfe8850..95fdcfc85 100644 --- a/weed/worker/tasks/iceberg/handler.go +++ b/weed/worker/tasks/iceberg/handler.go @@ -324,7 +324,7 @@ func (h *Handler) Descriptor() *plugin_pb.JobTypeDescriptor { }, AdminRuntimeDefaults: &plugin_pb.AdminRuntimeDefaults{ Enabled: false, // disabled by default - DetectionIntervalMinutes: 60, // 1 hour + DetectionIntervalMinutes: 60, // 1 hour DetectionTimeoutSeconds: 300, MaxJobsPerDetection: 100, GlobalExecutionConcurrency: 4,