Cloud Backup and Disaster Recovery: Business Continuity P...

requirements.txt:

boto3>=1.34.0
python-dateutil>=2.8.2

Complete Backup and Disaster Recovery Manager:

#!/usr/bin/env python3
"""
Cloud Backup & Disaster Recovery - Backup Manager
Production-ready backup and disaster recovery with comprehensive error handling
"""

import boto3
from botocore.exceptions import ClientError, BotoCoreError
from typing import Dict, List, Optional
from dataclasses import dataclass, asdict, field
from enum import Enum
from datetime import datetime, timedelta
import logging
import os
import json

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)


class BackupError(Exception):
    """Base exception for backup errors."""
    pass


class BackupNotFoundError(BackupError):
    """Raised when backup is not found."""
    pass


class RetentionPolicy(Enum):
    """Backup retention policies."""
    DAILY = "daily"
    WEEKLY = "weekly"
    MONTHLY = "monthly"
    YEARLY = "yearly"
    CUSTOM = "custom"


@dataclass
class BackupConfig:
    """Backup configuration."""
    source_bucket: str
    destination_bucket: str
    prefix: str = "backups"
    retention_days: int = 30
    retention_policy: RetentionPolicy = RetentionPolicy.DAILY
    encryption: bool = True
    versioning: bool = True
    cross_region: bool = False
    destination_region: Optional[str] = None
    tags: Dict[str, str] = field(default_factory=dict)
    
    def to_dict(self) -> Dict:
        """Convert to dictionary."""
        result = asdict(self)
        result['retention_policy'] = self.retention_policy.value
        return result


@dataclass
class BackupResult:
    """Result of backup operation."""
    backup_id: str
    source: str
    destination: str
    timestamp: datetime
    size_bytes: int
    status: str
    metadata: Dict = field(default_factory=dict)
    
    def to_dict(self) -> Dict:
        """Convert to dictionary."""
        result = asdict(self)
        result['timestamp'] = self.timestamp.isoformat()
        return result


