AWS Disaster Recovery: Backup, Pilot Light, and Multi-Region Strategies

#!/bin/bash
# Simulate RTO/RPO check for a critical workload
RTO_TARGET=3600  # 1 hour in seconds
RPO_TARGET=900   # 15 minutes in seconds

# Get latest snapshot timestamp from RDS
LATEST_SNAPSHOT=$(aws rds describe-db-snapshots \
  --db-instance-identifier prod-db \
  --query 'DBSnapshots[-1].SnapshotCreateTime' \
  --output text)

CURRENT_TIME=$(date -u +%s)
SNAPSHOT_TIME=$(date -u -d "$LATEST_SNAPSHOT" +%s)
RPO=$((CURRENT_TIME - SNAPSHOT_TIME))

if [ $RPO -le $RPO_TARGET ]; then
  echo "RPO OK: $RPO seconds (target <= $RPO_TARGET)"
else
  echo "RPO VIOLATION: $RPO seconds (target <= $RPO_TARGET)"
fi

# Simulate RTO: time to restore from snapshot (placeholder)
RTO=1800  # assume 30 minutes
if [ $RTO -le $RTO_TARGET ]; then
  echo "RTO OK: $RTO seconds (target <= $RTO_TARGET)"
else
  echo "RTO VIOLATION: $RTO seconds (target <= $RTO_TARGET)"
fi

import boto3
import time

ebs = boto3.client('ec2', region_name='us-east-1')
rds = boto3.client('rds', region_name='us-east-1')

def create_ebs_snapshot(volume_id):
    response = ebs.create_snapshot(
        VolumeId=volume_id,
        Description='Production backup snapshot',
        TagSpecifications=[{
            'ResourceType': 'snapshot',
            'Tags': [{'Key': 'Name', 'Value': 'prod-backup'}]
        }]
    )
    snapshot_id = response['SnapshotId']
    print(f"Created snapshot {snapshot_id}")
    return snapshot_id

def restore_ebs_volume(snapshot_id, availability_zone):
    response = ebs.create_volume(
        SnapshotId=snapshot_id,
        AvailabilityZone=availability_zone,
        VolumeType='gp3'
    )
    volume_id = response['VolumeId']
    print(f"Restored volume {volume_id} from snapshot {snapshot_id}")
    return volume_id

def create_rds_snapshot(db_instance_id):
    response = rds.create_db_snapshot(
        DBSnapshotIdentifier=f"{db_instance_id}-{int(time.time())}",
        DBInstanceIdentifier=db_instance_id
    )
    snapshot_arn = response['DBSnapshot']['DBSnapshotArn']
    print(f"Created RDS snapshot {snapshot_arn}")
    return snapshot_arn

def restore_rds_instance(snapshot_id, new_instance_id):
    response = rds.restore_db_instance_from_db_snapshot(
        DBInstanceIdentifier=new_instance_id,
        DBSnapshotIdentifier=snapshot_id
    )
    print(f"Restoring RDS instance {new_instance_id} from snapshot {snapshot_id}")
    return response['DBInstance']['DBInstanceArn']

if __name__ == '__main__':
    # Example usage
    vol_id = 'vol-0abcd1234'
    snap_id = create_ebs_snapshot(vol_id)
    restore_ebs_volume(snap_id, 'us-east-1a')
    
    db_id = 'prod-db'
    rds_snap = create_rds_snapshot(db_id)
    restore_rds_instance(rds_snap, 'prod-db-restored')

import boto3
import time

rds = boto3.client('rds', region_name='us-west-2')
route53 = boto3.client('route53')

def promote_rds_read_replica(replica_id):
    response = rds.promote_read_replica(
        DBInstanceIdentifier=replica_id,
        BackupRetentionPeriod=7
    )
    print(f"Promoted replica {replica_id} to primary")
    return response['DBInstance']['DBInstanceArn']

def update_dns_record(hosted_zone_id, record_name, new_endpoint):
    response = route53.change_resource_record_sets(
        HostedZoneId=hosted_zone_id,
        ChangeBatch={
            'Changes': [{
                'Action': 'UPSERT',
                'ResourceRecordSet': {
                    'Name': record_name,
                    'Type': 'CNAME',
                    'TTL': 60,
                    'ResourceRecords': [{'Value': new_endpoint}]
                }
            }]
        }
    )
    print(f"Updated DNS {record_name} -> {new_endpoint}")
    return response['ChangeInfo']['Id']

