feat(telemetry): Expose key stack metrics to Prometheus (#4541)

Export the following metrics on the /metrics endpoint:

- defined/active shape counts,
- replication lag (byte-based slot lag and a time-based receive-lag
histogram)
- retained WAL size
- electric.plug.serve_shape.requests.count (tagged by HTTP status,
`electric_plug_serve_shape_requests_count{status="200"|"409"|...}`)
- admission_control.(aquire|reject).count

These were already collected and exported to OTel/StatsD/Call-Home but
not Prometheus, which only served system-level metrics. A new
additional_prometheus_metrics telemetry option routes them through the
single shared Prometheus aggregator without double-reporting via the
other reporters. Single stack per instance.

Fixes #4535

Author: magnetised

.changeset/prometheus-stack-metrics.md

@@ -0,0 +1,18 @@
+---
+'@core/electric-telemetry': patch
+'@core/sync-service': patch
+---
+
+Expose key stack-level metrics on the Prometheus `/metrics` endpoint: number of
+defined shapes, number of active shapes, replication lag (byte-based slot lag
+and time-based receive-lag histogram), retained WAL size, and per-status-code
+counts of shape-endpoint HTTP responses
+(`electric_plug_serve_shape_requests_count{status="200"}`, `409`, `503`, ...),
+and admission-control concurrency/rejection counts
+(`electric_admission_control_acquire_current{kind=...}`,
+`electric_admission_control_reject_count{kind=...}`).
+These metrics were already collected and exported to OTel/StatsD/Call-Home but
+not to Prometheus, which previously only served system-level (CPU/RAM/BEAM)
+metrics. A new `additional_prometheus_metrics` telemetry option routes them
+through the single shared Prometheus aggregator without double-reporting via the
+other reporters. Assumes a single stack per instance.

packages/electric-telemetry/lib/electric/telemetry/application_telemetry.ex

@@ -42,13 +42,17 @@ defmodule ElectricTelemetry.ApplicationTelemetry do
   defp exporter_child_specs(opts) do
     metrics = metrics(opts)
 
+    # Metrics that should reach Prometheus only, e.g. stack-level metrics that the other
+    # reporters already export per-stack. Appending them here avoids double-reporting.
+    prometheus_metrics = metrics ++ Map.get(opts, :additional_prometheus_metrics, [])
+
     [
       Reporters.CallHomeReporter.child_spec(
         opts,
         metrics: Reporters.CallHomeReporter.application_metrics()
       ),
       Reporters.Otel.child_spec(opts, metrics: metrics),
-      Reporters.Prometheus.child_spec(opts, metrics: metrics),
+      Reporters.Prometheus.child_spec(opts, metrics: prometheus_metrics),
       Reporters.Statsd.child_spec(opts, metrics: metrics)
     ]
   end

packages/electric-telemetry/lib/electric/telemetry/opts.ex

@@ -76,6 +76,22 @@ defmodule ElectricTelemetry.Opts do
               {:struct, Telemetry.Metrics.Distribution}
             ]}}
       ],
+      # Metrics exported only via the Prometheus reporter, in addition to the shared
+      # `additional_metrics`. Used to expose stack-level metrics on the `/metrics` endpoint
+      # without double-reporting them through the OTel/StatsD/Call-Home reporters.
+      additional_prometheus_metrics: [
+        type:
+          {:list,
+           {:or,
+            [
+              {:struct, Telemetry.Metrics.Counter},
+              {:struct, Telemetry.Metrics.LastValue},
+              {:struct, Telemetry.Metrics.Sum},
+              {:struct, Telemetry.Metrics.Summary},
+              {:struct, Telemetry.Metrics.Distribution}
+            ]}},
+        default: []
+      ],
       otel_opts: [
         type: :keyword_list,
         keys: [

packages/electric-telemetry/test/electric/telemetry/application_telemetry_test.exs

@@ -3,6 +3,47 @@ defmodule ElectricTelemetry.ApplicationTelemetryTest do
 
   alias ElectricTelemetry.ApplicationTelemetry
 
+  describe "additional_prometheus_metrics" do
+    setup do
+      {:ok, opts} =
+        ElectricTelemetry.validate_options(
+          instance_id: "test-instance",
+          version: "1.0.0",
+          reporters: [prometheus?: true, statsd_host: "localhost", otel_metrics?: true],
+          additional_prometheus_metrics: [Telemetry.Metrics.last_value("test.custom.gauge")]
+        )
+
+      {:ok, {_flags, children}} = ApplicationTelemetry.init(opts)
+      %{children: children}
+    end
+
+    test "are exported to the Prometheus reporter", %{children: children} do
+      metrics = reporter_metrics(children, :prometheus_metrics)
+      assert Enum.any?(metrics, &(&1.name == [:test, :custom, :gauge]))
+    end
+
+    test "are not exported to the OTel or StatsD reporters", %{children: children} do
+      refute Enum.any?(
+               reporter_metrics(children, OtelMetricExporter),
+               &(&1.name == [:test, :custom, :gauge])
+             )
+
+      refute Enum.any?(
+               reporter_metrics(children, TelemetryMetricsStatsd),
+               &(&1.name == [:test, :custom, :gauge])
+             )
+    end
+  end
+
+  # `Supervisor.init/2` normalises child specs into maps, nesting the reporter's start args
+  # (which carry `:metrics`) inside `:start`. Reporters are identified by their child spec id.
+  defp reporter_metrics(children, id) do
+    Enum.find_value(children, [], fn
+      %{id: ^id, start: {_m, _f, [opts]}} when is_list(opts) -> Keyword.get(opts, :metrics, [])
+      _ -> false
+    end)
+  end
+
   describe "get_system_memory_usage" do
     test "returns calculated memory stats" do
       case :os.type() do

packages/sync-service/lib/electric/application.ex

@@ -413,6 +413,10 @@ defmodule Electric.Application do
           if(get_env(opts, :call_home_telemetry?), do: get_env(opts, :telemetry_url)),
         otel_metrics?: not is_nil(Application.get_env(:otel_metric_exporter, :otlp_endpoint))
       ],
+      # Export the key stack-level metrics (shapes, replication lag, retained WAL) to the
+      # Prometheus `/metrics` endpoint. They already reach the other reporters per-stack via
+      # StackTelemetry, so they go through this Prometheus-only seam to avoid double-reporting.
+      additional_prometheus_metrics: Electric.StackSupervisor.Telemetry.prometheus_metrics(),
       otel_opts: get_opts(opts, export_period: :otel_export_period)
     ]
   end

