Chapter 17: Disaster Recovery and Business Continuity

from dataclasses import dataclass
from datetime import datetime, timedelta
from typing import Optional

@dataclass
class RecoveryMetrics:
    """Key disaster recovery metrics"""
    rto: timedelta  # Recovery Time Objective
    rpo: timedelta  # Recovery Point Objective
    mttr: timedelta  # Mean Time To Recovery
    mtbf: timedelta  # Mean Time Between Failures
    availability_target: float  # e.g., 99.9%
    
    def calculate_downtime_budget(self, period_days: int = 30) -> timedelta:
        """Calculate allowed downtime for availability target"""
        total_time = timedelta(days=period_days)
        allowed_downtime = total_time * (1 - self.availability_target / 100)
        return allowed_downtime
    
    def is_rto_met(self, actual_recovery_time: timedelta) -> bool:
        """Check if recovery time objective is met"""
        return actual_recovery_time <= self.rto
    
    def is_rpo_met(self, data_loss_duration: timedelta) -> bool:
        """Check if recovery point objective is met"""
        return data_loss_duration <= self.rpo

# Example metrics for different service tiers
class ServiceTiers:
    CRITICAL = RecoveryMetrics(
        rto=timedelta(minutes=15),
        rpo=timedelta(minutes=5),
        mttr=timedelta(hours=1),
        mtbf=timedelta(days=30),
        availability_target=99.99
    )
    
    IMPORTANT = RecoveryMetrics(
        rto=timedelta(hours=1),
        rpo=timedelta(minutes=30),
        mttr=timedelta(hours=4),
        mtbf=timedelta(days=15),
        availability_target=99.9
    )
    
    STANDARD = RecoveryMetrics(
        rto=timedelta(hours=4),
        rpo=timedelta(hours=1),
        mttr=timedelta(hours=8),
        mtbf=timedelta(days=7),
        availability_target=99.5
    )

import os
import shutil
import json
import hashlib
from enum import Enum
from datetime import datetime
from typing import List, Dict, Optional
from abc import ABC, abstractmethod

class BackupType(Enum):
    FULL = "full"
    INCREMENTAL = "incremental"
    DIFFERENTIAL = "differential"

class BackupStatus(Enum):
    PENDING = "pending"
    RUNNING = "running"
    COMPLETED = "completed"
    FAILED = "failed"

@dataclass
class BackupJob:
    id: str
    backup_type: BackupType
    source_path: str
    destination_path: str
    created_at: datetime
    started_at: Optional[datetime] = None
    completed_at: Optional[datetime] = None
    status: BackupStatus = BackupStatus.PENDING
    size_bytes: Optional[int] = None
    error_message: Optional[str] = None

class BackupStrategy(ABC):
    """Abstract base class for backup strategies"""
    
    @abstractmethod
    def create_backup(self, source: str, destination: str) -> BackupJob:
        pass
    
    @abstractmethod
    def restore_backup(self, backup_job: BackupJob, restore_path: str) -> bool:
        pass

class FullBackupStrategy(BackupStrategy):
    """Full backup strategy implementation"""
    
    def create_backup(self, source: str, destination: str) -> BackupJob:
        """Create a full backup"""
        job = BackupJob(
            id=f"full_{datetime.now().strftime('%Y%m%d_%H%M%S')}",
            backup_type=BackupType.FULL,
            source_path=source,
            destination_path=destination,
            created_at=datetime.now()
        )
        
        try:
            job.started_at = datetime.now()
            job.status = BackupStatus.RUNNING
            
            # Create backup directory
            backup_dir = f"{destination}/{job.id}"
            os.makedirs(backup_dir, exist_ok=True)
            
            # Copy all files
            shutil.copytree(source, f"{backup_dir}/data", dirs_exist_ok=True)
            
            # Create metadata
            metadata = {
                'backup_id': job.id,
                'backup_type': job.backup_type.value,
                'source_path': source,
                'backup_time': job.started_at.isoformat(),
                'file_count': self._count_files(f"{backup_dir}/data")
            }
            
            with open(f"{backup_dir}/metadata.json", 'w') as f:
                json.dump(metadata, f, indent=2)
            
            job.size_bytes = self._get_directory_size(backup_dir)
            job.completed_at = datetime.now()
            job.status = BackupStatus.COMPLETED
            
        except Exception as e:
            job.status = BackupStatus.FAILED
            job.error_message = str(e)
        
        return job
    
    def restore_backup(self, backup_job: BackupJob, restore_path: str) -> bool:
        """Restore from full backup"""
        try:
            backup_dir = f"{backup_job.destination_path}/{backup_job.id}"
            data_dir = f"{backup_dir}/data"
            
            if not os.path.exists(data_dir):
                print(f"Backup data not found: {data_dir}")
                return False
            
            # Restore files
            shutil.copytree(data_dir, restore_path, dirs_exist_ok=True)
            return True
            
        except Exception as e:
            print(f"Full restore failed: {e}")
            return False
    
    def _count_files(self, directory: str) -> int:
        """Count files in directory recursively"""
        count = 0
        for root, dirs, files in os.walk(directory):
            count += len(files)
        return count
    
    def _get_directory_size(self, directory: str) -> int:
        """Calculate total size of directory"""
        total_size = 0
        for dirpath, dirnames, filenames in os.walk(directory):
            for filename in filenames:
                filepath = os.path.join(dirpath, filename)
                try:
                    total_size += os.path.getsize(filepath)
                except OSError:
                    pass
        return total_size

