diff --git a/weed/s3api/s3lifecycle/dailyrun/run_stats.go b/weed/s3api/s3lifecycle/dailyrun/run_stats.go index 6dee7eb07d..4142943b07 100644 --- a/weed/s3api/s3lifecycle/dailyrun/run_stats.go +++ b/weed/s3api/s3lifecycle/dailyrun/run_stats.go @@ -1,10 +1,14 @@ package dailyrun import ( + "context" "strings" "sync" "sync/atomic" + "google.golang.org/grpc" + + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" "github.com/seaweedfs/seaweedfs/weed/stats" ) @@ -46,7 +50,10 @@ type RunSummary struct { EntriesVisited int64 // Distinct object keys among the listed entries. ObjectsVisited int64 - EventsScanned int64 + // Every entry read from the filer, including the ones a walk skips + // without visiting (versioned pre-pass, resume start, other shards). + EntriesListed int64 + EventsScanned int64 // LifecycleDelete outcomes keyed by the dispatch_total outcome label. // They count RPC results, not reclaimed space. Outcomes map[string]int64 @@ -63,6 +70,7 @@ type RunSummary struct { type RunStats struct { entries atomic.Int64 objects atomic.Int64 + listed atomic.Int64 events atomic.Int64 mu sync.Mutex @@ -123,6 +131,41 @@ func (s *RunStats) visitEntry(newObject bool) { } } +// CountListed returns client with its ListEntries results counted as +// EntriesListed. +func (s *RunStats) CountListed(client filer_pb.SeaweedFilerClient) filer_pb.SeaweedFilerClient { + if s == nil || client == nil { + return client + } + return &listCountingClient{SeaweedFilerClient: client, stats: s} +} + +type listCountingClient struct { + filer_pb.SeaweedFilerClient + stats *RunStats +} + +func (c *listCountingClient) ListEntries(ctx context.Context, in *filer_pb.ListEntriesRequest, opts ...grpc.CallOption) (filer_pb.SeaweedFiler_ListEntriesClient, error) { + stream, err := c.SeaweedFilerClient.ListEntries(ctx, in, opts...) + if err != nil { + return nil, err + } + return &listCountingStream{SeaweedFiler_ListEntriesClient: stream, stats: c.stats}, nil +} + +type listCountingStream struct { + filer_pb.SeaweedFiler_ListEntriesClient + stats *RunStats +} + +func (s *listCountingStream) Recv() (*filer_pb.ListEntriesResponse, error) { + resp, err := s.SeaweedFiler_ListEntriesClient.Recv() + if err == nil && resp.GetEntry() != nil { + s.stats.listed.Add(1) + } + return resp, err +} + func (s *RunStats) scanEvent() { if s == nil { return @@ -176,6 +219,7 @@ func (s *RunStats) Summary() RunSummary { out := RunSummary{ EntriesVisited: s.entries.Load(), ObjectsVisited: s.objects.Load(), + EntriesListed: s.listed.Load(), EventsScanned: s.events.Load(), } s.mu.Lock() diff --git a/weed/s3api/s3lifecycle/dailyrun/run_stats_test.go b/weed/s3api/s3lifecycle/dailyrun/run_stats_test.go index a37badd81f..74e2ffca57 100644 --- a/weed/s3api/s3lifecycle/dailyrun/run_stats_test.go +++ b/weed/s3api/s3lifecycle/dailyrun/run_stats_test.go @@ -7,6 +7,7 @@ import ( "testing" "time" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" "github.com/seaweedfs/seaweedfs/weed/pb/s3_lifecycle_pb" "github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle" "github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/bootstrap" @@ -167,3 +168,24 @@ func TestRunStats_NilIsANoOp(t *testing.T) { runStats.setShardErrors(1) assert.Equal(t, RunSummary{}, runStats.Summary()) } + +// A walk can read a bucket without visiting anything: the versioned +// pre-pass reads plain files without emitting them, and a resumed walk +// skips every key up to its start. +func TestRunStats_CountListedCountsSkippedEntries(t *testing.T) { + mtime := time.Now().Add(-7 * 24 * time.Hour) + client := &fakeFiler{tree: map[string][]*filer_pb.Entry{ + "/buckets/bkt": {file("a.txt", mtime, 1), file("b.txt", mtime, 1), file("c.txt", mtime, 1)}, + }} + stats := &RunStats{} + emitted := 0 + require.NoError(t, FilerListFunc(stats.CountListed(client), "/buckets", nil)(context.Background(), "bkt", "c.txt", func(*bootstrap.Entry) error { + emitted++ + return nil + })) + + assert.Zero(t, emitted) + assert.Equal(t, int64(6), stats.Summary().EntriesListed, "both passes over the versioned bucket") + var nilStats *RunStats + assert.Same(t, client, nilStats.CountListed(client)) +} diff --git a/weed/worker/tasks/s3_lifecycle/handler.go b/weed/worker/tasks/s3_lifecycle/handler.go index 2abeb0e8a4..ffb6aa32f3 100644 --- a/weed/worker/tasks/s3_lifecycle/handler.go +++ b/weed/worker/tasks/s3_lifecycle/handler.go @@ -337,7 +337,7 @@ func (h *Handler) executeDailyReplay(ctx context.Context, request *plugin_pb.Exe 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, runStats) + dailyrun.FilerListFunc(runStats.CountListed(filerClient), bucketsPath, view), walkerDispatch, runStats) }) _ = sender.SendProgress(&plugin_pb.JobProgressUpdate{ diff --git a/weed/worker/tasks/s3_lifecycle/progress.go b/weed/worker/tasks/s3_lifecycle/progress.go index b97f3f8342..a4b9c04465 100644 --- a/weed/worker/tasks/s3_lifecycle/progress.go +++ b/weed/worker/tasks/s3_lifecycle/progress.go @@ -15,7 +15,7 @@ import ( const progressInterval = 5 * time.Minute func runWork(s dailyrun.RunSummary) int64 { - work := s.EntriesVisited + s.EventsScanned + work := s.EntriesVisited + s.EntriesListed + s.EventsScanned for _, n := range s.Outcomes { work += n } @@ -23,8 +23,8 @@ func runWork(s dailyrun.RunSummary) int64 { } 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"]) + return fmt.Sprintf("s3 lifecycle daily replay running: last_phase=%s last_bucket=%s entries=%d objects=%d listed=%d events=%d rpc_done=%d rpc_errors=%d", + s.Phase, s.Bucket, s.EntriesVisited, s.ObjectsVisited, s.EntriesListed, s.EventsScanned, s.Outcomes["DONE"], s.Outcomes["RPC_ERROR"]) } // startProgressReporter returns stop, which returns only once no further diff --git a/weed/worker/tasks/s3_lifecycle/progress_test.go b/weed/worker/tasks/s3_lifecycle/progress_test.go index d50a447e1e..0e55693d16 100644 --- a/weed/worker/tasks/s3_lifecycle/progress_test.go +++ b/weed/worker/tasks/s3_lifecycle/progress_test.go @@ -46,7 +46,7 @@ func (s *progressSender) tries() int { } type fakeRun struct { - entries, events, done, rpcErrors atomic.Int64 + entries, listed, events, done, rpcErrors atomic.Int64 } func (f *fakeRun) summary() dailyrun.RunSummary { @@ -54,6 +54,7 @@ func (f *fakeRun) summary() dailyrun.RunSummary { Phase: dailyrun.PhaseWalk, Bucket: "b1", EntriesVisited: f.entries.Load(), + EntriesListed: f.listed.Load(), EventsScanned: f.events.Load(), Outcomes: map[string]int64{"DONE": f.done.Load(), "RPC_ERROR": f.rpcErrors.Load()}, } @@ -80,7 +81,7 @@ func TestProgressReporter_ReportsEachKindOfWork(t *testing.T) { stop := startProgressReporter(progressRequest(), sender, run.summary, testInterval) defer stop() - bumps := []*atomic.Int64{&run.entries, &run.events, &run.done, &run.rpcErrors} + bumps := []*atomic.Int64{&run.entries, &run.listed, &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) @@ -96,6 +97,7 @@ func TestProgressReporter_ReportsEachKindOfWork(t *testing.T) { 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, "listed=1") require.Contains(t, last.Message, "events=1") require.Contains(t, last.Message, "rpc_done=1") require.Contains(t, last.Message, "rpc_errors=1")