Capturing Live Cost Signals from PostgreSQL Logical Replication #

This page taps a PostgreSQL logical replication slot with psycopg to capture live write and activity volume — the WAL a workload generates — as a real-time cost signal, mapping each decoded change to a tenant cost dimension and flushing consumer feedback so the slot never stalls.

Back to: Real-Time Metric Streaming Setup

CloudWatch and provider APIs tell you what a database costs after the fact; logical replication tells you what it is doing right now, straight from the write-ahead log, with no polling and no query load on the primary. Every INSERT, UPDATE, and DELETE a workload commits flows through a replication slot as a decoded change event, and the volume and shape of those changes is a direct, sub-second proxy for the I/O and storage growth that drives spend. Unlike the CloudWatch-to-Kafka poller, which samples pre-aggregated gauges on a cadence, this is a continuous push stream sourced from the WAL itself — the finest-grained live cost signal PostgreSQL can give you. The records it emits feed the same downstream schema validation for billing data contract and complement the point-in-time snapshots from reading pg_stat_activity for cost tracking.

A slot’s health is measured by replication lag — the byte distance between the WAL write position and the position the consumer has confirmed flushed:

If the consumer stops sending feedback, this distance grows without bound and WAL accumulates on disk, so flushing correctly is not optional — it is what keeps the primary from filling up.

Prerequisites #

Before creating a slot, confirm the server and role are configured for logical decoding.

• Server config: wal_level = logical (requires a restart), and max_replication_slots and max_wal_senders each at least 1. On RDS/Aurora PostgreSQL set rds.logical_replication = 1 in the parameter group and reboot.

• Role privileges: the connecting role needs the REPLICATION attribute (or rds_replication on RDS). Grant nothing broader — a decoding consumer reads change data across every table, so scope it like any cross-tenant reader per access control for cost data.

  Copy

  CREATE ROLE cost_decoder WITH LOGIN REPLICATION PASSWORD :'pw';
  -- On RDS instead: GRANT rds_replication TO cost_decoder;
  

• Output plugin: pgoutput ships with PostgreSQL 10+ (no install). wal2json emits ready-to-parse JSON and must be installed on the server; this page uses wal2json for readability and notes the pgoutput differences.

• Python: 3.10+ with psycopg 3 (not psycopg2), which exposes logical replication through its dedicated connection classes.

  Copy

  pip install "psycopg[binary]>=3.2"
  

Step-by-Step Implementation #

The consumer creates (or reuses) a slot, opens a replication-mode connection, starts streaming from the slot’s confirmed position, maps each decoded change to a cost dimension, and periodically sends feedback so the server can advance confirmed_flush_lsn and recycle WAL.

Step 1 — Create the replication slot #

The slot is a server-side cursor into the WAL that persists across disconnects, so it is created once and reused. Creating it separately from streaming means a consumer restart resumes exactly where feedback left off rather than losing changes.

import os
import psycopg

DSN = os.environ["PG_DSN"]          # e.g. "host=... dbname=... user=cost_decoder"
SLOT = "cost_signal_slot"

def ensure_slot() -> None:
    """Create the logical slot if it does not exist (idempotent)."""
    with psycopg.connect(DSN, autocommit=True) as conn:
        exists = conn.execute(
            "SELECT 1 FROM pg_replication_slots WHERE slot_name = %s", (SLOT,)
        ).fetchone()
        if not exists:
            # wal2json plugin; use 'pgoutput' with a PUBLICATION for the built-in path.
            conn.execute(
                "SELECT pg_create_logical_replication_slot(%s, 'wal2json')", (SLOT,)
            )

# Expected (first run):
#   pg_replication_slots now has one row: slot_name='cost_signal_slot',
#   plugin='wal2json', slot_type='logical', active='f'

Step 2 — Map a decoded change to a cost dimension #

Each wal2json change carries the schema, table, and operation. The mapping turns raw DML into a weighted cost signal: writes to large or high-churn tables cost more I/O than trivial lookups, and the tenant column identifies whose budget the write belongs to. This classifier is pure so it can be unit-tested against captured WAL fixtures.

