[Feature]: expose change stream consumer lag; a consumer 11 days behind presents as completely healthy

[lfriedman-netllama]

Prerequisites

• I searched existing issues

Code of Conduct

• I agree to follow NVSentinel's Code of Conduct

Feature Summary

Expose change stream consumer lag as a metric in the shared store-client, so every
change stream consumer gets it rather than each module reinventing it. Split out of #1704 at
@lalitadithya's suggestion, since it applies to all consumers and not just
health-events-analyzer.

Problem/Use Case

On a 288 node GB200 fleet, health-events-analyzer fell 11 days and 2.4 million events
behind its change stream. Nothing surfaced it. The pod presented as completely healthy:

• Running, zero restarts, uptime 6 days
• 2m CPU against a 1000m limit, 67Mi against 1Gi, so no throttling or OOM signal
• no errors in logs, and its own success counters climbing normally
• rule_matched_total incrementing, so it looked like it was working

It was working. Just on events from 11 days earlier. We only found it because the stale
pattern events it emitted looked like a current hardware fault, and chasing that down led to
decoding the resume token by hand.

The existing metric does not catch this. event-exporter has
ResumeTokenUpdateTimestamp (event-exporter/pkg/metrics/metrics.go:80, set at
worker_pool.go:171), but it records when the token was last written, not where the
stream position is. Our analyzer was persisting tokens continuously and successfully the
entire time it was 11 days behind, so a freshness gauge on token writes would have read as
perfectly healthy throughout. The two are independent: a consumer can commit progress every
few seconds while falling further behind every minute.

Proposed Solution

Emit the stream position as a gauge from store-client, where the token is already
loaded and stored, so it lands once and covers every consumer:

changestream_position_timestamp_seconds{client="health-events-analyzer"}

Lag is then time() - changestream_position_timestamp_seconds, which alerts cleanly:

time() - changestream_position_timestamp_seconds > 900

For MongoDB this needs no extra query. The resume token's _data encodes the BSON timestamp
in its leading bytes, so the position is derivable from the token the consumer already holds:
the seconds are parseInt(_data[2:10], 16). We decoded ours exactly that way to confirm the
2026-08-21 position. The PostgreSQL provider tracks a monotonic event ID rather than a
timestamp, so it may need its own derivation, or a changestream_events_behind{client} gauge
instead.

Natural home is alongside the existing token load and store in
store-client/pkg/datastore/providers/mongodb/watcher/watch_store.go, so consumers get it
without code changes.

Why this is worth doing generically

Every consumer of this change stream has the same failure mode, and it is silent in all of
them. On our fleet the consumers are health-events-analyzer, fault-quarantine, node-drainer
and fault-remediation, plus event-exporter when enabled. A consumer that silently stops
keeping up does not fail loudly; it just makes decisions on stale data, which for the action
modules would mean quarantining or rebooting nodes based on faults that are days old.

Related: #1594 raises the oplog window for change stream consumers, which is the other half of
this. If a consumer falls behind the oplog window it cannot resume at all, and today there is
no metric that would warn you as it approached that edge.

Acceptance criteria

• A gauge exposes the change stream position per consumer, labeled by client name.
• It is emitted from store-client so every consumer gets it without per-module code.
• The value updates as the stream advances, and does not merely reflect token write time.
• MongoDB derives the position from the existing resume token, with no extra query per event.
• The PostgreSQL provider either derives an equivalent or documents why it differs.
• Documentation includes an example lag alert.
• Unit tests cover position extraction, including a token from a known timestamp.

Component

Other

[lfriedman-netllama]

Additional findings from re-reading the code today. Two of them change the shape of this request, and one is a bug that exists right now.

The mechanism already exists, and exactly one consumer uses it. I had framed this as a missing capability. It is not:

• ChangeStreamMetrics.GetUnprocessedEventCount is defined at store-client/pkg/client/interfaces.go:80-84
• implemented for MongoDB at store-client/pkg/datastore/providers/mongodb/watcher/watch_store.go:561
• implemented for PostgreSQL at store-client/pkg/datastore/providers/postgresql/changestream.go:1354
• forwarded through the resume-control wrapper at store-client/pkg/client/resume_token.go:75

