Complete DevOps Automation Project

1┌─────────────────────────────────────────────────────────┐
2│                    API Gateway Layer                     │
3│  ┌──────────────────────────────────────────────────┐  │
4│  │  FastAPI + JWT Auth + Rate Limiting + RBAC      │  │
5│  └──────────────────────────────────────────────────┘  │
6└─────────────────────────────────────────────────────────┘
7                          │
8          ┌───────────────┼───────────────┐
9          │               │               │
10┌─────────▼─────────┐ ┌──▼──────────┐ ┌─▼──────────────┐
11│ Infrastructure    │ │  CI/CD      │ │  Monitoring    │
12│ Manager           │ │  Orchestrator│ │  System        │
13│                   │ │             │ │                │
14│ • AWS/Azure/GCP   │ │ • Jenkins   │ │ • Prometheus   │
15│ • Terraform       │ │ • GitHub    │ │ • Grafana      │
16│ • CloudFormation  │ │ • GitLab    │ │ • Alerting     │
17└─────────┬─────────┘ └──┬──────────┘ └─┬──────────────┘
18          │               │               │
19          └───────────────┼───────────────┘
20                          │
21          ┌───────────────┼───────────────┐
22          │               │               │
23┌─────────▼─────────┐ ┌──▼──────────┐ ┌─▼──────────────┐
24│ Container         │ │  Security   │ │  Disaster      │
25│ Orchestration     │ │  Scanner    │ │  Recovery      │
26│                   │ │             │ │                │
27│ • Docker          │ │ • Trivy     │ │ • Backups      │
28│ • Kubernetes      │ │ • OWASP ZAP │ │ • Failover     │
29│ • Helm            │ │ • Secrets   │ │ • DR Testing   │
30└───────────────────┘ └─────────────┘ └────────────────┘

1from dataclasses import dataclass
2from typing import Dict, List, Optional
3from enum import Enum
4
5class ComponentStatus(Enum):
6    HEALTHY = "healthy"
7    DEGRADED = "degraded"
8    UNHEALTHY = "unhealthy"
9
10@dataclass
11class InfrastructureConfig:
12    """Infrastructure configuration"""
13    cloud_provider: str  # aws, azure, gcp
14    region: str
15    vpc_cidr: str
16    cluster_name: str
17    node_count: int
18    node_type: str
19
20@dataclass
21class ApplicationConfig:
22    """Application deployment configuration"""
23    name: str
24    version: str
25    image: str
26    replicas: int
27    environment: Dict[str, str]
28    resources: Dict[str, str]
29
30class DevOpsPlatform:
31    """
32    Complete DevOps automation platform
33    Integrates all course concepts into unified system
34    """
35
36    def __init__(self, config_path: str):
37        self.infra_manager = InfrastructureManager()
38        self.container_orchestrator = KubernetesOrchestrator()
39        self.cicd_pipeline = CICDPipeline()
40        self.monitoring_system = MonitoringSystem()
41        self.security_scanner = SecurityScanner()
42        self.dr_manager = DisasterRecoveryManager()
43
44    def deploy_full_stack(
45        self,
46        infra_config: InfrastructureConfig,
47        app_config: ApplicationConfig
48    ) -> Dict:
49        """
50        Deploy complete application stack
51
52        Workflow:
53        1. Provision infrastructure (clouds + networking)
54        2. Deploy Kubernetes cluster
55        3. Configure monitoring and logging
56        4. Set up security policies
57        5. Deploy application
58        6. Configure DR and backups
59        7. Run health checks and tests
60        """
61        # Will implement in practical example
62        pass

1class ServiceRegistry:
2    """
3    Service discovery and registration
4    Enables components to find each other
5    """
6
7    def register_service(
8        self,
9        service_name: str,
10        endpoint: str,
11        health_check: str
12    ):
13        # Register service with Consul/etcd
14        pass
15
16    def discover_service(self, service_name: str) -> str:
17        # Look up service endpoint
18        pass

