Multi-Region Architecture: Building Globally Distributed Systems

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

-- Primary server configuration (postgresql.conf)
-- Enable WAL shipping for replication
 
-- Set replication parameters
wal_level = replica                    -- Enable replication-level logging
max_wal_senders = 10                   -- Max concurrent replication connections
wal_keep_size = 4GB                    -- Retain WAL for replicas that fall behind
synchronous_commit = local             -- Don't wait for remote confirmation (async)
archive_mode = on                      -- Enable WAL archiving for PITR
archive_command = 'aws s3 cp %p s3://wal-archive/%f'  -- Archive to S3
 
-- Network configuration for replication (pg_hba.conf)
-- Allow replication connections from secondary region
host    replication    replicator    10.1.0.0/16       scram-sha-256
host    replication    replicator    10.2.0.0/16       scram-sha-256
 
-- Create replication slot for reliable streaming
SELECT pg_create_physical_replication_slot('secondary_region_slot');
 
-- Secondary server initialization
-- Run on secondary to create replica from primary
pg_basebackup -h primary.us-east.internal \
    -D /var/lib/postgresql/data \
    -U replicator \
    -v -P \
    --wal-method=stream \
    --slot=secondary_region_slot
 
-- Secondary server configuration (postgresql.conf)
hot_standby = on                       -- Allow read queries on replica
hot_standby_feedback = on              -- Inform primary of replica queries
primary_conninfo = 'host=primary.us-east.internal port=5432 user=replicator application_name=us_west_replica'
primary_slot_name = 'secondary_region_slot'

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173

"""
Replication Lag Monitoring for Active-Passive Multi-Region
 
This module monitors database replication lag and triggers alerts
when lag exceeds acceptable thresholds. Critical for understanding
actual RPO and failover readiness.
"""
import psycopg2
from datetime import datetime, timedelta
from dataclasses import dataclass
from typing import Optional
import logging
 
@dataclass
class ReplicationStatus:
    """Status of replication between primary and replica."""
    replica_name: str
    replication_lag_bytes: int
    replication_lag_seconds: float
    replay_lag_seconds: float
    state: str  # 'streaming', 'catchup', 'startup'
    is_healthy: bool
    last_updated: datetime
 
