Github-Mirror/seaweedfs
1
0 Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2026-11-09 05:18:23 +01:00
Code Issues Packages Projects Releases Wiki Activity

shell: retry filer discovery until master knows a filer (#11989)

Browse Source 
* 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>
This commit is contained in:
Chris LuandDevin authored and GitHub committed 2026-10-23 22:52:51 -07:00
1 parent b86b2f3cee
commit 6a34d82f05
18 files changed
+246 -325
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

weed/shell/shell_liner_test.go
-241
View File
@@ -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)
}
}