Skip to main content

​
Regional Failover Testing

Test your application’s regional failover and disaster recovery capabilities across different geographic locations. This locustfile validates automatic failover, backup region activation, and service continuity during regional outages.

​
Use Cases

  • Disaster Recovery: Test automatic failover during regional outages
  • Load Balancing: Validate traffic distribution across regions
  • Health Monitoring: Test regional health checks and monitoring
  • Service Continuity: Ensure uninterrupted service during failover
  • Data Replication: Validate cross-region data synchronization

​
Locustfile

from locust import HttpUser, task, between
import random
import time
import json

class RegionalFailoverUser(HttpUser):
    """Test regional failover and disaster recovery"""
    wait_time = between(2, 4)
    
    def on_start(self):
        """Initialize regional failover testing"""
        self.regions = {
            'primary': {
                'name': 'us-east-1',
                'endpoint': '/api/v1',
                'health_check': '/health',
                'priority': 1
            },
            'secondary': {
                'name': 'us-west-2',
                'endpoint': '/api/v2',
                'health_check': '/health',
                'priority': 2
            },
            'tertiary': {
                'name': 'eu-west-1',
                'endpoint': '/api/v3',
                'health_check': '/health',
                'priority': 3
            }
        }
        
        self.current_region = 'primary'
        self.failover_attempts = 0
        self.max_failover_attempts = 3
        
        print(f"Regional failover testing initialized - starting with {self.current_region}")
    
    @task(4)
    def test_primary_region_health(self):
        """Test primary region health and availability"""
        region_config = self.regions[self.current_region]
        health_endpoint = f"{region_config['endpoint']}{region_config['health_check']}"
        
        with self.client.get(
            health_endpoint,
            name=f"Health Check - {region_config['name']}",
            catch_response=True
        ) as response:
            if response.status_code == 200:
                try:
                    health_data = response.json()
                    
                    if health_data.get('status') == 'healthy':
                        print(f"Region {region_config['name']} is healthy")
                        response.success()
                    elif health_data.get('status') == 'degraded':
                        print(f"Region {region_config['name']} is degraded - may trigger failover")
                        response.success()
                        self.consider_failover()
                    else:
                        print(f"Region {region_config['name']} is unhealthy")
                        response.failure("Region unhealthy")
                        self.trigger_failover()
                        
                except json.JSONDecodeError:
                    response.failure("Invalid health check response")
                    self.trigger_failover()
            else:
                print(f"Health check failed for {region_config['name']}: {response.status_code}")
                response.failure(f"Health check failed: {response.status_code}")
                self.trigger_failover()
    
    @task(3)
    def test_api_functionality(self):
        """Test core API functionality in current region"""
        region_config = self.regions[self.current_region]
        api_endpoint = f"{region_config['endpoint']}/data"
        
        with self.client.get(
            api_endpoint,
            name=f"API Test - {region_config['name']}",
            catch_response=True
        ) as response:
            if response.status_code == 200:
                try:
                    data = response.json()
                    
                    # Validate response includes region information
                    if 'region' in data:
                        response_region = data['region']
                        if response_region == region_config['name']:
                            print(f"API response from correct region: {response_region}")
                            response.success()
                        else:
                            print(f"API response from unexpected region: {response_region}")
                            response.success()  # Still functional, just different region
                    else:
                        response.success()
                        
                except json.JSONDecodeError:
                    response.failure("Invalid API response")
            else:
                print(f"API request failed in {region_config['name']}: {response.status_code}")
                response.failure(f"API failed: {response.status_code}")
                self.trigger_failover()
    
    @task(2)
    def test_data_consistency(self):
        """Test data consistency across regions"""
        region_config = self.regions[self.current_region]
        
        # Write data to current region
        test_data = {
            'id': f"test_{int(time.time())}",
            'region': region_config['name'],
            'timestamp': time.time()
        }
        
        with self.client.post(
            f"{region_config['endpoint']}/data",
            json=test_data,
            name=f"Data Write - {region_config['name']}",
            catch_response=True
        ) as write_response:
            if write_response.status_code in [200, 201]:
                # Wait briefly for replication
                time.sleep(1)
                
                # Read data back to verify consistency
                with self.client.get(
                    f"{region_config['endpoint']}/data/{test_data['id']}",
                    name=f"Data Read - {region_config['name']}",
                    catch_response=True
                ) as read_response:
                    if read_response.status_code == 200:
                        try:
                            read_data = read_response.json()
                            if read_data.get('id') == test_data['id']:
                                print(f"Data consistency verified in {region_config['name']}")
                                write_response.success()
                            else:
                                write_response.failure("Data consistency check failed")
                        except json.JSONDecodeError:
                            write_response.failure("Invalid data read response")
                    else:
                        write_response.failure("Data read failed after write")
            else:
                write_response.failure(f"Data write failed: {write_response.status_code}")
    
    @task(1)
    def test_cross_region_replication(self):
        """Test cross-region data replication"""
        if self.current_region == 'primary':
            # Test replication to secondary region
            secondary_config = self.regions['secondary']
            
            with self.client.get(
                f"{secondary_config['endpoint']}/replication/status",
                name=f"Replication Status - {secondary_config['name']}",
                catch_response=True
            ) as response:
                if response.status_code == 200:
                    try:
                        replication_data = response.json()
                        
                        lag_seconds = replication_data.get('lag_seconds', 0)
                        if lag_seconds < 60:  # Less than 1 minute lag
                            print(f"Replication lag acceptable: {lag_seconds}s")
                            response.success()
                        else:
                            print(f"High replication lag: {lag_seconds}s")
                            response.failure("High replication lag")
                            
                    except json.JSONDecodeError:
                        response.failure("Invalid replication status response")
                else:
                    response.failure(f"Replication status check failed: {response.status_code}")
    
    def consider_failover(self):
        """Consider failover based on degraded performance"""
        # Implement logic to decide if failover should be triggered
        # This could be based on response times, error rates, etc.
        if random.random() < 0.1:  # 10% chance to trigger failover on degraded status
            self.trigger_failover()
    
    def trigger_failover(self):
        """Trigger failover to next available region"""
        if self.failover_attempts >= self.max_failover_attempts:
            print("Maximum failover attempts reached")
            return
        
        current_priority = self.regions[self.current_region]['priority']
        
        # Find next available region
        next_region = None
        for region_key, region_config in self.regions.items():
            if region_config['priority'] > current_priority:
                if next_region is None or region_config['priority'] < self.regions[next_region]['priority']:
                    next_region = region_key
        
        if next_region:
            old_region = self.current_region
            self.current_region = next_region
            self.failover_attempts += 1
            
            print(f"FAILOVER: Switching from {self.regions[old_region]['name']} to {self.regions[next_region]['name']}")
            
            # Test new region immediately
            self.test_failover_success()
        else:
            print("No available regions for failover")
    
    def test_failover_success(self):
        """Test that failover was successful"""
        region_config = self.regions[self.current_region]
        
        with self.client.get(
            f"{region_config['endpoint']}{region_config['health_check']}",
            name=f"Failover Verification - {region_config['name']}",
            catch_response=True
        ) as response:
            if response.status_code == 200:
                try:
                    health_data = response.json()
                    if health_data.get('status') == 'healthy':
                        print(f"Failover successful - {region_config['name']} is healthy")
                    else:
                        print(f"Failover target {region_config['name']} is not healthy")
                        self.trigger_failover()  # Try next region
                except json.JSONDecodeError:
                    print(f"Invalid health response from failover target {region_config['name']}")
                    self.trigger_failover()
            else:
                print(f"Failover target {region_config['name']} is not responding")
                self.trigger_failover()