class ReplicationMonitor:
    """Monitor replication health between primary and secondary regions."""
    
    # Alert thresholds
    LAG_WARNING_SECONDS = 5.0
    LAG_CRITICAL_SECONDS = 30.0
    LAG_EMERGENCY_SECONDS = 300.0
    
    def __init__(self, primary_dsn: str, replica_dsn: str):
        self.primary_dsn = primary_dsn
        self.replica_dsn = replica_dsn
        self.logger = logging.getLogger(__name__)
    
    def get_primary_replication_status(self) -> list[ReplicationStatus]:
        """
        Query primary for status of all replicas.
        Uses pg_stat_replication to get streaming replication info.
        """
        query = """
        SELECT
            application_name,
            state,
            sent_lsn - write_lsn AS pending_bytes,
            COALESCE(
                EXTRACT(EPOCH FROM now() - write_lag),
                0
            ) AS write_lag_seconds,
            COALESCE(
                EXTRACT(EPOCH FROM now() - replay_lag),
                0
            ) AS replay_lag_seconds
        FROM pg_stat_replication
        WHERE state = 'streaming';
        """
        
        statuses = []
        with psycopg2.connect(self.primary_dsn) as conn:
            with conn.cursor() as cur:
                cur.execute(query)
                for row in cur.fetchall():
                    is_healthy = (
                        row[3] < self.LAG_CRITICAL_SECONDS and
                        row[4] < self.LAG_CRITICAL_SECONDS
                    )
                    statuses.append(ReplicationStatus(
                        replica_name=row[0],
                        replication_lag_bytes=row[2] or 0,
                        replication_lag_seconds=row[3],
                        replay_lag_seconds=row[4],
                        state=row[1],
                        is_healthy=is_healthy,
                        last_updated=datetime.now()
                    ))
        return statuses
    
    def get_replica_recovery_status(self) -> dict:
        """
        Query replica for its view of replication status.
        Provides ground-truth from replica perspective.
        """
        query = """
        SELECT
            pg_is_in_recovery() AS is_replica,
            pg_last_wal_receive_lsn() AS last_received,
            pg_last_wal_replay_lsn() AS last_replayed,
            pg_last_xact_replay_timestamp() AS last_replay_time,
            EXTRACT(EPOCH FROM now() - pg_last_xact_replay_timestamp()) 
                AS replay_lag_seconds
        """
        
        with psycopg2.connect(self.replica_dsn) as conn:
            with conn.cursor() as cur:
                cur.execute(query)
                row = cur.fetchone()
                return {
                    "is_replica": row[0],
                    "last_received_lsn": row[1],
                    "last_replayed_lsn": row[2],
                    "last_replay_time": row[3],
                    "replay_lag_seconds": row[4] or 0
                }
    
    def evaluate_failover_readiness(self) -> dict:
        """
        Evaluate whether the system is ready for a safe failover.
        Returns readiness status and estimated data loss.
        """
        primary_status = self.get_primary_replication_status()
        replica_status = self.get_replica_recovery_status()
        
        # Find the cross-region replica
        cross_region_replica = next(
            (s for s in primary_status if 'us_west' in s.replica_name),
            None
        )
        
        if not cross_region_replica:
            return {
                "ready": False,
                "reason": "No cross-region replica found",
                "estimated_data_loss_seconds": None
            }
        
        lag_seconds = max(
            cross_region_replica.replication_lag_seconds,
            cross_region_replica.replay_lag_seconds
        )
        
        # Determine readiness level
        if lag_seconds > self.LAG_EMERGENCY_SECONDS:
            readiness = "NOT_READY"
            reason = f"Replication lag ({lag_seconds:.1f}s) exceeds emergency threshold"
        elif lag_seconds > self.LAG_CRITICAL_SECONDS:
            readiness = "DEGRADED"
            reason = f"Elevated replication lag ({lag_seconds:.1f}s)"
        elif lag_seconds > self.LAG_WARNING_SECONDS:
            readiness = "READY_WITH_WARNINGS"
            reason = f"Minor replication lag ({lag_seconds:.1f}s)"
        else:
            readiness = "FULLY_READY"
            reason = "Replication healthy"
        
        return {
            "ready": readiness in ["FULLY_READY", "READY_WITH_WARNINGS"],
            "readiness_level": readiness,
            "reason": reason,
            "estimated_data_loss_seconds": lag_seconds,
            "replica_state": cross_region_replica.state,
            "replica_healthy": cross_region_replica.is_healthy
        }
    
    def emit_metrics(self) -> None:
        """Emit replication metrics for monitoring systems."""
        statuses = self.get_primary_replication_status()
        
        for status in statuses:
            # Emit to your metrics system (Prometheus, CloudWatch, etc.)
            self.logger.info(
                f"replication_lag_seconds{{replica="{status.replica_name}"}} "
                f"{status.replication_lag_seconds}"
            )
            self.logger.info(
                f"replay_lag_seconds{{replica="{status.replica_name}"}} "
                f"{status.replay_lag_seconds}"
            )
            self.logger.info(
                f"replication_healthy{{replica="{status.replica_name}"}} "
                f"{1 if status.is_healthy else 0}"
            )

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306

"""
Failover Orchestrator for Active-Passive Multi-Region
 
This module coordinates the complex sequence of operations required
for a controlled failover from primary to secondary region.
"""
import asyncio
from dataclasses import dataclass
from datetime import datetime
from enum import Enum
from typing import Optional
import logging
 
class FailoverType(Enum):
    PLANNED = "planned"      # Graceful, with full preparation
    EMERGENCY = "emergency"  # Immediate, potentially with data loss
 
class FailoverPhase(Enum):
    INITIALIZED = "initialized"
    PRIMARY_ISOLATED = "primary_isolated"
    REPLICATION_FINALIZED = "replication_finalized"
    DATABASE_PROMOTED = "database_promoted"
    APPLICATIONS_CONFIGURED = "applications_configured"
    TRAFFIC_REROUTED = "traffic_rerouted"
    CAPACITY_SCALED = "capacity_scaled"
    VERIFIED = "verified"
    COMPLETED = "completed"
    FAILED = "failed"
    ROLLED_BACK = "rolled_back"
 
@dataclass
class FailoverState:
    """Tracks the current state of a failover operation."""
    failover_id: str
    failover_type: FailoverType
    initiated_at: datetime
    initiated_by: str
    current_phase: FailoverPhase
    source_region: str
    target_region: str
    estimated_data_loss_seconds: float
    notes: list[str]
    
