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