War Story: How We Fixed a Cross-Region Replication Lag in Our PostgreSQL 16 Cluster Using Patroni and pglogical

-- Function to monitor pglogical replication lag across all active subscriptions
-- Returns: subscription_name, upstream_node, lag_bytes, lag_seconds, status
-- Includes error handling for missing pglogical extensions or inactive subscriptions
CREATE OR REPLACE FUNCTION monitor_pglogical_lag()
RETURNS TABLE(
    sub_name TEXT,
    upstream_node TEXT,
    lag_bytes BIGINT,
    lag_seconds NUMERIC(10,3),
    status TEXT
) AS $$
DECLARE
    rec RECORD;
    current_lsn pg_lsn;
    upstream_lsn pg_lsn;
    lag_bytes_calc BIGINT;
    lag_sec_calc NUMERIC(10,3);
BEGIN
    -- Check if pglogical extension is installed
    IF NOT EXISTS (SELECT 1 FROM pg_extension WHERE extname = 'pglogical') THEN
        RAISE WARNING 'pglogical extension not installed on node %', current_setting('cluster_name');
        RETURN;
    END IF;

    -- Get current local LSN
    SELECT pg_current_wal_lsn() INTO current_lsn;

    -- Iterate over all active pglogical subscriptions
    FOR rec IN 
        SELECT sub_name, sub_conninfo, sub_slot_name 
        FROM pglogical.subscription 
        WHERE sub_enabled = true
    LOOP
        BEGIN
            -- Get upstream LSN from replication slot (error handling for missing slots)
            SELECT slot_restart_lsn INTO upstream_lsn 
            FROM pg_replication_slots 
            WHERE slot_name = rec.sub_slot_name AND slot_type = 'logical';

            IF upstream_lsn IS NULL THEN
                status := 'ERROR: Replication slot ' || rec.sub_slot_name || ' not found';
                lag_bytes := -1;
                lag_seconds := -1;
                upstream_node := 'unknown';
                RETURN NEXT;
                CONTINUE;
            END IF;

            -- Calculate lag in bytes (using pg_wal_lsn_diff, available in PG10+)
            lag_bytes_calc := pg_wal_lsn_diff(current_lsn, upstream_lsn);

            -- Estimate lag in seconds (assuming average WAL write rate of 1.2MB/sec for our workload)
            -- This is a rough estimate; for production use, track WAL generation rate over time
            IF lag_bytes_calc < 0 THEN
                -- Upstream is ahead of local (should not happen for subscriptions)
                lag_sec_calc := 0.000;
                status := 'WARNING: Upstream ahead of local node';
            ELSE
                lag_sec_calc := (lag_bytes_calc::NUMERIC) / (1200::NUMERIC); -- 1200 bytes/sec = 1.2MB/sec
                status := CASE 
                    WHEN lag_sec_calc < 0.1 THEN 'HEALTHY'
                    WHEN lag_sec_calc < 1.0 THEN 'WARNING'
                    ELSE 'CRITICAL'
                END;
            END IF;

            -- Get upstream node name from connection info (parse host)
            SELECT split_part(split_part(rec.sub_conninfo, 'host=', 2), ' ', 1) INTO upstream_node;

            -- Return row
            sub_name := rec.sub_name;
            lag_bytes := lag_bytes_calc;
            lag_seconds := lag_sec_calc;
            RETURN NEXT;

        EXCEPTION WHEN OTHERS THEN
            -- Catch any unexpected errors per subscription to avoid function failure
            status := 'ERROR: ' || SQLERRM;
            sub_name := rec.sub_name;
            upstream_node := 'unknown';
            lag_bytes := -1;
            lag_seconds := -1;
            RETURN NEXT;
        END;
    END LOOP;

    RETURN;
END;
$$ LANGUAGE plpgsql SECURITY DEFINER;

-- Grant access to monitoring role
GRANT EXECUTE ON FUNCTION monitor_pglogical_lag() TO monitoring_role;

