From 495632730cbddc94cbd29e2a81ad1c9bfababc27 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Tue, 12 May 2026 12:15:20 -0700 Subject: [PATCH] =?UTF-8?q?feat(s3/lifecycle):=20daily-replay=20observabil?= =?UTF-8?q?ity=20=E2=80=94=20metrics=20+=20summary=20log=20(Phase=206)=20(?= =?UTF-8?q?#9462)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat(s3/lifecycle): daily-replay observability metrics + per-run summary log Operators have no Prometheus signal today for the daily_replay path beyond the cluster-rate-limiter wait histogram. Phase 6 adds the three baseline questions: how long does a shard take, how many events did it scan, and what did dispatch produce. - S3LifecycleDailyRunShardDurationSeconds (histogram, label=shard): wall-clock per shard. p95 climbing toward MaxRuntime means the shard is brushing its budget. - S3LifecycleDailyRunEventsScanned (counter, label=shard): meta-log events drainShardEvents processed. Pairs with the duration so a spike in events-per-shard correlates with a slow shard. - S3LifecycleDispatchCounter (existing, reused): processMatches now increments this with the outcome label, so streaming and daily_replay paths share one outcome view. Transport errors are counted under outcome="TRANSPORT_ERROR". dailyrun.Run logs a per-run summary at V(0): status / shards / errors / duration. The summary is the at-a-glance line operators read in /var/log to confirm a run completed. Test pins the dispatch-counter increment with a unique bucket/kind/outcome triple so a refactor that drops the instrumentation call surfaces as a test failure. * fix(s3/lifecycle): align dispatch error label + clean test labels Two PR-9462 review fixes from gemini: 1. processMatches' transport-failure label was "TRANSPORT_ERROR"; streaming's dispatcher uses "RPC_ERROR" for the same condition (see dispatcher/dispatcher.go). Use "RPC_ERROR" here too so the same Prometheus query covers both delete paths. 2. The dispatch-counter assertion test now deletes its label row on exit so the in-process Prometheus registry doesn't accumulate per-test state across the suite. --- .../dailyrun/process_matches_test.go | 39 +++++++++++++++++++ weed/s3api/s3lifecycle/dailyrun/run.go | 21 ++++++++++ weed/stats/metrics.go | 19 +++++++++ 3 files changed, 79 insertions(+) diff --git a/weed/s3api/s3lifecycle/dailyrun/process_matches_test.go b/weed/s3api/s3lifecycle/dailyrun/process_matches_test.go index 7ac21943f..20cd91c96 100644 --- a/weed/s3api/s3lifecycle/dailyrun/process_matches_test.go +++ b/weed/s3api/s3lifecycle/dailyrun/process_matches_test.go @@ -11,6 +11,8 @@ import ( "github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle" "github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/reader" "github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/router" + "github.com/seaweedfs/seaweedfs/weed/stats" + dto "github.com/prometheus/client_model/go" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) @@ -150,3 +152,40 @@ func TestProcessMatches_AllDueNoSkippedFlag(t *testing.T) { require.False(t, halted) require.False(t, skipped, "every match was past DueTime; skippedAny must be false") } + +func TestProcessMatches_DispatchCounterIncrements(t *testing.T) { + // Pin that dispatched matches increment S3LifecycleDispatchCounter + // with the outcome label so a refactor doesn't silently drop the + // observability hook. Use a bucket/kind no other test touches to + // keep the read-after-write stable. + runNow := time.Now() + rh := [8]byte{0xfe} + rule := s3lifecycle.ActionKey{Bucket: "metrics-pin-bkt", RuleHash: rh, ActionKind: s3lifecycle.ActionKindExpirationDays} + matches := []router.Match{ + {Key: rule, Bucket: "metrics-pin-bkt", ObjectKey: "obj", DueTime: runNow.Add(-time.Hour)}, + } + client := &recordingClient{} // default DONE + before := dispatchCounterValue("metrics-pin-bkt", "expiration_days", "DONE") + // Delete the label row on exit so this test doesn't leak into the + // in-process Prometheus registry that other tests share. + defer stats.S3LifecycleDispatchCounter.DeleteLabelValues("metrics-pin-bkt", "expiration_days", "DONE") + cfg := Config{Client: client} + _, _, err := processMatches(context.Background(), cfg, runNow, &reader.Event{}, matches) + require.NoError(t, err) + after := dispatchCounterValue("metrics-pin-bkt", "expiration_days", "DONE") + assert.Equal(t, before+1, after, "DONE outcome must increment the dispatch counter") +} + +// dispatchCounterValue reads the current value of the shared +// S3LifecycleDispatchCounter for the given (bucket, kind, outcome). +func dispatchCounterValue(bucket, kind, outcome string) float64 { + m := stats.S3LifecycleDispatchCounter.WithLabelValues(bucket, kind, outcome) + var pm dto.Metric + if err := m.Write(&pm); err != nil { + return 0 + } + if pm.Counter == nil { + return 0 + } + return pm.Counter.GetValue() +} diff --git a/weed/s3api/s3lifecycle/dailyrun/run.go b/weed/s3api/s3lifecycle/dailyrun/run.go index d334fc8fc..2ac02ecb6 100644 --- a/weed/s3api/s3lifecycle/dailyrun/run.go +++ b/weed/s3api/s3lifecycle/dailyrun/run.go @@ -4,6 +4,7 @@ import ( "context" "errors" "fmt" + "strconv" "sync" "time" @@ -86,6 +87,7 @@ func Run(ctx context.Context, cfg Config) error { } // Freeze "now" so shards agree on the boundary. runNow := now() + startedAt := time.Now() // Capture once so a mid-run Compile can't make shards disagree. snap := cfg.Engine.Snapshot() @@ -116,13 +118,21 @@ func Run(ctx context.Context, cfg Config) error { wg.Wait() close(errCh) var first error + errCount := 0 for err := range errCh { + errCount++ if first == nil { first = err } else { glog.V(1).Infof("daily_run: additional shard error: %v", err) } } + status := "ok" + if first != nil { + status = "error" + } + glog.V(0).Infof("daily_run: status=%s shards=%d errors=%d duration=%s", + status, len(cfg.Shards), errCount, time.Since(startedAt).Round(time.Millisecond)) return first } @@ -161,6 +171,11 @@ func validate(cfg Config) error { // scan_only-promoted rules fire every day even when replay rules // are unchanged. func runShard(ctx context.Context, cfg Config, snap *engine.Snapshot, runNow time.Time, shardID int) error { + shardLabel := strconv.Itoa(shardID) + shardStart := time.Now() + defer func() { + stats.S3LifecycleDailyRunShardDurationSeconds.WithLabelValues(shardLabel).Observe(time.Since(shardStart).Seconds()) + }() persisted, found, err := cfg.Persister.Load(ctx, shardID) if err != nil { return fmt.Errorf("shard=%d: load cursor: %w", shardID, err) @@ -307,6 +322,7 @@ drain: cancelReader() break drain } + stats.S3LifecycleDailyRunEventsScanned.WithLabelValues(strconv.Itoa(shardID)).Inc() matches := router.Route(ctx, snap, ev, runNow, cfg.Lister) eventSkipped, eventHalted, eventErr := processMatches(ctx, cfg, runNow, ev, matches) if eventErr != nil { @@ -357,8 +373,13 @@ func processMatches(ctx context.Context, cfg Config, runNow time.Time, ev *reade if dispatchErr != nil { glog.V(1).Infof("daily_run: transport error on %s/%s %s: %v", m.Bucket, m.ObjectKey, m.Key.ActionKind, dispatchErr) + // "RPC_ERROR" matches the streaming dispatcher's label + // (dispatcher/dispatcher.go) so transport failures + // aggregate under one outcome key across paths. + stats.S3LifecycleDispatchCounter.WithLabelValues(m.Bucket, m.Key.ActionKind.String(), "RPC_ERROR").Inc() return skippedAny, true, nil } + stats.S3LifecycleDispatchCounter.WithLabelValues(m.Bucket, m.Key.ActionKind.String(), outcome.String()).Inc() switch outcome { case s3_lifecycle_pb.LifecycleDeleteOutcome_DONE, s3_lifecycle_pb.LifecycleDeleteOutcome_NOOP_RESOLVED, diff --git a/weed/stats/metrics.go b/weed/stats/metrics.go index b225db356..2083d78f3 100644 --- a/weed/stats/metrics.go +++ b/weed/stats/metrics.go @@ -608,6 +608,23 @@ var ( Help: "Time spent waiting on the cluster rate limiter before issuing a LifecycleDelete RPC. Non-zero values indicate the cluster cap is binding.", Buckets: []float64{0.0001, 0.001, 0.01, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10}, }) + + S3LifecycleDailyRunShardDurationSeconds = prometheus.NewHistogramVec( + prometheus.HistogramOpts{ + Namespace: Namespace, + Subsystem: "s3_lifecycle", + Name: "daily_run_shard_duration_seconds", + Help: "Wall-clock seconds spent in one shard's daily_replay pass. p95 climbing toward MaxRuntime means the shard is brushing its budget.", + Buckets: []float64{0.1, 0.5, 1, 5, 15, 60, 300, 900, 1800, 3600}, + }, []string{"shard"}) + + S3LifecycleDailyRunEventsScanned = prometheus.NewCounterVec( + prometheus.CounterOpts{ + Namespace: Namespace, + Subsystem: "s3_lifecycle", + Name: "daily_run_events_scanned_total", + Help: "Counter of meta-log events drainShardEvents processed on the daily_replay path, partitioned by shard.", + }, []string{"shard"}) ) func init() { @@ -685,6 +702,8 @@ func init() { Gather.MustRegister(S3LifecycleBootstrapDispatchCounter) Gather.MustRegister(S3LifecycleMetadataOnlyCounter) Gather.MustRegister(S3LifecycleDispatchLimiterWaitSeconds) + Gather.MustRegister(S3LifecycleDailyRunShardDurationSeconds) + Gather.MustRegister(S3LifecycleDailyRunEventsScanned) Gather.MustRegister(UploadErrorCounter)