def scale_up_ec2_asg(asg_name, desired_capacity):
    asg = boto3.client('autoscaling', region_name='us-west-2')
    response = asg.update_auto_scaling_group(
        AutoScalingGroupName=asg_name,
        DesiredCapacity=desired_capacity,
        MinSize=desired_capacity
    )
    print(f"Scaled ASG {asg_name} to {desired_capacity} instances")
    return response

if __name__ == '__main__':
    # Promote RDS replica in us-west-2
    promote_rds_read_replica('prod-db-replica')
    
    # Update DNS to point to new primary
    update_dns_record('Z123456789', 'db.prod.example.com', 'prod-db-replica.c9w2x3y4z5.us-west-2.rds.amazonaws.com')
    
    # Scale up application servers
    scale_up_ec2_asg('prod-app-asg', 5)
    
    # Wait for DNS propagation
    time.sleep(60)
    print("Failover complete")

provider "aws" {
  region = "us-west-2"
}

resource "aws_rds_cluster" "aurora_global" {
  cluster_identifier      = "prod-aurora-global"
  engine                  = "aurora-mysql"
  engine_version          = "5.7.mysql_aurora.2.10.2"
  master_username         = "admin"
  master_password         = var.db_password
  backup_retention_period = 7
  preferred_backup_window = "02:00-03:00"
  skip_final_snapshot     = true
}

resource "aws_rds_cluster_instance" "primary" {
  identifier         = "prod-aurora-primary"
  cluster_identifier = aws_rds_cluster.aurora_global.id
  instance_class     = "db.r5.large"
  engine             = aws_rds_cluster.aurora_global.engine
  engine_version     = aws_rds_cluster.aurora_global.engine_version
}

resource "aws_rds_global_cluster" "global" {
  global_cluster_identifier = "prod-global"
  source_db_cluster_identifier = aws_rds_cluster.aurora_global.arn
}

resource "aws_rds_cluster" "secondary" {
  provider = aws.us-east-1
  cluster_identifier      = "prod-aurora-secondary"
  engine                  = "aurora-mysql"
  engine_version          = "5.7.mysql_aurora.2.10.2"
  master_username         = "admin"
  master_password         = var.db_password
  backup_retention_period = 7
  preferred_backup_window = "02:00-03:00"
  skip_final_snapshot     = true
  global_cluster_identifier = aws_rds_global_cluster.global.id
}

resource "aws_rds_cluster_instance" "secondary" {
  provider = aws.us-east-1
  identifier         = "prod-aurora-secondary-instance"
  cluster_identifier = aws_rds_cluster.secondary.id
  instance_class     = "db.r5.large"
  engine             = aws_rds_cluster.secondary.engine
  engine_version     = aws_rds_cluster.secondary.engine_version
}

resource "aws_autoscaling_group" "app_secondary" {
  provider = aws.us-east-1
  name     = "prod-app-secondary"
  min_size = 2
  max_size = 10
  desired_capacity = 2
  launch_template {
    id      = aws_launch_template.app.id
    version = "$Latest"
  }
  vpc_zone_identifier = aws_subnet.secondary[*].id
  tag {
    key                 = "Name"
    value               = "prod-app-secondary"
    propagate_at_launch = true
  }
}

resource "aws_launch_template" "app" {
  name_prefix   = "prod-app"
  image_id      = "ami-0c55b159cbfafe1f0"
  instance_type = "t3.medium"
  user_data = base64encode(file("user_data.sh"))
}

import boto3
from boto3.dynamodb.conditions import Key

dynamodb = boto3.resource('dynamodb', region_name='us-east-1')
table = dynamodb.Table('Users')

def put_user(user_id, name, region):
    # Write to local region; DynamoDB global tables replicate asynchronously
    table.put_item(
        Item={
            'user_id': user_id,
            'name': name,
            'region': region,
            'timestamp': int(time.time())
        },
        ConditionExpression='attribute_not_exists(user_id)'  # avoid overwrites
    )
    print(f"User {user_id} created in {region}")

def get_user(user_id):
    # Read from local region; may be stale if replication lag
    response = table.get_item(Key={'user_id': user_id})
    item = response.get('Item')
    if item:
        print(f"User: {item['name']}, region: {item['region']}")
    else:
        print("User not found")
    return item

if __name__ == '__main__':
    import time
    # Write in us-east-1
    put_user('123', 'Alice', 'us-east-1')
    # Read from us-west-2 (replica)
    dynamodb_west = boto3.resource('dynamodb', region_name='us-west-2')
    table_west = dynamodb_west.Table('Users')
    time.sleep(2)  # wait for replication
    response = table_west.get_item(Key={'user_id': '123'})
    if 'Item' in response:
        print(f"Read from us-west-2: {response['Item']['name']}")
    else:
        print("Replication not yet complete")

