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