mirrors/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2026-08-16 04:06:44 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: throughput limits for replicate, EC shard, and worker-driven moves (#10749)

Browse Source 
* 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.
This commit is contained in:
Chris Lu
2026-08-13 13:22:58 -07:00
committed by GitHub
parent 7d0fff32db
commit 4f50c5b0d4
32 changed files with 312 additions and 144 deletions
Download Patch File Download Diff File Expand all files Collapse all files
weed/operation/volume_move/ec_move.go
+15 -11
View File
@@ -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)
weed/operation/volume_move/ec_move_test.go
+3 -3
View File
@@ -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)
}
weed/operation/volume_move/volume_move.go
+7 -5
View File
@@ -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
weed/operation/volume_move/volume_move_test.go
+14 -2
View File
@@ -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()
weed/pb/volume_server.proto
+1
View File
@@ -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 {
}
weed/pb/volume_server_pb/volume_server.pb.go
+24 -14
View File
@@ -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" +
weed/pb/volume_server_pb/volume_server_grpc.pb.go
+51 -51
View File
@@ -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.
weed/pb/worker.proto
+4
View File
@@ -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 ==========
weed/pb/worker_pb/worker.pb.go
+47 -11
View File
@@ -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" +
weed/pb/worker_pb/worker_grpc.pb.go
+4 -4
View File
@@ -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.
weed/server/volume_grpc_erasure_coding.go
+13 -5
View File
@@ -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)
}
}
weed/shell/command_ec_balance.go
+2 -1
View File
@@ -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)
}
weed/shell/command_ec_common.go
+5 -3
View File
@@ -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
weed/shell/command_ec_encode.go
+1 -1
View File
@@ -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 {
weed/shell/command_fs_distribute_chunks.go
+1 -2
View File
@@ -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)
weed/shell/command_s3_lifecycle_run_shard.go
+1 -1
View File
@@ -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)),
weed/shell/command_volume_fix_replication.go
+1 -1
View File
@@ -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
weed/shell/command_volume_move.go
+2 -2
View File
@@ -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.
weed/shell/command_volume_tier_move.go
+3 -3
View File
@@ -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)
}
weed/shell/command_volume_tier_upload.go
+1 -1
View File
@@ -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)
}
weed/worker/tasks/balance/balance_task.go
+2 -1
View File
@@ -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)
weed/worker/tasks/balance/config.go
+18
View File
@@ -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
weed/worker/tasks/balance/detection.go
+5 -4
View File
@@ -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,
},
},
}
weed/worker/tasks/balance/plugin_handler.go
+36 -5
View File
@@ -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
weed/worker/tasks/balance/replica_placement_test.go
+1 -1
View File
@@ -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")
}
weed/worker/tasks/ec_balance/config.go
+19 -1
View File
@@ -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
weed/worker/tasks/ec_balance/detection.go
+4 -3
View File
@@ -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,
},
},
},
weed/worker/tasks/ec_balance/ec_balance_task.go
+2 -1
View File
@@ -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
weed/worker/tasks/ec_balance/plugin_handler.go
+19 -1
View File
@@ -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
weed/worker/tasks/erasure_coding/ec_task_encode_files_e2e_test.go
+2 -2
View File
@@ -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{
weed/worker/tasks/erasure_coding/ec_task_julor_layout_e2e_test.go
+3 -3
View File
@@ -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{
weed/worker/tasks/iceberg/handler.go
+1 -1
View File
@@ -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,