Add pipeline phase visibility via /status and OTel gauge

cmd/health_cmd.go

@@ -9,6 +9,7 @@ import (
 
 	"github.com/xataio/pgstream/cmd/config"
 	"github.com/xataio/pgstream/internal/health"
+	"github.com/xataio/pgstream/internal/phase"
 	pglib "github.com/xataio/pgstream/internal/postgres"
 	loglib "github.com/xataio/pgstream/pkg/log"
 )
@@ -28,8 +29,9 @@ const (
 // fails the command immediately rather than silently degrading.
 //
 // sourcePostgresURL, when non-empty, wires the /ready endpoint to ping the
-// source database.
-func startHealthServer(ctx context.Context, logger loglib.Logger, sourcePostgresURL string) (func(), error) {
+// source database. phaseTracker, when non-nil, wires the /status endpoint to
+// report the current pipeline phase.
+func startHealthServer(ctx context.Context, logger loglib.Logger, sourcePostgresURL string, phaseTracker *phase.Tracker) (func(), error) {
 	cfg, err := config.ParseHealthConfig()
 	if err != nil {
 		return nil, fmt.Errorf("parsing health config: %w", err)
@@ -43,6 +45,12 @@ func startHealthServer(ctx context.Context, logger loglib.Logger, sourcePostgres
 		health.WithVersion(version()),
 	}
 
+	if phaseTracker != nil {
+		opts = append(opts, health.WithPhaseProvider(func() string {
+			return string(phaseTracker.Get())
+		}))
+	}
+
 	var pool *pglib.Pool
 	if sourcePostgresURL != "" {
 		pool, err = pglib.NewConnPool(ctx, sourcePostgresURL, pglib.WithMaxConnections(healthReadinessPoolMaxConns))

cmd/run_cmd.go

@@ -11,6 +11,7 @@ import (
 	"github.com/spf13/viper"
 	"github.com/xataio/pgstream/cmd/config"
 	"github.com/xataio/pgstream/internal/log/zerolog"
+	"github.com/xataio/pgstream/internal/phase"
 	"github.com/xataio/pgstream/pkg/stream"
 )
 
@@ -71,7 +72,9 @@ func run(ctx context.Context) error {
 	}
 
 	stdLogger := zerolog.NewStdLogger(logger)
-	stopHealth, err := startHealthServer(ctx, stdLogger, streamConfig.SourcePostgresURL())
+	phaseTracker := phase.NewTracker()
+	opts = append(opts, stream.WithPhaseTracker(phaseTracker))
+	stopHealth, err := startHealthServer(ctx, stdLogger, streamConfig.SourcePostgresURL(), phaseTracker)
 	if err != nil {
 		return err
 	}

cmd/snapshot_cmd.go

@@ -10,6 +10,7 @@ import (
 	"github.com/spf13/viper"
 	"github.com/xataio/pgstream/cmd/config"
 	"github.com/xataio/pgstream/internal/log/zerolog"
+	"github.com/xataio/pgstream/internal/phase"
 	"github.com/xataio/pgstream/pkg/stream"
 )
 
@@ -44,13 +45,14 @@ func snapshot(ctx context.Context) error {
 	defer provider.Close()
 
 	stdLogger := zerolog.NewStdLogger(logger)
-	stopHealth, err := startHealthServer(ctx, stdLogger, streamConfig.SourcePostgresURL())
+	phaseTracker := phase.NewTracker()
+	stopHealth, err := startHealthServer(ctx, stdLogger, streamConfig.SourcePostgresURL(), phaseTracker)
 	if err != nil {
 		return err
 	}
 	defer stopHealth()
 
-	return stream.Snapshot(ctx, stdLogger, streamConfig, provider.NewInstrumentation("snapshot"))
+	return stream.Snapshot(ctx, stdLogger, streamConfig, provider.NewInstrumentation("snapshot"), stream.WithPhaseTracker(phaseTracker))
 }
 
 func snapshotFlagBinding(cmd *cobra.Command, args []string) error {

config_template.yaml

@@ -12,7 +12,7 @@ instrumentation:
     endpoint: "0.0.0.0:4317"
     sample_ratio: 0.5 # ratio of traces that will be sampled. Must be between 0.0-1.0, where 0 is no traces sampled, and 1 is all traces sampled.
   health:
-    enabled: false # whether to expose the /health (liveness) and /ready (readiness) HTTP endpoints. Only honoured by the run and snapshot commands. Defaults to false.
+    enabled: false # whether to expose the /health (liveness), /ready (readiness), and /status (pipeline phase) HTTP endpoints. Only honoured by the run and snapshot commands. Defaults to false.
     address: "localhost:9910" # address the health server binds to. Defaults to localhost:9910.
 
 source:

docs/Observability.md

@@ -64,6 +64,18 @@ All metrics follow the `pgstream.*` naming convention and include relevant attri
 
 **⚠️ Important:** This metric only tracks pgstream's consumer lag. It's strongly recommended to also monitor your source PostgreSQL metrics, particularly the built-in replication lag metrics (`pg_stat_replication.flush_lag`, `pg_stat_replication.replay_lag`) to get a complete picture of replication health.
 
+### Pipeline Phase
+
+| Metric                     | Type            | Unit | Description                                                                 |
+| -------------------------- | --------------- | ---- | --------------------------------------------------------------------------- |
+| `pgstream.pipeline.phase`  | ObservableGauge | 1    | Reports `1` for the active pipeline phase and `0` for the others (`phase=snapshot\|replication`) |
+
+**Attributes:**
+
+- `phase`: Phase this data point refers to (`snapshot` or `replication`)
+
+**Usage:** Distinguish whether a running process is performing the initial snapshot or streaming logical replication, without scraping logs. Both `phase="snapshot"` and `phase="replication"` series are reported on every collection, with `1` marking the active phase and `0` the inactive one, so queries like `sum by (phase)` stay unambiguous across transitions. Nothing is reported until the first phase transition. The gauge flips to `replication` when logical replication starts. The same value is also exposed via the health server `GET /status` endpoint when health checks are enabled.
+
 ### WAL Event Processing
 
 | Metric                               | Type      | Unit | Description                                    |

docs/configuration.md

@@ -440,7 +440,7 @@ One of exponential/constant/disable retries retry policies can be provided for t
 <details>
   <summary>Health endpoint</summary>
 
-Exposes `/health` (liveness, always 200) and `/ready` (readiness, pings the source postgres database when configured). Only the `run` and `snapshot` commands start the server. Responses are JSON.
+Exposes `/health` (liveness, always 200), `/ready` (readiness, pings the source postgres database when configured), and `/status` (current pipeline phase: `snapshot` or `replication`). Only the `run` and `snapshot` commands start the server. Responses are JSON.
 
 | Environment Variable           | Default          | Required | Description                                                                                                  |
 | ------------------------------ | ---------------- | -------- | ------------------------------------------------------------------------------------------------------------ |

internal/health/health.go

@@ -1,8 +1,8 @@
 // SPDX-License-Identifier: Apache-2.0
 
-// Package health exposes a small HTTP server with /health (liveness) and
-// /ready (readiness) endpoints so that operators and orchestrators can
-// monitor a running pgstream process.
+// Package health exposes a small HTTP server with /health (liveness),
+// /ready (readiness), and /status (pipeline phase) endpoints so that
+// operators and orchestrators can monitor a running pgstream process.
 package health
 
 import (
@@ -29,12 +29,13 @@ type Config struct {
 	Address string
 }
 
-// Server serves /health and /ready over HTTP.
+// Server serves /health, /ready, and /status over HTTP.
 type Server struct {
 	logger         loglib.Logger
 	address        string
 	version        string
 	readinessCheck func(ctx context.Context) error
+	phaseProvider  func() string
 	listener       net.Listener
 	httpServer     *http.Server
 }
@@ -84,6 +85,16 @@ func WithReadinessCheck(fn func(ctx context.Context) error) Option {
 	}
 }
 
+// WithPhaseProvider registers a function the /status handler will invoke to
+// report the current pipeline phase (e.g. "snapshot" or "replication").
+// If no provider is registered, /status still returns status/version with an
+// empty phase.
+func WithPhaseProvider(fn func() string) Option {
+	return func(s *Server) {
+		s.phaseProvider = fn
+	}
+}
+
 // Listen binds the configured address synchronously and prepares the HTTP
 // server. After it returns successfully, Serve must be called (typically
 // from a goroutine) to start accepting requests. Splitting bind from serve
@@ -98,6 +109,7 @@ func (s *Server) Listen() error {
 	mux := http.NewServeMux()
 	mux.HandleFunc("/health", s.handleHealth)
 	mux.HandleFunc("/ready", s.handleReady)
+	mux.HandleFunc("/status", s.handleStatus)
 
 	s.listener = ln
 	s.httpServer = &http.Server{
@@ -133,6 +145,10 @@ func (s *Server) handleHealth(w http.ResponseWriter, _ *http.Request) {
 	s.writeJSON(w, http.StatusOK, s.okPayload())
 }
 
+func (s *Server) handleStatus(w http.ResponseWriter, _ *http.Request) {
+	s.writeJSON(w, http.StatusOK, s.statusPayload())
+}
+
 func (s *Server) handleReady(w http.ResponseWriter, r *http.Request) {
 	if s.readinessCheck != nil {
 		ctx, cancel := context.WithTimeout(r.Context(), readinessCheckTimeout)
@@ -159,6 +175,16 @@ func (s *Server) okPayload() map[string]string {
 	return payload
 }
 
+func (s *Server) statusPayload() map[string]string {
+	payload := s.okPayload()
+	phase := ""
+	if s.phaseProvider != nil {
+		phase = s.phaseProvider()
+	}
+	payload["phase"] = phase
+	return payload
+}
+
 func (s *Server) writeJSON(w http.ResponseWriter, code int, body any) {
 	w.Header().Set("Content-Type", "application/json")
 	w.WriteHeader(code)

internal/health/health_test.go

@@ -101,6 +101,49 @@ func TestHandleReady_CheckFails(t *testing.T) {
 	require.Equal(t, "source unreachable", failure["error"])
 }
 
+func TestHandleStatus_NoProvider(t *testing.T) {
+	t.Parallel()
+	s := NewServer(Config{}, WithVersion("v1.2.3"))
+
+	req := httptest.NewRequest(http.MethodGet, "/status", nil)
+	rec := httptest.NewRecorder()
+	s.handleStatus(rec, req)
+
+	require.Equal(t, http.StatusOK, rec.Code)
+	require.Equal(t, "application/json", rec.Header().Get("Content-Type"))
+	body, _ := io.ReadAll(rec.Body)
+	var got map[string]string
+	require.NoError(t, json.Unmarshal(body, &got))
+	require.Equal(t, "ok", got["status"])
+	require.Equal(t, "v1.2.3", got["version"])
+	require.Equal(t, "", got["phase"])
+}
+
+func TestHandleStatus_WithProvider(t *testing.T) {
+	t.Parallel()
+	phase := "snapshot"
+	s := NewServer(Config{}, WithVersion("v1.2.3"), WithPhaseProvider(func() string {
+		return phase
+	}))
+
+	req := httptest.NewRequest(http.MethodGet, "/status", nil)
+	rec := httptest.NewRecorder()
+	s.handleStatus(rec, req)
+
+	require.Equal(t, http.StatusOK, rec.Code)
+	body, _ := io.ReadAll(rec.Body)
+	var got map[string]string
+	require.NoError(t, json.Unmarshal(body, &got))
+	require.Equal(t, "snapshot", got["phase"])
+
+	phase = "replication"
+	rec = httptest.NewRecorder()
+	s.handleStatus(rec, req)
+	body, _ = io.ReadAll(rec.Body)
+	require.NoError(t, json.Unmarshal(body, &got))
+	require.Equal(t, "replication", got["phase"])
+}
+
 func TestListenServeShutdown(t *testing.T) {
 	t.Parallel()
 	s := NewServer(Config{Address: "127.0.0.1:0"})

internal/phase/phase.go

@@ -0,0 +1,50 @@
+// SPDX-License-Identifier: Apache-2.0
+
+// Package phase tracks whether a running pgstream process is in the initial
+// snapshot or logical replication (streaming) phase.
+package phase
+
+import "sync/atomic"
+
+// Phase identifies the current pipeline stage.
+type Phase string
+
+const (
+	// Snapshot is the initial data/schema snapshot stage.
+	Snapshot Phase = "snapshot"
+	// Replication is the logical replication / streaming stage.
+	Replication Phase = "replication"
+)
+
+// Tracker holds the current pipeline phase. It is safe for concurrent use.
+type Tracker struct {
+	current atomic.Value // stores Phase
+}
+
+// NewTracker returns a tracker with an empty phase until the first Set.
+func NewTracker() *Tracker {
+	t := &Tracker{}
+	t.current.Store(Phase(""))
+	return t
+}
+
+// Set updates the current phase.
+func (t *Tracker) Set(p Phase) {
+	if t == nil {
+		return
+	}
+	t.current.Store(p)
+}
+
+// Get returns the current phase, or empty if unset / tracker is nil.
+func (t *Tracker) Get() Phase {
+	if t == nil {
+		return ""
+	}
+	v := t.current.Load()
+	p, ok := v.(Phase)
+	if !ok {
+		return ""
+	}
+	return p
+}

internal/phase/phase_test.go

@@ -0,0 +1,31 @@
+// SPDX-License-Identifier: Apache-2.0
+
+package phase
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestTracker_GetSet(t *testing.T) {
+	t.Parallel()
+
+	tr := NewTracker()
+	require.Equal(t, Phase(""), tr.Get())
+
+	tr.Set(Snapshot)
+	require.Equal(t, Snapshot, tr.Get())
+
+	tr.Set(Replication)
+	require.Equal(t, Replication, tr.Get())
+}
+
+func TestTracker_NilSafe(t *testing.T) {
+	t.Parallel()
+
+	var tr *Tracker
+	require.Equal(t, Phase(""), tr.Get())
+	tr.Set(Snapshot) // must not panic
+	require.Equal(t, Phase(""), tr.Get())
+}