Spaces:
Sleeping
Sleeping
| """ | |
| Deployment orchestration engine. | |
| Orchestrates end-to-end configuration deployment: | |
| - Config generation from templates | |
| - Pre-deployment validation | |
| - Device connectivity and deployment | |
| - Post-deployment verification | |
| - Automatic rollback on failure | |
| - Parallel deployment via Ray integration | |
| """ | |
| import logging | |
| from typing import Dict, List, Optional, Any, Tuple | |
| from dataclasses import dataclass, field | |
| from enum import Enum | |
| from datetime import datetime | |
| import time | |
| from agent.device_driver import ( | |
| DeviceDriver, DeviceCredentials, DeviceType, | |
| ConfigDeploymentResult, ConnectionStatus | |
| ) | |
| from agent.config_templates import ConfigTemplateEngine | |
| logger = logging.getLogger(__name__) | |
| class DeploymentStatus(Enum): | |
| """Deployment status""" | |
| PENDING = "pending" | |
| VALIDATING = "validating" | |
| DEPLOYING = "deploying" | |
| VERIFYING = "verifying" | |
| SUCCESS = "success" | |
| FAILED = "failed" | |
| ROLLED_BACK = "rolled_back" | |
| class DeploymentTask: | |
| """Single device deployment task""" | |
| device_id: str | |
| device_type: DeviceType | |
| hostname: str | |
| username: str | |
| password: str | |
| config: str | |
| dry_run: bool = False | |
| pre_checks: List[str] = field(default_factory=list) | |
| post_checks: List[str] = field(default_factory=list) | |
| class DeploymentResult: | |
| """Result from deployment orchestration""" | |
| device_id: str | |
| status: DeploymentStatus | |
| config_deployed: Optional[str] = None | |
| config_before: Optional[str] = None | |
| config_after: Optional[str] = None | |
| pre_check_results: Dict[str, bool] = field(default_factory=dict) | |
| post_check_results: Dict[str, bool] = field(default_factory=dict) | |
| deployment_output: Optional[str] = None | |
| error: Optional[str] = None | |
| duration_seconds: float = 0.0 | |
| rolled_back: bool = False | |
| timestamp: datetime = field(default_factory=datetime.now) | |
| class DeploymentEngine: | |
| """ | |
| Orchestrates configuration deployment to network devices. | |
| Features: | |
| - Multi-vendor device support (Cisco, Arista, Juniper) | |
| - Pre/post deployment validation | |
| - Automatic rollback on failure | |
| - Parallel deployment via Ray | |
| - Dry-run mode for testing | |
| """ | |
| def __init__(self, use_napalm: bool = True, use_ray: bool = False): | |
| """ | |
| Initialize deployment engine. | |
| Args: | |
| use_napalm: Prefer NAPALM over Netmiko | |
| use_ray: Use Ray for parallel deployment | |
| """ | |
| self.driver = DeviceDriver(use_napalm=use_napalm) | |
| self.template_engine = ConfigTemplateEngine() | |
| self.use_ray = use_ray | |
| if use_ray: | |
| try: | |
| from agent.ray_executor import RayExecutor | |
| self.ray_executor = RayExecutor() | |
| logger.info("DeploymentEngine using Ray for parallel execution") | |
| except ImportError: | |
| logger.warning("Ray not available - falling back to serial execution") | |
| self.use_ray = False | |
| self.ray_executor = None | |
| else: | |
| self.ray_executor = None | |
| self.deployment_history: List[DeploymentResult] = [] | |
| def deploy_single_device(self, task: DeploymentTask) -> DeploymentResult: | |
| """ | |
| Deploy configuration to a single device. | |
| Args: | |
| task: Deployment task specification | |
| Returns: | |
| DeploymentResult with outcome | |
| """ | |
| logger.info(f"Starting deployment to {task.device_id} (dry_run={task.dry_run})") | |
| start_time = time.time() | |
| result = DeploymentResult( | |
| device_id=task.device_id, | |
| status=DeploymentStatus.PENDING | |
| ) | |
| try: | |
| # 1. Connect to device | |
| logger.info(f"Connecting to {task.device_id}...") | |
| credentials = DeviceCredentials( | |
| hostname=task.hostname, | |
| username=task.username, | |
| password=task.password, | |
| device_type=task.device_type | |
| ) | |
| conn = self.driver.connect(credentials) | |
| if conn.status != ConnectionStatus.CONNECTED: | |
| raise Exception(f"Failed to connect: {conn.last_error}") | |
| # 2. Pre-deployment checks | |
| result.status = DeploymentStatus.VALIDATING | |
| logger.info(f"Running {len(task.pre_checks)} pre-deployment checks...") | |
| for check in task.pre_checks: | |
| check_result = self._run_validation_check(task.device_id, check) | |
| result.pre_check_results[check] = check_result | |
| if not check_result: | |
| raise Exception(f"Pre-deployment check failed: {check}") | |
| # 3. Get current config (for rollback) | |
| result.config_before = self.driver.get_config(task.device_id, "running") | |
| # 4. Deploy configuration | |
| result.status = DeploymentStatus.DEPLOYING | |
| logger.info(f"Deploying configuration to {task.device_id}...") | |
| deploy_result = self.driver.deploy_config( | |
| device_id=task.device_id, | |
| config=task.config, | |
| dry_run=task.dry_run, | |
| replace=False # Merge by default | |
| ) | |
| if not deploy_result.success: | |
| raise Exception(f"Deployment failed: {deploy_result.error}") | |
| result.config_deployed = task.config | |
| result.config_after = deploy_result.config_after | |
| result.deployment_output = deploy_result.output | |
| # 5. Post-deployment verification | |
| if not task.dry_run: | |
| result.status = DeploymentStatus.VERIFYING | |
| logger.info(f"Running {len(task.post_checks)} post-deployment checks...") | |
| # Give device time to apply config | |
| time.sleep(2) | |
| for check in task.post_checks: | |
| check_result = self._run_validation_check(task.device_id, check) | |
| result.post_check_results[check] = check_result | |
| if not check_result: | |
| # Post-check failed - rollback! | |
| logger.error(f"Post-deployment check failed: {check}") | |
| logger.warning(f"Initiating rollback on {task.device_id}") | |
| rollback_result = self.driver.rollback_config( | |
| device_id=task.device_id, | |
| config=result.config_before | |
| ) | |
| result.rolled_back = rollback_result.success | |
| result.status = DeploymentStatus.ROLLED_BACK if result.rolled_back else DeploymentStatus.FAILED | |
| raise Exception(f"Post-deployment check failed: {check} (rollback {'succeeded' if result.rolled_back else 'FAILED'})") | |
| # Success! | |
| result.status = DeploymentStatus.SUCCESS | |
| result.duration_seconds = time.time() - start_time | |
| logger.info(f"✓ Deployment to {task.device_id} completed successfully in {result.duration_seconds:.1f}s") | |
| except Exception as e: | |
| result.status = DeploymentStatus.FAILED | |
| result.error = str(e) | |
| result.duration_seconds = time.time() - start_time | |
| logger.error(f"✗ Deployment to {task.device_id} failed: {e}") | |
| finally: | |
| # Disconnect | |
| self.driver.disconnect(task.device_id) | |
| # Store in history | |
| self.deployment_history.append(result) | |
| return result | |
| def deploy_multiple_devices(self, tasks: List[DeploymentTask], | |
| parallel: bool = False, | |
| max_failures: Optional[int] = None) -> List[DeploymentResult]: | |
| """ | |
| Deploy configuration to multiple devices. | |
| Args: | |
| tasks: List of deployment tasks | |
| parallel: If True, use parallel execution (Ray if available) | |
| max_failures: Stop if this many devices fail (None = deploy all) | |
| Returns: | |
| List of DeploymentResults | |
| """ | |
| logger.info(f"Deploying to {len(tasks)} devices (parallel={parallel})") | |
| if parallel and self.use_ray and self.ray_executor: | |
| return self._deploy_parallel_ray(tasks, max_failures) | |
| elif parallel: | |
| logger.warning("Parallel mode requested but Ray not available - using serial") | |
| # Serial deployment | |
| results = [] | |
| failures = 0 | |
| for task in tasks: | |
| result = self.deploy_single_device(task) | |
| results.append(result) | |
| if result.status == DeploymentStatus.FAILED: | |
| failures += 1 | |
| if max_failures and failures >= max_failures: | |
| logger.error(f"Max failures ({max_failures}) reached - stopping deployment") | |
| # Mark remaining as pending | |
| remaining = len(tasks) - len(results) | |
| logger.warning(f"Skipping {remaining} remaining devices") | |
| break | |
| return results | |
| def _deploy_parallel_ray(self, tasks: List[DeploymentTask], | |
| max_failures: Optional[int]) -> List[DeploymentResult]: | |
| """Deploy using Ray parallel execution""" | |
| logger.info(f"Using Ray to deploy to {len(tasks)} devices in parallel") | |
| # This would integrate with Ray executor | |
| # For now, fall back to serial | |
| logger.warning("Ray parallel deployment not yet implemented - using serial") | |
| return self.deploy_multiple_devices(tasks, parallel=False, max_failures=max_failures) | |
| def _run_validation_check(self, device_id: str, check: str) -> bool: | |
| """ | |
| Run validation check on device. | |
| Args: | |
| device_id: Device identifier | |
| check: Check command or test | |
| Returns: | |
| True if check passes | |
| """ | |
| try: | |
| # Parse check type | |
| if check.startswith("ping:"): | |
| # Ping test | |
| target = check.split(":", 1)[1] | |
| result = self.driver.send_command(device_id, f"ping {target}") | |
| return result.success and "success" in result.output.lower() | |
| elif check.startswith("interface:"): | |
| # Interface status check | |
| interface = check.split(":", 1)[1] | |
| result = self.driver.send_command(device_id, f"show interface {interface}") | |
| return result.success and "up" in result.output.lower() | |
| elif check.startswith("command:"): | |
| # Generic command check (success = command runs without error) | |
| command = check.split(":", 1)[1] | |
| result = self.driver.send_command(device_id, command) | |
| return result.success | |
| else: | |
| # Default: run as show command | |
| result = self.driver.send_command(device_id, check) | |
| return result.success | |
| except Exception as e: | |
| logger.error(f"Validation check '{check}' failed on {device_id}: {e}") | |
| return False | |
| def generate_and_deploy(self, device: Any, network_context: Dict[str, Any], | |
| credentials: Dict[str, str], | |
| dry_run: bool = False, | |
| pre_checks: Optional[List[str]] = None, | |
| post_checks: Optional[List[str]] = None) -> DeploymentResult: | |
| """ | |
| Generate configuration from template and deploy to device. | |
| Args: | |
| device: Device object from NetworkModel | |
| network_context: Network-level context (VLANs, routing, etc.) | |
| credentials: Device credentials (username, password) | |
| dry_run: If True, generate and validate but don't deploy | |
| pre_checks: Optional pre-deployment validation checks | |
| post_checks: Optional post-deployment validation checks | |
| Returns: | |
| DeploymentResult | |
| """ | |
| device_name = device.get('name') if isinstance(device, dict) else device.name | |
| logger.info(f"Generating and deploying config for {device_name}") | |
| # 1. Generate configuration from template | |
| config = self.template_engine.generate_device_config(device, network_context) | |
| # 2. Map device vendor to driver type (handle both dict and object) | |
| vendor = device.get('vendor') if isinstance(device, dict) else device.vendor | |
| model = device.get('model', '') if isinstance(device, dict) else getattr(device, 'model', '') | |
| device_type = self._map_vendor_to_device_type(vendor, model) | |
| # 3. Create deployment task | |
| device_name = device.get('name') if isinstance(device, dict) else device.name | |
| mgmt_ip = device.get('mgmt_ip') if isinstance(device, dict) else device.mgmt_ip | |
| task = DeploymentTask( | |
| device_id=device_name, | |
| device_type=device_type, | |
| hostname=mgmt_ip, | |
| username=credentials.get('username', 'admin'), | |
| password=credentials.get('password', 'admin'), | |
| config=config, | |
| dry_run=dry_run, | |
| pre_checks=pre_checks or [], | |
| post_checks=post_checks or [] | |
| ) | |
| # 4. Deploy | |
| return self.deploy_single_device(task) | |
| def _map_vendor_to_device_type(self, vendor: str, model: str) -> DeviceType: | |
| """Map vendor/model to DeviceType""" | |
| vendor_lower = vendor.lower() | |
| model_lower = model.lower() | |
| if 'cisco' in vendor_lower: | |
| if 'nexus' in model_lower or 'nxos' in model_lower: | |
| return DeviceType.CISCO_NXOS | |
| elif 'xe' in model_lower or 'asr' in model_lower or 'isr' in model_lower: | |
| return DeviceType.CISCO_XE | |
| else: | |
| return DeviceType.CISCO_IOS | |
| elif 'arista' in vendor_lower: | |
| return DeviceType.ARISTA_EOS | |
| elif 'juniper' in vendor_lower: | |
| return DeviceType.JUNIPER_JUNOS | |
| else: | |
| logger.warning(f"Unknown vendor {vendor}, defaulting to Cisco IOS") | |
| return DeviceType.CISCO_IOS | |
| def get_deployment_summary(self) -> Dict[str, Any]: | |
| """Get summary of deployment history""" | |
| total = len(self.deployment_history) | |
| if total == 0: | |
| return { | |
| 'total_deployments': 0, | |
| 'success_count': 0, | |
| 'failed_count': 0, | |
| 'rolled_back_count': 0, | |
| 'success_rate': 0.0 | |
| } | |
| success = sum(1 for r in self.deployment_history if r.status == DeploymentStatus.SUCCESS) | |
| failed = sum(1 for r in self.deployment_history if r.status == DeploymentStatus.FAILED) | |
| rolled_back = sum(1 for r in self.deployment_history if r.rolled_back) | |
| return { | |
| 'total_deployments': total, | |
| 'success_count': success, | |
| 'failed_count': failed, | |
| 'rolled_back_count': rolled_back, | |
| 'success_rate': (success / total) * 100 if total > 0 else 0.0, | |
| 'avg_duration': sum(r.duration_seconds for r in self.deployment_history) / total, | |
| 'latest_deployments': [ | |
| { | |
| 'device_id': r.device_id, | |
| 'status': r.status.value, | |
| 'timestamp': r.timestamp.isoformat(), | |
| 'duration': r.duration_seconds | |
| } | |
| for r in self.deployment_history[-10:] # Last 10 | |
| ] | |
| } | |
| def verify_deployment(self, device_id: str, expected_config_snippet: Optional[str] = None) -> Dict[str, Any]: | |
| """ | |
| Verify deployment was successful. | |
| Args: | |
| device_id: Device identifier | |
| expected_config_snippet: Optional config snippet to verify is present | |
| Returns: | |
| Verification results | |
| """ | |
| results = { | |
| 'device_id': device_id, | |
| 'connectivity': False, | |
| 'config_retrieved': False, | |
| 'snippet_found': None, | |
| 'facts': None, | |
| 'errors': [] | |
| } | |
| try: | |
| # Check if we have a connection | |
| if device_id not in self.driver.connections: | |
| results['errors'].append("Device not connected") | |
| return results | |
| # Test connectivity | |
| results['connectivity'] = self.driver.verify_connectivity(device_id) | |
| if not results['connectivity']: | |
| results['errors'].append("Device not responsive") | |
| return results | |
| # Get current config | |
| config = self.driver.get_config(device_id, "running") | |
| results['config_retrieved'] = config is not None | |
| if not results['config_retrieved']: | |
| results['errors'].append("Could not retrieve configuration") | |
| return results | |
| # Check for expected snippet | |
| if expected_config_snippet: | |
| results['snippet_found'] = expected_config_snippet in config | |
| if not results['snippet_found']: | |
| results['errors'].append(f"Expected config snippet not found: {expected_config_snippet[:50]}...") | |
| # Get device facts | |
| facts = self.driver.get_device_facts(device_id) | |
| results['facts'] = facts | |
| except Exception as e: | |
| results['errors'].append(str(e)) | |
| logger.error(f"Verification failed for {device_id}: {e}") | |
| return results | |
| def cleanup(self): | |
| """Disconnect all devices and cleanup resources""" | |
| self.driver.disconnect_all() | |
| if self.ray_executor: | |
| self.ray_executor.shutdown() | |
| logger.info("DeploymentEngine cleanup complete") | |