class BackupManager:
    """Manages cloud backups with comprehensive error handling."""
    
    def __init__(
        self,
        region_name: str = 'us-east-1',
        aws_access_key_id: Optional[str] = None,
        aws_secret_access_key: Optional[str] = None
    ):
        """Initialize backup manager.
        
        Args:
            region_name: AWS region (default: us-east-1)
            aws_access_key_id: AWS access key (defaults to env/credentials)
            aws_secret_access_key: AWS secret key (defaults to env/credentials)
        """
        self.region_name = region_name
        
        try:
            session = boto3.Session(
                aws_access_key_id=aws_access_key_id or os.getenv('AWS_ACCESS_KEY_ID'),
                aws_secret_access_key=aws_secret_access_key or os.getenv('AWS_SECRET_ACCESS_KEY'),
                region_name=region_name
            )
            
            self.s3 = session.client('s3', region_name=region_name)
            self.backup_history: List[BackupResult] = []
            
            logger.info(f"Initialized BackupManager for region: {region_name}")
            
        except (ClientError, BotoCoreError) as e:
            error_msg = f"Failed to initialize AWS clients: {e}"
            logger.error(error_msg)
            raise BackupError(error_msg) from e
    
    def create_backup(
        self,
        config: BackupConfig,
        sync: bool = True
    ) -> BackupResult:
        """Create backup with comprehensive error handling.
        
        Args:
            config: Backup configuration
            sync: If True, wait for backup to complete (default: True)
            
        Returns:
            BackupResult with backup details
            
        Raises:
            BackupError: If backup creation fails
        """
        try:
            backup_id = datetime.utcnow().strftime('%Y%m%d-%H%M%S')
            timestamp = datetime.utcnow()
            
            # Create backup key
            backup_key = f"{config.prefix}/{backup_id}/"
            
            logger.info(f"Starting backup: {config.source_bucket} -> {config.destination_bucket}/{backup_key}")
            
            # Ensure destination bucket exists and has proper configuration
            self._ensure_backup_bucket(config)
            
            # Perform backup
            if sync:
                backup_size = self._copy_all_objects(
                    config.source_bucket,
                    config.destination_bucket,
                    backup_key,
                    config
                )
            else:
                # Trigger async backup (e.g., using S3 replication or Lambda)
                backup_size = 0  # Unknown for async
                logger.info("Triggered async backup")
            
            # Store backup metadata
            metadata = {
                'config': config.to_dict(),
                'backup_key': backup_key
            }
            
            backup_result = BackupResult(
                backup_id=backup_id,
                source=config.source_bucket,
                destination=f"{config.destination_bucket}/{backup_key}",
                timestamp=timestamp,
                size_bytes=backup_size,
                status='completed',
                metadata=metadata
            )
            
            # Save backup metadata
            self._save_backup_metadata(config.destination_bucket, backup_key, backup_result)
            
            self.backup_history.append(backup_result)
            logger.info(f"Backup completed: {backup_id} ({backup_size:,} bytes)")
            
            return backup_result
        
        except Exception as e:
            error_msg = f"Failed to create backup: {e}"
            logger.error(error_msg, exc_info=True)
            raise BackupError(error_msg) from e
    
    def _ensure_backup_bucket(self, config: BackupConfig) -> None:
        """Ensure backup bucket exists with proper configuration.
        
        Args:
            config: Backup configuration
        """
        try:
            # Check if bucket exists
            try:
                self.s3.head_bucket(Bucket=config.destination_bucket)
            except ClientError as e:
                error_code = e.response['Error']['Code']
                if error_code == '404':
                    # Create bucket
                    create_params = {'Bucket': config.destination_bucket}
                    if config.destination_region and config.destination_region != self.region_name:
                        create_params['CreateBucketConfiguration'] = {
                            'LocationConstraint': config.destination_region
                        }
                    
                    self.s3.create_bucket(**create_params)
                    logger.info(f"Created backup bucket: {config.destination_bucket}")
                else:
                    raise
            
            # Enable versioning if requested
            if config.versioning:
                try:
                    versioning = self.s3.get_bucket_versioning(Bucket=config.destination_bucket)
                    if versioning.get('Status') != 'Enabled':
                        self.s3.put_bucket_versioning(
                            Bucket=config.destination_bucket,
                            VersioningConfiguration={'Status': 'Enabled'}
                        )
                        logger.info(f"Enabled versioning on bucket: {config.destination_bucket}")
                except ClientError as e:
                    logger.warning(f"Failed to enable versioning: {e}")
            
            # Enable encryption if requested
            if config.encryption:
                try:
                    encryption = self.s3.get_bucket_encryption(Bucket=config.destination_bucket)
                    if not encryption.get('ServerSideEncryptionConfiguration'):
                        self.s3.put_bucket_encryption(
                            Bucket=config.destination_bucket,
                            ServerSideEncryptionConfiguration={
                                'Rules': [{
                                    'ApplyServerSideEncryptionByDefault': {
                                        'SSEAlgorithm': 'AES256'
                                    }
                                }]
                            }
                        )
                        logger.info(f"Enabled encryption on bucket: {config.destination_bucket}")
                except ClientError as e:
                    if e.response['Error']['Code'] != 'ServerSideEncryptionConfigurationNotFoundError':
                        logger.warning(f"Failed to enable encryption: {e}")
        
        except ClientError as e:
            raise BackupError(f"Failed to configure backup bucket: {e}") from e
    
    def _copy_all_objects(
        self,
        source_bucket: str,
        dest_bucket: str,
        dest_prefix: str,
        config: BackupConfig
    ) -> int:
        """Copy all objects from source to destination.
        
        Args:
            source_bucket: Source S3 bucket
            dest_bucket: Destination S3 bucket
            dest_prefix: Destination prefix
            config: Backup configuration
            
        Returns:
            Total size of copied objects in bytes
        """
        total_size = 0
        copied_count = 0
        
        try:
            paginator = self.s3.get_paginator('list_objects_v2')
            pages = paginator.paginate(Bucket=source_bucket)
            
            for page in pages:
                if 'Contents' not in page:
                    continue
                
                for obj in page['Contents']:
                    source_key = obj['Key']
                    dest_key = f"{dest_prefix}{source_key}"
                    
                    try:
                        # Copy object
                        copy_source = {'Bucket': source_bucket, 'Key': source_key}
                        self.s3.copy_object(
                            CopySource=copy_source,
                            Bucket=dest_bucket,
                            Key=dest_key
                        )
                        
                        total_size += obj['Size']
                        copied_count += 1
                        
                        if copied_count % 100 == 0:
                            logger.debug(f"Copied {copied_count} objects...")
                    
                    except ClientError as e:
                        logger.warning(f"Failed to copy {source_key}: {e}")
                        continue
            
            logger.info(f"Copied {copied_count} objects ({total_size:,} bytes)")
            return total_size
        
        except ClientError as e:
            raise BackupError(f"Failed to copy objects: {e}") from e
    
    def _save_backup_metadata(
        self,
        bucket: str,
        backup_key: str,
        backup_result: BackupResult
    ) -> None:
        """Save backup metadata.
        
        Args:
            bucket: S3 bucket
            backup_key: Backup key prefix
            backup_result: Backup result to save
        """
        try:
            metadata_key = f"{backup_key}backup-metadata.json"
            self.s3.put_object(
                Bucket=bucket,
                Key=metadata_key,
                Body=json.dumps(backup_result.to_dict(), indent=2),
                ContentType='application/json'
            )
        except ClientError as e:
            logger.warning(f"Failed to save backup metadata: {e}")
    
    def restore_backup(
        self,
        backup_id: str,
        destination_bucket: str,
        destination_prefix: str = "",
        source_backup_bucket: Optional[str] = None
    ) -> Dict:
        """Restore from backup.
        
        Args:
            backup_id: Backup ID to restore
            destination_bucket: Destination bucket for restore
            destination_prefix: Destination prefix
            source_backup_bucket: Source backup bucket (if different from config)
            
        Returns:
            Restore result dictionary
            
        Raises:
            BackupNotFoundError: If backup not found
        """
        try:
            # Find backup metadata
            backup_result = self._find_backup(backup_id, source_backup_bucket)
            
            if not backup_result:
                raise BackupNotFoundError(f"Backup {backup_id} not found")
            
            logger.info(f"Restoring backup {backup_id} to {destination_bucket}/{destination_prefix}")
            
            # Restore objects
            restored_count = 0
            source_bucket = backup_result.metadata['config']['destination_bucket']
            backup_key = backup_result.metadata['backup_key']
            
            paginator = self.s3.get_paginator('list_objects_v2')
            pages = paginator.paginate(Bucket=source_bucket, Prefix=backup_key)
            
            for page in pages:
                if 'Contents' not in page:
                    continue
                
                for obj in page['Contents']:
                    # Skip metadata file
                    if obj['Key'].endswith('backup-metadata.json'):
                        continue
                    
                    source_key = obj['Key']
                    # Remove backup prefix to get original key
                    original_key = source_key.replace(backup_key, '')
                    dest_key = f"{destination_prefix}{original_key}"
                    
                    try:
                        copy_source = {'Bucket': source_bucket, 'Key': source_key}
                        self.s3.copy_object(
                            CopySource=copy_source,
                            Bucket=destination_bucket,
                            Key=dest_key
                        )
                        restored_count += 1
                    except ClientError as e:
                        logger.warning(f"Failed to restore {source_key}: {e}")
            
            logger.info(f"Restored {restored_count} objects from backup {backup_id}")
            
            return {
                'backup_id': backup_id,
                'restored_count': restored_count,
                'destination': f"{destination_bucket}/{destination_prefix}"
            }
        
        except BackupNotFoundError:
            raise
        except Exception as e:
            error_msg = f"Failed to restore backup: {e}"
            logger.error(error_msg, exc_info=True)
            raise BackupError(error_msg) from e
    
    def cleanup_old_backups(
        self,
        destination_bucket: str,
        retention_days: int = 30
    ) -> Dict:
        """Cleanup backups older than retention period.
        
        Args:
            destination_bucket: Backup bucket
            retention_days: Retention period in days
            
        Returns:
            Cleanup result dictionary
        """
        try:
            cutoff_date = datetime.utcnow() - timedelta(days=retention_days)
            deleted_count = 0
            deleted_size = 0
            
            paginator = self.s3.get_paginator('list_objects_v2')
            pages = paginator.paginate(Bucket=destination_bucket, Prefix='backups/')
            
            for page in pages:
                if 'Contents' not in page:
                    continue
                
                for obj in page['Contents']:
                    if obj['LastModified'].replace(tzinfo=None) < cutoff_date:
                        try:
                            self.s3.delete_object(Bucket=destination_bucket, Key=obj['Key'])
                            deleted_count += 1
                            deleted_size += obj['Size']
                        except ClientError as e:
                            logger.warning(f"Failed to delete {obj['Key']}: {e}")
            
            logger.info(
                f"Cleaned up {deleted_count} backup objects "
                f"({deleted_size:,} bytes) older than {retention_days} days"
            )
            
            return {
                'deleted_count': deleted_count,
                'deleted_size_bytes': deleted_size,
                'cutoff_date': cutoff_date.isoformat()
            }
        
        except ClientError as e:
            raise BackupError(f"Failed to cleanup old backups: {e}") from e
    
    def _find_backup(
        self,
        backup_id: str,
        backup_bucket: Optional[str] = None
    ) -> Optional[BackupResult]:
        """Find backup by ID.
        
        Args:
            backup_id: Backup ID to find
            backup_bucket: Optional backup bucket to search
            
        Returns:
            BackupResult if found, None otherwise
        """
        # Search in backup history first
        for backup in self.backup_history:
            if backup.backup_id == backup_id:
                return backup
        
        # Search in S3 if bucket provided
        if backup_bucket:
            try:
                metadata_key = f"backups/{backup_id}/backup-metadata.json"
                response = self.s3.get_object(Bucket=backup_bucket, Key=metadata_key)
                metadata = json.loads(response['Body'].read())
                return BackupResult(**metadata)
            except ClientError:
                pass
        
        return None