class LoadBalancingUser(HttpUser):
    """Test load balancing across multiple regions"""
    wait_time = between(1, 2)
    
    def on_start(self):
        """Initialize load balancing testing"""
        self.regions = ['us-east-1', 'us-west-2', 'eu-west-1']
        self.request_counts = {region: 0 for region in self.regions}
        
    @task(5)
    def test_load_distribution(self):
        """Test that load is distributed across regions"""
        with self.client.get(
            '/api/v1/load-balanced',
            name="Load Balanced Request",
            catch_response=True
        ) as response:
            if response.status_code == 200:
                try:
                    data = response.json()
                    serving_region = data.get('region', 'unknown')
                    
                    if serving_region in self.request_counts:
                        self.request_counts[serving_region] += 1
                        print(f"Request served by {serving_region}")
                        response.success()
                    else:
                        print(f"Request served by unknown region: {serving_region}")
                        response.success()
                        
                except json.JSONDecodeError:
                    response.failure("Invalid load balanced response")
            else:
                response.failure(f"Load balanced request failed: {response.status_code}")
    
    def on_stop(self):
        """Report load distribution statistics"""
        total_requests = sum(self.request_counts.values())
        if total_requests > 0:
            print("Load Distribution Summary:")
            for region, count in self.request_counts.items():
                percentage = (count / total_requests) * 100
                print(f"  {region}: {count} requests ({percentage:.1f}%)")

​
Configuration

Update these settings for your regional failover testing: 
# Regional Configuration
REGIONS = {
    'primary': {'name': 'us-east-1', 'priority': 1},
    'secondary': {'name': 'us-west-2', 'priority': 2},
    'tertiary': {'name': 'eu-west-1', 'priority': 3}
}

# Failover Settings
MAX_FAILOVER_ATTEMPTS = 3
HEALTH_CHECK_INTERVAL = 30  # seconds
REPLICATION_LAG_THRESHOLD = 60  # seconds

​
LoadForge Multi-Region Failover Testing

​
Testing Strategy

Use LoadForge’s region selection to simulate realistic failover scenarios:
  1. Primary Region Testing: Run from primary region (e.g., NYC) to test normal operations
  2. Failover Simulation: Simulate primary region failure and test secondary activation
  3. Cross-Region Testing: Test from multiple regions simultaneously to validate load balancing
  4. Recovery Testing: Test failback to primary region after recovery

​
Key Metrics to Monitor

  • Failover Time: Time taken to switch to backup region
  • Service Availability: Uptime during failover events
  • Data Consistency: Cross-region data synchronization
  • Load Distribution: Traffic distribution across healthy regions
  • Recovery Time: Time to restore primary region service

​
Best Practices

  • Health Check Frequency: Regular monitoring of regional health
  • Automated Failover: Implement automatic failover triggers
  • Data Replication: Ensure real-time cross-region data sync
  • Graceful Degradation: Maintain partial functionality during outages
  • Failback Planning: Plan for returning to primary region

​
Troubleshooting

  • Slow Failover: Check health check frequency and failover triggers
  • Data Inconsistency: Review replication lag and synchronization
  • Load Imbalance: Verify load balancing algorithms and region weights
  • Failed Recovery: Check primary region restoration procedures
This locustfile provides comprehensive regional failover testing, ensuring your application maintains high availability across geographic regions and handles disaster recovery scenarios effectively.
⌘I