From 80e0bf97dd8dbdfb1f66aa32d87c44e62bbeb847 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Sun, 1 Nov 2026 18:50:48 -0800 Subject: [PATCH] shell: dial volume servers on their registered gRPC port in volume.scrub, ec.scrub, volume.server.state (#12315) * shell: dial scrubbed volume servers on their registered gRPC port volume.scrub, ec.scrub and volume.server.state built each data node's address from its HTTP address alone, so a volume server whose gRPC port is not the HTTP port + 10000 was dialed on the wrong port. Co-Authored-By: Claude Opus 5.5 * shell: resolve -node volume servers to their registered gRPC port An explicit -node/-nodes list was dialed on HTTP port + 10000 too. Match each entry against the topology and take the data node's gRPC port; an address the master does not know is used as given. Co-Authored-By: Claude Opus 5.5 * shell: keep a gRPC port given in -node A node given as host:port.grpcPort was rewritten to the registered gRPC port; only a bare host:port takes it from the topology. Co-Authored-By: Claude Opus 5.5 * shell: dial -node volume servers when the topology is unavailable An explicit node list did not need the master before; keep using it as given if the volume list cannot be fetched. Selecting every node still requires the topology. Co-Authored-By: Claude Opus 5.5 * shell: bound the topology lookup for an explicit -node list With the master down, volumeServer.state -nodes waited on the master forever instead of reaching the given volume servers. Give the lookup a 30s deadline and fall back to the addresses as given. Co-Authored-By: Claude Opus 5.5 * shell: resolve volume server addresses from a topology-only listing Only the data nodes are needed to find each server's gRPC port, so ask the master for the topology without its volumes and ec shards. A large cluster no longer risks running past the -node lookup deadline and falling back to the HTTP port + 10000. Co-Authored-By: Claude Opus 5.5 * shell: skip the topology lookup when every -node names its gRPC port A host:port.grpcPort entry is dialed as given, so a list made only of those does not need the master and no longer waits on it during an outage. Co-Authored-By: Claude Opus 5.5 --------- Co-authored-by: Claude Opus 5.5 --- weed/shell/command_ec_common.go | 85 ++++++++++++++++++++--- weed/shell/command_ec_scrub.go | 21 +----- weed/shell/command_volume_scrub.go | 15 +--- weed/shell/command_volume_scrub_test.go | 54 ++++++++++++++ weed/shell/command_volume_server_state.go | 16 +---- 5 files changed, 139 insertions(+), 52 deletions(-) create mode 100644 weed/shell/command_volume_scrub_test.go 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 {