import asyncio
import aiohttp
from typing import List, Callable, Any
from dataclasses import dataclass
from enum import Enum

class HealthStatus(Enum):
    HEALTHY = "healthy"
    UNHEALTHY = "unhealthy"
    UNKNOWN = "unknown"

@dataclass
class ServiceEndpoint:
    name: str
    url: str
    priority: int  # Lower number = higher priority
    health_check_path: str = "/health"
    timeout_seconds: int = 5
    status: HealthStatus = HealthStatus.UNKNOWN
    last_check: Optional[datetime] = None
    consecutive_failures: int = 0

class FailoverManager:
    """Manages automatic failover between service endpoints"""
    
    def __init__(self, endpoints: List[ServiceEndpoint], 
                 failure_threshold: int = 3,
                 check_interval: int = 30):
        self.endpoints = sorted(endpoints, key=lambda x: x.priority)
        self.failure_threshold = failure_threshold
        self.check_interval = check_interval
        self.active_endpoint: Optional[ServiceEndpoint] = None
        self.failover_callbacks: List[Callable] = []
        self._running = False
    
    async def start_monitoring(self):
        """Start health check monitoring"""
        self._running = True
        while self._running:
            await self._check_all_endpoints()
            await self._update_active_endpoint()
            await asyncio.sleep(self.check_interval)
    
    def stop_monitoring(self):
        """Stop health check monitoring"""
        self._running = False
    
    async def _check_all_endpoints(self):
        """Check health of all endpoints"""
        tasks = [self._check_endpoint_health(endpoint) 
                for endpoint in self.endpoints]
        await asyncio.gather(*tasks, return_exceptions=True)
    
    async def _check_endpoint_health(self, endpoint: ServiceEndpoint):
        """Check health of a single endpoint"""
        try:
            async with aiohttp.ClientSession() as session:
                health_url = f"{endpoint.url.rstrip('/')}{endpoint.health_check_path}"
                async with session.get(
                    health_url, 
                    timeout=aiohttp.ClientTimeout(total=endpoint.timeout_seconds)
                ) as response:
                    if response.status == 200:
                        endpoint.status = HealthStatus.HEALTHY
                        endpoint.consecutive_failures = 0
                    else:
                        endpoint.status = HealthStatus.UNHEALTHY
                        endpoint.consecutive_failures += 1
        except Exception:
            endpoint.status = HealthStatus.UNHEALTHY
            endpoint.consecutive_failures += 1
        
        endpoint.last_check = datetime.now()
    
    async def _update_active_endpoint(self):
        """Update active endpoint based on health checks"""
        # Find the highest priority healthy endpoint
        healthy_endpoints = [
            ep for ep in self.endpoints 
            if ep.status == HealthStatus.HEALTHY
        ]
        
        if not healthy_endpoints:
            if self.active_endpoint:
                print("No healthy endpoints available!")
                await self._trigger_failover(None)
            return
        
        best_endpoint = healthy_endpoints[0]  # Already sorted by priority
        
        # Check if we need to failover
        if (self.active_endpoint is None or 
            self.active_endpoint != best_endpoint or
            self.active_endpoint.consecutive_failures >= self.failure_threshold):
            
            await self._trigger_failover(best_endpoint)
    
    async def _trigger_failover(self, new_endpoint: Optional[ServiceEndpoint]):
        """Trigger failover to new endpoint"""
        old_endpoint = self.active_endpoint
        self.active_endpoint = new_endpoint
        
        print(f"Failover triggered: {old_endpoint.name if old_endpoint else 'None'} -> {new_endpoint.name if new_endpoint else 'None'}")
        
        # Execute failover callbacks
        for callback in self.failover_callbacks:
            try:
                await callback(old_endpoint, new_endpoint)
            except Exception as e:
                print(f"Failover callback failed: {e}")
    
    def add_failover_callback(self, callback: Callable):
        """Add callback to be executed on failover"""
        self.failover_callbacks.append(callback)
    
    def get_active_endpoint(self) -> Optional[ServiceEndpoint]:
        """Get currently active endpoint"""
        return self.active_endpoint

