From 1854101125423689dc315a08051ad20b6bd000e9 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Sat, 9 May 2026 13:42:31 -0700 Subject: [PATCH] feat(s3/lifecycle): bootstrap re-walk cadence + operator hooks (Phase 8) (#9386) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat(s3/lifecycle): bootstrap re-walk cadence + operator hooks (Phase 8) scan_only actions only fire from the bootstrap walk: the engine classifies a rule as scan_only when its retention horizon exceeds the meta-log retention, so event-driven routing can't be trusted. Today each bucket walks once per process, so a long-running worker never revisits — scan_only retention only catches up when the worker restarts. Replace BucketBootstrapper.known (set) with BucketBootstrapper.lastWalk (name -> completion time). KickOffNew now re-walks a bucket whose last walk completed more than BootstrapInterval ago. Zero interval preserves the legacy walk-once-per-process behavior so existing deployments don't change cadence by default. walkBucket re-stamps on success and clears the stamp on failure (via MarkDirty), so the next KickOffNew picks failed walks back up. Add MarkDirty / MarkAllDirty operator hooks for forced re-walks, and a Now func() for testable time travel. weed shell run-shard grows --bootstrap-interval (cadence knob) and --force-bootstrap (drop in-memory state at startup so every bucket walks again immediately, useful when a config change should take effect without a restart). Tests: cadence respected (skip inside interval, re-walk past it); zero interval keeps once-per-process; MarkDirty forces re-walk under a 24h interval; MarkAllDirty resets every record. The fakeClock helper guards the test clock with a mutex so race-detector runs are clean. * fix(s3/lifecycle): split walk state, thread BootstrapInterval through worker, drop dead flag Three issues with the Phase 8 cadence work as it landed: 1. lastWalk did double duty as both completed-walk timestamp and in-flight debounce. A walk that took longer than BootstrapInterval would have a fresh KickOffNew start a duplicate goroutine on the next refresh tick because the stamp from KickOffNew looked stale against the interval. Split into lastCompleted (set on success) and inFlight (set on dispatch, cleared after the walk goroutine returns success or failure). KickOffNew skips inFlight buckets regardless of cadence. 2. The cadence knob existed on `weed shell` but not on the production path: scheduler.Scheduler constructed BucketBootstrapper without BootstrapInterval, and weed/worker/tasks/s3_lifecycle/Config had no field for it. Add Scheduler.BootstrapInterval, parse `bootstrap_interval_minutes` in ParseConfig (zero = legacy walk- once-per-process; negative clamps to zero), and forward it from the handler. Tests cover default, override, clamp, and explicit-zero. 3. --force-bootstrap was a no-op: BucketBootstrapper is freshly allocated at command start, so MarkAllDirty on empty state does nothing, and the flag couldn't influence an already-running process anyway. Remove it; a real runtime trigger (SIGHUP, control RPC) is a separate change. In-flight regression: a blockingInjector pins the first walk in progress while the test advances the clock past the interval. The second KickOffNew is a no-op (inFlight check). After release, the post-completion KickOffNew within the interval is also a no-op. * test(s3/lifecycle): wait for lastCompleted stamp before advancing fake clock The cadence test polled listedN to know "the walk happened" — but that fires once both list passes are issued, while the success-stamp lands later, after walkBucketDir returns. A clock.Advance(30m) between those two events would record the stamp at clock+30m instead of T0; the next assertion would then see now.Sub(last) < 1h and skip the expected re-walk. Tight in practice but exposed under -race / load. Add a waitForCompleted helper that polls b.lastCompleted directly, and use it before each clock advance in both the cadence and zero- interval tests. * fix(s3/lifecycle): expose bootstrap interval in worker UI; honor MarkDirty during walks Two follow-ups on Phase 8. The worker config descriptor had no bootstrap_interval_minutes field, so the production operator UI couldn't enable the cadence — only the internal ParseConfig + Scheduler wiring knew about it. Add the field to the cadence section (MinValue=0 since 0 is the legacy default) and include the default in DefaultValues so existing deployments see the knob with the right preset. MarkDirty / MarkAllDirty silently lost their effect when a walk was in flight: the methods cleared lastCompleted, but the walk's success path then wrote a fresh timestamp, hiding the operator's invalidation. Track a pendingDirty set; the walk goroutine consumes the flag on exit and skips the success stamp, so the next KickOffNew picks the bucket up immediately. Regression: pin a walk in progress with a blockingInjector, MarkDirty the bucket, release the walk, and assert lastCompleted stayed empty plus the next KickOffNew triggers a new walk inside the BootstrapInterval window. * refactor(s3/lifecycle): drop unused MarkDirty / MarkAllDirty + pendingDirty These methods were the operator-hook half of Phase 8, but the only caller (--force-bootstrap on the shell command) was removed when it turned out to be a no-op against a freshly-allocated bootstrapper. Nothing in production calls them anymore. Strip the dead surface: MarkDirty, MarkAllDirty, the pendingDirty set, the dirty-suppression branch in walkBucket, and the three tests that only exercised those methods. BootstrapInterval-driven re-bootstrap is the live mechanism. A real runtime trigger (SIGHUP, control RPC) is a separate change with a real call site. --- weed/s3api/s3lifecycle/scheduler/bootstrap.go | 72 +++++- .../s3lifecycle/scheduler/bootstrap_test.go | 208 +++++++++++++++++- weed/s3api/s3lifecycle/scheduler/scheduler.go | 18 +- weed/shell/command_s3_lifecycle_run_shard.go | 8 +- weed/worker/tasks/s3_lifecycle/config.go | 40 ++-- weed/worker/tasks/s3_lifecycle/config_test.go | 40 +++- weed/worker/tasks/s3_lifecycle/handler.go | 40 ++-- 7 files changed, 359 insertions(+), 67 deletions(-) diff --git a/weed/s3api/s3lifecycle/scheduler/bootstrap.go b/weed/s3api/s3lifecycle/scheduler/bootstrap.go index ce17f64d4..af85c91ad 100644 --- a/weed/s3api/s3lifecycle/scheduler/bootstrap.go +++ b/weed/s3api/s3lifecycle/scheduler/bootstrap.go @@ -78,33 +78,70 @@ func listAll(ctx context.Context, client filer_pb.SeaweedFilerClient, dir string // Synthesized events carry TsNs=0 so dispatcher.advance is a no-op for // them — the reader still resumes from its persisted cursor on restart. // -// Bucket completion is in-memory per process; a fresh worker run walks -// each bucket once on first refresh. +// Walk state is tracked in-memory per process via two maps: +// - lastCompleted: time of the last successful walk, drives the +// BootstrapInterval cadence; zero interval means "walk once per +// process". +// - inFlight: buckets whose walk goroutines are still running. +// Skipped regardless of cadence so a walk that takes longer than +// BootstrapInterval can't trigger a duplicate goroutine on the +// next refresh. type BucketBootstrapper struct { FilerClient filer_pb.SeaweedFilerClient BucketsPath string Injector EventInjector - mu sync.Mutex - known map[string]bool + // BootstrapInterval gates re-walks. Zero means "walk once per + // process". Non-zero means "walk again once it's been at least + // this long since the last completed walk" — the cadence scan_only + // actions rely on, since they can only fire from bootstrap. + BootstrapInterval time.Duration + + // Now overrides time.Now for tests. + Now func() time.Time + + mu sync.Mutex + lastCompleted map[string]time.Time + inFlight map[string]bool +} + +func (b *BucketBootstrapper) now() time.Time { + if b.Now != nil { + return b.Now() + } + return time.Now() } // KickOffNew launches a one-shot walker goroutine for every bucket -// in `buckets` that hasn't been seen before. +// that's not currently in flight and either has never completed a walk +// or whose last successful walk finished more than BootstrapInterval ago. func (b *BucketBootstrapper) KickOffNew(ctx context.Context, buckets []string) { if b.Injector == nil { return } + now := b.now() b.mu.Lock() - if b.known == nil { - b.known = map[string]bool{} + if b.lastCompleted == nil { + b.lastCompleted = map[string]time.Time{} + } + if b.inFlight == nil { + b.inFlight = map[string]bool{} } fresh := make([]string, 0, len(buckets)) for _, bucket := range buckets { - if b.known[bucket] { + if b.inFlight[bucket] { + // Walk still running from an earlier KickOffNew — never + // double up regardless of cadence. A large bucket that + // takes longer than BootstrapInterval would otherwise + // have a fresh goroutine fire on every refresh tick. continue } - b.known[bucket] = true + if last, ok := b.lastCompleted[bucket]; ok { + if b.BootstrapInterval <= 0 || now.Sub(last) < b.BootstrapInterval { + continue + } + } + b.inFlight[bucket] = true fresh = append(fresh, bucket) } b.mu.Unlock() @@ -152,9 +189,22 @@ func (b *BucketBootstrapper) walkBucket(ctx context.Context, bucket string) { count++ return b.Injector.InjectEvent(ctx, ev) } - if err := walkBucketDir(ctx, b.FilerClient, root, root, cb); err != nil { + walkErr := walkBucketDir(ctx, b.FilerClient, root, root, cb) + b.mu.Lock() + delete(b.inFlight, bucket) + if walkErr == nil { + // Stamp completion so BootstrapInterval cadence measures from + // end-of-walk. Failures leave lastCompleted alone, so the next + // KickOffNew sees no record and walks the bucket again. + if b.lastCompleted == nil { + b.lastCompleted = map[string]time.Time{} + } + b.lastCompleted[bucket] = b.now() + } + b.mu.Unlock() + if walkErr != nil { if ctx.Err() == nil { - glog.V(0).Infof("lifecycle bootstrap %s: %v", bucket, err) + glog.V(0).Infof("lifecycle bootstrap %s: %v", bucket, walkErr) } return } diff --git a/weed/s3api/s3lifecycle/scheduler/bootstrap_test.go b/weed/s3api/s3lifecycle/scheduler/bootstrap_test.go index 2f4b1b450..620398f90 100644 --- a/weed/s3api/s3lifecycle/scheduler/bootstrap_test.go +++ b/weed/s3api/s3lifecycle/scheduler/bootstrap_test.go @@ -396,9 +396,11 @@ func TestBucketBootstrapper_KickOffNew_LaunchesPerBucket(t *testing.T) { b.mu.Lock() defer b.mu.Unlock() - assert.True(t, b.known["bucketA"]) - assert.True(t, b.known["bucketB"]) - assert.Len(t, b.known, 2) + _, hasA := b.lastCompleted["bucketA"] + _, hasB := b.lastCompleted["bucketB"] + assert.True(t, hasA) + assert.True(t, hasB) + assert.Len(t, b.lastCompleted, 2) } func TestBucketBootstrapper_KickOffNew_SkipsAlreadyKnown(t *testing.T) { @@ -442,8 +444,9 @@ func TestBucketBootstrapper_KickOffNew_SkipsAlreadyKnown(t *testing.T) { b.mu.Lock() defer b.mu.Unlock() - assert.Len(t, b.known, 3) - assert.True(t, b.known["bucketC"]) + assert.Len(t, b.lastCompleted, 3) + _, hasC := b.lastCompleted["bucketC"] + assert.True(t, hasC) } func TestBucketBootstrapper_KickOffNew_NilInjectorIsNoop(t *testing.T) { @@ -456,12 +459,13 @@ func TestBucketBootstrapper_KickOffNew_NilInjectorIsNoop(t *testing.T) { require.NotPanics(t, func() { b.KickOffNew(context.Background(), []string{"bucketA"}) }) - // No walks must have been kicked off, and known must remain empty. + // No walks must have been kicked off, and the lastWalk map must + // remain empty. time.Sleep(20 * time.Millisecond) assert.Equal(t, int32(0), atomic.LoadInt32(&client.listedN)) b.mu.Lock() defer b.mu.Unlock() - assert.Empty(t, b.known) + assert.Empty(t, b.lastCompleted) } func TestBucketBootstrapper_KickOffNew_EmptyBucketListIsNoop(t *testing.T) { @@ -1100,3 +1104,193 @@ func TestWalkBucketDir_PaginatesBeyondListingLimit(t *testing.T) { // entries at page 2 = 3 paginated calls per pass = 6 total. assert.Equal(t, 6, calls) } + +// fakeClock is a thread-safe time source for tests that need to fast- +// forward across a BootstrapInterval boundary. Bootstrap goroutines +// read it concurrently with the test advancing it, so a plain +// `clock := time.Now()` plus closure write would race under -race. +type fakeClock struct { + mu sync.Mutex + t time.Time +} + +func newFakeClock() *fakeClock { return &fakeClock{t: time.Now()} } + +func (c *fakeClock) Now() time.Time { + c.mu.Lock() + defer c.mu.Unlock() + return c.t +} + +func (c *fakeClock) Advance(d time.Duration) { + c.mu.Lock() + c.t = c.t.Add(d) + c.mu.Unlock() +} + +// waitForCompleted blocks until the bootstrapper has stamped a +// lastCompleted entry for the given bucket. Polling listedN is not +// enough — that fires once both list passes have started, but +// walkBucket stamps lastCompleted only after walkBucketDir returns, +// so a clock.Advance between those events would record the stamp at +// post-advance time and skew BootstrapInterval cadence assertions. +func waitForCompleted(t *testing.T, b *BucketBootstrapper, bucket string) { + t.Helper() + waitFor(t, func() bool { + b.mu.Lock() + _, ok := b.lastCompleted[bucket] + b.mu.Unlock() + return ok + }, "lastCompleted stamp for "+bucket) +} + +func TestBucketBootstrapper_KickOffNew_BootstrapIntervalRevisitsBucket(t *testing.T) { + // scan_only actions only fire from bootstrap, so a long-running + // worker has to revisit each bucket on a cadence. With + // BootstrapInterval set, KickOffNew re-walks once enough wall-clock + // has passed since the last completed walk. + client := newEmptyFilerClient() + inj := &recordingInjector{} + clock := newFakeClock() + b := &BucketBootstrapper{ + FilerClient: client, + BucketsPath: "/buckets", + Injector: inj, + BootstrapInterval: time.Hour, + Now: clock.Now, + } + + // First wave: walks once. Wait for the goroutine to actually stamp + // lastCompleted before advancing the clock — otherwise the stamp + // could land at clock+30m instead of T0 and the cadence assertion + // would race. + b.KickOffNew(context.Background(), []string{"bucketA"}) + waitForCompleted(t, b, "bucketA") + firstCount := atomic.LoadInt32(&client.listedN) + + // Inside the interval: skip. + clock.Advance(30 * time.Minute) + b.KickOffNew(context.Background(), []string{"bucketA"}) + time.Sleep(20 * time.Millisecond) + if got := atomic.LoadInt32(&client.listedN); got != firstCount { + t.Fatalf("inside interval, must not re-walk; listedN=%d, want %d", got, firstCount) + } + + // Past the interval: re-walk. + clock.Advance(45 * time.Minute) // total elapsed > 1h + b.KickOffNew(context.Background(), []string{"bucketA"}) + waitFor(t, func() bool { return atomic.LoadInt32(&client.listedN) >= firstCount+2 }, "re-walk after interval") +} + +func TestBucketBootstrapper_KickOffNew_ZeroIntervalLegacyOnceOnly(t *testing.T) { + // BootstrapInterval == 0 preserves the original "walk once per + // process" behavior so existing deployments don't get a different + // cadence by default. + client := newEmptyFilerClient() + inj := &recordingInjector{} + clock := newFakeClock() + b := &BucketBootstrapper{ + FilerClient: client, + BucketsPath: "/buckets", + Injector: inj, + Now: clock.Now, + } + + b.KickOffNew(context.Background(), []string{"bucketA"}) + waitForCompleted(t, b, "bucketA") + firstCount := atomic.LoadInt32(&client.listedN) + + // Even after 100 hours, KickOffNew skips the bucket. + clock.Advance(100 * time.Hour) + b.KickOffNew(context.Background(), []string{"bucketA"}) + time.Sleep(20 * time.Millisecond) + if got := atomic.LoadInt32(&client.listedN); got != firstCount { + t.Fatalf("zero interval must keep the once-per-process behavior; listedN=%d, want %d", got, firstCount) + } +} + +// blockingInjector lets the test pin a walk in progress until it +// signals release. Useful for asserting in-flight debounce. +type blockingInjector struct { + mu sync.Mutex + events []*reader.Event + released chan struct{} +} + +func newBlockingInjector() *blockingInjector { + return &blockingInjector{released: make(chan struct{})} +} + +func (b *blockingInjector) InjectEvent(ctx context.Context, ev *reader.Event) error { + select { + case <-b.released: + case <-ctx.Done(): + return ctx.Err() + } + b.mu.Lock() + b.events = append(b.events, ev) + b.mu.Unlock() + return nil +} + +func (b *blockingInjector) release() { close(b.released) } + +func (b *blockingInjector) eventCount() int { + b.mu.Lock() + defer b.mu.Unlock() + return len(b.events) +} + +func TestBucketBootstrapper_KickOffNew_InFlightDebounceBlocksDuplicate(t *testing.T) { + // A walk that takes longer than BootstrapInterval would otherwise + // have a fresh KickOffNew start a duplicate goroutine on the next + // refresh tick. The inFlight set prevents that. Verify by: + // 1) Pinning a walk in progress via a blockingInjector, + // 2) Advancing the clock past BootstrapInterval, + // 3) Confirming the second KickOffNew is a no-op while the first + // is still running, + // 4) Releasing the first walk and asserting only one walk + // completed (one bucket-root listing pair). + client := &fakeFilerClient{ + tree: map[string][]*filer_pb.Entry{ + "/buckets/bucketA": {fileEntry("a.txt")}, + }, + } + inj := newBlockingInjector() + clock := newFakeClock() + b := &BucketBootstrapper{ + FilerClient: client, + BucketsPath: "/buckets", + Injector: inj, + BootstrapInterval: time.Hour, + Now: clock.Now, + } + + b.KickOffNew(context.Background(), []string{"bucketA"}) + // Wait for the walk goroutine to actually start listing (pass 1 + // fires a ListEntries before InjectEvent). + waitFor(t, func() bool { return atomic.LoadInt32(&client.listedN) >= 1 }, "first walk to begin listing") + + // Advance past the interval — a stale-state KickOffNew would now + // see the lastCompleted as expired and try again. + clock.Advance(2 * time.Hour) + b.KickOffNew(context.Background(), []string{"bucketA"}) + time.Sleep(20 * time.Millisecond) + if got := inj.eventCount(); got != 0 { + t.Fatalf("first walk still blocked, got %d injected events from a phantom second walk", got) + } + + // Release: the first walk completes. eventCount goes to 1. + inj.release() + waitFor(t, func() bool { return inj.eventCount() == 1 }, "first walk to drain") + // Even after another KickOffNew at the same simulated time, the + // in-flight is now cleared and lastCompleted is fresh — second + // KickOffNew within interval is a no-op. + prevListed := atomic.LoadInt32(&client.listedN) + b.KickOffNew(context.Background(), []string{"bucketA"}) + time.Sleep(20 * time.Millisecond) + if got := atomic.LoadInt32(&client.listedN); got != prevListed { + t.Fatalf("post-release KickOffNew within interval must be a no-op; listedN=%d, want %d", got, prevListed) + } +} + diff --git a/weed/s3api/s3lifecycle/scheduler/scheduler.go b/weed/s3api/s3lifecycle/scheduler/scheduler.go index c4571b49b..fc697cc94 100644 --- a/weed/s3api/s3lifecycle/scheduler/scheduler.go +++ b/weed/s3api/s3lifecycle/scheduler/scheduler.go @@ -33,11 +33,12 @@ type Scheduler struct { ClientID int32 ClientName string - Workers int - DispatchTick time.Duration - CheckpointTick time.Duration - RefreshInterval time.Duration - RetryBackoff time.Duration + Workers int + DispatchTick time.Duration + CheckpointTick time.Duration + RefreshInterval time.Duration + BootstrapInterval time.Duration + RetryBackoff time.Duration } // Run blocks until ctx is canceled. Spawns Workers + 1 goroutines: one @@ -100,9 +101,10 @@ func (s *Scheduler) Run(ctx context.Context) error { } bs := &BucketBootstrapper{ - FilerClient: s.FilerClient, - BucketsPath: s.BucketsPath, - Injector: pipelineFanout(pipelinesByShard), + FilerClient: s.FilerClient, + BucketsPath: s.BucketsPath, + Injector: pipelineFanout(pipelinesByShard), + BootstrapInterval: s.BootstrapInterval, } s.refreshEngine(ctx, bs) diff --git a/weed/shell/command_s3_lifecycle_run_shard.go b/weed/shell/command_s3_lifecycle_run_shard.go index ace7f489f..1f84b6b15 100644 --- a/weed/shell/command_s3_lifecycle_run_shard.go +++ b/weed/shell/command_s3_lifecycle_run_shard.go @@ -68,6 +68,7 @@ func (c *commandS3LifecycleRunShard) Do(args []string, env *CommandEnv, writer i dispatchTick := fs.Duration("dispatch", 5*time.Second, "dispatcher tick cadence") checkpointTick := fs.Duration("checkpoint", 30*time.Second, "cursor checkpoint cadence") refreshInterval := fs.Duration("refresh", 5*time.Minute, "interval for rebuilding the engine snapshot from filer-backed bucket configs; 0 = compile once at startup") + bootstrapInterval := fs.Duration("bootstrap-interval", 0, "cadence for revisiting each bucket's bootstrap walk; 0 = walk once per process. scan_only actions only fire from bootstrap, so a long-running worker needs a non-zero value to handle their retention horizon") runtime := fs.Duration("runtime", 0, "wall-clock cap on the run; 0 = no timeout. -events alone can hang on quiet shards") if err := fs.Parse(args); err != nil { return err @@ -119,9 +120,10 @@ func (c *commandS3LifecycleRunShard) Do(args []string, env *CommandEnv, writer i } bsr := &scheduler.BucketBootstrapper{ - FilerClient: filerClient, - BucketsPath: bucketsPath, - Injector: pipeline, + FilerClient: filerClient, + BucketsPath: bucketsPath, + Injector: pipeline, + BootstrapInterval: *bootstrapInterval, } bootstrapCtx, bootstrapCancel := context.WithCancel(context.Background()) defer bootstrapCancel() diff --git a/weed/worker/tasks/s3_lifecycle/config.go b/weed/worker/tasks/s3_lifecycle/config.go index 07d8288f4..f39b8b06e 100644 --- a/weed/worker/tasks/s3_lifecycle/config.go +++ b/weed/worker/tasks/s3_lifecycle/config.go @@ -9,31 +9,34 @@ import ( const ( jobType = "s3_lifecycle" - defaultWorkers = 1 - defaultDispatchTickMinutes = int64(1) - defaultCheckpointTickSeconds = int64(30) - defaultRefreshIntervalMinutes = int64(5) - defaultMaxRuntimeMinutes = int64(60) + defaultWorkers = 1 + defaultDispatchTickMinutes = int64(1) + defaultCheckpointTickSeconds = int64(30) + defaultRefreshIntervalMinutes = int64(5) + defaultMaxRuntimeMinutes = int64(60) + defaultBootstrapIntervalMinutes = int64(0) // 0 = walk once per process ) // Config is the parsed AdminConfigForm + WorkerConfigForm view. type Config struct { - Workers int - DispatchTick time.Duration - CheckpointTick time.Duration - RefreshInterval time.Duration - MaxRuntime time.Duration + Workers int + DispatchTick time.Duration + CheckpointTick time.Duration + RefreshInterval time.Duration + BootstrapInterval time.Duration + MaxRuntime time.Duration } // ParseConfig pulls the lifecycle Handler config from the merged // admin+worker config values. Missing fields fall back to defaults. func ParseConfig(adminValues, workerValues map[string]*plugin_pb.ConfigValue) Config { cfg := Config{ - Workers: int(readInt64(adminValues, "workers", defaultWorkers)), - DispatchTick: time.Duration(readInt64(workerValues, "dispatch_tick_minutes", defaultDispatchTickMinutes)) * time.Minute, - CheckpointTick: time.Duration(readInt64(workerValues, "checkpoint_tick_seconds", defaultCheckpointTickSeconds)) * time.Second, - RefreshInterval: time.Duration(readInt64(workerValues, "refresh_interval_minutes", defaultRefreshIntervalMinutes)) * time.Minute, - MaxRuntime: time.Duration(readInt64(workerValues, "max_runtime_minutes", defaultMaxRuntimeMinutes)) * time.Minute, + Workers: int(readInt64(adminValues, "workers", defaultWorkers)), + DispatchTick: time.Duration(readInt64(workerValues, "dispatch_tick_minutes", defaultDispatchTickMinutes)) * time.Minute, + CheckpointTick: time.Duration(readInt64(workerValues, "checkpoint_tick_seconds", defaultCheckpointTickSeconds)) * time.Second, + RefreshInterval: time.Duration(readInt64(workerValues, "refresh_interval_minutes", defaultRefreshIntervalMinutes)) * time.Minute, + BootstrapInterval: time.Duration(readInt64(workerValues, "bootstrap_interval_minutes", defaultBootstrapIntervalMinutes)) * time.Minute, + MaxRuntime: time.Duration(readInt64(workerValues, "max_runtime_minutes", defaultMaxRuntimeMinutes)) * time.Minute, } if cfg.Workers <= 0 { cfg.Workers = defaultWorkers @@ -47,6 +50,13 @@ func ParseConfig(adminValues, workerValues map[string]*plugin_pb.ConfigValue) Co if cfg.RefreshInterval <= 0 { cfg.RefreshInterval = time.Duration(defaultRefreshIntervalMinutes) * time.Minute } + // BootstrapInterval is intentionally NOT clamped — zero means + // "walk once per process", which is the legacy default for any + // deployment that hasn't opted into a cadence yet. Negative values + // fall back to zero. + if cfg.BootstrapInterval < 0 { + cfg.BootstrapInterval = 0 + } if cfg.MaxRuntime <= 0 { cfg.MaxRuntime = time.Duration(defaultMaxRuntimeMinutes) * time.Minute } diff --git a/weed/worker/tasks/s3_lifecycle/config_test.go b/weed/worker/tasks/s3_lifecycle/config_test.go index d2afae189..840648dbd 100644 --- a/weed/worker/tasks/s3_lifecycle/config_test.go +++ b/weed/worker/tasks/s3_lifecycle/config_test.go @@ -24,6 +24,9 @@ func TestParseConfigDefaults(t *testing.T) { if cfg.MaxRuntime != 60*time.Minute { t.Errorf("MaxRuntime default=%v, want 60m", cfg.MaxRuntime) } + if cfg.BootstrapInterval != 0 { + t.Errorf("BootstrapInterval default=%v, want 0 (walk-once-per-process)", cfg.BootstrapInterval) + } } func TestParseConfigOverrides(t *testing.T) { @@ -31,10 +34,11 @@ func TestParseConfigOverrides(t *testing.T) { "workers": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 4}}, } worker := map[string]*plugin_pb.ConfigValue{ - "dispatch_tick_minutes": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 2}}, - "checkpoint_tick_seconds": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 15}}, - "refresh_interval_minutes": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 10}}, - "max_runtime_minutes": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 120}}, + "dispatch_tick_minutes": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 2}}, + "checkpoint_tick_seconds": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 15}}, + "refresh_interval_minutes": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 10}}, + "bootstrap_interval_minutes": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 120}}, + "max_runtime_minutes": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 120}}, } cfg := ParseConfig(admin, worker) if cfg.Workers != 4 { @@ -49,6 +53,9 @@ func TestParseConfigOverrides(t *testing.T) { if cfg.RefreshInterval != 10*time.Minute { t.Errorf("RefreshInterval=%v, want 10m", cfg.RefreshInterval) } + if cfg.BootstrapInterval != 120*time.Minute { + t.Errorf("BootstrapInterval=%v, want 120m", cfg.BootstrapInterval) + } if cfg.MaxRuntime != 120*time.Minute { t.Errorf("MaxRuntime=%v, want 120m", cfg.MaxRuntime) } @@ -56,10 +63,11 @@ func TestParseConfigOverrides(t *testing.T) { func TestParseConfigClampsZeroAndNegative(t *testing.T) { worker := map[string]*plugin_pb.ConfigValue{ - "dispatch_tick_minutes": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 0}}, - "checkpoint_tick_seconds": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 0}}, - "refresh_interval_minutes": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 0}}, - "max_runtime_minutes": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: -5}}, + "dispatch_tick_minutes": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 0}}, + "checkpoint_tick_seconds": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 0}}, + "refresh_interval_minutes": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 0}}, + "bootstrap_interval_minutes": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: -5}}, + "max_runtime_minutes": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: -5}}, } cfg := ParseConfig(nil, worker) if cfg.DispatchTick != 1*time.Minute { @@ -71,7 +79,23 @@ func TestParseConfigClampsZeroAndNegative(t *testing.T) { if cfg.RefreshInterval != 5*time.Minute { t.Errorf("zero RefreshInterval should clamp to default, got %v", cfg.RefreshInterval) } + if cfg.BootstrapInterval != 0 { + t.Errorf("negative BootstrapInterval should clamp to 0, got %v", cfg.BootstrapInterval) + } if cfg.MaxRuntime != 60*time.Minute { t.Errorf("negative MaxRuntime should clamp to default, got %v", cfg.MaxRuntime) } } + +func TestParseConfigBootstrapIntervalZeroIsLegacy(t *testing.T) { + // Explicit zero stays zero (walk-once-per-process). Existing + // deployments that don't set bootstrap_interval_minutes get the + // legacy behavior unchanged. + worker := map[string]*plugin_pb.ConfigValue{ + "bootstrap_interval_minutes": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 0}}, + } + cfg := ParseConfig(nil, worker) + if cfg.BootstrapInterval != 0 { + t.Errorf("zero BootstrapInterval must stay zero, got %v", cfg.BootstrapInterval) + } +} diff --git a/weed/worker/tasks/s3_lifecycle/handler.go b/weed/worker/tasks/s3_lifecycle/handler.go index 247616b77..d86f32f62 100644 --- a/weed/worker/tasks/s3_lifecycle/handler.go +++ b/weed/worker/tasks/s3_lifecycle/handler.go @@ -122,6 +122,14 @@ func (h *Handler) Descriptor() *plugin_pb.JobTypeDescriptor { Widget: plugin_pb.ConfigWidget_CONFIG_WIDGET_NUMBER, MinValue: &plugin_pb.ConfigValue{Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 1}}, }, + { + Name: "bootstrap_interval_minutes", + Label: "Bootstrap Re-walk Interval (minutes)", + Description: "How often each bucket is re-walked. scan_only rules — those whose retention horizon exceeds meta-log retention — only fire from the bootstrap walk, so a non-zero value is required to enforce them on a long-running worker. 0 keeps the legacy walk-once-per-process behavior.", + FieldType: plugin_pb.ConfigFieldType_CONFIG_FIELD_TYPE_INT64, + Widget: plugin_pb.ConfigWidget_CONFIG_WIDGET_NUMBER, + MinValue: &plugin_pb.ConfigValue{Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 0}}, + }, { Name: "max_runtime_minutes", Label: "Max Runtime (minutes)", @@ -134,10 +142,11 @@ func (h *Handler) Descriptor() *plugin_pb.JobTypeDescriptor { }, }, DefaultValues: map[string]*plugin_pb.ConfigValue{ - "dispatch_tick_minutes": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: defaultDispatchTickMinutes}}, - "checkpoint_tick_seconds": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: defaultCheckpointTickSeconds}}, - "refresh_interval_minutes": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: defaultRefreshIntervalMinutes}}, - "max_runtime_minutes": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: defaultMaxRuntimeMinutes}}, + "dispatch_tick_minutes": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: defaultDispatchTickMinutes}}, + "checkpoint_tick_seconds": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: defaultCheckpointTickSeconds}}, + "refresh_interval_minutes": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: defaultRefreshIntervalMinutes}}, + "bootstrap_interval_minutes": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: defaultBootstrapIntervalMinutes}}, + "max_runtime_minutes": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: defaultMaxRuntimeMinutes}}, }, }, AdminRuntimeDefaults: &plugin_pb.AdminRuntimeDefaults{ @@ -240,17 +249,18 @@ func (h *Handler) Execute(ctx context.Context, request *plugin_pb.ExecuteJobRequ rpc := s3_lifecycle_pb.NewSeaweedS3LifecycleInternalClient(s3Conn) sched := &scheduler.Scheduler{ - BucketsPath: bucketsPath, - Engine: engine.New(), - Persister: &dispatcher.FilerPersister{Store: dispatcher.NewFilerStoreClient(filerClient)}, - Client: lifecycleRPCAdapter{c: rpc}, - FilerClient: filerClient, - ClientID: util.RandomInt32(), - ClientName: "worker-s3-lifecycle", - Workers: cfg.Workers, - DispatchTick: cfg.DispatchTick, - CheckpointTick: cfg.CheckpointTick, - RefreshInterval: cfg.RefreshInterval, + BucketsPath: bucketsPath, + Engine: engine.New(), + Persister: &dispatcher.FilerPersister{Store: dispatcher.NewFilerStoreClient(filerClient)}, + Client: lifecycleRPCAdapter{c: rpc}, + FilerClient: filerClient, + ClientID: util.RandomInt32(), + ClientName: "worker-s3-lifecycle", + Workers: cfg.Workers, + DispatchTick: cfg.DispatchTick, + CheckpointTick: cfg.CheckpointTick, + RefreshInterval: cfg.RefreshInterval, + BootstrapInterval: cfg.BootstrapInterval, } if err := sched.Run(runCtx); err != nil { glog.Warningf("s3 lifecycle execute: %v", err)