1from typing import Callable
2import asyncio
3
4class EventBus:
5    """
6    Event bus for component communication
7    Decouples components with async messaging
8    """
9
10    def __init__(self):
11        self.subscribers: Dict[str, List[Callable]] = {}
12
13    def subscribe(self, event_type: str, handler: Callable):
14        if event_type not in self.subscribers:
15            self.subscribers[event_type] = []
16        self.subscribers[event_type].append(handler)
17
18    async def publish(self, event_type: str, data: Dict):
19        if event_type in self.subscribers:
20            await asyncio.gather(*[
21                handler(data)
22                for handler in self.subscribers[event_type]
23            ])
24
25# Usage
26event_bus = EventBus()
27
28async def on_deployment_complete(data: Dict):
29    # Trigger monitoring setup
30    await monitoring.setup_for_deployment(data['deployment_id'])
31    # Trigger security scan
32    await security.scan_deployment(data['image'])
33
34event_bus.subscribe('deployment.complete', on_deployment_complete)

1import os
2from pathlib import Path
3import yaml
4from typing import Any
5
6class ConfigManager:
7    """
8    Centralized configuration management
9    Supports multiple environments and secret management
10    """
11
12    def __init__(self, config_dir: str = "./config"):
13        self.config_dir = Path(config_dir)
14        self.env = os.getenv('ENVIRONMENT', 'development')
15        self.config = self._load_config()
16
17    def _load_config(self) -> Dict[str, Any]:
18        # Load base config
19        base_config = self._load_yaml('base.yaml')
20
21        # Load environment-specific config
22        env_config = self._load_yaml(f'{self.env}.yaml')
23
24        # Merge configs (env overrides base)
25        return {**base_config, **env_config}
26
27    def _load_yaml(self, filename: str) -> Dict:
28        config_file = self.config_dir / filename
29        if not config_file.exists():
30            return {}
31        with open(config_file) as f:
32            return yaml.safe_load(f)
33
34    def get(self, key: str, default: Any = None) -> Any:
35        keys = key.split('.')
36        value = self.config
37        for k in keys:
38            if isinstance(value, dict):
39                value = value.get(k)
40            else:
41                return default
42        return value if value is not None else default
43
44# Usage
45config = ConfigManager()
46aws_region = config.get('aws.region', 'us-east-1')
47cluster_size = config.get('kubernetes.cluster.size', 3)

1import json
2from typing import Any
3from datetime import datetime
4
5class StateManager:
6    """
7    Manage platform state across operations
8    Enables idempotent operations and recovery
9    """
10
11    def __init__(self, state_backend: str = "local"):
12        self.state_backend = state_backend  # local, s3, consul
13        self.state = self._load_state()
14
15    def _load_state(self) -> Dict[str, Any]:
16        if self.state_backend == "local":
17            try:
18                with open('.devops-state.json') as f:
19                    return json.load(f)
20            except FileNotFoundError:
21                return {}
22        # Implement S3/Consul backends
23        return {}
24
25    def save_state(self):
26        if self.state_backend == "local":
27            with open('.devops-state.json', 'w') as f:
28                json.dump(self.state, f, indent=2)
29
30    def update(self, key: str, value: Any):
31        self.state[key] = {
32            'value': value,
33            'timestamp': datetime.now().isoformat(),
34            'version': self.state.get(key, {}).get('version', 0) + 1
35        }
36        self.save_state()
37
38    def get(self, key: str, default: Any = None) -> Any:
39        return self.state.get(key, {}).get('value', default)
40
41# Usage
42state = StateManager()
43state.update('infrastructure.vpc_id', 'vpc-123456')
44state.update('kubernetes.cluster_endpoint', 'https://...')
45vpc_id = state.get('infrastructure.vpc_id')

