Disaster Recovery: RTO and RPO Planning

def calculate_downtime_cost(
    revenue_per_hour: float,
    rto_hours: float,
    incidents_per_year: int = 2,
    data_loss_multiplier: float = 1.5
) -> dict:
    """
    Estimate annual cost of downtime based on RTO.
    data_loss_multiplier accounts for additional costs beyond lost revenue
    (customer trust, compliance penalties, recovery efforts).
    """
    downtime_cost_per_incident = revenue_per_hour * rto_hours
    data_recovery_cost = downtime_cost_per_incident * data_loss_multiplier
    
    annual_downtime_cost = downtime_cost_per_incident * incidents_per_year
    annual_total_cost = (downtime_cost_per_incident + data_recovery_cost) * incidents_per_year
    
    return {
        "cost_per_incident": downtime_cost_per_incident,
        "annual_downtime_cost": annual_downtime_cost,
        "annual_total_cost": annual_total_cost,
        "max_acceptable_dr_investment": annual_total_cost * 3  # 3-year ROI
    }

# Example: SaaS platform generating $50K/hour
costs = calculate_downtime_cost(
    revenue_per_hour=50000,
    rto_hours=4,
    incidents_per_year=2
)
print(f"Annual DR cost: ${costs['annual_total_cost']:,.2f}")
print(f"Max DR investment: ${costs['max_acceptable_dr_investment']:,.2f}")

service_tiers:
  tier_1_critical:
    description: "Revenue-generating, customer-facing systems"
    rto_target: "1 hour"
    rpo_target: "5 minutes"
    examples:
      - payment_processing
      - order_management
      - authentication_service
    backup_strategy: "continuous_replication"
    infrastructure: "multi_region_active_active"
    estimated_cost_multiplier: 10x
    
  tier_2_important:
    description: "Business operations, internal tools"
    rto_target: "4 hours"
    rpo_target: "1 hour"
    examples:
      - inventory_management
      - reporting_systems
      - admin_dashboards
    backup_strategy: "hourly_snapshots"
    infrastructure: "multi_region_active_passive"
    estimated_cost_multiplier: 3x
    
  tier_3_standard:
    description: "Non-critical systems, can operate manually temporarily"
    rto_target: "24 hours"
    rpo_target: "24 hours"
    examples:
      - analytics_pipeline
      - marketing_automation
      - documentation_sites
    backup_strategy: "daily_backups"
    infrastructure: "single_region_with_backups"
    estimated_cost_multiplier: 1x

#!/usr/bin/env python3
"""
Automated backup orchestration with configurable RPO intervals.
Supports full and incremental backups with retention management.
"""
import subprocess
import boto3
from datetime import datetime, timedelta
from typing import List
import json

class BackupOrchestrator:
    def __init__(self, config_path: str):
        with open(config_path) as f:
            self.config = json.load(f)
        self.s3 = boto3.client('s3')
    
    def execute_postgres_backup(self, database: str, backup_type: str = "incremental"):
        """Execute PostgreSQL backup with WAL archiving for PITR."""
        timestamp = datetime.utcnow().strftime('%Y%m%d_%H%M%S')
        backup_file = f"{database}_{backup_type}_{timestamp}.dump"
        
        if backup_type == "full":
            cmd = [
                "pg_basebackup",
                "-h", self.config['db_host'],
                "-D", f"/backups/{backup_file}",
                "-Ft", "-z", "-P"
            ]
        else:
            # Incremental via WAL archiving
            cmd = [
                "pg_dump",
                "-h", self.config['db_host'],
                "-Fc",  # Custom format for parallel restore
                "-f", f"/backups/{backup_file}",
                database
            ]
        
        subprocess.run(cmd, check=True)
        self.upload_to_s3(backup_file)
        return backup_file
    
    def upload_to_s3(self, local_file: str):
        """Upload backup to S3 with appropriate storage class."""
        bucket = self.config['backup_bucket']
        key = f"backups/{local_file}"
        
        # Use Intelligent-Tiering for automatic cost optimization
        self.s3.upload_file(
            f"/backups/{local_file}",
            bucket,
            key,
            ExtraArgs={'StorageClass': 'INTELLIGENT_TIERING'}
        )
    
    def enforce_retention_policy(self):
        """Delete backups older than retention period."""
        retention_days = self.config['retention_days']
        cutoff_date = datetime.utcnow() - timedelta(days=retention_days)
        
        bucket = self.config['backup_bucket']
        response = self.s3.list_objects_v2(Bucket=bucket, Prefix='backups/')
        
        for obj in response.get('Contents', []):
            if obj['LastModified'].replace(tzinfo=None) < cutoff_date:
                self.s3.delete_object(Bucket=bucket, Key=obj['Key'])
                print(f"Deleted old backup: {obj['Key']}")

# Configuration example
config = {
    "db_host": "prod-db.example.com",
    "backup_bucket": "company-dr-backups",
    "retention_days": 30,
    "rpo_minutes": 15
}

# Schedule this based on RPO requirements
# RPO 15 min = run every 15 min
# RPO 1 hour = run every hour

# Terraform configuration for rapid DR environment provisioning
terraform {
  required_providers {
    aws = {
      source  = "hashicorp/aws"
      version = "~> 5.0"
    }
  }
  
  backend "s3" {
    bucket = "company-terraform-state"
    key    = "dr/production-replica.tfstate"
    region = "us-west-2"
  }
}

# DR region configuration
provider "aws" {
  alias  = "dr"
  region = var.dr_region
}

# Replicate production VPC in DR region
module "dr_vpc" {
  source = "./modules/vpc"
  providers = {
    aws = aws.dr
  }
  
