From f068fe7f09c9259aae576d1e4ba06a18918bb5c2 Mon Sep 17 00:00:00 2001 From: MorezMartin Date: Wed, 11 Nov 2026 06:24:04 +0100 Subject: [PATCH] fix(shell): fs.log.purge purges every filer, not just one (#12524) * fix(shell): fs.log.purge purges every filer, not just one Each filer keeps its own metadata logs in its own store, so purging a single filer leaves the chunks referenced by the others and the volume vacuum cannot reclaim anything. The shell picked that single filer at random (commands.go:208) when no -filer flag was given, and the admin script worker pinned it to filers[0] (admin_script_handler.go:574), so the same command purged a different filer on every run and never the whole cluster. fs.log.purge now resolves the filer list from the master and purges each of them, continuing when one is unreachable and reporting the failures at the end. -filer=,... restricts the purge to a subset. Also stops mutating the shared command environment per filer, so a long-lived shell keeps its original filer. * fix(shell): bound filer discovery and cover the cluster path Address review feedback: - resolveLogPurgeFilers waited on the master with context.Background(): a shell whose master connection dropped would hang before purging anything, where it used to purge the connected filer. Bound the master round trip with a timeout and fall back to the connected filer when the master cannot be reached or lists no filer. - the tests only exercised the single-filer fallback. Rename that test to what it actually covers and add one asserting every filer of the list is attempted and the shared command environment keeps its original filer. * shell: reject an fs.log.purge -filer list with no addresses A comma or whitespace only -filer value fell through to master discovery and purged every filer instead of the intended subset. Co-Authored-By: Claude Opus 5.5 * shell: fall back to the shell's filer without a second master lookup When the master listed no filer, FilerAddress() ran filer discovery again and could wait out the shell's ConnectTimeout before failing. Also cover master discovery with the fake discovery master. Co-Authored-By: Claude Opus 5.5 * shell: test the fs.log.purge filer loop against a fake filer The previous test looped over the addresses itself, so it passed no matter what Do did. Co-Authored-By: Claude Opus 5.5 --------- Co-authored-by: Martin Morez Co-authored-by: Chris Lu Co-authored-by: Claude Opus 5.5 --- weed/shell/command_fs_log.go | 115 ++++++++++++++++++++++-- weed/shell/command_fs_log_test.go | 141 ++++++++++++++++++++++++++++++ 2 files changed, 251 insertions(+), 5 deletions(-) create mode 100644 weed/shell/command_fs_log_test.go diff --git a/weed/shell/command_fs_log.go b/weed/shell/command_fs_log.go index b93c2b9cdf..c613b79637 100644 --- a/weed/shell/command_fs_log.go +++ b/weed/shell/command_fs_log.go @@ -5,10 +5,14 @@ import ( "flag" "fmt" "io" + "strings" "time" + "github.com/seaweedfs/seaweedfs/weed/cluster" "github.com/seaweedfs/seaweedfs/weed/filer" + "github.com/seaweedfs/seaweedfs/weed/pb" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" ) func init() { @@ -18,6 +22,9 @@ func init() { type commandFsLogPurge struct { } +// filerDiscoveryTimeout bounds the master round trip used to list the filers. +const filerDiscoveryTimeout = 30 * time.Second + func (c *commandFsLogPurge) Name() string { return "fs.log.purge" } @@ -25,7 +32,11 @@ func (c *commandFsLogPurge) Name() string { func (c *commandFsLogPurge) Help() string { return `purge filer logs - fs.log.purge [-v] [-daysAgo 365] + fs.log.purge [-v] [-daysAgo 365] [-filer=,...] + + Each filer keeps its own metadata logs, so the purge runs on every filer + registered with the master by default. Use -filer to restrict it to a + subset of the cluster. ` } @@ -37,23 +48,117 @@ func (c *commandFsLogPurge) Do(args []string, commandEnv *CommandEnv, writer io. fsLogPurgeCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError) daysAgo := fsLogPurgeCommand.Uint("daysAgo", 365, "purge logs older than N days") verbose := fsLogPurgeCommand.Bool("v", false, "verbose mode") + filerList := fsLogPurgeCommand.String("filer", "", "comma separated filer addresses to purge (default: all filers in the cluster)") if err = fsLogPurgeCommand.Parse(args); err != nil { return err } + filers, err := resolveLogPurgeFilers(*filerList, commandEnv, *verbose, writer) + if err != nil { + return err + } + modificationTimeAgo := time.Now().Add(-time.Hour * 24 * time.Duration(*daysAgo)).Unix() - err = filer_pb.ReadDirAllEntries(context.Background(), commandEnv, filer.SystemLogDir, "", func(entry *filer_pb.Entry, isLast bool) error { + + var failed []string + for _, filerAddress := range filers { + if *verbose { + fmt.Fprintf(writer, "purging filer %s\n", filerAddress) + } + if purgeErr := purgeFilerLogs(commandEnv, filerAddress, modificationTimeAgo, *verbose, writer); purgeErr != nil { + // one unreachable filer must not stop the purge of the others + fmt.Fprintf(writer, "purge filer %s: %v\n", filerAddress, purgeErr) + failed = append(failed, fmt.Sprintf("%s: %v", filerAddress, purgeErr)) + } + } + if len(failed) > 0 { + return fmt.Errorf("failed to purge %d of %d filers: %s", len(failed), len(filers), strings.Join(failed, "; ")) + } + return nil +} + +// resolveLogPurgeFilers returns the filers to purge: the explicit -filer list +// when given, otherwise every filer registered with the master. +func resolveLogPurgeFilers(filerList string, commandEnv *CommandEnv, verbose bool, writer io.Writer) ([]pb.ServerAddress, error) { + if filerList != "" { + var filers []pb.ServerAddress + for _, address := range strings.Split(filerList, ",") { + if address = strings.TrimSpace(address); address != "" { + filers = append(filers, pb.ServerAddress(address)) + } + } + if len(filers) == 0 { + return nil, fmt.Errorf("-filer %q lists no filer address", filerList) + } + return filers, nil + } + + if commandEnv.MasterClient != nil && commandEnv.option.FilerGroup != nil { + // bound the master round trip: a shell whose master connection dropped + // must still fall back to the filer it is connected to instead of + // waiting forever before purging anything. + ctx, cancel := context.WithTimeout(context.Background(), filerDiscoveryTimeout) + defer cancel() + + var filers []pb.ServerAddress + listErr := commandEnv.MasterClient.WithClient(ctx, false, func(client master_pb.SeaweedClient) error { + resp, err := client.ListClusterNodes(ctx, &master_pb.ListClusterNodesRequest{ + ClientType: cluster.FilerType, + FilerGroup: *commandEnv.option.FilerGroup, + }) + if err != nil { + return err + } + for _, node := range resp.ClusterNodes { + filers = append(filers, pb.ServerAddress(node.Address)) + } + return nil + }) + if listErr == nil && len(filers) > 0 { + return filers, nil + } + // a master that cannot be reached or lists nothing is not fatal: the + // filer this shell talks to is still worth purging. + if listErr != nil && verbose { + fmt.Fprintf(writer, "cannot list filers from master: %v\n", listErr) + } + } + + commandEnv.filerMu.Lock() + address := commandEnv.option.FilerAddress + commandEnv.filerMu.Unlock() + if address != "" { + return []pb.ServerAddress{address}, nil + } + return nil, fmt.Errorf("no filer found to purge") +} + +// purgeFilerLogs purges the logs of a single filer, leaving the caller's +// command environment untouched. +func purgeFilerLogs(commandEnv *CommandEnv, filerAddress pb.ServerAddress, modificationTimeAgo int64, verbose bool, writer io.Writer) error { + options := *commandEnv.option + options.FilerAddress = filerAddress + filerEnv := &CommandEnv{ + env: commandEnv.env, + MasterClient: commandEnv.MasterClient, + option: &options, + locker: commandEnv.locker, + noLock: commandEnv.noLock, + forceNoLock: commandEnv.forceNoLock, + verbose: commandEnv.verbose, + } + + return filer_pb.ReadDirAllEntries(context.Background(), filerEnv, filer.SystemLogDir, "", func(entry *filer_pb.Entry, isLast bool) error { if entry.Attributes.Mtime > modificationTimeAgo { return nil } - if errDel := filer_pb.Remove(context.Background(), commandEnv, filer.SystemLogDir, entry.Name, true, true, true, false, nil); errDel != nil { + if errDel := filer_pb.Remove(context.Background(), filerEnv, filer.SystemLogDir, entry.Name, true, true, true, false, nil); errDel != nil { return errDel } - if *verbose { + if verbose { fmt.Fprintf(writer, "delete %s\n", entry.Name) } return nil }) - return err } diff --git a/weed/shell/command_fs_log_test.go b/weed/shell/command_fs_log_test.go new file mode 100644 index 0000000000..183035cb88 --- /dev/null +++ b/weed/shell/command_fs_log_test.go @@ -0,0 +1,141 @@ +package shell + +import ( + "slices" + "strings" + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/weed/filer" + "github.com/seaweedfs/seaweedfs/weed/pb" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" +) + +func TestResolveLogPurgeFilersExplicitList(t *testing.T) { + commandEnv := &CommandEnv{} + + filers, err := resolveLogPurgeFilers(" filer-a:8888, filer-b:8888 ,, filer-c:8888 ", commandEnv, false, &strings.Builder{}) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + + want := []pb.ServerAddress{"filer-a:8888", "filer-b:8888", "filer-c:8888"} + if len(filers) != len(want) { + t.Fatalf("got %d filers, want %d: %v", len(filers), len(want), filers) + } + for i, address := range want { + if filers[i] != address { + t.Errorf("filer[%d] = %q, want %q", i, filers[i], address) + } + } +} + +func TestResolveLogPurgeFilersRejectsEmptyExplicitList(t *testing.T) { + commandEnv := &CommandEnv{option: &ShellOptions{FilerAddress: "filer-only:8888"}} + + for _, filerList := range []string{" ", ",,", " , "} { + if filers, err := resolveLogPurgeFilers(filerList, commandEnv, false, &strings.Builder{}); err == nil { + t.Errorf("-filer=%q: got %v, want an error", filerList, filers) + } + } +} + +func TestResolveLogPurgeFilersFallsBackToConnectedFiler(t *testing.T) { + // no explicit list and no master client: fall back to the connected filer + commandEnv := &CommandEnv{option: &ShellOptions{FilerAddress: "filer-only:8888"}} + + filers, err := resolveLogPurgeFilers("", commandEnv, false, &strings.Builder{}) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if len(filers) != 1 || filers[0] != "filer-only:8888" { + t.Fatalf("got %v, want [filer-only:8888]", filers) + } +} + +func TestResolveLogPurgeFilersFromMaster(t *testing.T) { + srv := &fakeDiscoveryServer{} + addr := startFakeDiscoveryMaster(t, srv) + srv.leader = addr + srv.filers.Store(3) + + commandEnv, cancel := discoveryEnv(addr, 10*time.Second) + defer cancel() + commandEnv.option.FilerAddress = "filer-connected:8888" + + filers, err := resolveLogPurgeFilers("", commandEnv, false, &strings.Builder{}) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + want := []pb.ServerAddress{"127.0.0.1:8800", "127.0.0.1:8801", "127.0.0.1:8802"} + if !slices.Equal(filers, want) { + t.Fatalf("got %v, want %v", filers, want) + } +} + +func TestResolveLogPurgeFilersEmptyMasterSkipsSecondLookup(t *testing.T) { + srv := &fakeDiscoveryServer{} + addr := startFakeDiscoveryMaster(t, srv) + srv.leader = addr + + commandEnv, cancel := discoveryEnv(addr, 10*time.Second) + defer cancel() + + start := time.Now() + if filers, err := resolveLogPurgeFilers("", commandEnv, false, &strings.Builder{}); err == nil { + t.Fatalf("got %v, want an error", filers) + } + if elapsed := time.Since(start); elapsed > 5*time.Second { + t.Fatalf("resolve took %v, it should not wait out the shell's filer discovery", elapsed) + } + + commandEnv.option.FilerAddress = "filer-connected:8888" + filers, err := resolveLogPurgeFilers("", commandEnv, false, &strings.Builder{}) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if !slices.Equal(filers, []pb.ServerAddress{"filer-connected:8888"}) { + t.Fatalf("got %v, want [filer-connected:8888]", filers) + } +} + +func TestResolveLogPurgeFilersNoneFound(t *testing.T) { + commandEnv := &CommandEnv{option: &ShellOptions{}} + + if _, err := resolveLogPurgeFilers("", commandEnv, false, &strings.Builder{}); err == nil { + t.Fatal("expected an error when no filer can be resolved") + } +} + +func TestPurgeFilerLogsKeepsCommandEnvFiler(t *testing.T) { + // the per-filer purge must not mutate the shared command environment, + // otherwise the next filer of the loop would be purged twice and the + // original one never. + commandEnv := &CommandEnv{option: &ShellOptions{FilerAddress: "filer-original:8888"}} + + _ = purgeFilerLogs(commandEnv, "filer-other:8888", 0, false, &strings.Builder{}) + + if got := commandEnv.option.FilerAddress; got != "filer-original:8888" { + t.Fatalf("command env filer changed to %q, want filer-original:8888", got) + } +} + +func TestFsLogPurgeContinuesPastUnreachableFiler(t *testing.T) { + stub := &stubFilerServer{entries: map[string]*filer_pb.Entry{ + filer.SystemLogDir + "/2020-01-01": {IsDirectory: true, Attributes: &filer_pb.FuseAttributes{Mtime: 1}}, + filer.SystemLogDir + "/today": {IsDirectory: true, Attributes: &filer_pb.FuseAttributes{Mtime: time.Now().Unix()}}, + }} + commandEnv := startStubFiler(t, stub) + reachable := commandEnv.option.FilerAddress + + err := (&commandFsLogPurge{}).Do([]string{"-daysAgo=1", "-filer=127.0.0.1:1.1," + string(reachable) + ",127.0.0.1:2.2"}, commandEnv, &strings.Builder{}) + if err == nil || !strings.Contains(err.Error(), "failed to purge 2 of 3 filers") { + t.Fatalf("got error %v, want both unreachable filers reported", err) + } + if !slices.Equal(stub.deleted, []string{filer.SystemLogDir + "/2020-01-01"}) { + t.Fatalf("reachable filer deleted %v, want only the old log directory", stub.deleted) + } + if commandEnv.option.FilerAddress != reachable { + t.Fatalf("command env filer changed to %q, want %q", commandEnv.option.FilerAddress, reachable) + } +}