packages/sync-service/lib/electric/stack_supervisor/telemetry.ex

@@ -26,6 +26,52 @@ defmodule Electric.StackSupervisor.Telemetry do
     ]
   end
 
+  @doc """
+  The curated subset of stack-level metrics exported to the Prometheus `/metrics` endpoint.
+
+  These are emitted by the periodic measurements above (and, for `receive_lag`, by the
+  replication client) and are already exported to OTel/StatsD/Call-Home via
+  `ElectricTelemetry.StackTelemetry`. They are passed to `ApplicationTelemetry` as
+  `additional_prometheus_metrics` so they reach the single shared Prometheus aggregator
+  without being double-reported through the other reporters.
+
+  Note: these definitions carry no `stack_id` tag, so they assume a single stack per
+  instance. With multiple stacks the series would collide in the shared aggregator.
+  """
+  def prometheus_metrics do
+    [
+      # number of defined shapes
+      Telemetry.Metrics.last_value("electric.shapes.total_shapes.count"),
+      # number of active shapes
+      Telemetry.Metrics.last_value("electric.shapes.active_shapes.count"),
+      # retained WAL size (bytes Postgres is keeping for Electric's replication slot)
+      Telemetry.Metrics.last_value("electric.postgres.replication.slot_retained_wal_size",
+        unit: :byte
+      ),
+      # replication lag, byte-based: how far behind Postgres' WAL Electric is
+      Telemetry.Metrics.last_value("electric.postgres.replication.slot_confirmed_flush_lsn_lag",
+        unit: :byte
+      ),
+      # replication lag, time-based: wall-clock staleness of received transactions
+      Telemetry.Metrics.distribution(
+        "electric.postgres.replication.transaction_received.receive_lag",
+        unit: :millisecond
+      ),
+      # HTTP response status-code counts for the shape endpoint (200s, 409s, 50xs, ...).
+      # The event also carries :known_error and :live; we tag by :status only so the series
+      # are a plain per-code count.
+      Telemetry.Metrics.counter("electric.plug.serve_shape.requests.count",
+        event_name: [:electric, :plug, :serve_shape],
+        measurement: :count,
+        tags: [:status]
+      ),
+      # admission control: current in-flight concurrency per kind
+      Telemetry.Metrics.last_value("electric.admission_control.acquire.current", tags: [:kind]),
+      # admission control: number of rejected requests per kind
+      Telemetry.Metrics.sum("electric.admission_control.reject.count", tags: [:kind])
+    ]
+  end
+
   def count_shapes(stack_id, _telemetry_opts) do
     # Emit active_shapes first so that a failure in shape_counts (e.g. shape cache not yet
     # started during stack startup) doesn't drop this metric for the tick.

packages/sync-service/test/electric/config_test.exs

@@ -57,6 +57,16 @@ defmodule Electric.ConfigTest do
       )
     end
 
+    test "configuration/1 exposes the curated stack metrics to Prometheus", ctx do
+      config =
+        Electric.Application.configuration(
+          Keyword.take(ctx.initial_config, [:replication_connection_opts])
+        )
+
+      assert config[:telemetry_opts][:additional_prometheus_metrics] ==
+               Electric.StackSupervisor.Telemetry.prometheus_metrics()
+    end
+
     test "api/0" do
       Electric.Application.api()
     end

packages/sync-service/test/electric/stack_supervisor/telemetry_test.exs