class FailoverOrchestrator:
    """
    Coordinates multi-region failover operations.
    
    Implements the complete failover sequence with proper
    error handling, rollback capabilities, and audit logging.
    """
    
    # Timeout for each phase (seconds)
    PHASE_TIMEOUTS = {
        FailoverPhase.PRIMARY_ISOLATED: 60,
        FailoverPhase.REPLICATION_FINALIZED: 300,
        FailoverPhase.DATABASE_PROMOTED: 120,
        FailoverPhase.APPLICATIONS_CONFIGURED: 60,
        FailoverPhase.TRAFFIC_REROUTED: 30,
        FailoverPhase.CAPACITY_SCALED: 300,
        FailoverPhase.VERIFIED: 120,
    }
    
    def __init__(
        self,
        database_manager,      # Handles DB promotion
        load_balancer_manager, # Handles traffic routing
        infrastructure_manager, # Handles scaling
        monitoring_client,     # For health verification
        notification_service   # For stakeholder alerts
    ):
        self.db = database_manager
        self.lb = load_balancer_manager
        self.infra = infrastructure_manager
        self.monitoring = monitoring_client
        self.notifications = notification_service
        self.logger = logging.getLogger(__name__)
        self.state: Optional[FailoverState] = None
    
    async def execute_failover(
        self,
        source_region: str,
        target_region: str,
        failover_type: FailoverType,
        operator: str,
        skip_confirmation: bool = False
    ) -> FailoverState:
        """
        Execute a complete failover from source to target region.
        
        Args:
            source_region: The current primary region
            target_region: The region to promote
            failover_type: PLANNED or EMERGENCY
            operator: ID of person/system initiating failover
            skip_confirmation: For automation, skip human confirmation
        """
        # Initialize failover state
        self.state = FailoverState(
            failover_id=self._generate_failover_id(),
            failover_type=failover_type,
            initiated_at=datetime.now(),
            initiated_by=operator,
            current_phase=FailoverPhase.INITIALIZED,
            source_region=source_region,
            target_region=target_region,
            estimated_data_loss_seconds=0,
            notes=[]
        )
        
        self.logger.warning(
            f"FAILOVER INITIATED: {self.state.failover_id} "
            f"({failover_type.value}) {source_region} -> {target_region}"
        )
        
        try:
            # Phase 1: Pre-flight checks
            await self._preflight_checks()
            
            # Phase 2: Isolate primary (if reachable and planned)
            if failover_type == FailoverType.PLANNED:
                await self._isolate_primary()
            else:
                self._log("Skipping primary isolation (emergency failover)")
            
            # Phase 3: Finalize replication (wait for sync if possible)
            await self._finalize_replication(failover_type)
            
            # Phase 4: Promote database
            await self._promote_database()
            
            # Point of no return checkpoint
            if not skip_confirmation and failover_type == FailoverType.PLANNED:
                self._log("DATABASE PROMOTED - Awaiting confirmation to continue")
                # In real implementation, wait for human confirmation
            
            # Phase 5: Configure applications
            await self._configure_applications()
            
            # Phase 6: Reroute traffic
            await self._reroute_traffic()
            
            # Phase 7: Scale capacity
            await self._scale_capacity()
            
            # Phase 8: Verify health
            await self._verify_health()
            
            # Complete
            self.state.current_phase = FailoverPhase.COMPLETED
            self._log("FAILOVER COMPLETED SUCCESSFULLY")
            
            # Notify stakeholders
            await self.notifications.send_failover_complete(self.state)
            
        except Exception as e:
            self.state.current_phase = FailoverPhase.FAILED
            self.state.notes.append(f"FAILURE: {str(e)}")
            self.logger.error(f"Failover failed: {e}")
            await self.notifications.send_failover_failed(self.state, str(e))
            raise
        
        return self.state
    
    async def _preflight_checks(self) -> None:
        """Verify system is ready for failover."""
        self._log("Running pre-flight checks...")
        
        # Check target region health
        target_health = await self.monitoring.check_region_health(
            self.state.target_region
        )
        if not target_health.is_healthy:
            raise FailoverError(
                f"Target region {self.state.target_region} is not healthy"
            )
        
        # Check replication status
        replication = await self.db.get_replication_status(
            self.state.source_region,
            self.state.target_region
        )
        self.state.estimated_data_loss_seconds = replication.lag_seconds
        
        if replication.lag_seconds > 300:
            self._log(
                f"WARNING: High replication lag ({replication.lag_seconds}s). "
                f"Estimated data loss: {replication.lag_seconds} seconds"
            )
        
        self._log(f"Pre-flight checks passed. Lag: {replication.lag_seconds}s")
    
    async def _isolate_primary(self) -> None:
        """Stop accepting new connections at primary."""
        self._log("Isolating primary region...")
        self.state.current_phase = FailoverPhase.PRIMARY_ISOLATED
        
        # Disable load balancer backends in primary
        await self.lb.disable_backends(self.state.source_region)
        
        # Wait for connection draining
        await asyncio.sleep(10)  # Allow existing requests to complete
        
        self._log("Primary region isolated")
    
    async def _finalize_replication(self, failover_type: FailoverType) -> None:
        """Wait for replication to catch up if possible."""
        self._log("Finalizing replication...")
        self.state.current_phase = FailoverPhase.REPLICATION_FINALIZED
        
        if failover_type == FailoverType.PLANNED:
            # Wait for full sync (with timeout)
            timeout = self.PHASE_TIMEOUTS[FailoverPhase.REPLICATION_FINALIZED]
            await self.db.wait_for_replication_sync(
                self.state.source_region,
                self.state.target_region,
                timeout_seconds=timeout
            )
            self.state.estimated_data_loss_seconds = 0
            self._log("Replication fully synchronized (zero data loss)")
        else:
            # Emergency: accept current lag state
            current_lag = await self.db.get_replication_lag_seconds(
                self.state.source_region,
                self.state.target_region
            )
            self.state.estimated_data_loss_seconds = current_lag
            self._log(f"Proceeding with {current_lag}s replication lag")
    
    async def _promote_database(self) -> None:
        """Promote replica to primary status."""
        self._log("PROMOTING DATABASE - Point of no return")
        self.state.current_phase = FailoverPhase.DATABASE_PROMOTED
        
        await self.db.promote_replica(
            self.state.target_region,
            new_role="primary"
        )
        
        # Verify promotion succeeded
        is_primary = await self.db.verify_is_primary(self.state.target_region)
        if not is_primary:
            raise FailoverError("Database promotion verification failed")
        
        self._log("Database promoted successfully")
    
    async def _configure_applications(self) -> None:
        """Update application configurations for new primary."""
        self._log("Configuring applications...")
        self.state.current_phase = FailoverPhase.APPLICATIONS_CONFIGURED
        
        # Update connection strings, feature flags, etc.
        await self.infra.update_application_configs(
            region=self.state.target_region,
            database_endpoint=await self.db.get_primary_endpoint(),
            enable_background_jobs=True
        )
        
        self._log("Applications configured")
    
    async def _reroute_traffic(self) -> None:
        """Update global load balancer to route to new primary."""
        self._log("Rerouting traffic...")
        self.state.current_phase = FailoverPhase.TRAFFIC_REROUTED
        
        await self.lb.set_primary_region(self.state.target_region)
        
        self._log("Traffic rerouted to new primary")
    
    async def _scale_capacity(self) -> None:
        """Scale up target region to handle full load."""
        self._log("Scaling capacity...")
        self.state.current_phase = FailoverPhase.CAPACITY_SCALED
        
        await self.infra.scale_to_production(self.state.target_region)
        
        self._log("Capacity scaled")
    
    async def _verify_health(self) -> None:
        """Verify new primary is healthy and serving traffic."""
        self._log("Verifying health...")
        self.state.current_phase = FailoverPhase.VERIFIED
        
        # Wait for health checks to stabilize
        await asyncio.sleep(30)
        
        health = await self.monitoring.comprehensive_health_check(
            self.state.target_region
        )
        
        if not health.is_healthy:
            raise FailoverError(f"Health verification failed: {health.issues}")
        
        self._log(f"Health verified: {health.summary}")
    
    def _log(self, message: str) -> None:
        """Log and record message in state."""
        self.logger.info(f"[{self.state.failover_id}] {message}")
        self.state.notes.append(f"{datetime.now().isoformat()}: {message}")
    
    def _generate_failover_id(self) -> str:
        """Generate unique failover identifier."""
        return f"FO-{datetime.now().strftime('%Y%m%d-%H%M%S')}"
 
