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) + } +}