@@ -0,0 +1,130 @@
+if Electric.telemetry_enabled?() and Code.ensure_loaded?(ElectricTelemetry.Reporters.Prometheus) do
+  defmodule Electric.StackSupervisor.TelemetryTest do
+    use ExUnit.Case, async: true
+
+    alias Electric.StackSupervisor.Telemetry
+
+    describe "prometheus_metrics/0" do
+      test "the curated stack metrics are scrapable via the Prometheus reporter" do
+        name = :"test_prometheus_#{System.unique_integer([:positive])}"
+
+        # Build the core through the real reporter so distribution buckets are applied, exactly
+        # as ApplicationTelemetry does for the /metrics endpoint.
+        {mod, opts} =
+          ElectricTelemetry.Reporters.Prometheus.child_spec(
+            %{reporters: %{prometheus?: true}},
+            name: name,
+            metrics: Telemetry.prometheus_metrics()
+          )
+
+        start_supervised!({mod, opts})
+
+        # The core attaches its telemetry handlers asynchronously (via a `:setup` message).
+        # A scrape is a synchronous call to the registry, so it drains that message and
+        # guarantees the handlers are attached before we emit events below.
+        _ = TelemetryMetricsPrometheus.Core.scrape(name)
+
+        stack_id = "test-stack"
+
+        :telemetry.execute(
+          [:electric, :shapes, :total_shapes],
+          %{count: 7, count_indexed: 4, count_unindexed: 3},
+          %{stack_id: stack_id}
+        )
+
+        :telemetry.execute(
+          [:electric, :shapes, :active_shapes],
+          %{count: 3},
+          %{stack_id: stack_id}
+        )
+
+        :telemetry.execute(
+          [:electric, :postgres, :replication],
+          %{pg_wal_offset: 1, slot_retained_wal_size: 1234, slot_confirmed_flush_lsn_lag: 56},
+          %{stack_id: stack_id}
+        )
+
+        :telemetry.execute(
+          [:electric, :postgres, :replication, :transaction_received],
+          %{receive_lag: 12, bytes: 0, count: 1, operations: 1},
+          %{stack_id: stack_id}
+        )
+
+        scrape = TelemetryMetricsPrometheus.Core.scrape(name)
+
+        # defined shapes
+        assert scrape =~ "electric_shapes_total_shapes_count 7"
+        # active shapes
+        assert scrape =~ "electric_shapes_active_shapes_count 3"
+        # retained WAL size
+        assert scrape =~ "electric_postgres_replication_slot_retained_wal_size 1234"
+        # byte-based replication lag
+        assert scrape =~ "electric_postgres_replication_slot_confirmed_flush_lsn_lag 56"
+        # time-based replication lag (histogram)
+        assert scrape =~ "electric_postgres_replication_transaction_received_receive_lag"
+      end
+
+      test "per-status HTTP request counts are scrapable" do
+        name = :"test_prometheus_status_#{System.unique_integer([:positive])}"
+
+        {mod, opts} =
+          ElectricTelemetry.Reporters.Prometheus.child_spec(
+            %{reporters: %{prometheus?: true}},
+            name: name,
+            metrics: Telemetry.prometheus_metrics()
+          )
+
+        start_supervised!({mod, opts})
+        _ = TelemetryMetricsPrometheus.Core.scrape(name)
+
+        for status <- [200, 200, 409] do
+          :telemetry.execute(
+            [:electric, :plug, :serve_shape],
+            %{count: 1},
+            %{status: status, known_error: false, live: false, stack_id: "test-stack"}
+          )
+        end
+
+        scrape = TelemetryMetricsPrometheus.Core.scrape(name)
+
+        assert scrape =~ ~r/electric_plug_serve_shape_requests_count\{status="200"\} 2/
+        assert scrape =~ ~r/electric_plug_serve_shape_requests_count\{status="409"\} 1/
+      end
+
+      test "admission control metrics are scrapable" do
+        name = :"test_prometheus_admission_#{System.unique_integer([:positive])}"
+
+        {mod, opts} =
+          ElectricTelemetry.Reporters.Prometheus.child_spec(
+            %{reporters: %{prometheus?: true}},
+            name: name,
+            metrics: Telemetry.prometheus_metrics()
+          )
+
+        start_supervised!({mod, opts})
+        _ = TelemetryMetricsPrometheus.Core.scrape(name)
+
+        :telemetry.execute(
+          [:electric, :admission_control, :acquire],
+          %{count: 1, current: 5, limit: 10},
+          %{stack_id: "test-stack", kind: :shape}
+        )
+
+        for _ <- 1..2 do
+          :telemetry.execute(
+            [:electric, :admission_control, :reject],
+            %{count: 1, limit: 10},
+            %{stack_id: "test-stack", reason: :overloaded, kind: :shape, current: 11}
+          )
+        end
+
+        scrape = TelemetryMetricsPrometheus.Core.scrape(name)
+
+        # current concurrency gauge
+        assert scrape =~ ~r/electric_admission_control_acquire_current\{kind="shape"\} 5/
+        # rejection count
+        assert scrape =~ ~r/electric_admission_control_reject_count\{kind="shape"\} 2/
+      end
+    end
+  end
+end