diff --git a/adapter/redis_delta_compactor.go b/adapter/redis_delta_compactor.go index fe57ecaa1..4b5ea2f6a 100644 --- a/adapter/redis_delta_compactor.go +++ b/adapter/redis_delta_compactor.go @@ -10,6 +10,7 @@ import ( "github.com/bootjp/elastickv/keyviz" "github.com/bootjp/elastickv/kv" + "github.com/bootjp/elastickv/monitoring" "github.com/bootjp/elastickv/store" "github.com/cockroachdb/errors" ) @@ -65,6 +66,12 @@ type DeltaCompactor struct { // called from a test while Run is active). cursorMu sync.Mutex cursors map[string][]byte + // ttlInlinePlans tracks each TTL-inline scan plan's pass, keyed like + // cursors by the plan's cursor name. Guarded by cursorMu so a pass moves + // together with its plan's cursor. + ttlInlinePlans map[string]*ttlInlineMigrationPlanState + // ttlInlineObserver, when set, receives every completed TTL-inline pass. + ttlInlineObserver monitoring.RedisTTLInlineMigrationObserver // urgentCh receives requests to immediately compact a single key whose // delta count has exceeded MaxDeltaScanLimit, making reads fail. The Run @@ -81,6 +88,11 @@ type DeltaCompactor struct { ttlInlinePlanCursor int ttlInlineGroupsUnknownOnce sync.Once + + // ttlInlineMu lets one tick at a time run the TTL-inline migration, so a + // plan's pages are scanned, tallied and reported in order. A tick that + // finds it held skips the migration. + ttlInlineMu sync.Mutex } // DeltaCompactorOption configures a DeltaCompactor. @@ -134,6 +146,14 @@ func WithDeltaCompactorLogger(l *slog.Logger) DeltaCompactorOption { } } +// WithDeltaCompactorTTLInlineMigrationObserver reports every pass the +// TTL-inline migration completes over an anchor type and data group. +func WithDeltaCompactorTTLInlineMigrationObserver(o monitoring.RedisTTLInlineMigrationObserver) DeltaCompactorOption { + return func(c *DeltaCompactor) { + c.ttlInlineObserver = o + } +} + // NewDeltaCompactor creates a DeltaCompactor that operates on st using coord. func NewDeltaCompactor(st store.MVCCStore, coord kv.Coordinator, opts ...DeltaCompactorOption) *DeltaCompactor { c := &DeltaCompactor{ @@ -145,6 +165,7 @@ func NewDeltaCompactor(st store.MVCCStore, coord kv.Coordinator, opts ...DeltaCo timeout: defaultDeltaCompactorTimeout, timeoutBackoff: defaultDeltaCompactorTimeoutBackoff, cursors: make(map[string][]byte), + ttlInlinePlans: make(map[string]*ttlInlineMigrationPlanState), urgentCh: make(chan urgentCompactionRequest, urgentCompactionChanSize), } for _, opt := range opts { @@ -318,6 +339,9 @@ func (c *DeltaCompactor) SyncOnce(ctx context.Context) error { } tickCtx, cancel := context.WithTimeout(ctx, c.timeout) defer cancel() + // snapshotAt precedes the read timestamp, so a TTL-inline pass that reads + // at readTS never reports a start later than its snapshot. + snapshotAt := time.Now() readTimestamp, err := kv.BeginReadTimestampThrough(tickCtx, c.coord, snapshotTS(c.coord.Clock(), c.st), "redis delta compactor: begin read timestamp") if err != nil { @@ -328,7 +352,7 @@ func (c *DeltaCompactor) SyncOnce(ctx context.Context) error { combined := c.compactBackgroundHandlers(tickCtx, readTS) if tickCtx.Err() == nil { - if err := c.migrateTTLInlineOnce(tickCtx, readTS); err != nil && !errors.Is(err, context.Canceled) { + if err := c.migrateTTLInlineOnce(tickCtx, readTS, snapshotAt); err != nil && !errors.Is(err, context.Canceled) { combined = errors.CombineErrors(combined, err) } } @@ -490,11 +514,19 @@ func (c *DeltaCompactor) scanDeltaRaw(ctx context.Context, h collectionDeltaHand func (c *DeltaCompactor) advanceCursor(typeName string, lastScannedKey []byte, truncated bool) { c.cursorMu.Lock() defer c.cursorMu.Unlock() + c.advanceCursorLocked(typeName, lastScannedKey, truncated) +} + +// advanceCursorLocked is advanceCursor for callers holding cursorMu. It +// reports whether the cursor was reset, i.e. the scan reached the end of the +// prefix and the next one starts from the beginning. +func (c *DeltaCompactor) advanceCursorLocked(typeName string, lastScannedKey []byte, truncated bool) (reset bool) { if truncated && len(lastScannedKey) > 0 { c.cursors[typeName] = bytes.Clone(lastScannedKey) - } else { - delete(c.cursors, typeName) + return false } + delete(c.cursors, typeName) + return true } // compactHandler scans the delta prefix for h, groups entries by userKey, diff --git a/adapter/redis_delta_compactor_test.go b/adapter/redis_delta_compactor_test.go index 247ea507c..b6fedb4e6 100644 --- a/adapter/redis_delta_compactor_test.go +++ b/adapter/redis_delta_compactor_test.go @@ -4,6 +4,9 @@ import ( "bytes" "context" "encoding/binary" + "encoding/json" + "fmt" + "log/slog" "strconv" "sync" "testing" @@ -12,7 +15,9 @@ import ( "github.com/bootjp/elastickv/distribution" "github.com/bootjp/elastickv/keyviz" "github.com/bootjp/elastickv/kv" + "github.com/bootjp/elastickv/monitoring" "github.com/bootjp/elastickv/store" + "github.com/cockroachdb/errors" "github.com/stretchr/testify/require" ) @@ -664,7 +669,7 @@ func TestDeltaCompactor_TTLInlineRotatesPlansAcrossTicksThatRunOutOfBudget(t *te for range plans { ctx, cancel := context.WithCancel(context.Background()) st.endTick = cancel - require.ErrorIs(t, c.migrateTTLInlineOnce(ctx, 10), context.Canceled) + require.ErrorIs(t, c.migrateTTLInlineOnce(ctx, 10, time.Now()), context.Canceled) cancel() } @@ -2279,3 +2284,500 @@ func TestListInlineMetaCompaction(t *testing.T) { require.NoError(t, err) require.Empty(t, remaining, "all list delta keys must be removed after inline compaction") } + +// ttlInlinePassTestCoordinator leads the data groups in groupIDs and, when +// leadsKeys is set, every key. Its failDispatch-th non-empty dispatch fails. +type ttlInlinePassTestCoordinator struct { + *localAdapterCoordinator + + mu sync.Mutex + groupIDs []uint64 + leadsKeys bool + dispatches int + failDispatch int +} + +func (c *ttlInlinePassTestCoordinator) IsLeader() bool { return false } + +func (c *ttlInlinePassTestCoordinator) IsLeaderForKey([]byte) bool { + c.mu.Lock() + defer c.mu.Unlock() + return c.leadsKeys +} + +func (c *ttlInlinePassTestCoordinator) LocalLeaderGroupIDs() ([]uint64, bool) { + c.mu.Lock() + defer c.mu.Unlock() + return append([]uint64(nil), c.groupIDs...), true +} + +func (c *ttlInlinePassTestCoordinator) setGroupIDs(ids []uint64) { + c.mu.Lock() + defer c.mu.Unlock() + c.groupIDs = ids +} + +func (c *ttlInlinePassTestCoordinator) Dispatch(ctx context.Context, req *kv.OperationGroup[kv.OP]) (*kv.CoordinateResponse, error) { + if req != nil && len(req.Elems) > 0 { + c.mu.Lock() + c.dispatches++ + fail := c.dispatches == c.failDispatch + c.mu.Unlock() + if fail { + return nil, errors.New("injected dispatch failure") + } + } + return c.localAdapterCoordinator.Dispatch(ctx, req) +} + +type ttlInlinePassObservation struct { + typeName string + groupID uint64 + outcome string +} + +type recordedTTLInlinePass struct { + ttlInlinePassObservation + startedAt time.Time +} + +type recordingTTLInlinePassObserver struct { + mu sync.Mutex + passes []recordedTTLInlinePass +} + +func (o *recordingTTLInlinePassObserver) ObserveRedisTTLInlineMigrationPass(typeName string, groupID uint64, outcome string, startedAt time.Time) { + o.mu.Lock() + defer o.mu.Unlock() + o.passes = append(o.passes, recordedTTLInlinePass{ + ttlInlinePassObservation: ttlInlinePassObservation{typeName: typeName, groupID: groupID, outcome: outcome}, + startedAt: startedAt, + }) +} + +func (o *recordingTTLInlinePassObserver) forType(typeName string) []ttlInlinePassObservation { + var out []ttlInlinePassObservation + for _, p := range o.recordedForType(typeName) { + out = append(out, p.ttlInlinePassObservation) + } + return out +} + +func (o *recordingTTLInlinePassObserver) recordedForType(typeName string) []recordedTTLInlinePass { + o.mu.Lock() + defer o.mu.Unlock() + var out []recordedTTLInlinePass + for _, p := range o.passes { + if p.typeName == typeName { + out = append(out, p) + } + } + return out +} + +type lockedLogBuffer struct { + mu sync.Mutex + buf bytes.Buffer +} + +func (b *lockedLogBuffer) Write(p []byte) (int, error) { + b.mu.Lock() + defer b.mu.Unlock() + n, err := b.buf.Write(p) + return n, errors.WithStack(err) +} + +// ttlInlinePassLine is the part of a "pass complete" log line the tests check. +type ttlInlinePassLine struct { + Level string `json:"level"` + Type string `json:"type"` + GroupID uint64 `json:"group_id"` + Clean bool `json:"clean"` + FullPass bool `json:"full_pass"` + Pages int `json:"pages"` + KeysScanned int `json:"keys_scanned"` + KeysMigrated int `json:"keys_migrated"` + KeysSkipped int `json:"keys_skipped"` + KeysFailed int `json:"keys_failed"` +} + +func (b *lockedLogBuffer) ttlInlinePasses(t *testing.T, typeName string) []ttlInlinePassLine { + t.Helper() + b.mu.Lock() + defer b.mu.Unlock() + var out []ttlInlinePassLine + for _, line := range bytes.Split(b.buf.Bytes(), []byte("\n")) { + if len(line) == 0 { + continue + } + var rec struct { + Msg string `json:"msg"` + ttlInlinePassLine + } + require.NoError(t, json.Unmarshal(line, &rec)) + if rec.Msg == "ttl inline migrator: pass complete" && rec.Type == typeName { + out = append(out, rec.ttlInlinePassLine) + } + } + return out +} + +// wantTTLInlinePassOutcome is the monitoring outcome a pass should report: a +// pass that is not clean but found nothing to rewrite, skip or fail did not +// read its whole prefix, so it is partial. +func wantTTLInlinePassOutcome(clean, foundNothing bool) string { + switch { + case clean: + return monitoring.RedisTTLInlinePassClean + case foundNothing: + return monitoring.RedisTTLInlinePassPartial + default: + return monitoring.RedisTTLInlinePassNotClean + } +} + +func seedTTLInlinePassStrings(t *testing.T, st store.MVCCStore, n uint64, legacy bool) { + t.Helper() + ctx := context.Background() + for i := uint64(0); i < n; i++ { + key := []byte(fmt.Sprintf("ttl:pass:%04d", i)) + value := []byte("value") + if !legacy { + value = encodeRedisStr(value, nil) + } + require.NoError(t, st.PutAt(ctx, redisStrKey(key), value, i+1, 0)) + } +} + +// The migrator reports each pass a scan plan completes over its prefix, so an +// operator can tell from a clean pass on every anchor type and group that the +// legacy TTL backlog has drained. A pass runs from the first page of the +// prefix to the first page that is not truncated, its counters restart with +// the next pass, and a pass that resumed a cursor from an earlier leadership +// term, migrated keys, or left anchors unchecked is not clean. +func TestDeltaCompactor_TTLInlineReportsCompletedPasses(t *testing.T) { + t.Parallel() + + const typeName = "ttl-inline-string" + group1 := []uint64{1} + type pass struct { + level string + clean bool + full bool + pages int + keysScanned int + keysMigrated int + keysSkipped int + keysFailed int + } + clean := func(level string, pages, scanned int) pass { + return pass{level: level, clean: true, full: true, pages: pages, keysScanned: scanned} + } + tests := []struct { + name string + seed func(*testing.T, store.MVCCStore) + notLeader bool + failDispatch int + // ticks lists the groups this node leads at each tick. + ticks [][]uint64 + want []pass + }{ + { + name: "clean pass", + seed: func(t *testing.T, st store.MVCCStore) { seedTTLInlinePassStrings(t, st, 3, false) }, + ticks: [][]uint64{group1}, + want: []pass{clean("INFO", 1, 3)}, + }, + { + name: "pass that migrated keys, then clean passes", + seed: func(t *testing.T, st store.MVCCStore) { seedTTLInlinePassStrings(t, st, 3, true) }, + ticks: [][]uint64{group1, group1, group1}, + want: []pass{ + {level: "INFO", full: true, pages: 1, keysScanned: 3, keysMigrated: 3}, + clean("INFO", 1, 3), + clean("DEBUG", 1, 3), + }, + }, + { + name: "pass spanning several ticks", + seed: func(t *testing.T, st store.MVCCStore) { + seedTTLInlinePassStrings(t, st, ttlInlineMigrationTickScanLimit+88, true) + }, + ticks: [][]uint64{group1, group1, group1, group1}, + want: []pass{ + {level: "INFO", full: true, pages: 2, keysScanned: ttlInlineMigrationTickScanLimit + 88, keysMigrated: ttlInlineMigrationTickScanLimit + 88}, + clean("INFO", 2, ttlInlineMigrationTickScanLimit+88), + }, + }, + { + name: "leadership moves away mid-pass and back", + seed: func(t *testing.T, st store.MVCCStore) { + seedTTLInlinePassStrings(t, st, ttlInlineMigrationTickScanLimit+88, false) + }, + ticks: [][]uint64{group1, {}, group1, group1, group1}, + want: []pass{ + {level: "INFO", full: false, pages: 1, keysScanned: 88}, + clean("INFO", 2, ttlInlineMigrationTickScanLimit+88), + }, + }, + { + name: "anchors this node does not lead", + seed: func(t *testing.T, st store.MVCCStore) { seedTTLInlinePassStrings(t, st, 3, false) }, + notLeader: true, + ticks: [][]uint64{group1}, + want: []pass{{level: "INFO", full: true, pages: 1, keysScanned: 3, keysSkipped: 3}}, + }, + { + name: "anchor whose rewrite cannot be built", + seed: func(t *testing.T, st store.MVCCStore) { + ctx := context.Background() + require.NoError(t, st.PutAt(ctx, redisStrKey([]byte("ttl:pass:bad")), []byte("value"), 1, 0)) + require.NoError(t, st.PutAt(ctx, redisTTLKey([]byte("ttl:pass:bad")), []byte{1}, 2, 0)) + }, + ticks: [][]uint64{group1}, + want: []pass{{level: "INFO", full: true, pages: 1, keysScanned: 1, keysFailed: 1}}, + }, + { + // The first batch of 32 commits and the second fails. The next + // tick reads the page again and rewrites the 33rd key; the pass + // counts the page once and both batches' keys. + name: "page retried after a failed batch", + seed: func(t *testing.T, st store.MVCCStore) { + seedTTLInlinePassStrings(t, st, ttlInlineMigrationBatchKeyLimit+1, true) + }, + failDispatch: 2, + ticks: [][]uint64{group1, group1}, + want: []pass{ + {level: "INFO", full: true, pages: 1, keysScanned: ttlInlineMigrationBatchKeyLimit + 1, keysMigrated: ttlInlineMigrationBatchKeyLimit + 1}, + }, + }, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + ctx := context.Background() + st := &recordingTTLInlineGroupScanStore{MVCCStore: store.NewMVCCStore()} + tc.seed(t, st) + coord := &ttlInlinePassTestCoordinator{ + localAdapterCoordinator: newLocalAdapterCoordinator(st), + leadsKeys: !tc.notLeader, + failDispatch: tc.failDispatch, + } + logs := &lockedLogBuffer{} + observer := &recordingTTLInlinePassObserver{} + c := NewDeltaCompactor(st, coord, + WithDeltaCompactorTimeout(time.Minute), + WithDeltaCompactorLogger(slog.New(slog.NewJSONHandler(logs, &slog.HandlerOptions{Level: slog.LevelDebug}))), + WithDeltaCompactorTTLInlineMigrationObserver(observer), + ) + + for _, groups := range tc.ticks { + coord.setGroupIDs(groups) + err := c.SyncOnce(ctx) + if tc.failDispatch == 0 { + require.NoError(t, err) + } + } + + lines := logs.ttlInlinePasses(t, typeName) + got := make([]pass, 0, len(lines)) + for _, l := range lines { + require.Equal(t, uint64(1), l.GroupID) + got = append(got, pass{ + level: l.Level, clean: l.Clean, full: l.FullPass, pages: l.Pages, + keysScanned: l.KeysScanned, keysMigrated: l.KeysMigrated, + keysSkipped: l.KeysSkipped, keysFailed: l.KeysFailed, + }) + } + require.Equal(t, tc.want, got) + + wantObserved := make([]ttlInlinePassObservation, 0, len(tc.want)) + for _, p := range tc.want { + outcome := wantTTLInlinePassOutcome(p.clean, p.keysMigrated+p.keysSkipped+p.keysFailed == 0) + wantObserved = append(wantObserved, ttlInlinePassObservation{typeName: typeName, groupID: 1, outcome: outcome}) + } + require.Equal(t, wantObserved, observer.forType(typeName)) + }) + } +} + +// hookedTTLInlineScanStore runs onScan before each group scan of prefix. +type hookedTTLInlineScanStore struct { + store.MVCCStore + + prefix []byte + onScan func() +} + +func (s *hookedTTLInlineScanStore) ScanGroupOwnedAt(ctx context.Context, groupID uint64, start []byte, end []byte, limit int, ts uint64) ([]*store.KVPair, []byte, error) { + if bytes.HasPrefix(start, s.prefix) && s.onScan != nil { + s.onScan() + } + return scanPageWithResume(ctx, s.MVCCStore, start, end, limit, ts) +} + +func newHookedTTLInlinePassCompactor(t *testing.T, st *hookedTTLInlineScanStore) (*DeltaCompactor, *lockedLogBuffer) { + t.Helper() + coord := &ttlInlinePassTestCoordinator{ + localAdapterCoordinator: newLocalAdapterCoordinator(st), + groupIDs: []uint64{1}, + leadsKeys: true, + } + logs := &lockedLogBuffer{} + c := NewDeltaCompactor(st, coord, + WithDeltaCompactorTimeout(time.Minute), + WithDeltaCompactorLogger(slog.New(slog.NewJSONHandler(logs, &slog.HandlerOptions{Level: slog.LevelDebug}))), + ) + return c, logs +} + +// TTL-inline migration ticks never overlap. A tick that starts while another +// is migrating skips migration, so a page is scanned, tallied and reported by +// one tick only. Overlapping ticks double-counted a middle page; on a one-page +// plan, where the cursor resets to empty, the second tick even reported the +// same page as a second pass. +func TestDeltaCompactor_TTLInlineTicksDoNotOverlap(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + keys uint64 + ticks int + pages int + }{ + {name: "one-page plan", keys: 3, ticks: 1, pages: 1}, + {name: "middle page", keys: 2*ttlInlineMigrationTickScanLimit + 88, ticks: 3, pages: 3}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + ctx := context.Background() + st := &hookedTTLInlineScanStore{MVCCStore: store.NewMVCCStore(), prefix: []byte(redisStrPrefix)} + seedTTLInlinePassStrings(t, st, tc.keys, false) + c, logs := newHookedTTLInlinePassCompactor(t, st) + + // The scan of the last page runs a whole tick in between, as a + // concurrent SyncOnce would. + scans, nested := 0, false + st.onScan = func() { + scans++ + if scans == tc.pages && !nested { + nested = true + require.NoError(t, c.SyncOnce(ctx)) + } + } + for range tc.ticks { + require.NoError(t, c.SyncOnce(ctx)) + } + + require.Equal(t, tc.pages, scans, "the overlapping tick must not scan") + passes := logs.ttlInlinePasses(t, "ttl-inline-string") + require.Len(t, passes, 1) + require.Equal(t, tc.pages, passes[0].Pages) + require.EqualValues(t, tc.keys, passes[0].KeysScanned) + require.True(t, passes[0].Clean) + }) + } +} + +// A pass's duration includes its first page: the pass starts before that +// page's scan, not when the page has been migrated. +func TestDeltaCompactor_TTLInlinePassDurationIncludesFirstPage(t *testing.T) { + t.Parallel() + + const scanDelay = 50 * time.Millisecond + st := &hookedTTLInlineScanStore{ + MVCCStore: store.NewMVCCStore(), + prefix: []byte(redisStrPrefix), + onScan: func() { time.Sleep(scanDelay) }, + } + seedTTLInlinePassStrings(t, st, 3, false) + c, logs := newHookedTTLInlinePassCompactor(t, st) + observer := &recordingTTLInlinePassObserver{} + c.ttlInlineObserver = observer + + before := time.Now() + require.NoError(t, c.SyncOnce(context.Background())) + after := time.Now() + + durations := logs.ttlInlinePassDurations(t, "ttl-inline-string") + require.Len(t, durations, 1) + require.GreaterOrEqual(t, durations[0], scanDelay) + + // The observer gets the pass's start, which precedes its first page's + // scan: a gate can then require a pass that started after an event. + passes := observer.recordedForType("ttl-inline-string") + require.Len(t, passes, 1) + require.False(t, passes[0].startedAt.Before(before)) + require.LessOrEqual(t, passes[0].startedAt, after.Add(-scanDelay)) +} + +func (b *lockedLogBuffer) ttlInlinePassDurations(t *testing.T, typeName string) []time.Duration { + t.Helper() + b.mu.Lock() + defer b.mu.Unlock() + var out []time.Duration + for _, line := range bytes.Split(b.buf.Bytes(), []byte("\n")) { + if len(line) == 0 { + continue + } + var rec struct { + Msg string `json:"msg"` + Type string `json:"type"` + Duration time.Duration `json:"duration"` + } + require.NoError(t, json.Unmarshal(line, &rec)) + if rec.Msg == "ttl inline migrator: pass complete" && rec.Type == typeName { + out = append(out, rec.Duration) + } + } + return out +} + +// snapshotTimingCoordinator records when the compactor first asks for its +// clock, which is when SyncOnce acquires the tick's read timestamp, and then +// delays, as a slow read-timestamp acquisition would. +type snapshotTimingCoordinator struct { + *ttlInlinePassTestCoordinator + + once sync.Once + snapshotAt time.Time +} + +func (c *snapshotTimingCoordinator) Clock() *kv.HLC { + c.once.Do(func() { + c.snapshotAt = time.Now() + time.Sleep(10 * time.Millisecond) + }) + return c.ttlInlinePassTestCoordinator.Clock() +} + +// A pass reads its pages at the tick's read timestamp, so it starts no later +// than that timestamp is acquired. Stamping it when the scan began let a route +// change between the two give a pass that read a snapshot from before the +// change a start after it. +func TestDeltaCompactor_TTLInlinePassStartsNoLaterThanItsReadTimestamp(t *testing.T) { + t.Parallel() + + st := &recordingTTLInlineGroupScanStore{MVCCStore: store.NewMVCCStore()} + seedTTLInlinePassStrings(t, st, 3, false) + coord := &snapshotTimingCoordinator{ttlInlinePassTestCoordinator: &ttlInlinePassTestCoordinator{ + localAdapterCoordinator: newLocalAdapterCoordinator(st), + groupIDs: []uint64{1}, + leadsKeys: true, + }} + observer := &recordingTTLInlinePassObserver{} + c := NewDeltaCompactor(st, coord, + WithDeltaCompactorTimeout(time.Minute), + WithDeltaCompactorTTLInlineMigrationObserver(observer), + ) + + require.NoError(t, c.SyncOnce(context.Background())) + + passes := observer.recordedForType("ttl-inline-string") + require.Len(t, passes, 1) + require.False(t, coord.snapshotAt.IsZero(), "SyncOnce must read the coordinator's clock") + require.LessOrEqual(t, passes[0].startedAt, coord.snapshotAt) +} diff --git a/adapter/redis_ttl_inline_migration_pass.go b/adapter/redis_ttl_inline_migration_pass.go new file mode 100644 index 000000000..05d84d81b --- /dev/null +++ b/adapter/redis_ttl_inline_migration_pass.go @@ -0,0 +1,185 @@ +package adapter + +import ( + "context" + "log/slog" + "time" + + "github.com/bootjp/elastickv/monitoring" +) + +// ttlInlineMigrationTally counts what the TTL-inline migrator did with the +// anchors it read. +type ttlInlineMigrationTally struct { + // keysScanned counts the anchors read, i.e. the rows whose user key the + // handler could extract. + keysScanned int + // keysMigrated counts the anchors rewritten by a dispatched batch. + keysMigrated int + // keysSkipped counts the anchors left alone because this node did not + // lead their user key's route. + keysSkipped int + // keysFailed counts the anchors whose rewrite could not be built. + keysFailed int +} + +func (t *ttlInlineMigrationTally) add(o ttlInlineMigrationTally) { + t.keysScanned += o.keysScanned + t.keysMigrated += o.keysMigrated + t.keysSkipped += o.keysSkipped + t.keysFailed += o.keysFailed +} + +// ttlInlineMigrationPass is one pass of a scan plan over its prefix: the pages +// from the start of the prefix through the page that reaches its end. +type ttlInlineMigrationPass struct { + ttlInlineMigrationTally + // full reports that the pass started at the beginning of the prefix and + // its group stayed in every tick's plans since. A pass that resumes a + // cursor left before its group dropped out of the plans is not full. + full bool + pages int + startedAt time.Time +} + +// clean reports that the pass read every anchor of its plan and found each one +// already in the inline format: it read the whole prefix, rewrote nothing, and +// left nothing unchecked. +func (p ttlInlineMigrationPass) clean() bool { + return p.full && p.foundNothing() +} + +// foundNothing reports that the pass rewrote, skipped and failed no anchor. +func (p ttlInlineMigrationPass) foundNothing() bool { + return p.keysMigrated == 0 && p.keysSkipped == 0 && p.keysFailed == 0 +} + +// outcome is the pass's monitoring outcome: clean, partial (found nothing but +// did not read the whole prefix), or not_clean. +func (p ttlInlineMigrationPass) outcome() string { + switch { + case p.clean(): + return monitoring.RedisTTLInlinePassClean + case p.foundNothing(): + return monitoring.RedisTTLInlinePassPartial + default: + return monitoring.RedisTTLInlinePassNotClean + } +} + +// ttlInlineMigrationPlanState is a scan plan's pass bookkeeping. +type ttlInlineMigrationPlanState struct { + // pass is the open pass, nil between passes. + pass *ttlInlineMigrationPass + // lastClean reports that the plan's previous pass was clean. It keeps + // repeated clean passes out of the info log. + lastClean bool +} + +// openTTLInlineMigrationPassLocked returns the open pass of the plan named +// name, starting one if needed. A page that starts at the beginning of the +// prefix always opens a new pass unless the open pass has not completed a page +// yet. A pass the page opens starts at the page's snapshot time, no later than +// its read timestamp was acquired. Callers hold cursorMu. +func (c *DeltaCompactor) openTTLInlineMigrationPassLocked(name string, page ttlInlineMigrationPage) *ttlInlineMigrationPass { + state := c.ttlInlinePlans[name] + if state == nil { + state = &ttlInlineMigrationPlanState{} + c.ttlInlinePlans[name] = state + } + if state.pass == nil || (page.fromStart && state.pass.pages > 0) { + state.pass = &ttlInlineMigrationPass{full: page.fromStart, startedAt: page.snapshotAt} + } + return state.pass +} + +// noteTTLInlineMigrationPageFailed adds the keys a failed page already +// rewrote to its plan's pass. The page itself does not count: its cursor stays, +// and the next tick reads it again. +func (c *DeltaCompactor) noteTTLInlineMigrationPageFailed(name string, page ttlInlineMigrationPage, migrated int) { + if migrated == 0 { + return + } + c.cursorMu.Lock() + defer c.cursorMu.Unlock() + c.openTTLInlineMigrationPassLocked(name, page).keysMigrated += migrated +} + +// finishTTLInlineMigrationPage adds a migrated page to its plan's pass and +// moves the plan's cursor past it. When the page reached the end of the prefix +// it closes the pass and returns it with done set, and whether the plan's +// previous pass was clean. +func (c *DeltaCompactor) finishTTLInlineMigrationPage( + name string, + page ttlInlineMigrationPage, + tally ttlInlineMigrationTally, +) (pass ttlInlineMigrationPass, prevClean bool, done bool) { + c.cursorMu.Lock() + defer c.cursorMu.Unlock() + open := c.openTTLInlineMigrationPassLocked(name, page) + open.add(tally) + open.pages++ + if !c.advanceCursorLocked(name, page.cursorKey, page.truncated) { + return ttlInlineMigrationPass{}, false, false + } + state := c.ttlInlinePlans[name] + pass, prevClean = *open, state.lastClean + state.pass = nil + state.lastClean = pass.clean() + return pass, prevClean, true +} + +// pruneTTLInlineMigrationPlanStates drops the bookkeeping of plans this tick +// no longer has, such as a group whose leadership moved to another node. The +// plan's cursor stays, so migration resumes where it stopped if the group +// comes back; the pass it resumes is not full and so cannot be clean. +func (c *DeltaCompactor) pruneTTLInlineMigrationPlanStates(plans []ttlInlineMigrationTickPlan) { + c.cursorMu.Lock() + defer c.cursorMu.Unlock() + if len(c.ttlInlinePlans) == 0 { + return + } + live := make(map[string]struct{}, len(plans)) + for _, p := range plans { + live[p.plan.cursorName] = struct{}{} + } + for name := range c.ttlInlinePlans { + if _, ok := live[name]; !ok { + delete(c.ttlInlinePlans, name) + } + } +} + +// reportTTLInlineMigrationPass logs a completed pass and hands it to the +// observer. A pass that is not clean, or the first clean pass after one that +// was not, is logged at info; repeated clean passes are logged at debug so an +// idle migrator does not log every tick. group_id 0 is the whole-prefix scan +// used when the coordinator does not report its local leader groups. +func (c *DeltaCompactor) reportTTLInlineMigrationPass( + ctx context.Context, + h ttlInlineMigrationHandler, + plan ttlInlineMigrationScanPlan, + pass ttlInlineMigrationPass, + prevClean bool, +) { + clean := pass.clean() + level := slog.LevelInfo + if clean && prevClean { + level = slog.LevelDebug + } + c.logger.Log(ctx, level, "ttl inline migrator: pass complete", + "type", h.typeName, + "group_id", plan.groupID, + "clean", clean, + "full_pass", pass.full, + "pages", pass.pages, + "keys_scanned", pass.keysScanned, + "keys_migrated", pass.keysMigrated, + "keys_skipped", pass.keysSkipped, + "keys_failed", pass.keysFailed, + "duration", time.Since(pass.startedAt), + ) + if c.ttlInlineObserver != nil { + c.ttlInlineObserver.ObserveRedisTTLInlineMigrationPass(h.typeName, plan.groupID, pass.outcome(), pass.startedAt) + } +} diff --git a/adapter/redis_ttl_inline_migrator.go b/adapter/redis_ttl_inline_migrator.go index 2f992ea10..6d0e4466e 100644 --- a/adapter/redis_ttl_inline_migrator.go +++ b/adapter/redis_ttl_inline_migrator.go @@ -49,8 +49,17 @@ type ttlInlineMigrationTickPlan struct { // the last one the previous tick ran, as compactBackgroundHandlers rotates its // handlers; starting from the first plan every tick starved the later handlers // and groups whenever the earlier plans used up the budget. -func (c *DeltaCompactor) migrateTTLInlineOnce(ctx context.Context, readTS uint64) error { +// +// snapshotAt is a wall-clock time no later than the acquisition of readTS. A +// pass that opens this tick reports it as its start. +func (c *DeltaCompactor) migrateTTLInlineOnce(ctx context.Context, readTS uint64, snapshotAt time.Time) error { + if !c.ttlInlineMu.TryLock() { + c.logger.DebugContext(ctx, "ttl inline migrator: another tick is migrating; skipping") + return nil + } + defer c.ttlInlineMu.Unlock() plans := c.ttlInlineMigrationTickPlans() + c.pruneTTLInlineMigrationPlanStates(plans) start := c.rotationStart(&c.ttlInlinePlanCursor, len(plans)) var combined error for offset := range plans { @@ -58,7 +67,7 @@ func (c *DeltaCompactor) migrateTTLInlineOnce(ctx context.Context, readTS uint64 return errors.WithStack(errors.CombineErrors(combined, ctx.Err())) } idx := (start + offset) % len(plans) - err := c.migrateTTLInlineScanPlan(ctx, plans[idx].handler, plans[idx].plan, readTS) + err := c.migrateTTLInlineScanPlan(ctx, plans[idx].handler, plans[idx].plan, readTS, snapshotAt) c.advanceRotation(&c.ttlInlinePlanCursor, idx, len(plans)) if err != nil { combined = errors.CombineErrors(combined, err) @@ -211,49 +220,85 @@ func (c *DeltaCompactor) migrateTTLInlineScanPlan( h ttlInlineMigrationHandler, plan ttlInlineMigrationScanPlan, readTS uint64, + snapshotAt time.Time, ) error { - kvs, cursorKey, truncated, err := c.scanTTLInlineMigrationRaw(ctx, h, plan, readTS) + page, err := c.scanTTLInlineMigrationRaw(ctx, h, plan, readTS, snapshotAt) + if err != nil { + return err + } + tally, err := c.migrateTTLInlinePage(ctx, h, page.kvs, readTS) if err != nil { + // The next tick reads this page again, so only the keys its earlier + // batches already rewrote count toward the pass. + c.noteTTLInlineMigrationPageFailed(plan.cursorName, page, tally.keysMigrated) return err } + if pass, prevClean, done := c.finishTTLInlineMigrationPage(plan.cursorName, page, tally); done { + c.reportTTLInlineMigrationPass(ctx, h, plan, pass, prevClean) + } + return nil +} + +// migrateTTLInlinePage rewrites the page's legacy anchors in batches of at most +// ttlInlineMigrationBatchKeyLimit keys and tallies what it did with each +// anchor. On a dispatch error the tally's keysMigrated still counts the keys +// of the batches dispatched before it. +func (c *DeltaCompactor) migrateTTLInlinePage( + ctx context.Context, + h ttlInlineMigrationHandler, + kvs []*store.KVPair, + readTS uint64, +) (ttlInlineMigrationTally, error) { var ( + tally ttlInlineMigrationTally elems []*kv.Elem[kv.OP] batchKeys int + rewrites int flushBatch = func() error { if err := c.dispatchCompaction(ctx, readTS, elems); err != nil { c.logger.WarnContext(ctx, "ttl inline migrator: batch dispatch failed", "type", h.typeName, "error", err) return err } + tally.keysMigrated += rewrites elems = nil batchKeys = 0 + rewrites = 0 return nil } ) for _, pair := range kvs { userKey := h.extractUserKey(pair.Key) - if userKey == nil || !c.coord.IsLeaderForKey(redisUserRouteKey(userKey)) { + if userKey == nil { + continue + } + tally.keysScanned++ + if !c.coord.IsLeaderForKey(redisUserRouteKey(userKey)) { + tally.keysSkipped++ continue } built, buildErr := h.buildElems(ctx, pair, readTS) if buildErr != nil { + tally.keysFailed++ c.logger.WarnContext(ctx, "ttl inline migrator: failed to build elems", "type", h.typeName, "key", string(userKey), "error", buildErr) continue } + if len(built) > 0 { + rewrites++ + } elems = append(elems, built...) batchKeys++ if batchKeys >= ttlInlineMigrationBatchKeyLimit { if err := flushBatch(); err != nil { - return err + return tally, err } } } - if err := flushBatch(); err != nil { - return err - } - c.advanceCursor(plan.cursorName, cursorKey, truncated) - return nil + // flushBatch adds the last batch's keys to tally, so call it before + // returning tally. + err := flushBatch() + return tally, err } // ttlInlineMigrationScanPlans scans only the groups this node leads when the @@ -294,34 +339,52 @@ func ttlInlineMigrationGroupCursorName(typeName string, groupID uint64) string { return typeName + ":group:" + strconv.FormatUint(groupID, 10) } -// scanTTLInlineMigrationRaw returns the scanned rows, the key the plan's cursor -// moves to, and whether the scan stopped before the end of the prefix. +// ttlInlineMigrationPage is one page a scan plan read in a tick. +type ttlInlineMigrationPage struct { + kvs []*store.KVPair + // cursorKey is the key the plan's cursor moves to. + cursorKey []byte + // truncated reports that the scan stopped before the end of the prefix. + truncated bool + // fromStart reports that the page starts at the beginning of the prefix, + // so it is the first page of a pass. + fromStart bool + // snapshotAt is no later than the acquisition of the read timestamp the + // page was scanned at; a pass opened by this page starts then. + snapshotAt time.Time +} + func (c *DeltaCompactor) scanTTLInlineMigrationRaw( ctx context.Context, h ttlInlineMigrationHandler, plan ttlInlineMigrationScanPlan, readTS uint64, -) ([]*store.KVPair, []byte, bool, error) { + snapshotAt time.Time, +) (ttlInlineMigrationPage, error) { c.cursorMu.Lock() start := h.prefix - if cur := c.cursors[plan.cursorName]; len(cur) > 0 { + cur := c.cursors[plan.cursorName] + if len(cur) > 0 { start = append(bytes.Clone(cur), cursorNextByte) } c.cursorMu.Unlock() + page := ttlInlineMigrationPage{fromStart: len(cur) == 0, snapshotAt: snapshotAt} end := store.PrefixScanEnd(h.prefix) if plan.scanner != nil { kvs, resume, err := plan.scanner.ScanGroupOwnedAt(ctx, plan.groupID, start, end, ttlInlineMigrationTickScanLimit, readTS) if err != nil { - return nil, nil, false, errors.WithStack(err) + return ttlInlineMigrationPage{}, errors.WithStack(err) } - return kvs, resume, resume != nil, nil + page.kvs, page.cursorKey, page.truncated = kvs, resume, resume != nil + return page, nil } kvs, err := c.st.ScanAt(ctx, start, end, ttlInlineMigrationTickScanLimit, readTS) if err != nil { - return nil, nil, false, errors.WithStack(err) + return ttlInlineMigrationPage{}, errors.WithStack(err) } - return kvs, lastKVKey(kvs), len(kvs) == ttlInlineMigrationTickScanLimit, nil + page.kvs, page.cursorKey, page.truncated = kvs, lastKVKey(kvs), len(kvs) == ttlInlineMigrationTickScanLimit + return page, nil } func (c *DeltaCompactor) migrateStringTTLInlineElems(ctx context.Context, pair *store.KVPair, readTS uint64) ([]*kv.Elem[kv.OP], error) { diff --git a/docs/design/2026_04_17_implemented_ttl_inline_value.md b/docs/design/2026_04_17_implemented_ttl_inline_value.md index b081b9052..5b36ceecd 100644 --- a/docs/design/2026_04_17_implemented_ttl_inline_value.md +++ b/docs/design/2026_04_17_implemented_ttl_inline_value.md @@ -315,6 +315,106 @@ from `ttlAt()` after the anchor-specific probes miss. The default remains fallback-enabled so rolling upgrades and not-yet-migrated stores preserve behavior. +"All nodes have migrated" is decided from the migrator's pass reports, not +from elapsed time. + +**What a pass is.** The migrator scans one plan per anchor type and per data +group the node leads. There are eight anchor types: `ttl-inline-string`, +`-hll`, `-indexed-collection`, `-list`, `-hash`, `-set`, `-zset` and +`-stream`. A plan reads at most 512 rows a tick, so a pass over a large prefix +spans several ticks. A pass runs from the first page of the prefix to the first +page that reaches its end. It closes on that page, and the next pass starts +again at the top with fresh counters. + +**What each pass reports.** The leader running the plan logs one line per +completed pass, `ttl inline migrator: pass complete`, with these keys: + +| Key | Meaning | +|---|---| +| `type` | Anchor type. | +| `group_id` | Data group. `0` is the whole-prefix fallback scan, used only when the coordinator does not report its local leader groups. | +| `full_pass` | The pass started at the top of the prefix, and every tick since planned its group on this node. A plan whose group is missing from a tick's plans drops its counters; if the group comes back, the resumed pass is not full. Leadership lost and regained between two ticks is not noticed (see "What a clean pass does not prove"). | +| `pages` | Pages the pass read. | +| `keys_scanned` | Anchors the pass read. | +| `keys_migrated` | Anchors rewritten by committed batches. A page retried after a failed batch still counts the keys its earlier batches rewrote. | +| `keys_skipped` | Anchors left alone because this node did not lead their user key's route. | +| `keys_failed` | Anchors whose rewrite could not be built. | +| `clean` | `full_pass` is true and the other three counts are all zero. | +| `duration` | How long the pass took. | + +A pass that is not clean, and the first clean pass after one that was not, are +logged at info. Repeated clean passes are logged at debug. + +Each pass also has a monitoring outcome: +- `clean`: as in the log. +- `partial`: rewrote, skipped and failed nothing, but was not full. +- `not_clean`: rewrote, skipped or failed at least one anchor. + +Two metrics per node carry it, labelled `type`, `group` and `outcome`: +- `elastickv_redis_ttl_inline_migration_passes_total`; +- `elastickv_redis_ttl_inline_migration_last_pass_start_timestamp_seconds`, + the time the latest pass this node completed with that outcome began. A pass + begins when the tick that reads its first page starts, before that tick + acquires the read timestamp the page is read at. The gauge records the start, + not the end, because a pass spanning several ticks read its first pages then. + +**Condition to check before disabling the fallback.** + +Let T0 be the latest of: +- the time the last node was upgraded to a build that reports passes (it also + migrates every locally led group, bootjp/elastickv#1278); +- the last route catalog change that moved a range between groups, since a + pass covers only the rows its group served at its read timestamps; +- the last restart of any node, since the metrics are process-local and a + restart drops that node's history. + +Then all of the following must hold: +1. No live build writes the legacy format. Every node runs a build that writes + inline TTLs (Phase 1 or later). +2. The Redis delta compactor (`ELASTICKV_REDIS_DELTA_COMPACTOR_ENABLED`) is + enabled on every node. A group whose leader runs no compactor gets no pass. +Pass starts come from each node's wall clock, and T0 from whichever clock +recorded the event. Let S be the largest wall-clock offset between any two +nodes (or between a node and the clock T0 was read from) that the deployment +guarantees, for example its NTP bound. Every comparison below keeps a margin +of S, so skew cannot order a pass on the wrong side of an event. + +3. For all eight types and every data group, the latest clean pass across all + nodes started more than S after T0: + `max by (type, group) (elastickv_redis_ttl_inline_migration_last_pass_start_timestamp_seconds{outcome="clean"})` + has a series for every type and group, each later than T0 + S. A pass that + only ended after T0 does not count: it read its first pages before T0, so a + route change at T0 could have put a legacy anchor behind its cursor. +4. For each type and group, that clean pass started more than S after every + `not_clean` pass on any node. Leadership moves, so a clean sample from a + former leader does not outrank a later `not_clean` pass on the current one. + With S in seconds, this query must return nothing: + + ``` + max by (type, group) (elastickv_redis_ttl_inline_migration_last_pass_start_timestamp_seconds{outcome="not_clean"}) + S + >= on (type, group) + max by (type, group) (elastickv_redis_ttl_inline_migration_last_pass_start_timestamp_seconds{outcome="clean"}) + ``` + +`partial` passes do not affect the gate. + +**What a clean pass does not prove.** +- Live writers only produce the inline format (condition 1). A clean pass + therefore shows that the anchors its group's scan serves were all inline + when read. +- Leadership lost and regained between two ticks leaves the pass open and + full. The anchors behind the cursor were read inline earlier in the pass, and + without a legacy writer (condition 1) the interim leader could not have made + them legacy again. +- The scan skips some legacy-route hash, set, zset and stream rows without + counting them: those whose user key belongs to a group another node leads. + They are not migrated, and they do not keep a pass from being clean. The gate + for them is milestone M4 of the cross-leader design (bootjp/elastickv#1280), + which Phase 3 must also pass. +- The scan also defers, without counting, rows whose route is under staged + visibility. T0 excludes them: staging only happens while a route moves. +- Legacy bare string keys are outside the anchor prefixes, as noted above. + ### Phase 4 — Remove TTL buffer The `TTLBuffer`, `runTTLFlusher`, and `flushTTLBuffer` have been removed. TTL @@ -394,5 +494,5 @@ adapter/ | 0 | Add dual-read fallback; new writes use new encoding | Low — reads always fall back | | 1 | Add inline TTL to list/hash/set/zset/stream metadata and HLL payloads | Medium — delta compaction must preserve TTL | | 2 | Background migration of existing keys | Medium — shipped as bounded compactor pass | -| 3 | Remove legacy `!redis\|ttl\|` read path | Medium — safe only after prefix-scanned anchors and legacy bare strings are cleared | +| 3 | Remove legacy `!redis\|ttl\|` read path | Medium — safe only after a clean migration pass for every anchor type and group (§5, Phase 3), the cross-leader legacy-route gate, and legacy bare strings are cleared | | 4 | Remove TTL buffer | Low — shipped | diff --git a/main.go b/main.go index 6efe3f584..e1e23364b 100644 --- a/main.go +++ b/main.go @@ -408,9 +408,13 @@ func startFSMCompactorIfEnabled(ctx context.Context, eg *errgroup.Group, runtime slog.Info("fsm compactor disabled", "env", fsmCompactorEnabledEnvVar) } -func newRedisDeltaCompactorIfEnabled(shardStore *kv.ShardStore, coordinate kv.Coordinator) *adapter.DeltaCompactor { +func newRedisDeltaCompactorIfEnabled(shardStore *kv.ShardStore, coordinate kv.Coordinator, metricsRegistry *monitoring.Registry) *adapter.DeltaCompactor { if enabledEnv(redisDeltaCompactorEnabledEnvVar) { - return adapter.NewDeltaCompactor(shardStore, coordinate) + var opts []adapter.DeltaCompactorOption + if obs := metricsRegistry.RedisTTLInlineMigrationObserver(); obs != nil { + opts = append(opts, adapter.WithDeltaCompactorTTLInlineMigrationObserver(obs)) + } + return adapter.NewDeltaCompactor(shardStore, coordinate, opts...) } slog.Info("redis delta compactor disabled", "env", redisDeltaCompactorEnabledEnvVar) return nil @@ -3482,7 +3486,7 @@ func prepareRedisServer(ctx context.Context, lc *net.ListenConfig, redisAddr str if err != nil { return nil, nil, nil, errors.Wrapf(err, "failed to listen on %s", redisAddr) } - deltaCompactor := newRedisDeltaCompactorIfEnabled(shardStore, coordinate) + deltaCompactor := newRedisDeltaCompactorIfEnabled(shardStore, coordinate, metricsRegistry) redisOpts := []adapter.RedisServerOption{ adapter.WithRedisActiveTimestampTracker(readTracker), adapter.WithRedisRequestObserver(metricsRegistry.RedisObserver()), diff --git a/main_background_env_test.go b/main_background_env_test.go index 17792ca0d..4f6461017 100644 --- a/main_background_env_test.go +++ b/main_background_env_test.go @@ -28,14 +28,14 @@ func TestEnabledEnv(t *testing.T) { func TestRedisDeltaCompactorEnv(t *testing.T) { t.Run("default enabled", func(t *testing.T) { t.Setenv(redisDeltaCompactorEnabledEnvVar, "") - if newRedisDeltaCompactorIfEnabled(nil, nil) == nil { + if newRedisDeltaCompactorIfEnabled(nil, nil, nil) == nil { t.Fatal("newRedisDeltaCompactorIfEnabled() = nil, want compactor") } }) t.Run("false disables", func(t *testing.T) { t.Setenv(redisDeltaCompactorEnabledEnvVar, "false") - if newRedisDeltaCompactorIfEnabled(nil, nil) != nil { + if newRedisDeltaCompactorIfEnabled(nil, nil, nil) != nil { t.Fatal("newRedisDeltaCompactorIfEnabled() returned compactor, want nil") } }) diff --git a/monitoring/redis_ttl_inline.go b/monitoring/redis_ttl_inline.go new file mode 100644 index 000000000..9162ef812 --- /dev/null +++ b/monitoring/redis_ttl_inline.go @@ -0,0 +1,77 @@ +package monitoring + +import ( + "strconv" + "time" + + "github.com/prometheus/client_golang/prometheus" +) + +// Outcomes of a Redis TTL-inline migration pass. +const ( + // RedisTTLInlinePassClean is a pass that read its whole prefix while its + // group stayed in this node's plans and found every anchor already in the + // inline format. + RedisTTLInlinePassClean = "clean" + // RedisTTLInlinePassPartial is a pass that rewrote, skipped and failed + // nothing but did not read its whole prefix: it resumed a cursor left + // before its group dropped out of this node's plans. + RedisTTLInlinePassPartial = "partial" + // RedisTTLInlinePassNotClean is a pass that rewrote, skipped or failed at + // least one anchor. + RedisTTLInlinePassNotClean = "not_clean" +) + +// RedisTTLInlineMigrationObserver records the passes the Redis TTL-inline +// migrator completes. A pass covers one anchor type (typeName, such as +// "ttl-inline-string") in one data group; group 0 is the whole-prefix scan the +// migrator falls back to when the coordinator does not report its groups. +// outcome is one of the RedisTTLInlinePass constants; startedAt is no later +// than the read timestamp of the pass's first page was acquired. +type RedisTTLInlineMigrationObserver interface { + ObserveRedisTTLInlineMigrationPass(typeName string, groupID uint64, outcome string, startedAt time.Time) +} + +// RedisTTLInlineMigrationMetrics exports the TTL-inline migrator's passes. +// Labels are the anchor type, data group and outcome only: eight types times +// the data groups a node leads times three outcomes. +type RedisTTLInlineMigrationMetrics struct { + passes *prometheus.CounterVec + lastPassStart *prometheus.GaugeVec +} + +func newRedisTTLInlineMigrationMetrics(registerer prometheus.Registerer) *RedisTTLInlineMigrationMetrics { + m := &RedisTTLInlineMigrationMetrics{ + passes: prometheus.NewCounterVec( + prometheus.CounterOpts{ + Name: "elastickv_redis_ttl_inline_migration_passes_total", + Help: "Completed Redis TTL-inline migration passes by anchor type, data group, and outcome (clean, partial, not_clean).", + }, + []string{"type", "group", "outcome"}, + ), + lastPassStart: prometheus.NewGaugeVec( + prometheus.GaugeOpts{ + Name: "elastickv_redis_ttl_inline_migration_last_pass_start_timestamp_seconds", + Help: "Unix time at which the latest Redis TTL-inline migration pass this node completed began, by anchor type, data group, and outcome (clean, partial, not_clean).", + }, + []string{"type", "group", "outcome"}, + ), + } + if registerer != nil { + registerer.MustRegister(m.passes, m.lastPassStart) + } + return m +} + +// ObserveRedisTTLInlineMigrationPass counts a completed pass and records when +// it began. The start, not the end, is what a gate compares against: a pass +// that ended after an event but began before it read part of its prefix +// before the event. +func (m *RedisTTLInlineMigrationMetrics) ObserveRedisTTLInlineMigrationPass(typeName string, groupID uint64, outcome string, startedAt time.Time) { + if m == nil { + return + } + group := strconv.FormatUint(groupID, 10) + m.passes.WithLabelValues(typeName, group, outcome).Inc() + m.lastPassStart.WithLabelValues(typeName, group, outcome).Set(float64(startedAt.UnixNano()) / float64(time.Second)) +} diff --git a/monitoring/redis_ttl_inline_test.go b/monitoring/redis_ttl_inline_test.go new file mode 100644 index 000000000..ed1ba45d4 --- /dev/null +++ b/monitoring/redis_ttl_inline_test.go @@ -0,0 +1,54 @@ +package monitoring + +import ( + "testing" + "time" + + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/testutil" + "github.com/stretchr/testify/require" +) + +// Each outcome keeps the start time of its latest pass, so a not_clean pass +// after a clean one stays visible next to it, and a gate can require a clean +// pass that started, not just ended, after an event. +func TestRedisTTLInlineMigrationMetricsExportPasses(t *testing.T) { + t.Parallel() + reg := prometheus.NewRegistry() + metrics := newRedisTTLInlineMigrationMetrics(reg) + + start := time.Unix(1_700_000_000, 0) + metrics.ObserveRedisTTLInlineMigrationPass("ttl-inline-string", 2, RedisTTLInlinePassNotClean, start) + metrics.ObserveRedisTTLInlineMigrationPass("ttl-inline-string", 2, RedisTTLInlinePassClean, start.Add(time.Minute)) + metrics.ObserveRedisTTLInlineMigrationPass("ttl-inline-string", 2, RedisTTLInlinePassClean, start.Add(2*time.Minute)) + metrics.ObserveRedisTTLInlineMigrationPass("ttl-inline-hash", 3, RedisTTLInlinePassPartial, start) + + require.Equal(t, 1.0, testutil.ToFloat64(metrics.passes.WithLabelValues("ttl-inline-string", "2", "not_clean"))) + require.Equal(t, 2.0, testutil.ToFloat64(metrics.passes.WithLabelValues("ttl-inline-string", "2", "clean"))) + require.Equal(t, 1.0, testutil.ToFloat64(metrics.passes.WithLabelValues("ttl-inline-hash", "3", "partial"))) + for labels, want := range map[[3]string]time.Time{ + {"ttl-inline-string", "2", "not_clean"}: start, + {"ttl-inline-string", "2", "clean"}: start.Add(2 * time.Minute), + {"ttl-inline-hash", "3", "partial"}: start, + } { + got := testutil.ToFloat64(metrics.lastPassStart.WithLabelValues(labels[0], labels[1], labels[2])) + require.InDelta(t, float64(want.Unix()), got, 0, labels) + } + require.Equal(t, 3, testutil.CollectAndCount(metrics.lastPassStart)) + + families, err := reg.Gather() + require.NoError(t, err) + require.True(t, hasMetricFamily(families, "elastickv_redis_ttl_inline_migration_passes_total")) + require.True(t, hasMetricFamily(families, "elastickv_redis_ttl_inline_migration_last_pass_start_timestamp_seconds")) + require.False(t, hasMetricFamily(families, "elastickv_redis_ttl_inline_migration_last_pass_timestamp_seconds")) +} + +func TestRedisTTLInlineMigrationObserverIsNilSafe(t *testing.T) { + t.Parallel() + var metrics *RedisTTLInlineMigrationMetrics + require.NotPanics(t, func() { + metrics.ObserveRedisTTLInlineMigrationPass("ttl-inline-string", 1, RedisTTLInlinePassClean, time.Now()) + }) + var r *Registry + require.Nil(t, r.RedisTTLInlineMigrationObserver()) +} diff --git a/monitoring/registry.go b/monitoring/registry.go index 159bbe822..2ae7698dd 100644 --- a/monitoring/registry.go +++ b/monitoring/registry.go @@ -14,25 +14,26 @@ type Registry struct { registerer prometheus.Registerer gatherer prometheus.Gatherer - dynamo *DynamoDBMetrics - redis *RedisMetrics - raft *RaftMetrics - lua *LuaMetrics - hotPath *HotPathMetrics - pebble *PebbleMetrics - writeConflict *WriteConflictMetrics - sqs *SQSMetrics - sqsObserver *SQSObserver - fs *FileSystemMetrics - s3 *S3Metrics - hlc *HLCMetrics - hlcObserver *HLCObserver - coldStart *ColdStartMetrics - coldStartObs *ColdStartObserver - tso *TSOMetrics - tsoObserver *TSOObserver - encryption *EncryptionMetrics - encryptionObs *EncryptionStateObserver + dynamo *DynamoDBMetrics + redis *RedisMetrics + raft *RaftMetrics + lua *LuaMetrics + hotPath *HotPathMetrics + pebble *PebbleMetrics + writeConflict *WriteConflictMetrics + sqs *SQSMetrics + sqsObserver *SQSObserver + fs *FileSystemMetrics + s3 *S3Metrics + hlc *HLCMetrics + hlcObserver *HLCObserver + coldStart *ColdStartMetrics + coldStartObs *ColdStartObserver + tso *TSOMetrics + tsoObserver *TSOObserver + encryption *EncryptionMetrics + encryptionObs *EncryptionStateObserver + redisTTLInline *RedisTTLInlineMigrationMetrics } // NewRegistry builds a registry with constant labels that identify the local node. @@ -67,6 +68,7 @@ func NewRegistry(nodeID string, nodeAddress string) *Registry { r.tsoObserver = newTSOObserver(r.tso) r.encryption = newEncryptionMetrics(registerer) r.encryptionObs = newEncryptionStateObserver(r.encryption) + r.redisTTLInline = newRedisTTLInlineMigrationMetrics(registerer) return r } @@ -288,6 +290,15 @@ func (r *Registry) HLCObserver() *HLCObserver { return r.hlcObserver } +// RedisTTLInlineMigrationObserver returns the observer the Redis delta +// compactor reports its completed TTL-inline migration passes to. +func (r *Registry) RedisTTLInlineMigrationObserver() RedisTTLInlineMigrationObserver { + if r == nil || r.redisTTLInline == nil { + return nil + } + return r.redisTTLInline +} + // TSOObserver returns the shared runtime TSO observer. The kv package consumes // it through its narrow observer interface, keeping Prometheus ownership here. func (r *Registry) TSOObserver() *TSOObserver {