# Example usage
if __name__ == "__main__":
    manager = BackupManager(region_name='us-east-1')
    
    # Configure backup
    backup_config = BackupConfig(
        source_bucket='production-data',
        destination_bucket='backup-data',
        prefix='backups',
        retention_days=30,
        encryption=True,
        versioning=True
    )
    
    # Create backup
    result = manager.create_backup(backup_config)
    print(f"Backup completed: {result.backup_id}")
    print(f"Size: {result.size_bytes:,} bytes")
    print(f"Destination: {result.destination}")
    
    # Cleanup old backups
    cleanup_result = manager.cleanup_old_backups(
        destination_bucket='backup-data',
        retention_days=30
    )
    print(f"Cleaned up {cleanup_result['deleted_count']} old backups")

Complete Disaster Recovery Plan Manager:

#!/usr/bin/env python3
"""
Cloud Backup & Disaster Recovery - Disaster Recovery Plan Manager
Production-ready DR plan management with automated recovery procedures
"""

from typing import Dict, List, Optional
from dataclasses import dataclass, asdict, field
from enum import Enum
from datetime import datetime, timedelta
import logging
import json

logger = logging.getLogger(__name__)


class DRPlanError(Exception):
    """Base exception for DR plan errors."""
    pass


@dataclass
class DRMetrics:
    """Disaster Recovery metrics."""
    rpo_hours: float  # Recovery Point Objective
    rto_hours: float  # Recovery Time Objective
    mttr_hours: float  # Mean Time To Recovery
    
    def to_dict(self) -> Dict:
        """Convert to dictionary."""
        return asdict(self)