class ActivePassiveCluster:
    """Active-Passive failover pattern"""
    
    def __init__(self, primary_endpoint: ServiceEndpoint, 
                 secondary_endpoint: ServiceEndpoint):
        self.primary = primary_endpoint
        self.secondary = secondary_endpoint
        self.is_failed_over = False
    
    async def failover_callback(self, old_endpoint: Optional[ServiceEndpoint], 
                               new_endpoint: Optional[ServiceEndpoint]):
        """Handle failover between primary and secondary"""
        if new_endpoint == self.secondary and not self.is_failed_over:
            print("Activating secondary endpoint")
            await self._activate_secondary()
            self.is_failed_over = True
        elif new_endpoint == self.primary and self.is_failed_over:
            print("Failing back to primary endpoint")
            await self._deactivate_secondary()
            self.is_failed_over = False
    
    async def _activate_secondary(self):
        """Activate secondary endpoint"""
        # Implementation specific to your infrastructure
        # e.g., update load balancer, DNS records, etc.
        pass
    
    async def _deactivate_secondary(self):
        """Deactivate secondary endpoint"""
        # Implementation specific to your infrastructure
        pass

class ActiveActiveCluster:
    """Active-Active load balancing pattern"""
    
    def __init__(self, endpoints: List[ServiceEndpoint]):
        self.endpoints = endpoints
        self.active_endpoints: List[ServiceEndpoint] = []
    
    async def failover_callback(self, old_endpoint: Optional[ServiceEndpoint], 
                               new_endpoint: Optional[ServiceEndpoint]):
        """Update active endpoints list"""
        # Update list of active endpoints
        self.active_endpoints = [
            ep for ep in self.endpoints 
            if ep.status == HealthStatus.HEALTHY
        ]
        
        print(f"Active endpoints updated: {[ep.name for ep in self.active_endpoints]}")
        await self._update_load_balancer()
    
    async def _update_load_balancer(self):
        """Update load balancer configuration"""
        # Implementation specific to your load balancer
        # e.g., update upstream servers, weights, etc.
        pass

import asyncio
from typing import Dict, List
from dataclasses import dataclass

@dataclass
class Region:
    name: str
    endpoint: str
    is_primary: bool = False
    replication_lag_ms: int = 0
    last_sync: Optional[datetime] = None

class CrossRegionReplication:
    """Manages data replication across regions"""
    
    def __init__(self, regions: List[Region]):
        self.regions = {region.name: region for region in regions}
        self.primary_region = next(
            (r for r in regions if r.is_primary), 
            regions[0] if regions else None
        )
        self.replication_tasks: Dict[str, asyncio.Task] = {}
    
    async def start_replication(self):
        """Start replication to all secondary regions"""
        for region_name, region in self.regions.items():
            if not region.is_primary:
                task = asyncio.create_task(
                    self._replicate_to_region(region)
                )
                self.replication_tasks[region_name] = task
    
    async def stop_replication(self):
        """Stop all replication tasks"""
        for task in self.replication_tasks.values():
            task.cancel()
        
        await asyncio.gather(
            *self.replication_tasks.values(), 
            return_exceptions=True
        )
        self.replication_tasks.clear()
    
    async def _replicate_to_region(self, target_region: Region):
        """Replicate data to a specific region"""
        while True:
            try:
                # Get changes from primary region
                changes = await self._get_changes_since(
                    target_region.last_sync
                )
                
                if changes:
                    # Apply changes to target region
                    await self._apply_changes(target_region, changes)
                    target_region.last_sync = datetime.now()
                
                await asyncio.sleep(1)  # Replication interval
                
            except Exception as e:
                print(f"Replication to {target_region.name} failed: {e}")
                await asyncio.sleep(5)  # Retry delay
    
    async def _get_changes_since(self, since: Optional[datetime]) -> List[Dict]:
        """Get changes from primary region since timestamp"""
        # Implementation depends on your data store
        # e.g., query change log, WAL, etc.
        return []
    
    async def _apply_changes(self, region: Region, changes: List[Dict]):
        """Apply changes to target region"""
        # Implementation depends on your data store
        # e.g., batch write, streaming replication, etc.
        pass
    
    async def promote_region_to_primary(self, region_name: str):
        """Promote a secondary region to primary"""
        if region_name not in self.regions:
            raise ValueError(f"Region {region_name} not found")
        
        new_primary = self.regions[region_name]
        
        # Stop replication
        await self.stop_replication()
        
        # Update primary designation
        if self.primary_region:
            self.primary_region.is_primary = False
        
        new_primary.is_primary = True
        self.primary_region = new_primary
        
        print(f"Promoted {region_name} to primary region")
        
        # Restart replication from new primary
        await self.start_replication()

