PostgreSQL Replication & High Availability Best Practices¶

# monitoring/replication_monitor.py
import psycopg2
import time
import logging
from datetime import datetime, timedelta

class PostgreSQLReplicationMonitor:
    def __init__(self, primary_config, standby_configs):
        self.primary_config = primary_config
        self.standby_configs = standby_configs
        self.setup_logging()

    def setup_logging(self):
        """Setup logging."""
        logging.basicConfig(level=logging.INFO)
        self.logger = logging.getLogger(__name__)

    def check_primary_status(self):
        """Check primary server status."""
        try:
            conn = psycopg2.connect(**self.primary_config)
            with conn.cursor() as cur:
                # Check if primary is accepting connections
                cur.execute("SELECT 1")
                result = cur.fetchone()

                # Check replication slots
                cur.execute("""
                    SELECT slot_name, active, pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn)) as lag
                    FROM pg_replication_slots;
                """)
                slots = cur.fetchall()

                # Check WAL sender processes
                cur.execute("""
                    SELECT count(*) FROM pg_stat_replication;
                """)
                replication_count = cur.fetchone()[0]

                return {
                    'status': 'healthy',
                    'replication_slots': slots,
                    'replication_count': replication_count
                }
        except Exception as e:
            self.logger.error(f"Primary check failed: {e}")
            return {'status': 'unhealthy', 'error': str(e)}

    def check_standby_status(self, standby_config):
        """Check standby server status."""
        try:
            conn = psycopg2.connect(**standby_config)
            with conn.cursor() as cur:
                # Check if standby is in recovery mode
                cur.execute("SELECT pg_is_in_recovery();")
                in_recovery = cur.fetchone()[0]

                if not in_recovery:
                    return {'status': 'not_standby', 'error': 'Server is not in recovery mode'}

                # Check replication lag
                cur.execute("""
                    SELECT 
                        pg_wal_lsn_diff(pg_last_wal_receive_lsn(), pg_last_wal_replay_lsn()) as lag_bytes,
                        pg_last_wal_receive_lsn(),
                        pg_last_wal_replay_lsn()
                """)
                lag_info = cur.fetchone()

                # Check last activity
                cur.execute("""
                    SELECT 
                        pg_last_xact_replay_timestamp(),
                        pg_last_wal_replay_timestamp()
                """)
                activity_info = cur.fetchone()

                return {
                    'status': 'healthy',
                    'lag_bytes': lag_info[0],
                    'receive_lsn': lag_info[1],
                    'replay_lsn': lag_info[2],
                    'last_xact_replay': activity_info[0],
                    'last_wal_replay': activity_info[1]
                }
        except Exception as e:
            self.logger.error(f"Standby check failed: {e}")
            return {'status': 'unhealthy', 'error': str(e)}

    def check_replication_lag(self, threshold_mb=100):
        """Check replication lag across all standbys."""
        lag_alerts = []

        for standby_name, standby_config in self.standby_configs.items():
            status = self.check_standby_status(standby_config)

            if status['status'] == 'healthy':
                lag_mb = status['lag_bytes'] / (1024 * 1024)
                if lag_mb > threshold_mb:
                    lag_alerts.append({
                        'standby': standby_name,
                        'lag_mb': lag_mb,
                        'threshold_mb': threshold_mb
                    })

        return lag_alerts

    def monitor_replication(self):
        """Monitor replication health."""
        # Check primary
        primary_status = self.check_primary_status()
        self.logger.info(f"Primary status: {primary_status['status']}")

        # Check standbys
        for standby_name, standby_config in self.standby_configs.items():
            standby_status = self.check_standby_status(standby_config)
            self.logger.info(f"Standby {standby_name} status: {standby_status['status']}")

            if standby_status['status'] == 'healthy':
                lag_mb = standby_status['lag_bytes'] / (1024 * 1024)
                self.logger.info(f"Standby {standby_name} lag: {lag_mb:.2f} MB")

        # Check for lag alerts
        lag_alerts = self.check_replication_lag()
        if lag_alerts:
            for alert in lag_alerts:
                self.logger.warning(f"High replication lag on {alert['standby']}: {alert['lag_mb']:.2f} MB")

        return {
            'primary': primary_status,
            'standbys': {name: self.check_standby_status(config) for name, config in self.standby_configs.items()},
            'lag_alerts': lag_alerts
        }

# Usage
if __name__ == "__main__":
    primary_config = {
        'host': '192.168.1.10',
        'database': 'production',
        'user': 'monitor_user',
        'password': 'monitor_password'
    }

    standby_configs = {
        'standby-1': {
            'host': '192.168.1.11',
            'database': 'production',
            'user': 'monitor_user',
            'password': 'monitor_password'
        },
        'standby-2': {
            'host': '192.168.1.12',
            'database': 'production',
            'user': 'monitor_user',
            'password': 'monitor_password'
        }
    }

    monitor = PostgreSQLReplicationMonitor(primary_config, standby_configs)

    while True:
        status = monitor.monitor_replication()
        time.sleep(30)

# failover/automated_failover.py
import psycopg2
import subprocess
import time
import logging
from datetime import datetime

