From d57015c995b3e9fd9755be5cef239341dab62af2 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Mon, 11 May 2026 19:27:55 -0700 Subject: [PATCH] Revert "chore(s3/lifecycle): trim verbose comments" This reverts commit 11b98ee8738cbdafbe731361aa72f5568594bfbe. --- weed/admin/plugin/cluster_rate_limit.go | 62 ++++++++++++++++--- weed/admin/plugin/cluster_rate_limit_test.go | 42 ++++++++++--- weed/admin/plugin/plugin.go | 6 ++ weed/admin/plugin/registry.go | 6 +- weed/stats/metrics.go | 7 +++ .../tasks/s3_lifecycle/cluster_rate_limit.go | 34 +++++++--- .../s3_lifecycle/cluster_rate_limit_test.go | 10 ++- weed/worker/tasks/s3_lifecycle/config.go | 24 ++++++- weed/worker/tasks/s3_lifecycle/handler.go | 28 +++++++-- 9 files changed, 179 insertions(+), 40 deletions(-) diff --git a/weed/admin/plugin/cluster_rate_limit.go b/weed/admin/plugin/cluster_rate_limit.go index 14519c413..e1802a5f7 100644 --- a/weed/admin/plugin/cluster_rate_limit.go +++ b/weed/admin/plugin/cluster_rate_limit.go @@ -7,9 +7,23 @@ import ( "github.com/seaweedfs/seaweedfs/weed/pb/plugin_pb" ) -// String constants must match weed/worker/tasks/s3_lifecycle/cluster_rate_limit.go. -// Duplicated rather than imported so the admin package doesn't depend -// on the worker handler package; tests pin both sides to the same values. +// Job types whose ExecuteJobRequest needs per-worker rate-allocation +// metadata injected. Keyed by the job-type string so plugin.go's +// generic dispatch path stays job-agnostic. +// +// To add a new job type to the share-allocation pipeline: register an +// entry here that knows how to read its admin-config field(s) and +// produce the metadata keys/values the worker reads. + +// s3LifecycleClusterDeletesPerSecondKey, s3LifecycleClusterDeletesBurstKey, +// s3LifecycleMetadataDeletesPerSecond, and s3LifecycleMetadataDeletesBurst +// are the contract between admin and worker. The values must match the +// constants exported from weed/worker/tasks/s3_lifecycle/cluster_rate_limit.go +// — duplicated here as plain strings rather than imported so the admin +// plugin package doesn't pull a dependency on the worker handler +// package. A mismatch on either side would silently disable rate +// limiting; tests pin the constants in both packages against the same +// values. const ( s3LifecycleJobType = "s3_lifecycle" s3LifecycleClusterDeletesPerSecondKey = "cluster_deletes_per_second" @@ -18,10 +32,23 @@ const ( s3LifecycleMetadataDeletesBurst = "s3_lifecycle.deletes_burst" ) -// decorateClusterContextForJob injects per-job rate-allocation metadata -// when the job type opts in. Divisor is min(executors, maxJobsPerDetection) -// so singleton jobs (maxJobs=1) get the full cluster budget on the -// single active worker. +// decorateClusterContextForJob returns a new ClusterContext with any +// per-job-type metadata the admin needs to inject before the +// ExecuteJobRequest is sent. Returns the input cc unchanged when no +// decoration applies. +// +// Today only s3_lifecycle decorates; the function exists so a future +// job type's plumbing slots in alongside without touching +// executeJobWithExecutor. +// +// maxJobsPerDetection is the job-type's AdminRuntimeConfig.MaxJobsPerDetection +// — the cap on how many parallel instances of this job the scheduler will +// dispatch per detection cycle. For singleton jobs (s3_lifecycle has +// MaxJobsPerDetection=1) only one worker is ever active, so the cluster +// budget must go to that worker undivided. For parallel-dispatch jobs the +// budget divides across the actually-running set, not across every +// capable worker. The divisor is min(executors, maxJobsPerDetection), +// clamped to ≥1. func (r *Plugin) decorateClusterContextForJob(cc *plugin_pb.ClusterContext, jobType string, adminConfigValues map[string]*plugin_pb.ConfigValue, maxJobsPerDetection int) *plugin_pb.ClusterContext { if cc == nil { return cc @@ -32,13 +59,21 @@ func (r *Plugin) decorateClusterContextForJob(cc *plugin_pb.ClusterContext, jobT rps := readNonNegativeInt(adminConfigValues, s3LifecycleClusterDeletesPerSecondKey) burst := readNonNegativeInt(adminConfigValues, s3LifecycleClusterDeletesBurstKey) if rps <= 0 { + // Operator hasn't configured a cluster cap; nothing to allocate. + // The worker treats missing metadata keys as "unlimited," which + // is the legacy behavior. return cc } executors := r.registry.CountCapableExecutors(jobType) if executors <= 0 { + // No executors means the job won't dispatch at all; metadata + // would be discarded. Log so the case is visible in ops. glog.V(2).Infof("decorateClusterContext: %s rps=%d but no execute-capable workers; skipping allocation", jobType, rps) return cc } + // Divide by the number of *concurrently-running* workers, not the + // number of capable ones. A singleton job (maxJobs=1) gets the full + // budget on its single active worker. activeWorkers := executors if maxJobsPerDetection > 0 && maxJobsPerDetection < activeWorkers { activeWorkers = maxJobsPerDetection @@ -48,11 +83,12 @@ func (r *Plugin) decorateClusterContextForJob(cc *plugin_pb.ClusterContext, jobT if burst > 0 { perWorkerBurst = burst / activeWorkers if perWorkerBurst < 1 { - // rate.Limiter with burst<1 never refills; floor. perWorkerBurst = 1 } } + // Clone so we don't mutate the shared base context. The metadata + // map is small; a fresh allocation per ExecuteJob is fine. out := cloneClusterContext(cc) if out.Metadata == nil { out.Metadata = map[string]string{} @@ -66,8 +102,11 @@ func (r *Plugin) decorateClusterContextForJob(cc *plugin_pb.ClusterContext, jobT return out } -// cloneClusterContext duplicates the Metadata map so callers can mutate -// it without racing the shared base context. Address slices are aliased. +// cloneClusterContext returns a shallow-but-safe copy: the top-level +// fields are reassigned, and the Metadata map is duplicated so the +// caller can mutate it without racing other consumers of the input. +// Slices of strings (master/filer/volume/s3 addresses) are copied by +// reference — those are treated as immutable elsewhere in the codebase. func cloneClusterContext(in *plugin_pb.ClusterContext) *plugin_pb.ClusterContext { if in == nil { return nil @@ -87,6 +126,9 @@ func cloneClusterContext(in *plugin_pb.ClusterContext) *plugin_pb.ClusterContext return out } +// readNonNegativeInt reads an int64 admin config value, treating +// missing fields and non-int kinds as 0. Negative values are clamped +// to 0 since the AdminConfigForm declares MinValue=0 on both fields. func readNonNegativeInt(values map[string]*plugin_pb.ConfigValue, field string) int { v, ok := values[field] if !ok || v == nil { diff --git a/weed/admin/plugin/cluster_rate_limit_test.go b/weed/admin/plugin/cluster_rate_limit_test.go index ac3841f60..9a336db40 100644 --- a/weed/admin/plugin/cluster_rate_limit_test.go +++ b/weed/admin/plugin/cluster_rate_limit_test.go @@ -9,8 +9,10 @@ import ( "github.com/stretchr/testify/require" ) -// pluginWithExecutors builds a Plugin with n execute-capable workers -// for jobType, bypassing UpsertFromHello. +// pluginWithExecutors returns a Plugin whose registry contains n +// non-stale execute-capable workers for jobType. Helper for the +// allocator tests. Bypasses UpsertFromHello so tests don't have to +// build a full Hello message. func pluginWithExecutors(t *testing.T, jobType string, n int) *Plugin { t.Helper() reg := NewRegistry() @@ -29,6 +31,7 @@ func pluginWithExecutors(t *testing.T, jobType string, n int) *Plugin { return &Plugin{registry: reg} } +// adminConfig builds an int64 admin config map for the given fields. func adminConfig(pairs ...interface{}) map[string]*plugin_pb.ConfigValue { if len(pairs)%2 != 0 { panic("adminConfig expects key/value pairs") @@ -51,6 +54,9 @@ func adminConfig(pairs ...interface{}) map[string]*plugin_pb.ConfigValue { } func TestDecorateClusterContext_NonS3LifecycleIsPassThrough(t *testing.T) { + // Any job type other than s3_lifecycle gets the input cc back + // unchanged. Future allocators add their own branch; the default + // is pass-through. r := pluginWithExecutors(t, s3LifecycleJobType, 4) in := &plugin_pb.ClusterContext{Metadata: map[string]string{"unrelated": "v"}} out := r.decorateClusterContextForJob(in, "some_other_job", adminConfig(s3LifecycleClusterDeletesPerSecondKey, 100), 1) @@ -58,7 +64,10 @@ func TestDecorateClusterContext_NonS3LifecycleIsPassThrough(t *testing.T) { } func TestDecorateClusterContext_RpsZeroSkipsAllocation(t *testing.T) { - // rps=0 must NOT write "0" — worker would read it as zero-throughput. + // rps=0 means "operator hasn't configured a cap"; the worker + // treats missing keys as unlimited. We must NOT inject any + // metadata (in particular, not "0") because that would force the + // worker into a no-throughput state on a misconfigured cluster. r := pluginWithExecutors(t, s3LifecycleJobType, 4) in := &plugin_pb.ClusterContext{} out := r.decorateClusterContextForJob(in, s3LifecycleJobType, adminConfig(s3LifecycleClusterDeletesPerSecondKey, 0), 1) @@ -79,26 +88,34 @@ func TestDecorateClusterContext_NoExecutorsSkipsAllocation(t *testing.T) { } func TestDecorateClusterContext_SingletonJobGetsFullBudget(t *testing.T) { - // maxJobs=1: budget must go undivided to the single active worker. + // s3_lifecycle has MaxJobsPerDetection=1: only ONE worker runs the + // job at a time. The cluster budget must go to that worker undivided + // — dividing by N capable executors would starve the active worker + // to 1/N of the configured rps. Pin the singleton behavior. r := pluginWithExecutors(t, s3LifecycleJobType, 4) in := &plugin_pb.ClusterContext{} out := r.decorateClusterContextForJob(in, s3LifecycleJobType, adminConfig(s3LifecycleClusterDeletesPerSecondKey, 100), 1) require.NotNil(t, out.Metadata) - assert.Equal(t, "100", out.Metadata[s3LifecycleMetadataDeletesPerSecond]) + assert.Equal(t, "100", out.Metadata[s3LifecycleMetadataDeletesPerSecond], "singleton job: full budget to the single active worker") } func TestDecorateClusterContext_SharedEvenlyWhenParallelLimited(t *testing.T) { + // Hypothetical parallel-dispatch job type (maxJobs=4): budget + // divides across the running-set, which equals min(executors, + // maxJobs)=4. 100/4=25. r := pluginWithExecutors(t, s3LifecycleJobType, 4) in := &plugin_pb.ClusterContext{} out := r.decorateClusterContextForJob(in, s3LifecycleJobType, adminConfig(s3LifecycleClusterDeletesPerSecondKey, 100), 4) require.NotNil(t, out.Metadata) - assert.Equal(t, "25", out.Metadata[s3LifecycleMetadataDeletesPerSecond]) + assert.Equal(t, "25", out.Metadata[s3LifecycleMetadataDeletesPerSecond], "maxJobs=4 across 4 executors = 25/worker") } func TestDecorateClusterContext_MaxJobsExceedsExecutors(t *testing.T) { - // divisor = min(executors=4, maxJobs=10). + // maxJobs=10 but only 4 executors exist — the divisor is the + // smaller value (executors) since you can't run more jobs in + // parallel than there are workers to run them. r := pluginWithExecutors(t, s3LifecycleJobType, 4) in := &plugin_pb.ClusterContext{} out := r.decorateClusterContextForJob(in, s3LifecycleJobType, @@ -118,7 +135,9 @@ func TestDecorateClusterContext_BurstSharedWhenParallel(t *testing.T) { } func TestDecorateClusterContext_BurstZeroOmitsKey(t *testing.T) { - // burst=0 means "let the worker default to 2*rps" — omit the key. + // burst=0 means "let the worker default it." Don't write the key — + // the worker's parsePositiveInt would then take the unset path + // and compute 2 × rps automatically. r := pluginWithExecutors(t, s3LifecycleJobType, 4) in := &plugin_pb.ClusterContext{} out := r.decorateClusterContextForJob(in, s3LifecycleJobType, @@ -128,7 +147,8 @@ func TestDecorateClusterContext_BurstZeroOmitsKey(t *testing.T) { } func TestDecorateClusterContext_BurstFloorIsOneWhenDividesBelowOne(t *testing.T) { - // rate.Limiter with burst<1 never refills. + // burst=1 across 4 active workers would round to 0; clamp to 1 so + // the limiter doesn't become "single-token bucket that never refills." r := pluginWithExecutors(t, s3LifecycleJobType, 4) in := &plugin_pb.ClusterContext{} out := r.decorateClusterContextForJob(in, s3LifecycleJobType, @@ -137,7 +157,9 @@ func TestDecorateClusterContext_BurstFloorIsOneWhenDividesBelowOne(t *testing.T) } func TestDecorateClusterContext_DoesNotMutateInput(t *testing.T) { - // Base ClusterContext is shared across parallel ExecuteJob calls. + // The same base ClusterContext is shared across many parallel + // ExecuteJob calls. The decorator must produce a fresh map so it + // can't race / leak per-job metadata into the base. r := pluginWithExecutors(t, s3LifecycleJobType, 4) baseMeta := map[string]string{"existing": "value"} in := &plugin_pb.ClusterContext{Metadata: baseMeta} diff --git a/weed/admin/plugin/plugin.go b/weed/admin/plugin/plugin.go index e65c642de..d50d7d2a3 100644 --- a/weed/admin/plugin/plugin.go +++ b/weed/admin/plugin/plugin.go @@ -649,6 +649,12 @@ func (r *Plugin) executeJobWithExecutor( return nil, err } + // Apply per-job-type cluster-allocation decoration (e.g. s3_lifecycle + // divides cluster_deletes_per_second by min(workers, maxJobsPerDetection) + // and ships the share via ClusterContext.Metadata). No-op for job + // types without an allocator registered. MaxJobsPerDetection caps + // the divisor so a singleton job (maxJobs=1) gets the full budget on + // the single active worker, not 1/N of it. clusterContext = r.decorateClusterContextForJob(clusterContext, job.JobType, adminConfigValues, int(adminRuntime.GetMaxJobsPerDetection())) completedCh := make(chan *plugin_pb.JobCompleted, 1) diff --git a/weed/admin/plugin/registry.go b/weed/admin/plugin/registry.go index 0ab9cb092..46d51e79d 100644 --- a/weed/admin/plugin/registry.go +++ b/weed/admin/plugin/registry.go @@ -143,7 +143,11 @@ func (r *Registry) HasCapableWorker(jobType string) bool { } // CountCapableExecutors returns the number of non-stale workers that -// can EXECUTE the given job type. +// can EXECUTE the given job type. Used by per-job-type cluster +// allocators (e.g. the s3_lifecycle delete-rate divider) to compute a +// per-worker share at dispatch time. Returns 0 when no executor is +// available — callers should treat that as "skip allocation" rather +// than dividing by zero. func (r *Registry) CountCapableExecutors(jobType string) int { r.mu.RLock() defer r.mu.RUnlock() diff --git a/weed/stats/metrics.go b/weed/stats/metrics.go index 094a492f6..b225db356 100644 --- a/weed/stats/metrics.go +++ b/weed/stats/metrics.go @@ -593,6 +593,13 @@ var ( Help: "Counter of LifecycleDelete completions that skipped per-chunk delete (volume TTL reclaim).", }, []string{"bucket", "rule_hash"}) + // S3LifecycleDispatchLimiterWaitSeconds is the cluster-wide rate + // limiter's per-dispatch wait time on the daily-replay path. The + // limiter blocks just before each LifecycleDelete RPC; near-zero + // observations mean the cluster cap isn't binding, a long-tail at + // the configured 1/rate ceiling means the cluster cap is the + // active throttle. Operators tune cluster_deletes_per_second by + // reading p95/p99 on this histogram. S3LifecycleDispatchLimiterWaitSeconds = prometheus.NewHistogram( prometheus.HistogramOpts{ Namespace: Namespace, diff --git a/weed/worker/tasks/s3_lifecycle/cluster_rate_limit.go b/weed/worker/tasks/s3_lifecycle/cluster_rate_limit.go index 784f3e832..1e92dd6df 100644 --- a/weed/worker/tasks/s3_lifecycle/cluster_rate_limit.go +++ b/weed/worker/tasks/s3_lifecycle/cluster_rate_limit.go @@ -1,18 +1,34 @@ package s3_lifecycle -// Contract with weed/admin/plugin/cluster_rate_limit.go: admin computes -// the per-worker share from the *AdminKey fields and writes it to -// ExecuteJobRequest.ClusterContext.Metadata under the MetadataKey* keys. -// Changing a name on one side without the other silently disables rate -// limiting. +// Cluster-wide rate-limit configuration plumbing for the daily-replay +// worker. The admin holds a single "cluster delete budget" knob, divides +// it by the number of execute-capable s3_lifecycle workers at job-dispatch +// time, and ships the per-worker share to the worker via +// ExecuteJobRequest.ClusterContext.Metadata. The worker reads the share, +// constructs a rate.Limiter, and passes it to dailyrun.Run. +// +// These constants are the contract between admin (weed/admin/plugin/plugin.go +// computes the share and writes the keys) and worker (this package's +// handler.go reads them). Changing a name on one side without the other +// would silently disable rate limiting — both sides must read these +// exact values. const ( - // 0 = unlimited. + // ClusterDeletesPerSecondAdminKey is the admin-config field that + // holds the cluster-wide budget in delete RPCs per second. 0 means + // unlimited (legacy behavior). Set via the AdminConfigForm in + // handler.go's "Scope" section. ClusterDeletesPerSecondAdminKey = "cluster_deletes_per_second" - // 0 = 2 * rps. + // ClusterDeletesBurstAdminKey holds the token-bucket burst. 0 means + // "2 × rps" (computed by the admin allocator). ClusterDeletesBurstAdminKey = "cluster_deletes_burst" - // Per-worker share. Missing/empty/zero -> cfg.Limiter stays nil. + // MetadataKeyDeletesPerSecond is the per-worker share value the + // admin writes into ClusterContext.Metadata at ExecuteJob time. + // Stored as a string of a non-negative float64; empty/missing/zero + // means "no rate limit on this run" (cfg.Limiter stays nil). MetadataKeyDeletesPerSecond = "s3_lifecycle.deletes_per_second" - MetadataKeyDeletesBurst = "s3_lifecycle.deletes_burst" + // MetadataKeyDeletesBurst is the per-worker burst share. Stored as + // a string of a non-negative integer. + MetadataKeyDeletesBurst = "s3_lifecycle.deletes_burst" ) diff --git a/weed/worker/tasks/s3_lifecycle/cluster_rate_limit_test.go b/weed/worker/tasks/s3_lifecycle/cluster_rate_limit_test.go index 2c435a210..4b3c1b2a4 100644 --- a/weed/worker/tasks/s3_lifecycle/cluster_rate_limit_test.go +++ b/weed/worker/tasks/s3_lifecycle/cluster_rate_limit_test.go @@ -29,7 +29,9 @@ func TestBuildLimiterFromClusterContext_MissingRateKey(t *testing.T) { } func TestBuildLimiterFromClusterContext_NonPositiveRate(t *testing.T) { - // rate<=0 must yield nil, not a zero-throughput limiter. + // 0 or negative rate means the admin didn't allocate; the worker + // must NOT construct a limiter that throttles every request to + // zero-throughput. for _, raw := range []string{"0", "-1", "0.0", "not-a-number", ""} { l, desc := buildLimiterFromClusterContext(&plugin_pb.ClusterContext{ Metadata: map[string]string{MetadataKeyDeletesPerSecond: raw}, @@ -54,6 +56,8 @@ func TestBuildLimiterFromClusterContext_PositiveRateBuildsLimiter(t *testing.T) } func TestBuildLimiterFromClusterContext_BurstMissingDefaultsTo2xRate(t *testing.T) { + // burst omitted (admin allocator wrote nothing) → worker computes + // 2 × rps so a single tick has headroom for two refills. l, _ := buildLimiterFromClusterContext(&plugin_pb.ClusterContext{ Metadata: map[string]string{MetadataKeyDeletesPerSecond: "10"}, }) @@ -63,7 +67,9 @@ func TestBuildLimiterFromClusterContext_BurstMissingDefaultsTo2xRate(t *testing. } func TestBuildLimiterFromClusterContext_TinyRateClampsBurstToOne(t *testing.T) { - // rate.Limiter with burst<1 never refills; floor. + // A sub-1-rps allocation (e.g. 0.5/s across few workers) would + // compute 2 × 0.5 = 1, but if the int truncation produced 0 the + // limiter would never refill. Clamp to at least 1. l, _ := buildLimiterFromClusterContext(&plugin_pb.ClusterContext{ Metadata: map[string]string{MetadataKeyDeletesPerSecond: "0.1"}, }) diff --git a/weed/worker/tasks/s3_lifecycle/config.go b/weed/worker/tasks/s3_lifecycle/config.go index 2f0717f69..ed3ec9f8c 100644 --- a/weed/worker/tasks/s3_lifecycle/config.go +++ b/weed/worker/tasks/s3_lifecycle/config.go @@ -9,7 +9,10 @@ import ( const ( jobType = "s3_lifecycle" - // In-process fan-out across the 16 shards. + // shardPipelineGoroutines is the in-process fan-out across the + // 16-shard space. Kept as a hardcoded internal default — formerly + // an admin form field, removed because it's a per-worker tuning + // knob, not a cluster-coordination concern. shardPipelineGoroutines = 1 defaultDispatchTickMinutes = int64(1) @@ -18,12 +21,21 @@ const ( defaultMaxRuntimeMinutes = int64(60) defaultBootstrapIntervalMinutes = int64(0) // 0 = walk once per process + // AlgorithmDailyReplay routes the worker through dailyrun.Run for + // one bounded pass per Execute. Currently Phase 2 / replay-only: + // buckets with walker-bound action kinds are refused. Phase 4 + // extends this to handle every kind. Default — the streaming path + // stays in the tree as a runtime escape hatch only. AlgorithmDailyReplay = "daily_replay" - AlgorithmStreaming = "streaming" + // AlgorithmStreaming is the legacy event-driven dispatcher path + // (reader + heap + per-shard pipeline). Kept as a fallback knob for + // rollout; deleted by Phase 5 once Phase 4 walker integration ships. + AlgorithmStreaming = "streaming" defaultAlgorithm = AlgorithmDailyReplay ) +// Config is the parsed AdminConfigForm + WorkerConfigForm view. type Config struct { Workers int DispatchTick time.Duration @@ -34,6 +46,8 @@ type Config struct { Algorithm string } +// 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: shardPipelineGoroutines, @@ -53,7 +67,10 @@ func ParseConfig(adminValues, workerValues map[string]*plugin_pb.ConfigValue) Co if cfg.RefreshInterval <= 0 { cfg.RefreshInterval = time.Duration(defaultRefreshIntervalMinutes) * time.Minute } - // Zero means walk-once-per-process; clamp negatives only. + // 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 } @@ -62,6 +79,7 @@ func ParseConfig(adminValues, workerValues map[string]*plugin_pb.ConfigValue) Co } switch cfg.Algorithm { case AlgorithmStreaming, AlgorithmDailyReplay: + // valid default: cfg.Algorithm = defaultAlgorithm } diff --git a/weed/worker/tasks/s3_lifecycle/handler.go b/weed/worker/tasks/s3_lifecycle/handler.go index 102e6cef6..98c2404fd 100644 --- a/weed/worker/tasks/s3_lifecycle/handler.go +++ b/weed/worker/tasks/s3_lifecycle/handler.go @@ -297,8 +297,10 @@ func (h *Handler) Execute(ctx context.Context, request *plugin_pb.ExecuteJobRequ return nil } -// executeDailyReplay is the bounded daily-replay path (Phase 2: -// replay-only, refuses walker-bound action kinds). +// executeDailyReplay runs the bounded daily-replay path. Reuses the +// streaming path's filer / s3 / engine setup but routes the per-shard +// loop through dailyrun.Run instead of scheduler.Scheduler. Phase 2: +// replay-only, refuses walker-bound action kinds with a typed error. func (h *Handler) executeDailyReplay(ctx context.Context, request *plugin_pb.ExecuteJobRequest, bucketsPath string, filerClient filer_pb.SeaweedFilerClient, rpc s3_lifecycle_pb.SeaweedS3LifecycleInternalClient, cfg Config, sender pluginworker.ExecutionSender) error { eng := engine.New() inputs, parseErrors, err := scheduler.LoadCompileInputs(ctx, filerClient, bucketsPath) @@ -335,6 +337,12 @@ func (h *Handler) executeDailyReplay(ctx context.Context, request *plugin_pb.Exe Limiter: limiter, ClientName: "worker-s3-lifecycle-daily", }) + if dailyrun.IsUnsupportedRule(runErr) { + // Surface the typed error verbatim so admin marks the run as + // failed with the user-facing reason in the activity log. + glog.Warningf("daily_replay: %v", runErr) + return runErr + } if runErr != nil { glog.Warningf("daily_replay: %v", runErr) return runErr @@ -342,9 +350,17 @@ func (h *Handler) executeDailyReplay(ctx context.Context, request *plugin_pb.Exe return nil } -// buildLimiterFromClusterContext returns (nil, "unlimited") for any -// missing/malformed/non-positive value. The admin allocator owns the -// "should there be a cap" decision; the worker doesn't second-guess. +// buildLimiterFromClusterContext parses the per-worker share the admin +// wrote into ClusterContext.Metadata (see weed/admin/plugin/plugin.go's +// s3_lifecycle injection) and returns a rate.Limiter, or nil when no +// rate cap applies. The description string is for the JobProgressUpdate +// so operators can see "rate=unlimited" / "rate=12.5/s burst=25" in +// the activity log. +// +// Tolerant of missing keys, empty strings, malformed numbers, and +// non-positive values — all treated as "no limit" rather than failing +// the run. The admin allocator is the single point that decides whether +// to populate these keys; the worker doesn't second-guess. func buildLimiterFromClusterContext(cc *plugin_pb.ClusterContext) (*rate.Limiter, string) { if cc == nil || cc.Metadata == nil { return nil, "unlimited" @@ -355,6 +371,8 @@ func buildLimiterFromClusterContext(cc *plugin_pb.ClusterContext) (*rate.Limiter } burst, _ := parsePositiveInt(cc.Metadata[MetadataKeyDeletesBurst]) if burst <= 0 { + // Sensible default: enough headroom for one tick's worth of + // throughput. Caller may also supply 0 to opt into this default. burst = int(rps * 2) if burst < 1 { burst = 1