1from typing import Dict, Any
2import asyncio
3from datetime import datetime
4
5class DeploymentWorkflow:
6    """
7    End-to-end deployment automation workflow
8    Integrates all platform components
9    """
10
11    def __init__(self, platform: DevOpsPlatform):
12        self.platform = platform
13        self.logger = logging.getLogger(__name__)
14
15    async def deploy_application(
16        self,
17        app_name: str,
18        version: str,
19        environment: str
20    ) -> Dict[str, Any]:
21        """
22        Complete application deployment workflow
23
24        Steps:
25        1. Pre-deployment validation
26        2. Infrastructure provisioning/validation
27        3. Build and test application
28        4. Security scanning
29        5. Deploy to Kubernetes
30        6. Configure monitoring
31        7. Run smoke tests
32        8. Update DNS/load balancers
33        9. Post-deployment validation
34        """
35
36        deployment_id = f"{app_name}-{version}-{int(datetime.now().timestamp())}"
37        self.logger.info(f"🚀 Starting deployment: {deployment_id}")
38
39        try:
40            # Step 1: Pre-deployment validation
41            await self._validate_prerequisites(app_name, environment)
42
43            # Step 2: Ensure infrastructure exists
44            await self._ensure_infrastructure(environment)
45
46            # Step 3: Build application
47            build_result = await self._build_application(app_name, version)
48            image_uri = build_result['image_uri']
49
50            # Step 4: Security scan
51            scan_result = await self._security_scan(image_uri)
52            if scan_result['critical_vulns'] > 0:
53                raise Exception(f"Critical vulnerabilities found: {scan_result['critical_vulns']}")
54
55            # Step 5: Deploy to Kubernetes
56            deploy_result = await self._deploy_to_kubernetes(
57                app_name, version, image_uri, environment
58            )
59
60            # Step 6: Configure monitoring
61            await self._setup_monitoring(app_name, environment)
62
63            # Step 7: Run smoke tests
64            await self._run_smoke_tests(app_name, environment)
65
66            # Step 8: Update traffic routing
67            await self._update_routing(app_name, environment)
68
69            # Step 9: Post-deployment validation
70            await self._validate_deployment(app_name, environment)
71
72            self.logger.info(f"✅ Deployment successful: {deployment_id}")
73
74            return {
75                'deployment_id': deployment_id,
76                'status': 'success',
77                'app_name': app_name,
78                'version': version,
79                'environment': environment,
80                'image_uri': image_uri,
81                'timestamp': datetime.now().isoformat()
82            }
83
84        except Exception as e:
85            self.logger.error(f"❌ Deployment failed: {e}")
86            # Rollback on failure
87            await self._rollback_deployment(app_name, environment)
88            raise
89
90    async def _validate_prerequisites(self, app_name: str, environment: str):
91        """Validate all prerequisites before deployment"""
92        self.logger.info("🔍 Validating prerequisites...")
93
94        # Check configuration exists
95        if not self.platform.config.get(f'apps.{app_name}'):
96            raise Exception(f"No configuration found for {app_name}")
97
98        # Check credentials
99        if environment == 'production':
100            if not self._has_production_approvals(app_name):
101                raise Exception("Production deployment requires approval")
102
103        # Check cluster health
104        cluster_status = await self.platform.k8s.get_cluster_health()
105        if cluster_status != 'healthy':
106            raise Exception(f"Cluster unhealthy: {cluster_status}")
107
108        self.logger.info("✅ Prerequisites validated")
109
110    async def _ensure_infrastructure(self, environment: str):
111        """Ensure required infrastructure exists"""
112        self.logger.info("🏗️  Ensuring infrastructure...")
113
114        # Check if VPC exists
115        vpc_id = self.platform.state.get(f'{environment}.vpc_id')
116        if not vpc_id:
117            # Provision VPC
118            vpc_result = await self.platform.infra.provision_vpc(environment)
119            vpc_id = vpc_result['vpc_id']
120            self.platform.state.update(f'{environment}.vpc_id', vpc_id)
121
122        # Check if Kubernetes cluster exists
123        cluster_endpoint = self.platform.state.get(f'{environment}.cluster_endpoint')
124        if not cluster_endpoint:
125            # Provision cluster
126            cluster_result = await self.platform.k8s.provision_cluster(
127                environment=environment,
128                vpc_id=vpc_id
129            )
130            self.platform.state.update(
131                f'{environment}.cluster_endpoint',
132                cluster_result['endpoint']
133            )
134
135        self.logger.info("✅ Infrastructure ready")
136
137    async def _build_application(self, app_name: str, version: str) -> Dict:
138        """Build application container image"""
139        self.logger.info(f"🔨 Building {app_name}:{version}...")
140
141        # Trigger CI/CD pipeline
142        build_result = await self.platform.cicd.trigger_build(
143            app_name=app_name,
144            version=version,
145            git_ref=f"v{version}"
146        )
147
148        # Wait for build to complete
149        while build_result['status'] == 'building':
150            await asyncio.sleep(10)
151            build_result = await self.platform.cicd.get_build_status(
152                build_result['build_id']
153            )
154
155        if build_result['status'] != 'success':
156            raise Exception(f"Build failed: {build_result['error']}")
157
158        self.logger.info(f"✅ Build complete: {build_result['image_uri']}")
159        return build_result
160
161    async def _security_scan(self, image_uri: str) -> Dict:
162        """Scan container image for vulnerabilities"""
163        self.logger.info(f"🔒 Scanning {image_uri}...")
164
165        scan_result = await self.platform.security.scan_image(image_uri)
166
167        self.logger.info(
168            f"Vulnerabilities found: "
169            f"Critical={scan_result['critical_vulns']}, "
170            f"High={scan_result['high_vulns']}, "
171            f"Medium={scan_result['medium_vulns']}"
172        )
173
174        return scan_result
175
176    async def _deploy_to_kubernetes(
177        self,
178        app_name: str,
179        version: str,
180        image_uri: str,
181        environment: str
182    ) -> Dict:
183        """Deploy application to Kubernetes"""
184        self.logger.info(f"☸️  Deploying to Kubernetes ({environment})...")
185
186        # Create/update deployment
187        deploy_result = await self.platform.k8s.deploy_application(
188            name=app_name,
189            image=image_uri,
190            version=version,
191            namespace=environment,
192            replicas=self.platform.config.get(f'apps.{app_name}.replicas', 3),
193            resources=self.platform.config.get(f'apps.{app_name}.resources', {})
194        )
195
196        # Wait for rollout to complete
197        await self.platform.k8s.wait_for_rollout(
198            name=app_name,
199            namespace=environment,
200            timeout=300
201        )
202
203        self.logger.info("✅ Kubernetes deployment complete")
204        return deploy_result
205
206    async def _setup_monitoring(self, app_name: str, environment: str):
207        """Configure monitoring for application"""
208        self.logger.info("📊 Setting up monitoring...")
209
210        # Create Prometheus ServiceMonitor
211        await self.platform.monitoring.create_service_monitor(
212            app_name=app_name,
213            namespace=environment,
214            metrics_path='/metrics',
215            interval='30s'
216        )
217
218        # Create Grafana dashboard
219        await self.platform.monitoring.create_dashboard(
220            app_name=app_name,
221            environment=environment
222        )
223
224        # Configure alerts
225        await self.platform.monitoring.create_alerts(
226            app_name=app_name,
227            environment=environment,
228            rules=[
229                {
230                    'alert': 'HighErrorRate',
231                    'expr': f'rate(http_requests_total{{job="{app_name}",status=~"5.."}}[5m]) > 0.05',
232                    'for': '5m',
233                    'severity': 'critical'
234                },
235                {
236                    'alert': 'PodCrashLooping',
237                    'expr': f'rate(kube_pod_container_status_restarts_total{{pod=~"{app_name}.*"}}[15m]) > 0',
238                    'for': '5m',
239                    'severity': 'warning'
240                }
241            ]
242        )
243
244        self.logger.info("✅ Monitoring configured")
245
246    async def _run_smoke_tests(self, app_name: str, environment: str):
247        """Run smoke tests to validate deployment"""
248        self.logger.info("🧪 Running smoke tests...")
249
250        # Get service endpoint
251        service_url = await self.platform.k8s.get_service_url(
252            name=app_name,
253            namespace=environment
254        )
255
256        # Run basic health check
257        import aiohttp
258        async with aiohttp.ClientSession() as session:
259            async with session.get(f"{service_url}/health") as resp:
260                if resp.status != 200:
261                    raise Exception(f"Health check failed: {resp.status}")
262
263        # Run additional smoke tests from config
264        smoke_tests = self.platform.config.get(f'apps.{app_name}.smoke_tests', [])
265        for test in smoke_tests:
266            async with aiohttp.ClientSession() as session:
267                async with session.request(
268                    method=test['method'],
269                    url=f"{service_url}{test['path']}"
270                ) as resp:
271                    if resp.status not in test.get('expected_status', [200]):
272                        raise Exception(f"Smoke test failed: {test['path']}")
273
274        self.logger.info("✅ Smoke tests passed")
275
276    async def _update_routing(self, app_name: str, environment: str):
277        """Update traffic routing to new version"""
278        self.logger.info("🔀 Updating traffic routing...")
279
280        if environment == 'production':
281            # Blue-green deployment: gradually shift traffic
282            await self.platform.k8s.update_ingress(
283                app_name=app_name,
284                strategy='blue-green',
285                canary_weight=10  # Start with 10% traffic
286            )
287
288            # Monitor metrics for 5 minutes
289            await asyncio.sleep(300)
290
291            # If no errors, shift 100% traffic
292            await self.platform.k8s.update_ingress(
293                app_name=app_name,
294                strategy='blue-green',
295                canary_weight=100
296            )
297        else:
298            # Dev/staging: immediate cutover
299            await self.platform.k8s.update_ingress(
300                app_name=app_name,
301                strategy='replace'
302            )
303
304        self.logger.info("✅ Traffic routing updated")
305
306    async def _validate_deployment(self, app_name: str, environment: str):
307        """Validate deployment was successful"""
308        self.logger.info("✔️  Validating deployment...")
309
310        # Check all pods are running
311        pods_status = await self.platform.k8s.get_pods_status(
312            app_name=app_name,
313            namespace=environment
314        )
315
316        running_pods = sum(1 for p in pods_status if p['status'] == 'Running')
317        expected_pods = self.platform.config.get(f'apps.{app_name}.replicas', 3)
318
319        if running_pods < expected_pods:
320            raise Exception(f"Only {running_pods}/{expected_pods} pods running")
321
322        # Check metrics are being collected
323        metrics_available = await self.platform.monitoring.check_metrics_available(
324            app_name=app_name
325        )
326        if not metrics_available:
327            self.logger.warning("⚠️  Metrics not yet available")
328
329        self.logger.info("✅ Deployment validated")
330
331    async def _rollback_deployment(self, app_name: str, environment: str):
332        """Rollback failed deployment"""
333        self.logger.warning(f"⏪ Rolling back {app_name} in {environment}...")
334
335        try:
336            await self.platform.k8s.rollback_deployment(
337                name=app_name,
338                namespace=environment
339            )
340            self.logger.info("✅ Rollback complete")
341        except Exception as e:
342            self.logger.error(f"❌ Rollback failed: {e}")

