Add pipeline phase visibility via /status and OTel gauge#1000
Conversation
Operators can query whether a running process is in snapshot or replication without scraping logs. The phase flips when logical replication starts, covering all source modes. Signed-off-by: shaurya2k06 <shaurya2k06@gmail.com>
Cover registerPhaseMetric with an OTel ManualReader and add an integration test that polls GET /status through snapshot and replication during a real pg2pg snapshot_and_replication run. Signed-off-by: shaurya2k06 <shaurya2k06@gmail.com>
|
You can ignore the coverage failure, but please fix the issues reported by the linter. |
… Lint fixes Signed-off-by: shaurya2k06 <shaurya2k06@gmail.com>
| } | ||
| // snapshotAndListen already entered the listen loop; only reached on | ||
| // clean return (context cancel) or if snapshot path returned early. | ||
| return nil |
There was a problem hiding this comment.
Could you please explain the reasoning behind this change?
There was a problem hiding this comment.
The old fallthrough was unreachable in the current code path, because snapshotAndListen() always ends by calling listen() , and listen() only returns a non-nil error or ctx.Err(), not nil. So the explicit return nil isn't really a behavior fix, it's mainly about safety, it makes it obvious that once the snapshot path is used, the function should stop there and not continue into the normal replication startup path.It also prevents a future refactor from accidentally making the old fallthrough reachable and then calling StartReplication() twice on the same listener flow.
How do you think this should be dealt with? Again, It's guard-railing/documentation, not a bug fix.
kvch
left a comment
kvch
left a comment
There was a problem hiding this comment.
Thank your for getting back to the PR. I've posted a few notes.
Signed-off-by: shaurya2k06 <shaurya2k06@gmail.com>
2 participants
Description
Adds queryable pipeline phase visibility so operators can tell whether a running pgstream process is in the initial snapshot or replication (streaming) phase without scraping logs.
In
snapshot_and_replicationmode, the only signal today is log order:initial snapshot enabled→snapshot generation completed→logical replication started.pgstream statusis a static config/init check against Postgres and has no phase field.This PR exposes phase via:
GET /statuson the existing health server (alongside/healthand/ready)pgstream.pipeline.phaseThe phase flips to
replicationwhen logical replication starts (after successfulStartReplication/StartReplicationFromLSN), matching the existing log signal.Related Issue(s)
snapshot_and_replicationmode #984Type of Change
Changes Made
internal/phase.Tracker— thread-safe atomic tracker withsnapshotandreplicationphasesGET /statusandWithPhaseProvider; response includesphasefieldrun/snapshotcommands into health server and stream pipelinesnapshot_and_replication:snapshotat start ofsnapshotAndListen,replicationafterStartReplicationFromLSNsucceedsreplicationonly:replicationafterStartReplicationsucceedspgstream snapshot:snapshotwhen snapshotListenstartsreplicationwhen kafkaListenstartsInt64ObservableGaugepgstream.pipeline.phase(value1withphaseattribute)/statusindocs/configuration.mdandconfig_template.yamlpgstream.pipeline.phaseindocs/Observability.mdOut of scope (per issue discussion): enriching CLI
pgstream status— it runs as a separate process and cannot observe in-memory pipeline state; slot/LSN heuristics would be unreliable.Before / After
GET /status{"status":"ok","version":"...","phase":"snapshot"|"replication"|""}pgstream statuspgstream.pipeline.phase{phase="snapshot"|"replication"} = 1Testing
internal/phase/phase_test.go— tracker get/set, nil-safeinternal/health/health_test.go—/statushandler with/without provider, phase transitionspkg/wal/listener/postgres/wal_pg_listener_test.go— phase flips atStartReplication/StartReplicationFromLSNpkg/stream/phase_metrics_test.go—registerPhaseMetricvia OTelManualReaderpkg/stream/integration/phase_status_integration_test.go— pollsGET /statusthrough snapshot → replication on a real pg2pgsnapshot_and_replicationrun; verifies post-snapshot insert replicates to target/statusduring snapshot and replication (automated equivalent of manualcurl)make testpassesChecklist
Additional Notes
/statusis opt-in via existinginstrumentation.health.enabled(PGSTREAM_HEALTH_CHECK_ENABLED); no new config knob added"") until the first lifecycle transition/healthand/readybehavior is unchanged/healthand/ready#882) and observable gauges (pgstream.replication.lag, dropped-queries counter Add strict mode toWALWriterand counter for dropped messages #959)