class FailoverError(Exception):
    """Exception raised during failover operations."""
    pass

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113

#!/bin/bash
# Failback Procedure: US-West (current primary) -> US-East (original primary)
# 
# Prerequisites:
# - Original primary (US-East) is recovered and healthy
# - No active incidents
# - Change approval obtained
# - Maintenance window scheduled
 
set -euo pipefail
 
CURRENT_PRIMARY="us-west"
TARGET_REGION="us-east"
LOGFILE="/var/log/failback-$(date +%Y%m%d-%H%M%S).log"
 
log() {
    echo "[$(date '+%Y-%m-%d %H:%M:%S')] $1" | tee -a "$LOGFILE"
}
 
checkpoint() {
    log "CHECKPOINT: $1"
    read -p "Press enter to continue or Ctrl+C to abort..."
}
 
# 1. Pre-flight verification
log "=== Starting failback pre-flight checks ==="
 
log "Checking target region (US-East) health..."
if ! health-check.sh "$TARGET_REGION"; then
    log "ERROR: Target region is not healthy. Aborting."
    exit 1
fi
 
log "Checking current primary (US-West) health..."
if ! health-check.sh "$CURRENT_PRIMARY"; then
    log "ERROR: Current primary is not healthy. Aborting."
    exit 1
fi
 