@dataclass
class BackupStrategy:
    """Backup strategy configuration."""
    frequency: str  # hourly, daily, weekly
    retention_days: int
    locations: List[str]  # Regions
    encryption: bool = True
    replication: bool = True
    
    def to_dict(self) -> Dict:
        """Convert to dictionary."""
        return asdict(self)


@dataclass
class RecoveryProcedure:
    """Recovery procedure step."""
    step_number: int
    description: str
    command: Optional[str] = None
    validation: Optional[str] = None
    estimated_duration_minutes: int = 15
    
    def to_dict(self) -> Dict:
        """Convert to dictionary."""
        return asdict(self)


@dataclass
class DisasterRecoveryPlan:
    """Complete disaster recovery plan."""
    name: str
    description: str
    metrics: DRMetrics
    backup_strategy: BackupStrategy
    recovery_procedures: List[RecoveryProcedure]
    contacts: List[Dict[str, str]] = field(default_factory=list)
    last_updated: datetime = field(default_factory=datetime.utcnow)
    
    def to_dict(self) -> Dict:
        """Convert to dictionary."""
        result = asdict(self)
        result['last_updated'] = self.last_updated.isoformat()
        result['recovery_procedures'] = [rp.to_dict() for rp in self.recovery_procedures]
        return result
    
    def to_yaml(self) -> str:
        """Convert to YAML string."""
        import yaml
        return yaml.dump(self.to_dict(), default_flow_style=False)


class DRPlanManager:
    """Manages disaster recovery plans."""
    
    def __init__(self):
        """Initialize DR plan manager."""
        self.plans: Dict[str, DisasterRecoveryPlan] = {}
    
    def create_dr_plan(self, plan: DisasterRecoveryPlan) -> None:
        """Create or update DR plan.
        
        Args:
            plan: Disaster recovery plan
        """
        self.plans[plan.name] = plan
        logger.info(f"Created/updated DR plan: {plan.name}")
    
    def get_dr_plan(self, name: str) -> Optional[DisasterRecoveryPlan]:
        """Get DR plan by name.
        
        Args:
            name: Plan name
            
        Returns:
            DisasterRecoveryPlan if found, None otherwise
        """
        return self.plans.get(name)
    
    def list_dr_plans(self) -> List[str]:
        """List all DR plan names.
        
        Returns:
            List of plan names
        """
        return list(self.plans.keys())