So this is "adopt what is already there, consistently", which is cheaper than building it.

Current state per consumer:

Consumer	Stream backlog metric	Notes
fault-quarantine	yes, works	Ticker at pkg/eventwatcher/event_watcher.go:383-395 sets fault_quarantine_event_backlog_count. Sets -1 when the watcher lacks the interface, which is a nice touch
event-exporter	declared but never set	see below
health-events-analyzer	none	no reference to ChangeStreamMetrics, GetUnprocessedEventCount or any backlog metric anywhere in the module. This is the one that went 2.4M events behind
fault-remediation	none
janitor	none
lifecycle-manager	none
labeler	none

node_drainer_queue_depth exists but measures node-drainer's own in-process queue, not stream position, so it does not cover this.

A live bug worth fixing independently of this feature: event-exporter creates health_events_exporter_event_backlog_size at pkg/metrics/metrics.go:73 and never assigns it anywhere in the module. A registered gauge that is never written reports 0, so event-exporter currently advertises "no backlog" permanently. That is worse than having no metric at all, and it is a small fix. Happy to send a PR for just that if useful.

On resume token age specifically: nothing in the repo exports it (grep for token_age / TokenAge returns nothing). It is the cheaper and more robust of the two signals: no countDocuments per poll, and it grows on its own even when a consumer is completely stalled, whereas a backlog count still needs the query to notice.

Suggested shape, emitted from store-client and labelled by the existing client name so every consumer gets it without per-module wiring:

changestream_events_behind{client}                            gauge
changestream_resume_token_age_seconds{client}                 gauge
changestream_last_event_processed_timestamp_seconds{client}   gauge

I would keep fault_quarantine_event_backlog_count as-is for dashboard compatibility rather than breaking it.

[Nikhil-cp1905]

May i take up this issue?

[lalitadithya]

Sure!

Can you please share a design on how you'd like to implement this first since this seems like it is going to be rather large feature?

[lfriedman-netllama]

@lalitadithya design is up as #1738 (ADR-052), no code in it so the shape can be agreed first.

Three things the write-up changed about the ask, worth flagging here rather than making you read the whole ADR:

It is smaller than I implied. ChangeStreamMetrics.GetUnprocessedEventCount already exists and is implemented for both providers; exactly one consumer (fault-quarantine) wired it up. So most of this is "adopt what is there, from one place" rather than build it.

The obvious signal is wrong, so the ADR does not use it. "Resume token age" means now - position_time, which reports lag on a caught-up but idle consumer: no events for an hour looks like an hour of lag. The ADR proposes head_time - position_time instead, so an idle stream reports zero. There is a specific regression test for that case, because it is the one that would pass under the naive approach.

Decoding the Mongo token is not shippable. The token's _data hex encodes the cluster time and that is genuinely how I measured the 11 days by hand, but it is an undocumented driver detail. The ADR records the position timestamp at save time in MarkProcessed instead, which already holds the event's clusterTime. PostgreSQL needs nothing new, datastore_changelog.changed_at already exists.

Four open questions are listed at the end of the PR: metric naming, which signal is default-on, whether the small positionTime field on the Mongo token document is acceptable, and whether you want event-exporter's dead gauge fixed here or as its own small PR first. I am happy to send that gauge fix separately today, since it is a one-line bug independent of this design.

[lfriedman-netllama]

The dead-gauge fix is up separately as #1739, as offered: health_events_exporter_event_backlog_size is registered and never set, so event-exporter has been reporting "no backlog" unconditionally.

Worth noting why it was never wired, since it also validates the ADR's approach: GetUnprocessedEventCount takes a lastProcessedID, and event-exporter tracks only resume tokens with no per-document id to pass. fault-quarantine can call it because it keeps a LastProcessedObjectIDStore.

So wiring it in place would mean threading a document id through the concurrent worker pool so "last processed" respects the contiguous-completion ordering sequenceTracker already owns. That is exactly the duplication ADR-052 avoids by computing position in the shared layer, which already has it. #1739 therefore just deletes the misleading metric, and real visibility arrives with this issue.