1class InfrastructureWorkflow:
2    """Manage complete infrastructure lifecycle"""
3
4    async def provision_environment(
5        self,
6        environment: str,
7        region: str
8    ) -> Dict:
9        """
10        Provision complete environment infrastructure
11        - VPC with public/private subnets
12        - EKS/AKS/GKE cluster
13        - RDS/CloudSQL database
14        - ElastiCache/Redis
15        - S3/Blob storage
16        - IAM roles and policies
17        """
18        self.logger.info(f"🏗️  Provisioning {environment} environment...")
19
20        # Use Terraform/Pulumi for IaC
21        terraform_vars = {
22            'environment': environment,
23            'region': region,
24            'cluster_size': 3,
25            'node_type': 't3.medium',
26            'db_instance_type': 'db.t3.medium'
27        }
28
29        result = await self.platform.infra.apply_terraform(
30            workspace=environment,
31            variables=terraform_vars
32        )
33
34        return result

1from prometheus_client import Counter, Histogram, Gauge, generate_latest
2from typing import Dict, List
3import asyncio
4
5class MonitoringSystem:
6    """
7    Complete monitoring and observability system
8    Integrates metrics, logs, traces, and alerts
9    """
10
11    def __init__(self, config: ConfigManager):
12        self.config = config
13        self._init_metrics()
14
15    def _init_metrics(self):
16        """Initialize Prometheus metrics"""
17
18        # Deployment metrics
19        self.deployments_total = Counter(
20            'devops_deployments_total',
21            'Total number of deployments',
22            ['app', 'environment', 'status']
23        )
24
25        self.deployment_duration = Histogram(
26            'devops_deployment_duration_seconds',
27            'Deployment duration in seconds',
28            ['app', 'environment']
29        )
30
31        # Infrastructure metrics
32        self.cluster_nodes = Gauge(
33            'devops_cluster_nodes',
34            'Number of cluster nodes',
35            ['environment', 'status']
36        )
37
38        self.infrastructure_cost = Gauge(
39            'devops_infrastructure_cost_dollars',
40            'Estimated monthly infrastructure cost',
41            ['environment', 'service']
42        )
43
44        # Security metrics
45        self.vulnerabilities_found = Gauge(
46            'devops_vulnerabilities_total',
47            'Number of vulnerabilities found',
48            ['severity', 'app']
49        )
50
51        # Operational metrics
52        self.backup_age_hours = Gauge(
53            'devops_backup_age_hours',
54            'Age of latest backup in hours',
55            ['database', 'environment']
56        )
57
58    async def setup_for_deployment(self, deployment_id: str):
59        """Set up monitoring for new deployment"""
60        # Implementation from practical example
61        pass
62
63    def record_deployment(
64        self,
65        app: str,
66        environment: str,
67        status: str,
68        duration: float
69    ):
70        """Record deployment metrics"""
71        self.deployments_total.labels(
72            app=app,
73            environment=environment,
74            status=status
75        ).inc()
76
77        if status == 'success':
78            self.deployment_duration.labels(
79                app=app,
80                environment=environment
81            ).observe(duration)