class PostgreSQLFailoverManager:
    def __init__(self, config):
        self.config = config
        self.setup_logging()

    def setup_logging(self):
        """Setup logging."""
        logging.basicConfig(level=logging.INFO)
        self.logger = logging.getLogger(__name__)

    def check_primary_health(self):
        """Check if primary is healthy."""
        try:
            conn = psycopg2.connect(**self.config['primary'])
            with conn.cursor() as cur:
                cur.execute("SELECT 1")
                result = cur.fetchone()
                return result[0] == 1
        except Exception as e:
            self.logger.error(f"Primary health check failed: {e}")
            return False

    def promote_standby(self, standby_name):
        """Promote standby to primary."""
        try:
            # Create trigger file to promote standby
            trigger_file = f"/tmp/postgresql.trigger.{standby_name}"
            with open(trigger_file, 'w') as f:
                f.write("")

            self.logger.info(f"Trigger file created: {trigger_file}")

            # Wait for promotion to complete
            time.sleep(10)

            # Verify promotion
            standby_config = self.config['standbys'][standby_name]
            conn = psycopg2.connect(**standby_config)
            with conn.cursor() as cur:
                cur.execute("SELECT pg_is_in_recovery();")
                in_recovery = cur.fetchone()[0]

                if not in_recovery:
                    self.logger.info(f"Standby {standby_name} successfully promoted to primary")
                    return True
                else:
                    self.logger.error(f"Standby {standby_name} promotion failed")
                    return False

        except Exception as e:
            self.logger.error(f"Promotion failed: {e}")
            return False

    def update_load_balancer(self, new_primary):
        """Update load balancer configuration."""
        try:
            # Update HAProxy configuration
            haproxy_config = f"""
listen postgresql
    bind *:5432
    mode tcp
    balance roundrobin
    option tcp-check
    tcp-check connect port 5432

    server {new_primary} {self.config['standbys'][new_primary]['host']}:5432 check port 8008 inter 5s rise 3 fall 3
"""

            with open('/etc/haproxy/haproxy.cfg', 'w') as f:
                f.write(haproxy_config)

            # Reload HAProxy
            subprocess.run(['systemctl', 'reload', 'haproxy'], check=True)
            self.logger.info(f"Load balancer updated to use {new_primary}")

        except Exception as e:
            self.logger.error(f"Load balancer update failed: {e}")

    def execute_failover(self):
        """Execute automated failover."""
        self.logger.info("Starting automated failover process")

        # Check if primary is actually down
        if self.check_primary_health():
            self.logger.info("Primary is healthy, no failover needed")
            return False

        # Select best standby for promotion
        best_standby = self.select_best_standby()
        if not best_standby:
            self.logger.error("No healthy standby available for failover")
            return False

        # Promote standby
        if self.promote_standby(best_standby):
            # Update load balancer
            self.update_load_balancer(best_standby)

            # Send notifications
            self.send_failover_notification(best_standby)

            self.logger.info(f"Failover completed successfully to {best_standby}")
            return True
        else:
            self.logger.error("Failover failed")
            return False

    def select_best_standby(self):
        """Select the best standby for promotion."""
        best_standby = None
        min_lag = float('inf')

        for standby_name, standby_config in self.config['standbys'].items():
            try:
                conn = psycopg2.connect(**standby_config)
                with conn.cursor() as cur:
                    # Check if standby is healthy
                    cur.execute("SELECT pg_is_in_recovery();")
                    in_recovery = cur.fetchone()[0]

                    if in_recovery:
                        # Check replication lag
                        cur.execute("""
                            SELECT pg_wal_lsn_diff(pg_last_wal_receive_lsn(), pg_last_wal_replay_lsn())
                        """)
                        lag = cur.fetchone()[0]

                        if lag < min_lag:
                            min_lag = lag
                            best_standby = standby_name

            except Exception as e:
                self.logger.error(f"Error checking standby {standby_name}: {e}")
                continue

        return best_standby

    def send_failover_notification(self, new_primary):
        """Send failover notification."""
        message = f"""
PostgreSQL Failover Alert

Time: {datetime.now().isoformat()}
Action: Automated failover executed
New Primary: {new_primary}
Previous Primary: {self.config['primary']['host']}

Please verify the new primary is functioning correctly.
"""

        # Send email notification
        # Implementation depends on your notification system
        self.logger.info(f"Failover notification sent: {message}")

# Usage
if __name__ == "__main__":
    config = {
        'primary': {
            'host': '192.168.1.10',
            'database': 'production',
            'user': 'monitor_user',
            'password': 'monitor_password'
        },
        'standbys': {
            'standby-1': {
                'host': '192.168.1.11',
                'database': 'production',
                'user': 'monitor_user',
                'password': 'monitor_password'
            },
            'standby-2': {
                'host': '192.168.1.12',
                'database': 'production',
                'user': 'monitor_user',
                'password': 'monitor_password'
            }
        }
    }

    failover_manager = PostgreSQLFailoverManager(config)

    # Check primary health every 30 seconds
    while True:
        if not failover_manager.check_primary_health():
            failover_manager.execute_failover()
        time.sleep(30)

# Complete PostgreSQL replication and HA setup
def setup_postgresql_replication_ha():
    # 1. Streaming replication
    # 2. Logical replication
    # 3. High availability with Patroni
    # 4. Load balancing
    # 5. Monitoring and health checks
    # 6. Automated failover
    # 7. Disaster recovery
    # 8. Testing procedures

    print("PostgreSQL replication and HA setup complete!")