# Example DR Plan
def create_example_dr_plan() -> DisasterRecoveryPlan:
    """Create example disaster recovery plan."""
    metrics = DRMetrics(
        rpo_hours=1.0,
        rto_hours=4.0,
        mttr_hours=3.5
    )
    
    backup_strategy = BackupStrategy(
        frequency="hourly",
        retention_days=30,
        locations=["us-east-1", "us-west-2"],
        encryption=True,
        replication=True
    )
    
    recovery_procedures = [
        RecoveryProcedure(
            step_number=1,
            description="Assess damage and identify affected systems",
            command="aws s3 ls s3://backup-data/backups/ | tail -5",
            estimated_duration_minutes=15
        ),
        RecoveryProcedure(
            step_number=2,
            description="Restore from latest backup",
            command="python restore_backup.py --backup-id <BACKUP_ID>",
            validation="Verify restored data integrity",
            estimated_duration_minutes=60
        ),
        RecoveryProcedure(
            step_number=3,
            description="Validate data integrity",
            command="python validate_backup.py --backup-id <BACKUP_ID>",
            estimated_duration_minutes=30
        ),
        RecoveryProcedure(
            step_number=4,
            description="Resume operations and notify stakeholders",
            command="notify_team.sh --status restored",
            estimated_duration_minutes=15
        )
    ]
    
    contacts = [
        {"role": "Incident Commander", "name": "John Doe", "email": "john@example.com", "phone": "+1-555-0100"},
        {"role": "Backup Admin", "name": "Jane Smith", "email": "jane@example.com", "phone": "+1-555-0101"}
    ]
    
    return DisasterRecoveryPlan(
        name="production-dr-plan",
        description="Disaster recovery plan for production environment",
        metrics=metrics,
        backup_strategy=backup_strategy,
        recovery_procedures=recovery_procedures,
        contacts=contacts
    )


# Example usage
if __name__ == "__main__":
    manager = DRPlanManager()
    
    # Create DR plan
    dr_plan = create_example_dr_plan()
    manager.create_dr_plan(dr_plan)
    
    # Save to file
    with open('dr-plan.json', 'w') as f:
        json.dump(dr_plan.to_dict(), f, indent=2)
    
    print(f"Created DR plan: {dr_plan.name}")
    print(f"RPO: {dr_plan.metrics.rpo_hours} hours")
    print(f"RTO: {dr_plan.metrics.rto_hours} hours")
    print(f"Recovery steps: {len(dr_plan.recovery_procedures)}")

YAML DR Plan Example:

disaster_recovery_plan:
  name: "production-dr-plan"
  description: "Disaster recovery plan for production environment"
  
  metrics:
    rpo_hours: 1.0  # Recovery Point Objective: 1 hour of data loss acceptable
    rto_hours: 4.0  # Recovery Time Objective: 4 hours to restore
    mttr_hours: 3.5  # Mean Time To Recovery: 3.5 hours average
  
  backup_strategy:
    frequency: "hourly"
    retention: "30 days"
    locations:
      - "us-east-1"  # Primary region
      - "us-west-2"  # Secondary region
    encryption: true
    replication: true
  
  recovery_procedures:
    - step: 1
      description: "Assess damage and identify affected systems"
      command: "aws s3 ls s3://backup-data/backups/ | tail -5"
      estimated_duration_minutes: 15
    
    - step: 2
      description: "Restore from latest backup"
      command: "python restore_backup.py --backup-id <BACKUP_ID>"
      validation: "Verify restored data integrity"
      estimated_duration_minutes: 60
    
    - step: 3
      description: "Validate data integrity"
      command: "python validate_backup.py --backup-id <BACKUP_ID>"
      estimated_duration_minutes: 30
    
    - step: 4
      description: "Resume operations and notify stakeholders"
      command: "notify_team.sh --status restored"
      estimated_duration_minutes: 15
  
  contacts:
    - role: "Incident Commander"
      name: "John Doe"
      email: "john@example.com"
      phone: "+1-555-0100"
    - role: "Backup Admin"
      name: "Jane Smith"
      email: "jane@example.com"
      phone: "+1-555-0101"