feat(s3/lifecycle): daily-replay observability — metrics + summary log (Phase 6) (#9462)

* 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.

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()
+}

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,

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)