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,