-- Example usage:
-- SELECT * FROM monitor_pglogical_lag();

#!/usr/bin/env python3
"""
Patroni Failover Automator for pglogical Replication Lag
Monitors replication lag via the pglogical monitor function, triggers Patroni failover
if lag exceeds 500ms for 3 consecutive checks. Includes error handling for API failures,
network issues, and invalid cluster states.
"""

import requests
import time
import logging
import sys
from typing import Dict, Optional, Tuple

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[logging.StreamHandler(sys.stdout)]
)
logger = logging.getLogger(__name__)

# Configuration (load from env in production, hardcoded here for example)
PATRONI_API_URL = "http://localhost:8008"  # Default Patroni REST API port
DB_CONNECTION_STRING = "host=localhost dbname=postgres user=monitoring_role password=monitor_pass"
LAG_THRESHOLD_MS = 500  # Failover if lag exceeds this
CONSECUTIVE_CHECKS = 3  # Number of consecutive high lag checks before failover
CHECK_INTERVAL_SEC = 10  # Time between lag checks

def get_pglogical_lag() -> Optional[float]:
    """Fetch p99 replication lag from pglogical monitor function. Returns lag in ms, or None on error."""
    try:
        import psycopg2  # Import here to avoid hard dependency if not used
        with psycopg2.connect(DB_CONNECTION_STRING) as conn:
            conn.autocommit = True
            with conn.cursor() as cur:
                cur.execute("SELECT MAX(lag_seconds) * 1000 FROM monitor_pglogical_lag()")
                result = cur.fetchone()
                if result and result[0] is not None:
                    return float(result[0])
                logger.warning("No active pglogical subscriptions found")
                return 0.0  # Treat no subscriptions as 0 lag (nothing to replicate)
    except ImportError:
        logger.error("psycopg2 not installed. Install with: pip install psycopg2-binary")
        return None
    except psycopg2.Error as e:
        logger.error(f"Database error fetching lag: {e}")
        return None
    except Exception as e:
        logger.error(f"Unexpected error fetching lag: {e}")
        return None

def get_patroni_cluster_state() -> Optional[Dict]:
    """Fetch current Patroni cluster state from REST API."""
    try:
        resp = requests.get(f"{PATRONI_API_URL}/cluster", timeout=5)
        resp.raise_for_status()
        return resp.json()
    except requests.exceptions.RequestException as e:
        logger.error(f"Failed to fetch Patroni cluster state: {e}")
        return None

def trigger_patroni_failover() -> bool:
    """Trigger a manual failover via Patroni REST API. Returns True on success."""
    try:
        # Get current leader first
        state = get_patroni_cluster_state()
        if not state:
            return False
        leader = state.get("leader")
        if not leader:
            logger.error("No leader found in Patroni cluster")
            return False

        # Trigger failover (Patroni will elect a new leader automatically)
        resp = requests.post(
            f"{PATRONI_API_URL}/failover",
            json={"leader": leader},
            timeout=10
        )
        resp.raise_for_status()
        logger.info(f"Triggered Patroni failover from leader {leader}")
        return True
    except requests.exceptions.RequestException as e:
        logger.error(f"Failed to trigger failover: {e}")
        return False
    except Exception as e:
        logger.error(f"Unexpected error during failover: {e}")
        return False

