diff --git a/other/metrics/grafana_seaweedfs.json b/other/metrics/grafana_seaweedfs.json index ee0b590d7c..b4e0547bf4 100644 --- a/other/metrics/grafana_seaweedfs.json +++ b/other/metrics/grafana_seaweedfs.json @@ -10263,6 +10263,102 @@ } ], "pluginVersion": "10.3.1" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "title": "Time Since Last Full Lifecycle Pass", + "description": "Time since a pass covered every shard and lifecycle policy without an error; never excluded", + "type": "timeseries", + "id": 261, + "gridPos": { + "h": 8, + "w": 12, + "x": 12, + "y": 51 + }, + "fieldConfig": { + "defaults": { + "color": { + "mode": "palette-classic" + }, + "custom": { + "axisBorderShow": false, + "axisCenteredZero": false, + "axisColorMode": "text", + "axisLabel": "", + "axisPlacement": "auto", + "barAlignment": 0, + "drawStyle": "line", + "fillOpacity": 10, + "gradientMode": "none", + "hideFrom": { + "legend": false, + "tooltip": false, + "viz": false + }, + "insertNulls": false, + "lineInterpolation": "linear", + "lineWidth": 1, + "pointSize": 4, + "scaleDistribution": { + "type": "linear" + }, + "showPoints": "never", + "spanNulls": false, + "stacking": { + "group": "A", + "mode": "none" + }, + "thresholdsStyle": { + "mode": "off" + } + }, + "mappings": [], + "unit": "s", + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + } + ] + } + }, + "overrides": [] + }, + "options": { + "legend": { + "calcs": [ + "lastNotNull", + "max" + ], + "displayMode": "table", + "placement": "bottom", + "showLegend": true + }, + "tooltip": { + "mode": "multi", + "sort": "desc" + } + }, + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "editorMode": "code", + "expr": "time() - max(SeaweedFS_s3_lifecycle_daily_run_last_full_pass_timestamp_seconds{cluster=~\"$cluster\"} > 0)", + "range": true, + "refId": "A", + "legendFormat": "last full pass" + } + ], + "pluginVersion": "10.3.1" } ] }, diff --git a/weed/admin/plugin/plugin_monitor.go b/weed/admin/plugin/plugin_monitor.go index eb360b639e..32359e3161 100644 --- a/weed/admin/plugin/plugin_monitor.go +++ b/weed/admin/plugin/plugin_monitor.go @@ -1120,13 +1120,15 @@ func (r *Plugin) trackExecutionCompletion(completed *plugin_pb.JobCompleted) *Tr if completed.RequestId != "" { tracked.RequestID = completed.RequestId } + if completed.Result != nil { + tracked.ResultSummary = completed.Result.Summary + } else { + tracked.ResultSummary = "" + } if completed.Success { tracked.State = strings.ToLower(plugin_pb.JobState_JOB_STATE_SUCCEEDED.String()) tracked.Progress = 100 tracked.Stage = "completed" - if completed.Result != nil { - tracked.ResultSummary = completed.Result.Summary - } tracked.Message = tracked.ResultSummary if tracked.Message == "" { tracked.Message = "completed" @@ -1159,7 +1161,7 @@ func (r *Plugin) trackExecutionCompletion(completed *plugin_pb.JobCompleted) *Tr detail.Message = clone.Message detail.ErrorMessage = clone.ErrorMessage detail.ResultSummary = clone.ResultSummary - if completed.Success && completed.Result != nil { + if completed.Result != nil { detail.ResultOutputValues = configValueMapToPlain(completed.Result.OutputValues) } else { detail.ResultOutputValues = nil diff --git a/weed/admin/plugin/plugin_monitor_test.go b/weed/admin/plugin/plugin_monitor_test.go index b4a6b82ca2..5fe83868d6 100644 --- a/weed/admin/plugin/plugin_monitor_test.go +++ b/weed/admin/plugin/plugin_monitor_test.go @@ -594,6 +594,95 @@ func TestBuildJobDetailIncludesActivitiesAndRunRecord(t *testing.T) { } } +func TestBuildJobDetailKeepsFailedJobResult(t *testing.T) { + t.Parallel() + + pluginSvc, err := New(Options{DataDir: t.TempDir()}) + if err != nil { + t.Fatalf("New: %v", err) + } + defer pluginSvc.Shutdown() + + pluginSvc.trackExecutionStart("req-failed", "worker-f", &plugin_pb.JobSpec{ + JobId: "job-failed", + JobType: "s3_lifecycle", + }, 1) + pluginSvc.handleJobCompleted(&plugin_pb.JobCompleted{ + RequestId: "req-failed", + JobId: "job-failed", + JobType: "s3_lifecycle", + Success: false, + ErrorMessage: "walk bad-a: filer down", + Result: &plugin_pb.JobResult{ + Summary: "pass failed", + OutputValues: map[string]*plugin_pb.ConfigValue{ + "buckets_failed": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 2}}, + }, + }, + CompletedAt: timestamppb.Now(), + }) + pluginSvc.Shutdown() + + detail, found, err := pluginSvc.BuildJobDetail("job-failed", 100, 5) + if err != nil { + t.Fatalf("BuildJobDetail error: %v", err) + } + if !found || detail == nil || detail.Job == nil { + t.Fatalf("expected job detail") + } + if detail.Job.State != "job_state_failed" || detail.Job.ErrorMessage != "walk bad-a: filer down" { + t.Fatalf("expected the failure to be kept, got state=%q error=%q", detail.Job.State, detail.Job.ErrorMessage) + } + if detail.Job.ResultSummary != "pass failed" { + t.Fatalf("expected the failed job's result summary, got %q", detail.Job.ResultSummary) + } + if detail.Job.ResultOutputValues["buckets_failed"] == nil { + t.Fatalf("expected the failed job's result output values, got %+v", detail.Job.ResultOutputValues) + } +} + +func TestRetryWithoutResultDropsFailedAttemptSummary(t *testing.T) { + t.Parallel() + + pluginSvc, err := New(Options{DataDir: t.TempDir()}) + if err != nil { + t.Fatalf("New: %v", err) + } + defer pluginSvc.Shutdown() + + job := &plugin_pb.JobSpec{JobId: "job-retry", JobType: "s3_lifecycle"} + pluginSvc.trackExecutionStart("req-retry", "worker-r", job, 1) + pluginSvc.handleJobCompleted(&plugin_pb.JobCompleted{ + RequestId: "req-retry", + JobId: "job-retry", + JobType: "s3_lifecycle", + Success: false, + ErrorMessage: "walk bad-a: filer down", + Result: &plugin_pb.JobResult{Summary: "pass failed"}, + CompletedAt: timestamppb.Now(), + }) + pluginSvc.trackExecutionStart("req-retry", "worker-r", job, 2) + pluginSvc.handleJobCompleted(&plugin_pb.JobCompleted{ + RequestId: "req-retry", + JobId: "job-retry", + JobType: "s3_lifecycle", + Success: true, + CompletedAt: timestamppb.Now(), + }) + pluginSvc.Shutdown() + + detail, found, err := pluginSvc.BuildJobDetail("job-retry", 100, 5) + if err != nil { + t.Fatalf("BuildJobDetail error: %v", err) + } + if !found || detail == nil || detail.Job == nil { + t.Fatalf("expected job detail") + } + if detail.Job.ResultSummary != "" || detail.Job.Message != "completed" { + t.Fatalf("expected the failed attempt's summary to be dropped, got summary=%q message=%q", detail.Job.ResultSummary, detail.Job.Message) + } +} + func TestBuildJobDetailLoadsFromDiskWhenMemoryCleared(t *testing.T) { t.Parallel() diff --git a/weed/s3api/s3lifecycle/dailyrun/cursor.go b/weed/s3api/s3lifecycle/dailyrun/cursor.go index fc73a46471..5e46f5b634 100644 --- a/weed/s3api/s3lifecycle/dailyrun/cursor.go +++ b/weed/s3api/s3lifecycle/dailyrun/cursor.go @@ -18,6 +18,7 @@ import ( "encoding/json" "errors" "fmt" + "time" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" "github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/dispatcher" @@ -161,3 +162,58 @@ func (p *FilerCursorPersister) Save(ctx context.Context, shardID int, c Cursor) } return nil } + +const fullPassFileName = "full-pass.json" + +// FullPass is the last pass that covered every shard and every lifecycle +// policy without an error. Shard cursors advance on partial passes too, +// so they cannot answer that. +type FullPass struct { + StartedAt time.Time + CompletedAt time.Time +} + +type fullPassFile struct { + Version int `json:"version"` + StartedAtNs int64 `json:"started_at_ns"` + CompletedAtNs int64 `json:"completed_at_ns"` +} + +func (p *FilerCursorPersister) LoadFullPass(ctx context.Context) (FullPass, bool, error) { + if p.Store == nil { + return FullPass{}, false, errors.New("FilerCursorPersister: nil Store") + } + content, err := p.Store.Read(ctx, CursorDir, fullPassFileName) + if err != nil { + if errors.Is(err, filer_pb.ErrNotFound) { + return FullPass{}, false, nil + } + return FullPass{}, false, fmt.Errorf("full pass read: %w", err) + } + var f fullPassFile + if err := json.Unmarshal(content, &f); err != nil { + return FullPass{}, false, fmt.Errorf("full pass decode: %w", err) + } + if f.Version != cursorFileVersion { + return FullPass{}, false, fmt.Errorf("full pass version: got %d, want %d", f.Version, cursorFileVersion) + } + return FullPass{StartedAt: time.Unix(0, f.StartedAtNs).UTC(), CompletedAt: time.Unix(0, f.CompletedAtNs).UTC()}, true, nil +} + +func (p *FilerCursorPersister) SaveFullPass(ctx context.Context, fp FullPass) error { + if p.Store == nil { + return errors.New("FilerCursorPersister: nil Store") + } + content, err := json.Marshal(fullPassFile{ + Version: cursorFileVersion, + StartedAtNs: fp.StartedAt.UnixNano(), + CompletedAtNs: fp.CompletedAt.UnixNano(), + }) + if err != nil { + return fmt.Errorf("full pass encode: %w", err) + } + if err := p.Store.Save(ctx, CursorDir, fullPassFileName, content); err != nil { + return fmt.Errorf("full pass save: %w", err) + } + return nil +} diff --git a/weed/s3api/s3lifecycle/dailyrun/cursor_test.go b/weed/s3api/s3lifecycle/dailyrun/cursor_test.go index 9a9279a61d..7e09f51666 100644 --- a/weed/s3api/s3lifecycle/dailyrun/cursor_test.go +++ b/weed/s3api/s3lifecycle/dailyrun/cursor_test.go @@ -4,6 +4,7 @@ import ( "context" "sync" "testing" + "time" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" "github.com/stretchr/testify/assert" @@ -153,3 +154,32 @@ func TestFilerCursorPersister_NilStoreErrors(t *testing.T) { require.Error(t, err) require.Error(t, p.Save(context.Background(), 0, Cursor{})) } + +func TestFilerCursorPersister_FullPassRoundTrip(t *testing.T) { + store := newFakeStore() + p := &FilerCursorPersister{Store: store} + _, found, err := p.LoadFullPass(context.Background()) + require.NoError(t, err) + assert.False(t, found) + + want := FullPass{ + StartedAt: time.Unix(1_700_000_000, 123).UTC(), + CompletedAt: time.Unix(1_700_003_600, 456).UTC(), + } + require.NoError(t, p.SaveFullPass(context.Background(), want)) + got, found, err := p.LoadFullPass(context.Background()) + require.NoError(t, err) + require.True(t, found) + assert.Equal(t, want, got) + + _, found, err = p.Load(context.Background(), 0) + require.NoError(t, err) + assert.False(t, found, "the full-pass record must not read as a shard cursor") +} + +func TestFilerCursorPersister_FullPassRejectsUnknownVersion(t *testing.T) { + store := newFakeStore() + require.NoError(t, store.Save(context.Background(), CursorDir, fullPassFileName, []byte(`{"version":99,"completed_at_ns":1}`))) + _, _, err := (&FilerCursorPersister{Store: store}).LoadFullPass(context.Background()) + assert.Error(t, err) +} diff --git a/weed/s3api/s3lifecycle/dailyrun/run.go b/weed/s3api/s3lifecycle/dailyrun/run.go index beeacfa6d4..1adb83a144 100644 --- a/weed/s3api/s3lifecycle/dailyrun/run.go +++ b/weed/s3api/s3lifecycle/dailyrun/run.go @@ -91,6 +91,10 @@ type Config struct { // 0 -> defaultSubscriptionReceiveTimeout. SubscriptionReceiveTimeout time.Duration + + // nil -> no per-pass summary. Share it with the walker's + // WalkBuckets and WalkerDispatcher to cover both paths. + Stats *RunStats } // Above the filer's 15m maxGapStall, so a subscriber parked on a gap is @@ -240,6 +244,7 @@ func Run(ctx context.Context, cfg Config) error { first = readerErr } } + cfg.Stats.setShardErrors(errCount) status := "ok" if first != nil { status = "error" @@ -562,6 +567,8 @@ func runShard(ctx context.Context, cfg Config, snap *engine.Snapshot, runNow tim lastWalkedNs = runNow.UnixNano() walkedThisPass = true } + } else if cfg.Walker != nil { + countDeferredWalk(cfg.Stats, snap, shardID, retentionWindow) } _ = walkedThisPass // empty-replay branch returns; no downstream walker. return saveCursorAndPublish(ctx, cfg.Persister, shardID, Cursor{ @@ -625,6 +632,8 @@ func runShard(ctx context.Context, cfg Config, snap *engine.Snapshot, runNow tim } lastWalkedNs = runNow.UnixNano() } + } else if cfg.Walker != nil && !walkedThisPass { + countDeferredWalk(cfg.Stats, snap, shardID, retentionWindow) } // Cold start: scan from now-maxTTL so already-due objects within @@ -644,7 +653,10 @@ func runShard(ctx context.Context, cfg Config, snap *engine.Snapshot, runNow tim // Walk-bound rules are never due inside the retention window; keep them // out of the drain so their matches don't pin the cursor. replayView := snap.ReplayView(retentionWindow) - lastOK, _, drainErr := drainShardEvents(ctx, cfg, runNow, shardID, replayView, startTsNs, events) + lastOK, halted, drainErr := drainShardEvents(ctx, cfg, runNow, shardID, replayView, startTsNs, events) + if halted && drainErr == nil { + cfg.Stats.shardHalted() + } // Fresh ctx: a wall-clock cap on the pass (the shell driver's // -runtime) cancels the drain's, and saving with it would drop the // cursor and re-replay from the same floor next pass. @@ -684,6 +696,17 @@ func saveCursorAndPublish(ctx context.Context, p CursorPersister, shardID int, c return nil } +// countDeferredWalk counts a shard whose walk-only rules waited for +// WalkerInterval, so the pass left them unevaluated. +func countDeferredWalk(s *RunStats, snap *engine.Snapshot, shardID int, retentionWindow time.Duration) { + if s == nil { + return + } + if _, walkView := snap.RulesForShard(shardID, retentionWindow); walkView != nil && len(walkView.AllActions()) > 0 { + s.walkDeferred() + } +} + // walkerDue answers the persisted-state throttle: has enough time // elapsed since the last walker fire on this shard? interval == 0 // means "fire every pass" (the prior, unconditional behavior). @@ -719,6 +742,7 @@ func drainShardEvents(ctx context.Context, cfg Config, runNow time.Time, shardID cursorAdvanceTo := startTsNs stuck := false halted := false + cfg.Stats.startReplay() drain: for { @@ -739,6 +763,7 @@ drain: continue } stats.S3LifecycleDailyRunEventsScanned.WithLabelValues(strconv.Itoa(shardID)).Inc() + cfg.Stats.scanEvent() matches := router.Route(ctx, snap, ev, runNow, cfg.Lister) eventSkipped, eventHalted, eventErr := processMatches(ctx, cfg, runNow, ev, matches) if eventErr != nil { @@ -786,10 +811,10 @@ func processMatches(ctx context.Context, cfg Config, runNow time.Time, ev *reade // "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() + countDispatch(cfg.Stats, m.Bucket, m.Key.ActionKind.String(), "RPC_ERROR") return skippedAny, true, nil } - stats.S3LifecycleDispatchCounter.WithLabelValues(m.Bucket, m.Key.ActionKind.String(), outcome.String()).Inc() + countDispatch(cfg.Stats, m.Bucket, m.Key.ActionKind.String(), outcome.String()) switch outcome { case s3_lifecycle_pb.LifecycleDeleteOutcome_DONE, s3_lifecycle_pb.LifecycleDeleteOutcome_NOOP_RESOLVED, diff --git a/weed/s3api/s3lifecycle/dailyrun/run_fanout_test.go b/weed/s3api/s3lifecycle/dailyrun/run_fanout_test.go index 59c849654d..533689d54b 100644 --- a/weed/s3api/s3lifecycle/dailyrun/run_fanout_test.go +++ b/weed/s3api/s3lifecycle/dailyrun/run_fanout_test.go @@ -126,6 +126,7 @@ func TestRun_HaltedShardDoesNotStarveOthers(t *testing.T) { Persister: persister, Lister: stubSiblingLister{}, Now: func() time.Time { return runNow }, + Stats: &RunStats{}, } done := make(chan error, 1) @@ -145,4 +146,11 @@ func TestRun_HaltedShardDoesNotStarveOthers(t *testing.T) { require.NoError(t, err) require.True(t, found) assert.Greater(t, starvedCursor.TsNs, seeded.TsNs, "starved shard's cursor must advance") + + sum := cfg.Stats.Summary() + assert.Equal(t, PhaseReplay, sum.Phase) + assert.Equal(t, int64(2), sum.EventsScanned, "the halted shard stops after its first event") + assert.Equal(t, map[string]int64{"BLOCKED": 1, "DONE": 1}, sum.Outcomes) + assert.Zero(t, sum.ShardErrors) + assert.Equal(t, 1, sum.ShardsHalted, "a BLOCKED delete halts its shard without an error") } diff --git a/weed/s3api/s3lifecycle/dailyrun/run_stats.go b/weed/s3api/s3lifecycle/dailyrun/run_stats.go new file mode 100644 index 0000000000..6dee7eb07d --- /dev/null +++ b/weed/s3api/s3lifecycle/dailyrun/run_stats.go @@ -0,0 +1,221 @@ +package dailyrun + +import ( + "strings" + "sync" + "sync/atomic" + + "github.com/seaweedfs/seaweedfs/weed/stats" +) + +// Bounds what a summary carries per pass; the counts stay exact. +const ( + maxReportedBucketFailures = 10 + maxReportedErrorLen = 256 +) + +const ( + PhaseWalk = "walk" + PhaseReplay = "replay" +) + +type bucketState uint8 + +const ( + bucketSkipped bucketState = iota + 1 + bucketWalked + bucketFailed +) + +type BucketFailure struct { + Bucket string + Error string +} + +// RunSummary is a copy of a RunStats at one point in time. A bucket walked +// more than once in a pass counts once, as failed if any walk failed. +type RunSummary struct { + Phase string + Bucket string + BucketsWalked int + BucketsFailed int + BucketsSkipped int + // The first maxReportedBucketFailures failed buckets. + BucketFailures []BucketFailure + // Every listed entry, versions and delete markers included. + EntriesVisited int64 + // Distinct object keys among the listed entries. + ObjectsVisited int64 + EventsScanned int64 + // LifecycleDelete outcomes keyed by the dispatch_total outcome label. + // They count RPC results, not reclaimed space. + Outcomes map[string]int64 + ShardErrors int + // Shards whose replay stopped on an unresolved delete; their due + // events wait for a later pass. + ShardsHalted int + // Shards whose walk-only rules waited for WalkerInterval. + WalksDeferred int +} + +// RunStats collects one pass's work. Safe for concurrent use; a nil +// *RunStats ignores every call. +type RunStats struct { + entries atomic.Int64 + objects atomic.Int64 + events atomic.Int64 + + mu sync.Mutex + phase string + bucket string + buckets map[string]bucketState + failures []BucketFailure + outcomes map[string]int64 + shardErrors int + shardsHalted int + deferred int +} + +func (s *RunStats) startBucket(bucket string) { + if s == nil { + return + } + s.mu.Lock() + s.phase, s.bucket = PhaseWalk, bucket + s.mu.Unlock() +} + +func (s *RunStats) finishBucket(bucket string, state bucketState, err error) { + if s == nil { + return + } + s.mu.Lock() + defer s.mu.Unlock() + if s.buckets == nil { + s.buckets = map[string]bucketState{} + } + prev := s.buckets[bucket] + if state <= prev { + return + } + s.buckets[bucket] = state + if state == bucketFailed && len(s.failures) < maxReportedBucketFailures { + s.failures = append(s.failures, BucketFailure{Bucket: bucket, Error: truncateError(err)}) + } +} + +func (s *RunStats) startReplay() { + if s == nil { + return + } + s.mu.Lock() + s.phase, s.bucket = PhaseReplay, "" + s.mu.Unlock() +} + +func (s *RunStats) visitEntry(newObject bool) { + if s == nil { + return + } + s.entries.Add(1) + if newObject { + s.objects.Add(1) + } +} + +func (s *RunStats) scanEvent() { + if s == nil { + return + } + s.events.Add(1) +} + +func (s *RunStats) recordOutcome(outcome string) { + if s == nil { + return + } + s.mu.Lock() + if s.outcomes == nil { + s.outcomes = map[string]int64{} + } + s.outcomes[outcome]++ + s.mu.Unlock() +} + +func (s *RunStats) setShardErrors(n int) { + if s == nil { + return + } + s.mu.Lock() + s.shardErrors = n + s.mu.Unlock() +} + +func (s *RunStats) shardHalted() { + if s == nil { + return + } + s.mu.Lock() + s.shardsHalted++ + s.mu.Unlock() +} + +func (s *RunStats) walkDeferred() { + if s == nil { + return + } + s.mu.Lock() + s.deferred++ + s.mu.Unlock() +} + +func (s *RunStats) Summary() RunSummary { + if s == nil { + return RunSummary{} + } + out := RunSummary{ + EntriesVisited: s.entries.Load(), + ObjectsVisited: s.objects.Load(), + EventsScanned: s.events.Load(), + } + s.mu.Lock() + defer s.mu.Unlock() + out.Phase, out.Bucket = s.phase, s.bucket + for _, state := range s.buckets { + switch state { + case bucketWalked: + out.BucketsWalked++ + case bucketFailed: + out.BucketsFailed++ + case bucketSkipped: + out.BucketsSkipped++ + } + } + out.BucketFailures = append([]BucketFailure(nil), s.failures...) + if len(s.outcomes) > 0 { + out.Outcomes = make(map[string]int64, len(s.outcomes)) + for k, v := range s.outcomes { + out.Outcomes[k] = v + } + } + out.ShardErrors = s.shardErrors + out.ShardsHalted = s.shardsHalted + out.WalksDeferred = s.deferred + return out +} + +func countDispatch(s *RunStats, bucket, kind, outcome string) { + stats.S3LifecycleDispatchCounter.WithLabelValues(bucket, kind, outcome).Inc() + s.recordOutcome(outcome) +} + +func truncateError(err error) string { + if err == nil { + return "" + } + msg := err.Error() + if len(msg) <= maxReportedErrorLen { + return msg + } + return strings.ToValidUTF8(msg[:maxReportedErrorLen], "") + "..." +} diff --git a/weed/s3api/s3lifecycle/dailyrun/run_stats_test.go b/weed/s3api/s3lifecycle/dailyrun/run_stats_test.go new file mode 100644 index 0000000000..a37badd81f --- /dev/null +++ b/weed/s3api/s3lifecycle/dailyrun/run_stats_test.go @@ -0,0 +1,169 @@ +package dailyrun + +import ( + "context" + "errors" + "strings" + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/weed/pb/s3_lifecycle_pb" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/bootstrap" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/reader" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/router" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func allShards() []int { + out := make([]int, s3lifecycle.ShardCount) + for i := range out { + out[i] = i + } + return out +} + +func TestWalkBuckets_RecordsEveryBucketFailure(t *testing.T) { + list := func(_ context.Context, bucket, _ string, _ func(*bootstrap.Entry) error) error { + if strings.HasPrefix(bucket, "bad") { + return errors.New("list " + bucket + " failed") + } + return nil + } + runStats := &RunStats{} + err := WalkBuckets(context.Background(), snapshotForBucketRule(t, "good", 7), []int{0}, + []string{"bad-a", "good", "bad-b"}, list, &recordingDispatcher{}, runStats) + require.Error(t, err) + assert.Contains(t, err.Error(), "bad-a", "the returned error stays the first bucket's") + + sum := runStats.Summary() + assert.Equal(t, 1, sum.BucketsWalked) + assert.Equal(t, 2, sum.BucketsFailed) + assert.Equal(t, 0, sum.BucketsSkipped) + require.Len(t, sum.BucketFailures, 2) + assert.Equal(t, "bad-a", sum.BucketFailures[0].Bucket) + assert.Contains(t, sum.BucketFailures[0].Error, "list bad-a failed") + assert.Equal(t, "bad-b", sum.BucketFailures[1].Bucket) + assert.Equal(t, PhaseWalk, sum.Phase) + assert.Equal(t, "bad-b", sum.Bucket) +} + +func TestWalkBuckets_CancelMarksRemainingBucketsSkipped(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + list := func(ctx context.Context, bucket, _ string, _ func(*bootstrap.Entry) error) error { + if bucket == "second" { + cancel() + return ctx.Err() + } + return nil + } + runStats := &RunStats{} + err := WalkBuckets(ctx, snapshotForBucketRule(t, "first", 7), []int{0}, + []string{"first", "second", "third"}, list, &recordingDispatcher{}, runStats) + assert.ErrorIs(t, err, context.Canceled) + + sum := runStats.Summary() + assert.Equal(t, 1, sum.BucketsWalked) + assert.Equal(t, 0, sum.BucketsFailed, "a cancelled walk is not a bucket failure") + assert.Equal(t, 2, sum.BucketsSkipped) + assert.Empty(t, sum.BucketFailures) +} + +func TestRunStats_FailedWalkWinsAcrossWalksOfOneBucket(t *testing.T) { + runStats := &RunStats{} + runStats.finishBucket("b", bucketSkipped, nil) + runStats.finishBucket("b", bucketWalked, nil) + runStats.finishBucket("b", bucketFailed, errors.New("first")) + runStats.finishBucket("b", bucketFailed, errors.New("second")) + runStats.finishBucket("b", bucketWalked, nil) + + sum := runStats.Summary() + assert.Equal(t, 1, sum.BucketsFailed) + assert.Equal(t, 0, sum.BucketsWalked) + require.Len(t, sum.BucketFailures, 1) + assert.Equal(t, "first", sum.BucketFailures[0].Error) +} + +func TestRunStats_BoundsReportedFailures(t *testing.T) { + runStats := &RunStats{} + long := strings.Repeat("x", 2*maxReportedErrorLen) + for i := 0; i < maxReportedBucketFailures+5; i++ { + runStats.finishBucket("b-"+intToStr(i), bucketFailed, errors.New(long)) + } + sum := runStats.Summary() + assert.Equal(t, maxReportedBucketFailures+5, sum.BucketsFailed, "the count stays exact") + require.Len(t, sum.BucketFailures, maxReportedBucketFailures) + assert.LessOrEqual(t, len(sum.BucketFailures[0].Error), maxReportedErrorLen+len("...")) +} + +func TestWalkBuckets_CountsEntriesAndObjects(t *testing.T) { + old := time.Now().Add(-90 * 24 * time.Hour) + entries := []*bootstrap.Entry{ + {Path: "a", VersionID: "v3", ModTime: old, IsLatest: true}, + {Path: "a", VersionID: "v2", ModTime: old}, + {Path: "a", VersionID: "v1", ModTime: old}, + {Path: "b", ModTime: old, IsLatest: true}, + {Path: "dir", IsDirectory: true}, + {Path: ".uploads/u1", DestKey: "c", IsDirectory: true, IsMPUInit: true, ModTime: old}, + } + runStats := &RunStats{} + require.NoError(t, WalkBuckets(context.Background(), snapshotForBucketRule(t, "bkt", 7), allShards(), + []string{"bkt"}, bootstrap.EntryCallback(entries), &recordingDispatcher{}, runStats)) + + sum := runStats.Summary() + assert.Equal(t, int64(5), sum.EntriesVisited, "versions and upload records count, plain directories do not") + assert.Equal(t, int64(2), sum.ObjectsVisited, "versions of one key count as one object") + assert.Equal(t, 1, sum.BucketsWalked) +} + +func TestWalkerDispatcher_CountsOutcomes(t *testing.T) { + runStats := &RunStats{} + a := sampleAction(t, s3lifecycle.ActionKindExpirationDays) + e := &bootstrap.Entry{Path: "obj"} + + d := &WalkerDispatcher{Client: &walkerStubClient{outcome: s3_lifecycle_pb.LifecycleDeleteOutcome_DONE}, Stats: runStats} + require.NoError(t, d.Delete(context.Background(), a, e)) + require.NoError(t, d.Delete(context.Background(), a, e)) + d.Client = &walkerStubClient{outcome: s3_lifecycle_pb.LifecycleDeleteOutcome_SKIPPED_OBJECT_LOCK} + require.NoError(t, d.Delete(context.Background(), a, e)) + d.Client = &walkerStubClient{err: errors.New("unavailable")} + require.Error(t, d.Delete(context.Background(), a, e)) + d.Client = &walkerStubClient{nilResp: true} + require.Error(t, d.Delete(context.Background(), a, e)) + + assert.Equal(t, map[string]int64{"DONE": 2, "SKIPPED_OBJECT_LOCK": 1, "RPC_ERROR": 2}, runStats.Summary().Outcomes) +} + +func TestProcessMatches_CountsOutcomes(t *testing.T) { + runNow := time.Now() + rule := s3lifecycle.ActionKey{Bucket: "b", RuleHash: [8]byte{1}, ActionKind: s3lifecycle.ActionKindExpirationDays} + matches := []router.Match{ + {Key: rule, Bucket: "b", ObjectKey: "a", DueTime: runNow.Add(-time.Hour)}, + {Key: rule, Bucket: "b", ObjectKey: "b", DueTime: runNow.Add(-time.Hour)}, + {Key: rule, Bucket: "b", ObjectKey: "c", DueTime: runNow.Add(-time.Hour)}, + } + client := &recordingClient{responses: []s3_lifecycle_pb.LifecycleDeleteOutcome{ + s3_lifecycle_pb.LifecycleDeleteOutcome_DONE, + s3_lifecycle_pb.LifecycleDeleteOutcome_NOOP_RESOLVED, + s3_lifecycle_pb.LifecycleDeleteOutcome_RETRY_LATER, + }} + runStats := &RunStats{} + _, halted, err := processMatches(context.Background(), Config{Client: client, Stats: runStats}, runNow, &reader.Event{}, matches) + require.NoError(t, err) + assert.True(t, halted) + assert.Equal(t, map[string]int64{"DONE": 1, "NOOP_RESOLVED": 1, "RETRY_LATER": 1}, runStats.Summary().Outcomes) +} + +func TestRunStats_NilIsANoOp(t *testing.T) { + var runStats *RunStats + runStats.startBucket("b") + runStats.finishBucket("b", bucketFailed, errors.New("x")) + runStats.startReplay() + runStats.visitEntry(true) + runStats.scanEvent() + runStats.recordOutcome("DONE") + runStats.setShardErrors(1) + assert.Equal(t, RunSummary{}, runStats.Summary()) +} diff --git a/weed/s3api/s3lifecycle/dailyrun/walk_buckets.go b/weed/s3api/s3lifecycle/dailyrun/walk_buckets.go index 2a0eac3589..fe7dcfe4ad 100644 --- a/weed/s3api/s3lifecycle/dailyrun/walk_buckets.go +++ b/weed/s3api/s3lifecycle/dailyrun/walk_buckets.go @@ -14,12 +14,13 @@ import ( // WalkBuckets runs bootstrap.Walk for each bucket using a per-bucket // ListFunc wrapped to drop entries whose ShardID isn't in shards. // Returns the first bucket error; remaining buckets log and continue -// so one bucket's filer error doesn't kill the whole walk. +// so one bucket's filer error doesn't kill the whole walk. Every +// bucket's result goes to runStats when it is non-nil. // // Composable shape: callers (the worker handler) supply a real // filer-backed ListFunc and a real Dispatcher (WalkerDispatcher). // Tests pass bootstrap.EntryCallback and a stub Dispatcher. -func WalkBuckets(ctx context.Context, view *engine.Snapshot, shards []int, buckets []string, list bootstrap.ListFunc, dispatch bootstrap.Dispatcher) error { +func WalkBuckets(ctx context.Context, view *engine.Snapshot, shards []int, buckets []string, list bootstrap.ListFunc, dispatch bootstrap.Dispatcher, runStats *RunStats) error { if view == nil { return errors.New("WalkBuckets: nil view") } @@ -33,13 +34,26 @@ func WalkBuckets(ctx context.Context, view *engine.Snapshot, shards []int, bucke for _, sh := range shards { shardSet[sh] = true } - filtered := perShardListFunc(list, shardSet) + filtered := perShardListFunc(list, shardSet, runStats) var firstErr error - for _, b := range buckets { + for i, b := range buckets { if err := ctx.Err(); err != nil { + for _, rest := range buckets[i:] { + runStats.finishBucket(rest, bucketSkipped, nil) + } return err } - if _, err := bootstrap.Walk(ctx, view, b, filtered, dispatch, bootstrap.WalkOptions{}); err != nil { + runStats.startBucket(b) + _, err := bootstrap.Walk(ctx, view, b, filtered, dispatch, bootstrap.WalkOptions{}) + switch { + case err == nil: + runStats.finishBucket(b, bucketWalked, nil) + case ctx.Err() != nil: + runStats.finishBucket(b, bucketSkipped, nil) + default: + runStats.finishBucket(b, bucketFailed, err) + } + if err != nil { if firstErr == nil { firstErr = fmt.Errorf("walk %s: %w", b, err) } else { @@ -57,8 +71,10 @@ func WalkBuckets(ctx context.Context, view *engine.Snapshot, shards []int, bucke // MPU init records carry the user's destination in DestKey, so the // shard predicate must use DestKey there to match what the dispatcher // will send. -func perShardListFunc(inner bootstrap.ListFunc, shardSet map[int]bool) bootstrap.ListFunc { +func perShardListFunc(inner bootstrap.ListFunc, shardSet map[int]bool, runStats *RunStats) bootstrap.ListFunc { return func(ctx context.Context, bucket, start string, cb func(*bootstrap.Entry) error) error { + // Listings emit every version of a key back to back. + lastPath := "" return inner(ctx, bucket, start, func(e *bootstrap.Entry) error { if e == nil { return nil @@ -66,6 +82,10 @@ func perShardListFunc(inner bootstrap.ListFunc, shardSet map[int]bool) bootstrap if !shardSet[entryShardID(bucket, e)] { return nil } + if !e.IsDirectory || e.IsMPUInit { + runStats.visitEntry(!e.IsMPUInit && e.Path != lastPath) + lastPath = e.Path + } return cb(e) }) } diff --git a/weed/s3api/s3lifecycle/dailyrun/walk_buckets_test.go b/weed/s3api/s3lifecycle/dailyrun/walk_buckets_test.go index b311b4f5ff..c4260a395f 100644 --- a/weed/s3api/s3lifecycle/dailyrun/walk_buckets_test.go +++ b/weed/s3api/s3lifecycle/dailyrun/walk_buckets_test.go @@ -105,7 +105,7 @@ func TestWalkBuckets_DispatchesOnlyShardMatchingEntries(t *testing.T) { snap := snapshotForBucketRule(t, bucket, 7) // 30-day-old objects expire d := &recordingDispatcher{} err := WalkBuckets(context.Background(), snap, []int{targetShard}, []string{bucket}, - bootstrap.EntryCallback(all), d) + bootstrap.EntryCallback(all), d, nil) require.NoError(t, err) gotPaths := make([]string, 0, len(d.calls)) @@ -124,9 +124,9 @@ func TestWalkBuckets_DispatchesOnlyShardMatchingEntries(t *testing.T) { } func TestWalkBuckets_NilGuards(t *testing.T) { - require.Error(t, WalkBuckets(context.Background(), nil, []int{0}, nil, bootstrap.EntryCallback(nil), &recordingDispatcher{})) - require.Error(t, WalkBuckets(context.Background(), snapshotForBucketRule(t, "b", 7), []int{0}, nil, nil, &recordingDispatcher{})) - require.Error(t, WalkBuckets(context.Background(), snapshotForBucketRule(t, "b", 7), []int{0}, nil, bootstrap.EntryCallback(nil), nil)) + require.Error(t, WalkBuckets(context.Background(), nil, []int{0}, nil, bootstrap.EntryCallback(nil), &recordingDispatcher{}, nil)) + require.Error(t, WalkBuckets(context.Background(), snapshotForBucketRule(t, "b", 7), []int{0}, nil, nil, &recordingDispatcher{}, nil)) + require.Error(t, WalkBuckets(context.Background(), snapshotForBucketRule(t, "b", 7), []int{0}, nil, bootstrap.EntryCallback(nil), nil, nil)) } func TestWalkBuckets_OneBucketErrorDoesNotStopOthers(t *testing.T) { @@ -141,7 +141,7 @@ func TestWalkBuckets_OneBucketErrorDoesNotStopOthers(t *testing.T) { } snap := snapshotForBucketRule(t, "good", 7) - err := WalkBuckets(context.Background(), snap, []int{0}, []string{"bad", "good"}, list, &recordingDispatcher{}) + err := WalkBuckets(context.Background(), snap, []int{0}, []string{"bad", "good"}, list, &recordingDispatcher{}, nil) require.Error(t, err) assert.ErrorIs(t, err, listErr) } @@ -153,7 +153,7 @@ func TestWalkBuckets_HonorsContextCancellationBetweenBuckets(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) cancel() err := WalkBuckets(ctx, snapshotForBucketRule(t, "b", 7), []int{0}, []string{"b"}, - bootstrap.EntryCallback(nil), &recordingDispatcher{}) + bootstrap.EntryCallback(nil), &recordingDispatcher{}, nil) assert.ErrorIs(t, err, context.Canceled) } diff --git a/weed/s3api/s3lifecycle/dailyrun/walker_dispatcher.go b/weed/s3api/s3lifecycle/dailyrun/walker_dispatcher.go index 2ab26162ba..6ac582ba23 100644 --- a/weed/s3api/s3lifecycle/dailyrun/walker_dispatcher.go +++ b/weed/s3api/s3lifecycle/dailyrun/walker_dispatcher.go @@ -24,6 +24,8 @@ type WalkerDispatcher struct { // daily-run's processMatches uses so the walker and replay paths // can't combine to burst past the cap. nil disables throttling. Limiter *rate.Limiter + // nil skips per-pass outcome counts. + Stats *RunStats } // Compile-time check. @@ -75,17 +77,17 @@ func (d *WalkerDispatcher) Delete(ctx context.Context, action *engine.CompiledAc if err != nil { // "RPC_ERROR" matches the streaming dispatcher and processMatches // labels so transport-class failures aggregate under one key. - stats.S3LifecycleDispatchCounter.WithLabelValues(action.Bucket, kindLabel, "RPC_ERROR").Inc() + countDispatch(d.Stats, action.Bucket, kindLabel, "RPC_ERROR") return fmt.Errorf("walker dispatch %s/%s %s: %w", action.Bucket, objectPath, action.Key.ActionKind, err) } if resp == nil { // A misbehaving server stub returning (nil, nil) would panic on // the switch below. Bucket under RPC_ERROR rather than a new // label — operationally it's the same class of failure. - stats.S3LifecycleDispatchCounter.WithLabelValues(action.Bucket, kindLabel, "RPC_ERROR").Inc() + countDispatch(d.Stats, action.Bucket, kindLabel, "RPC_ERROR") return fmt.Errorf("walker dispatch %s/%s %s: nil response", action.Bucket, objectPath, action.Key.ActionKind) } - stats.S3LifecycleDispatchCounter.WithLabelValues(action.Bucket, kindLabel, resp.Outcome.String()).Inc() + countDispatch(d.Stats, action.Bucket, kindLabel, resp.Outcome.String()) switch resp.Outcome { case s3_lifecycle_pb.LifecycleDeleteOutcome_DONE, s3_lifecycle_pb.LifecycleDeleteOutcome_NOOP_RESOLVED, diff --git a/weed/s3api/s3lifecycle/dailyrun/walker_interval_test.go b/weed/s3api/s3lifecycle/dailyrun/walker_interval_test.go index 27f9230d8a..6cb95ebd6b 100644 --- a/weed/s3api/s3lifecycle/dailyrun/walker_interval_test.go +++ b/weed/s3api/s3lifecycle/dailyrun/walker_interval_test.go @@ -136,13 +136,16 @@ func TestRunShard_WalkerThrottle(t *testing.T) { runNow2 := runNow.Add(tc.secondPassAfter) closedEvents2 := make(chan *readerEventAlias) close(closedEvents2) + cfg.Stats = &RunStats{} require.NoError(t, runShard(context.Background(), cfg, snap, runNow2, 0, closedEvents2, nil)) afterPass2, _, _ := p.Load(context.Background(), 0) assert.Equal(t, tc.wantTotalCalls, calls, "walker calls after 2 passes") if tc.wantSecondAdvNs { assert.Equal(t, runNow2.UnixNano(), afterPass2.LastWalkedNs, "throttle allowed the second walk") + assert.Zero(t, cfg.Stats.Summary().WalksDeferred) } else { assert.Equal(t, runNow.UnixNano(), afterPass2.LastWalkedNs, "throttle suppressed the second walk; anchor unchanged") + assert.Equal(t, 1, cfg.Stats.Summary().WalksDeferred, "a throttled walk leaves its rules unevaluated") } }) } diff --git a/weed/shell/command_s3_lifecycle_run_shard.go b/weed/shell/command_s3_lifecycle_run_shard.go index 47025bf8c2..21926fd314 100644 --- a/weed/shell/command_s3_lifecycle_run_shard.go +++ b/weed/shell/command_s3_lifecycle_run_shard.go @@ -167,7 +167,7 @@ func (c *commandS3LifecycleRunShard) Do(args []string, env *CommandEnv, writer i } walker := dailyrun.WalkerFunc(func(walkCtx context.Context, view *engine.Snapshot, shards []int) error { return dailyrun.WalkBuckets(walkCtx, view, shards, buckets, - dailyrun.FilerListFunc(filerClient, bucketsPath, view), walkerDispatch) + dailyrun.FilerListFunc(filerClient, bucketsPath, view), walkerDispatch, nil) }) return dailyrun.Run(passCtx, dailyrun.Config{ Shards: shards, diff --git a/weed/stats/metrics.go b/weed/stats/metrics.go index b39e9c410f..64ff6d373b 100644 --- a/weed/stats/metrics.go +++ b/weed/stats/metrics.go @@ -954,6 +954,18 @@ var ( Help: "Per-shard timestamp (UnixNano) of the most recent successful walker fire. 0 means the shard hasn't completed a walk yet.", }, []string{"shard"}) + // S3LifecycleDailyRunLastFullPassTimestampSeconds is when the last + // lifecycle pass that covered every shard and every lifecycle policy + // without an error finished. Partial, failed and cancelled passes leave + // it alone, unlike the per-shard cursor and walk gauges. + S3LifecycleDailyRunLastFullPassTimestampSeconds = prometheus.NewGauge( + prometheus.GaugeOpts{ + Namespace: Namespace, + Subsystem: subsystemS3Lifecycle, + Name: "daily_run_last_full_pass_timestamp_seconds", + Help: "Unix time the last fully successful lifecycle pass finished. 0 means none is known yet.", + }) + AdminMaintenanceTasksByStatus = prometheus.NewGaugeVec( prometheus.GaugeOpts{ Namespace: Namespace, @@ -1141,6 +1153,7 @@ func init() { Gather.MustRegister(S3LifecycleDailyRunShardDurationSeconds) Gather.MustRegister(S3LifecycleDailyRunEventsScanned) Gather.MustRegister(S3LifecycleDailyRunLastWalkedNs) + Gather.MustRegister(S3LifecycleDailyRunLastFullPassTimestampSeconds) Gather.MustRegister(UploadErrorCounter) diff --git a/weed/worker/tasks/s3_lifecycle/handler.go b/weed/worker/tasks/s3_lifecycle/handler.go index 50d9d4a021..3bdf41f66b 100644 --- a/weed/worker/tasks/s3_lifecycle/handler.go +++ b/weed/worker/tasks/s3_lifecycle/handler.go @@ -19,6 +19,7 @@ import ( "github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/dispatcher" "github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/engine" "github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/scheduler" + "github.com/seaweedfs/seaweedfs/weed/stats" "golang.org/x/time/rate" "google.golang.org/grpc" ) @@ -257,25 +258,27 @@ func (h *Handler) Execute(ctx context.Context, request *plugin_pb.ExecuteJobRequ defer s3Conn.Close() rpc := s3_lifecycle_pb.NewSeaweedS3LifecycleInternalClient(s3Conn) - if err := h.executeDailyReplay(runCtx, request, bucketsPath, filerClient, rpc, cfg, sender); err != nil { + report, err := h.executeDailyReplay(runCtx, request, bucketsPath, filerClient, rpc, cfg, sender) + if err != nil { return err } - return sendSuccessCompletion(request, sender) + return sendCompletion(request, sender, report) } -const dailyReplaySuccessSummary = "s3 lifecycle daily replay completed" - // JobCompleted must set JobType to the handler's jobType constant; // routed requests may leave request.Job.JobType empty, and the admin -// ignores completions with empty JobType. -func sendSuccessCompletion(request *plugin_pb.ExecuteJobRequest, sender pluginworker.ExecutionSender) error { +// ignores completions with empty JobType. Failed passes carry their +// result too, so every bucket failure stays in the job record. +func sendCompletion(request *plugin_pb.ExecuteJobRequest, sender pluginworker.ExecutionSender, report *runReport) error { return sender.SendCompleted(&plugin_pb.JobCompleted{ - JobId: request.Job.JobId, - JobType: jobType, - Success: true, + JobId: request.Job.JobId, + JobType: jobType, + Success: report.err == nil, + ErrorMessage: report.errorMessage(), Result: &plugin_pb.JobResult{ - Summary: dailyReplaySuccessSummary, + Summary: report.summaryLine(), + OutputValues: report.outputValues(), }, }) } @@ -283,14 +286,17 @@ func sendSuccessCompletion(request *plugin_pb.ExecuteJobRequest, sender pluginwo // executeDailyReplay runs one bounded daily-replay pass via // dailyrun.Run. The walker fires inside runShard on rule-content edits // and against the steady-state walk view; all rule kinds are serviced. -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 { +// The returned error covers setup only; the pass's own error is in the +// report. +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) (*runReport, error) { + startedAt := time.Now() eng := engine.New() inputs, parseErrors, err := scheduler.LoadCompileInputs(ctx, filerClient, bucketsPath) if err != nil { - return fmt.Errorf("daily_replay: load lifecycle inputs: %w", err) + return nil, fmt.Errorf("daily_replay: load lifecycle inputs: %w", err) } for _, pe := range parseErrors { - glog.V(1).Infof("daily_replay: %s: %v", pe.Bucket, pe.Err) + glog.Warningf("daily_replay: invalid lifecycle policy on bucket %s: %v", pe.Bucket, pe.Err) } eng.Compile(inputs, engine.CompileOptions{PriorStates: scheduler.AllActivePriorStates(inputs)}) @@ -313,28 +319,33 @@ func (h *Handler) executeDailyReplay(ctx context.Context, request *plugin_pb.Exe buckets = append(buckets, in.Bucket) } } + persister := &dailyrun.FilerCursorPersister{Store: dispatcher.NewFilerStoreClient(filerClient)} + report := &runReport{bucketsTotal: len(buckets) + len(parseErrors), invalidPolicies: parseErrors} + loadLastFullPass(ctx, persister, report) + + runStats := &dailyrun.RunStats{} // Share the limiter with processMatches so walker + replay can't // combine to burst past the cluster cap. - walkerDispatch := &dailyrun.WalkerDispatcher{Client: client, Limiter: limiter} + walkerDispatch := &dailyrun.WalkerDispatcher{Client: client, Limiter: limiter, Stats: runStats} walker := dailyrun.WalkerFunc(func(walkCtx context.Context, view *engine.Snapshot, shards []int) error { return dailyrun.WalkBuckets(walkCtx, view, shards, buckets, - dailyrun.FilerListFunc(filerClient, bucketsPath, view), walkerDispatch) + dailyrun.FilerListFunc(filerClient, bucketsPath, view), walkerDispatch, runStats) }) _ = sender.SendProgress(&plugin_pb.JobProgressUpdate{ JobId: request.Job.JobId, JobType: jobType, State: plugin_pb.JobState_JOB_STATE_RUNNING, Stage: "starting", - Message: fmt.Sprintf("daily_replay shards=%d workers=%d rate=%s buckets=%d walker=on", - len(shards), cfg.Workers, limiterDesc, len(buckets)), + Message: fmt.Sprintf("daily_replay shards=%d workers=%d rate=%s buckets=%d invalid_policies=%d walker=on", + len(shards), cfg.Workers, limiterDesc, len(buckets), len(parseErrors)), }) - runErr := dailyrun.Run(ctx, dailyrun.Config{ + report.err = dailyrun.Run(ctx, dailyrun.Config{ Shards: shards, BucketsPath: bucketsPath, Engine: eng, FilerClient: filerClient, Client: client, - Persister: &dailyrun.FilerCursorPersister{Store: dispatcher.NewFilerStoreClient(filerClient)}, + Persister: persister, Lister: dispatcher.NewFilerSiblingLister(filerClient, bucketsPath), Workers: cfg.Workers, Limiter: limiter, @@ -342,12 +353,50 @@ func (h *Handler) executeDailyReplay(ctx context.Context, request *plugin_pb.Exe Walker: walker, WalkerInterval: cfg.WalkerInterval, ClientName: "worker-s3-lifecycle-daily", + Stats: runStats, }) - if runErr != nil { - glog.Warningf("daily_replay: %v", runErr) - return runErr + report.canceled = ctx.Err() != nil + report.elapsed = time.Since(startedAt) + report.summary = runStats.Summary() + if report.err != nil { + glog.Warningf("daily_replay: %v", report.err) } - return nil + recordFullPass(ctx, persister, report, startedAt) + return report, nil +} + +func loadLastFullPass(ctx context.Context, persister *dailyrun.FilerCursorPersister, report *runReport) { + fp, found, err := persister.LoadFullPass(ctx) + if err != nil { + glog.Warningf("daily_replay: %v", err) + return + } + if found { + report.lastFullPass = fp.CompletedAt + publishFullPass(fp) + } +} + +// recordFullPass advances the full-pass record only when the pass +// covered every shard and every policy. +func recordFullPass(ctx context.Context, persister *dailyrun.FilerCursorPersister, report *runReport, startedAt time.Time) { + if !report.fullPass() { + return + } + fp := dailyrun.FullPass{StartedAt: startedAt.UTC(), CompletedAt: time.Now().UTC()} + saveCtx, saveCancel := context.WithTimeout(ctx, 10*time.Second) + defer saveCancel() + if err := persister.SaveFullPass(saveCtx, fp); err != nil { + glog.Warningf("daily_replay: %v", err) + report.fullPassSaveErr = err + return + } + report.lastFullPass = fp.CompletedAt + publishFullPass(fp) +} + +func publishFullPass(fp dailyrun.FullPass) { + stats.S3LifecycleDailyRunLastFullPassTimestampSeconds.Set(float64(fp.CompletedAt.UnixNano()) / 1e9) } // buildLimiterFromClusterContext parses the per-worker share the admin diff --git a/weed/worker/tasks/s3_lifecycle/handler_test.go b/weed/worker/tasks/s3_lifecycle/handler_test.go index f4a1481bdc..4a43be7906 100644 --- a/weed/worker/tasks/s3_lifecycle/handler_test.go +++ b/weed/worker/tasks/s3_lifecycle/handler_test.go @@ -4,10 +4,16 @@ import ( "context" "errors" "math" + "strings" "testing" + "time" + "github.com/prometheus/client_golang/prometheus/testutil" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" "github.com/seaweedfs/seaweedfs/weed/pb/plugin_pb" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/dailyrun" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/scheduler" + "github.com/seaweedfs/seaweedfs/weed/stats" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "google.golang.org/grpc" @@ -468,15 +474,15 @@ func TestExecute_EmptyJobTypeAccepted(t *testing.T) { assert.Contains(t, err.Error(), "no s3 servers", "validation flowed past the type check") } -// ---------- sendSuccessCompletion ---------- +// ---------- sendCompletion ---------- -func TestSendSuccessCompletion_EmitsCanonicalCompletion(t *testing.T) { +func TestSendCompletion_EmitsCanonicalCompletion(t *testing.T) { sender := &recordingExecSender{} request := &plugin_pb.ExecuteJobRequest{ Job: &plugin_pb.JobSpec{JobId: "job-123"}, // empty JobType (broadcast routing) } - require.NoError(t, sendSuccessCompletion(request, sender)) + require.NoError(t, sendCompletion(request, sender, &runReport{bucketsTotal: 2})) require.Len(t, sender.completed, 1, "exactly one completion must be sent") c := sender.completed[0] @@ -484,20 +490,140 @@ func TestSendSuccessCompletion_EmitsCanonicalCompletion(t *testing.T) { assert.NotEmpty(t, c.JobType, "an empty JobType is dropped by the admin") assert.Equal(t, "job-123", c.JobId, "completion must echo the original JobId") assert.True(t, c.Success, "a clean replay is a success completion") + assert.Empty(t, c.ErrorMessage) require.NotNil(t, c.Result) - assert.NotEmpty(t, c.Result.Summary, "admin UI surfaces a non-empty Result.Summary") + assert.True(t, strings.HasPrefix(c.Result.Summary, "s3 lifecycle daily replay completed:"), c.Result.Summary) + assert.Equal(t, runStatusComplete, c.Result.OutputValues["status"].GetStringValue()) + assert.Equal(t, int64(2), c.Result.OutputValues["buckets_total"].GetInt64Value()) } -func TestSendSuccessCompletion_PropagatesSendError(t *testing.T) { +func TestSendCompletion_PropagatesSendError(t *testing.T) { sentinel := errors.New("stream closed") sender := &recordingExecSender{completedErr: sentinel} request := &plugin_pb.ExecuteJobRequest{Job: &plugin_pb.JobSpec{JobId: "job-456"}} - err := sendSuccessCompletion(request, sender) + err := sendCompletion(request, sender, &runReport{}) require.Error(t, err) assert.ErrorIs(t, err, sentinel) } +func TestSendCompletion_FailedPassKeepsEveryBucketFailure(t *testing.T) { + sender := &recordingExecSender{} + report := &runReport{ + err: errors.New("shard=3: steady walk: walk bad-a: filer down"), + bucketsTotal: 3, + summary: dailyrun.RunSummary{ + BucketsWalked: 1, + BucketsFailed: 2, + BucketFailures: []dailyrun.BucketFailure{ + {Bucket: "bad-a", Error: "filer down"}, + {Bucket: "bad-b", Error: "outcome=BLOCKED"}, + }, + Outcomes: map[string]int64{"DONE": 4, "RPC_ERROR": 1}, + }, + } + require.NoError(t, sendCompletion(&plugin_pb.ExecuteJobRequest{Job: &plugin_pb.JobSpec{JobId: "j"}}, sender, report)) + + require.Len(t, sender.completed, 1) + c := sender.completed[0] + assert.False(t, c.Success) + assert.Contains(t, c.ErrorMessage, "walk bad-a: filer down") + assert.Contains(t, c.ErrorMessage, "2 of 3 buckets failed") + require.NotNil(t, c.Result, "a failed pass still reports its result") + values := c.Result.OutputValues + assert.Equal(t, runStatusFailed, values["status"].GetStringValue()) + assert.Equal(t, int64(2), values["buckets_failed"].GetInt64Value()) + assert.Equal(t, []string{"bad-a: filer down", "bad-b: outcome=BLOCKED"}, values["bucket_failures"].GetStringList().GetValues()) + assert.Equal(t, int64(4), values["rpc_done"].GetInt64Value()) + assert.Equal(t, int64(1), values["rpc_error"].GetInt64Value()) + assert.True(t, strings.HasPrefix(c.Result.Summary, "s3 lifecycle daily replay failed:"), c.Result.Summary) +} + +func TestRunReport_InvalidPolicyIsPartial(t *testing.T) { + report := &runReport{ + bucketsTotal: 2, + invalidPolicies: []scheduler.ParseError{{Bucket: "broken", Err: errors.New("bad xml")}}, + } + assert.Equal(t, runStatusPartial, report.status()) + assert.True(t, strings.HasPrefix(report.summaryLine(), "s3 lifecycle daily replay partial:"), report.summaryLine()) + assert.Contains(t, report.summaryLine(), "invalid_policies=1") + values := report.outputValues() + assert.Equal(t, int64(1), values["invalid_policies"].GetInt64Value()) + assert.Equal(t, []string{"broken: bad xml"}, values["invalid_policy_buckets"].GetStringList().GetValues()) +} + +func TestRunReport_CanceledWinsOverError(t *testing.T) { + report := &runReport{canceled: true, err: context.Canceled} + assert.Equal(t, runStatusCanceled, report.status()) + report.err = nil + assert.Equal(t, runStatusCanceled, report.status(), "a drain cut short returns no error") +} + +// ---------- full pass record ---------- + +type memFilerStore struct { + files map[string][]byte + saveErr error +} + +func (s *memFilerStore) Read(_ context.Context, dir, name string) ([]byte, error) { + b, ok := s.files[dir+"/"+name] + if !ok { + return nil, filer_pb.ErrNotFound + } + return b, nil +} + +func (s *memFilerStore) Save(_ context.Context, dir, name string, content []byte) error { + if s.saveErr != nil { + return s.saveErr + } + s.files[dir+"/"+name] = content + return nil +} + +func TestRecordFullPass_OnlyACompletePassAdvances(t *testing.T) { + previous := dailyrun.FullPass{StartedAt: time.Unix(1_700_000_000, 0).UTC(), CompletedAt: time.Unix(1_700_000_600, 0).UTC()} + for name, report := range map[string]*runReport{ + "failed": {err: errors.New("walk failed")}, + "canceled": {canceled: true}, + "partial": {invalidPolicies: []scheduler.ParseError{{Bucket: "b", Err: errors.New("bad")}}}, + "halted": {summary: dailyrun.RunSummary{ShardsHalted: 1}}, + "deferred": {summary: dailyrun.RunSummary{WalksDeferred: 16}}, + } { + t.Run(name, func(t *testing.T) { + persister := &dailyrun.FilerCursorPersister{Store: &memFilerStore{files: map[string][]byte{}}} + require.NoError(t, persister.SaveFullPass(context.Background(), previous)) + loadLastFullPass(context.Background(), persister, report) + recordFullPass(context.Background(), persister, report, time.Now()) + + got, found, err := persister.LoadFullPass(context.Background()) + require.NoError(t, err) + require.True(t, found) + assert.Equal(t, previous, got) + assert.Equal(t, previous.CompletedAt, report.lastFullPass) + assert.Equal(t, "2023-11-14T22:23:20Z", report.outputValues()["last_full_pass_at"].GetStringValue()) + }) + } + + t.Run("complete", func(t *testing.T) { + persister := &dailyrun.FilerCursorPersister{Store: &memFilerStore{files: map[string][]byte{}}} + require.NoError(t, persister.SaveFullPass(context.Background(), previous)) + report := &runReport{} + startedAt := time.Now().Add(-time.Minute) + recordFullPass(context.Background(), persister, report, startedAt) + + got, found, err := persister.LoadFullPass(context.Background()) + require.NoError(t, err) + require.True(t, found) + assert.True(t, got.CompletedAt.After(previous.CompletedAt)) + assert.Equal(t, startedAt.UTC(), got.StartedAt) + assert.Equal(t, got.CompletedAt, report.lastFullPass) + assert.InDelta(t, float64(got.CompletedAt.UnixNano())/1e9, + testutil.ToFloat64(stats.S3LifecycleDailyRunLastFullPassTimestampSeconds), 1e-3) + }) +} + // ---------- lookupBucketsPath ---------- // stubFilerConfigClient implements filer_pb.SeaweedFilerClient for the @@ -548,3 +674,19 @@ func TestLookupBucketsPath_EmptyDirBucketsFallsBackToDefault(t *testing.T) { require.NoError(t, err) assert.Equal(t, "/buckets", got) } + +func TestRecordFullPass_FailedSaveKeepsThePreviousTimestamp(t *testing.T) { + store := &memFilerStore{files: map[string][]byte{}} + persister := &dailyrun.FilerCursorPersister{Store: store} + previous := dailyrun.FullPass{StartedAt: time.Unix(1_700_000_000, 0).UTC(), CompletedAt: time.Unix(1_700_000_600, 0).UTC()} + require.NoError(t, persister.SaveFullPass(context.Background(), previous)) + + report := &runReport{} + loadLastFullPass(context.Background(), persister, report) + store.saveErr = errors.New("filer unavailable") + recordFullPass(context.Background(), persister, report, time.Now()) + + assert.Equal(t, previous.CompletedAt, report.lastFullPass, "an unsaved pass must not replace the saved timestamp") + assert.InDelta(t, float64(previous.CompletedAt.Unix()), testutil.ToFloat64(stats.S3LifecycleDailyRunLastFullPassTimestampSeconds), 1e-3) + assert.Contains(t, report.outputValues()["full_pass_save_error"].GetStringValue(), "filer unavailable") +} diff --git a/weed/worker/tasks/s3_lifecycle/run_report.go b/weed/worker/tasks/s3_lifecycle/run_report.go new file mode 100644 index 0000000000..3ec2e9cad2 --- /dev/null +++ b/weed/worker/tasks/s3_lifecycle/run_report.go @@ -0,0 +1,142 @@ +package s3_lifecycle + +import ( + "fmt" + "sort" + "strings" + "time" + + "github.com/seaweedfs/seaweedfs/weed/pb/plugin_pb" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/dailyrun" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/scheduler" + "google.golang.org/protobuf/types/known/durationpb" +) + +const ( + runStatusComplete = "complete" + runStatusPartial = "partial" + runStatusFailed = "failed" + runStatusCanceled = "canceled" +) + +const maxReportedInvalidPolicies = 10 + +// runReport is one pass's outcome as the job record shows it. +type runReport struct { + err error + canceled bool + elapsed time.Duration + bucketsTotal int + invalidPolicies []scheduler.ParseError + summary dailyrun.RunSummary + lastFullPass time.Time + fullPassSaveErr error +} + +// Bucket walk and shard failures always surface as err. A pass without +// one is still partial when a policy could not be loaded or a shard's +// replay halted on an unresolved delete. +func (r *runReport) status() string { + switch { + case r.canceled: + return runStatusCanceled + case r.err != nil: + return runStatusFailed + case len(r.invalidPolicies) > 0, r.summary.ShardsHalted > 0: + return runStatusPartial + } + return runStatusComplete +} + +// fullPass also needs every walk-only rule evaluated: a walk deferred by +// WalkerInterval is by design, but it leaves those rules unchecked. +func (r *runReport) fullPass() bool { + return r.status() == runStatusComplete && r.summary.WalksDeferred == 0 +} + +func (r *runReport) summaryLine() string { + verb := r.status() + if verb == runStatusComplete { + verb = "completed" + } + s := r.summary + return fmt.Sprintf("s3 lifecycle daily replay %s: buckets=%d walked=%d failed=%d skipped=%d invalid_policies=%d entries=%d events=%d rpc_done=%d rpc_errors=%d elapsed=%s", + verb, r.bucketsTotal, s.BucketsWalked, s.BucketsFailed, s.BucketsSkipped, len(r.invalidPolicies), + s.EntriesVisited, s.EventsScanned, s.Outcomes["DONE"], s.Outcomes["RPC_ERROR"], r.elapsed.Round(time.Second)) +} + +func (r *runReport) errorMessage() string { + if r.err == nil { + return "" + } + if n := r.summary.BucketsFailed; n > 1 { + return fmt.Sprintf("%v (%d of %d buckets failed)", r.err, n, r.bucketsTotal) + } + return r.err.Error() +} + +func (r *runReport) outputValues() map[string]*plugin_pb.ConfigValue { + s := r.summary + out := map[string]*plugin_pb.ConfigValue{ + "status": stringValue(r.status()), + "buckets_total": int64Value(int64(r.bucketsTotal)), + "buckets_walked": int64Value(int64(s.BucketsWalked)), + "buckets_failed": int64Value(int64(s.BucketsFailed)), + "buckets_skipped": int64Value(int64(s.BucketsSkipped)), + "invalid_policies": int64Value(int64(len(r.invalidPolicies))), + "entries_visited": int64Value(s.EntriesVisited), + "objects_visited": int64Value(s.ObjectsVisited), + "events_scanned": int64Value(s.EventsScanned), + "shard_errors": int64Value(int64(s.ShardErrors)), + "shards_halted": int64Value(int64(s.ShardsHalted)), + "walks_deferred": int64Value(int64(s.WalksDeferred)), + "full_pass": {Kind: &plugin_pb.ConfigValue_BoolValue{BoolValue: r.fullPass()}}, + "elapsed": {Kind: &plugin_pb.ConfigValue_DurationValue{DurationValue: durationpb.New(r.elapsed.Round(time.Millisecond))}}, + } + for outcome, n := range s.Outcomes { + out["rpc_"+strings.ToLower(strings.TrimPrefix(outcome, "RPC_"))] = int64Value(n) + } + if len(s.BucketFailures) > 0 { + failures := make([]string, 0, len(s.BucketFailures)) + for _, f := range s.BucketFailures { + failures = append(failures, f.Bucket+": "+f.Error) + } + out["bucket_failures"] = stringListValue(failures) + } + if len(r.invalidPolicies) > 0 { + invalid := make([]string, 0, maxReportedInvalidPolicies) + for _, pe := range r.invalidPolicies { + if len(invalid) == maxReportedInvalidPolicies { + break + } + invalid = append(invalid, fmt.Sprintf("%s: %v", pe.Bucket, pe.Err)) + } + sort.Strings(invalid) + out["invalid_policy_buckets"] = stringListValue(invalid) + } + if s.Phase != "" { + out["last_phase"] = stringValue(s.Phase) + } + if s.Bucket != "" { + out["last_bucket"] = stringValue(s.Bucket) + } + if r.fullPassSaveErr != nil { + out["full_pass_save_error"] = stringValue(r.fullPassSaveErr.Error()) + } + if !r.lastFullPass.IsZero() { + out["last_full_pass_at"] = stringValue(r.lastFullPass.UTC().Format(time.RFC3339)) + } + return out +} + +func stringValue(v string) *plugin_pb.ConfigValue { + return &plugin_pb.ConfigValue{Kind: &plugin_pb.ConfigValue_StringValue{StringValue: v}} +} + +func int64Value(v int64) *plugin_pb.ConfigValue { + return &plugin_pb.ConfigValue{Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: v}} +} + +func stringListValue(v []string) *plugin_pb.ConfigValue { + return &plugin_pb.ConfigValue{Kind: &plugin_pb.ConfigValue_StringList{StringList: &plugin_pb.StringList{Values: v}}} +}