From 6a34d82f05aa8d22eb00d24542ac4f9dae8e00e4 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Fri, 23 Oct 2026 22:52:51 -0700 Subject: [PATCH] shell: retry filer discovery until master knows a filer (#11989) * shell: discover the filer on first use instead of at startup The startup retry loop made every weed shell invocation wait out the connect timeout on clusters that have no filer, even for commands that only talk to the master. Discovery now runs once, on the first command that actually needs a filer, keeping the retry-until-registered fix for the fresh-cluster race without stalling master-only commands. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * shell: stop filer discovery retries on non-transient errors A master that keeps answering ListClusterNodes with a permanent error (permission denied, unimplemented) cannot gain a filer by retrying, so discovery returns instead of polling until the deadline. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * shell: explain an empty filer list when discovery times out * shell: retry filer discovery while the address is still empty Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --------- Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- weed/shell/command_fs_configure.go | 2 +- weed/shell/command_fs_meta_save.go | 4 +- weed/shell/command_fs_verify.go | 2 +- weed/shell/command_mq_topic_compact.go | 2 +- weed/shell/command_remote_meta_sync.go | 2 +- weed/shell/command_remote_mount.go | 4 +- weed/shell/command_remote_mount_buckets.go | 2 +- weed/shell/command_remote_uncache.go | 2 +- weed/shell/command_remote_unmount.go | 4 +- weed/shell/command_s3_bucket_quota_check.go | 2 +- weed/shell/command_s3_clean_uploads.go | 2 +- weed/shell/command_s3_iam_client.go | 2 +- weed/shell/command_volume_fsck.go | 2 +- weed/shell/commands.go | 71 +++++- weed/shell/commands_test.go | 159 +++++++++++++ weed/shell/s3tables_helpers.go | 2 +- weed/shell/shell_liner.go | 66 ------ weed/shell/shell_liner_test.go | 241 -------------------- 18 files changed, 246 insertions(+), 325 deletions(-) create mode 100644 weed/shell/commands_test.go diff --git a/weed/shell/command_fs_configure.go b/weed/shell/command_fs_configure.go index 4bc29e3984..165e0b8e65 100644 --- a/weed/shell/command_fs_configure.go +++ b/weed/shell/command_fs_configure.go @@ -83,7 +83,7 @@ func (c *commandFsConfigure) Do(args []string, commandEnv *CommandEnv, writer io return nil } - fc, err := filer.ReadFilerConf(commandEnv.option.FilerAddress, commandEnv.option.GrpcDialOption, commandEnv.MasterClient) + fc, err := filer.ReadFilerConf(commandEnv.FilerAddress(), commandEnv.option.GrpcDialOption, commandEnv.MasterClient) if err != nil { return err } diff --git a/weed/shell/command_fs_meta_save.go b/weed/shell/command_fs_meta_save.go index 4034655350..c52cc421dc 100644 --- a/weed/shell/command_fs_meta_save.go +++ b/weed/shell/command_fs_meta_save.go @@ -75,7 +75,7 @@ func (c *commandFsMetaSave) Do(args []string, commandEnv *CommandEnv, writer io. if fileName == "" { t := time.Now() fileName = fmt.Sprintf("%s-%4d%02d%02d-%02d%02d%02d.meta.gz", - commandEnv.option.FilerAddress.ToHttpAddress(), t.Year(), t.Month(), t.Day(), t.Hour(), t.Minute(), t.Second()) + commandEnv.FilerAddress().ToHttpAddress(), t.Year(), t.Month(), t.Day(), t.Hour(), t.Minute(), t.Second()) } var dst io.Writer @@ -143,7 +143,7 @@ func (c *commandFsMetaSave) Do(args []string, commandEnv *CommandEnv, writer io. }) if err == nil { - fmt.Fprintf(writer, "meta data for http://%s%s is saved to %s\n", commandEnv.option.FilerAddress.ToHttpAddress(), path, fileName) + fmt.Fprintf(writer, "meta data for http://%s%s is saved to %s\n", commandEnv.FilerAddress().ToHttpAddress(), path, fileName) } return err diff --git a/weed/shell/command_fs_verify.go b/weed/shell/command_fs_verify.go index 32128b4b6f..24007793f0 100644 --- a/weed/shell/command_fs_verify.go +++ b/weed/shell/command_fs_verify.go @@ -254,7 +254,7 @@ func (c *commandFsVerify) verifyProcessMetadata(path string, wg *sync.WaitGroup) StopTsNs: time.Now().UnixNano(), EventErrorType: pb.DontLogError, } - err = pb.FollowMetadata(c.env.option.FilerAddress, c.env.option.GrpcDialOption, metadataFollowOption, processEventFn) + err = pb.FollowMetadata(c.env.FilerAddress(), c.env.option.GrpcDialOption, metadataFollowOption, processEventFn) return fileCount, errCount, prunedCount, err } diff --git a/weed/shell/command_mq_topic_compact.go b/weed/shell/command_mq_topic_compact.go index ab3552022c..8171144b1f 100644 --- a/weed/shell/command_mq_topic_compact.go +++ b/weed/shell/command_mq_topic_compact.go @@ -73,7 +73,7 @@ func (c *commandMqTopicCompact) Do(args []string, commandEnv *CommandEnv, writer // read topic configuration fca := filer_client.NewFilerClientAccessor( - []pb.ServerAddress{commandEnv.option.FilerAddress}, + []pb.ServerAddress{commandEnv.FilerAddress()}, commandEnv.option.GrpcDialOption, ) t := topic.NewTopic(*namespace, *topicName) diff --git a/weed/shell/command_remote_meta_sync.go b/weed/shell/command_remote_meta_sync.go index b1ce83140b..ca9320d3f5 100644 --- a/weed/shell/command_remote_meta_sync.go +++ b/weed/shell/command_remote_meta_sync.go @@ -84,7 +84,7 @@ func (c *commandRemoteMetaSync) Do(args []string, commandEnv *CommandEnv, writer } func detectMountInfo(commandEnv *CommandEnv, writer io.Writer, dir string) (*remote_pb.RemoteStorageMapping, string, *remote_pb.RemoteStorageLocation, *remote_pb.RemoteConf, error) { - return filer.DetectMountInfo(commandEnv.option.GrpcDialOption, commandEnv.option.FilerAddress, dir) + return filer.DetectMountInfo(commandEnv.option.GrpcDialOption, commandEnv.FilerAddress(), dir) } /* diff --git a/weed/shell/command_remote_mount.go b/weed/shell/command_remote_mount.go index 1eeb43bf53..debfa27a63 100644 --- a/weed/shell/command_remote_mount.go +++ b/weed/shell/command_remote_mount.go @@ -85,7 +85,7 @@ func (c *commandRemoteMount) Do(args []string, commandEnv *CommandEnv, writer io } // find configuration for remote storage - remoteConf, err := filer.ReadRemoteStorageConf(commandEnv.option.GrpcDialOption, commandEnv.option.FilerAddress, remote_storage.ParseLocationName(*remote)) + remoteConf, err := filer.ReadRemoteStorageConf(commandEnv.option.GrpcDialOption, commandEnv.FilerAddress(), remote_storage.ParseLocationName(*remote)) if err != nil { return fmt.Errorf("find configuration for %s: %v", *remote, err) } @@ -133,7 +133,7 @@ func (c *commandRemoteMount) Do(args []string, commandEnv *CommandEnv, writer io func listExistingRemoteStorageMounts(commandEnv *CommandEnv, writer io.Writer) (mappings *remote_pb.RemoteStorageMapping, err error) { // read current mapping - mappings, err = filer.ReadMountMappings(commandEnv.option.GrpcDialOption, commandEnv.option.FilerAddress) + mappings, err = filer.ReadMountMappings(commandEnv.option.GrpcDialOption, commandEnv.FilerAddress()) if err != nil { return mappings, err } diff --git a/weed/shell/command_remote_mount_buckets.go b/weed/shell/command_remote_mount_buckets.go index 1fa464141c..c011a15e57 100644 --- a/weed/shell/command_remote_mount_buckets.go +++ b/weed/shell/command_remote_mount_buckets.go @@ -62,7 +62,7 @@ func (c *commandRemoteMountBuckets) Do(args []string, commandEnv *CommandEnv, wr } // find configuration for remote storage - remoteConf, err := filer.ReadRemoteStorageConf(commandEnv.option.GrpcDialOption, commandEnv.option.FilerAddress, *remote) + remoteConf, err := filer.ReadRemoteStorageConf(commandEnv.option.GrpcDialOption, commandEnv.FilerAddress(), *remote) if err != nil { return fmt.Errorf("find configuration for %s: %v", *remote, err) } diff --git a/weed/shell/command_remote_uncache.go b/weed/shell/command_remote_uncache.go index e7281c791f..20383a8f40 100644 --- a/weed/shell/command_remote_uncache.go +++ b/weed/shell/command_remote_uncache.go @@ -57,7 +57,7 @@ func (c *commandRemoteUncache) Do(args []string, commandEnv *CommandEnv, writer return nil } - mappings, listErr := filer.ReadMountMappings(commandEnv.option.GrpcDialOption, commandEnv.option.FilerAddress) + mappings, listErr := filer.ReadMountMappings(commandEnv.option.GrpcDialOption, commandEnv.FilerAddress()) if listErr != nil { return listErr } diff --git a/weed/shell/command_remote_unmount.go b/weed/shell/command_remote_unmount.go index 030dd9021a..fd905d00ee 100644 --- a/weed/shell/command_remote_unmount.go +++ b/weed/shell/command_remote_unmount.go @@ -52,7 +52,7 @@ func (c *commandRemoteUnmount) Do(args []string, commandEnv *CommandEnv, writer return nil } - mappings, listErr := filer.ReadMountMappings(commandEnv.option.GrpcDialOption, commandEnv.option.FilerAddress) + mappings, listErr := filer.ReadMountMappings(commandEnv.option.GrpcDialOption, commandEnv.FilerAddress()) if listErr != nil { return listErr } @@ -79,7 +79,7 @@ func (c *commandRemoteUnmount) Do(args []string, commandEnv *CommandEnv, writer } // reset remote sync offset in case the folder is mounted again - if err = remote_storage.SetSyncOffset(commandEnv.option.GrpcDialOption, commandEnv.option.FilerAddress, *dir, time.Now().UnixNano()); err != nil { + if err = remote_storage.SetSyncOffset(commandEnv.option.GrpcDialOption, commandEnv.FilerAddress(), *dir, time.Now().UnixNano()); err != nil { return fmt.Errorf("reset remote.sync offset for %s: %v", *dir, err) } diff --git a/weed/shell/command_s3_bucket_quota_check.go b/weed/shell/command_s3_bucket_quota_check.go index 7313514ec2..eda0c42842 100644 --- a/weed/shell/command_s3_bucket_quota_check.go +++ b/weed/shell/command_s3_bucket_quota_check.go @@ -60,7 +60,7 @@ func (c *commandS3BucketQuotaEnforce) Do(args []string, commandEnv *CommandEnv, } // read existing filer configuration - fc, err := filer.ReadFilerConf(commandEnv.option.FilerAddress, commandEnv.option.GrpcDialOption, commandEnv.MasterClient) + fc, err := filer.ReadFilerConf(commandEnv.FilerAddress(), commandEnv.option.GrpcDialOption, commandEnv.MasterClient) if err != nil { return err } diff --git a/weed/shell/command_s3_clean_uploads.go b/weed/shell/command_s3_clean_uploads.go index 1da7a02d40..1d7bf986c0 100644 --- a/weed/shell/command_s3_clean_uploads.go +++ b/weed/shell/command_s3_clean_uploads.go @@ -106,7 +106,7 @@ func (c *commandS3CleanUploads) cleanupUploads(commandEnv *CommandEnv, writer io fmt.Fprintf(writer, "skip %s: %v\n", staleUpload.Name, checkErr) continue } - deleteUrl := fmt.Sprintf("http://%s%s/%s?recursive=true&ignoreRecursiveError=true", commandEnv.option.FilerAddress.ToHttpAddress(), uploadsDir, staleUpload.Name) + deleteUrl := fmt.Sprintf("http://%s%s/%s?recursive=true&ignoreRecursiveError=true", commandEnv.FilerAddress().ToHttpAddress(), uploadsDir, staleUpload.Name) if completed { deleteUrl += "&skipChunkDeletion=true" } diff --git a/weed/shell/command_s3_iam_client.go b/weed/shell/command_s3_iam_client.go index eb619100a1..832471d0bf 100644 --- a/weed/shell/command_s3_iam_client.go +++ b/weed/shell/command_s3_iam_client.go @@ -28,7 +28,7 @@ func (ce *CommandEnv) withIamClient(fn func(ctx context.Context, client iam_pb.S ctx, cancel := context.WithTimeout(iamAdminAuthContext(context.Background()), iamRequestTimeout) defer cancel() return fn(ctx, iam_pb.NewSeaweedIdentityAccessManagementClient(conn)) - }, ce.option.FilerAddress.ToGrpcAddress(), false, ce.option.GrpcDialOption) + }, ce.FilerAddress().ToGrpcAddress(), false, ce.option.GrpcDialOption) } func iamAdminAuthContext(ctx context.Context) context.Context { diff --git a/weed/shell/command_volume_fsck.go b/weed/shell/command_volume_fsck.go index e3c301992d..ef6b61bfab 100644 --- a/weed/shell/command_volume_fsck.go +++ b/weed/shell/command_volume_fsck.go @@ -790,7 +790,7 @@ func (c *commandVolumeFsck) httpDelete(path util.FullPath) { req.URL = &url.URL{ Scheme: "http", - Host: c.env.option.FilerAddress.ToHttpAddress(), + Host: c.env.FilerAddress().ToHttpAddress(), Path: string(path), } diff --git a/weed/shell/commands.go b/weed/shell/commands.go index 3bbb363b18..43b681d4d7 100644 --- a/weed/shell/commands.go +++ b/weed/shell/commands.go @@ -4,7 +4,10 @@ import ( "context" "fmt" "io" + "math/rand/v2" + "os" "strings" + "sync" "time" "github.com/seaweedfs/seaweedfs/weed/cluster" @@ -43,6 +46,7 @@ type CommandEnv struct { noLock bool forceNoLock bool verbose bool + filerMu sync.Mutex } func NewCommandEnv(options *ShellOptions) *CommandEnv { @@ -147,10 +151,75 @@ var _ = filer_pb.FilerClient(&CommandEnv{}) func (ce *CommandEnv) WithFilerClient(streamingMode bool, fn func(filer_pb.SeaweedFilerClient) error) error { - return pb.WithGrpcFilerClient(streamingMode, 0, ce.option.FilerAddress, ce.option.GrpcDialOption, fn) + return pb.WithGrpcFilerClient(streamingMode, 0, ce.FilerAddress(), ce.option.GrpcDialOption, fn) } +// FilerAddress returns the discovered or configured filer, running discovery +// on first use so commands that never need a filer skip the wait entirely. +// A failed or empty discovery is not cached: each later call retries until a +// filer registers, so a long-lived shell can still find one. +func (ce *CommandEnv) FilerAddress() pb.ServerAddress { + ce.filerMu.Lock() + defer ce.filerMu.Unlock() + ce.discoverFiler() + return ce.option.FilerAddress +} + +func (ce *CommandEnv) discoverFiler() { + if ce.option.FilerAddress != "" || ce.MasterClient == nil || ce.option.FilerGroup == nil { + return + } + connectTimeout := ce.option.ConnectTimeout + if connectTimeout <= 0 { + connectTimeout = 30 * time.Second + } + ctx, cancel := context.WithTimeout(context.Background(), connectTimeout) + defer cancel() + var filers []pb.ServerAddress + var listErr error + for len(filers) == 0 && ctx.Err() == nil { + filers = filers[:0] + listErr = ce.MasterClient.WithClient(ctx, false, func(client master_pb.SeaweedClient) error { + resp, err := client.ListClusterNodes(ctx, &master_pb.ListClusterNodesRequest{ + ClientType: cluster.FilerType, + FilerGroup: *ce.option.FilerGroup, + }) + if err != nil { + return err + } + + for _, clusterNode := range resp.ClusterNodes { + filers = append(filers, pb.ServerAddress(clusterNode.Address)) + } + return nil + }) + if len(filers) == 0 { + if listErr != nil && !util.IsTransientError(listErr) { + break + } + select { + case <-ctx.Done(): + case <-time.After(time.Second): + } + } + } + if len(filers) > 0 { + ce.option.FilerAddress = filers[rand.IntN(len(filers))] + } else if listErr != nil { + fmt.Fprintf(os.Stderr, "cannot list filers from master %s: %v\n", *ce.option.Masters, listErr) + } else { + fmt.Fprintf(os.Stderr, "no filer registered with master %s within %s\n", *ce.option.Masters, connectTimeout) + } + if ce.option.Debug { + if len(filers) > 0 { + fmt.Fprintf(os.Stderr, "master: %s filers: %v\n", *ce.option.Masters, filers) + } else { + fmt.Fprintf(os.Stderr, "master: %s\n", *ce.option.Masters) + } + } +} + func (ce *CommandEnv) AdjustedUrl(location *filer_pb.Location) string { return location.Url } diff --git a/weed/shell/commands_test.go b/weed/shell/commands_test.go new file mode 100644 index 0000000000..8456413b7a --- /dev/null +++ b/weed/shell/commands_test.go @@ -0,0 +1,159 @@ +package shell + +import ( + "context" + "fmt" + "net" + "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 fakeDiscoveryServer struct { + master_pb.UnimplementedSeaweedServer + leader string + filers atomic.Int32 + listErr atomic.Value +} + +func (s *fakeDiscoveryServer) KeepConnected(stream master_pb.Seaweed_KeepConnectedServer) error { + if _, err := stream.Recv(); err != nil { + return err + } + if err := stream.Send(&master_pb.KeepConnectedResponse{ + VolumeLocation: &master_pb.VolumeLocation{Leader: s.leader}, + }); err != nil { + return err + } + <-stream.Context().Done() + return nil +} + +func (s *fakeDiscoveryServer) ListClusterNodes(context.Context, *master_pb.ListClusterNodesRequest) (*master_pb.ListClusterNodesResponse, error) { + if err, _ := s.listErr.Load().(error); err != nil { + return nil, err + } + resp := &master_pb.ListClusterNodesResponse{} + for i := int32(0); i < s.filers.Load(); i++ { + resp.ClusterNodes = append(resp.ClusterNodes, &master_pb.ListClusterNodesResponse_ClusterNode{ + Address: fmt.Sprintf("127.0.0.1:88%02d", i), + }) + } + return resp, nil +} + +func startFakeDiscoveryMaster(t *testing.T, srv *fakeDiscoveryServer) string { + t.Helper() + lis, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + grpcServer := grpc.NewServer() + master_pb.RegisterSeaweedServer(grpcServer, srv) + go func() { _ = grpcServer.Serve(lis) }() + t.Cleanup(grpcServer.GracefulStop) + + _, port, _ := net.SplitHostPort(lis.Addr().String()) + return fmt.Sprintf("127.0.0.1:0.%s", port) +} + +func discoveryEnv(addr string, timeout time.Duration) (*CommandEnv, context.CancelFunc) { + masters := addr + filerGroup := "" + options := &ShellOptions{ + Masters: &masters, + GrpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()), + FilerGroup: &filerGroup, + Directory: "/", + ConnectTimeout: timeout, + } + ce := NewCommandEnv(options) + ctx, cancel := context.WithCancel(context.Background()) + go ce.MasterClient.KeepConnectedToMaster(ctx) + ce.MasterClient.WaitUntilConnected(ctx) + return ce, cancel +} + +func TestFilerDiscoveryRetriesUntilFilerAppears(t *testing.T) { + srv := &fakeDiscoveryServer{} + addr := startFakeDiscoveryMaster(t, srv) + srv.leader = addr + + ce, cancel := discoveryEnv(addr, 10*time.Second) + defer cancel() + + go func() { + time.Sleep(1200 * time.Millisecond) + srv.filers.Store(1) + }() + + got := ce.FilerAddress() + if got != "127.0.0.1:8800" { + t.Fatalf("expected the late-registering filer, got %q", got) + } +} + +func TestFilerDiscoveryReturnsEmptyAtDeadline(t *testing.T) { + srv := &fakeDiscoveryServer{} + addr := startFakeDiscoveryMaster(t, srv) + srv.leader = addr + + ce, cancel := discoveryEnv(addr, 1500*time.Millisecond) + defer cancel() + + start := time.Now() + if got := ce.FilerAddress(); got != "" { + t.Fatalf("expected no filer, got %q", got) + } + if elapsed := time.Since(start); elapsed < time.Second { + t.Fatalf("discovery returned after %v, expected it to poll until the deadline", elapsed) + } +} + +func TestFilerDiscoveryRetriesAfterEmptyResult(t *testing.T) { + srv := &fakeDiscoveryServer{} + addr := startFakeDiscoveryMaster(t, srv) + srv.leader = addr + + ce, cancel := discoveryEnv(addr, 1500*time.Millisecond) + defer cancel() + + if got := ce.FilerAddress(); got != "" { + t.Fatalf("expected no filer, got %q", got) + } + srv.filers.Store(1) + if got := ce.FilerAddress(); got != "127.0.0.1:8800" { + t.Fatalf("expected the late-registered filer on retry, got %q", got) + } +} + +func TestFilerDiscoveryStopsOnNonTransientError(t *testing.T) { + srv := &fakeDiscoveryServer{} + addr := startFakeDiscoveryMaster(t, srv) + srv.leader = addr + srv.listErr.Store(status.Error(codes.PermissionDenied, "denied")) + + ce, cancel := discoveryEnv(addr, 30*time.Second) + defer cancel() + + start := time.Now() + if got := ce.FilerAddress(); got != "" { + t.Fatalf("expected no filer, got %q", got) + } + if elapsed := time.Since(start); elapsed > 10*time.Second { + t.Fatalf("discovery kept polling a non-transient error for %v", elapsed) + } +} + +func TestFilerAddressConfiguredSkipsDiscovery(t *testing.T) { + ce := &CommandEnv{option: &ShellOptions{FilerAddress: "127.0.0.1:8888"}} + if got := ce.FilerAddress(); got != "127.0.0.1:8888" { + t.Fatalf("expected the configured filer, got %q", got) + } +} diff --git a/weed/shell/s3tables_helpers.go b/weed/shell/s3tables_helpers.go index d595a5af74..18c5f90a72 100644 --- a/weed/shell/s3tables_helpers.go +++ b/weed/shell/s3tables_helpers.go @@ -20,7 +20,7 @@ func withFilerClient(commandEnv *CommandEnv, fn func(client filer_pb.SeaweedFile return pb.WithGrpcClient(context.Background(), false, 0, func(conn *grpc.ClientConn) error { client := filer_pb.NewSeaweedFilerClient(conn) return fn(client) - }, commandEnv.option.FilerAddress.ToGrpcAddress(), false, commandEnv.option.GrpcDialOption) + }, commandEnv.FilerAddress().ToGrpcAddress(), false, commandEnv.option.GrpcDialOption) } func executeS3Tables(commandEnv *CommandEnv, operation string, req interface{}, resp interface{}, accountID string) error { diff --git a/weed/shell/shell_liner.go b/weed/shell/shell_liner.go index 87d18c9b6c..303f2ab2ad 100644 --- a/weed/shell/shell_liner.go +++ b/weed/shell/shell_liner.go @@ -3,19 +3,14 @@ package shell import ( "bufio" "context" - "errors" "fmt" "io" - "math/rand/v2" "os" "path" "slices" "strings" "time" - "github.com/seaweedfs/seaweedfs/weed/cluster" - "github.com/seaweedfs/seaweedfs/weed/pb" - "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" "github.com/seaweedfs/seaweedfs/weed/util" "github.com/seaweedfs/seaweedfs/weed/util/grace" @@ -26,47 +21,6 @@ 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 { @@ -116,26 +70,6 @@ func RunShell(options ShellOptions) error { return err } - if commandEnv.option.FilerAddress == "" { - discoverCtx, cancelDiscover := context.WithTimeout(ctx, connectTimeout) - defer cancelDiscover() - 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 { - fmt.Fprintf(os.Stderr, "master: %s filers: %v\n", *options.Masters, filers) - } else { - fmt.Fprintf(os.Stderr, "master: %s\n", *options.Masters) - } - } - } - if interactive { for { cmd, err := line.Prompt("> ") diff --git a/weed/shell/shell_liner_test.go b/weed/shell/shell_liner_test.go index d8268f887f..21a707ac8a 100644 --- a/weed/shell/shell_liner_test.go +++ b/weed/shell/shell_liner_test.go @@ -1,22 +1,11 @@ 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 { @@ -94,233 +83,3 @@ 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) - } -}