def main():
    consecutive_high_lag = 0
    logger.info("Starting Patroni failover automator for pglogical lag")

    while True:
        try:
            # Check lag
            lag_ms = get_pglogical_lag()
            if lag_ms is None:
                logger.warning("Failed to fetch lag, skipping check")
                time.sleep(CHECK_INTERVAL_SEC)
                continue

            logger.info(f"Current p99 replication lag: {lag_ms:.2f}ms")

            # Check if lag exceeds threshold
            if lag_ms > LAG_THRESHOLD_MS:
                consecutive_high_lag += 1
                logger.warning(f"High lag detected ({lag_ms:.2f}ms). Consecutive checks: {consecutive_high_lag}/{CONSECUTIVE_CHECKS}")
                if consecutive_high_lag >= CONSECUTIVE_CHECKS:
                    logger.error(f"Lag exceeded threshold for {CONSECUTIVE_CHECKS} consecutive checks. Triggering failover.")
                    if trigger_patroni_failover():
                        consecutive_high_lag = 0  # Reset after failover
                        time.sleep(60)  # Wait for failover to complete
                    else:
                        logger.error("Failover failed, retrying in 30 seconds")
                        time.sleep(30)
            else:
                consecutive_high_lag = 0  # Reset if lag is healthy

            time.sleep(CHECK_INTERVAL_SEC)
        except KeyboardInterrupt:
            logger.info("Shutting down failover automator")
            break
        except Exception as e:
            logger.error(f"Unexpected error in main loop: {e}")
            time.sleep(CHECK_INTERVAL_SEC)

if __name__ == "__main__":
    main()

-- Full pglogical 2.4.2 Setup Script for PostgreSQL 16 Cross-Region Replication
-- Run this on the provider (us-east-1 primary) first, then on the subscriber (eu-central-1)
-- Includes error handling, idempotent checks, and validation steps

-- =========================
-- STEP 1: Provider Setup (us-east-1 Primary)
-- =========================

-- Enable pglogical extension (idempotent)
CREATE EXTENSION IF NOT EXISTS pglogical;

-- Create pglogical node for provider
DO $$
BEGIN
    IF NOT EXISTS (SELECT 1 FROM pglogical.node WHERE node_name = 'provider_useast1') THEN
        PERFORM pglogical.create_node(
            node_name := 'provider_useast1',
            dsn := 'host=useast1-primary.example.com dbname=app_db user=pglogical_user password=secure_pass'
        );
        RAISE NOTICE 'Created pglogical provider node: provider_useast1';
    ELSE
        RAISE NOTICE 'pglogical provider node already exists: provider_useast1';
    END IF;
END;
$$;

-- Create replication set for all tables (idempotent)
DO $$
BEGIN
    IF NOT EXISTS (SELECT 1 FROM pglogical.replication_set WHERE set_name = 'app_replication_set') THEN
        PERFORM pglogical.create_replication_set(
            set_name := 'app_replication_set',
            replicate_insert := true,
            replicate_update := true,
            replicate_delete := true,
            replicate_truncate := false  -- We handle truncates manually
        );
        RAISE NOTICE 'Created replication set: app_replication_set';
    ELSE
        RAISE NOTICE 'Replication set already exists: app_replication_set';
    END IF;
END;
$$;

-- Add all tables in app schema to replication set (exclude temporary tables)
DO $$
DECLARE
    tbl RECORD;
BEGIN
    FOR tbl IN 
        SELECT tablename FROM pg_tables 
        WHERE schemaname = 'app' AND tablename NOT LIKE 'tmp_%'
    LOOP
        BEGIN
            PERFORM pglogical.replication_set_add_table(
                set_name := 'app_replication_set',
                table_name := 'app.' || tbl.tablename,
                synchronize_data := true  -- Sync existing data on add
            );
            RAISE NOTICE 'Added table app.% to replication set', tbl.tablename;
        EXCEPTION WHEN OTHERS THEN
            RAISE WARNING 'Failed to add table app.% to replication set: %', tbl.tablename, SQLERRM;
        END;
    END LOOP;
END;
$$;

-- Create replication slot for subscriber (idempotent)
DO $$
BEGIN
    IF NOT EXISTS (SELECT 1 FROM pg_replication_slots WHERE slot_name = 'pglogical_subscriber_eucentral1') THEN
        PERFORM pglogical.create_subscription_slot(
            slot_name := 'pglogical_subscriber_eucentral1',
            plugin := 'pglogical_output'
        );
        RAISE NOTICE 'Created replication slot: pglogical_subscriber_eucentral1';
    ELSE
        RAISE NOTICE 'Replication slot already exists: pglogical_subscriber_eucentral1';
    END IF;
