diff --git a/weed/shell/command_ec_common.go b/weed/shell/command_ec_common.go index ea7a9fd9e2..bf49b02b7c 100644 --- a/weed/shell/command_ec_common.go +++ b/weed/shell/command_ec_common.go @@ -93,10 +93,13 @@ func collectTopologyInfo(commandEnv *CommandEnv, delayBeforeCollecting time.Dura if delayBeforeCollecting > 0 { time.Sleep(delayBeforeCollecting) } + return fetchTopologyInfo(context.Background(), commandEnv, &master_pb.VolumeListRequest{}) +} +func fetchTopologyInfo(ctx context.Context, commandEnv *CommandEnv, request *master_pb.VolumeListRequest) (topoInfo *master_pb.TopologyInfo, volumeSizeLimitMb uint64, err error) { var resp *master_pb.VolumeListResponse - err = commandEnv.MasterClient.WithClient(context.Background(), false, func(client master_pb.SeaweedClient) error { - resp, err = pb.CollectVolumeList(context.Background(), client, &master_pb.VolumeListRequest{}) + err = commandEnv.MasterClient.WithClient(ctx, false, func(client master_pb.SeaweedClient) error { + resp, err = pb.CollectVolumeList(ctx, client, request) return err }) if err != nil { @@ -107,22 +110,88 @@ func collectTopologyInfo(commandEnv *CommandEnv, delayBeforeCollecting time.Dura } func collectDataNodes(commandEnv *CommandEnv, delayBeforeCollecting time.Duration) ([]*master_pb.DataNodeInfo, error) { - dataNodes := []*master_pb.DataNodeInfo{} - topo, _, err := collectTopologyInfo(commandEnv, delayBeforeCollecting) if err != nil { return nil, err } + return topologyDataNodes(topo), nil +} +func topologyDataNodes(topo *master_pb.TopologyInfo) []*master_pb.DataNodeInfo { + dataNodes := []*master_pb.DataNodeInfo{} for _, dci := range topo.GetDataCenterInfos() { for _, r := range dci.GetRackInfos() { - for _, dn := range r.GetDataNodeInfos() { - dataNodes = append(dataNodes, dn) - } + dataNodes = append(dataNodes, r.GetDataNodeInfos()...) } } + return dataNodes +} - return dataNodes, nil +// collectVolumeServerAddresses resolves nodes with volumeServerAddresses from +// a topology-only listing. An explicit node list bounds the lookup and is used +// as given when the topology is unavailable, so it still works while the +// master is down. +func collectVolumeServerAddresses(commandEnv *CommandEnv, nodes string) ([]pb.ServerAddress, error) { + if nodes != "" && everyNodeNamesGrpcPort(nodes) { + return volumeServerAddresses(nil, nodes), nil + } + ctx := context.Background() + if nodes != "" { + var cancel context.CancelFunc + ctx, cancel = context.WithTimeout(ctx, 30*time.Second) + defer cancel() + } + topo, _, err := fetchTopologyInfo(ctx, commandEnv, &master_pb.VolumeListRequest{TopologyOnly: true}) + if err != nil { + if nodes == "" { + return nil, err + } + glog.V(0).Infof("dialing volume servers as given, topology unavailable: %v", err) + } + return volumeServerAddresses(topologyDataNodes(topo), nodes), nil +} + +func everyNodeNamesGrpcPort(nodes string) bool { + for _, node := range strings.Split(nodes, ",") { + node = strings.TrimSpace(node) + if node == "" { + continue + } + portsSepIndex := strings.LastIndex(node, ":") + if portsSepIndex < 0 || !strings.Contains(node[portsSepIndex+1:], ".") { + return false + } + } + return true +} + +// volumeServerAddresses returns the comma-separated nodes, or every data node +// when nodes is empty, dialing each registered server on its gRPC port unless +// the node names one. +func volumeServerAddresses(dataNodes []*master_pb.DataNodeInfo, nodes string) []pb.ServerAddress { + registered := make(map[string]pb.ServerAddress, len(dataNodes)) + all := make([]pb.ServerAddress, 0, len(dataNodes)) + for _, dn := range dataNodes { + addr := pb.NewServerAddressFromDataNode(dn) + registered[addr.ToHttpAddress()] = addr + all = append(all, addr) + } + if nodes == "" { + return all + } + addrs := []pb.ServerAddress{} + for _, node := range strings.Split(nodes, ",") { + node = strings.TrimSpace(node) + if node == "" { + continue + } + if addr, found := registered[node]; found { + addrs = append(addrs, addr) + } else { + addrs = append(addrs, pb.ServerAddress(node)) + } + } + return addrs } func collectEcNodesForDC(commandEnv *CommandEnv, selectedDataCenter string, diskType types.DiskType) (ecNodes []*EcNode, totalFreeEcSlots int, err error) { diff --git a/weed/shell/command_ec_scrub.go b/weed/shell/command_ec_scrub.go index 13414fba3e..00a1fa072d 100644 --- a/weed/shell/command_ec_scrub.go +++ b/weed/shell/command_ec_scrub.go @@ -8,7 +8,6 @@ import ( "strings" "github.com/seaweedfs/seaweedfs/weed/ec" - "github.com/seaweedfs/seaweedfs/weed/pb" "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb" ) @@ -53,23 +52,9 @@ func (c *commandEcVolumeScrub) Do(args []string, commandEnv *CommandEnv, writer return } - volumeServerAddrs := []pb.ServerAddress{} - if *nodesStr != "" { - for _, addr := range strings.Split(*nodesStr, ",") { - addr = strings.TrimSpace(addr) - if addr == "" { - continue - } - volumeServerAddrs = append(volumeServerAddrs, pb.ServerAddress(addr)) - } - } else { - dns, err := collectDataNodes(commandEnv, 0) - if err != nil { - return err - } - for _, dn := range dns { - volumeServerAddrs = append(volumeServerAddrs, pb.ServerAddress(dn.Address)) - } + volumeServerAddrs, err := collectVolumeServerAddresses(commandEnv, *nodesStr) + if err != nil { + return err } volumeIDs := []uint32{} diff --git a/weed/shell/command_volume_scrub.go b/weed/shell/command_volume_scrub.go index 95afd3a329..0671075eea 100644 --- a/weed/shell/command_volume_scrub.go +++ b/weed/shell/command_volume_scrub.go @@ -62,19 +62,8 @@ func (c *commandVolumeScrub) Do(args []string, commandEnv *CommandEnv, writer io return } - c.volumeServerAddrs = []pb.ServerAddress{} - if *nodesStr != "" { - for _, addr := range strings.Split(*nodesStr, ",") { - c.volumeServerAddrs = append(c.volumeServerAddrs, pb.ServerAddress(addr)) - } - } else { - dns, err := collectDataNodes(commandEnv, 0) - if err != nil { - return err - } - for _, dn := range dns { - c.volumeServerAddrs = append(c.volumeServerAddrs, pb.ServerAddress(dn.Address)) - } + if c.volumeServerAddrs, err = collectVolumeServerAddresses(commandEnv, *nodesStr); err != nil { + return err } c.volumeIDs = []uint32{} diff --git a/weed/shell/command_volume_scrub_test.go b/weed/shell/command_volume_scrub_test.go new file mode 100644 index 0000000000..8a2d29d867 --- /dev/null +++ b/weed/shell/command_volume_scrub_test.go @@ -0,0 +1,54 @@ +package shell + +import ( + "reflect" + "testing" + + "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" +) + +func TestVolumeServerAddressesUseRegisteredGrpcPort(t *testing.T) { + dataNodes := []*master_pb.DataNodeInfo{ + {Id: "10.0.0.1:8080", Address: "10.0.0.1:8080", GrpcPort: 18080}, + {Id: "10.0.0.2:7000", Address: "10.0.0.2:7000", GrpcPort: 7500}, + {Id: "10.0.0.3:8081"}, + } + tests := []struct { + nodes string + wantGrpc []string + }{ + {"", []string{"10.0.0.1:18080", "10.0.0.2:7500", "10.0.0.3:18081"}}, + {"10.0.0.2:7000", []string{"10.0.0.2:7500"}}, + {" 10.0.0.2:7000 ,,10.0.0.1:8080", []string{"10.0.0.2:7500", "10.0.0.1:18080"}}, + {"10.0.0.9:7000.7600", []string{"10.0.0.9:7600"}}, + {"10.0.0.2:7000.7600", []string{"10.0.0.2:7600"}}, + } + for _, tt := range tests { + var gotGrpc []string + for _, addr := range volumeServerAddresses(dataNodes, tt.nodes) { + gotGrpc = append(gotGrpc, addr.ToGrpcAddress()) + } + if !reflect.DeepEqual(gotGrpc, tt.wantGrpc) { + t.Errorf("volumeServerAddresses(%q) dials %v, want %v", tt.nodes, gotGrpc, tt.wantGrpc) + } + } +} + +func TestEveryNodeNamesGrpcPort(t *testing.T) { + tests := []struct { + nodes string + want bool + }{ + {"10.0.0.9:7000.7600", true}, + {" 10.0.0.9:7000.7600 ,,[::1]:8080.18080", true}, + {"10.0.0.9:7000.7600,10.0.0.2:7000", false}, + {"10.0.0.2:7000", false}, + {"volume.example.com", false}, + {"volume.example.com:8080", false}, + } + for _, tt := range tests { + if got := everyNodeNamesGrpcPort(tt.nodes); got != tt.want { + t.Errorf("everyNodeNamesGrpcPort(%q) = %v, want %v", tt.nodes, got, tt.want) + } + } +} diff --git a/weed/shell/command_volume_server_state.go b/weed/shell/command_volume_server_state.go index 9ab850f568..03fd5814f0 100644 --- a/weed/shell/command_volume_server_state.go +++ b/weed/shell/command_volume_server_state.go @@ -86,19 +86,9 @@ func (c *commandVolumeServerState) Do(args []string, commandEnv *CommandEnv, wri return fmt.Errorf("--maintenanceOn and --maintenanceOff are mutually exclusive") } - volumeServerAddrs := []pb.ServerAddress{} - if *nodesStr != "" { - for _, addr := range strings.Split(*nodesStr, ",") { - volumeServerAddrs = append(volumeServerAddrs, pb.ServerAddress(addr)) - } - } else { - dns, err := collectDataNodes(commandEnv, 0) - if err != nil { - return err - } - for _, dn := range dns { - volumeServerAddrs = append(volumeServerAddrs, pb.ServerAddress(dn.Address)) - } + volumeServerAddrs, err := collectVolumeServerAddresses(commandEnv, *nodesStr) + if err != nil { + return err } slices.Sort(volumeServerAddrs) if len(volumeServerAddrs) == 0 {