from dataclasses import dataclass
from datetime import datetime, timezone

# Relative I/O weight per operation — tune from your workload's write amplification.
OP_WEIGHT = {"insert": 1.0, "update": 1.5, "delete": 0.8}

@dataclass(frozen=True)
class CostEvent:
    tenant: str
    schema: str
    table: str
    op: str
    weight: float
    observed_at: datetime

def to_cost_event(change: dict) -> CostEvent | None:
    """Map one wal2json change object to a weighted cost signal."""
    op = change.get("kind")                       # insert | update | delete
    if op not in OP_WEIGHT:
        return None                               # skip truncate/message events
    cols = dict(zip(change.get("columnnames", []), change.get("columnvalues", [])))
    tenant = str(cols.get("tenant_id", "unattributed"))
    return CostEvent(
        tenant=tenant,
        schema=change["schema"],
        table=change["table"],
        op=op,
        weight=OP_WEIGHT[op],
        observed_at=datetime.now(timezone.utc),
    )

# Expected:
#   to_cost_event({"kind":"update","schema":"public","table":"orders",
#                  "columnnames":["tenant_id"],"columnvalues":["acme"]})
#   -> CostEvent(tenant='acme', schema='public', table='orders',
#               op='update', weight=1.5, observed_at=...)

Step 3 — Stream changes and flush feedback #

psycopg exposes logical decoding through LogicalReplicationConnection and a start_replication cursor that yields ReplicationMessage objects. The critical discipline is feedback: after processing a message you call send_feedback(flush_lsn=msg.data_start) so the server knows it can recycle that WAL. Flushing only after a message is durably handled downstream is what makes the stream at-least-once rather than lossy.

The flow below shows the streaming loop: the primary pushes decoded changes, the consumer maps and forwards each one, and periodic feedback advances the slot so WAL is recycled.

import json
import logging
from datetime import datetime, timezone, timedelta
from psycopg import connect
from psycopg.rows import tuple_row

FEEDBACK_INTERVAL = timedelta(seconds=10)

def stream_cost_signals(publish) -> None:
    """Stream the slot, mapping changes to cost events; publish() forwards them."""
    # Replication requires a dedicated connection type.
    from psycopg import Connection
    with connect(DSN, autocommit=True) as conn:
        cur = conn.cursor()
        # wal2json options: format-version 2 emits one message per change.
        cur.execute(
            "START_REPLICATION SLOT %s LOGICAL 0/0 "
            "(\"format-version\", '2', \"include-timestamp\", 'on')" % SLOT
        )
        last_feedback = datetime.now(timezone.utc)
        for msg in cur.stream():                          # yields replication messages
            payload = json.loads(msg.payload)
            if payload.get("action") in ("I", "U", "D"):
                change = _wal2json_v2_to_change(payload)  # normalize v2 shape
                event = to_cost_event(change)
                if event is not None:
                    publish(event)                        # forward downstream first
            now = datetime.now(timezone.utc)
            # Only confirm flush AFTER downstream accepted the event.
            if now - last_feedback >= FEEDBACK_INTERVAL:
                cur.send_feedback(flush_lsn=msg.wal_end)
                last_feedback = now
                logging.info("flushed up to LSN %s", msg.wal_end)

Expected log output while a workload writes:

INFO:root:flushed up to LSN 3A/7F0021C8
INFO:root:flushed up to LSN 3A/7F00A410

The START_REPLICATION syntax above uses psycopg’s logical replication cursor; with pgoutput you instead pass proto_version and publication_names options and decode the binary protocol rather than JSON — wal2json is chosen here so the decode step is a single json.loads.

Verification #

Confirm the slot is active and lag is bounded before treating the stream as a live signal.

1. Inspect the slot on the primary:

   Copy

   SELECT slot_name, active, confirmed_flush_lsn,
          pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), confirmed_flush_lsn))
            AS retained_wal
   FROM pg_replication_slots WHERE slot_name = 'cost_signal_slot';
   

2. Expected cost-event shape emitted downstream:

   Copy

   {"tenant": "acme", "schema": "public", "table": "orders", "op": "update", "weight": 1.5, "observed_at": "2026-07-18T14:03:11Z"}
   