import boto3
import os

def lambda_handler(event, context):
    # Triggered by CloudWatch alarm on health check failure
    region = os.environ['SECONDARY_REGION']
    rds = boto3.client('rds', region_name=region)
    route53 = boto3.client('route53')
    
    # Promote RDS replica
    replica_id = os.environ['REPLICA_ID']
    rds.promote_read_replica(DBInstanceIdentifier=replica_id)
    
    # Update DNS
    hosted_zone_id = os.environ['HOSTED_ZONE_ID']
    record_name = os.environ['RECORD_NAME']
    new_endpoint = rds.describe_db_instances(DBInstanceIdentifier=replica_id)['DBInstances'][0]['Endpoint']['Address']
    
    route53.change_resource_record_sets(
        HostedZoneId=hosted_zone_id,
        ChangeBatch={
            'Changes': [{
                'Action': 'UPSERT',
                'ResourceRecordSet': {
                    'Name': record_name,
                    'Type': 'CNAME',
                    'TTL': 60,
                    'ResourceRecords': [{'Value': new_endpoint}]
                }
            }]
        }
    )
    
    # Scale up ASG
    asg = boto3.client('autoscaling', region_name=region)
    asg.update_auto_scaling_group(
        AutoScalingGroupName=os.environ['ASG_NAME'],
        DesiredCapacity=int(os.environ['DESIRED_CAPACITY'])
    )
    
    return {'statusCode': 200, 'body': 'Failover initiated'}

# Environment variables:
# SECONDARY_REGION, REPLICA_ID, HOSTED_ZONE_ID, RECORD_NAME, ASG_NAME, DESIRED_CAPACITY

import boto3
import time

cloudwatch = boto3.client('cloudwatch', region_name='us-west-2')

def get_replication_lag(db_instance_id):
    response = cloudwatch.get_metric_statistics(
        Namespace='AWS/RDS',
        MetricName='ReplicaLag',
        Dimensions=[
            {'Name': 'DBInstanceIdentifier', 'Value': db_instance_id}
        ],
        StartTime=time.time() - 300,
        EndTime=time.time(),
        Period=60,
        Statistics=['Average']
    )
    if response['Datapoints']:
        lag = response['Datapoints'][-1]['Average']
        print(f"Replication lag for {db_instance_id}: {lag:.2f} seconds")
        return lag
    else:
        print("No data points")
        return None

def check_s3_crr_lag(bucket, rule_id):
    s3 = boto3.client('s3')
    response = s3.get_bucket_replication(Bucket=bucket)
    # CRR metrics are in CloudWatch as 'ReplicationLatency'
    # This is a simplified check
    print(f"Checking CRR for bucket {bucket}, rule {rule_id}")
    # Actual implementation would use CloudWatch metrics
    return True

if __name__ == '__main__':
    get_replication_lag('prod-db-replica')
    check_s3_crr_lag('prod-data-bucket', 'crr-rule-1')

import boto3
import time

fis = boto3.client('fis', region_name='us-east-1')

def create_ec2_stop_experiment(instance_ids):
    response = fis.create_experiment_template(
        description='Stop EC2 instances in production',
        targets={
            'Instances': {
                'resourceType': 'aws:ec2:instance',
                'resourceArns': [f'arn:aws:ec2:us-east-1:123456789012:instance/{i}' for i in instance_ids],
                'selectionMode': 'ALL'
            }
        },
        actions={
            'stopInstances': {
                'actionId': 'aws:ec2:stop-instances',
                'parameters': {},
                'targets': {'Instances': 'Instances'}
            }
        },
        stopConditions=[
            {'source': 'aws:cloudwatch:alarm', 'value': 'arn:aws:cloudwatch:us-east-1:123456789012:alarm:DRTestAlarm'}
        ],
        roleArn='arn:aws:iam::123456789012:role/FISRole'
    )
    template_id = response['experimentTemplate']['id']
    print(f"Created experiment template {template_id}")
    return template_id

def start_experiment(template_id):
    response = fis.start_experiment(experimentTemplateId=template_id)
    experiment_id = response['experiment']['id']
    print(f"Started experiment {experiment_id}")
    return experiment_id

if __name__ == '__main__':
    # Use with caution: this will stop instances!
    # template_id = create_ec2_stop_experiment(['i-0abcd1234'])
    # start_experiment(template_id)
    print("DR test scenario: simulate region failure by stopping all app instances")
    print("Monitor RTO and RPO during the experiment")

