From 4ce027c2f3bf5235cd2620308ae11791d89a2548 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Tue, 12 May 2026 18:29:18 -0700 Subject: [PATCH] fix(shell): restore s3.lifecycle.run-shard for CI/integration-test compatibility (#9475) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit fix(shell): restore s3.lifecycle.run-shard as a dailyrun.Run wrapper PR #9466 deleted weed/shell/command_s3_lifecycle_run_shard.go on the premise that it was a debug-only tool. It wasn't: the s3tests CI workflow (.github/workflows/s3tests.yml) and the test/s3/lifecycle/ integration tests invoke it via `weed shell` to drive lifecycle expirations on demand. Both started failing with "unknown command: s3.lifecycle.run-shard". This PR restores the command with the same flag set so existing callers (CI scripts and integration tests) work unchanged. The implementation no longer drives the streaming dispatcher.Pipeline + scheduler.BucketBootstrapper (deleted) — instead it does one bounded dailyrun.Run pass through the same daily-replay code path the production worker exercises. The walker fires for walker-bound rules just like in the worker. Obsolete streaming flags (-dispatch / -checkpoint / -refresh / -bootstrap-interval) are accepted-but-ignored so existing scripts don't need to drop them. --- weed/shell/command_s3_lifecycle_run_shard.go | 293 ++++++++++++++++++ .../command_s3_lifecycle_run_shard_test.go | 92 ++++++ 2 files changed, 385 insertions(+) create mode 100644 weed/shell/command_s3_lifecycle_run_shard.go create mode 100644 weed/shell/command_s3_lifecycle_run_shard_test.go diff --git a/weed/shell/command_s3_lifecycle_run_shard.go b/weed/shell/command_s3_lifecycle_run_shard.go new file mode 100644 index 000000000..b11132641 --- /dev/null +++ b/weed/shell/command_s3_lifecycle_run_shard.go @@ -0,0 +1,293 @@ +package shell + +import ( + "context" + "flag" + "fmt" + "io" + "sort" + "strconv" + "strings" + "time" + + "github.com/seaweedfs/seaweedfs/weed/pb" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/pb/s3_lifecycle_pb" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/dailyrun" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/dispatcher" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/engine" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/scheduler" +) + +func init() { + Commands = append(Commands, &commandS3LifecycleRunShard{}) +} + +type commandS3LifecycleRunShard struct{} + +func (c *commandS3LifecycleRunShard) Name() string { + return "s3.lifecycle.run-shard" +} + +func (c *commandS3LifecycleRunShard) Help() string { + return `manually run one daily-replay pass for the given shards + +Drives dailyrun.Run once against the live filer + S3 server: builds +the engine snapshot from filer-backed bucket configs, opens the +meta-log subscription per shard, dispatches due actions via +LifecycleDelete, and walks the live tree for any walker-bound rules. +Persists each shard's cursor to /etc/s3/lifecycle/daily-cursors/ +so subsequent runs resume. + +Used by the s3-tests CI workflow and the test/s3/lifecycle/ +integration tests to drive expirations on demand without standing up +the full admin+worker plugin stack. + + # single shard + s3.lifecycle.run-shard -shard 0 -s3 localhost:8333 -events 100 + + # contiguous range + s3.lifecycle.run-shard -shards 0-15 -s3 localhost:8333 -events 5000 + + # explicit set + s3.lifecycle.run-shard -shards 0,3,7 -s3 localhost:8333 + + # bounded wall-clock + s3.lifecycle.run-shard -shards 0-15 -s3 localhost:8333 -runtime 10s +` +} + +func (c *commandS3LifecycleRunShard) HasTag(CommandTag) bool { return false } + +func (c *commandS3LifecycleRunShard) Do(args []string, env *CommandEnv, writer io.Writer) error { + fs := flag.NewFlagSet(c.Name(), flag.ContinueOnError) + shard := fs.Int("shard", -1, "single shard id in [0, 16); use -shards for a range or set") + shardsSpec := fs.String("shards", "", "shard range \"lo-hi\" or comma list \"a,b,c\"; mutually exclusive with -shard") + s3Endpoint := fs.String("s3", "", "s3 server gRPC endpoint, host:port") + eventBudget := fs.Int("events", 1000, "max in-shard events to process before returning (0 = unbounded)") + runtime := fs.Duration("runtime", 0, "wall-clock cap on the run; 0 = no timeout") + // Obsolete flags kept for back-compat with existing CI scripts and + // integration tests. The daily-replay path has no streaming-tick + // cadence; accept and ignore. + _ = fs.Duration("dispatch", 0, "ignored (legacy streaming-dispatcher flag)") + _ = fs.Duration("checkpoint", 0, "ignored (legacy streaming-checkpoint flag)") + _ = fs.Duration("refresh", 0, "ignored (legacy streaming-refresh flag)") + _ = fs.Duration("bootstrap-interval", 0, "ignored (legacy streaming-bootstrap flag)") + if err := fs.Parse(args); err != nil { + return err + } + + shards, err := resolveShardSelection(*shard, *shardsSpec) + if err != nil { + return err + } + if *s3Endpoint == "" { + return fmt.Errorf("-s3 required (host:port of s3 server gRPC)") + } + if *eventBudget < 0 { + return fmt.Errorf("-events must be >= 0 (0 = unbounded)") + } + + bucketsPath, err := resolveBucketsPath(env) + if err != nil { + return fmt.Errorf("resolve buckets path: %w", err) + } + fmt.Fprintf(writer, "buckets path: %s\n", bucketsPath) + + dialCtx, dialCancel := context.WithTimeout(context.Background(), 30*time.Second) + conn, err := pb.GrpcDial(dialCtx, *s3Endpoint, false, env.option.GrpcDialOption) + dialCancel() + if err != nil { + return fmt.Errorf("dial s3 %s: %w", *s3Endpoint, err) + } + defer conn.Close() + rpcClient := s3_lifecycle_pb.NewSeaweedS3LifecycleInternalClient(conn) + + return env.WithFilerClient(true, func(filerClient filer_pb.SeaweedFilerClient) error { + eng := engine.New() + inputs, parseErrors, err := scheduler.LoadCompileInputs(context.Background(), filerClient, bucketsPath) + if err != nil { + return fmt.Errorf("load lifecycle configs: %w", err) + } + for i, pe := range parseErrors { + if i < 3 { + fmt.Fprintf(writer, "warning: %s: %v\n", pe.Bucket, pe.Err) + } + } + if extra := len(parseErrors) - 3; extra > 0 { + fmt.Fprintf(writer, "warning: %d additional bucket(s) had malformed lifecycle config\n", extra) + } + eng.Compile(inputs, engine.CompileOptions{PriorStates: scheduler.AllActivePriorStates(inputs)}) + if len(inputs) == 0 { + fmt.Fprintln(writer, "no buckets with enabled lifecycle rules; nothing to run") + return nil + } + + buckets := make([]string, 0, len(inputs)) + for _, in := range inputs { + if in.Bucket != "" { + buckets = append(buckets, in.Bucket) + } + } + client := &lifecycleClientCallable{c: rpcClient} + listFn := dailyrun.FilerListFunc(filerClient, bucketsPath) + walkerDispatch := &dailyrun.WalkerDispatcher{Client: client} + walker := dailyrun.WalkerFunc(func(walkCtx context.Context, view *engine.Snapshot, shardID int) error { + return dailyrun.WalkBuckets(walkCtx, view, shardID, buckets, listFn, walkerDispatch) + }) + + ctx := context.Background() + var cancel context.CancelFunc + if *runtime > 0 { + ctx, cancel = context.WithTimeout(ctx, *runtime) + defer cancel() + } + + fmt.Fprintf(writer, "running shards %s (event budget=%d, runtime=%s)…\n", + formatShardLabel(shards), *eventBudget, *runtime) + runErr := dailyrun.Run(ctx, dailyrun.Config{ + Shards: shards, + BucketsPath: bucketsPath, + Engine: eng, + FilerClient: filerClient, + Client: client, + Persister: &dailyrun.FilerCursorPersister{Store: dispatcher.NewFilerStoreClient(filerClient)}, + Lister: dispatcher.NewFilerSiblingLister(filerClient, bucketsPath), + Walker: walker, + EventBudget: *eventBudget, + ClientName: fmt.Sprintf("shell-lifecycle-%s", formatShardLabel(shards)), + }) + if runErr != nil { + return fmt.Errorf("daily_run: %w", runErr) + } + fmt.Fprintf(writer, "shards %s complete; cursors checkpointed\n", formatShardLabel(shards)) + return nil + }) +} + +// resolveShardSelection turns the -shard / -shards flags into a sorted, +// deduplicated []int. Exactly one form must be specified. +func resolveShardSelection(singleShard int, shardsSpec string) ([]int, error) { + if singleShard >= 0 && shardsSpec != "" { + return nil, fmt.Errorf("-shard and -shards are mutually exclusive") + } + if singleShard < 0 && shardsSpec == "" { + return nil, fmt.Errorf("specify -shard or -shards ") + } + if singleShard >= 0 { + if singleShard >= s3lifecycle.ShardCount { + return nil, fmt.Errorf("-shard %d out of [0,%d)", singleShard, s3lifecycle.ShardCount) + } + return []int{singleShard}, nil + } + return parseShardsSpec(shardsSpec) +} + +// parseShardsSpec accepts "lo-hi" (inclusive) or "a,b,c" and returns a +// sorted, deduplicated, in-range []int. +func parseShardsSpec(spec string) ([]int, error) { + spec = strings.TrimSpace(spec) + seen := map[int]struct{}{} + add := func(v int) error { + if v < 0 || v >= s3lifecycle.ShardCount { + return fmt.Errorf("shard %d out of [0,%d)", v, s3lifecycle.ShardCount) + } + seen[v] = struct{}{} + return nil + } + if strings.Contains(spec, "-") && !strings.Contains(spec, ",") { + parts := strings.SplitN(spec, "-", 2) + lo, err := strconv.Atoi(strings.TrimSpace(parts[0])) + if err != nil { + return nil, fmt.Errorf("range lo: %w", err) + } + hi, err := strconv.Atoi(strings.TrimSpace(parts[1])) + if err != nil { + return nil, fmt.Errorf("range hi: %w", err) + } + if lo > hi { + return nil, fmt.Errorf("range lo %d > hi %d", lo, hi) + } + for v := lo; v <= hi; v++ { + if err := add(v); err != nil { + return nil, err + } + } + } else { + for _, part := range strings.Split(spec, ",") { + part = strings.TrimSpace(part) + if part == "" { + continue + } + v, err := strconv.Atoi(part) + if err != nil { + return nil, fmt.Errorf("shard list: %w", err) + } + if err := add(v); err != nil { + return nil, err + } + } + } + if len(seen) == 0 { + return nil, fmt.Errorf("empty shard set") + } + out := make([]int, 0, len(seen)) + for v := range seen { + out = append(out, v) + } + sort.Ints(out) + return out, nil +} + +func formatShardLabel(shards []int) string { + if len(shards) == 1 { + return fmt.Sprintf("%d", shards[0]) + } + contiguous := true + for i := 1; i < len(shards); i++ { + if shards[i] != shards[i-1]+1 { + contiguous = false + break + } + } + if contiguous { + return fmt.Sprintf("%d-%d", shards[0], shards[len(shards)-1]) + } + parts := make([]string, len(shards)) + for i, v := range shards { + parts[i] = strconv.Itoa(v) + } + return strings.Join(parts, ",") +} + +// lifecycleClientCallable adapts the generated grpc client (variadic +// CallOption tail) to dailyrun.LifecycleClient. +type lifecycleClientCallable struct { + c s3_lifecycle_pb.SeaweedS3LifecycleInternalClient +} + +func (l *lifecycleClientCallable) LifecycleDelete(ctx context.Context, req *s3_lifecycle_pb.LifecycleDeleteRequest) (*s3_lifecycle_pb.LifecycleDeleteResponse, error) { + return l.c.LifecycleDelete(ctx, req) +} + +// resolveBucketsPath fetches the filer's configured buckets directory. +// Falls back to /buckets when the filer doesn't return one. +func resolveBucketsPath(env *CommandEnv) (string, error) { + var path string + err := env.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + resp, err := client.GetFilerConfiguration(context.Background(), &filer_pb.GetFilerConfigurationRequest{}) + if err != nil { + return err + } + path = resp.GetDirBuckets() + return nil + }) + if err != nil { + return "", err + } + if path == "" { + path = "/buckets" + } + return path, nil +} diff --git a/weed/shell/command_s3_lifecycle_run_shard_test.go b/weed/shell/command_s3_lifecycle_run_shard_test.go new file mode 100644 index 000000000..a1f5321ef --- /dev/null +++ b/weed/shell/command_s3_lifecycle_run_shard_test.go @@ -0,0 +1,92 @@ +package shell + +import ( + "reflect" + "testing" +) + +func TestParseShardsSpec_Range(t *testing.T) { + got, err := parseShardsSpec("3-7") + if err != nil { + t.Fatalf("err: %v", err) + } + want := []int{3, 4, 5, 6, 7} + if !reflect.DeepEqual(got, want) { + t.Fatalf("got %v, want %v", got, want) + } +} + +func TestParseShardsSpec_Set(t *testing.T) { + got, err := parseShardsSpec("0,3,7") + if err != nil { + t.Fatalf("err: %v", err) + } + want := []int{0, 3, 7} + if !reflect.DeepEqual(got, want) { + t.Fatalf("got %v, want %v", got, want) + } +} + +func TestParseShardsSpec_DedupSort(t *testing.T) { + got, err := parseShardsSpec("7,3,3,0") + if err != nil { + t.Fatalf("err: %v", err) + } + want := []int{0, 3, 7} + if !reflect.DeepEqual(got, want) { + t.Fatalf("got %v, want %v", got, want) + } +} + +func TestParseShardsSpec_OutOfRange(t *testing.T) { + if _, err := parseShardsSpec("16"); err == nil { + t.Fatal("expected out-of-range error for 16") + } + if _, err := parseShardsSpec("-1"); err == nil { + t.Fatal("expected out-of-range error for -1") + } + if _, err := parseShardsSpec("0-16"); err == nil { + t.Fatal("expected out-of-range error for 0-16") + } +} + +func TestParseShardsSpec_BadRange(t *testing.T) { + if _, err := parseShardsSpec("7-3"); err == nil { + t.Fatal("expected lo>hi error") + } +} + +func TestResolveShardSelection_Mutex(t *testing.T) { + if _, err := resolveShardSelection(0, "1,2"); err == nil { + t.Fatal("expected mutex error when both -shard and -shards set") + } + if _, err := resolveShardSelection(-1, ""); err == nil { + t.Fatal("expected error when neither set") + } +} + +func TestResolveShardSelection_SingleShard(t *testing.T) { + got, err := resolveShardSelection(5, "") + if err != nil { + t.Fatalf("err: %v", err) + } + if !reflect.DeepEqual(got, []int{5}) { + t.Fatalf("got %v, want [5]", got) + } +} + +func TestFormatShardLabel(t *testing.T) { + cases := []struct { + in []int + want string + }{ + {[]int{5}, "5"}, + {[]int{0, 1, 2, 3}, "0-3"}, + {[]int{0, 2, 5}, "0,2,5"}, + } + for _, tc := range cases { + if got := formatShardLabel(tc.in); got != tc.want { + t.Errorf("formatShardLabel(%v)=%q, want %q", tc.in, got, tc.want) + } + } +}