  vpc_cidr = var.dr_vpc_cidr
  environment = "dr-production"
}

# Database with automated failover
resource "aws_rds_cluster" "primary" {
  cluster_identifier     = "prod-db-cluster"
  engine                = "aurora-postgresql"
  database_name         = "production"
  master_username       = var.db_username
  master_password       = var.db_password
  
  # Enable global database for cross-region replication
  global_cluster_identifier = aws_rds_global_cluster.main.id
  
  backup_retention_period = 35
  preferred_backup_window = "03:00-04:00"
  
  # Enable automated backups for PITR
  enabled_cloudwatch_logs_exports = ["postgresql"]
}

resource "aws_rds_global_cluster" "main" {
  global_cluster_identifier = "prod-global-db"
  engine                   = "aurora-postgresql"
  engine_version          = "14.6"
  database_name           = "production"
}

# DR region cluster (read replica that can be promoted)
resource "aws_rds_cluster" "dr" {
  provider              = aws.dr
  cluster_identifier    = "prod-db-cluster-dr"
  engine               = "aurora-postgresql"
  global_cluster_identifier = aws_rds_global_cluster.main.id
  
  # This cluster can be promoted to standalone during failover
}

# Application auto-scaling in DR region (scaled to zero normally)
resource "aws_autoscaling_group" "dr_app" {
  provider            = aws.dr
  name               = "app-dr-asg"
  vpc_zone_identifier = module.dr_vpc.private_subnet_ids
  min_size           = 0  # Scale to zero for cost savings
  max_size           = 10
  desired_capacity   = 0
  
  # During DR activation, set desired_capacity to match production
  
  launch_template {
    id      = aws_launch_template.app_dr.id
    version = "$Latest"
  }
}

#!/usr/bin/env python3
"""
DR test orchestration with automated validation.
Simulates failover and validates RTO/RPO compliance.
"""
import time
import requests
from datetime import datetime
from typing import Dict, List
import boto3

class DRTestOrchestrator:
    def __init__(self, config: Dict):
        self.config = config
        self.results = []
        
    def execute_dr_test(self) -> Dict:
        """Run complete DR test and measure actual RTO/RPO."""
        print("Starting DR test...")
        test_start = datetime.utcnow()
        
        # Step 1: Mark test data for RPO validation
        last_transaction_id = self.record_test_transaction()
        
        # Step 2: Simulate failure
        self.simulate_primary_failure()
        failure_time = datetime.utcnow()
        
        # Step 3: Execute failover procedures
        self.initiate_failover()
        
        # Step 4: Wait for DR environment to be healthy
        recovery_time = self.wait_for_recovery()
        
        # Step 5: Validate data consistency
        recovered_transaction_id = self.validate_data_recovery()
        
        # Calculate actual metrics
        actual_rto = (recovery_time - failure_time).total_seconds() / 60
        data_loss_minutes = self.calculate_data_loss(
            last_transaction_id,
            recovered_transaction_id
        )
        
        results = {
            "test_timestamp": test_start.isoformat(),
            "target_rto_minutes": self.config['target_rto'],
            "actual_rto_minutes": actual_rto,
            "rto_met": actual_rto <= self.config['target_rto'],
            "target_rpo_minutes": self.config['target_rpo'],
            "actual_data_loss_minutes": data_loss_minutes,
            "rpo_met": data_loss_minutes <= self.config['target_rpo'],
            "test_passed": (
                actual_rto <= self.config['target_rto'] and
                data_loss_minutes <= self.config['target_rpo']
            )
        }
        
        self.publish_results(results)
        return results
    
    def simulate_primary_failure(self):
        """Simulate primary region failure (in test environment)."""
        # In production DR test, this might be:
        # - Disable primary region routing in Route53
        # - Stop primary database cluster
        # - Terminate primary application instances
        print("Simulating primary region failure...")
        time.sleep(2)
    
    def initiate_failover(self):
        """Execute automated failover procedures."""
        # Promote DR database to primary
        # Update DNS to point to DR region
        # Scale up DR application instances
        print("Initiating failover to DR region...")
    
    def wait_for_recovery(self) -> datetime:
        """Poll DR environment until healthy."""
        dr_endpoint = self.config['dr_health_check_url']
        max_attempts = 120  # 10 minutes with 5-second intervals
        
        for attempt in range(max_attempts):
            try:
                response = requests.get(dr_endpoint, timeout=5)
                if response.status_code == 200:
                    return datetime.utcnow()
            except requests.RequestException:
                pass
            time.sleep(5)
        
        raise Exception("DR environment failed to become healthy")
    
    def validate_data_recovery(self) -> str:
        """Verify data consistency in DR environment."""
        # Query DR database for last transaction
        # Compare with pre-failure state
        return "last_recovered_transaction_id"
    
    def calculate_data_loss(self, last_tx: str, recovered_tx: str) -> float:
        """Calculate actual data loss in minutes."""
        # Implementation depends on your transaction ID schema
        return 2.5  # Example: 2.5 minutes of data loss
    
    def publish_results(self, results: Dict):
        """Send results to monitoring system."""
        # Push to CloudWatch, Datadog, or your monitoring platform
        print(f"DR Test Results: {results}")
        
        if not results['test_passed']:
            self.send_alert(results)
    
    def send_alert(self, results: Dict):
        """Alert on DR test failures."""
        message = f"""
        DR Test Failed!
        Target RTO: {results['target_rto_minutes']} min
        Actual RTO: {results['actual_rto_minutes']:.1f} min
        Target RPO: {results['target_rpo_minutes']} min
        Actual Data Loss: {results['actual_data_loss_minutes']:.1f} min
        """
        # Send to PagerDuty, Slack, etc.
        print(message)