diff --git a/weed/worker/tasks/s3_lifecycle/cluster_rate_limit.go b/weed/worker/tasks/s3_lifecycle/cluster_rate_limit.go index d927f2b37..34866a38e 100644 --- a/weed/worker/tasks/s3_lifecycle/cluster_rate_limit.go +++ b/weed/worker/tasks/s3_lifecycle/cluster_rate_limit.go @@ -30,6 +30,18 @@ const ( // recovery branch on the next run. 0 = unbounded (current behavior, // falls back to maxTTL in runShard so PromotedHash stays empty). MetaLogRetentionDaysAdminKey = "meta_log_retention_days" + // WalkerIntervalMinutesAdminKey throttles the per-shard steady-state + // and empty-replay walker fires. dailyrun.runShard checks the time + // since the persisted Cursor.LastWalkedNs and skips the walk when + // less than this interval has elapsed; cold-start and recovery walker + // fires (RecoveryView) stay unconditional. 0 means "fire on every + // run" (the prior behavior — appropriate when the worker is driven + // at the operator's intended walk cadence, e.g. once per hour). + // Production deployments running the worker at multi-second cadence + // (CI ticks, sub-minute admin schedules) should set this to roughly + // the per-shard walk budget — typically 60 (1h) for small clusters, + // 360+ (6h+) for large ones. + WalkerIntervalMinutesAdminKey = "walker_interval_minutes" // MetadataKeyDeletesPerSecond is the per-worker share value the // admin writes into ClusterContext.Metadata at ExecuteJob time. diff --git a/weed/worker/tasks/s3_lifecycle/config.go b/weed/worker/tasks/s3_lifecycle/config.go index 6bf38dd11..a041194ee 100644 --- a/weed/worker/tasks/s3_lifecycle/config.go +++ b/weed/worker/tasks/s3_lifecycle/config.go @@ -19,6 +19,11 @@ type Config struct { Workers int MaxRuntime time.Duration MetaLogRetention time.Duration + // WalkerInterval is the minimum time between steady-state walker + // fires per shard. 0 means "fire on every run", preserving prior + // behavior; positive values gate the walker via Cursor.LastWalkedNs + // inside dailyrun.runShard. + WalkerInterval time.Duration } func ParseConfig(adminValues map[string]*plugin_pb.ConfigValue, workerValues map[string]*plugin_pb.ConfigValue) Config { @@ -36,6 +41,13 @@ func ParseConfig(adminValues map[string]*plugin_pb.ConfigValue, workerValues map if days := readInt64(adminValues, MetaLogRetentionDaysAdminKey, 0); days > 0 { cfg.MetaLogRetention = time.Duration(days*24) * time.Hour } + // Walker throttle. Negative / zero stay zero so dailyrun.runShard + // keeps the prior "fire every pass" semantics — important for in- + // repo integration tests and s3tests's sub-minute driver. Positive + // values throttle the steady-state walker per shard. + if mins := readInt64(adminValues, WalkerIntervalMinutesAdminKey, 0); mins > 0 { + cfg.WalkerInterval = time.Duration(mins) * time.Minute + } return cfg } diff --git a/weed/worker/tasks/s3_lifecycle/config_test.go b/weed/worker/tasks/s3_lifecycle/config_test.go index 287c9fb74..ae6bee00f 100644 --- a/weed/worker/tasks/s3_lifecycle/config_test.go +++ b/weed/worker/tasks/s3_lifecycle/config_test.go @@ -67,3 +67,37 @@ func TestParseConfigMetaLogRetentionNegativeStaysZero(t *testing.T) { t.Errorf("negative MetaLogRetention should stay 0, got %v", cfg.MetaLogRetention) } } + +func TestParseConfigWalkerIntervalDefaultsToZero(t *testing.T) { + // Unset key keeps WalkerInterval at 0 so dailyrun.runShard fires the + // walker every pass (the pre-throttle behavior the s3tests fast + // driver and the in-repo integration tests rely on). + cfg := ParseConfig(nil, nil) + if cfg.WalkerInterval != 0 { + t.Errorf("WalkerInterval default=%v, want 0", cfg.WalkerInterval) + } +} + +func TestParseConfigWalkerIntervalMinutesConvertsToDuration(t *testing.T) { + admin := map[string]*plugin_pb.ConfigValue{ + WalkerIntervalMinutesAdminKey: {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 90}}, + } + cfg := ParseConfig(admin, nil) + if want := 90 * time.Minute; cfg.WalkerInterval != want { + t.Errorf("WalkerInterval=%v, want %v", cfg.WalkerInterval, want) + } +} + +func TestParseConfigWalkerIntervalNegativeStaysZero(t *testing.T) { + // Negative declarations stay at 0 so the worker keeps "fire every + // pass" rather than treating the negative as past-due (which would + // fire every pass anyway — but via a less obvious code path that + // future readers would have to trace). + admin := map[string]*plugin_pb.ConfigValue{ + WalkerIntervalMinutesAdminKey: {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: -10}}, + } + cfg := ParseConfig(admin, nil) + if cfg.WalkerInterval != 0 { + t.Errorf("negative WalkerInterval should stay 0, got %v", cfg.WalkerInterval) + } +} diff --git a/weed/worker/tasks/s3_lifecycle/handler.go b/weed/worker/tasks/s3_lifecycle/handler.go index 947036036..3b4aad99f 100644 --- a/weed/worker/tasks/s3_lifecycle/handler.go +++ b/weed/worker/tasks/s3_lifecycle/handler.go @@ -99,6 +99,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: 0}}, }, + { + Name: WalkerIntervalMinutesAdminKey, + Label: "Walker Interval (minutes)", + Description: "Minimum time between steady-state walker fires per shard. Cold-start and rule-change recovery walks ignore this — they run unconditionally. 0 = fire on every run (use when the worker is scheduled at the desired walk cadence, e.g. hourly). Set to a positive value when the worker runs at a tighter cadence than the desired walk frequency, to avoid hammering filer with a full subtree scan per run.", + 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}}, + }, }, }, }, @@ -106,6 +114,7 @@ func (h *Handler) Descriptor() *plugin_pb.JobTypeDescriptor { ClusterDeletesPerSecondAdminKey: {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 0}}, ClusterDeletesBurstAdminKey: {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 0}}, MetaLogRetentionDaysAdminKey: {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 0}}, + WalkerIntervalMinutesAdminKey: {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 0}}, }, }, WorkerConfigForm: &plugin_pb.ConfigForm{ @@ -295,6 +304,7 @@ func (h *Handler) executeDailyReplay(ctx context.Context, request *plugin_pb.Exe Limiter: limiter, RetentionWindow: cfg.MetaLogRetention, Walker: walker, + WalkerInterval: cfg.WalkerInterval, ClientName: "worker-s3-lifecycle-daily", }) if runErr != nil {