From 7b43dec665f18cf29a1718f8576f00f10d3a036a Mon Sep 17 00:00:00 2001 From: Junker der Provinz Date: Sat, 24 Oct 2026 19:29:38 +0200 Subject: [PATCH] shell: run volume.configure.replication in parallel per volume server (#12041) * shell: run volume.configure.replication in parallel per volume server Volume servers were configured one after another, and each server's result overwrote the previous error, so a failure on one server was lost when a later server succeeded. Run one job per volume server, bounded by -maxParallelization like the other shell commands, and return the errors from all servers. VolumeConfigure unmounts and remounts the volume, so replicas of the same volume on different servers take turns instead of all going offline at once. Co-authored-by: NyaMisty <5344431+NyaMisty@users.noreply.github.com> * shell: test that volume.configure.replication overlaps servers up to the limit * shell: assert every server ran in the parallel configure-replication test Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --------- Co-authored-by: NyaMisty <5344431+NyaMisty@users.noreply.github.com> Co-authored-by: chrislusf Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- .../command_volume_configure_replication.go | 72 ++++-- ...mmand_volume_configure_replication_test.go | 226 ++++++++++++++++++ 2 files changed, 277 insertions(+), 21 deletions(-) create mode 100644 weed/shell/command_volume_configure_replication_test.go diff --git a/weed/shell/command_volume_configure_replication.go b/weed/shell/command_volume_configure_replication.go index 8add4aad80..cc85bb5f34 100644 --- a/weed/shell/command_volume_configure_replication.go +++ b/weed/shell/command_volume_configure_replication.go @@ -6,6 +6,7 @@ import ( "flag" "fmt" "io" + "sync" "github.com/seaweedfs/seaweedfs/weed/pb" @@ -33,6 +34,9 @@ func (c *commandVolumeConfigureReplication) Help() string { This command changes a volume replication value. It should be followed by "volume.fix.replication". + Up to -maxParallelization volume servers are configured at a time, one volume at a time on each, + and replicas of the same volume on different servers take turns. + ` } @@ -46,6 +50,7 @@ func (c *commandVolumeConfigureReplication) Do(args []string, commandEnv *Comman volumeIdInt := configureReplicationCommand.Int("volumeId", 0, "the volume id") replicationString := configureReplicationCommand.String("replication", "", "the intended replication value") collectionPattern := configureReplicationCommand.String("collectionPattern", "", "comma-separated collection names, with '*' and '?' wildcards; empty matches all") + maxParallelization := configureReplicationCommand.Int("maxParallelization", DefaultMaxParallelization, "run up to X tasks in parallel, whenever possible") if err = configureReplicationCommand.Parse(args); err != nil { return nil } @@ -75,6 +80,19 @@ func (c *commandVolumeConfigureReplication) Do(args []string, commandEnv *Comman return err } + return configureReplicationOnVolumeServers(commandEnv, topologyInfo, volumeFilter, replicaPlacement.String(), *maxParallelization) +} + +func configureReplicationOnVolumeServers(commandEnv *CommandEnv, topologyInfo *master_pb.TopologyInfo, volumeFilter func(*master_pb.VolumeInformationMessage) bool, replication string, maxParallelization int) error { + type serverVolumes struct { + addr pb.ServerAddress + volumeIds []uint32 + } + var servers []serverVolumes + // VolumeConfigure unmounts and remounts the volume, so replicas take turns + // to keep at least one of them serving. + volumeLocks := make(map[uint32]*sync.Mutex) + // find all data nodes with volumes that needs replication change eachDataNode(topologyInfo, func(dc DataCenterId, rack RackId, dn *master_pb.DataNodeInfo) { var targetVolumeIds []uint32 @@ -82,33 +100,45 @@ func (c *commandVolumeConfigureReplication) Do(args []string, commandEnv *Comman for _, v := range diskInfo.VolumeInfos { if volumeFilter(v) { targetVolumeIds = append(targetVolumeIds, v.Id) + if volumeLocks[v.Id] == nil { + volumeLocks[v.Id] = &sync.Mutex{} + } } } } - if len(targetVolumeIds) == 0 { - return - } - err = operation.WithVolumeServerClient(false, pb.NewServerAddressFromDataNode(dn), commandEnv.option.GrpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error { - for _, targetVolumeId := range targetVolumeIds { - resp, configureErr := volumeServerClient.VolumeConfigure(context.Background(), &volume_server_pb.VolumeConfigureRequest{ - VolumeId: targetVolumeId, - Replication: replicaPlacement.String(), - }) - if configureErr != nil { - return configureErr - } - if resp.Error != "" { - return errors.New(resp.Error) - } - } - return nil - }) - if err != nil { - return + if len(targetVolumeIds) > 0 { + servers = append(servers, serverVolumes{pb.NewServerAddressFromDataNode(dn), targetVolumeIds}) } }) - return err + ewg := NewErrorWaitGroup(maxParallelization) + for _, server := range servers { + ewg.Add(func() error { + err := operation.WithVolumeServerClient(false, server.addr, commandEnv.option.GrpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error { + for _, targetVolumeId := range server.volumeIds { + volumeLocks[targetVolumeId].Lock() + resp, configureErr := volumeServerClient.VolumeConfigure(context.Background(), &volume_server_pb.VolumeConfigureRequest{ + VolumeId: targetVolumeId, + Replication: replication, + }) + volumeLocks[targetVolumeId].Unlock() + if configureErr != nil { + return configureErr + } + if resp.Error != "" { + return errors.New(resp.Error) + } + } + return nil + }) + if err != nil { + return fmt.Errorf("volume server %s: %w", server.addr, err) + } + return nil + }) + } + + return ewg.Wait() } func getVolumeFilter(replicaPlacement *super_block.ReplicaPlacement, volumeId uint32, collectionPattern string) (func(message *master_pb.VolumeInformationMessage) bool, error) { diff --git a/weed/shell/command_volume_configure_replication_test.go b/weed/shell/command_volume_configure_replication_test.go new file mode 100644 index 0000000000..f24e65b835 --- /dev/null +++ b/weed/shell/command_volume_configure_replication_test.go @@ -0,0 +1,226 @@ +package shell + +import ( + "context" + "fmt" + "maps" + "net" + "strings" + "sync" + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" + "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb" + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" +) + +type configureReplicationTestVolumeServer struct { + volume_server_pb.UnimplementedVolumeServerServer + failure string + mu sync.Mutex + configured map[uint32]string +} + +func (s *configureReplicationTestVolumeServer) VolumeConfigure(_ context.Context, req *volume_server_pb.VolumeConfigureRequest) (*volume_server_pb.VolumeConfigureResponse, error) { + s.mu.Lock() + defer s.mu.Unlock() + if s.configured == nil { + s.configured = make(map[uint32]string) + } + s.configured[req.VolumeId] = req.Replication + return &volume_server_pb.VolumeConfigureResponse{Error: s.failure}, nil +} + +// configureReplicationOverlapServer can back several volume servers at once and +// records whether two calls for the same volume were ever in flight together. +type configureReplicationOverlapServer struct { + volume_server_pb.UnimplementedVolumeServerServer + mu sync.Mutex + inFlight map[uint32]int + overlapped bool +} + +func (s *configureReplicationOverlapServer) VolumeConfigure(_ context.Context, req *volume_server_pb.VolumeConfigureRequest) (*volume_server_pb.VolumeConfigureResponse, error) { + s.mu.Lock() + s.inFlight[req.VolumeId]++ + if s.inFlight[req.VolumeId] > 1 { + s.overlapped = true + } + s.mu.Unlock() + + time.Sleep(20 * time.Millisecond) + + s.mu.Lock() + s.inFlight[req.VolumeId]-- + s.mu.Unlock() + return &volume_server_pb.VolumeConfigureResponse{}, nil +} + +func startConfigureReplicationTestVolumeServer(t *testing.T, server volume_server_pb.VolumeServerServer, volumeIds ...uint32) *master_pb.DataNodeInfo { + t.Helper() + lis, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + srv := grpc.NewServer() + volume_server_pb.RegisterVolumeServerServer(srv, server) + go srv.Serve(lis) + t.Cleanup(srv.Stop) + + port := lis.Addr().(*net.TCPAddr).Port + disk := &master_pb.DiskInfo{} + for _, vid := range volumeIds { + disk.VolumeInfos = append(disk.VolumeInfos, &master_pb.VolumeInformationMessage{Id: vid}) + } + return &master_pb.DataNodeInfo{ + Id: fmt.Sprintf("127.0.0.1:%d", port), + GrpcPort: uint32(port), + DiskInfos: map[string]*master_pb.DiskInfo{"": disk}, + } +} + +func configureReplicationTestEnv(dataNodes ...*master_pb.DataNodeInfo) (*CommandEnv, *master_pb.TopologyInfo) { + commandEnv := &CommandEnv{option: &ShellOptions{ + GrpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()), + }} + topologyInfo := &master_pb.TopologyInfo{DataCenterInfos: []*master_pb.DataCenterInfo{{ + RackInfos: []*master_pb.RackInfo{{DataNodeInfos: dataNodes}}, + }}} + return commandEnv, topologyInfo +} + +func allVolumes(*master_pb.VolumeInformationMessage) bool { return true } + +func TestConfigureReplicationOnVolumeServersReportsEveryFailingServer(t *testing.T) { + for _, maxParallelization := range []int{1, DefaultMaxParallelization} { + t.Run(fmt.Sprintf("maxParallelization=%d", maxParallelization), func(t *testing.T) { + first := &configureReplicationTestVolumeServer{failure: "first server failed"} + healthy := &configureReplicationTestVolumeServer{} + last := &configureReplicationTestVolumeServer{failure: "last server failed"} + commandEnv, topologyInfo := configureReplicationTestEnv( + startConfigureReplicationTestVolumeServer(t, first, 1), + startConfigureReplicationTestVolumeServer(t, healthy, 2, 3), + startConfigureReplicationTestVolumeServer(t, last, 4), + ) + + err := configureReplicationOnVolumeServers(commandEnv, topologyInfo, allVolumes, "001", maxParallelization) + + if err == nil { + t.Fatal("expected an error") + } + for _, failure := range []string{first.failure, last.failure} { + if !strings.Contains(err.Error(), failure) { + t.Errorf("error %q does not mention %q", err, failure) + } + } + healthy.mu.Lock() + defer healthy.mu.Unlock() + if expected := map[uint32]string{2: "001", 3: "001"}; !maps.Equal(healthy.configured, expected) { + t.Errorf("healthy server configured %v, expected %v", healthy.configured, expected) + } + }) + } +} + +func TestConfigureReplicationOnVolumeServersConfiguresReplicasOneAtATime(t *testing.T) { + server := &configureReplicationOverlapServer{inFlight: make(map[uint32]int)} + commandEnv, topologyInfo := configureReplicationTestEnv( + startConfigureReplicationTestVolumeServer(t, server, 1, 2), + startConfigureReplicationTestVolumeServer(t, server, 1, 2), + startConfigureReplicationTestVolumeServer(t, server, 1, 2), + ) + + if err := configureReplicationOnVolumeServers(commandEnv, topologyInfo, allVolumes, "002", DefaultMaxParallelization); err != nil { + t.Fatal(err) + } + + server.mu.Lock() + defer server.mu.Unlock() + if server.overlapped { + t.Error("replicas of one volume were configured at the same time") + } +} + +// configureReplicationGateServer blocks inside VolumeConfigure until the test +// releases it, so the test can observe how many servers run at once. +type configureReplicationGateServer struct { + volume_server_pb.UnimplementedVolumeServerServer + gate *configureReplicationGate +} + +type configureReplicationGate struct { + entered chan struct{} + release chan struct{} + mu sync.Mutex + active int + maxActive int + calls int +} + +func (s *configureReplicationGateServer) VolumeConfigure(_ context.Context, _ *volume_server_pb.VolumeConfigureRequest) (*volume_server_pb.VolumeConfigureResponse, error) { + s.gate.mu.Lock() + s.gate.calls++ + s.gate.active++ + if s.gate.active > s.gate.maxActive { + s.gate.maxActive = s.gate.active + } + s.gate.mu.Unlock() + + s.gate.entered <- struct{}{} + <-s.gate.release + + s.gate.mu.Lock() + s.gate.active-- + s.gate.mu.Unlock() + return &volume_server_pb.VolumeConfigureResponse{}, nil +} + +func TestConfigureReplicationOnVolumeServersRunsServersInParallelUpToTheLimit(t *testing.T) { + const servers = 4 + const maxParallelization = 2 + gate := &configureReplicationGate{ + entered: make(chan struct{}, servers), + release: make(chan struct{}), + } + var dataNodes []*master_pb.DataNodeInfo + for i := 0; i < servers; i++ { + dataNodes = append(dataNodes, startConfigureReplicationTestVolumeServer(t, + &configureReplicationGateServer{gate: gate}, uint32(i+1))) + } + commandEnv, topologyInfo := configureReplicationTestEnv(dataNodes...) + + done := make(chan error, 1) + go func() { + done <- configureReplicationOnVolumeServers(commandEnv, topologyInfo, allVolumes, "001", maxParallelization) + }() + + for i := 0; i < maxParallelization; i++ { + select { + case <-gate.entered: + case <-time.After(10 * time.Second): + t.Fatalf("only %d servers started; serial execution would hold the second call", i) + } + } + + select { + case <-gate.entered: + t.Fatalf("more than %d servers ran while calls were held open", maxParallelization) + case <-time.After(100 * time.Millisecond): + } + + close(gate.release) + if err := <-done; err != nil { + t.Fatal(err) + } + + gate.mu.Lock() + defer gate.mu.Unlock() + if gate.maxActive != maxParallelization { + t.Errorf("at most %d servers ran together, expected %d", gate.maxActive, maxParallelization) + } + if gate.calls != servers { + t.Errorf("%d servers ran, expected %d", gate.calls, servers) + } +}