From 512f898600b6284282cfe0624b5014e4a9cbedb0 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Sat, 14 Nov 2026 10:55:08 -0800 Subject: [PATCH] s3 lifecycle: report work-based job progress during a pass (#12644) * s3 lifecycle: report work-based job progress during a pass The daily pass sent progress only while starting, so a healthy scan that ran for many hours could reach the admin's 24-hour no-progress guard. Worker heartbeats prove the worker is alive, not that this job moves. While dailyrun.Run is in progress the handler now sends a progress update every 5 minutes, and only when the pass's counters (entries visited, replay events, RPC outcomes including errors) moved since the last delivered update. An idle tick sends nothing, so a stalled pass still expires. A failed send is retried on the next tick. The reporter is stopped and joined before the completion is sent, so no progress can follow the terminal result. Co-Authored-By: Claude Opus 5.5 * admin: test that progress, not heartbeats, keeps an old job active Co-Authored-By: Claude Opus 5.5 * s3 lifecycle: label progress phase and bucket as the last started Co-Authored-By: Claude Opus 5.5 --------- Co-authored-by: Claude Opus 5.5 --- weed/admin/plugin/plugin_monitor_test.go | 58 ++++++++ weed/worker/tasks/s3_lifecycle/handler.go | 2 + weed/worker/tasks/s3_lifecycle/progress.go | 71 ++++++++++ .../tasks/s3_lifecycle/progress_test.go | 131 ++++++++++++++++++ 4 files changed, 262 insertions(+) create mode 100644 weed/worker/tasks/s3_lifecycle/progress.go create mode 100644 weed/worker/tasks/s3_lifecycle/progress_test.go diff --git a/weed/admin/plugin/plugin_monitor_test.go b/weed/admin/plugin/plugin_monitor_test.go index 5fe83868d6..e936709c3b 100644 --- a/weed/admin/plugin/plugin_monitor_test.go +++ b/weed/admin/plugin/plugin_monitor_test.go @@ -793,6 +793,64 @@ func TestExpireStaleJobsCancelsTheJobOnItsWorker(t *testing.T) { expectCancelFor(t, session, "req-stale") } +// A job older than the stale limit stays active while its worker reports +// progress for it; the worker's heartbeat alone does not keep it alive. +func TestExpireStaleJobsKeepsAJobThatReportsProgress(t *testing.T) { + t.Parallel() + + pluginSvc, err := New(Options{}) + if err != nil { + t.Fatalf("New: %v", err) + } + defer pluginSvc.Shutdown() + + const workerID = "worker-1" + session := &streamSession{workerID: workerID, outgoing: make(chan *plugin_pb.AdminToWorkerMessage, 8), done: make(chan struct{})} + pluginSvc.putSession(session) + pluginSvc.handleWorkerMessage(session, &plugin_pb.WorkerToAdminMessage{ + Body: &plugin_pb.WorkerToAdminMessage_Hello{Hello: &plugin_pb.WorkerHello{WorkerId: workerID}}, + }) + pluginSvc.trackExecutionStart("req-working", workerID, &plugin_pb.JobSpec{JobId: "job-working", JobType: "s3_lifecycle"}, 1) + pluginSvc.trackExecutionStart("req-stalled", workerID, &plugin_pb.JobSpec{JobId: "job-stalled", JobType: "s3_lifecycle"}, 1) + + startedAt := time.Now().UTC().Add(-defaultStaleActiveJobTimeout - time.Hour) + pluginSvc.jobsMu.Lock() + for _, job := range pluginSvc.jobs { + job.CreatedAt = timeToPtr(startedAt) + job.UpdatedAt = timeToPtr(startedAt) + } + pluginSvc.jobsMu.Unlock() + + pluginSvc.handleWorkerMessage(session, &plugin_pb.WorkerToAdminMessage{ + Body: &plugin_pb.WorkerToAdminMessage_Heartbeat{Heartbeat: &plugin_pb.WorkerHeartbeat{ + WorkerId: workerID, + ExecutionSlotsTotal: 4, + ExecutionSlotsUsed: 2, + RunningWork: []*plugin_pb.RunningWork{ + {WorkId: "job-working", Kind: plugin_pb.WorkKind_WORK_KIND_EXECUTION, JobType: "s3_lifecycle"}, + {WorkId: "job-stalled", Kind: plugin_pb.WorkKind_WORK_KIND_EXECUTION, JobType: "s3_lifecycle"}, + }, + }}, + }) + pluginSvc.handleWorkerMessage(session, &plugin_pb.WorkerToAdminMessage{ + Body: &plugin_pb.WorkerToAdminMessage_JobProgressUpdate{JobProgressUpdate: &plugin_pb.JobProgressUpdate{ + JobId: "job-working", JobType: "s3_lifecycle", RequestId: "req-working", + State: plugin_pb.JobState_JOB_STATE_RUNNING, Stage: "walk", Message: "entries=10", + }}, + }) + + if n := pluginSvc.expireStaleJobs(time.Now().UTC()); n != 1 { + t.Fatalf("expired %d jobs, want only the stalled one", n) + } + if got, _ := pluginSvc.GetTrackedJob("job-working"); got == nil || got.State != "job_state_running" { + t.Fatalf("expected the job reporting progress to stay running, got %+v", got) + } + if got, _ := pluginSvc.GetTrackedJob("job-stalled"); got == nil || got.State != StateFailed { + t.Fatalf("expected the heartbeat-only job to be failed, got %+v", got) + } + expectCancelFor(t, session, "req-stalled") +} + func expectNoCancel(t *testing.T, session *streamSession) { t.Helper() select { diff --git a/weed/worker/tasks/s3_lifecycle/handler.go b/weed/worker/tasks/s3_lifecycle/handler.go index 48c84a704e..2abeb0e8a4 100644 --- a/weed/worker/tasks/s3_lifecycle/handler.go +++ b/weed/worker/tasks/s3_lifecycle/handler.go @@ -347,6 +347,7 @@ func (h *Handler) executeDailyReplay(ctx context.Context, request *plugin_pb.Exe len(shards), cfg.Workers, limiterDesc, len(buckets), len(parseErrors)), }) + stopProgress := startProgressReporter(request, sender, runStats.Summary, progressInterval) report.err = dailyrun.Run(ctx, dailyrun.Config{ Shards: shards, BucketsPath: bucketsPath, @@ -363,6 +364,7 @@ func (h *Handler) executeDailyReplay(ctx context.Context, request *plugin_pb.Exe ClientName: "worker-s3-lifecycle-daily", Stats: runStats, }) + stopProgress() report.canceled = ctx.Err() != nil report.elapsed = time.Since(startedAt) report.summary = runStats.Summary() diff --git a/weed/worker/tasks/s3_lifecycle/progress.go b/weed/worker/tasks/s3_lifecycle/progress.go new file mode 100644 index 0000000000..b97f3f8342 --- /dev/null +++ b/weed/worker/tasks/s3_lifecycle/progress.go @@ -0,0 +1,71 @@ +package s3_lifecycle + +import ( + "fmt" + "time" + + "github.com/seaweedfs/seaweedfs/weed/glog" + "github.com/seaweedfs/seaweedfs/weed/pb/plugin_pb" + pluginworker "github.com/seaweedfs/seaweedfs/weed/plugin/worker" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/dailyrun" +) + +// A long pass must keep refreshing the admin's stale-job clock, but only +// with work it actually did: an idle tick sends nothing. +const progressInterval = 5 * time.Minute + +func runWork(s dailyrun.RunSummary) int64 { + work := s.EntriesVisited + s.EventsScanned + for _, n := range s.Outcomes { + work += n + } + return work +} + +func progressMessage(s dailyrun.RunSummary) string { + return fmt.Sprintf("s3 lifecycle daily replay running: last_phase=%s last_bucket=%s entries=%d objects=%d events=%d rpc_done=%d rpc_errors=%d", + s.Phase, s.Bucket, s.EntriesVisited, s.ObjectsVisited, s.EventsScanned, s.Outcomes["DONE"], s.Outcomes["RPC_ERROR"]) +} + +// startProgressReporter returns stop, which returns only once no further +// progress can be sent, so the caller can send the completion after it. +func startProgressReporter(request *plugin_pb.ExecuteJobRequest, sender pluginworker.ExecutionSender, summarize func() dailyrun.RunSummary, interval time.Duration) (stop func()) { + done := make(chan struct{}) + finished := make(chan struct{}) + go func() { + defer close(finished) + ticker := time.NewTicker(interval) + defer ticker.Stop() + var reported int64 + for { + select { + case <-done: + return + case <-ticker.C: + } + summary := summarize() + work := runWork(summary) + if work == reported { + continue + } + select { + case <-done: + return + default: + } + if err := sender.SendProgress(&plugin_pb.JobProgressUpdate{ + JobId: request.Job.JobId, JobType: jobType, + State: plugin_pb.JobState_JOB_STATE_RUNNING, Stage: "running", + Message: progressMessage(summary), + }); err != nil { + glog.V(1).Infof("s3 lifecycle job %s: send progress: %v", request.Job.JobId, err) + continue + } + reported = work + } + }() + return func() { + close(done) + <-finished + } +} diff --git a/weed/worker/tasks/s3_lifecycle/progress_test.go b/weed/worker/tasks/s3_lifecycle/progress_test.go new file mode 100644 index 0000000000..d50a447e1e --- /dev/null +++ b/weed/worker/tasks/s3_lifecycle/progress_test.go @@ -0,0 +1,131 @@ +package s3_lifecycle + +import ( + "errors" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/weed/pb/plugin_pb" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/dailyrun" + "github.com/stretchr/testify/require" +) + +type progressSender struct { + mu sync.Mutex + progress []*plugin_pb.JobProgressUpdate + failures int + attempts int +} + +func (s *progressSender) SendProgress(p *plugin_pb.JobProgressUpdate) error { + s.mu.Lock() + defer s.mu.Unlock() + s.attempts++ + if s.failures > 0 { + s.failures-- + return errors.New("stream closed") + } + s.progress = append(s.progress, p) + return nil +} + +func (s *progressSender) SendCompleted(*plugin_pb.JobCompleted) error { return nil } + +func (s *progressSender) sent() []*plugin_pb.JobProgressUpdate { + s.mu.Lock() + defer s.mu.Unlock() + return append([]*plugin_pb.JobProgressUpdate(nil), s.progress...) +} + +func (s *progressSender) tries() int { + s.mu.Lock() + defer s.mu.Unlock() + return s.attempts +} + +type fakeRun struct { + entries, events, done, rpcErrors atomic.Int64 +} + +func (f *fakeRun) summary() dailyrun.RunSummary { + return dailyrun.RunSummary{ + Phase: dailyrun.PhaseWalk, + Bucket: "b1", + EntriesVisited: f.entries.Load(), + EventsScanned: f.events.Load(), + Outcomes: map[string]int64{"DONE": f.done.Load(), "RPC_ERROR": f.rpcErrors.Load()}, + } +} + +func progressRequest() *plugin_pb.ExecuteJobRequest { + return &plugin_pb.ExecuteJobRequest{Job: &plugin_pb.JobSpec{JobId: "job-1"}} +} + +const testInterval = 5 * time.Millisecond + +func TestProgressReporter_IdleTicksSendNothing(t *testing.T) { + sender := &progressSender{} + run := &fakeRun{} + stop := startProgressReporter(progressRequest(), sender, run.summary, testInterval) + time.Sleep(20 * testInterval) + stop() + require.Empty(t, sender.sent()) +} + +func TestProgressReporter_ReportsEachKindOfWork(t *testing.T) { + sender := &progressSender{} + run := &fakeRun{} + stop := startProgressReporter(progressRequest(), sender, run.summary, testInterval) + defer stop() + + bumps := []*atomic.Int64{&run.entries, &run.events, &run.done, &run.rpcErrors} + for i, counter := range bumps { + counter.Add(1) + require.Eventually(t, func() bool { return len(sender.sent()) == i+1 }, time.Second, testInterval) + } + time.Sleep(10 * testInterval) + updates := sender.sent() + require.Len(t, updates, len(bumps), "unchanged work must not refresh the job") + + last := updates[len(updates)-1] + require.Equal(t, "job-1", last.JobId) + require.Equal(t, jobType, last.JobType) + require.Equal(t, plugin_pb.JobState_JOB_STATE_RUNNING, last.State) + require.Equal(t, "running", last.Stage) + require.Contains(t, last.Message, "last_phase=walk last_bucket=b1") + require.Contains(t, last.Message, "entries=1") + require.Contains(t, last.Message, "events=1") + require.Contains(t, last.Message, "rpc_done=1") + require.Contains(t, last.Message, "rpc_errors=1") +} + +func TestProgressReporter_RetriesAFailedSendWithoutNewWork(t *testing.T) { + sender := &progressSender{failures: 2} + run := &fakeRun{} + stop := startProgressReporter(progressRequest(), sender, run.summary, testInterval) + defer stop() + + run.entries.Add(3) + require.Eventually(t, func() bool { return len(sender.sent()) == 1 }, time.Second, testInterval) + require.Equal(t, 3, sender.tries()) + time.Sleep(10 * testInterval) + require.Equal(t, 3, sender.tries(), "a delivered report must not be resent") + require.Contains(t, sender.sent()[0].Message, "entries=3") +} + +func TestProgressReporter_NoProgressAfterStop(t *testing.T) { + sender := &progressSender{} + run := &fakeRun{} + stop := startProgressReporter(progressRequest(), sender, run.summary, testInterval) + run.done.Add(1) + require.Eventually(t, func() bool { return len(sender.sent()) == 1 }, time.Second, testInterval) + + stop() + tries := sender.tries() + run.done.Add(1) + run.entries.Add(1) + time.Sleep(10 * testInterval) + require.Equal(t, tries, sender.tries()) +}