import asyncio
from typing import List, Dict, Any, Callable
from dataclasses import dataclass
from enum import Enum

class TestType(Enum):
    BACKUP_RESTORE = "backup_restore"
    FAILOVER = "failover"
    REPLICATION = "replication"
    RTO_VALIDATION = "rto_validation"
    RPO_VALIDATION = "rpo_validation"

class TestStatus(Enum):
    PENDING = "pending"
    RUNNING = "running"
    PASSED = "passed"
    FAILED = "failed"
    SKIPPED = "skipped"

@dataclass
class DRTestResult:
    test_name: str
    test_type: TestType
    status: TestStatus
    start_time: datetime
    end_time: Optional[datetime] = None
    duration: Optional[timedelta] = None
    error_message: Optional[str] = None
    metrics: Dict[str, Any] = None

class DRTestSuite:
    """Disaster Recovery testing framework"""
    
    def __init__(self):
        self.test_results: List[DRTestResult] = []
        self.test_functions: Dict[TestType, Callable] = {
            TestType.BACKUP_RESTORE: self._test_backup_restore,
            TestType.FAILOVER: self._test_failover,
            TestType.REPLICATION: self._test_replication,
            TestType.RTO_VALIDATION: self._test_rto,
            TestType.RPO_VALIDATION: self._test_rpo
        }
    
    async def run_all_tests(self) -> List[DRTestResult]:
        """Run all DR tests"""
        self.test_results.clear()
        
        for test_type in TestType:
            result = await self._run_test(test_type)
            self.test_results.append(result)
        
        return self.test_results
    
    async def run_specific_test(self, test_type: TestType) -> DRTestResult:
        """Run a specific DR test"""
        return await self._run_test(test_type)
    
    async def _run_test(self, test_type: TestType) -> DRTestResult:
        """Run a single DR test"""
        result = DRTestResult(
            test_name=test_type.value,
            test_type=test_type,
            status=TestStatus.PENDING,
            start_time=datetime.now()
        )
        
        try:
            result.status = TestStatus.RUNNING
            test_function = self.test_functions[test_type]
            
            # Run the test
            test_metrics = await test_function()
            
            result.status = TestStatus.PASSED
            result.metrics = test_metrics
            
        except Exception as e:
            result.status = TestStatus.FAILED
            result.error_message = str(e)
        
        finally:
            result.end_time = datetime.now()
            result.duration = result.end_time - result.start_time
        
        return result
    
    async def _test_backup_restore(self) -> Dict[str, Any]:
        """Test backup and restore functionality"""
        # Create test data
        test_data_path = "/tmp/dr_test_data"
        os.makedirs(test_data_path, exist_ok=True)
        
        # Write test files
        test_files = []
        for i in range(10):
            file_path = f"{test_data_path}/test_file_{i}.txt"
            with open(file_path, 'w') as f:
                f.write(f"Test data {i} - {datetime.now()}")
            test_files.append(file_path)
        
        # Create backup
        backup_strategy = FullBackupStrategy()
        backup_job = backup_strategy.create_backup(
            test_data_path, 
            "/tmp/dr_test_backup"
        )
        
        if backup_job.status != BackupStatus.COMPLETED:
            raise Exception(f"Backup failed: {backup_job.error_message}")
        
        # Delete original data
        shutil.rmtree(test_data_path)
        
        # Restore from backup
        restore_path = "/tmp/dr_test_restore"
        restore_success = backup_strategy.restore_backup(backup_job, restore_path)
        
        if not restore_success:
            raise Exception("Restore failed")
        
        # Verify restored data
        for i, original_file in enumerate(test_files):
            restored_file = original_file.replace(test_data_path, restore_path)
            if not os.path.exists(restored_file):
                raise Exception(f"Restored file missing: {restored_file}")
        
        # Cleanup
        shutil.rmtree("/tmp/dr_test_backup", ignore_errors=True)
        shutil.rmtree(restore_path, ignore_errors=True)
        
        return {
            "backup_size_bytes": backup_job.size_bytes,
            "backup_duration_seconds": (backup_job.completed_at - backup_job.started_at).total_seconds(),
            "files_backed_up": len(test_files)
        }
    
    async def _test_failover(self) -> Dict[str, Any]:
        """Test failover functionality"""
        # Create test endpoints
        primary = ServiceEndpoint(
            name="primary",
            url="http://primary.example.com",
            priority=1
        )
        secondary = ServiceEndpoint(
            name="secondary",
            url="http://secondary.example.com",
            priority=2
        )
        
        failover_manager = FailoverManager([primary, secondary])
        
        # Simulate primary failure
        primary.status = HealthStatus.UNHEALTHY
        primary.consecutive_failures = 5
        
        # Simulate secondary healthy
        secondary.status = HealthStatus.HEALTHY
        secondary.consecutive_failures = 0
        
        # Test failover
        await failover_manager._update_active_endpoint()
        
        if failover_manager.get_active_endpoint() != secondary:
            raise Exception("Failover did not switch to secondary")
        
        # Test failback
        primary.status = HealthStatus.HEALTHY
        primary.consecutive_failures = 0
        
        await failover_manager._update_active_endpoint()
        
        if failover_manager.get_active_endpoint() != primary:
            raise Exception("Failback did not switch to primary")
        
        return {
            "failover_successful": True,
            "failback_successful": True
        }
    
    async def _test_replication(self) -> Dict[str, Any]:
        """Test cross-region replication"""
        # Create test regions
        primary_region = Region(
            name="us-east-1",
            endpoint="https://us-east-1.example.com",
            is_primary=True
        )
        secondary_region = Region(
            name="us-west-2",
            endpoint="https://us-west-2.example.com",
            is_primary=False
        )
        
        replication = CrossRegionReplication([primary_region, secondary_region])
        
        # Start replication
        await replication.start_replication()
        
        # Wait for initial sync
        await asyncio.sleep(2)
        
        # Stop replication
        await replication.stop_replication()
        
        return {
            "replication_started": True,
            "replication_stopped": True,
            "regions_configured": len(replication.regions)
        }
    
    async def _test_rto(self) -> Dict[str, Any]:
        """Test Recovery Time Objective"""
        target_rto = timedelta(minutes=15)
        
        # Simulate failure detection and recovery
        start_time = datetime.now()
        
        # Simulate recovery steps
        await asyncio.sleep(0.1)  # Simulate failure detection
        await asyncio.sleep(0.2)  # Simulate failover
        await asyncio.sleep(0.1)  # Simulate service restart
        
        end_time = datetime.now()
        actual_rto = end_time - start_time
        
        rto_met = actual_rto <= target_rto
        
        if not rto_met:
            raise Exception(f"RTO not met: {actual_rto} > {target_rto}")
        
        return {
            "target_rto_seconds": target_rto.total_seconds(),
            "actual_rto_seconds": actual_rto.total_seconds(),
            "rto_met": rto_met
        }
    
    async def _test_rpo(self) -> Dict[str, Any]:
        """Test Recovery Point Objective"""
        target_rpo = timedelta(minutes=5)
        
        # Simulate data loss scenario
        last_backup_time = datetime.now() - timedelta(minutes=2)
        failure_time = datetime.now()
        
        data_loss_duration = failure_time - last_backup_time
        rpo_met = data_loss_duration <= target_rpo
        
        if not rpo_met:
            raise Exception(f"RPO not met: {data_loss_duration} > {target_rpo}")
        
        return {
            "target_rpo_seconds": target_rpo.total_seconds(),
            "actual_data_loss_seconds": data_loss_duration.total_seconds(),
            "rpo_met": rpo_met
        }
    
    def generate_test_report(self) -> str:
        """Generate comprehensive test report"""
        report = ["\n=== Disaster Recovery Test Report ==="]
        report.append(f"Test Run Date: {datetime.now()}")
        report.append(f"Total Tests: {len(self.test_results)}")
        
        passed = sum(1 for r in self.test_results if r.status == TestStatus.PASSED)
        failed = sum(1 for r in self.test_results if r.status == TestStatus.FAILED)
        
        report.append(f"Passed: {passed}")
        report.append(f"Failed: {failed}")
        report.append("")
        
        for result in self.test_results:
            report.append(f"Test: {result.test_name}")
            report.append(f"  Status: {result.status.value}")
            report.append(f"  Duration: {result.duration}")
            
            if result.error_message:
                report.append(f"  Error: {result.error_message}")
            
            if result.metrics:
                report.append("  Metrics:")
                for key, value in result.metrics.items():
                    report.append(f"    {key}: {value}")
            
            report.append("")
        
        return "\n".join(report)