resource "aws_autoscaling_group" "secondary_spot" {
  provider = aws.us-east-1
  name     = "prod-secondary-spot"
  min_size = 1
  max_size = 10
  desired_capacity = 1
  mixed_instances_policy {
    launch_template {
      launch_template_specification {
        launch_template_id = aws_launch_template.app.id
        version = "$Latest"
      }
      override {
        instance_type = "t3.medium"
        weighted_capacity = "1"
      }
      override {
        instance_type = "t3.large"
        weighted_capacity = "2"
      }
    }
    instances_distribution {
      on_demand_percentage_above_base_capacity = 0
      spot_allocation_strategy = "capacity-optimized"
    }
  }
  vpc_zone_identifier = aws_subnet.secondary[*].id
}

resource "aws_db_instance" "secondary_small" {
  provider = aws.us-east-1
  identifier = "prod-db-secondary-small"
  instance_class = "db.t3.small"  # smaller than primary
  allocated_storage = 20
  engine = "mysql"
  engine_version = "8.0"
  username = "admin"
  password = var.db_password
  skip_final_snapshot = true
  replicate_source_db = aws_db_instance.primary.arn
  storage_type = "gp2"
  backup_retention_period = 0  # no backups needed for standby
}

import boto3

def check_dr_readiness():
    issues = []
    
    # Check RDS replicas
    rds = boto3.client('rds', region_name='us-east-1')
    replicas = rds.describe_db_instances(Filters=[{'Name': 'read-replica', 'Values': ['true']}])
    for rep in replicas['DBInstances']:
        if rep['DBInstanceStatus'] != 'available':
            issues.append(f"Replica {rep['DBInstanceIdentifier']} not available")
    
    # Check S3 CRR
    s3 = boto3.client('s3')
    buckets = s3.list_buckets()['Buckets']
    for bucket in buckets:
        try:
            repl = s3.get_bucket_replication(Bucket=bucket['Name'])
            if 'Rules' not in repl:
                issues.append(f"Bucket {bucket['Name']} has no replication rules")
        except:
            issues.append(f"Bucket {bucket['Name']} replication not configured")
    
    # Check Route 53 health checks
    route53 = boto3.client('route53')
    health_checks = route53.list_health_checks()['HealthChecks']
    if len(health_checks) == 0:
        issues.append("No Route 53 health checks configured")
    
    # Check IAM roles for cross-region access
    iam = boto3.client('iam')
    try:
        iam.get_role(RoleName='DRFailoverRole')
    except:
        issues.append("DRFailoverRole not found")
    
    if issues:
        print("DR readiness issues found:")
        for issue in issues:
            print(f"- {issue}")
    else:
        print("DR readiness check passed")
    return issues

if __name__ == '__main__':
    check_dr_readiness()

# Disaster Recovery Runbook

## 1. Pre-requisites
- AWS CLI configured with cross-account access
- Access to secondary region (us-east-1)
- IAM role `DRFailoverRole` with permissions to promote replicas, update DNS, scale ASG

## 2. Detection
- CloudWatch alarm `PrimaryRegionHealth` triggers if health check fails for 3 consecutive periods
- Manual confirmation via AWS Health Dashboard

## 3. Decision
- If primary region is unavailable for >5 minutes, declare disaster
- Notify on-call via PagerDuty

## 4. Execution (Automated)
```bash
# Promote RDS replica
aws rds promote-read-replica --db-instance-identifier prod-db-replica --region us-east-1

# Update DNS
aws route53 change-resource-record-sets --hosted-zone-id Z123456789 --change-batch file://dns-update.json

# Scale up ASG
aws autoscaling update-auto-scaling-group --auto-scaling-group-name prod-app-asg --desired-capacity 5 --region us-east-1
```

## 5. Validation
- Run smoke tests: `curl https://api.prod.example.com/health`
- Verify database connectivity: `mysql -h <new-endpoint> -u admin -p`
- Check CloudWatch metrics for errors

## 6. Failback
- After primary region recovers, reverse the process:
  1. Scale down secondary ASG
  2. Promote primary database back (or rebuild from secondary)
  3. Update DNS to primary
  4. Run validation tests

def recommend_dr_pattern(rto_seconds, rpo_seconds, budget):
    if rto_seconds <= 60 and rpo_seconds <= 5:
        return "Active-Active"
    elif rto_seconds <= 300 and rpo_seconds <= 60:
        return "Warm Standby"
    elif rto_seconds <= 1800 and rpo_seconds <= 300:
        return "Pilot Light"
    else:
        return "Backup & Restore"

if __name__ == '__main__':
    # Example: RTO=10 min, RPO=1 min, budget=moderate
    pattern = recommend_dr_pattern(600, 60, 'moderate')
    print(f"Recommended pattern: {pattern}")