Create deployment/blue_green.py

Adding Blue-Green Deployment Strategy

deployment/blue_green.py

@@ -0,0 +1,761 @@
+import json
+import time
+import logging
+import threading
+import numpy as np
+from enum import Enum
+from pathlib import Path
+from datetime import datetime, timedelta
+from dataclasses import dataclass, asdict
+from typing import Dict, List, Optional, Any, Tuple
+
+
+logger = logging.getLogger(__name__)
+
+class DeploymentStatus(Enum):
+    INACTIVE = "inactive"
+    PREPARING = "preparing"
+    STAGING = "staging"
+    DEPLOYING = "deploying"
+    ACTIVE = "active"
+    ROLLING_BACK = "rolling_back"
+    FAILED = "failed"
+    COMPLETED = "completed"
+
+@dataclass
+class ModelVersion:
+    """Model version metadata"""
+    version_id: str
+    model_path: str
+    vectorizer_path: str
+    metadata_path: str
+    created_at: str
+    status: str
+    performance_metrics: Dict[str, float]
+    deployment_config: Dict[str, Any]
+
+@dataclass
+class DeploymentPlan:
+    """Deployment plan configuration"""
+    deployment_id: str
+    source_version: str
+    target_version: str
+    strategy: str  # 'blue_green', 'canary', 'rolling'
+    traffic_stages: List[Dict[str, Any]]
+    health_checks: Dict[str, Any]
+    rollback_conditions: Dict[str, Any]
+    created_at: str
+    status: str
+
+class BlueGreenDeploymentManager:
+    """Manages blue-green deployments with traffic routing and health monitoring"""
+    
+    def __init__(self, base_dir: Path = None):
+        self.base_dir = base_dir or Path("/tmp")
+        self.setup_deployment_paths()
+        self.setup_deployment_config()
+        
+        # Current deployment state
+        self.current_deployment = None
+        self.active_version = None
+        self.staging_version = None
+        self.traffic_split = {"blue": 100, "green": 0}
+        
+        # Monitoring
+        self.deployment_monitor = None
+        self.monitor_thread = None
+        self.monitoring_active = False
+        
+        # Load existing state
+        self.load_deployment_state()
+    
+    def setup_deployment_paths(self):
+        """Setup deployment-specific paths"""
+        self.deployment_dir = self.base_dir / "deployment"
+        self.deployment_dir.mkdir(parents=True, exist_ok=True)
+        
+        # Model storage
+        self.models_dir = self.deployment_dir / "models"
+        self.models_dir.mkdir(parents=True, exist_ok=True)
+        
+        # Deployment logs
+        self.deployment_log_path = self.deployment_dir / "deployment_log.json"
+        self.deployment_state_path = self.deployment_dir / "deployment_state.json"
+        self.traffic_log_path = self.deployment_dir / "traffic_log.json"
+        
+        # Blue-Green specific directories
+        self.blue_dir = self.models_dir / "blue"
+        self.green_dir = self.models_dir / "green"
+        self.blue_dir.mkdir(parents=True, exist_ok=True)
+        self.green_dir.mkdir(parents=True, exist_ok=True)
+    
+    def setup_deployment_config(self):
+        """Setup deployment configuration"""
+        self.deployment_config = {
+            'traffic_stages': [
+                {'percentage': 10, 'duration_minutes': 15, 'success_threshold': 0.95},
+                {'percentage': 25, 'duration_minutes': 15, 'success_threshold': 0.95},
+                {'percentage': 50, 'duration_minutes': 30, 'success_threshold': 0.95},
+                {'percentage': 75, 'duration_minutes': 30, 'success_threshold': 0.95},
+                {'percentage': 100, 'duration_minutes': 0, 'success_threshold': 0.95}
+            ],
+            'health_checks': {
+                'response_time_threshold': 5.0,  # seconds
+                'error_rate_threshold': 0.05,    # 5%
+                'confidence_threshold': 0.6,     # minimum confidence
+                'check_interval': 30,            # seconds
+                'failure_threshold': 3           # consecutive failures
+            },
+            'rollback_conditions': {
+                'error_rate_spike': 0.15,        # 15% error rate
+                'response_time_spike': 10.0,     # 10 seconds
+                'confidence_drop': 0.4,          # below 40% confidence
+                'health_check_failures': 5       # consecutive failures
+            },
+            'deployment_timeouts': {
+                'stage_timeout_minutes': 60,
+                'total_deployment_hours': 6,
+                'rollback_timeout_minutes': 15
+            }
+        }
+    
+    def create_model_version(self, model_path: str, vectorizer_path: str, 
+                           metadata: Dict) -> str:
+        """Create a new model version"""
+        try:
+            version_id = f"v{datetime.now().strftime('%Y%m%d_%H%M%S')}"
+            
+            # Create version directory
+            version_dir = self.models_dir / version_id
+            version_dir.mkdir(parents=True, exist_ok=True)
+            
+            # Copy model files to version directory
+            import shutil
+            model_dest = version_dir / "model.pkl"
+            vectorizer_dest = version_dir / "vectorizer.pkl"
+            metadata_dest = version_dir / "metadata.json"
+            
+            shutil.copy2(model_path, model_dest)
+            shutil.copy2(vectorizer_path, vectorizer_dest)
+            
+            # Save metadata
+            version_metadata = {
+                **metadata,
+                'version_id': version_id,
+                'created_at': datetime.now().isoformat(),
+                'model_path': str(model_dest),
+                'vectorizer_path': str(vectorizer_dest),
+                'status': 'created'
+            }
+            
+            with open(metadata_dest, 'w') as f:
+                json.dump(version_metadata, f, indent=2)
+            
+            # Create ModelVersion object
+            model_version = ModelVersion(
+                version_id=version_id,
+                model_path=str(model_dest),
+                vectorizer_path=str(vectorizer_dest),
+                metadata_path=str(metadata_dest),
+                created_at=version_metadata['created_at'],
+                status='created',
+                performance_metrics=metadata.get('performance_metrics', {}),
+                deployment_config={}
+            )
+            
+            # Log version creation
+            self.log_deployment_event("version_created", f"Created model version {version_id}", {
+                'version_id': version_id,
+                'performance_metrics': model_version.performance_metrics
+            })
+            
+            logger.info(f"Created model version: {version_id}")
+            return version_id
+            
+        except Exception as e:
+            logger.error(f"Failed to create model version: {e}")
+            raise e
+    
+    def prepare_deployment(self, target_version_id: str, 
+                         deployment_strategy: str = "blue_green") -> str:
+        """Prepare a new deployment"""
+        try:
+            # Generate deployment ID
+            deployment_id = f"deploy_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
+            
+            # Get current active version
+            current_version = self.get_active_version()
+            
+            # Validate target version exists
+            if not self.version_exists(target_version_id):
+                raise ValueError(f"Target version {target_version_id} does not exist")
+            
+            # Create deployment plan
+            deployment_plan = DeploymentPlan(
+                deployment_id=deployment_id,
+                source_version=current_version['version_id'] if current_version else None,
+                target_version=target_version_id,
+                strategy=deployment_strategy,
+                traffic_stages=self.deployment_config['traffic_stages'].copy(),
+                health_checks=self.deployment_config['health_checks'].copy(),
+                rollback_conditions=self.deployment_config['rollback_conditions'].copy(),
+                created_at=datetime.now().isoformat(),
+                status=DeploymentStatus.PREPARING.value
+            )
+            
+            # Stage the new version
+            self.stage_version(target_version_id, deployment_plan)
+            
+            # Update deployment state
+            self.current_deployment = deployment_plan
+            self.save_deployment_state()
+            
+            self.log_deployment_event("deployment_prepared", f"Prepared deployment {deployment_id}", {
+                'deployment_plan': asdict(deployment_plan)
+            })
+            
+            logger.info(f"Prepared deployment: {deployment_id}")
+            return deployment_id
+            
+        except Exception as e:
+            logger.error(f"Failed to prepare deployment: {e}")
+            raise e
+    
+    def stage_version(self, version_id: str, deployment_plan: DeploymentPlan):
+        """Stage a model version for deployment"""
+        try:
+            # Determine staging environment (blue or green)
+            staging_env = self.determine_staging_environment()
+            
+            # Copy version to staging directory
+            version_dir = self.models_dir / version_id
+            staging_dir = self.blue_dir if staging_env == "blue" else self.green_dir
+            
+            # Clear staging directory
+            import shutil
+            if staging_dir.exists():
+                shutil.rmtree(staging_dir)
+            staging_dir.mkdir(parents=True, exist_ok=True)
+            
+            # Copy model files
+            for file_name in ["model.pkl", "vectorizer.pkl", "metadata.json"]:
+                source_file = version_dir / file_name
+                if source_file.exists():
+                    shutil.copy2(source_file, staging_dir / file_name)
+            
+            # Update staging version
+            self.staging_version = {
+                'version_id': version_id,
+                'environment': staging_env,
+                'staged_at': datetime.now().isoformat(),
+                'status': 'staged'
+            }
+            
+            # Update deployment status
+            deployment_plan.status = DeploymentStatus.STAGING.value
+            
+            logger.info(f"Staged version {version_id} in {staging_env} environment")
+            
+        except Exception as e:
+            logger.error(f"Failed to stage version: {e}")
+            raise e
+    
+    def start_deployment(self, deployment_id: str) -> bool:
+        """Start the deployment process"""
+        try:
+            if not self.current_deployment or self.current_deployment.deployment_id != deployment_id:
+                raise ValueError(f"Deployment {deployment_id} not found or not current")
+            
+            # Update status
+            self.current_deployment.status = DeploymentStatus.DEPLOYING.value
+            
+            # Start monitoring
+            self.start_deployment_monitoring()
+            
+            # Begin traffic shifting
+            success = self.execute_traffic_stages()
+            
+            if success:
+                self.current_deployment.status = DeploymentStatus.COMPLETED.value
+                self.finalize_deployment()
+            else:
+                self.current_deployment.status = DeploymentStatus.FAILED.value
+                self.initiate_rollback("Deployment failed during traffic shifting")
+            
+            self.save_deployment_state()
+            
+            return success
+            
+        except Exception as e:
+            logger.error(f"Failed to start deployment: {e}")
+            self.initiate_rollback(f"Deployment error: {str(e)}")
+            return False
+    
+    def execute_traffic_stages(self) -> bool:
+        """Execute gradual traffic shifting"""
+        try:
+            stages = self.current_deployment.traffic_stages
+            
+            for i, stage in enumerate(stages):
+                logger.info(f"Starting stage {i+1}/{len(stages)}: {stage['percentage']}% traffic")
+                
+                # Update traffic split
+                self.update_traffic_split(stage['percentage'])
+                
+                # Wait for stage duration
+                if stage['duration_minutes'] > 0:
+                    stage_success = self.monitor_stage_health(
+                        stage['duration_minutes'], 
+                        stage['success_threshold']
+                    )
+                    
+                    if not stage_success:
+                        logger.error(f"Stage {i+1} failed health checks")
+                        return False
+                
+                self.log_deployment_event("stage_completed", f"Stage {i+1} completed", {
+                    'stage': stage,
+                    'traffic_split': self.traffic_split
+                })
+            
+            return True
+            
+        except Exception as e:
+            logger.error(f"Traffic stage execution failed: {e}")
+            return False
+    
+    def update_traffic_split(self, green_percentage: int):
+        """Update traffic routing split"""
+        self.traffic_split = {
+            "blue": 100 - green_percentage,
+            "green": green_percentage
+        }
+        
+        # Log traffic change
+        self.log_traffic_change(self.traffic_split)
+        
+        logger.info(f"Updated traffic split: Blue {self.traffic_split['blue']}%, Green {self.traffic_split['green']}%")
+    
+    def monitor_stage_health(self, duration_minutes: int, success_threshold: float) -> bool:
+        """Monitor health during a deployment stage"""
+        try:
+            start_time = datetime.now()
+            end_time = start_time + timedelta(minutes=duration_minutes)
+            check_interval = self.deployment_config['health_checks']['check_interval']
+            
+            consecutive_failures = 0
+            max_failures = self.deployment_config['health_checks']['failure_threshold']
+            
+            while datetime.now() < end_time:
+                # Perform health check
+                health_result = self.perform_health_check()
+                
+                if health_result['healthy']:
+                    consecutive_failures = 0
+                else:
+                    consecutive_failures += 1
+                    logger.warning(f"Health check failed: {health_result['issues']}")
+                    
+                    if consecutive_failures >= max_failures:
+                        logger.error(f"Too many consecutive failures: {consecutive_failures}")
+                        return False
+                
+                # Check for immediate rollback conditions
+                if self.should_trigger_immediate_rollback(health_result):
+                    logger.error("Immediate rollback conditions met")
+                    return False
+                
+                time.sleep(check_interval)
+            
+            return True
+            
+        except Exception as e:
+            logger.error(f"Stage health monitoring failed: {e}")
+            return False
+    
+    def perform_health_check(self) -> Dict[str, Any]:
+        """Perform comprehensive health check"""
+        try:
+            health_result = {
+                'healthy': True,
+                'issues': [],
+                'metrics': {},
+                'timestamp': datetime.now().isoformat()
+            }
+            
+            # Check response times
+            avg_response_time = self.get_average_response_time()
+            threshold = self.deployment_config['health_checks']['response_time_threshold']
+            
+            health_result['metrics']['response_time'] = avg_response_time
+            
+            if avg_response_time > threshold:
+                health_result['healthy'] = False
+                health_result['issues'].append(f"High response time: {avg_response_time:.2f}s")
+            
+            # Check error rates
+            error_rate = self.get_current_error_rate()
+            error_threshold = self.deployment_config['health_checks']['error_rate_threshold']
+            
+            health_result['metrics']['error_rate'] = error_rate
+            
+            if error_rate > error_threshold:
+                health_result['healthy'] = False
+                health_result['issues'].append(f"High error rate: {error_rate:.2%}")
+            
+            # Check prediction confidence
+            avg_confidence = self.get_average_confidence()
+            confidence_threshold = self.deployment_config['health_checks']['confidence_threshold']
+            
+            health_result['metrics']['confidence'] = avg_confidence
+            
+            if avg_confidence < confidence_threshold:
+                health_result['healthy'] = False
+                health_result['issues'].append(f"Low confidence: {avg_confidence:.2f}")
+            
+            return health_result
+            
+        except Exception as e:
+            logger.error(f"Health check failed: {e}")
+            return {
+                'healthy': False,
+                'issues': [f"Health check error: {str(e)}"],
+                'metrics': {},
+                'timestamp': datetime.now().isoformat()
+            }
+    
+    def should_trigger_immediate_rollback(self, health_result: Dict) -> bool:
+        """Check if immediate rollback should be triggered"""
+        rollback_conditions = self.deployment_config['rollback_conditions']
+        metrics = health_result['metrics']
+        
+        # Check error rate spike
+        if metrics.get('error_rate', 0) > rollback_conditions['error_rate_spike']:
+            return True
+        
+        # Check response time spike
+        if metrics.get('response_time', 0) > rollback_conditions['response_time_spike']:
+            return True
+        
+        # Check confidence drop
+        if metrics.get('confidence', 1) < rollback_conditions['confidence_drop']:
+            return True
+        
+        return False
+    
+    def initiate_rollback(self, reason: str) -> bool:
+        """Initiate deployment rollback"""
+        try:
+            logger.warning(f"Initiating rollback: {reason}")
+            
+            if self.current_deployment:
+                self.current_deployment.status = DeploymentStatus.ROLLING_BACK.value
+            
+            # Immediately route all traffic to blue (current production)
+            self.update_traffic_split(0)  # 0% to green, 100% to blue
+            
+            # Clear staging environment
+            self.clear_staging_environment()
+            
+            # Update deployment state
+            if self.current_deployment:
+                self.current_deployment.status = DeploymentStatus.FAILED.value
+            
+            self.save_deployment_state()
+            
+            self.log_deployment_event("rollback_initiated", f"Rollback initiated: {reason}", {
+                'reason': reason,
+                'traffic_split': self.traffic_split
+            })
+            
+            logger.info("Rollback completed successfully")
+            return True
+            
+        except Exception as e:
+            logger.error(f"Rollback failed: {e}")
+            return False
+    
+    def finalize_deployment(self):
+        """Finalize successful deployment"""
+        try:
+            if not self.staging_version:
+                raise ValueError("No staging version to finalize")
+            
+            # Move staging to active
+            staging_env = self.staging_version['environment']
+            
+            # Update active version
+            self.active_version = {
+                **self.staging_version,
+                'activated_at': datetime.now().isoformat(),
+                'status': 'active'
+            }
+            
+            # Clear staging
+            self.staging_version = None
+            
+            # Update traffic split to 100% green if that's where new version is
+            if staging_env == "green":
+                self.update_traffic_split(100)
+            else:
+                self.update_traffic_split(0)
+            
+            # Archive old version if exists
+            self.archive_old_version()
+            
+            self.log_deployment_event("deployment_finalized", "Deployment successfully finalized", {
+                'active_version': self.active_version,
+                'traffic_split': self.traffic_split
+            })
+            
+            logger.info("Deployment finalized successfully")
+            
+        except Exception as e:
+            logger.error(f"Failed to finalize deployment: {e}")
+            raise e
+    
+    def get_active_version(self) -> Optional[Dict]:
+        """Get currently active model version"""
+        return self.active_version
+    
+    def get_staging_version(self) -> Optional[Dict]:
+        """Get currently staged model version"""
+        return self.staging_version
+    
+    def get_traffic_split(self) -> Dict[str, int]:
+        """Get current traffic split configuration"""
+        return self.traffic_split.copy()
+    
+    def determine_staging_environment(self) -> str:
+        """Determine which environment to use for staging"""
+        if not self.active_version:
+            return "blue"  # Default to blue if no active version
+        
+        current_env = self.active_version.get('environment', 'blue')
+        return "green" if current_env == "blue" else "blue"
+    
+    def version_exists(self, version_id: str) -> bool:
+        """Check if a version exists"""
+        version_dir = self.models_dir / version_id
+        return version_dir.exists()
+    
+    def clear_staging_environment(self):
+        """Clear the staging environment"""
+        if self.staging_version:
+            staging_env = self.staging_version['environment']
+            staging_dir = self.blue_dir if staging_env == "blue" else self.green_dir
+            
+            import shutil
+            if staging_dir.exists():
+                shutil.rmtree(staging_dir)
+                staging_dir.mkdir(parents=True, exist_ok=True)
+            
+            self.staging_version = None
+    
+    def archive_old_version(self):
+        """Archive the previously active version"""
+        # Implementation for archiving old versions
+        # This could move old versions to an archive directory
+        pass
+    
+    def start_deployment_monitoring(self):
+        """Start background deployment monitoring"""
+        if not self.monitoring_active:
+            self.monitoring_active = True
+            self.monitor_thread = threading.Thread(target=self.deployment_monitoring_loop, daemon=True)
+            self.monitor_thread.start()
+    
+    def stop_deployment_monitoring(self):
+        """Stop deployment monitoring"""
+        self.monitoring_active = False
+        if self.monitor_thread:
+            self.monitor_thread.join(timeout=10)
+    
+    def deployment_monitoring_loop(self):
+        """Background monitoring loop for deployments"""
+        while self.monitoring_active:
+            try:
+                if (self.current_deployment and 
+                    self.current_deployment.status == DeploymentStatus.DEPLOYING.value):
+                    
+                    # Perform periodic health checks
+                    health_result = self.perform_health_check()
+                    
+                    if self.should_trigger_immediate_rollback(health_result):
+                        self.initiate_rollback("Automated rollback due to health check failures")
+                        break
+                
+                time.sleep(30)  # Check every 30 seconds
+                
+            except Exception as e:
+                logger.error(f"Deployment monitoring error: {e}")
+                time.sleep(60)
+    
+    def get_average_response_time(self) -> float:
+        """Get average response time from recent requests"""
+        # This would integrate with your monitoring system
+        # For now, return a simulated value
+        return np.random.normal(2.0, 0.5)
+    
+    def get_current_error_rate(self) -> float:
+        """Get current error rate"""
+        # This would integrate with your monitoring system
+        # For now, return a simulated value
+        return np.random.beta(1, 20)  # Typically low error rate
+    
+    def get_average_confidence(self) -> float:
+        """Get average prediction confidence"""
+        # This would integrate with your monitoring system
+        # For now, return a simulated value
+        return np.random.beta(8, 2)  # Typically high confidence
+    
+    def log_deployment_event(self, event: str, message: str, details: Dict = None):
+        """Log deployment events"""
+        try:
+            log_entry = {
+                'timestamp': datetime.now().isoformat(),
+                'event': event,
+                'message': message,
+                'details': details or {}
+            }
+            
+            # Load existing logs
+            logs = []
+            if self.deployment_log_path.exists():
+                try:
+                    with open(self.deployment_log_path, 'r') as f:
+                        logs = json.load(f)
+                except:
+                    logs = []
+            
+            logs.append(log_entry)
+            
+            # Keep only last 1000 entries
+            if len(logs) > 1000:
+                logs = logs[-1000:]
+            
+            # Save logs
+            with open(self.deployment_log_path, 'w') as f:
+                json.dump(logs, f, indent=2)
+            
+        except Exception as e:
+            logger.error(f"Failed to log deployment event: {e}")
+    
+    def log_traffic_change(self, traffic_split: Dict[str, int]):
+        """Log traffic routing changes"""
+        try:
+            traffic_entry = {
+                'timestamp': datetime.now().isoformat(),
+                'traffic_split': traffic_split,
+                'deployment_id': self.current_deployment.deployment_id if self.current_deployment else None
+            }
+            
+            # Load existing traffic logs
+            logs = []
+            if self.traffic_log_path.exists():
+                try:
+                    with open(self.traffic_log_path, 'r') as f:
+                        logs = json.load(f)
+                except:
+                    logs = []
+            
+            logs.append(traffic_entry)
+            
+            # Keep only last 500 entries
+            if len(logs) > 500:
+                logs = logs[-500:]
+            
+            # Save logs
+            with open(self.traffic_log_path, 'w') as f:
+                json.dump(logs, f, indent=2)
+            
+        except Exception as e:
+            logger.error(f"Failed to log traffic change: {e}")
+    
+    def save_deployment_state(self):
+        """Save current deployment state"""
+        try:
+            state = {
+                'current_deployment': asdict(self.current_deployment) if self.current_deployment else None,
+                'active_version': self.active_version,
+                'staging_version': self.staging_version,
+                'traffic_split': self.traffic_split,
+                'last_updated': datetime.now().isoformat()
+            }
+            
+            with open(self.deployment_state_path, 'w') as f:
+                json.dump(state, f, indent=2)
+            
+        except Exception as e:
+            logger.error(f"Failed to save deployment state: {e}")
+    
+    def load_deployment_state(self):
+        """Load existing deployment state"""
+        try:
+            if self.deployment_state_path.exists():
+                with open(self.deployment_state_path, 'r') as f:
+                    state = json.load(f)
+                
+                # Restore state
+                if state.get('current_deployment'):
+                    self.current_deployment = DeploymentPlan(**state['current_deployment'])
+                
+                self.active_version = state.get('active_version')
+                self.staging_version = state.get('staging_version')
+                self.traffic_split = state.get('traffic_split', {"blue": 100, "green": 0})
+                
+                logger.info("Loaded deployment state from file")
+            
+        except Exception as e:
+            logger.warning(f"Failed to load deployment state: {e}")
+    
+    def get_deployment_status(self) -> Dict[str, Any]:
+        """Get comprehensive deployment status"""
+        try:
+            return {
+                'timestamp': datetime.now().isoformat(),
+                'current_deployment': asdict(self.current_deployment) if self.current_deployment else None,
+                'active_version': self.active_version,
+                'staging_version': self.staging_version,
+                'traffic_split': self.traffic_split,
+                'monitoring_active': self.monitoring_active,
+                'available_versions': self.list_available_versions(),
+                'recent_deployments': self.get_recent_deployments(limit=5)
+            }
+        except Exception as e:
+            logger.error(f"Failed to get deployment status: {e}")
+            return {'error': str(e)}
+    
+    def list_available_versions(self) -> List[str]:
+        """List all available model versions"""
+        try:
+            versions = []
+            if self.models_dir.exists():
+                for item in self.models_dir.iterdir():
+                    if item.is_dir() and item.name.startswith('v'):
+                        versions.append(item.name)
+            return sorted(versions, reverse=True)
+        except Exception as e:
+            logger.error(f"Failed to list versions: {e}")
+            return []
+    
+    def get_recent_deployments(self, limit: int = 10) -> List[Dict]:
+        """Get recent deployment history"""
+        try:
+            if not self.deployment_log_path.exists():
+                return []
+            
+            with open(self.deployment_log_path, 'r') as f:
+                logs = json.load(f)
+            
+            # Filter deployment events
+            deployment_events = [
+                log for log in logs 
+                if log.get('event') in ['deployment_prepared', 'deployment_finalized', 'rollback_initiated']
+            ]
+            
+            return deployment_events[-limit:]
+            
+        except Exception as e:
+            logger.error(f"Failed to get recent deployments: {e}")
+            return []