1class IncidentResponse:
2    """
3    Automated incident detection and response
4    """
5
6    def __init__(self, platform: DevOpsPlatform):
7        self.platform = platform
8        self.alert_rules = self._load_alert_rules()
9
10    async def handle_alert(self, alert: Dict):
11        """Handle incoming alert"""
12        alert_name = alert['labels']['alertname']
13        severity = alert['labels']['severity']
14
15        self.logger.warning(
16            f"🚨 Alert triggered: {alert_name} (severity: {severity})"
17        )
18
19        # Auto-remediation based on alert type
20        if alert_name == 'PodCrashLooping':
21            await self._handle_pod_crash(alert)
22        elif alert_name == 'HighMemoryUsage':
23            await self._handle_high_memory(alert)
24        elif alert_name == 'DiskSpaceLow':
25            await self._handle_disk_space(alert)
26
27        # Always notify on-call team
28        await self._notify_oncall(alert)
29
30    async def _handle_pod_crash(self, alert: Dict):
31        """Auto-remediate crashing pods"""
32        pod_name = alert['labels']['pod']
33        namespace = alert['labels']['namespace']
34
35        # Get pod logs
36        logs = await self.platform.k8s.get_pod_logs(
37            pod=pod_name,
38            namespace=namespace,
39            tail=100
40        )
41
42        # Check for common issues
43        if 'OutOfMemoryError' in logs:
44            # Increase memory limit
45            await self.platform.k8s.update_resources(
46                deployment=pod_name.split('-')[0],
47                namespace=namespace,
48                memory_limit='512Mi'  # Increase from 256Mi
49            )
50        elif 'Connection refused' in logs:
51            # Check dependent services
52            await self._check_dependencies(pod_name, namespace)
53
54    async def _handle_high_memory(self, alert: Dict):
55        """Auto-scale on high memory usage"""
56        deployment = alert['labels']['deployment']
57        namespace = alert['labels']['namespace']
58
59        # Scale up if not at max
60        current_replicas = await self.platform.k8s.get_replica_count(
61            deployment=deployment,
62            namespace=namespace
63        )
64
65        max_replicas = self.platform.config.get(
66            f'apps.{deployment}.max_replicas',
67            10
68        )
69
70        if current_replicas < max_replicas:
71            await self.platform.k8s.scale_deployment(
72                deployment=deployment,
73                namespace=namespace,
74                replicas=current_replicas + 2
75            )
76            self.logger.info(
77                f"📈 Scaled {deployment} from {current_replicas} "
78                f"to {current_replicas + 2} replicas"
79            )

1class CostOptimization:
2    """Automated cost optimization"""
3
4    async def optimize_infrastructure(self, environment: str):
5        """Run cost optimization analysis"""
6
7        # Find idle resources
8        idle_resources = await self._find_idle_resources(environment)
9
10        # Recommend instance right-sizing
11        sizing_recommendations = await self._analyze_instance_sizing(environment)
12
13        # Identify unused volumes
14        unused_volumes = await self._find_unused_volumes(environment)
15
16        # Auto-apply low-risk optimizations
17        savings = 0
18        for resource in idle_resources:
19            if resource['idle_days'] > 7:
20                await self._terminate_resource(resource)
21                savings += resource['monthly_cost']
22
23        return {
24            'total_savings': savings,
25            'recommendations': sizing_recommendations,
26            'actions_taken': len(idle_resources)
27        }