mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-08-21 06:36:54 +00:00
feat(s3/lifecycle): dispatcher + per-shard pipeline (Phase 3 PR-D) (#9356)
feat(s3/lifecycle): dispatcher + blocker store + per-shard pipeline Dispatcher consumes due Matches from the schedule, calls LifecycleDelete, and routes outcomes: DONE / NOOP_RESOLVED / SKIPPED_OBJECT_LOCK -> Cursor.Advance RETRY_LATER (within budget) -> re-schedule with backoff RETRY_LATER (budget exhausted) / BLOCKED -> BlockerStore.Put + Freeze BlockerStore is a small interface with InMemoryBlockerStore for tests; the filer-backed impl follows when the worker task registration lands. Pipeline composes Reader + Router + Dispatcher into a single Run loop keyed by shard. Cursor is restored on start, blockers are replayed as freezes, checkpoints write at a configurable cadence, and a final save fires on shutdown. The meta-log itself is the durable buffer for in-flight schedule entries — restart re-derives them from the cursor's MinTsNs.
This commit is contained in:
Chris Lu
@@ -0,0 +1,77 @@
|
||||
package dispatcher
|
||||
|
||||
import (
|
||||
"context"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle"
|
||||
)
|
||||
|
||||
// BlockerRecord persists the state needed to re-freeze a cursor on worker
|
||||
// restart. Stored under /etc/s3/lifecycle/blockers/<shard>/<rule_hash>/<kind>.
|
||||
//
|
||||
// Operator action via the blocker-resolve flow either:
|
||||
// - Quarantine: keep the freeze; mark the (object, version) skipped via a
|
||||
// side-table so subsequent re-evaluations don't trip the same blocker.
|
||||
// - Retry: clear the record and Unfreeze; the cursor advances on the next
|
||||
// successful dispatch.
|
||||
type BlockerRecord struct {
|
||||
ShardID int
|
||||
Key s3lifecycle.ActionKey
|
||||
FrozenAtNs int64 // tsNs of the event that tripped the blocker
|
||||
Reason string // RPC outcome reason
|
||||
CreatedAt time.Time
|
||||
}
|
||||
|
||||
// BlockerStore persists BlockerRecords. The contract:
|
||||
// - Put replaces any existing record for (ShardID, Key) atomically.
|
||||
// - Delete is idempotent (no error on missing).
|
||||
// - List returns all records for ShardID for restart-time freeze replay.
|
||||
type BlockerStore interface {
|
||||
Put(ctx context.Context, rec BlockerRecord) error
|
||||
Delete(ctx context.Context, shardID int, key s3lifecycle.ActionKey) error
|
||||
List(ctx context.Context, shardID int) ([]BlockerRecord, error)
|
||||
}
|
||||
|
||||
// InMemoryBlockerStore is a BlockerStore for tests.
|
||||
type InMemoryBlockerStore struct {
|
||||
mu sync.Mutex
|
||||
records map[int]map[s3lifecycle.ActionKey]BlockerRecord
|
||||
}
|
||||
|
||||
func NewInMemoryBlockerStore() *InMemoryBlockerStore {
|
||||
return &InMemoryBlockerStore{records: map[int]map[s3lifecycle.ActionKey]BlockerRecord{}}
|
||||
}
|
||||
|
||||
func (s *InMemoryBlockerStore) Put(ctx context.Context, rec BlockerRecord) error {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
m, ok := s.records[rec.ShardID]
|
||||
if !ok {
|
||||
m = map[s3lifecycle.ActionKey]BlockerRecord{}
|
||||
s.records[rec.ShardID] = m
|
||||
}
|
||||
m[rec.Key] = rec
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *InMemoryBlockerStore) Delete(ctx context.Context, shardID int, key s3lifecycle.ActionKey) error {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
if m, ok := s.records[shardID]; ok {
|
||||
delete(m, key)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *InMemoryBlockerStore) List(ctx context.Context, shardID int) ([]BlockerRecord, error) {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
m := s.records[shardID]
|
||||
out := make([]BlockerRecord, 0, len(m))
|
||||
for _, rec := range m {
|
||||
out = append(out, rec)
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
@@ -0,0 +1,217 @@
|
||||
package dispatcher
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"github.com/seaweedfs/seaweedfs/weed/glog"
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb/s3_lifecycle_pb"
|
||||
"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle"
|
||||
"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/reader"
|
||||
"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/router"
|
||||
)
|
||||
|
||||
// LifecycleClient abstracts the LifecycleDelete RPC so the dispatcher is
|
||||
// testable without a live S3 server.
|
||||
type LifecycleClient interface {
|
||||
LifecycleDelete(ctx context.Context, req *s3_lifecycle_pb.LifecycleDeleteRequest) (*s3_lifecycle_pb.LifecycleDeleteResponse, error)
|
||||
}
|
||||
|
||||
// Dispatcher consumes due Matches, calls LifecycleDelete, and routes the
|
||||
// outcome back to the per-shard cursor and the blocker store.
|
||||
//
|
||||
// State machine for one Match:
|
||||
// DONE / NOOP_RESOLVED / SKIPPED_OBJECT_LOCK -> Cursor.Advance
|
||||
// RETRY_LATER (within budget) -> back into the schedule with backoff
|
||||
// RETRY_LATER (budget exhausted) -> escalate to BLOCKED
|
||||
// BLOCKED -> BlockerStore.Put + Cursor.Freeze
|
||||
// FATAL_EVENT_ERROR / unknown -> treat as BLOCKED
|
||||
type Dispatcher struct {
|
||||
ShardID int
|
||||
Client LifecycleClient
|
||||
Cursor *reader.Cursor
|
||||
Blockers BlockerStore
|
||||
Schedule *router.Schedule
|
||||
|
||||
// RetryBudget caps RETRY_LATER attempts before escalating to BLOCKED.
|
||||
// Zero defaults to defaultRetryBudget.
|
||||
RetryBudget int
|
||||
|
||||
// RetryBackoff is the wait between RETRY_LATER re-schedules. Zero
|
||||
// defaults to defaultRetryBackoff.
|
||||
RetryBackoff time.Duration
|
||||
|
||||
// retries[Match.Key+ObjectKey] = attempts so far. In-memory only:
|
||||
// worker restart resets the budget, which is fine because BLOCKED is
|
||||
// durable and the cursor is durable.
|
||||
retries map[retryKey]int
|
||||
}
|
||||
|
||||
const (
|
||||
defaultRetryBudget = 5
|
||||
defaultRetryBackoff = 30 * time.Second
|
||||
)
|
||||
|
||||
type retryKey struct {
|
||||
bucket string
|
||||
ruleHash [8]byte
|
||||
kind s3lifecycle.ActionKind
|
||||
objectKey string
|
||||
versionID string
|
||||
}
|
||||
|
||||
func keyOf(m router.Match) retryKey {
|
||||
return retryKey{
|
||||
bucket: m.Key.Bucket,
|
||||
ruleHash: m.Key.RuleHash,
|
||||
kind: m.Key.ActionKind,
|
||||
objectKey: m.ObjectKey,
|
||||
versionID: m.VersionID,
|
||||
}
|
||||
}
|
||||
|
||||
func (d *Dispatcher) budget() int {
|
||||
if d.RetryBudget > 0 {
|
||||
return d.RetryBudget
|
||||
}
|
||||
return defaultRetryBudget
|
||||
}
|
||||
|
||||
func (d *Dispatcher) backoff() time.Duration {
|
||||
if d.RetryBackoff > 0 {
|
||||
return d.RetryBackoff
|
||||
}
|
||||
return defaultRetryBackoff
|
||||
}
|
||||
|
||||
// Tick drains the schedule for due Matches and dispatches each. Returns the
|
||||
// count of Matches processed. Safe to call repeatedly.
|
||||
func (d *Dispatcher) Tick(ctx context.Context, now time.Time) int {
|
||||
if d.retries == nil {
|
||||
d.retries = map[retryKey]int{}
|
||||
}
|
||||
due := d.Schedule.Drain(now)
|
||||
for _, m := range due {
|
||||
if err := ctx.Err(); err != nil {
|
||||
// Re-queue and return; the caller is shutting down.
|
||||
d.Schedule.Add(m)
|
||||
return 0
|
||||
}
|
||||
d.dispatchOne(ctx, m, now)
|
||||
}
|
||||
return len(due)
|
||||
}
|
||||
|
||||
func (d *Dispatcher) dispatchOne(ctx context.Context, m router.Match, now time.Time) {
|
||||
// A frozen cursor means a prior BLOCKED is still active for this
|
||||
// (shard, ActionKey); skip until operator clears it.
|
||||
if d.Cursor.IsFrozen(m.Key) {
|
||||
return
|
||||
}
|
||||
|
||||
ruleHash := m.Key.RuleHash
|
||||
req := &s3_lifecycle_pb.LifecycleDeleteRequest{
|
||||
Bucket: m.Bucket,
|
||||
ObjectPath: m.ObjectKey,
|
||||
VersionId: m.VersionID,
|
||||
RuleHash: ruleHash[:],
|
||||
ActionKind: toProtoActionKind(m.Key.ActionKind),
|
||||
ExpectedIdentity: toProtoIdentity(m.Identity),
|
||||
}
|
||||
resp, err := d.Client.LifecycleDelete(ctx, req)
|
||||
if err != nil {
|
||||
// Transport error: classify as RETRY_LATER. The remote handler
|
||||
// already classifies its own filer-side errors; the only path
|
||||
// that hits this branch is the RPC itself failing.
|
||||
d.handleRetryLater(ctx, m, fmt.Sprintf("RPC: %v", err), now)
|
||||
return
|
||||
}
|
||||
switch resp.Outcome {
|
||||
case s3_lifecycle_pb.LifecycleDeleteOutcome_DONE,
|
||||
s3_lifecycle_pb.LifecycleDeleteOutcome_NOOP_RESOLVED,
|
||||
s3_lifecycle_pb.LifecycleDeleteOutcome_SKIPPED_OBJECT_LOCK:
|
||||
d.advance(m)
|
||||
case s3_lifecycle_pb.LifecycleDeleteOutcome_RETRY_LATER:
|
||||
d.handleRetryLater(ctx, m, resp.Reason, now)
|
||||
case s3_lifecycle_pb.LifecycleDeleteOutcome_BLOCKED:
|
||||
d.handleBlocked(ctx, m, resp.Reason)
|
||||
default:
|
||||
d.handleBlocked(ctx, m, fmt.Sprintf("unknown outcome %v: %s", resp.Outcome, resp.Reason))
|
||||
}
|
||||
}
|
||||
|
||||
func (d *Dispatcher) advance(m router.Match) {
|
||||
delete(d.retries, keyOf(m))
|
||||
d.Cursor.Advance(m.Key, m.EventTs.UnixNano())
|
||||
}
|
||||
|
||||
func (d *Dispatcher) handleRetryLater(ctx context.Context, m router.Match, reason string, now time.Time) {
|
||||
rk := keyOf(m)
|
||||
d.retries[rk]++
|
||||
if d.retries[rk] > d.budget() {
|
||||
d.handleBlocked(ctx, m, fmt.Sprintf("retry budget exhausted: %s", reason))
|
||||
return
|
||||
}
|
||||
// Re-schedule with backoff; same Match, new DueTime.
|
||||
m.DueTime = now.Add(d.backoff())
|
||||
d.Schedule.Add(m)
|
||||
}
|
||||
|
||||
func (d *Dispatcher) handleBlocked(ctx context.Context, m router.Match, reason string) {
|
||||
delete(d.retries, keyOf(m))
|
||||
rec := BlockerRecord{
|
||||
ShardID: d.ShardID,
|
||||
Key: m.Key,
|
||||
FrozenAtNs: m.EventTs.UnixNano(),
|
||||
Reason: reason,
|
||||
CreatedAt: time.Now(),
|
||||
}
|
||||
if err := d.Blockers.Put(ctx, rec); err != nil {
|
||||
// Persistence failure: retry; cursor stays frozen in-memory.
|
||||
glog.Errorf("lifecycle blocker persist: shard=%d key=%v: %v", d.ShardID, m.Key, err)
|
||||
}
|
||||
d.Cursor.Freeze(m.Key, m.EventTs.UnixNano())
|
||||
}
|
||||
|
||||
// ReplayBlockers re-applies in-memory freezes from the durable BlockerStore.
|
||||
// Call once on Pipeline startup before the reader begins emitting.
|
||||
func (d *Dispatcher) ReplayBlockers(ctx context.Context) error {
|
||||
recs, err := d.Blockers.List(ctx, d.ShardID)
|
||||
if err != nil {
|
||||
return fmt.Errorf("blocker list: %w", err)
|
||||
}
|
||||
for _, r := range recs {
|
||||
d.Cursor.Freeze(r.Key, r.FrozenAtNs)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func toProtoActionKind(k s3lifecycle.ActionKind) s3_lifecycle_pb.ActionKind {
|
||||
switch k {
|
||||
case s3lifecycle.ActionKindExpirationDays:
|
||||
return s3_lifecycle_pb.ActionKind_EXPIRATION_DAYS
|
||||
case s3lifecycle.ActionKindExpirationDate:
|
||||
return s3_lifecycle_pb.ActionKind_EXPIRATION_DATE
|
||||
case s3lifecycle.ActionKindNoncurrentDays:
|
||||
return s3_lifecycle_pb.ActionKind_NONCURRENT_DAYS
|
||||
case s3lifecycle.ActionKindNewerNoncurrent:
|
||||
return s3_lifecycle_pb.ActionKind_NEWER_NONCURRENT
|
||||
case s3lifecycle.ActionKindAbortMPU:
|
||||
return s3_lifecycle_pb.ActionKind_ABORT_MPU
|
||||
case s3lifecycle.ActionKindExpiredDeleteMarker:
|
||||
return s3_lifecycle_pb.ActionKind_EXPIRED_DELETE_MARKER
|
||||
}
|
||||
return s3_lifecycle_pb.ActionKind_ACTION_KIND_UNSPECIFIED
|
||||
}
|
||||
|
||||
func toProtoIdentity(id *router.EntryIdentity) *s3_lifecycle_pb.EntryIdentity {
|
||||
if id == nil {
|
||||
return nil
|
||||
}
|
||||
return &s3_lifecycle_pb.EntryIdentity{
|
||||
MtimeNs: id.MtimeNs,
|
||||
Size: id.Size,
|
||||
HeadFid: id.HeadFid,
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,240 @@
|
||||
package dispatcher
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"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/reader"
|
||||
"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/router"
|
||||
)
|
||||
|
||||
type fakeClient struct {
|
||||
calls int
|
||||
respond func(call int) (*s3_lifecycle_pb.LifecycleDeleteResponse, error)
|
||||
}
|
||||
|
||||
func (f *fakeClient) LifecycleDelete(ctx context.Context, req *s3_lifecycle_pb.LifecycleDeleteRequest) (*s3_lifecycle_pb.LifecycleDeleteResponse, error) {
|
||||
f.calls++
|
||||
return f.respond(f.calls)
|
||||
}
|
||||
|
||||
func mkMatch(eventTs time.Time, due time.Time, key string) router.Match {
|
||||
return router.Match{
|
||||
Key: s3lifecycle.ActionKey{Bucket: "bk", ActionKind: s3lifecycle.ActionKindExpirationDays},
|
||||
EventTs: eventTs,
|
||||
DueTime: due,
|
||||
Bucket: "bk",
|
||||
ObjectKey: key,
|
||||
}
|
||||
}
|
||||
|
||||
func newDispatcher(client LifecycleClient) (*Dispatcher, *router.Schedule) {
|
||||
sched := router.NewSchedule()
|
||||
d := &Dispatcher{
|
||||
ShardID: 0,
|
||||
Client: client,
|
||||
Cursor: reader.NewCursor(),
|
||||
Blockers: NewInMemoryBlockerStore(),
|
||||
Schedule: sched,
|
||||
RetryBudget: 3,
|
||||
RetryBackoff: time.Millisecond,
|
||||
}
|
||||
return d, sched
|
||||
}
|
||||
|
||||
func TestDispatchDoneAdvancesCursor(t *testing.T) {
|
||||
client := &fakeClient{
|
||||
respond: func(int) (*s3_lifecycle_pb.LifecycleDeleteResponse, error) {
|
||||
return &s3_lifecycle_pb.LifecycleDeleteResponse{
|
||||
Outcome: s3_lifecycle_pb.LifecycleDeleteOutcome_DONE,
|
||||
}, nil
|
||||
},
|
||||
}
|
||||
d, sched := newDispatcher(client)
|
||||
t0 := time.Now()
|
||||
m := mkMatch(t0, t0, "obj")
|
||||
sched.Add(m)
|
||||
|
||||
got := d.Tick(context.Background(), t0)
|
||||
if got != 1 {
|
||||
t.Fatalf("Tick processed=%d, want 1", got)
|
||||
}
|
||||
if d.Cursor.Get(m.Key) != t0.UnixNano() {
|
||||
t.Fatalf("cursor not advanced: %d", d.Cursor.Get(m.Key))
|
||||
}
|
||||
if d.Cursor.IsFrozen(m.Key) {
|
||||
t.Fatal("cursor should not be frozen")
|
||||
}
|
||||
}
|
||||
|
||||
func TestDispatchNoopAdvancesCursor(t *testing.T) {
|
||||
client := &fakeClient{
|
||||
respond: func(int) (*s3_lifecycle_pb.LifecycleDeleteResponse, error) {
|
||||
return &s3_lifecycle_pb.LifecycleDeleteResponse{
|
||||
Outcome: s3_lifecycle_pb.LifecycleDeleteOutcome_NOOP_RESOLVED,
|
||||
Reason: "STALE_IDENTITY",
|
||||
}, nil
|
||||
},
|
||||
}
|
||||
d, sched := newDispatcher(client)
|
||||
t0 := time.Now()
|
||||
m := mkMatch(t0, t0, "obj")
|
||||
sched.Add(m)
|
||||
d.Tick(context.Background(), t0)
|
||||
if d.Cursor.Get(m.Key) != t0.UnixNano() {
|
||||
t.Fatal("NOOP_RESOLVED should advance cursor")
|
||||
}
|
||||
}
|
||||
|
||||
func TestDispatchRetryLaterReSchedules(t *testing.T) {
|
||||
client := &fakeClient{
|
||||
respond: func(int) (*s3_lifecycle_pb.LifecycleDeleteResponse, error) {
|
||||
return &s3_lifecycle_pb.LifecycleDeleteResponse{
|
||||
Outcome: s3_lifecycle_pb.LifecycleDeleteOutcome_RETRY_LATER,
|
||||
Reason: "TRANSPORT_ERROR",
|
||||
}, nil
|
||||
},
|
||||
}
|
||||
d, sched := newDispatcher(client)
|
||||
t0 := time.Now()
|
||||
m := mkMatch(t0, t0, "obj")
|
||||
sched.Add(m)
|
||||
|
||||
// First tick: dispatched, retry-budget = 1, re-scheduled.
|
||||
d.Tick(context.Background(), t0)
|
||||
if sched.Len() != 1 {
|
||||
t.Fatalf("expected re-schedule on RETRY_LATER, sched.Len=%d", sched.Len())
|
||||
}
|
||||
if d.Cursor.Get(m.Key) != 0 {
|
||||
t.Fatal("cursor must not advance on RETRY_LATER")
|
||||
}
|
||||
if d.Cursor.IsFrozen(m.Key) {
|
||||
t.Fatal("cursor must not freeze within budget")
|
||||
}
|
||||
}
|
||||
|
||||
func TestDispatchRetryBudgetEscalatesToBlocked(t *testing.T) {
|
||||
client := &fakeClient{
|
||||
respond: func(int) (*s3_lifecycle_pb.LifecycleDeleteResponse, error) {
|
||||
return &s3_lifecycle_pb.LifecycleDeleteResponse{
|
||||
Outcome: s3_lifecycle_pb.LifecycleDeleteOutcome_RETRY_LATER,
|
||||
Reason: "stuck",
|
||||
}, nil
|
||||
},
|
||||
}
|
||||
d, sched := newDispatcher(client)
|
||||
d.RetryBudget = 2
|
||||
t0 := time.Now()
|
||||
m := mkMatch(t0, t0, "obj")
|
||||
sched.Add(m)
|
||||
|
||||
// Tick repeatedly; each pushes the re-scheduled entry forward by backoff,
|
||||
// so we advance "now" past each backoff to drain it.
|
||||
now := t0
|
||||
for i := 0; i < 5 && sched.Len() > 0; i++ {
|
||||
now = now.Add(d.RetryBackoff + time.Millisecond)
|
||||
d.Tick(context.Background(), now)
|
||||
}
|
||||
if !d.Cursor.IsFrozen(m.Key) {
|
||||
t.Fatal("expected freeze after budget exhausted")
|
||||
}
|
||||
recs, _ := d.Blockers.List(context.Background(), 0)
|
||||
if len(recs) != 1 {
|
||||
t.Fatalf("expected 1 blocker record, got %d", len(recs))
|
||||
}
|
||||
if recs[0].Reason == "" {
|
||||
t.Fatal("blocker record missing reason")
|
||||
}
|
||||
}
|
||||
|
||||
func TestDispatchBlockedFreezesCursor(t *testing.T) {
|
||||
client := &fakeClient{
|
||||
respond: func(int) (*s3_lifecycle_pb.LifecycleDeleteResponse, error) {
|
||||
return &s3_lifecycle_pb.LifecycleDeleteResponse{
|
||||
Outcome: s3_lifecycle_pb.LifecycleDeleteOutcome_BLOCKED,
|
||||
Reason: "FATAL_EVENT_ERROR: empty bucket",
|
||||
}, nil
|
||||
},
|
||||
}
|
||||
d, sched := newDispatcher(client)
|
||||
t0 := time.Now()
|
||||
m := mkMatch(t0, t0, "obj")
|
||||
sched.Add(m)
|
||||
d.Tick(context.Background(), t0)
|
||||
if !d.Cursor.IsFrozen(m.Key) {
|
||||
t.Fatal("BLOCKED must freeze cursor")
|
||||
}
|
||||
if d.Cursor.Get(m.Key) != t0.UnixNano() {
|
||||
t.Fatal("frozen cursor should be pinned at event ts")
|
||||
}
|
||||
}
|
||||
|
||||
func TestDispatchTransportErrorRetries(t *testing.T) {
|
||||
// gRPC error: classified as RETRY_LATER. After the budget the cursor freezes.
|
||||
client := &fakeClient{
|
||||
respond: func(int) (*s3_lifecycle_pb.LifecycleDeleteResponse, error) {
|
||||
return nil, errors.New("connection refused")
|
||||
},
|
||||
}
|
||||
d, sched := newDispatcher(client)
|
||||
d.RetryBudget = 1
|
||||
t0 := time.Now()
|
||||
m := mkMatch(t0, t0, "obj")
|
||||
sched.Add(m)
|
||||
|
||||
now := t0
|
||||
for i := 0; i < 5 && sched.Len() > 0 && !d.Cursor.IsFrozen(m.Key); i++ {
|
||||
now = now.Add(d.RetryBackoff + time.Millisecond)
|
||||
d.Tick(context.Background(), now)
|
||||
}
|
||||
if !d.Cursor.IsFrozen(m.Key) {
|
||||
t.Fatal("transport-error retries should escalate to BLOCKED past budget")
|
||||
}
|
||||
}
|
||||
|
||||
func TestDispatchSkipsFrozenCursor(t *testing.T) {
|
||||
client := &fakeClient{
|
||||
respond: func(int) (*s3_lifecycle_pb.LifecycleDeleteResponse, error) {
|
||||
t.Fatal("frozen cursor should not call RPC")
|
||||
return nil, nil
|
||||
},
|
||||
}
|
||||
d, sched := newDispatcher(client)
|
||||
t0 := time.Now()
|
||||
m := mkMatch(t0, t0, "obj")
|
||||
d.Cursor.Freeze(m.Key, t0.UnixNano())
|
||||
sched.Add(m)
|
||||
d.Tick(context.Background(), t0)
|
||||
if client.calls != 0 {
|
||||
t.Fatalf("expected 0 RPC calls, got %d", client.calls)
|
||||
}
|
||||
}
|
||||
|
||||
func TestReplayBlockersRefreezes(t *testing.T) {
|
||||
store := NewInMemoryBlockerStore()
|
||||
t0 := time.Now()
|
||||
key := s3lifecycle.ActionKey{Bucket: "bk", ActionKind: s3lifecycle.ActionKindExpirationDays}
|
||||
store.Put(context.Background(), BlockerRecord{
|
||||
ShardID: 0,
|
||||
Key: key,
|
||||
FrozenAtNs: t0.UnixNano(),
|
||||
Reason: "prior run",
|
||||
CreatedAt: t0,
|
||||
})
|
||||
d := &Dispatcher{
|
||||
ShardID: 0,
|
||||
Cursor: reader.NewCursor(),
|
||||
Blockers: store,
|
||||
Schedule: router.NewSchedule(),
|
||||
}
|
||||
if err := d.ReplayBlockers(context.Background()); err != nil {
|
||||
t.Fatalf("ReplayBlockers: %v", err)
|
||||
}
|
||||
if !d.Cursor.IsFrozen(key) {
|
||||
t.Fatal("ReplayBlockers should refreeze cursor")
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,172 @@
|
||||
package dispatcher
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/seaweedfs/seaweedfs/weed/glog"
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
|
||||
"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/engine"
|
||||
"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/reader"
|
||||
"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/router"
|
||||
)
|
||||
|
||||
// Pipeline composes the per-shard reader, router, dispatcher, and cursor
|
||||
// checkpoint into a single Run loop. One Pipeline per (worker, shard).
|
||||
type Pipeline struct {
|
||||
ShardID int
|
||||
BucketsPath string
|
||||
|
||||
Engine *engine.Engine
|
||||
Cursor *reader.Cursor
|
||||
Persister reader.Persister
|
||||
Blockers BlockerStore
|
||||
Client LifecycleClient
|
||||
|
||||
FilerClient filer_pb.SeaweedFilerClient
|
||||
ClientID int32
|
||||
ClientName string
|
||||
|
||||
// Tick cadence for the dispatcher. Zero = defaultDispatchTick.
|
||||
DispatchTick time.Duration
|
||||
// Cadence for cursor checkpoint writes. Zero = defaultCheckpointTick.
|
||||
CheckpointTick time.Duration
|
||||
// EventBudget caps reader events per Run; zero = unbounded (run until
|
||||
// ctx cancellation). Used by the worker scheduler to bound a single
|
||||
// READ task.
|
||||
EventBudget int
|
||||
|
||||
// EventBuffer sets the channel capacity between reader and router
|
||||
// goroutines. Zero = defaultEventBuffer.
|
||||
EventBuffer int
|
||||
}
|
||||
|
||||
const (
|
||||
defaultDispatchTick = 5 * time.Second
|
||||
defaultCheckpointTick = 30 * time.Second
|
||||
defaultEventBuffer = 1024
|
||||
)
|
||||
|
||||
// Run blocks until ctx is canceled or a fatal error occurs. On exit, the
|
||||
// cursor is persisted; in-flight schedule entries are dropped (the meta-log
|
||||
// is the durable buffer, so a restart re-derives them).
|
||||
func (p *Pipeline) Run(ctx context.Context) error {
|
||||
if p.Engine == nil || p.Cursor == nil || p.Persister == nil ||
|
||||
p.Blockers == nil || p.Client == nil || p.FilerClient == nil {
|
||||
return errors.New("pipeline: missing required dependency")
|
||||
}
|
||||
if p.BucketsPath == "" {
|
||||
return errors.New("pipeline: BucketsPath required")
|
||||
}
|
||||
|
||||
// 1. Restore cursor + replay blocker freezes.
|
||||
state, err := p.Persister.Load(ctx, p.ShardID)
|
||||
if err != nil {
|
||||
return fmt.Errorf("cursor load: %w", err)
|
||||
}
|
||||
p.Cursor.Restore(state)
|
||||
|
||||
dispatch := &Dispatcher{
|
||||
ShardID: p.ShardID,
|
||||
Client: p.Client,
|
||||
Cursor: p.Cursor,
|
||||
Blockers: p.Blockers,
|
||||
Schedule: router.NewSchedule(),
|
||||
}
|
||||
if err := dispatch.ReplayBlockers(ctx); err != nil {
|
||||
return fmt.Errorf("blocker replay: %w", err)
|
||||
}
|
||||
|
||||
// 2. Wire reader -> router -> schedule via a buffered channel.
|
||||
bufSize := p.EventBuffer
|
||||
if bufSize <= 0 {
|
||||
bufSize = defaultEventBuffer
|
||||
}
|
||||
events := make(chan *reader.Event, bufSize)
|
||||
rd := &reader.Reader{
|
||||
ShardID: p.ShardID,
|
||||
BucketsPath: p.BucketsPath,
|
||||
Cursor: p.Cursor,
|
||||
Events: events,
|
||||
EventBudget: p.EventBudget,
|
||||
}
|
||||
rd.LogStartup()
|
||||
|
||||
runCtx, cancel := context.WithCancel(ctx)
|
||||
defer cancel()
|
||||
|
||||
var wg sync.WaitGroup
|
||||
var readerErr error
|
||||
|
||||
// Reader goroutine.
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
defer close(events)
|
||||
readerErr = rd.Run(runCtx, p.FilerClient, p.ClientName, p.ClientID)
|
||||
if readerErr != nil && !errors.Is(readerErr, context.Canceled) {
|
||||
glog.Errorf("lifecycle reader: shard=%d: %v", p.ShardID, readerErr)
|
||||
}
|
||||
cancel() // wake the dispatcher goroutine to drain & exit
|
||||
}()
|
||||
|
||||
// Router/dispatcher goroutine: pulls events, routes them, ticks schedule.
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
dispatchTick := p.DispatchTick
|
||||
if dispatchTick <= 0 {
|
||||
dispatchTick = defaultDispatchTick
|
||||
}
|
||||
checkpointTick := p.CheckpointTick
|
||||
if checkpointTick <= 0 {
|
||||
checkpointTick = defaultCheckpointTick
|
||||
}
|
||||
dt := time.NewTicker(dispatchTick)
|
||||
defer dt.Stop()
|
||||
ct := time.NewTicker(checkpointTick)
|
||||
defer ct.Stop()
|
||||
snap := p.Engine.Snapshot()
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-runCtx.Done():
|
||||
dispatch.Tick(context.Background(), time.Now())
|
||||
return
|
||||
case ev, ok := <-events:
|
||||
if !ok {
|
||||
dispatch.Tick(context.Background(), time.Now())
|
||||
return
|
||||
}
|
||||
if snap == nil {
|
||||
snap = p.Engine.Snapshot()
|
||||
}
|
||||
for _, m := range router.Route(snap, ev, time.Now()) {
|
||||
dispatch.Schedule.Add(m)
|
||||
}
|
||||
case <-dt.C:
|
||||
snap = p.Engine.Snapshot() // refresh between tick boundaries
|
||||
dispatch.Tick(runCtx, time.Now())
|
||||
case <-ct.C:
|
||||
if err := p.Persister.Save(runCtx, p.ShardID, p.Cursor.Snapshot()); err != nil {
|
||||
glog.Warningf("lifecycle cursor checkpoint: shard=%d: %v", p.ShardID, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
}()
|
||||
|
||||
wg.Wait()
|
||||
|
||||
// Final cursor checkpoint on graceful shutdown.
|
||||
if err := p.Persister.Save(context.Background(), p.ShardID, p.Cursor.Snapshot()); err != nil {
|
||||
glog.Warningf("lifecycle cursor final save: shard=%d: %v", p.ShardID, err)
|
||||
}
|
||||
|
||||
if readerErr != nil && !errors.Is(readerErr, context.Canceled) {
|
||||
return readerErr
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -0,0 +1,84 @@
|
||||
package dispatcher
|
||||
|
||||
import (
|
||||
"context"
|
||||
"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/engine"
|
||||
"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/reader"
|
||||
"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/router"
|
||||
)
|
||||
|
||||
// TestPipelineIntegrationInMemory exercises router + dispatcher end-to-end
|
||||
// without the Reader (which requires a live filer client). The Reader's
|
||||
// behavior is covered separately in the reader package; this test pins the
|
||||
// composition: an event flows through Route -> Schedule -> Dispatcher and
|
||||
// the cursor advances on a successful RPC.
|
||||
func TestPipelineIntegrationInMemory(t *testing.T) {
|
||||
rule := &s3lifecycle.Rule{ID: "r", Status: s3lifecycle.StatusEnabled, ExpirationDays: 1}
|
||||
hash := s3lifecycle.RuleHash(rule)
|
||||
prior := map[s3lifecycle.ActionKey]engine.PriorState{}
|
||||
for _, k := range s3lifecycle.RuleActionKinds(rule) {
|
||||
prior[s3lifecycle.ActionKey{Bucket: "bk", RuleHash: hash, ActionKind: k}] = engine.PriorState{
|
||||
BootstrapComplete: true,
|
||||
Mode: engine.ModeEventDriven,
|
||||
}
|
||||
}
|
||||
e := engine.New()
|
||||
snap := e.Compile([]engine.CompileInput{{Bucket: "bk", Rules: []*s3lifecycle.Rule{rule}}},
|
||||
engine.CompileOptions{PriorStates: prior})
|
||||
|
||||
now := time.Now()
|
||||
old := now.Add(-48 * time.Hour)
|
||||
|
||||
ev := &reader.Event{
|
||||
TsNs: old.UnixNano(),
|
||||
Bucket: "bk",
|
||||
Key: "obj.txt",
|
||||
NewEntry: &filer_pb.Entry{
|
||||
Name: "obj.txt",
|
||||
Attributes: &filer_pb.FuseAttributes{
|
||||
Mtime: old.Unix(),
|
||||
FileSize: 1,
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
matches := router.Route(snap, ev, now)
|
||||
if len(matches) != 1 {
|
||||
t.Fatalf("expected 1 match, got %v", matches)
|
||||
}
|
||||
|
||||
client := &fakeClient{
|
||||
respond: func(int) (*s3_lifecycle_pb.LifecycleDeleteResponse, error) {
|
||||
return &s3_lifecycle_pb.LifecycleDeleteResponse{
|
||||
Outcome: s3_lifecycle_pb.LifecycleDeleteOutcome_DONE,
|
||||
}, nil
|
||||
},
|
||||
}
|
||||
d, sched := newDispatcher(client)
|
||||
d.ShardID = s3lifecycle.ShardID("bk", "obj.txt")
|
||||
for _, m := range matches {
|
||||
sched.Add(m)
|
||||
}
|
||||
|
||||
processed := d.Tick(context.Background(), now.Add(time.Hour)) // past DueTime
|
||||
if processed != 1 {
|
||||
t.Fatalf("Tick processed=%d, want 1", processed)
|
||||
}
|
||||
if d.Cursor.Get(matches[0].Key) != ev.TsNs {
|
||||
t.Fatalf("cursor not at event TsNs: %d", d.Cursor.Get(matches[0].Key))
|
||||
}
|
||||
}
|
||||
|
||||
func TestPipelineRunRequiresDependencies(t *testing.T) {
|
||||
p := &Pipeline{}
|
||||
err := p.Run(context.Background())
|
||||
if err == nil {
|
||||
t.Fatal("expected error for empty Pipeline")
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user