diff --git a/01_advanced_python_language_features.md b/01_advanced_python_language_features.md new file mode 100644 index 0000000..48611a2 --- /dev/null +++ b/01_advanced_python_language_features.md @@ -0,0 +1,1234 @@ +# Advanced Python Language Features for DevOps Engineers + +## Table of Contents +1. [Object-Oriented Programming (OOP)](#object-oriented-programming-oop) +2. [Functional Programming](#functional-programming) +3. [Concurrency and Parallelism](#concurrency-and-parallelism) +4. [Error Handling and Debugging](#error-handling-and-debugging) + +## Object-Oriented Programming (OOP) + +### Classes, Inheritance, Polymorphism, Encapsulation + +**DevOps Usage Scenario**: Creating reusable infrastructure components and cloud resource managers. + +```python +from abc import ABC, abstractmethod +from typing import Dict, List, Optional +import boto3 +import logging + +class CloudResource(ABC): + """Abstract base class for cloud resources""" + + def __init__(self, name: str, region: str, tags: Dict[str, str] = None): + self._name = name + self._region = region + self._tags = tags or {} + self._logger = logging.getLogger(self.__class__.__name__) + + @property + def name(self) -> str: + return self._name + + @property + def region(self) -> str: + return self._region + + @abstractmethod + def create(self) -> bool: + """Create the cloud resource""" + pass + + @abstractmethod + def delete(self) -> bool: + """Delete the cloud resource""" + pass + + @abstractmethod + def get_status(self) -> str: + """Get current status of the resource""" + pass + + def add_tags(self, tags: Dict[str, str]) -> None: + """Add tags to the resource""" + self._tags.update(tags) + +class EC2Instance(CloudResource): + """EC2 Instance implementation""" + + def __init__(self, name: str, region: str, instance_type: str = 't2.micro', + ami_id: str = 'ami-0abcdef1234567890', tags: Dict[str, str] = None): + super().__init__(name, region, tags) + self.instance_type = instance_type + self.ami_id = ami_id + self.instance_id: Optional[str] = None + self._ec2_client = boto3.client('ec2', region_name=region) + + def create(self) -> bool: + """Create EC2 instance""" + try: + response = self._ec2_client.run_instances( + ImageId=self.ami_id, + MinCount=1, + MaxCount=1, + InstanceType=self.instance_type, + TagSpecifications=[ + { + 'ResourceType': 'instance', + 'Tags': [{'Key': k, 'Value': v} for k, v in self._tags.items()] + } + ] + ) + self.instance_id = response['Instances'][0]['InstanceId'] + self._logger.info(f"Created EC2 instance: {self.instance_id}") + return True + except Exception as e: + self._logger.error(f"Failed to create EC2 instance: {e}") + return False + + def delete(self) -> bool: + """Terminate EC2 instance""" + if not self.instance_id: + self._logger.warning("No instance ID available for deletion") + return False + + try: + self._ec2_client.terminate_instances(InstanceIds=[self.instance_id]) + self._logger.info(f"Terminated EC2 instance: {self.instance_id}") + return True + except Exception as e: + self._logger.error(f"Failed to terminate EC2 instance: {e}") + return False + + def get_status(self) -> str: + """Get instance status""" + if not self.instance_id: + return "not_created" + + try: + response = self._ec2_client.describe_instances(InstanceIds=[self.instance_id]) + return response['Reservations'][0]['Instances'][0]['State']['Name'] + except Exception as e: + self._logger.error(f"Failed to get instance status: {e}") + return "unknown" + +class RDSInstance(CloudResource): + """RDS Instance implementation""" + + def __init__(self, name: str, region: str, db_instance_class: str = 'db.t3.micro', + engine: str = 'mysql', tags: Dict[str, str] = None): + super().__init__(name, region, tags) + self.db_instance_class = db_instance_class + self.engine = engine + self._rds_client = boto3.client('rds', region_name=region) + + def create(self) -> bool: + """Create RDS instance""" + try: + self._rds_client.create_db_instance( + DBInstanceIdentifier=self._name, + DBInstanceClass=self.db_instance_class, + Engine=self.engine, + MasterUsername='admin', + MasterUserPassword='temp_password_123', + AllocatedStorage=20, + Tags=[{'Key': k, 'Value': v} for k, v in self._tags.items()] + ) + self._logger.info(f"Created RDS instance: {self._name}") + return True + except Exception as e: + self._logger.error(f"Failed to create RDS instance: {e}") + return False + + def delete(self) -> bool: + """Delete RDS instance""" + try: + self._rds_client.delete_db_instance( + DBInstanceIdentifier=self._name, + SkipFinalSnapshot=True + ) + self._logger.info(f"Deleted RDS instance: {self._name}") + return True + except Exception as e: + self._logger.error(f"Failed to delete RDS instance: {e}") + return False + + def get_status(self) -> str: + """Get RDS instance status""" + try: + response = self._rds_client.describe_db_instances( + DBInstanceIdentifier=self._name + ) + return response['DBInstances'][0]['DBInstanceStatus'] + except Exception as e: + self._logger.error(f"Failed to get RDS status: {e}") + return "unknown" + +# Infrastructure Manager using polymorphism +class InfrastructureManager: + """Manages multiple cloud resources""" + + def __init__(self): + self.resources: List[CloudResource] = [] + + def add_resource(self, resource: CloudResource) -> None: + """Add a resource to management""" + self.resources.append(resource) + + def deploy_all(self) -> Dict[str, bool]: + """Deploy all resources""" + results = {} + for resource in self.resources: + results[resource.name] = resource.create() + return results + + def destroy_all(self) -> Dict[str, bool]: + """Destroy all resources""" + results = {} + for resource in self.resources: + results[resource.name] = resource.delete() + return results + + def get_status_all(self) -> Dict[str, str]: + """Get status of all resources""" + status = {} + for resource in self.resources: + status[resource.name] = resource.get_status() + return status + +# Usage example +def main(): + # Create infrastructure manager + infra_manager = InfrastructureManager() + + # Add resources + web_server = EC2Instance( + name="web-server-01", + region="us-east-1", + instance_type="t3.small", + tags={"Environment": "production", "Team": "web"} + ) + + database = RDSInstance( + name="app-database", + region="us-east-1", + db_instance_class="db.t3.small", + engine="postgresql", + tags={"Environment": "production", "Team": "database"} + ) + + infra_manager.add_resource(web_server) + infra_manager.add_resource(database) + + # Deploy infrastructure + deployment_results = infra_manager.deploy_all() + print("Deployment results:", deployment_results) + + # Check status + status = infra_manager.get_status_all() + print("Resource status:", status) + +if __name__ == "__main__": + main() +``` + +### Design Patterns for DevOps + +**Singleton Pattern for Configuration Management**: + +```python +import threading +import yaml +from typing import Dict, Any + +class ConfigurationManager: + """Singleton configuration manager for DevOps applications""" + + _instance = None + _lock = threading.Lock() + + def __new__(cls): + if cls._instance is None: + with cls._lock: + if cls._instance is None: + cls._instance = super().__new__(cls) + cls._instance._initialized = False + return cls._instance + + def __init__(self): + if not self._initialized: + self._config: Dict[str, Any] = {} + self._config_file: str = "config.yaml" + self._load_config() + self._initialized = True + + def _load_config(self) -> None: + """Load configuration from file""" + try: + with open(self._config_file, 'r') as file: + self._config = yaml.safe_load(file) or {} + except FileNotFoundError: + self._config = {} + + def get(self, key: str, default: Any = None) -> Any: + """Get configuration value""" + keys = key.split('.') + value = self._config + for k in keys: + if isinstance(value, dict) and k in value: + value = value[k] + else: + return default + return value + + def set(self, key: str, value: Any) -> None: + """Set configuration value""" + keys = key.split('.') + config = self._config + for k in keys[:-1]: + if k not in config: + config[k] = {} + config = config[k] + config[keys[-1]] = value + + def reload(self) -> None: + """Reload configuration from file""" + self._load_config() + +# Factory Pattern for Cloud Providers +class CloudProviderFactory: + """Factory for creating cloud provider clients""" + + @staticmethod + def create_provider(provider: str, region: str = None): + """Create cloud provider client""" + if provider.lower() == 'aws': + return AWSProvider(region) + elif provider.lower() == 'azure': + return AzureProvider(region) + elif provider.lower() == 'gcp': + return GCPProvider(region) + else: + raise ValueError(f"Unsupported cloud provider: {provider}") + +class CloudProvider(ABC): + """Abstract cloud provider""" + + @abstractmethod + def create_vm(self, **kwargs) -> str: + pass + + @abstractmethod + def delete_vm(self, vm_id: str) -> bool: + pass + +class AWSProvider(CloudProvider): + def __init__(self, region: str = 'us-east-1'): + self.region = region + self.ec2 = boto3.client('ec2', region_name=region) + + def create_vm(self, **kwargs) -> str: + # AWS-specific VM creation logic + pass + + def delete_vm(self, vm_id: str) -> bool: + # AWS-specific VM deletion logic + pass + +# Observer Pattern for Monitoring +class MonitoringSubject: + """Subject for monitoring events""" + + def __init__(self): + self._observers = [] + self._state = {} + + def attach(self, observer): + self._observers.append(observer) + + def detach(self, observer): + self._observers.remove(observer) + + def notify(self, event_type: str, data: Dict[str, Any]): + for observer in self._observers: + observer.update(event_type, data) + + def set_state(self, key: str, value: Any): + self._state[key] = value + self.notify('state_change', {'key': key, 'value': value}) + +class AlertObserver: + """Observer for sending alerts""" + + def update(self, event_type: str, data: Dict[str, Any]): + if event_type == 'state_change': + if data['key'] == 'cpu_usage' and data['value'] > 80: + self.send_alert(f"High CPU usage: {data['value']}%") + + def send_alert(self, message: str): + print(f"ALERT: {message}") + +class LoggingObserver: + """Observer for logging events""" + + def update(self, event_type: str, data: Dict[str, Any]): + print(f"LOG: {event_type} - {data}") +``` + +## Functional Programming + +### Decorators and Context Managers + +**DevOps Usage Scenario**: Adding monitoring, logging, and error handling to infrastructure operations. + +```python +import functools +import time +import logging +from contextlib import contextmanager +from typing import Callable, Any, Generator +import boto3 +from botocore.exceptions import ClientError + +# Timing decorator for performance monitoring +def monitor_execution_time(func: Callable) -> Callable: + """Decorator to monitor function execution time""" + @functools.wraps(func) + def wrapper(*args, **kwargs): + start_time = time.time() + try: + result = func(*args, **kwargs) + execution_time = time.time() - start_time + logging.info(f"{func.__name__} executed in {execution_time:.2f} seconds") + return result + except Exception as e: + execution_time = time.time() - start_time + logging.error(f"{func.__name__} failed after {execution_time:.2f} seconds: {e}") + raise + return wrapper + +# Retry decorator for handling transient failures +def retry(max_attempts: int = 3, delay: float = 1.0, backoff: float = 2.0): + """Decorator for retrying functions with exponential backoff""" + def decorator(func: Callable) -> Callable: + @functools.wraps(func) + def wrapper(*args, **kwargs): + attempts = 0 + current_delay = delay + + while attempts < max_attempts: + try: + return func(*args, **kwargs) + except Exception as e: + attempts += 1 + if attempts == max_attempts: + logging.error(f"{func.__name__} failed after {max_attempts} attempts") + raise e + + logging.warning(f"{func.__name__} attempt {attempts} failed: {e}. Retrying in {current_delay}s...") + time.sleep(current_delay) + current_delay *= backoff + + return wrapper + return decorator + +# Authentication decorator +def require_aws_credentials(func: Callable) -> Callable: + """Decorator to ensure AWS credentials are available""" + @functools.wraps(func) + def wrapper(*args, **kwargs): + try: + # Test AWS credentials + sts = boto3.client('sts') + sts.get_caller_identity() + return func(*args, **kwargs) + except Exception as e: + logging.error(f"AWS credentials not available: {e}") + raise ValueError("Valid AWS credentials required") + return wrapper + +# Context manager for AWS resource management +@contextmanager +def aws_session(region: str = 'us-east-1') -> Generator[boto3.Session, None, None]: + """Context manager for AWS session""" + session = boto3.Session(region_name=region) + try: + logging.info(f"Created AWS session for region: {region}") + yield session + finally: + logging.info("AWS session closed") + +@contextmanager +def temporary_security_group(ec2_client, group_name: str, vpc_id: str) -> Generator[str, None, None]: + """Context manager for temporary security group""" + group_id = None + try: + # Create security group + response = ec2_client.create_security_group( + GroupName=group_name, + Description="Temporary security group", + VpcId=vpc_id + ) + group_id = response['GroupId'] + logging.info(f"Created temporary security group: {group_id}") + yield group_id + finally: + # Clean up security group + if group_id: + try: + ec2_client.delete_security_group(GroupId=group_id) + logging.info(f"Deleted temporary security group: {group_id}") + except Exception as e: + logging.error(f"Failed to delete security group {group_id}: {e}") + +# Practical usage example +class InfrastructureDeployer: + def __init__(self, region: str = 'us-east-1'): + self.region = region + + @monitor_execution_time + @retry(max_attempts=3, delay=2.0) + @require_aws_credentials + def deploy_instance(self, instance_config: dict) -> str: + """Deploy EC2 instance with monitoring and retry logic""" + with aws_session(self.region) as session: + ec2 = session.client('ec2') + + response = ec2.run_instances( + ImageId=instance_config['ami_id'], + MinCount=1, + MaxCount=1, + InstanceType=instance_config['instance_type'], + KeyName=instance_config.get('key_name'), + SecurityGroupIds=instance_config.get('security_groups', []) + ) + + instance_id = response['Instances'][0]['InstanceId'] + logging.info(f"Successfully deployed instance: {instance_id}") + return instance_id + + @monitor_execution_time + def setup_networking(self, vpc_id: str) -> dict: + """Setup networking with temporary resources""" + with aws_session(self.region) as session: + ec2 = session.client('ec2') + + with temporary_security_group(ec2, "temp-sg", vpc_id) as sg_id: + # Add rules to security group + ec2.authorize_security_group_ingress( + GroupId=sg_id, + IpPermissions=[ + { + 'IpProtocol': 'tcp', + 'FromPort': 80, + 'ToPort': 80, + 'IpRanges': [{'CidrIp': '0.0.0.0/0'}] + } + ] + ) + + return {'security_group_id': sg_id, 'vpc_id': vpc_id} + +# Higher-order functions for pipeline processing +def create_pipeline(*functions): + """Create a processing pipeline from functions""" + def pipeline(data): + result = data + for func in functions: + result = func(result) + return result + return pipeline + +def validate_config(config: dict) -> dict: + """Validate configuration""" + required_fields = ['ami_id', 'instance_type'] + for field in required_fields: + if field not in config: + raise ValueError(f"Missing required field: {field}") + return config + +def add_default_tags(config: dict) -> dict: + """Add default tags to configuration""" + default_tags = { + 'Environment': 'development', + 'Team': 'devops', + 'CreatedBy': 'automation' + } + config.setdefault('tags', {}).update(default_tags) + return config + +def normalize_instance_type(config: dict) -> dict: + """Normalize instance type""" + instance_type_map = { + 'small': 't3.small', + 'medium': 't3.medium', + 'large': 't3.large' + } + if config['instance_type'] in instance_type_map: + config['instance_type'] = instance_type_map[config['instance_type']] + return config + +# Create deployment pipeline +deployment_pipeline = create_pipeline( + validate_config, + add_default_tags, + normalize_instance_type +) + +# Usage example +def main(): + deployer = InfrastructureDeployer('us-west-2') + + # Raw configuration + config = { + 'ami_id': 'ami-0abcdef1234567890', + 'instance_type': 'small', + 'key_name': 'my-key-pair' + } + + # Process through pipeline + processed_config = deployment_pipeline(config) + print("Processed config:", processed_config) + + # Deploy instance + try: + instance_id = deployer.deploy_instance(processed_config) + print(f"Deployed instance: {instance_id}") + except Exception as e: + print(f"Deployment failed: {e}") + +if __name__ == "__main__": + main() +``` + +## Concurrency and Parallelism + +### Asyncio for DevOps Operations + +**DevOps Usage Scenario**: Managing multiple cloud resources concurrently for faster deployment and monitoring. + +```python +import asyncio +import aiohttp +import time +from typing import List, Dict, Any +from concurrent.futures import ThreadPoolExecutor, as_completed +import boto3 +from dataclasses import dataclass + +@dataclass +class DeploymentTask: + name: str + region: str + instance_type: str + ami_id: str + +class AsyncCloudManager: + def __init__(self, max_concurrent: int = 10): + self.max_concurrent = max_concurrent + self.semaphore = asyncio.Semaphore(max_concurrent) + + async def deploy_instance_async(self, task: DeploymentTask) -> Dict[str, Any]: + """Deploy EC2 instance asynchronously""" + async with self.semaphore: + try: + # Simulate async AWS API call (in reality, you'd use aioboto3) + await asyncio.sleep(2) # Simulate deployment time + + # Create instance (simplified) + instance_id = f"i-{task.name}-{int(time.time())}" + + return { + 'name': task.name, + 'instance_id': instance_id, + 'status': 'success', + 'region': task.region + } + except Exception as e: + return { + 'name': task.name, + 'status': 'failed', + 'error': str(e) + } + + async def deploy_multiple_instances(self, tasks: List[DeploymentTask]) -> List[Dict[str, Any]]: + """Deploy multiple instances concurrently""" + deployment_tasks = [ + self.deploy_instance_async(task) for task in tasks + ] + + results = await asyncio.gather(*deployment_tasks, return_exceptions=True) + return [r for r in results if not isinstance(r, Exception)] + + async def health_check(self, url: str) -> Dict[str, Any]: + """Perform health check on a service""" + async with aiohttp.ClientSession() as session: + try: + async with session.get(url, timeout=10) as response: + return { + 'url': url, + 'status_code': response.status, + 'response_time': time.time(), + 'healthy': response.status == 200 + } + except Exception as e: + return { + 'url': url, + 'error': str(e), + 'healthy': False + } + + async def monitor_services(self, urls: List[str]) -> List[Dict[str, Any]]: + """Monitor multiple services concurrently""" + health_tasks = [self.health_check(url) for url in urls] + results = await asyncio.gather(*health_tasks, return_exceptions=True) + return [r for r in results if not isinstance(r, Exception)] + +# Thread-based parallel processing for CPU-intensive tasks +class ParallelProcessor: + def __init__(self, max_workers: int = 4): + self.max_workers = max_workers + + def process_logs_parallel(self, log_files: List[str]) -> Dict[str, Any]: + """Process multiple log files in parallel""" + def process_single_log(log_file: str) -> Dict[str, Any]: + # Simulate log processing + time.sleep(1) # CPU-intensive work + return { + 'file': log_file, + 'lines_processed': 1000, + 'errors_found': 5, + 'warnings_found': 20 + } + + results = {} + with ThreadPoolExecutor(max_workers=self.max_workers) as executor: + # Submit all tasks + future_to_file = { + executor.submit(process_single_log, log_file): log_file + for log_file in log_files + } + + # Collect results as they complete + for future in as_completed(future_to_file): + log_file = future_to_file[future] + try: + result = future.result() + results[log_file] = result + except Exception as e: + results[log_file] = {'error': str(e)} + + return results + + def backup_databases_parallel(self, databases: List[str]) -> Dict[str, bool]: + """Backup multiple databases in parallel""" + def backup_database(db_name: str) -> bool: + # Simulate database backup + print(f"Starting backup for {db_name}") + time.sleep(3) # Simulate backup time + print(f"Completed backup for {db_name}") + return True + + results = {} + with ThreadPoolExecutor(max_workers=self.max_workers) as executor: + future_to_db = { + executor.submit(backup_database, db): db + for db in databases + } + + for future in as_completed(future_to_db): + db_name = future_to_db[future] + try: + success = future.result() + results[db_name] = success + except Exception as e: + results[db_name] = False + print(f"Backup failed for {db_name}: {e}") + + return results + +# Advanced async patterns for DevOps +class AsyncDevOpsOrchestrator: + def __init__(self): + self.cloud_manager = AsyncCloudManager() + self.processor = ParallelProcessor() + + async def rolling_deployment(self, instances: List[str], new_ami: str) -> Dict[str, Any]: + """Perform rolling deployment with health checks""" + results = {'updated': [], 'failed': [], 'total_time': 0} + start_time = time.time() + + for instance in instances: + try: + # Update instance + print(f"Updating instance {instance} with AMI {new_ami}") + await asyncio.sleep(5) # Simulate instance update + + # Health check after update + health_result = await self.cloud_manager.health_check( + f"http://{instance}.example.com/health" + ) + + if health_result.get('healthy', False): + results['updated'].append(instance) + print(f"Successfully updated {instance}") + else: + results['failed'].append(instance) + print(f"Health check failed for {instance}") + break # Stop rolling deployment on failure + + except Exception as e: + results['failed'].append(instance) + print(f"Failed to update {instance}: {e}") + break + + results['total_time'] = time.time() - start_time + return results + + async def disaster_recovery_drill(self, backup_regions: List[str]) -> Dict[str, Any]: + """Simulate disaster recovery across multiple regions""" + recovery_tasks = [] + + for region in backup_regions: + task = asyncio.create_task(self._restore_region(region)) + recovery_tasks.append(task) + + results = await asyncio.gather(*recovery_tasks, return_exceptions=True) + + recovery_summary = { + 'successful_regions': [], + 'failed_regions': [], + 'total_recovery_time': 0 + } + + for i, result in enumerate(results): + if isinstance(result, Exception): + recovery_summary['failed_regions'].append(backup_regions[i]) + else: + recovery_summary['successful_regions'].append(backup_regions[i]) + + return recovery_summary + + async def _restore_region(self, region: str) -> Dict[str, Any]: + """Restore services in a specific region""" + print(f"Starting recovery in region {region}") + + # Simulate recovery steps + await asyncio.sleep(10) # Database restore + await asyncio.sleep(5) # Application deployment + await asyncio.sleep(3) # Configuration sync + + print(f"Recovery completed in region {region}") + return {'region': region, 'status': 'recovered'} + +# Usage examples +async def main(): + orchestrator = AsyncDevOpsOrchestrator() + + # Example 1: Parallel instance deployment + deployment_tasks = [ + DeploymentTask("web-01", "us-east-1", "t3.small", "ami-12345"), + DeploymentTask("web-02", "us-east-1", "t3.small", "ami-12345"), + DeploymentTask("api-01", "us-west-2", "t3.medium", "ami-67890"), + DeploymentTask("api-02", "us-west-2", "t3.medium", "ami-67890"), + ] + + print("Starting parallel deployment...") + deployment_results = await orchestrator.cloud_manager.deploy_multiple_instances(deployment_tasks) + print("Deployment results:", deployment_results) + + # Example 2: Service health monitoring + service_urls = [ + "http://web-01.example.com/health", + "http://web-02.example.com/health", + "http://api-01.example.com/health", + "http://api-02.example.com/health", + ] + + print("\nStarting health checks...") + health_results = await orchestrator.cloud_manager.monitor_services(service_urls) + print("Health check results:", health_results) + + # Example 3: Rolling deployment + instances = ["web-01", "web-02", "api-01", "api-02"] + print("\nStarting rolling deployment...") + rolling_results = await orchestrator.rolling_deployment(instances, "ami-new-version") + print("Rolling deployment results:", rolling_results) + +def sync_processing_example(): + """Example of synchronous parallel processing""" + processor = ParallelProcessor() + + # Process log files in parallel + log_files = [f"app-{i}.log" for i in range(1, 6)] + print("Processing log files in parallel...") + log_results = processor.process_logs_parallel(log_files) + print("Log processing results:", log_results) + + # Backup databases in parallel + databases = ["user_db", "inventory_db", "analytics_db", "logs_db"] + print("\nBacking up databases in parallel...") + backup_results = processor.backup_databases_parallel(databases) + print("Backup results:", backup_results) + +if __name__ == "__main__": + # Run async examples + asyncio.run(main()) + + # Run sync parallel processing examples + sync_processing_example() +``` + +## Error Handling and Debugging + +### Advanced Error Handling for DevOps + +**DevOps Usage Scenario**: Robust error handling for infrastructure operations with proper logging and recovery mechanisms. + +```python +import logging +import traceback +import functools +import sys +from typing import Optional, Dict, Any, Callable +from enum import Enum +import json +from datetime import datetime + +# Custom exception hierarchy for DevOps operations +class DevOpsException(Exception): + """Base exception for DevOps operations""" + def __init__(self, message: str, error_code: str = None, details: Dict[str, Any] = None): + super().__init__(message) + self.error_code = error_code or self.__class__.__name__ + self.details = details or {} + self.timestamp = datetime.utcnow() + +class InfrastructureException(DevOpsException): + """Infrastructure-related exceptions""" + pass + +class DeploymentException(DevOpsException): + """Deployment-related exceptions""" + pass + +class ConfigurationException(DevOpsException): + """Configuration-related exceptions""" + pass + +class MonitoringException(DevOpsException): + """Monitoring-related exceptions""" + pass + +# Error severity levels +class ErrorSeverity(Enum): + LOW = "low" + MEDIUM = "medium" + HIGH = "high" + CRITICAL = "critical" + +# Advanced logging configuration +class DevOpsLogger: + def __init__(self, name: str, log_file: str = "devops.log"): + self.logger = logging.getLogger(name) + self.logger.setLevel(logging.DEBUG) + + # Create formatters + detailed_formatter = logging.Formatter( + '%(asctime)s - %(name)s - %(levelname)s - %(funcName)s:%(lineno)d - %(message)s' + ) + + json_formatter = JsonFormatter() + + # File handler with rotation + from logging.handlers import RotatingFileHandler + file_handler = RotatingFileHandler( + log_file, maxBytes=10*1024*1024, backupCount=5 + ) + file_handler.setLevel(logging.DEBUG) + file_handler.setFormatter(json_formatter) + + # Console handler + console_handler = logging.StreamHandler(sys.stdout) + console_handler.setLevel(logging.INFO) + console_handler.setFormatter(detailed_formatter) + + # Add handlers + self.logger.addHandler(file_handler) + self.logger.addHandler(console_handler) + + def log_exception(self, exc: Exception, severity: ErrorSeverity = ErrorSeverity.MEDIUM, + context: Dict[str, Any] = None): + """Log exception with context""" + context = context or {} + + log_data = { + 'exception_type': type(exc).__name__, + 'exception_message': str(exc), + 'severity': severity.value, + 'context': context, + 'traceback': traceback.format_exc() + } + + if hasattr(exc, 'error_code'): + log_data['error_code'] = exc.error_code + if hasattr(exc, 'details'): + log_data['details'] = exc.details + + if severity in [ErrorSeverity.HIGH, ErrorSeverity.CRITICAL]: + self.logger.error(json.dumps(log_data)) + else: + self.logger.warning(json.dumps(log_data)) + +class JsonFormatter(logging.Formatter): + """JSON formatter for structured logging""" + def format(self, record): + log_data = { + 'timestamp': datetime.utcnow().isoformat(), + 'level': record.levelname, + 'logger': record.name, + 'message': record.getMessage(), + 'module': record.module, + 'function': record.funcName, + 'line': record.lineno + } + + if record.exc_info: + log_data['exception'] = self.formatException(record.exc_info) + + return json.dumps(log_data) + +# Error handling decorators +def handle_devops_errors(severity: ErrorSeverity = ErrorSeverity.MEDIUM, + reraise: bool = True): + """Decorator for handling DevOps errors""" + def decorator(func: Callable) -> Callable: + @functools.wraps(func) + def wrapper(*args, **kwargs): + logger = DevOpsLogger(func.__module__) + try: + return func(*args, **kwargs) + except DevOpsException as e: + logger.log_exception(e, severity, { + 'function': func.__name__, + 'args': str(args), + 'kwargs': str(kwargs) + }) + if reraise: + raise + return None + except Exception as e: + # Convert generic exceptions to DevOpsException + devops_exc = DevOpsException( + f"Unexpected error in {func.__name__}: {str(e)}", + error_code="UNEXPECTED_ERROR", + details={'original_exception': type(e).__name__} + ) + logger.log_exception(devops_exc, ErrorSeverity.HIGH, { + 'function': func.__name__, + 'args': str(args), + 'kwargs': str(kwargs) + }) + if reraise: + raise devops_exc + return None + return wrapper + return decorator + +# Circuit breaker pattern for external services +class CircuitBreaker: + def __init__(self, failure_threshold: int = 5, recovery_timeout: int = 60): + self.failure_threshold = failure_threshold + self.recovery_timeout = recovery_timeout + self.failure_count = 0 + self.last_failure_time = None + self.state = 'CLOSED' # CLOSED, OPEN, HALF_OPEN + + def call(self, func: Callable, *args, **kwargs): + """Call function with circuit breaker protection""" + if self.state == 'OPEN': + if self._should_attempt_reset(): + self.state = 'HALF_OPEN' + else: + raise DeploymentException( + "Circuit breaker is OPEN - service unavailable", + error_code="CIRCUIT_BREAKER_OPEN" + ) + + try: + result = func(*args, **kwargs) + self._on_success() + return result + except Exception as e: + self._on_failure() + raise e + + def _should_attempt_reset(self) -> bool: + """Check if we should attempt to reset the circuit breaker""" + if self.last_failure_time is None: + return False + return (datetime.utcnow() - self.last_failure_time).seconds >= self.recovery_timeout + + def _on_success(self): + """Handle successful call""" + self.failure_count = 0 + self.state = 'CLOSED' + + def _on_failure(self): + """Handle failed call""" + self.failure_count += 1 + self.last_failure_time = datetime.utcnow() + + if self.failure_count >= self.failure_threshold: + self.state = 'OPEN' + +# Practical DevOps error handling examples +class InfrastructureManager: + def __init__(self): + self.logger = DevOpsLogger(__name__) + self.circuit_breaker = CircuitBreaker() + + @handle_devops_errors(severity=ErrorSeverity.HIGH) + def deploy_infrastructure(self, config: Dict[str, Any]) -> Dict[str, Any]: + """Deploy infrastructure with comprehensive error handling""" + try: + # Validate configuration + self._validate_config(config) + + # Deploy resources + results = {} + for resource_type, resource_config in config.items(): + try: + result = self._deploy_resource(resource_type, resource_config) + results[resource_type] = result + except Exception as e: + # Log error and continue with other resources + self.logger.log_exception(e, ErrorSeverity.MEDIUM, { + 'resource_type': resource_type, + 'resource_config': resource_config + }) + results[resource_type] = {'status': 'failed', 'error': str(e)} + + return results + + except ConfigurationException as e: + # Configuration errors are critical + raise e + except Exception as e: + raise InfrastructureException( + f"Infrastructure deployment failed: {str(e)}", + error_code="DEPLOYMENT_FAILED" + ) + + def _validate_config(self, config: Dict[str, Any]): + """Validate infrastructure configuration""" + required_fields = ['region', 'environment'] + missing_fields = [field for field in required_fields if field not in config] + + if missing_fields: + raise ConfigurationException( + f"Missing required configuration fields: {missing_fields}", + error_code="INVALID_CONFIG", + details={'missing_fields': missing_fields} + ) + + def _deploy_resource(self, resource_type: str, config: Dict[str, Any]) -> Dict[str, Any]: + """Deploy a single resource with error handling""" + try: + # Use circuit breaker for external API calls + return self.circuit_breaker.call(self._call_cloud_api, resource_type, config) + except Exception as e: + raise InfrastructureException( + f"Failed to deploy {resource_type}: {str(e)}", + error_code="RESOURCE_DEPLOYMENT_FAILED", + details={'resource_type': resource_type, 'config': config} + ) + + def _call_cloud_api(self, resource_type: str, config: Dict[str, Any]) -> Dict[str, Any]: + """Simulate cloud API call that might fail""" + import random + if random.random() < 0.2: # 20% chance of failure + raise Exception("Simulated API failure") + + return { + 'status': 'success', + 'resource_id': f"{resource_type}-{random.randint(1000, 9999)}", + 'config': config + } + +# Debugging utilities for DevOps +class DevOpsDebugger: + @staticmethod + def create_debug_context(func_name: str, **kwargs) -> Dict[str, Any]: + """Create debugging context for troubleshooting""" + return { + 'function': func_name, + 'timestamp': datetime.utcnow().isoformat(), + 'parameters': kwargs, + 'system_info': { + 'python_version': sys.version, + 'platform': sys.platform + } + } + + @staticmethod + def debug_on_exception(func: Callable) -> Callable: + """Decorator to enter debugger on exception""" + @functools.wraps(func) + def wrapper(*args, **kwargs): + try: + return func(*args, **kwargs) + except Exception: + import pdb + pdb.post_mortem() + raise + return wrapper + +# Usage examples +def main(): + manager = InfrastructureManager() + + # Example 1: Valid configuration + valid_config = { + 'region': 'us-east-1', + 'environment': 'production', + 'ec2_instances': { + 'instance_type': 't3.small', + 'count': 2 + }, + 'rds_database': { + 'engine': 'postgresql', + 'instance_class': 'db.t3.micro' + } + } + + try: + results = manager.deploy_infrastructure(valid_config) + print("Deployment results:", results) + except DevOpsException as e: + print(f"DevOps error: {e}") + print(f"Error code: {e.error_code}") + print(f"Details: {e.details}") + + # Example 2: Invalid configuration + invalid_config = { + 'ec2_instances': { + 'instance_type': 't3.small' + } + } + + try: + results = manager.deploy_infrastructure(invalid_config) + print("Deployment results:", results) + except ConfigurationException as e: + print(f"Configuration error: {e}") + print(f"Missing fields: {e.details.get('missing_fields', [])}") + +if __name__ == "__main__": + main() +``` + +This comprehensive guide covers advanced Python language features specifically tailored for DevOps engineers. Each section includes practical examples and real-world usage scenarios that demonstrate how these concepts apply to infrastructure management, deployment automation, monitoring, and troubleshooting. + +The examples show how to build robust, maintainable DevOps tools using advanced Python features like OOP design patterns, functional programming concepts, concurrency for performance, and comprehensive error handling strategies. \ No newline at end of file diff --git a/02_configuration_management.md b/02_configuration_management.md new file mode 100644 index 0000000..16616b0 --- /dev/null +++ b/02_configuration_management.md @@ -0,0 +1,1885 @@ +# Configuration Management with Python for DevOps + +## Table of Contents +1. [YAML and JSON Processing](#yaml-and-json-processing) +2. [Infrastructure as Code (IaC)](#infrastructure-as-code-iac) +3. [Configuration Management Tools](#configuration-management-tools) +4. [Best Practices](#best-practices) + +## YAML and JSON Processing + +### Advanced YAML Operations + +**DevOps Usage Scenario**: Managing complex application configurations, Kubernetes manifests, and CI/CD pipeline definitions. + +```python +import yaml +import json +import jsonschema +from typing import Dict, Any, List, Optional +from pathlib import Path +import os +from jinja2 import Environment, FileSystemLoader, DictLoader +from dataclasses import dataclass, asdict +import logging + +# Configuration data classes for type safety +@dataclass +class DatabaseConfig: + host: str + port: int + name: str + username: str + password: str + ssl_enabled: bool = True + connection_pool_size: int = 10 + +@dataclass +class ApplicationConfig: + name: str + version: str + environment: str + debug: bool + database: DatabaseConfig + redis_url: str + secret_key: str + allowed_hosts: List[str] + +class ConfigurationManager: + """Advanced configuration management with validation and templating""" + + def __init__(self, config_dir: str = "configs"): + self.config_dir = Path(config_dir) + self.config_dir.mkdir(exist_ok=True) + self.logger = logging.getLogger(__name__) + + # YAML loader that preserves order and handles custom tags + yaml.add_constructor('!env', self._env_constructor) + yaml.add_constructor('!file', self._file_constructor) + yaml.add_constructor('!base64', self._base64_constructor) + + def _env_constructor(self, loader, node): + """Custom YAML constructor for environment variables""" + env_var = loader.construct_scalar(node) + parts = env_var.split(':', 1) + env_name = parts[0] + default_value = parts[1] if len(parts) > 1 else None + + value = os.getenv(env_name, default_value) + if value is None: + raise ValueError(f"Environment variable {env_name} not found") + return value + + def _file_constructor(self, loader, node): + """Custom YAML constructor for file contents""" + file_path = loader.construct_scalar(node) + try: + with open(file_path, 'r') as f: + return f.read().strip() + except FileNotFoundError: + raise ValueError(f"File {file_path} not found") + + def _base64_constructor(self, loader, node): + """Custom YAML constructor for base64 decoding""" + import base64 + encoded_value = loader.construct_scalar(node) + return base64.b64decode(encoded_value).decode('utf-8') + + def load_config(self, config_file: str) -> Dict[str, Any]: + """Load configuration from YAML file with custom constructors""" + config_path = self.config_dir / config_file + try: + with open(config_path, 'r') as f: + config = yaml.safe_load(f) + self.logger.info(f"Loaded configuration from {config_path}") + return config + except Exception as e: + self.logger.error(f"Failed to load config from {config_path}: {e}") + raise + + def load_config_with_includes(self, config_file: str) -> Dict[str, Any]: + """Load configuration with support for includes""" + config = self.load_config(config_file) + + # Process includes + if 'includes' in config: + for include_file in config['includes']: + include_config = self.load_config(include_file) + config = self._deep_merge(config, include_config) + del config['includes'] + + return config + + def _deep_merge(self, base: Dict[str, Any], override: Dict[str, Any]) -> Dict[str, Any]: + """Deep merge two dictionaries""" + result = base.copy() + for key, value in override.items(): + if key in result and isinstance(result[key], dict) and isinstance(value, dict): + result[key] = self._deep_merge(result[key], value) + else: + result[key] = value + return result + + def validate_config(self, config: Dict[str, Any], schema: Dict[str, Any]) -> bool: + """Validate configuration against JSON schema""" + try: + jsonschema.validate(config, schema) + self.logger.info("Configuration validation passed") + return True + except jsonschema.ValidationError as e: + self.logger.error(f"Configuration validation failed: {e.message}") + raise ValueError(f"Invalid configuration: {e.message}") + + def save_config(self, config: Dict[str, Any], config_file: str) -> None: + """Save configuration to YAML file""" + config_path = self.config_dir / config_file + try: + with open(config_path, 'w') as f: + yaml.dump(config, f, default_flow_style=False, sort_keys=False) + self.logger.info(f"Saved configuration to {config_path}") + except Exception as e: + self.logger.error(f"Failed to save config to {config_path}: {e}") + raise + +# Configuration schema definitions +APPLICATION_SCHEMA = { + "type": "object", + "properties": { + "name": {"type": "string"}, + "version": {"type": "string", "pattern": r"^\d+\.\d+\.\d+$"}, + "environment": {"type": "string", "enum": ["development", "staging", "production"]}, + "debug": {"type": "boolean"}, + "database": { + "type": "object", + "properties": { + "host": {"type": "string"}, + "port": {"type": "integer", "minimum": 1, "maximum": 65535}, + "name": {"type": "string"}, + "username": {"type": "string"}, + "password": {"type": "string"}, + "ssl_enabled": {"type": "boolean"}, + "connection_pool_size": {"type": "integer", "minimum": 1} + }, + "required": ["host", "port", "name", "username", "password"] + }, + "redis_url": {"type": "string"}, + "secret_key": {"type": "string"}, + "allowed_hosts": { + "type": "array", + "items": {"type": "string"} + } + }, + "required": ["name", "version", "environment", "database"] +} + +# Environment-specific configuration templates +class ConfigTemplateManager: + """Manage configuration templates for different environments""" + + def __init__(self, template_dir: str = "templates"): + self.template_dir = Path(template_dir) + self.template_dir.mkdir(exist_ok=True) + self.jinja_env = Environment( + loader=FileSystemLoader(str(self.template_dir)), + trim_blocks=True, + lstrip_blocks=True + ) + + def render_config(self, template_name: str, variables: Dict[str, Any]) -> str: + """Render configuration template with variables""" + template = self.jinja_env.get_template(template_name) + return template.render(**variables) + + def generate_config_for_environment(self, + environment: str, + base_config: Dict[str, Any]) -> Dict[str, Any]: + """Generate environment-specific configuration""" + env_overrides = { + 'development': { + 'debug': True, + 'database': { + 'host': 'localhost', + 'port': 5432, + 'ssl_enabled': False + }, + 'redis_url': 'redis://localhost:6379/0' + }, + 'staging': { + 'debug': False, + 'database': { + 'host': 'staging-db.example.com', + 'port': 5432, + 'ssl_enabled': True + }, + 'redis_url': 'redis://staging-redis.example.com:6379/0' + }, + 'production': { + 'debug': False, + 'database': { + 'host': 'prod-db.example.com', + 'port': 5432, + 'ssl_enabled': True, + 'connection_pool_size': 20 + }, + 'redis_url': 'redis://prod-redis.example.com:6379/0' + } + } + + if environment not in env_overrides: + raise ValueError(f"Unknown environment: {environment}") + + # Deep merge base config with environment overrides + config_manager = ConfigurationManager() + return config_manager._deep_merge(base_config, env_overrides[environment]) + +# Kubernetes configuration management +class KubernetesConfigManager: + """Manage Kubernetes configurations with Python""" + + def __init__(self, namespace: str = "default"): + self.namespace = namespace + self.logger = logging.getLogger(__name__) + + def create_deployment_config(self, + app_name: str, + image: str, + replicas: int = 1, + resources: Dict[str, Any] = None, + env_vars: Dict[str, str] = None) -> Dict[str, Any]: + """Create Kubernetes deployment configuration""" + + resources = resources or { + 'requests': {'cpu': '100m', 'memory': '128Mi'}, + 'limits': {'cpu': '500m', 'memory': '512Mi'} + } + + env_vars = env_vars or {} + + deployment = { + 'apiVersion': 'apps/v1', + 'kind': 'Deployment', + 'metadata': { + 'name': app_name, + 'namespace': self.namespace, + 'labels': { + 'app': app_name + } + }, + 'spec': { + 'replicas': replicas, + 'selector': { + 'matchLabels': { + 'app': app_name + } + }, + 'template': { + 'metadata': { + 'labels': { + 'app': app_name + } + }, + 'spec': { + 'containers': [{ + 'name': app_name, + 'image': image, + 'ports': [{'containerPort': 8080}], + 'resources': resources, + 'env': [{'name': k, 'value': v} for k, v in env_vars.items()] + }] + } + } + } + } + + return deployment + + def create_service_config(self, + app_name: str, + port: int = 80, + target_port: int = 8080, + service_type: str = 'ClusterIP') -> Dict[str, Any]: + """Create Kubernetes service configuration""" + + service = { + 'apiVersion': 'v1', + 'kind': 'Service', + 'metadata': { + 'name': f"{app_name}-service", + 'namespace': self.namespace, + 'labels': { + 'app': app_name + } + }, + 'spec': { + 'selector': { + 'app': app_name + }, + 'ports': [{ + 'port': port, + 'targetPort': target_port, + 'protocol': 'TCP' + }], + 'type': service_type + } + } + + return service + + def create_configmap(self, + name: str, + data: Dict[str, str], + binary_data: Dict[str, bytes] = None) -> Dict[str, Any]: + """Create Kubernetes ConfigMap""" + + configmap = { + 'apiVersion': 'v1', + 'kind': 'ConfigMap', + 'metadata': { + 'name': name, + 'namespace': self.namespace + }, + 'data': data + } + + if binary_data: + import base64 + configmap['binaryData'] = { + k: base64.b64encode(v).decode('utf-8') + for k, v in binary_data.items() + } + + return configmap + + def save_manifests(self, manifests: List[Dict[str, Any]], output_dir: str) -> None: + """Save Kubernetes manifests to files""" + output_path = Path(output_dir) + output_path.mkdir(exist_ok=True) + + for manifest in manifests: + kind = manifest.get('kind', 'unknown') + name = manifest.get('metadata', {}).get('name', 'unnamed') + filename = f"{kind.lower()}-{name}.yaml" + + with open(output_path / filename, 'w') as f: + yaml.dump(manifest, f, default_flow_style=False) + + self.logger.info(f"Saved {kind} manifest: {filename}") + +# Example usage scenarios +def main(): + # Example 1: Application configuration management + config_manager = ConfigurationManager() + + # Base application configuration + base_config = { + 'name': 'my-app', + 'version': '1.0.0', + 'database': { + 'name': 'myapp_db', + 'username': 'app_user', + 'password': '!env DATABASE_PASSWORD:default_password' + }, + 'secret_key': '!env SECRET_KEY', + 'allowed_hosts': ['localhost', '127.0.0.1'] + } + + # Generate environment-specific configurations + template_manager = ConfigTemplateManager() + + for env in ['development', 'staging', 'production']: + env_config = template_manager.generate_config_for_environment(env, base_config) + env_config['environment'] = env + + # Validate configuration + config_manager.validate_config(env_config, APPLICATION_SCHEMA) + + # Save configuration + config_manager.save_config(env_config, f"app-{env}.yaml") + print(f"Generated configuration for {env}") + + # Example 2: Kubernetes manifest generation + k8s_manager = KubernetesConfigManager(namespace="production") + + # Create deployment + deployment = k8s_manager.create_deployment_config( + app_name="web-app", + image="nginx:1.20", + replicas=3, + resources={ + 'requests': {'cpu': '200m', 'memory': '256Mi'}, + 'limits': {'cpu': '1000m', 'memory': '1Gi'} + }, + env_vars={ + 'ENV': 'production', + 'DEBUG': 'false' + } + ) + + # Create service + service = k8s_manager.create_service_config( + app_name="web-app", + port=80, + target_port=8080, + service_type="LoadBalancer" + ) + + # Create ConfigMap + app_config = { + 'database_url': 'postgresql://user:pass@db:5432/myapp', + 'redis_url': 'redis://redis:6379/0', + 'log_level': 'info' + } + + configmap = k8s_manager.create_configmap( + name="web-app-config", + data={k: str(v) for k, v in app_config.items()} + ) + + # Save all manifests + manifests = [deployment, service, configmap] + k8s_manager.save_manifests(manifests, "k8s-manifests") + + print("Generated Kubernetes manifests") + +if __name__ == "__main__": + main() +``` + +## Infrastructure as Code (IaC) + +### Terraform Integration with Python + +**DevOps Usage Scenario**: Managing Terraform configurations, state files, and automating infrastructure deployments. + +```python +import subprocess +import json +import os +from typing import Dict, Any, List, Optional +from pathlib import Path +import tempfile +from dataclasses import dataclass, asdict +import logging + +@dataclass +class TerraformResource: + """Base class for Terraform resources""" + resource_type: str + resource_name: str + provider: str = "aws" + +@dataclass +class EC2Instance(TerraformResource): + """EC2 instance resource""" + ami: str + instance_type: str + key_name: Optional[str] = None + vpc_security_group_ids: Optional[List[str]] = None + subnet_id: Optional[str] = None + tags: Optional[Dict[str, str]] = None + + def __post_init__(self): + self.resource_type = "aws_instance" + +@dataclass +class S3Bucket(TerraformResource): + """S3 bucket resource""" + bucket_name: str + acl: str = "private" + versioning_enabled: bool = True + tags: Optional[Dict[str, str]] = None + + def __post_init__(self): + self.resource_type = "aws_s3_bucket" + +class TerraformManager: + """Manage Terraform operations with Python""" + + def __init__(self, working_dir: str = "terraform"): + self.working_dir = Path(working_dir) + self.working_dir.mkdir(exist_ok=True) + self.logger = logging.getLogger(__name__) + self.state_file = self.working_dir / "terraform.tfstate" + + def generate_provider_config(self, providers: Dict[str, Dict[str, Any]]) -> str: + """Generate Terraform provider configuration""" + config_lines = ['terraform {', ' required_providers {'] + + for provider, settings in providers.items(): + config_lines.append(f' {provider} = {{') + for key, value in settings.items(): + if isinstance(value, str): + config_lines.append(f' {key} = "{value}"') + else: + config_lines.append(f' {key} = {json.dumps(value)}') + config_lines.append(' }') + + config_lines.extend([' }', '}', '']) + + # Add provider blocks + for provider in providers.keys(): + config_lines.extend([ + f'provider "{provider}" {{', + ' # Configuration options', + '}', + '' + ]) + + return '\n'.join(config_lines) + + def generate_resource_config(self, resources: List[TerraformResource]) -> str: + """Generate Terraform resource configuration""" + config_lines = [] + + for resource in resources: + config_lines.append( + f'resource "{resource.resource_type}" "{resource.resource_name}" {{' + ) + + # Convert dataclass to dict and filter out None values + resource_dict = asdict(resource) + resource_dict = {k: v for k, v in resource_dict.items() + if v is not None and k not in ['resource_type', 'resource_name', 'provider']} + + for key, value in resource_dict.items(): + if isinstance(value, str): + config_lines.append(f' {key} = "{value}"') + elif isinstance(value, bool): + config_lines.append(f' {key} = {str(value).lower()}') + elif isinstance(value, list): + config_lines.append(f' {key} = {json.dumps(value)}') + elif isinstance(value, dict): + config_lines.append(f' {key} = {{') + for sub_key, sub_value in value.items(): + if isinstance(sub_value, str): + config_lines.append(f' {sub_key} = "{sub_value}"') + else: + config_lines.append(f' {sub_key} = {json.dumps(sub_value)}') + config_lines.append(' }') + else: + config_lines.append(f' {key} = {value}') + + config_lines.extend(['}', '']) + + return '\n'.join(config_lines) + + def generate_variables(self, variables: Dict[str, Dict[str, Any]]) -> str: + """Generate Terraform variables configuration""" + config_lines = [] + + for var_name, var_config in variables.items(): + config_lines.append(f'variable "{var_name}" {{') + + for key, value in var_config.items(): + if isinstance(value, str): + config_lines.append(f' {key} = "{value}"') + else: + config_lines.append(f' {key} = {json.dumps(value)}') + + config_lines.extend(['}', '']) + + return '\n'.join(config_lines) + + def generate_outputs(self, outputs: Dict[str, Dict[str, str]]) -> str: + """Generate Terraform outputs configuration""" + config_lines = [] + + for output_name, output_config in outputs.items(): + config_lines.append(f'output "{output_name}" {{') + + for key, value in output_config.items(): + config_lines.append(f' {key} = {value}') + + config_lines.extend(['}', '']) + + return '\n'.join(config_lines) + + def write_terraform_files(self, + providers: Dict[str, Dict[str, Any]], + resources: List[TerraformResource], + variables: Dict[str, Dict[str, Any]] = None, + outputs: Dict[str, Dict[str, str]] = None) -> None: + """Write Terraform configuration files""" + + # Main configuration file + main_config = self.generate_provider_config(providers) + '\n' + main_config += self.generate_resource_config(resources) + + with open(self.working_dir / "main.tf", 'w') as f: + f.write(main_config) + + # Variables file + if variables: + variables_config = self.generate_variables(variables) + with open(self.working_dir / "variables.tf", 'w') as f: + f.write(variables_config) + + # Outputs file + if outputs: + outputs_config = self.generate_outputs(outputs) + with open(self.working_dir / "outputs.tf", 'w') as f: + f.write(outputs_config) + + self.logger.info("Generated Terraform configuration files") + + def terraform_init(self) -> subprocess.CompletedProcess: + """Initialize Terraform""" + return self._run_terraform_command(['init']) + + def terraform_plan(self, var_file: str = None) -> subprocess.CompletedProcess: + """Run Terraform plan""" + cmd = ['plan'] + if var_file: + cmd.extend(['-var-file', var_file]) + return self._run_terraform_command(cmd) + + def terraform_apply(self, var_file: str = None, auto_approve: bool = False) -> subprocess.CompletedProcess: + """Apply Terraform configuration""" + cmd = ['apply'] + if var_file: + cmd.extend(['-var-file', var_file]) + if auto_approve: + cmd.append('-auto-approve') + return self._run_terraform_command(cmd) + + def terraform_destroy(self, var_file: str = None, auto_approve: bool = False) -> subprocess.CompletedProcess: + """Destroy Terraform-managed infrastructure""" + cmd = ['destroy'] + if var_file: + cmd.extend(['-var-file', var_file]) + if auto_approve: + cmd.append('-auto-approve') + return self._run_terraform_command(cmd) + + def terraform_output(self, output_name: str = None) -> Dict[str, Any]: + """Get Terraform outputs""" + cmd = ['output', '-json'] + if output_name: + cmd.append(output_name) + + result = self._run_terraform_command(cmd) + if result.returncode == 0: + return json.loads(result.stdout) + else: + raise RuntimeError(f"Failed to get Terraform output: {result.stderr}") + + def get_state(self) -> Dict[str, Any]: + """Get current Terraform state""" + if self.state_file.exists(): + with open(self.state_file, 'r') as f: + return json.load(f) + return {} + + def _run_terraform_command(self, cmd: List[str]) -> subprocess.CompletedProcess: + """Run Terraform command""" + full_cmd = ['terraform'] + cmd + + self.logger.info(f"Running: {' '.join(full_cmd)}") + + result = subprocess.run( + full_cmd, + cwd=self.working_dir, + capture_output=True, + text=True + ) + + if result.returncode != 0: + self.logger.error(f"Terraform command failed: {result.stderr}") + else: + self.logger.info("Terraform command completed successfully") + + return result + +# CloudFormation integration +class CloudFormationManager: + """Manage AWS CloudFormation with Python""" + + def __init__(self): + import boto3 + self.cf_client = boto3.client('cloudformation') + self.logger = logging.getLogger(__name__) + + def generate_template(self, + resources: Dict[str, Dict[str, Any]], + parameters: Dict[str, Dict[str, Any]] = None, + outputs: Dict[str, Dict[str, str]] = None) -> Dict[str, Any]: + """Generate CloudFormation template""" + + template = { + 'AWSTemplateFormatVersion': '2010-09-09', + 'Description': 'Generated CloudFormation template', + 'Resources': resources + } + + if parameters: + template['Parameters'] = parameters + + if outputs: + template['Outputs'] = outputs + + return template + + def create_stack(self, + stack_name: str, + template: Dict[str, Any], + parameters: List[Dict[str, str]] = None, + tags: List[Dict[str, str]] = None) -> str: + """Create CloudFormation stack""" + + kwargs = { + 'StackName': stack_name, + 'TemplateBody': json.dumps(template), + 'Capabilities': ['CAPABILITY_IAM', 'CAPABILITY_NAMED_IAM'] + } + + if parameters: + kwargs['Parameters'] = parameters + + if tags: + kwargs['Tags'] = tags + + response = self.cf_client.create_stack(**kwargs) + stack_id = response['StackId'] + + self.logger.info(f"Created CloudFormation stack: {stack_name} ({stack_id})") + return stack_id + + def update_stack(self, + stack_name: str, + template: Dict[str, Any], + parameters: List[Dict[str, str]] = None) -> str: + """Update CloudFormation stack""" + + kwargs = { + 'StackName': stack_name, + 'TemplateBody': json.dumps(template), + 'Capabilities': ['CAPABILITY_IAM', 'CAPABILITY_NAMED_IAM'] + } + + if parameters: + kwargs['Parameters'] = parameters + + response = self.cf_client.update_stack(**kwargs) + stack_id = response['StackId'] + + self.logger.info(f"Updated CloudFormation stack: {stack_name}") + return stack_id + + def delete_stack(self, stack_name: str) -> None: + """Delete CloudFormation stack""" + self.cf_client.delete_stack(StackName=stack_name) + self.logger.info(f"Initiated deletion of CloudFormation stack: {stack_name}") + + def get_stack_status(self, stack_name: str) -> str: + """Get CloudFormation stack status""" + response = self.cf_client.describe_stacks(StackName=stack_name) + return response['Stacks'][0]['StackStatus'] + + def wait_for_stack_completion(self, stack_name: str, operation: str = 'create') -> bool: + """Wait for stack operation to complete""" + import time + + success_statuses = { + 'create': 'CREATE_COMPLETE', + 'update': 'UPDATE_COMPLETE', + 'delete': 'DELETE_COMPLETE' + } + + failed_statuses = { + 'create': ['CREATE_FAILED', 'ROLLBACK_COMPLETE'], + 'update': ['UPDATE_FAILED', 'UPDATE_ROLLBACK_COMPLETE'], + 'delete': ['DELETE_FAILED'] + } + + target_status = success_statuses[operation] + fail_statuses = failed_statuses[operation] + + while True: + try: + status = self.get_stack_status(stack_name) + self.logger.info(f"Stack {stack_name} status: {status}") + + if status == target_status: + return True + elif status in fail_statuses: + return False + + time.sleep(30) # Wait 30 seconds before checking again + + except Exception as e: + if operation == 'delete' and 'does not exist' in str(e): + return True # Stack successfully deleted + self.logger.error(f"Error checking stack status: {e}") + return False + +# Pulumi integration +class PulumiManager: + """Manage Pulumi infrastructure with Python""" + + def __init__(self, project_name: str, stack_name: str): + self.project_name = project_name + self.stack_name = stack_name + self.logger = logging.getLogger(__name__) + + def create_pulumi_project(self, + template: str = "aws-python", + project_dir: str = None) -> str: + """Create new Pulumi project""" + if project_dir is None: + project_dir = self.project_name + + os.makedirs(project_dir, exist_ok=True) + + # Create Pulumi.yaml + pulumi_config = { + 'name': self.project_name, + 'runtime': 'python', + 'description': f'Infrastructure for {self.project_name}' + } + + pulumi_yaml_path = Path(project_dir) / "Pulumi.yaml" + with open(pulumi_yaml_path, 'w') as f: + yaml.dump(pulumi_config, f) + + # Create requirements.txt + requirements = [ + 'pulumi>=3.0.0,<4.0.0', + 'pulumi-aws>=4.0.0,<5.0.0' + ] + + requirements_path = Path(project_dir) / "requirements.txt" + with open(requirements_path, 'w') as f: + f.write('\n'.join(requirements)) + + self.logger.info(f"Created Pulumi project: {self.project_name}") + return project_dir + + def generate_main_program(self, resources: List[Dict[str, Any]]) -> str: + """Generate main Pulumi program""" + + program_lines = [ + 'import pulumi', + 'import pulumi_aws as aws', + '', + ] + + for resource in resources: + resource_type = resource['type'] + resource_name = resource['name'] + resource_args = resource.get('args', {}) + + program_lines.append(f'{resource_name} = aws.{resource_type}(') + program_lines.append(f' "{resource_name}",') + + for key, value in resource_args.items(): + if isinstance(value, str): + program_lines.append(f' {key}="{value}",') + else: + program_lines.append(f' {key}={value},') + + program_lines.append(')') + program_lines.append('') + + # Add exports + program_lines.append('# Exports') + for resource in resources: + resource_name = resource['name'] + program_lines.append(f'pulumi.export("{resource_name}_id", {resource_name}.id)') + + return '\n'.join(program_lines) + +# Example usage +def example_terraform_usage(): + """Example of using TerraformManager""" + + # Create Terraform manager + tf_manager = TerraformManager("./terraform-example") + + # Define providers + providers = { + 'aws': { + 'source': 'hashicorp/aws', + 'version': '~> 4.0' + } + } + + # Define resources + resources = [ + EC2Instance( + resource_name="web_server", + ami="ami-0abcdef1234567890", + instance_type="t3.micro", + tags={"Name": "WebServer", "Environment": "production"} + ), + S3Bucket( + resource_name="app_bucket", + bucket_name="my-app-bucket-unique-name", + acl="private", + tags={"Purpose": "Application Storage"} + ) + ] + + # Define variables + variables = { + 'region': { + 'description': 'AWS region', + 'type': 'string', + 'default': 'us-east-1' + }, + 'environment': { + 'description': 'Environment name', + 'type': 'string' + } + } + + # Define outputs + outputs = { + 'instance_id': { + 'value': 'aws_instance.web_server.id', + 'description': 'EC2 instance ID' + }, + 'bucket_name': { + 'value': 'aws_s3_bucket.app_bucket.bucket', + 'description': 'S3 bucket name' + } + } + + # Generate Terraform files + tf_manager.write_terraform_files(providers, resources, variables, outputs) + + # Initialize and plan + tf_manager.terraform_init() + tf_manager.terraform_plan() + + print("Terraform configuration generated and planned successfully") + +if __name__ == "__main__": + example_terraform_usage() +``` + +## Configuration Management Tools + +### Ansible Integration + +**DevOps Usage Scenario**: Automating server configuration, application deployment, and infrastructure management using Ansible with Python. + +```python +import yaml +import json +import subprocess +from typing import Dict, Any, List, Optional +from pathlib import Path +import tempfile +import logging +from dataclasses import dataclass, asdict + +@dataclass +class AnsibleTask: + """Represents an Ansible task""" + name: str + module: str + args: Dict[str, Any] + when: Optional[str] = None + tags: Optional[List[str]] = None + become: Optional[bool] = None + register: Optional[str] = None + +@dataclass +class AnsiblePlay: + """Represents an Ansible play""" + name: str + hosts: str + tasks: List[AnsibleTask] + vars: Optional[Dict[str, Any]] = None + become: Optional[bool] = None + gather_facts: Optional[bool] = None + +class AnsibleManager: + """Manage Ansible operations with Python""" + + def __init__(self, inventory_file: str = "inventory.ini"): + self.inventory_file = inventory_file + self.logger = logging.getLogger(__name__) + self.playbooks_dir = Path("playbooks") + self.playbooks_dir.mkdir(exist_ok=True) + + def create_inventory(self, inventory: Dict[str, Any]) -> str: + """Create Ansible inventory file""" + inventory_lines = [] + + # Process groups + for group_name, group_data in inventory.items(): + if group_name == '_meta': + continue + + inventory_lines.append(f"[{group_name}]") + + if isinstance(group_data, dict): + hosts = group_data.get('hosts', []) + vars_dict = group_data.get('vars', {}) + + # Add hosts + for host in hosts: + if isinstance(host, dict): + host_line = host['name'] + host_vars = host.get('vars', {}) + for key, value in host_vars.items(): + host_line += f" {key}={value}" + inventory_lines.append(host_line) + else: + inventory_lines.append(host) + + # Add group variables + if vars_dict: + inventory_lines.append(f"[{group_name}:vars]") + for key, value in vars_dict.items(): + inventory_lines.append(f"{key}={value}") + + inventory_lines.append("") + + inventory_content = "\n".join(inventory_lines) + + with open(self.inventory_file, 'w') as f: + f.write(inventory_content) + + self.logger.info(f"Created inventory file: {self.inventory_file}") + return inventory_content + + def create_playbook(self, plays: List[AnsiblePlay], playbook_name: str) -> str: + """Create Ansible playbook from play objects""" + playbook_data = [] + + for play in plays: + play_dict = { + 'name': play.name, + 'hosts': play.hosts, + 'tasks': [] + } + + if play.vars: + play_dict['vars'] = play.vars + if play.become is not None: + play_dict['become'] = play.become + if play.gather_facts is not None: + play_dict['gather_facts'] = play.gather_facts + + # Add tasks + for task in play.tasks: + task_dict = { + 'name': task.name, + task.module: task.args + } + + if task.when: + task_dict['when'] = task.when + if task.tags: + task_dict['tags'] = task.tags + if task.become is not None: + task_dict['become'] = task.become + if task.register: + task_dict['register'] = task.register + + play_dict['tasks'].append(task_dict) + + playbook_data.append(play_dict) + + playbook_content = yaml.dump(playbook_data, default_flow_style=False) + playbook_path = self.playbooks_dir / f"{playbook_name}.yml" + + with open(playbook_path, 'w') as f: + f.write(playbook_content) + + self.logger.info(f"Created playbook: {playbook_path}") + return str(playbook_path) + + def run_playbook(self, + playbook_path: str, + inventory: str = None, + extra_vars: Dict[str, Any] = None, + tags: List[str] = None, + limit: str = None, + check_mode: bool = False) -> subprocess.CompletedProcess: + """Run Ansible playbook""" + + cmd = ['ansible-playbook'] + + # Add inventory + if inventory: + cmd.extend(['-i', inventory]) + elif self.inventory_file: + cmd.extend(['-i', self.inventory_file]) + + # Add extra variables + if extra_vars: + extra_vars_json = json.dumps(extra_vars) + cmd.extend(['--extra-vars', extra_vars_json]) + + # Add tags + if tags: + cmd.extend(['--tags', ','.join(tags)]) + + # Add limit + if limit: + cmd.extend(['--limit', limit]) + + # Add check mode + if check_mode: + cmd.append('--check') + + # Add playbook path + cmd.append(playbook_path) + + self.logger.info(f"Running: {' '.join(cmd)}") + + result = subprocess.run( + cmd, + capture_output=True, + text=True + ) + + if result.returncode == 0: + self.logger.info("Playbook execution completed successfully") + else: + self.logger.error(f"Playbook execution failed: {result.stderr}") + + return result + + def run_ad_hoc_command(self, + hosts: str, + module: str, + args: str = "", + inventory: str = None, + become: bool = False) -> subprocess.CompletedProcess: + """Run ad-hoc Ansible command""" + + cmd = ['ansible'] + + # Add inventory + if inventory: + cmd.extend(['-i', inventory]) + elif self.inventory_file: + cmd.extend(['-i', self.inventory_file]) + + # Add become + if become: + cmd.append('--become') + + # Add hosts and module + cmd.extend([hosts, '-m', module]) + + # Add module arguments + if args: + cmd.extend(['-a', args]) + + self.logger.info(f"Running ad-hoc command: {' '.join(cmd)}") + + result = subprocess.run( + cmd, + capture_output=True, + text=True + ) + + return result + +class ApplicationDeploymentManager: + """Manage application deployments with Ansible""" + + def __init__(self, ansible_manager: AnsibleManager): + self.ansible_manager = ansible_manager + self.logger = logging.getLogger(__name__) + + def create_web_app_deployment(self, + app_name: str, + app_version: str, + git_repo: str, + app_port: int = 8080, + environment: str = "production") -> str: + """Create web application deployment playbook""" + + # Define tasks for web app deployment + tasks = [ + AnsibleTask( + name="Install required packages", + module="package", + args={ + "name": ["git", "python3", "python3-pip", "nginx"], + "state": "present" + }, + become=True + ), + AnsibleTask( + name="Create application user", + module="user", + args={ + "name": app_name, + "system": True, + "shell": "/bin/bash", + "home": f"/opt/{app_name}" + }, + become=True + ), + AnsibleTask( + name="Clone application repository", + module="git", + args={ + "repo": git_repo, + "dest": f"/opt/{app_name}/app", + "version": app_version, + "force": True + }, + become=True, + register="git_clone" + ), + AnsibleTask( + name="Install Python dependencies", + module="pip", + args={ + "requirements": f"/opt/{app_name}/app/requirements.txt", + "virtualenv": f"/opt/{app_name}/venv", + "virtualenv_python": "python3" + }, + become=True + ), + AnsibleTask( + name="Create application configuration", + module="template", + args={ + "src": "app_config.j2", + "dest": f"/opt/{app_name}/app/config.py", + "owner": app_name, + "group": app_name, + "mode": "0644" + }, + become=True + ), + AnsibleTask( + name="Create systemd service file", + module="template", + args={ + "src": "app_service.j2", + "dest": f"/etc/systemd/system/{app_name}.service", + "mode": "0644" + }, + become=True, + register="service_file" + ), + AnsibleTask( + name="Reload systemd daemon", + module="systemd", + args={ + "daemon_reload": True + }, + become=True, + when="service_file.changed" + ), + AnsibleTask( + name="Start and enable application service", + module="systemd", + args={ + "name": app_name, + "state": "started", + "enabled": True + }, + become=True + ), + AnsibleTask( + name="Configure Nginx reverse proxy", + module="template", + args={ + "src": "nginx_site.j2", + "dest": f"/etc/nginx/sites-available/{app_name}", + "mode": "0644" + }, + become=True, + register="nginx_config" + ), + AnsibleTask( + name="Enable Nginx site", + module="file", + args={ + "src": f"/etc/nginx/sites-available/{app_name}", + "dest": f"/etc/nginx/sites-enabled/{app_name}", + "state": "link" + }, + become=True + ), + AnsibleTask( + name="Restart Nginx", + module="systemd", + args={ + "name": "nginx", + "state": "restarted" + }, + become=True, + when="nginx_config.changed" + ) + ] + + # Create play + play = AnsiblePlay( + name=f"Deploy {app_name} application", + hosts="web_servers", + tasks=tasks, + vars={ + "app_name": app_name, + "app_version": app_version, + "app_port": app_port, + "environment": environment + }, + become=False, + gather_facts=True + ) + + # Create playbook + playbook_path = self.ansible_manager.create_playbook( + [play], + f"deploy_{app_name}" + ) + + return playbook_path + + def create_database_setup(self, + db_type: str = "postgresql", + db_name: str = "appdb", + db_user: str = "appuser", + db_password: str = "changeme") -> str: + """Create database setup playbook""" + + if db_type == "postgresql": + tasks = [ + AnsibleTask( + name="Install PostgreSQL", + module="package", + args={ + "name": ["postgresql", "postgresql-contrib", "python3-psycopg2"], + "state": "present" + }, + become=True + ), + AnsibleTask( + name="Start PostgreSQL service", + module="systemd", + args={ + "name": "postgresql", + "state": "started", + "enabled": True + }, + become=True + ), + AnsibleTask( + name="Create database", + module="postgresql_db", + args={ + "name": db_name, + "state": "present" + }, + become=True, + become_user="postgres" + ), + AnsibleTask( + name="Create database user", + module="postgresql_user", + args={ + "name": db_user, + "password": db_password, + "priv": f"{db_name}:ALL", + "state": "present" + }, + become=True, + become_user="postgres" + ) + ] + elif db_type == "mysql": + tasks = [ + AnsibleTask( + name="Install MySQL", + module="package", + args={ + "name": ["mysql-server", "python3-pymysql"], + "state": "present" + }, + become=True + ), + AnsibleTask( + name="Start MySQL service", + module="systemd", + args={ + "name": "mysql", + "state": "started", + "enabled": True + }, + become=True + ), + AnsibleTask( + name="Create database", + module="mysql_db", + args={ + "name": db_name, + "state": "present" + }, + become=True + ), + AnsibleTask( + name="Create database user", + module="mysql_user", + args={ + "name": db_user, + "password": db_password, + "priv": f"{db_name}.*:ALL", + "state": "present" + }, + become=True + ) + ] + else: + raise ValueError(f"Unsupported database type: {db_type}") + + play = AnsiblePlay( + name=f"Setup {db_type} database", + hosts="database_servers", + tasks=tasks, + become=False, + gather_facts=True + ) + + playbook_path = self.ansible_manager.create_playbook( + [play], + f"setup_{db_type}_database" + ) + + return playbook_path + +# Example usage +def main(): + # Create Ansible manager + ansible_manager = AnsibleManager() + + # Create inventory + inventory = { + 'web_servers': { + 'hosts': [ + {'name': 'web1.example.com', 'vars': {'ansible_user': 'ubuntu'}}, + {'name': 'web2.example.com', 'vars': {'ansible_user': 'ubuntu'}} + ], + 'vars': { + 'http_port': 80, + 'maxRequestsPerChild': 808 + } + }, + 'database_servers': { + 'hosts': ['db1.example.com', 'db2.example.com'], + 'vars': { + 'ansible_user': 'ubuntu' + } + } + } + + ansible_manager.create_inventory(inventory) + + # Create deployment manager + deployment_manager = ApplicationDeploymentManager(ansible_manager) + + # Create web app deployment playbook + web_app_playbook = deployment_manager.create_web_app_deployment( + app_name="myapp", + app_version="v1.2.0", + git_repo="https://github.com/example/myapp.git", + app_port=8080, + environment="production" + ) + + # Create database setup playbook + db_playbook = deployment_manager.create_database_setup( + db_type="postgresql", + db_name="myapp_db", + db_user="myapp_user", + db_password="secure_password_123" + ) + + print(f"Created web app deployment playbook: {web_app_playbook}") + print(f"Created database setup playbook: {db_playbook}") + + # Run playbooks (uncomment to actually execute) + # ansible_manager.run_playbook(db_playbook, check_mode=True) + # ansible_manager.run_playbook(web_app_playbook, check_mode=True) + +if __name__ == "__main__": + main() +``` + +## Best Practices + +### Configuration Security and Validation + +**DevOps Usage Scenario**: Implementing secure configuration management with encryption, validation, and audit trails. + +```python +import os +import hashlib +import base64 +from cryptography.fernet import Fernet +from cryptography.hazmat.primitives import hashes +from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC +import json +import yaml +from typing import Dict, Any, List +import logging +from datetime import datetime +from pathlib import Path + +class SecureConfigManager: + """Secure configuration management with encryption and validation""" + + def __init__(self, password: str = None): + self.logger = logging.getLogger(__name__) + self.encryption_key = self._derive_key(password) if password else None + self.audit_log = [] + + def _derive_key(self, password: str) -> bytes: + """Derive encryption key from password""" + password_bytes = password.encode() + salt = b'salt_1234567890' # In production, use random salt + kdf = PBKDF2HMAC( + algorithm=hashes.SHA256(), + length=32, + salt=salt, + iterations=100000, + ) + key = base64.urlsafe_b64encode(kdf.derive(password_bytes)) + return key + + def encrypt_value(self, value: str) -> str: + """Encrypt a configuration value""" + if not self.encryption_key: + raise ValueError("No encryption key available") + + f = Fernet(self.encryption_key) + encrypted_value = f.encrypt(value.encode()) + return base64.urlsafe_b64encode(encrypted_value).decode() + + def decrypt_value(self, encrypted_value: str) -> str: + """Decrypt a configuration value""" + if not self.encryption_key: + raise ValueError("No encryption key available") + + f = Fernet(self.encryption_key) + encrypted_bytes = base64.urlsafe_b64decode(encrypted_value.encode()) + decrypted_value = f.decrypt(encrypted_bytes) + return decrypted_value.decode() + + def encrypt_sensitive_values(self, config: Dict[str, Any], + sensitive_keys: List[str]) -> Dict[str, Any]: + """Encrypt sensitive values in configuration""" + encrypted_config = config.copy() + + for key_path in sensitive_keys: + keys = key_path.split('.') + current = encrypted_config + + # Navigate to the parent of the target key + for key in keys[:-1]: + if key in current and isinstance(current[key], dict): + current = current[key] + else: + break + else: + # Encrypt the final key's value + final_key = keys[-1] + if final_key in current and isinstance(current[final_key], str): + original_value = current[final_key] + encrypted_value = self.encrypt_value(original_value) + current[final_key] = f"ENC[{encrypted_value}]" + + self._log_audit( + "encrypt", + f"Encrypted sensitive value at {key_path}", + {"key_path": key_path} + ) + + return encrypted_config + + def decrypt_config(self, config: Dict[str, Any]) -> Dict[str, Any]: + """Decrypt all encrypted values in configuration""" + decrypted_config = self._deep_copy_and_decrypt(config) + return decrypted_config + + def _deep_copy_and_decrypt(self, obj: Any) -> Any: + """Recursively decrypt encrypted values in nested structures""" + if isinstance(obj, dict): + return {k: self._deep_copy_and_decrypt(v) for k, v in obj.items()} + elif isinstance(obj, list): + return [self._deep_copy_and_decrypt(item) for item in obj] + elif isinstance(obj, str) and obj.startswith("ENC[") and obj.endswith("]"): + encrypted_value = obj[4:-1] # Remove "ENC[" and "]" + return self.decrypt_value(encrypted_value) + else: + return obj + + def _log_audit(self, action: str, description: str, metadata: Dict[str, Any] = None): + """Log configuration management actions for audit trail""" + audit_entry = { + "timestamp": datetime.utcnow().isoformat(), + "action": action, + "description": description, + "metadata": metadata or {} + } + self.audit_log.append(audit_entry) + self.logger.info(f"Audit: {action} - {description}") + + def get_audit_log(self) -> List[Dict[str, Any]]: + """Get audit log entries""" + return self.audit_log.copy() + + def validate_config_integrity(self, config_file: str) -> bool: + """Validate configuration file integrity using checksums""" + try: + with open(config_file, 'rb') as f: + content = f.read() + + checksum = hashlib.sha256(content).hexdigest() + checksum_file = f"{config_file}.sha256" + + if os.path.exists(checksum_file): + with open(checksum_file, 'r') as f: + stored_checksum = f.read().strip() + + if checksum == stored_checksum: + self.logger.info(f"Config integrity verified: {config_file}") + return True + else: + self.logger.error(f"Config integrity check failed: {config_file}") + return False + else: + # Create checksum file + with open(checksum_file, 'w') as f: + f.write(checksum) + self.logger.info(f"Created checksum file: {checksum_file}") + return True + + except Exception as e: + self.logger.error(f"Error validating config integrity: {e}") + return False + +class ConfigurationValidator: + """Advanced configuration validation""" + + def __init__(self): + self.logger = logging.getLogger(__name__) + self.validation_rules = {} + + def add_validation_rule(self, key_path: str, validator_func, error_message: str): + """Add custom validation rule""" + self.validation_rules[key_path] = { + 'validator': validator_func, + 'error_message': error_message + } + + def validate_config(self, config: Dict[str, Any]) -> Dict[str, Any]: + """Validate configuration using custom rules""" + validation_results = { + 'valid': True, + 'errors': [], + 'warnings': [] + } + + for key_path, rule in self.validation_rules.items(): + try: + value = self._get_nested_value(config, key_path) + if value is not None: + if not rule['validator'](value): + validation_results['valid'] = False + validation_results['errors'].append({ + 'key_path': key_path, + 'message': rule['error_message'], + 'value': str(value) + }) + except KeyError: + validation_results['warnings'].append({ + 'key_path': key_path, + 'message': f"Configuration key {key_path} not found" + }) + + return validation_results + + def _get_nested_value(self, config: Dict[str, Any], key_path: str) -> Any: + """Get value from nested configuration using dot notation""" + keys = key_path.split('.') + value = config + + for key in keys: + if isinstance(value, dict) and key in value: + value = value[key] + else: + raise KeyError(f"Key {key_path} not found") + + return value + + def validate_environment_consistency(self, + configs: Dict[str, Dict[str, Any]]) -> Dict[str, Any]: + """Validate consistency across different environment configurations""" + consistency_results = { + 'consistent': True, + 'inconsistencies': [] + } + + if len(configs) < 2: + return consistency_results + + # Get all unique key paths from all configs + all_key_paths = set() + for config in configs.values(): + all_key_paths.update(self._get_all_key_paths(config)) + + # Check each key path across environments + for key_path in all_key_paths: + values = {} + for env_name, config in configs.items(): + try: + value = self._get_nested_value(config, key_path) + values[env_name] = value + except KeyError: + values[env_name] = None + + # Check for type consistency (all values should be same type) + non_none_values = [v for v in values.values() if v is not None] + if non_none_values: + first_type = type(non_none_values[0]) + if not all(type(v) == first_type for v in non_none_values): + consistency_results['consistent'] = False + consistency_results['inconsistencies'].append({ + 'key_path': key_path, + 'issue': 'type_mismatch', + 'values': {env: str(type(v)) for env, v in values.items()} + }) + + return consistency_results + + def _get_all_key_paths(self, config: Dict[str, Any], prefix: str = "") -> List[str]: + """Get all key paths in a nested configuration""" + key_paths = [] + + for key, value in config.items(): + current_path = f"{prefix}.{key}" if prefix else key + key_paths.append(current_path) + + if isinstance(value, dict): + key_paths.extend(self._get_all_key_paths(value, current_path)) + + return key_paths + +# Configuration management best practices example +def demonstrate_best_practices(): + """Demonstrate configuration management best practices""" + + # Initialize secure config manager + secure_manager = SecureConfigManager(password="my_secure_password_123") + + # Sample configuration with sensitive data + config = { + 'application': { + 'name': 'myapp', + 'version': '1.0.0', + 'debug': False + }, + 'database': { + 'host': 'db.example.com', + 'port': 5432, + 'name': 'myapp_db', + 'username': 'dbuser', + 'password': 'super_secret_password', + 'ssl_enabled': True + }, + 'api_keys': { + 'stripe': 'sk_live_abc123def456', + 'sendgrid': 'SG.xyz789.abc123' + }, + 'redis': { + 'url': 'redis://redis.example.com:6379/0', + 'password': 'redis_secret_password' + } + } + + # Define sensitive keys that should be encrypted + sensitive_keys = [ + 'database.password', + 'api_keys.stripe', + 'api_keys.sendgrid', + 'redis.password' + ] + + # Encrypt sensitive values + encrypted_config = secure_manager.encrypt_sensitive_values(config, sensitive_keys) + + # Save encrypted configuration + with open('config_encrypted.yaml', 'w') as f: + yaml.dump(encrypted_config, f, default_flow_style=False) + + print("Encrypted configuration saved to config_encrypted.yaml") + + # Validate file integrity + integrity_valid = secure_manager.validate_config_integrity('config_encrypted.yaml') + print(f"Configuration integrity valid: {integrity_valid}") + + # Load and decrypt configuration + with open('config_encrypted.yaml', 'r') as f: + loaded_encrypted_config = yaml.safe_load(f) + + decrypted_config = secure_manager.decrypt_config(loaded_encrypted_config) + print("Configuration decrypted successfully") + + # Set up validation rules + validator = ConfigurationValidator() + + # Add validation rules + validator.add_validation_rule( + 'database.port', + lambda x: isinstance(x, int) and 1 <= x <= 65535, + 'Database port must be a valid integer between 1 and 65535' + ) + + validator.add_validation_rule( + 'database.ssl_enabled', + lambda x: isinstance(x, bool), + 'Database SSL enabled must be a boolean value' + ) + + validator.add_validation_rule( + 'application.version', + lambda x: isinstance(x, str) and len(x.split('.')) == 3, + 'Application version must be in semantic versioning format (x.y.z)' + ) + + # Validate configuration + validation_results = validator.validate_config(decrypted_config) + + if validation_results['valid']: + print("Configuration validation passed") + else: + print("Configuration validation failed:") + for error in validation_results['errors']: + print(f" - {error['key_path']}: {error['message']}") + + # Print audit log + print("\nAudit Log:") + for entry in secure_manager.get_audit_log(): + print(f" {entry['timestamp']}: {entry['action']} - {entry['description']}") + +if __name__ == "__main__": + demonstrate_best_practices() +``` + +This comprehensive configuration management guide covers: + +1. **YAML and JSON Processing**: Advanced configuration loading, validation, and templating +2. **Infrastructure as Code**: Terraform, CloudFormation, and Pulumi integration +3. **Configuration Management Tools**: Ansible automation and playbook generation +4. **Best Practices**: Security, encryption, validation, and audit trails + +The examples demonstrate real-world DevOps scenarios like: +- Managing complex application configurations across environments +- Generating infrastructure code programmatically +- Automating server configuration and application deployment +- Implementing secure configuration management with encryption +- Validating configuration consistency and integrity + +Each section includes practical, production-ready code that DevOps engineers can adapt for their specific use cases. \ No newline at end of file diff --git a/03_cloud_platform_integration.md b/03_cloud_platform_integration.md new file mode 100644 index 0000000..260a7f3 --- /dev/null +++ b/03_cloud_platform_integration.md @@ -0,0 +1,1013 @@ +# Cloud Platform Integration with Python for DevOps + +## Table of Contents +1. [AWS Automation](#aws-automation) +2. [Azure Automation](#azure-automation) +3. [Google Cloud Platform](#google-cloud-platform) +4. [Multi-Cloud Management](#multi-cloud-management) + +## AWS Automation + +### Boto3 SDK Mastery + +**DevOps Usage Scenario**: Comprehensive AWS resource management, automation, and cost optimization. + +```python +import boto3 +import json +from typing import Dict, List, Any, Optional +from dataclasses import dataclass +from datetime import datetime, timedelta +import logging + +@dataclass +class AWSResource: + resource_type: str + resource_id: str + region: str + tags: Dict[str, str] + created_date: datetime + +class AWSManager: + """Comprehensive AWS resource management""" + + def __init__(self, region: str = 'us-east-1'): + self.region = region + self.session = boto3.Session(region_name=region) + self.logger = logging.getLogger(__name__) + + # Initialize clients + self.ec2 = self.session.client('ec2') + self.s3 = self.session.client('s3') + self.rds = self.session.client('rds') + self.lambda_client = self.session.client('lambda') + self.cloudformation = self.session.client('cloudformation') + self.cost_explorer = self.session.client('ce') + + def create_vpc_infrastructure(self, vpc_cidr: str = "10.0.0.0/16") -> Dict[str, str]: + """Create complete VPC infrastructure""" + try: + # Create VPC + vpc_response = self.ec2.create_vpc(CidrBlock=vpc_cidr) + vpc_id = vpc_response['Vpc']['VpcId'] + + # Create Internet Gateway + igw_response = self.ec2.create_internet_gateway() + igw_id = igw_response['InternetGateway']['InternetGatewayId'] + self.ec2.attach_internet_gateway(InternetGatewayId=igw_id, VpcId=vpc_id) + + # Create public subnet + public_subnet = self.ec2.create_subnet( + VpcId=vpc_id, + CidrBlock="10.0.1.0/24", + AvailabilityZone=f"{self.region}a" + ) + public_subnet_id = public_subnet['Subnet']['SubnetId'] + + # Create private subnet + private_subnet = self.ec2.create_subnet( + VpcId=vpc_id, + CidrBlock="10.0.2.0/24", + AvailabilityZone=f"{self.region}b" + ) + private_subnet_id = private_subnet['Subnet']['SubnetId'] + + # Create route table for public subnet + route_table = self.ec2.create_route_table(VpcId=vpc_id) + route_table_id = route_table['RouteTable']['RouteTableId'] + + # Add route to internet gateway + self.ec2.create_route( + RouteTableId=route_table_id, + DestinationCidrBlock='0.0.0.0/0', + GatewayId=igw_id + ) + + # Associate route table with public subnet + self.ec2.associate_route_table( + RouteTableId=route_table_id, + SubnetId=public_subnet_id + ) + + return { + 'vpc_id': vpc_id, + 'igw_id': igw_id, + 'public_subnet_id': public_subnet_id, + 'private_subnet_id': private_subnet_id, + 'route_table_id': route_table_id + } + + except Exception as e: + self.logger.error(f"Failed to create VPC infrastructure: {e}") + raise + + def launch_auto_scaling_group(self, + launch_template_name: str, + subnet_ids: List[str], + min_size: int = 1, + max_size: int = 3, + desired_capacity: int = 2) -> str: + """Launch Auto Scaling Group with launch template""" + + try: + autoscaling = self.session.client('autoscaling') + + asg_response = autoscaling.create_auto_scaling_group( + AutoScalingGroupName=f"{launch_template_name}-asg", + LaunchTemplate={ + 'LaunchTemplateName': launch_template_name, + 'Version': '$Latest' + }, + MinSize=min_size, + MaxSize=max_size, + DesiredCapacity=desired_capacity, + VPCZoneIdentifier=','.join(subnet_ids), + TargetGroupARNs=[], + HealthCheckType='ELB', + HealthCheckGracePeriod=300, + Tags=[ + { + 'Key': 'Name', + 'Value': f"{launch_template_name}-asg", + 'PropagateAtLaunch': True, + 'ResourceId': f"{launch_template_name}-asg", + 'ResourceType': 'auto-scaling-group' + } + ] + ) + + self.logger.info(f"Created Auto Scaling Group: {launch_template_name}-asg") + return f"{launch_template_name}-asg" + + except Exception as e: + self.logger.error(f"Failed to create Auto Scaling Group: {e}") + raise + + def setup_cloudwatch_monitoring(self, instance_ids: List[str]) -> List[str]: + """Setup CloudWatch monitoring and alarms""" + + cloudwatch = self.session.client('cloudwatch') + alarm_names = [] + + for instance_id in instance_ids: + try: + # CPU Utilization Alarm + cpu_alarm_name = f"{instance_id}-high-cpu" + cloudwatch.put_metric_alarm( + AlarmName=cpu_alarm_name, + ComparisonOperator='GreaterThanThreshold', + EvaluationPeriods=2, + MetricName='CPUUtilization', + Namespace='AWS/EC2', + Period=300, + Statistic='Average', + Threshold=80.0, + ActionsEnabled=True, + AlarmDescription='High CPU utilization', + Dimensions=[ + { + 'Name': 'InstanceId', + 'Value': instance_id + } + ] + ) + alarm_names.append(cpu_alarm_name) + + # Memory Utilization Alarm (requires CloudWatch agent) + memory_alarm_name = f"{instance_id}-high-memory" + cloudwatch.put_metric_alarm( + AlarmName=memory_alarm_name, + ComparisonOperator='GreaterThanThreshold', + EvaluationPeriods=2, + MetricName='MemoryUtilization', + Namespace='CWAgent', + Period=300, + Statistic='Average', + Threshold=85.0, + ActionsEnabled=True, + AlarmDescription='High memory utilization', + Dimensions=[ + { + 'Name': 'InstanceId', + 'Value': instance_id + } + ] + ) + alarm_names.append(memory_alarm_name) + + except Exception as e: + self.logger.error(f"Failed to create alarms for {instance_id}: {e}") + + return alarm_names + + def implement_cost_optimization(self) -> Dict[str, Any]: + """Implement cost optimization strategies""" + + optimization_results = { + 'unused_ebs_volumes': [], + 'unattached_eips': [], + 'old_snapshots': [], + 'rightsizing_recommendations': [] + } + + try: + # Find unused EBS volumes + volumes = self.ec2.describe_volumes( + Filters=[{'Name': 'state', 'Values': ['available']}] + ) + for volume in volumes['Volumes']: + optimization_results['unused_ebs_volumes'].append({ + 'volume_id': volume['VolumeId'], + 'size': volume['Size'], + 'volume_type': volume['VolumeType'], + 'created_time': volume['CreateTime'] + }) + + # Find unattached Elastic IPs + addresses = self.ec2.describe_addresses() + for address in addresses['Addresses']: + if 'InstanceId' not in address: + optimization_results['unattached_eips'].append({ + 'allocation_id': address.get('AllocationId'), + 'public_ip': address['PublicIp'] + }) + + # Find old snapshots (older than 30 days) + thirty_days_ago = datetime.utcnow() - timedelta(days=30) + snapshots = self.ec2.describe_snapshots(OwnerIds=['self']) + for snapshot in snapshots['Snapshots']: + if snapshot['StartTime'].replace(tzinfo=None) < thirty_days_ago: + optimization_results['old_snapshots'].append({ + 'snapshot_id': snapshot['SnapshotId'], + 'start_time': snapshot['StartTime'], + 'volume_size': snapshot['VolumeSize'] + }) + + # Get rightsizing recommendations using Cost Explorer + try: + recommendations = self.cost_explorer.get_rightsizing_recommendation( + Service='AmazonEC2', + Configuration={ + 'BenefitsConsidered': True, + 'RecommendationTarget': 'SAME_INSTANCE_FAMILY' + } + ) + optimization_results['rightsizing_recommendations'] = recommendations.get('RightsizingRecommendations', []) + except Exception as e: + self.logger.warning(f"Could not get rightsizing recommendations: {e}") + + return optimization_results + + except Exception as e: + self.logger.error(f"Cost optimization analysis failed: {e}") + raise + +class LambdaManager: + """AWS Lambda management and deployment""" + + def __init__(self, region: str = 'us-east-1'): + self.lambda_client = boto3.client('lambda', region_name=region) + self.iam_client = boto3.client('iam', region_name=region) + self.logger = logging.getLogger(__name__) + + def create_lambda_function(self, + function_name: str, + runtime: str, + handler: str, + code_zip: bytes, + role_arn: str, + environment_vars: Dict[str, str] = None) -> str: + """Create and deploy Lambda function""" + + try: + response = self.lambda_client.create_function( + FunctionName=function_name, + Runtime=runtime, + Role=role_arn, + Handler=handler, + Code={'ZipFile': code_zip}, + Timeout=30, + MemorySize=128, + Environment={ + 'Variables': environment_vars or {} + }, + Tags={ + 'Environment': 'production', + 'ManagedBy': 'python-devops' + } + ) + + function_arn = response['FunctionArn'] + self.logger.info(f"Created Lambda function: {function_name}") + return function_arn + + except Exception as e: + self.logger.error(f"Failed to create Lambda function: {e}") + raise + + def setup_lambda_trigger(self, + function_name: str, + trigger_type: str, + trigger_config: Dict[str, Any]) -> str: + """Setup Lambda triggers (S3, CloudWatch Events, etc.)""" + + try: + if trigger_type == 's3': + # Add S3 bucket notification + s3_client = boto3.client('s3') + bucket_name = trigger_config['bucket_name'] + + # Add permission for S3 to invoke Lambda + self.lambda_client.add_permission( + FunctionName=function_name, + StatementId=f"s3-trigger-{bucket_name}", + Action='lambda:InvokeFunction', + Principal='s3.amazonaws.com', + SourceArn=f"arn:aws:s3:::{bucket_name}" + ) + + # Configure S3 bucket notification + notification_config = { + 'LambdaConfigurations': [ + { + 'Id': f"lambda-trigger-{function_name}", + 'LambdaFunctionArn': self.lambda_client.get_function( + FunctionName=function_name + )['Configuration']['FunctionArn'], + 'Events': trigger_config.get('events', ['s3:ObjectCreated:*']) + } + ] + } + + s3_client.put_bucket_notification_configuration( + Bucket=bucket_name, + NotificationConfiguration=notification_config + ) + + elif trigger_type == 'cloudwatch_events': + # Create CloudWatch Events rule + events_client = boto3.client('events') + rule_name = trigger_config['rule_name'] + + events_client.put_rule( + Name=rule_name, + ScheduleExpression=trigger_config['schedule'], + State='ENABLED' + ) + + # Add Lambda as target + events_client.put_targets( + Rule=rule_name, + Targets=[ + { + 'Id': '1', + 'Arn': self.lambda_client.get_function( + FunctionName=function_name + )['Configuration']['FunctionArn'] + } + ] + ) + + # Add permission for CloudWatch Events + self.lambda_client.add_permission( + FunctionName=function_name, + StatementId=f"events-trigger-{rule_name}", + Action='lambda:InvokeFunction', + Principal='events.amazonaws.com', + SourceArn=f"arn:aws:events:*:*:rule/{rule_name}" + ) + + self.logger.info(f"Setup {trigger_type} trigger for {function_name}") + return f"{trigger_type}-trigger-configured" + + except Exception as e: + self.logger.error(f"Failed to setup Lambda trigger: {e}") + raise + +# Example usage +def main(): + # Initialize AWS manager + aws_manager = AWSManager(region='us-west-2') + + # Create VPC infrastructure + vpc_info = aws_manager.create_vpc_infrastructure() + print(f"Created VPC infrastructure: {vpc_info}") + + # Cost optimization analysis + cost_optimization = aws_manager.implement_cost_optimization() + print(f"Cost optimization opportunities: {cost_optimization}") + + # Lambda deployment example + lambda_manager = LambdaManager(region='us-west-2') + + # Example Lambda function code + lambda_code = """ +def lambda_handler(event, context): + print(f"Received event: {event}") + return { + 'statusCode': 200, + 'body': 'Hello from Lambda!' + } +""" + + # Create zip file (simplified) + import zipfile + import io + + zip_buffer = io.BytesIO() + with zipfile.ZipFile(zip_buffer, 'w') as zip_file: + zip_file.writestr('lambda_function.py', lambda_code) + + zip_buffer.seek(0) + code_zip = zip_buffer.read() + + # Would need to create IAM role first in real scenario + # function_arn = lambda_manager.create_lambda_function( + # function_name="example-function", + # runtime="python3.9", + # handler="lambda_function.lambda_handler", + # code_zip=code_zip, + # role_arn="arn:aws:iam::account:role/lambda-role" + # ) + +if __name__ == "__main__": + main() +``` + +## Azure Automation + +### Azure SDK for Python + +**DevOps Usage Scenario**: Managing Azure resources, deployments, and monitoring. + +```python +from azure.identity import DefaultAzureCredential +from azure.mgmt.resource import ResourceManagementClient +from azure.mgmt.compute import ComputeManagementClient +from azure.mgmt.network import NetworkManagementClient +from azure.mgmt.storage import StorageManagementClient +from azure.mgmt.monitor import MonitorManagementClient +from typing import Dict, List, Any +import logging + +class AzureManager: + """Comprehensive Azure resource management""" + + def __init__(self, subscription_id: str): + self.subscription_id = subscription_id + self.credential = DefaultAzureCredential() + self.logger = logging.getLogger(__name__) + + # Initialize clients + self.resource_client = ResourceManagementClient( + self.credential, subscription_id + ) + self.compute_client = ComputeManagementClient( + self.credential, subscription_id + ) + self.network_client = NetworkManagementClient( + self.credential, subscription_id + ) + self.storage_client = StorageManagementClient( + self.credential, subscription_id + ) + self.monitor_client = MonitorManagementClient( + self.credential, subscription_id + ) + + def create_resource_group(self, name: str, location: str) -> Dict[str, Any]: + """Create Azure Resource Group""" + try: + rg_result = self.resource_client.resource_groups.create_or_update( + name, + { + 'location': location, + 'tags': { + 'Environment': 'production', + 'ManagedBy': 'python-devops' + } + } + ) + + self.logger.info(f"Created resource group: {name}") + return { + 'name': rg_result.name, + 'location': rg_result.location, + 'id': rg_result.id + } + + except Exception as e: + self.logger.error(f"Failed to create resource group: {e}") + raise + + def deploy_vm_scale_set(self, + resource_group: str, + name: str, + location: str, + instance_count: int = 2) -> Dict[str, Any]: + """Deploy Virtual Machine Scale Set""" + + try: + # Create virtual network first + vnet_name = f"{name}-vnet" + self.network_client.virtual_networks.begin_create_or_update( + resource_group, + vnet_name, + { + 'location': location, + 'address_space': { + 'address_prefixes': ['10.0.0.0/16'] + }, + 'subnets': [{ + 'name': 'default', + 'address_prefix': '10.0.0.0/24' + }] + } + ).result() + + # Create VM Scale Set + vmss_parameters = { + 'location': location, + 'sku': { + 'name': 'Standard_B1s', + 'tier': 'Standard', + 'capacity': instance_count + }, + 'upgrade_policy': { + 'mode': 'Manual' + }, + 'virtual_machine_profile': { + 'os_profile': { + 'computer_name_prefix': name, + 'admin_username': 'azureuser', + 'linux_configuration': { + 'disable_password_authentication': True, + 'ssh': { + 'public_keys': [{ + 'path': '/home/azureuser/.ssh/authorized_keys', + 'key_data': 'ssh-rsa AAAAB3NzaC1yc2E...' # Replace with actual key + }] + } + } + }, + 'storage_profile': { + 'image_reference': { + 'publisher': 'Canonical', + 'offer': 'UbuntuServer', + 'sku': '18.04-LTS', + 'version': 'latest' + }, + 'os_disk': { + 'create_option': 'FromImage', + 'caching': 'ReadWrite', + 'managed_disk': { + 'storage_account_type': 'Standard_LRS' + } + } + }, + 'network_profile': { + 'network_interface_configurations': [{ + 'name': f"{name}-nic", + 'primary': True, + 'ip_configurations': [{ + 'name': f"{name}-ip", + 'subnet': { + 'id': f"/subscriptions/{self.subscription_id}/resourceGroups/{resource_group}/providers/Microsoft.Network/virtualNetworks/{vnet_name}/subnets/default" + } + }] + }] + } + } + } + + vmss_result = self.compute_client.virtual_machine_scale_sets.begin_create_or_update( + resource_group, + name, + vmss_parameters + ).result() + + self.logger.info(f"Created VM Scale Set: {name}") + return { + 'name': vmss_result.name, + 'id': vmss_result.id, + 'location': vmss_result.location, + 'capacity': vmss_result.sku.capacity + } + + except Exception as e: + self.logger.error(f"Failed to create VM Scale Set: {e}") + raise + + def setup_auto_scaling(self, + resource_group: str, + vmss_name: str, + min_capacity: int = 1, + max_capacity: int = 10) -> str: + """Setup auto-scaling for VM Scale Set""" + + try: + autoscale_setting_name = f"{vmss_name}-autoscale" + + # Get VM Scale Set resource ID + vmss = self.compute_client.virtual_machine_scale_sets.get( + resource_group, vmss_name + ) + + autoscale_setting = { + 'location': vmss.location, + 'profiles': [{ + 'name': 'default', + 'capacity': { + 'minimum': str(min_capacity), + 'maximum': str(max_capacity), + 'default': str(min_capacity) + }, + 'rules': [{ + 'metric_trigger': { + 'metric_name': 'Percentage CPU', + 'metric_namespace': 'microsoft.compute/virtualmachinescalesets', + 'metric_resource_uri': vmss.id, + 'time_grain': 'PT1M', + 'statistic': 'Average', + 'time_window': 'PT5M', + 'time_aggregation': 'Average', + 'operator': 'GreaterThan', + 'threshold': 70 + }, + 'scale_action': { + 'direction': 'Increase', + 'type': 'ChangeCount', + 'value': '1', + 'cooldown': 'PT5M' + } + }, { + 'metric_trigger': { + 'metric_name': 'Percentage CPU', + 'metric_namespace': 'microsoft.compute/virtualmachinescalesets', + 'metric_resource_uri': vmss.id, + 'time_grain': 'PT1M', + 'statistic': 'Average', + 'time_window': 'PT5M', + 'time_aggregation': 'Average', + 'operator': 'LessThan', + 'threshold': 30 + }, + 'scale_action': { + 'direction': 'Decrease', + 'type': 'ChangeCount', + 'value': '1', + 'cooldown': 'PT5M' + } + }] + }], + 'target_resource_uri': vmss.id, + 'enabled': True + } + + self.monitor_client.autoscale_settings.create_or_update( + resource_group, + autoscale_setting_name, + autoscale_setting + ) + + self.logger.info(f"Created autoscale setting: {autoscale_setting_name}") + return autoscale_setting_name + + except Exception as e: + self.logger.error(f"Failed to create autoscale setting: {e}") + raise + +# Example usage +def azure_example(): + azure_manager = AzureManager(subscription_id="your-subscription-id") + + # Create resource group + rg = azure_manager.create_resource_group( + name="python-devops-rg", + location="East US" + ) + + # Deploy VM Scale Set + vmss = azure_manager.deploy_vm_scale_set( + resource_group="python-devops-rg", + name="web-vmss", + location="East US", + instance_count=2 + ) + + # Setup auto-scaling + autoscale = azure_manager.setup_auto_scaling( + resource_group="python-devops-rg", + vmss_name="web-vmss", + min_capacity=1, + max_capacity=5 + ) +``` + +## Google Cloud Platform + +### GCP Client Libraries + +**DevOps Usage Scenario**: Managing GCP resources, GKE clusters, and BigQuery automation. + +```python +from google.cloud import compute_v1 +from google.cloud import container_v1 +from google.cloud import bigquery +from google.cloud import storage +import logging +from typing import Dict, List, Any + +class GCPManager: + """Comprehensive GCP resource management""" + + def __init__(self, project_id: str, zone: str = 'us-central1-a'): + self.project_id = project_id + self.zone = zone + self.logger = logging.getLogger(__name__) + + # Initialize clients + self.compute_client = compute_v1.InstancesClient() + self.container_client = container_v1.ClusterManagerClient() + self.bigquery_client = bigquery.Client(project=project_id) + self.storage_client = storage.Client(project=project_id) + + def create_gke_cluster(self, + cluster_name: str, + node_count: int = 3, + machine_type: str = 'e2-medium') -> Dict[str, Any]: + """Create GKE cluster""" + + try: + cluster_config = { + 'name': cluster_name, + 'initial_node_count': node_count, + 'node_config': { + 'machine_type': machine_type, + 'disk_size_gb': 100, + 'oauth_scopes': [ + 'https://www.googleapis.com/auth/cloud-platform' + ] + }, + 'master_auth': { + 'username': '', + 'password': '' + }, + 'logging_service': 'logging.googleapis.com/kubernetes', + 'monitoring_service': 'monitoring.googleapis.com/kubernetes', + 'network': 'default', + 'subnetwork': 'default' + } + + parent = f"projects/{self.project_id}/locations/{self.zone}" + + operation = self.container_client.create_cluster( + parent=parent, + cluster=cluster_config + ) + + self.logger.info(f"Creating GKE cluster: {cluster_name}") + return { + 'operation_name': operation.name, + 'cluster_name': cluster_name, + 'status': 'CREATING' + } + + except Exception as e: + self.logger.error(f"Failed to create GKE cluster: {e}") + raise + + def setup_bigquery_data_pipeline(self, + dataset_name: str, + table_name: str, + schema: List[Dict[str, str]]) -> str: + """Setup BigQuery data pipeline""" + + try: + # Create dataset + dataset_id = f"{self.project_id}.{dataset_name}" + dataset = bigquery.Dataset(dataset_id) + dataset.location = "US" + + dataset = self.bigquery_client.create_dataset( + dataset, exists_ok=True + ) + + # Create table + table_id = f"{dataset_id}.{table_name}" + + # Convert schema + bq_schema = [] + for field in schema: + bq_schema.append( + bigquery.SchemaField( + field['name'], + field['type'], + mode=field.get('mode', 'NULLABLE') + ) + ) + + table = bigquery.Table(table_id, schema=bq_schema) + table = self.bigquery_client.create_table(table, exists_ok=True) + + self.logger.info(f"Created BigQuery table: {table_id}") + return table_id + + except Exception as e: + self.logger.error(f"Failed to setup BigQuery pipeline: {e}") + raise + +# Example usage +def gcp_example(): + gcp_manager = GCPManager(project_id="your-project-id") + + # Create GKE cluster + cluster = gcp_manager.create_gke_cluster( + cluster_name="production-cluster", + node_count=3, + machine_type="e2-standard-2" + ) + + # Setup BigQuery pipeline + schema = [ + {'name': 'timestamp', 'type': 'TIMESTAMP'}, + {'name': 'user_id', 'type': 'STRING'}, + {'name': 'event_type', 'type': 'STRING'}, + {'name': 'data', 'type': 'JSON'} + ] + + table_id = gcp_manager.setup_bigquery_data_pipeline( + dataset_name="analytics", + table_name="user_events", + schema=schema + ) +``` + +## Multi-Cloud Management + +### Cloud-Agnostic Automation + +**DevOps Usage Scenario**: Managing resources across multiple cloud providers with unified interface. + +```python +from abc import ABC, abstractmethod +from typing import Dict, List, Any, Optional +from enum import Enum +import logging + +class CloudProvider(Enum): + AWS = "aws" + AZURE = "azure" + GCP = "gcp" + +class CloudResource(ABC): + """Abstract base class for cloud resources""" + + @abstractmethod + def create(self) -> Dict[str, Any]: + pass + + @abstractmethod + def delete(self) -> bool: + pass + + @abstractmethod + def get_status(self) -> str: + pass + +class MultiCloudManager: + """Unified interface for multi-cloud resource management""" + + def __init__(self): + self.providers = {} + self.logger = logging.getLogger(__name__) + + def add_provider(self, provider: CloudProvider, client: Any) -> None: + """Add cloud provider client""" + self.providers[provider] = client + + def deploy_across_clouds(self, deployment_config: Dict[str, Any]) -> Dict[str, Any]: + """Deploy resources across multiple clouds""" + + results = {} + + for provider_name, config in deployment_config.items(): + provider = CloudProvider(provider_name) + + if provider not in self.providers: + self.logger.warning(f"Provider {provider_name} not configured") + continue + + try: + if provider == CloudProvider.AWS: + result = self._deploy_aws_resources(config) + elif provider == CloudProvider.AZURE: + result = self._deploy_azure_resources(config) + elif provider == CloudProvider.GCP: + result = self._deploy_gcp_resources(config) + + results[provider_name] = result + + except Exception as e: + self.logger.error(f"Deployment failed for {provider_name}: {e}") + results[provider_name] = {'error': str(e)} + + return results + + def get_cost_analysis(self) -> Dict[str, Any]: + """Get cost analysis across all providers""" + + cost_data = {} + + for provider, client in self.providers.items(): + try: + if provider == CloudProvider.AWS: + cost_data[provider.value] = self._get_aws_costs(client) + elif provider == CloudProvider.AZURE: + cost_data[provider.value] = self._get_azure_costs(client) + elif provider == CloudProvider.GCP: + cost_data[provider.value] = self._get_gcp_costs(client) + + except Exception as e: + self.logger.error(f"Cost analysis failed for {provider.value}: {e}") + cost_data[provider.value] = {'error': str(e)} + + return cost_data + + def _deploy_aws_resources(self, config: Dict[str, Any]) -> Dict[str, Any]: + """Deploy AWS resources""" + # Implementation for AWS deployment + return {'status': 'deployed', 'provider': 'aws'} + + def _deploy_azure_resources(self, config: Dict[str, Any]) -> Dict[str, Any]: + """Deploy Azure resources""" + # Implementation for Azure deployment + return {'status': 'deployed', 'provider': 'azure'} + + def _deploy_gcp_resources(self, config: Dict[str, Any]) -> Dict[str, Any]: + """Deploy GCP resources""" + # Implementation for GCP deployment + return {'status': 'deployed', 'provider': 'gcp'} + + def _get_aws_costs(self, client) -> Dict[str, Any]: + """Get AWS cost data""" + # Implementation for AWS cost analysis + return {'total_cost': 1000, 'currency': 'USD'} + + def _get_azure_costs(self, client) -> Dict[str, Any]: + """Get Azure cost data""" + # Implementation for Azure cost analysis + return {'total_cost': 800, 'currency': 'USD'} + + def _get_gcp_costs(self, client) -> Dict[str, Any]: + """Get GCP cost data""" + # Implementation for GCP cost analysis + return {'total_cost': 600, 'currency': 'USD'} + +# Example usage +def multi_cloud_example(): + multi_cloud = MultiCloudManager() + + # Add providers (would initialize actual clients in real scenario) + multi_cloud.add_provider(CloudProvider.AWS, "aws_client") + multi_cloud.add_provider(CloudProvider.AZURE, "azure_client") + multi_cloud.add_provider(CloudProvider.GCP, "gcp_client") + + # Deploy across clouds + deployment_config = { + 'aws': { + 'region': 'us-east-1', + 'instance_type': 't3.micro', + 'count': 2 + }, + 'azure': { + 'location': 'East US', + 'vm_size': 'Standard_B1s', + 'count': 2 + }, + 'gcp': { + 'zone': 'us-central1-a', + 'machine_type': 'e2-micro', + 'count': 2 + } + } + + deployment_results = multi_cloud.deploy_across_clouds(deployment_config) + print(f"Deployment results: {deployment_results}") + + # Get cost analysis + cost_analysis = multi_cloud.get_cost_analysis() + print(f"Cost analysis: {cost_analysis}") + +if __name__ == "__main__": + multi_cloud_example() +``` + +This guide covers comprehensive cloud platform integration including: + +1. **AWS Automation**: Complete infrastructure management, Lambda deployment, cost optimization +2. **Azure Automation**: Resource groups, VM scale sets, auto-scaling configuration +3. **GCP Management**: GKE clusters, BigQuery pipelines, storage management +4. **Multi-Cloud**: Unified interface for managing resources across all major cloud providers + +Each section provides production-ready code examples that DevOps engineers can adapt for their specific cloud automation needs. \ No newline at end of file diff --git a/python_for_devops_mastery_topics.md b/python_for_devops_mastery_topics.md new file mode 100644 index 0000000..dd414f6 --- /dev/null +++ b/python_for_devops_mastery_topics.md @@ -0,0 +1,406 @@ +# Python Mastery Topics for Experienced DevOps Engineers + +## Core Python Programming Concepts + +### 1. Advanced Python Language Features +- **Object-Oriented Programming (OOP)** + - Classes, inheritance, polymorphism, encapsulation + - Abstract base classes and mixins + - Metaclasses and descriptors + - Design patterns (Singleton, Factory, Observer, Strategy) + +- **Functional Programming** + - Lambda functions and closures + - Decorators and context managers + - Generators and iterators + - Map, filter, reduce operations + - Partial functions and currying + +- **Concurrency and Parallelism** + - Threading and multiprocessing + - Asyncio and async/await patterns + - Concurrent.futures module + - Event loops and coroutines + - Thread-safe programming and locks + +- **Error Handling and Debugging** + - Exception handling best practices + - Custom exceptions + - Logging frameworks and best practices + - Debugging techniques (pdb, profiling) + - Unit testing and mocking + +## Infrastructure Automation and Management + +### 2. Configuration Management +- **YAML and JSON Processing** + - PyYAML for configuration files + - JSON schema validation + - Configuration templating with Jinja2 + - Environment-specific configurations + +- **Infrastructure as Code (IaC)** + - Terraform integration with Python + - CloudFormation template generation + - Pulumi for cloud infrastructure + - ARM templates automation + +- **Configuration Management Tools** + - Ansible automation and playbook development + - Salt states and pillars + - Fabric for remote execution + - Custom configuration management scripts + +### 3. Cloud Platform Integration +- **AWS Automation** + - Boto3 SDK mastery + - Lambda function development + - CloudFormation and CDK + - S3, EC2, RDS, and other service automation + - Cost optimization scripts + - AWS CLI automation + +- **Azure Automation** + - Azure SDK for Python + - Azure Resource Manager (ARM) + - Azure Functions + - Azure DevOps integration + +- **Google Cloud Platform** + - Google Cloud Client Libraries + - Cloud Functions development + - GKE and container management + - BigQuery automation + +- **Multi-Cloud Management** + - Cloud-agnostic automation scripts + - Resource comparison and migration + - Cost optimization across platforms + +## Container and Orchestration Technologies + +### 4. Docker and Containerization +- **Docker SDK for Python** + - Container lifecycle management + - Image building and registry operations + - Volume and network management + - Multi-stage build automation + +- **Container Security** + - Vulnerability scanning automation + - Secret management + - Security policy enforcement + - Compliance checking scripts + +### 5. Kubernetes Orchestration +- **Kubernetes Python Client** + - Cluster resource management + - Pod, Service, and Deployment automation + - Custom Resource Definitions (CRDs) + - RBAC and security policies + +- **Kubernetes Operators** + - Operator development with Kopf + - Custom controller patterns + - Helm chart automation + - GitOps implementation + +- **Service Mesh Management** + - Istio configuration automation + - Traffic management scripts + - Security policy automation + +## CI/CD and DevOps Pipeline Automation + +### 6. Continuous Integration/Continuous Deployment +- **Pipeline Automation** + - Jenkins integration and job automation + - GitHub Actions with Python + - GitLab CI/CD pipeline scripts + - Azure DevOps automation + - TeamCity integration + +- **Build and Deployment Automation** + - Automated testing frameworks + - Artifact management + - Blue-green deployment scripts + - Canary deployment automation + - Rollback mechanisms + +- **GitOps Implementation** + - Git repository automation + - Configuration drift detection + - Automated synchronization + - Branch management strategies + +### 7. Testing and Quality Assurance +- **Infrastructure Testing** + - Testinfra for infrastructure testing + - Molecule for Ansible testing + - Terratest for infrastructure validation + - Chaos engineering with Python + +- **Application Testing** + - Pytest framework mastery + - Integration testing strategies + - Performance testing automation + - Security testing scripts + +## Monitoring, Observability, and Alerting + +### 8. Monitoring and Metrics +- **Metrics Collection and Analysis** + - Prometheus integration and custom metrics + - InfluxDB time-series data handling + - Grafana dashboard automation + - Custom monitoring solutions + +- **Log Management** + - ELK Stack integration (Elasticsearch, Logstash, Kibana) + - Fluentd log forwarding + - Log parsing and analysis + - Centralized logging solutions + +- **Application Performance Monitoring** + - APM tool integration + - Custom performance metrics + - Distributed tracing + - Resource usage optimization + +### 9. Alerting and Incident Management +- **Alert Management** + - PagerDuty integration + - Slack/Teams notifications + - Email alerting systems + - Custom alert correlation + +- **Incident Response** + - Automated incident detection + - Runbook automation + - Post-incident analysis scripts + - SLA monitoring and reporting + +## Security and Compliance + +### 10. DevSecOps Integration +- **Security Automation** + - Vulnerability scanning automation + - Secret management (HashiCorp Vault, AWS Secrets Manager) + - Security policy enforcement + - Compliance checking scripts + +- **Access Management** + - IAM automation + - RBAC implementation + - SSH key management + - Certificate automation + +- **Security Monitoring** + - Security event correlation + - Threat detection scripts + - Audit log analysis + - Compliance reporting + +## Database and Data Management + +### 11. Database Automation +- **Database Management** + - Database backup automation + - Schema migration scripts + - Performance monitoring + - Connection pooling and optimization + +- **Popular Database Libraries** + - SQLAlchemy ORM + - psycopg2 for PostgreSQL + - PyMongo for MongoDB + - Redis integration + +## Network and System Administration + +### 12. Network Automation +- **Network Management** + - Paramiko for SSH automation + - SNMP monitoring + - Network configuration automation + - Load balancer management + +- **System Administration** + - psutil for system monitoring + - Process management automation + - Service management scripts + - Performance optimization + +### 13. API Development and Integration +- **API Development** + - Flask/FastAPI for microservices + - RESTful API design + - GraphQL integration + - API documentation automation + +- **Third-Party Integrations** + - Webhook handling + - External service integration + - Rate limiting and throttling + - API security implementation + +## Advanced DevOps Practices + +### 14. Machine Learning Operations (MLOps) +- **ML Pipeline Automation** + - Model training automation + - Model deployment scripts + - A/B testing frameworks + - Model monitoring and drift detection + +- **Data Pipeline Management** + - Apache Airflow workflows + - ETL process automation + - Data quality validation + - Stream processing with Kafka + +### 15. Performance Optimization and Scaling +- **Application Performance** + - Code profiling and optimization + - Memory management + - Caching strategies (Redis, Memcached) + - Database query optimization + +- **Infrastructure Scaling** + - Auto-scaling automation + - Load testing frameworks + - Capacity planning scripts + - Resource optimization + +## Essential Python Libraries for DevOps + +### 16. Core Libraries to Master +- **Infrastructure and Cloud** + - `boto3` (AWS) + - `azure-sdk-for-python` (Azure) + - `google-cloud` (GCP) + - `kubernetes` (K8s client) + - `docker` (Docker SDK) + +- **Configuration and Automation** + - `ansible` (Configuration management) + - `fabric` (Remote execution) + - `paramiko` (SSH client) + - `pyyaml` (YAML processing) + - `jinja2` (Templating) + +- **Monitoring and Observability** + - `prometheus_client` (Metrics) + - `elasticsearch` (Log management) + - `psutil` (System monitoring) + - `requests` (HTTP client) + +- **Data Processing** + - `pandas` (Data analysis) + - `numpy` (Numerical computing) + - `sqlalchemy` (Database ORM) + +- **Testing and Quality** + - `pytest` (Testing framework) + - `mock` (Mocking library) + - `coverage` (Code coverage) + +- **Web and API** + - `flask`/`fastapi` (Web frameworks) + - `celery` (Task queue) + - `gunicorn` (WSGI server) + +## Best Practices and Methodologies + +### 17. Code Quality and Maintenance +- **Code Standards** + - PEP 8 compliance + - Type hints and mypy + - Docstring conventions + - Code review practices + +- **Version Control** + - Git workflows and branching strategies + - Semantic versioning + - Release management + - Automated changelog generation + +### 18. Security Best Practices +- **Secure Coding** + - Input validation and sanitization + - Secure API development + - Cryptography implementation + - Secret management + +- **Compliance and Auditing** + - SOC 2 compliance automation + - GDPR compliance scripts + - Audit trail implementation + - Regulatory reporting + +## Advanced Topics + +### 19. Emerging Technologies +- **AI/ML Integration** + - TensorFlow/PyTorch for DevOps + - Natural language processing for log analysis + - Predictive analytics for infrastructure + - Automated anomaly detection + +- **Edge Computing** + - IoT device management + - Edge deployment automation + - Distributed computing patterns + +### 20. Architecture and Design Patterns +- **Microservices Architecture** + - Service discovery automation + - Circuit breaker patterns + - Distributed system debugging + - Event-driven architecture + +- **Serverless Computing** + - AWS Lambda development + - Azure Functions + - Google Cloud Functions + - Serverless framework automation + +## Recommended Learning Path + +1. **Foundation** (Weeks 1-4): Core Python concepts, OOP, and basic automation +2. **Infrastructure** (Weeks 5-8): Cloud platforms, IaC, and configuration management +3. **Containers** (Weeks 9-12): Docker, Kubernetes, and orchestration +4. **CI/CD** (Weeks 13-16): Pipeline automation, testing, and deployment +5. **Monitoring** (Weeks 17-20): Observability, alerting, and incident management +6. **Security** (Weeks 21-24): DevSecOps, compliance, and security automation +7. **Advanced** (Weeks 25-28): MLOps, performance optimization, and emerging technologies +8. **Mastery** (Weeks 29-32): Complex projects, integration, and leadership skills + +## Practical Projects to Build + +1. **Infrastructure Automation Platform**: Multi-cloud resource management system +2. **CI/CD Pipeline Framework**: Customizable deployment automation +3. **Monitoring and Alerting System**: Comprehensive observability platform +4. **Security Compliance Dashboard**: Automated security scanning and reporting +5. **MLOps Pipeline**: End-to-end machine learning deployment system +6. **Disaster Recovery Automation**: Backup, restore, and failover systems +7. **Cost Optimization Engine**: Cloud resource optimization and reporting +8. **Configuration Management System**: Custom tool for infrastructure management + +## Career Development + +### Certifications to Consider +- AWS Certified DevOps Engineer +- Azure DevOps Engineer Expert +- Google Cloud Professional DevOps Engineer +- Certified Kubernetes Administrator (CKA) +- Certified Information Systems Security Professional (CISSP) + +### Community Engagement +- Contribute to open-source DevOps projects +- Speak at conferences and meetups +- Write technical blogs and tutorials +- Mentor junior engineers +- Participate in DevOps communities and forums + +This comprehensive list provides a structured approach to mastering Python for DevOps engineering, building on existing experience while introducing advanced concepts and emerging technologies. \ No newline at end of file