log "Checking replication status..."
REPLICATION_STATUS=$(psql -h db-primary.us-west.internal -c \
    "SELECT state, replay_lag FROM pg_stat_replication WHERE application_name = 'us_east_replica';")
log "Replication status: $REPLICATION_STATUS"
 
checkpoint "Pre-flight checks complete. Ready to proceed?"
 
# 2. Configure US-East database as replica of US-West
log "=== Configuring US-East as replica ==="
 
log "Stopping application servers in US-East to prevent stale data access..."
kubectl --context=us-east scale deployment/api-service --replicas=0
 
log "Reconfiguring US-East database as replica..."
# This requires the database to re-sync from US-West
# For PostgreSQL, this typically involves pg_basebackup or pg_rewind
 
ssh db-admin@db1.us-east.internal << 'EOF'
    sudo systemctl stop postgresql
    
    # Use pg_rewind if WAL is available, otherwise full resync
    if pg_rewind --target-pgdata=/var/lib/postgresql/data                  --source-server='host=db-primary.us-west.internal port=5432'; then
        echo "pg_rewind successful - incremental sync"
    else
        echo "pg_rewind failed - performing full resync"
        rm -rf /var/lib/postgresql/data/*
        pg_basebackup -h db-primary.us-west.internal -D /var/lib/postgresql/data             -U replicator -v -P --wal-method=stream
    fi
    
    # Configure as standby
    touch /var/lib/postgresql/data/standby.signal
    
    cat >> /var/lib/postgresql/data/postgresql.auto.conf << CONF
primary_conninfo = 'host=db-primary.us-west.internal port=5432 user=replicator'
CONF
    
    sudo systemctl start postgresql
EOF
 
log "Waiting for US-East replica to sync..."
until psql -h db1.us-east.internal -c "SELECT pg_is_in_recovery();" | grep -q 't'; do
    sleep 5
    log "Waiting for replica to start..."
done
 
checkpoint "US-East database synchronized as replica. Ready for traffic migration?"
 
# 3. Migrate traffic gradually
log "=== Beginning gradual traffic migration ==="
 
log "Setting traffic split: 90% US-West, 10% US-East..."
gcloud compute url-maps update global-lb     --backend-service-weight=us-west-backend=90,us-east-backend=10
 
log "Monitoring for 10 minutes..."
sleep 600
 
# Check error rates
ERROR_RATE=$(prometheus-query "sum(rate(http_errors_total{region='us-east'}[10m])) / sum(rate(http_requests_total{region='us-east'}[10m]))")
if (( $(echo "$ERROR_RATE > 0.01" | bc -l) )); then
    log "ERROR: High error rate in US-East ($ERROR_RATE). Aborting."
    gcloud compute url-maps update global-lb         --backend-service-weight=us-west-backend=100,us-east-backend=0
    exit 1
fi
 
log "Error rate acceptable ($ERROR_RATE). Increasing to 50%..."
gcloud compute url-maps update global-lb     --backend-service-weight=us-west-backend=50,us-east-backend=50
 
checkpoint "Traffic at 50/50. Monitor for issues before continuing."
 
log "Finalizing: 100% to US-East..."
# The rest of the failover procedure continues...
 
log "=== Failback complete. US-East is now primary. ===