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