From c925869dcc9f9b0c6e7307d90027a99833023ec4 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Fri, 23 Oct 2026 15:21:41 -0700 Subject: [PATCH] shell: retry filer discovery until a filer registers (#11994) * shell: retry filer discovery until a filer registers When -filer is not given, the shell discovers a filer with a single ListClusterNodes call right after connecting to the master. A filer can take several seconds to register after its port reports ready, so the list routinely comes back empty and the shell proceeds with an empty FilerAddress. Commands then fail with "dial tcp: missing address", which is how the s3tables Dremio integration test flaked in CI. Treat an empty list as retryable and keep polling ListClusterNodes until a filer appears or -connectTimeout expires. On timeout, print a clear "no filer registered" notice instead of failing later inside a command; master-only shell commands keep working either way. * shell: report the last filer discovery attempt failure A successful empty ListClusterNodes followed by a master outage left the diagnostic claiming no filer registered, hiding the real RPC failure. Track the outcome of the latest attempt instead: if the final list call failed, report that error; only when the last response was genuinely empty report that no filer registered within the timeout. * shell: test filer discovery retries against a fake master Extract the discovery loop into listFilers and cover the retry behavior with a fake master_pb.SeaweedServer: an empty cluster turning populated, an empty cluster exhausting the window without error, the last attempt's RPC failure being reported, and master-only commands still running after the discovery timeout. --- weed/shell/shell_liner.go | 60 ++++++-- weed/shell/shell_liner_test.go | 241 +++++++++++++++++++++++++++++++++ 2 files changed, 286 insertions(+), 15 deletions(-) diff --git a/weed/shell/shell_liner.go b/weed/shell/shell_liner.go index 6aae477b0f..87d18c9b6c 100644 --- a/weed/shell/shell_liner.go +++ b/weed/shell/shell_liner.go @@ -3,6 +3,7 @@ package shell import ( "bufio" "context" + "errors" "fmt" "io" "math/rand/v2" @@ -25,6 +26,47 @@ import ( var historyPath = path.Join(os.TempDir(), "weed-shell") +// errNoFilersDiscovered marks a successful ListClusterNodes call that came +// back empty, which usually means the filer has not registered with the +// master yet. Unlike an RPC failure it is expected at startup, so filer +// discovery retries on it until the connect timeout expires. +var errNoFilersDiscovered = errors.New("no filer registered with master yet") + +// listFilers polls ListClusterNodes until at least one filer is registered +// with the master or ctx expires. An empty list is expected while a filer is +// still starting, so it retries like a transient failure. On return, a nil +// error with an empty list means the master answered but no filer registered +// in time; a non-nil error is the last attempt's failure. +func listFilers(ctx context.Context, commandEnv *CommandEnv, filerGroup string, maxWait time.Duration) (filers []pb.ServerAddress, listErr error) { + util.RetryWithBackoff(ctx, "filer discovery", maxWait, + func(err error) bool { + return errors.Is(err, errNoFilersDiscovered) || util.IsTransientError(err) + }, func() error { + filers = filers[:0] + listErr = commandEnv.MasterClient.WithClient(ctx, false, func(client master_pb.SeaweedClient) error { + resp, err := client.ListClusterNodes(ctx, &master_pb.ListClusterNodesRequest{ + ClientType: cluster.FilerType, + FilerGroup: filerGroup, + }) + if err != nil { + return err + } + for _, clusterNode := range resp.ClusterNodes { + filers = append(filers, pb.ServerAddress(clusterNode.Address)) + } + return nil + }) + if listErr != nil { + return listErr + } + if len(filers) == 0 { + return errNoFilersDiscovered + } + return nil + }) + return filers, listErr +} + // Piped stdin returns the last command failure; an interactive session shows // the error to the operator and keeps going. func RunShell(options ShellOptions) error { @@ -75,27 +117,15 @@ func RunShell(options ShellOptions) error { } if commandEnv.option.FilerAddress == "" { - var filers []pb.ServerAddress discoverCtx, cancelDiscover := context.WithTimeout(ctx, connectTimeout) defer cancelDiscover() - listErr := commandEnv.MasterClient.WithClient(discoverCtx, false, func(client master_pb.SeaweedClient) error { - resp, err := client.ListClusterNodes(discoverCtx, &master_pb.ListClusterNodesRequest{ - ClientType: cluster.FilerType, - FilerGroup: *options.FilerGroup, - }) - if err != nil { - return err - } - - for _, clusterNode := range resp.ClusterNodes { - filers = append(filers, pb.ServerAddress(clusterNode.Address)) - } - return nil - }) + filers, listErr := listFilers(discoverCtx, commandEnv, *options.FilerGroup, connectTimeout) if len(filers) > 0 { commandEnv.option.FilerAddress = filers[rand.IntN(len(filers))] } else if listErr != nil { fmt.Fprintf(os.Stderr, "cannot list filers from master %s: %v\n", *options.Masters, listErr) + } else { + fmt.Fprintf(os.Stderr, "no filer registered with master %s within %s\n", *options.Masters, connectTimeout) } if options.Debug { if len(filers) > 0 { diff --git a/weed/shell/shell_liner_test.go b/weed/shell/shell_liner_test.go index 21a707ac8a..d8268f887f 100644 --- a/weed/shell/shell_liner_test.go +++ b/weed/shell/shell_liner_test.go @@ -1,11 +1,22 @@ package shell import ( + "context" "errors" + "fmt" "io" + "net" + "os" "strings" + "sync/atomic" "testing" "time" + + "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/credentials/insecure" + "google.golang.org/grpc/status" ) type fakeCommand struct { @@ -83,3 +94,233 @@ func TestProcessEachCmdPreservesRegisteredCommandError(t *testing.T) { t.Errorf("a succeeding command must return no error, got exit=%v err=%v", exit, err) } } + +// fakeMaster implements just enough of master_pb.SeaweedServer for filer +// discovery: KeepConnected holds the stream open so the client marks the +// master connected, and ListClusterNodes answers from a script. +type fakeMaster struct { + master_pb.UnimplementedSeaweedServer + listCalls atomic.Int32 + onList func() (*master_pb.ListClusterNodesResponse, error) +} + +func (f *fakeMaster) KeepConnected(stream grpc.BidiStreamingServer[master_pb.KeepConnectedRequest, master_pb.KeepConnectedResponse]) error { + for { + if _, err := stream.Recv(); err != nil { + return err + } + } +} + +func (f *fakeMaster) ListClusterNodes(context.Context, *master_pb.ListClusterNodesRequest) (*master_pb.ListClusterNodesResponse, error) { + f.listCalls.Add(1) + return f.onList() +} + +func (f *fakeMaster) VolumeList(context.Context, *master_pb.VolumeListRequest) (*master_pb.VolumeListResponse, error) { + return &master_pb.VolumeListResponse{TopologyInfo: &master_pb.TopologyInfo{}}, nil +} + +// startFakeMaster serves the fake master on an ephemeral gRPC port and +// returns the http-form master address (whose gRPC port is port+10000) to +// hand to ShellOptions.Masters. +func startFakeMaster(t *testing.T) (*fakeMaster, string) { + t.Helper() + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("fake master listen: %v", err) + } + grpcPort := listener.Addr().(*net.TCPAddr).Port + if grpcPort < 10001 { + listener.Close() + t.Fatalf("fake master needs a port above 10000 for its http form, got %d", grpcPort) + } + fake := &fakeMaster{} + server := grpc.NewServer() + master_pb.RegisterSeaweedServer(server, fake) + go server.Serve(listener) + t.Cleanup(server.Stop) + return fake, fmt.Sprintf("127.0.0.1:%d", grpcPort-10000) +} + +func newTestCommandEnv(t *testing.T, masters string) *CommandEnv { + t.Helper() + filerGroup := "" + options := &ShellOptions{ + Masters: &masters, + FilerGroup: &filerGroup, + GrpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()), + } + commandEnv := NewCommandEnv(options) + ctx, cancel := context.WithCancel(context.Background()) + go commandEnv.MasterClient.KeepConnectedToMaster(ctx) + t.Cleanup(cancel) + return commandEnv +} + +func filerNode(address string) *master_pb.ListClusterNodesResponse { + return &master_pb.ListClusterNodesResponse{ + ClusterNodes: []*master_pb.ListClusterNodesResponse_ClusterNode{ + {Address: address}, + }, + } +} + +// A filer registering mid-discovery turns an empty list into a populated one +// on a later attempt. +func TestListFilersEmptyThenPopulated(t *testing.T) { + master, masters := startFakeMaster(t) + master.onList = func() (*master_pb.ListClusterNodesResponse, error) { + if master.listCalls.Load() <= 2 { + return &master_pb.ListClusterNodesResponse{}, nil + } + return filerNode("127.0.0.1:9888"), nil + } + + commandEnv := newTestCommandEnv(t, masters) + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + + filers, err := listFilers(ctx, commandEnv, "", 10*time.Second) + if err != nil { + t.Fatalf("listFilers failed: %v", err) + } + if len(filers) != 1 || filers[0] != "127.0.0.1:9888" { + t.Fatalf("expected the late-registered filer, got %v", filers) + } + if got := master.listCalls.Load(); got != 3 { + t.Fatalf("expected 3 discovery attempts, got %d", got) + } +} + +// A master that answers but never lists a filer exhausts the discovery +// window and reports an empty, error-free result — the caller then prints +// the "no filer registered" notice and continues without a filer. +func TestListFilersEmptyUntilTimeout(t *testing.T) { + master, masters := startFakeMaster(t) + master.onList = func() (*master_pb.ListClusterNodesResponse, error) { + return &master_pb.ListClusterNodesResponse{}, nil + } + commandEnv := newTestCommandEnv(t, masters) + + start := time.Now() + ctx, cancel := context.WithTimeout(context.Background(), 1500*time.Millisecond) + defer cancel() + + filers, err := listFilers(ctx, commandEnv, "", 1500*time.Millisecond) + if err != nil { + t.Fatalf("empty discovery must not return an error, got: %v", err) + } + if len(filers) != 0 { + t.Fatalf("expected no filers, got %v", filers) + } + if got := master.listCalls.Load(); got < 2 { + t.Fatalf("expected repeated polling, got %d attempts", got) + } + if elapsed := time.Since(start); elapsed > 10*time.Second { + t.Fatalf("discovery should give up near the timeout, took %v", elapsed) + } +} + +// A non-transient failure on the last attempt surfaces as the discovery +// error, so the diagnostic reports the RPC failure instead of a missing +// filer. +func TestListFilersReportsLastError(t *testing.T) { + master, masters := startFakeMaster(t) + master.onList = func() (*master_pb.ListClusterNodesResponse, error) { + return nil, status.Error(codes.PermissionDenied, "denied") + } + commandEnv := newTestCommandEnv(t, masters) + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + + filers, err := listFilers(ctx, commandEnv, "", 10*time.Second) + if err == nil { + t.Fatalf("expected the permission error to be reported") + } + if len(filers) != 0 { + t.Fatalf("expected no filers, got %v", filers) + } + if got := master.listCalls.Load(); got != 1 { + t.Fatalf("a non-transient error must stop retries, got %d attempts", got) + } +} + +// A transient RPC failure on the last attempt also surfaces as the discovery +// error rather than as a missing filer. +func TestListFilersErrorAfterEmptyList(t *testing.T) { + master, masters := startFakeMaster(t) + master.onList = func() (*master_pb.ListClusterNodesResponse, error) { + if master.listCalls.Load() <= 1 { + return &master_pb.ListClusterNodesResponse{}, nil + } + return nil, status.Error(codes.Unavailable, "master gone") + } + commandEnv := newTestCommandEnv(t, masters) + + ctx, cancel := context.WithTimeout(context.Background(), 1500*time.Millisecond) + defer cancel() + + filers, err := listFilers(ctx, commandEnv, "", 1500*time.Millisecond) + if err == nil { + t.Fatalf("expected the late RPC failure to be reported") + } + if len(filers) != 0 { + t.Fatalf("expected no filers, got %v", filers) + } +} + +// After discovery times out with no filer, master-only commands still run: +// the shell warns about the missing filer but proceeds to execute stdin. +func TestRunShellFilerDiscoveryTimeout(t *testing.T) { + master, masters := startFakeMaster(t) + master.onList = func() (*master_pb.ListClusterNodesResponse, error) { + return &master_pb.ListClusterNodesResponse{}, nil + } + filerGroup := "" + + oldStdin := os.Stdin + inReader, inWriter, err := os.Pipe() + if err != nil { + t.Fatalf("stdin pipe: %v", err) + } + os.Stdin = inReader + t.Cleanup(func() { os.Stdin = oldStdin }) + go func() { + fmt.Fprintln(inWriter, "cluster.ps") + fmt.Fprintln(inWriter, "exit") + inWriter.Close() + }() + + oldStderr := os.Stderr + errReader, errWriter, err := os.Pipe() + if err != nil { + t.Fatalf("stderr pipe: %v", err) + } + os.Stderr = errWriter + t.Cleanup(func() { os.Stderr = oldStderr }) + + done := make(chan error, 1) + go func() { + done <- RunShell(ShellOptions{ + Masters: &masters, + FilerGroup: &filerGroup, + GrpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()), + ConnectTimeout: 1500 * time.Millisecond, + }) + }() + + select { + case err = <-done: + case <-time.After(15 * time.Second): + t.Fatalf("RunShell did not return after discovery timeout") + } + errWriter.Close() + stderr, _ := io.ReadAll(errReader) + if err != nil { + t.Fatalf("master-only commands must run without a filer, got: %v", err) + } + if !strings.Contains(string(stderr), "no filer registered") { + t.Fatalf("expected the missing-filer notice on stderr, got: %q", stderr) + } +}