END;
$$;

-- =========================
-- STEP 2: Subscriber Setup (eu-central-1 Standby)
-- =========================

-- Enable pglogical extension (idempotent)
CREATE EXTENSION IF NOT EXISTS pglogical;

-- Create pglogical node for subscriber
DO $$
BEGIN
    IF NOT EXISTS (SELECT 1 FROM pglogical.node WHERE node_name = 'subscriber_eucentral1') THEN
        PERFORM pglogical.create_node(
            node_name := 'subscriber_eucentral1',
            dsn := 'host=eucentral1-standby.example.com dbname=app_db user=pglogical_user password=secure_pass'
        );
        RAISE NOTICE 'Created pglogical subscriber node: subscriber_eucentral1';
    ELSE
        RAISE NOTICE 'pglogical subscriber node already exists: subscriber_eucentral1';
    END IF;
END;
$$;

-- Create subscription to provider (idempotent)
DO $$
BEGIN
    IF NOT EXISTS (SELECT 1 FROM pglogical.subscription WHERE sub_name = 'sub_useast1_to_eucentral1') THEN
        PERFORM pglogical.create_subscription(
            subscription_name := 'sub_useast1_to_eucentral1',
            provider_dsn := 'host=useast1-primary.example.com dbname=app_db user=pglogical_user password=secure_pass',
            replication_sets := ARRAY['app_replication_set'],
            synchronize_structure := false,  -- We manage DDL separately via CI/CD
            synchronize_data := true,  -- Sync existing data on creation
            forward_origins := 'all'
        );
        RAISE NOTICE 'Created pglogical subscription: sub_useast1_to_eucentral1';
    ELSE
        RAISE NOTICE 'pglogical subscription already exists: sub_useast1_to_eucentral1';
    END IF;
END;
$$;

-- =========================
-- STEP 3: Validation
-- =========================

-- Check replication status on subscriber
SELECT 
    sub_name, 
    sub_enabled, 
    sub_conninfo, 
    pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), slot_restart_lsn)) AS lag_bytes
FROM pglogical.subscription s
LEFT JOIN pg_replication_slots r ON s.sub_slot_name = r.slot_name;

-- Verify data sync (run on subscriber)
SELECT COUNT(*) FROM app.users;  -- Should match provider count

-- Patroni custom health check SQL for pglogical slots
SELECT 
    CASE 
        WHEN COUNT(*) = 0 THEN true  -- No subscriptions, healthy
        WHEN MAX(pg_wal_lsn_diff(pg_current_wal_lsn(), slot_restart_lsn)) < 1200 THEN true  -- Lag <1s
        ELSE false
    END AS is_healthy
FROM pglogical.subscription s
LEFT JOIN pg_replication_slots r ON s.sub_slot_name = r.slot_name
WHERE s.sub_enabled = true;

-- Configure async pglogical subscription (no synchronous commit)
SELECT pglogical.create_subscription(
    subscription_name := 'sub_useast1_to_eucentral1',
    provider_dsn := 'host=useast1-primary dbname=app user=pglogical password=pass',
    replication_sets := ARRAY['app_replication_set'],
    synchronize_structure := false,
    synchronize_data := true,
    forward_origins := 'all',
    apply_delay := 0  -- No artificial delay
);

# pgbench custom script for benchmarking (save as pglogical_bench.sql)
\set user_id random(1, 1000000)
\set amount random(100, 10000)
BEGIN;
INSERT INTO app.transactions (user_id, amount, created_at) 
VALUES (:user_id, :amount, NOW());
UPDATE app.users SET balance = balance - :amount 
WHERE id = :user_id;
COMMIT;