3. Watch retained_wal fall to near zero after each feedback interval. A steadily growing retained_wal means feedback is not reaching the server (or the consumer died) — the single most important health check for any slot-based consumer.

Gotchas & Edge Cases #

• An inactive slot still retains WAL forever. A dropped or crashed consumer leaves the slot active = f while confirmed_flush_lsn stays frozen, and the primary keeps every WAL segment since then — this fills the data disk and takes the primary down. Always monitor retained_wal and drop abandoned slots with pg_drop_replication_slot.
• Feedback before durable handoff loses data. If you send_feedback before the downstream pipeline has durably accepted an event, a crash between flush and delivery drops that change permanently. Flush only after the event is safely handed off, exactly as the dead-letter and retry model requires for at-least-once delivery.
• UPDATE/DELETE old values need REPLICA IDENTITY. By default a change event carries only the new tuple and the primary key. To attribute a tenant on DELETE, or to see pre-image values, set ALTER TABLE ... REPLICA IDENTITY FULL on the relevant tables — otherwise columnvalues omits the fields your classifier reads.
• DDL is not replicated. Logical decoding streams row changes, not schema changes. A new table or renamed column does not appear in the stream and can break your mapping silently — version your table-to-dimension map and reconcile against the catalog, the same schema-drift discipline as schema validation for billing data.
• Large transactions buffer until commit. Standard logical decoding does not emit a transaction’s changes until it commits, so a long bulk load appears as a sudden burst, not a smooth stream. PostgreSQL 14+ supports streaming in-progress transactions (streaming = 'on' with pgoutput) if burst latency matters.
• Aurora and read replicas. Logical slots live on the writer only; a failover invalidates the slot and the new writer starts fresh. Persist your last flushed LSN externally so a post-failover consumer can reason about the gap rather than assuming continuity.

Frequently Asked Questions #

How is this different from polling CloudWatch or pg_stat_activity? #

CloudWatch and pg_stat_activity are sampled snapshots — you ask “what is the state now?” on a cadence and miss everything between samples. Logical replication is a continuous push of every committed change, so it captures write volume at full fidelity with zero query load on the primary. Use it when you need per-write attribution; use polling when a periodic gauge is enough.

Does streaming the WAL add load to the primary? #

Decoding runs in a WAL sender process and reads WAL that is written anyway, so the marginal cost is modest — mostly the plugin’s CPU to format each change. The real risk is not CPU but disk: a slow or dead consumer forces the primary to retain WAL, so the operational burden is keeping the consumer healthy and feedback flowing, not the decode itself.

Should I use wal2json or pgoutput? #

pgoutput is built in and efficient but emits a binary protocol you must decode, and it requires a PUBLICATION. wal2json emits JSON you parse with one json.loads, which is far simpler for a cost consumer and easy to unit-test against fixtures. Choose pgoutput when you need maximum throughput or in-progress transaction streaming; choose wal2json for clarity, as this page does.

What happens if my consumer falls behind? #

The slot retains all WAL from confirmed_flush_lsn forward, so no data is lost — but retained WAL grows and threatens the primary’s disk. A backing buffer decouples decode speed from downstream speed; see buffering cost metrics with Redis Streams for absorbing bursts so you can keep flushing the slot promptly.

How do I attribute cost when a table has no tenant_id column? #

Fall back to a lookup: map (schema, table) to an owning cost center from a maintained catalog, and route rows that match neither a tenant column nor the catalog to an unattributed dimension for review rather than silently dropping them. This mirrors the quarantine-don’t-guess rule the schema validation layer applies to missing attribution keys.

Related #

• Streaming RDS CloudWatch Metrics to Kafka with aiokafka — the polled, gauge-based counterpart to this push-based WAL stream.
• Buffering Cost Metrics with Redis Streams — a durable buffer that absorbs decode bursts so the slot can keep flushing.
• Real-Time Metric Streaming Setup — the parent topic covering the sub-minute streaming tier end to end.

Back to: Real-Time Metric Streaming Setup