""" Infrastructure Management Controller Provides API endpoints for automated infrastructure management, including service deployment, configuration, and monitoring setup. """ from fastapi import APIRouter, HTTPException, Depends from typing import List, Dict, Any, Optional, Union from pydantic import BaseModel, field_validator from src.controllers.base import BaseController from src.clients.portainer_client import get_portainer_client from src.clients.npm_client import get_npm_client from src.logging_config import get_logger from src import service_groups from src.auth.oidc import get_admin_user, get_forward_auth_admin logger = get_logger(__name__) # Response models class ServiceInfo(BaseModel): """Information about a deployed service""" name: str stack_id: Optional[int] status: Union[str, int] endpoint_id: Optional[int] ports: List[int] = [] domains: List[str] = [] running: bool = False # True if at least one container is running containers_running: int = 0 # Number of running containers containers_total: int = 0 # Total number of containers @field_validator('status', mode='before') @classmethod def convert_status(cls, v): """Convert status to string representation""" if isinstance(v, int): # Portainer status: 1=active, 2=inactive return "active" if v == 1 else "inactive" return v class PortInfo(BaseModel): """Information about an allocated port""" port: int service: str container_name: Optional[str] = None protocol: str = "tcp" internal_hostname: Optional[str] = None internal_ip: Optional[str] = None external_domains: List[str] = [] host_port: Optional[int] = None # Port exposed on host, if different from internal description: str = "" class DomainInfo(BaseModel): """Information about a configured domain""" domain: str service: str proxy_host_id: Optional[int] ssl_enabled: bool = False certificate_id: Optional[int] class InfrastructureHealth(BaseModel): """Overall infrastructure health status""" portainer_connected: bool npm_connected: bool total_stacks: int total_proxy_hosts: int # Request models for write operations class DeployServiceRequest(BaseModel): """Request to deploy a new service""" name: str compose_content: str endpoint_id: int = 3 # Default to local endpoint class UpdateServiceRequest(BaseModel): """Request to update an existing service""" compose_content: str prune: bool = False pull_image: bool = True class CreateProxyRequest(BaseModel): """Request to create a new proxy host""" domain_names: List[str] forward_host: str forward_port: int forward_scheme: str = "http" ssl_enabled: bool = False request_ssl_certificate: bool = False block_exploits: bool = True websocket_upgrade: bool = True http2_support: bool = True class OperationResult(BaseModel): """Result of an infrastructure operation""" success: bool message: str details: Optional[Dict[str, Any]] = None class InfrastructureController(BaseController): """ Controller for infrastructure management operations Provides endpoints for: - Service discovery and listing - Port allocation management - Domain/proxy configuration - Automated service deployment """ def __init__(self): super().__init__(prefix="/infrastructure", tags=["Infrastructure"]) def create_router(self) -> APIRouter: """Create and configure the router""" router = APIRouter(prefix=self.prefix, tags=self.tags) @router.get( "/health", response_model=InfrastructureHealth, summary="Infrastructure health check" ) async def get_infrastructure_health(): """ Check health of all infrastructure services Returns status of Portainer, NPM, and summary statistics. """ portainer = get_portainer_client() npm = get_npm_client() portainer_healthy = await portainer.health_check() npm_healthy = await npm.health_check() total_stacks = 0 total_proxy_hosts = 0 if portainer_healthy: try: stacks = await portainer.get_stacks() total_stacks = len(stacks) except Exception as e: logger.error(f"Failed to get stacks count: {e}") if npm_healthy: try: proxy_hosts = await npm.get_proxy_hosts() total_proxy_hosts = len(proxy_hosts) except Exception as e: logger.error(f"Failed to get proxy hosts count: {e}") return InfrastructureHealth( portainer_connected=portainer_healthy, npm_connected=npm_healthy, total_stacks=total_stacks, total_proxy_hosts=total_proxy_hosts ) @router.get( "/services", response_model=List[ServiceInfo], summary="List all deployed services (Docker Compose stacks)", description="List all services deployed via Portainer stacks. Each 'service' represents a Docker Compose stack." ) async def list_services(): """ List all deployed Docker Compose stacks from Portainer Returns comprehensive stack/service information including: - Stack name (the service name) - Status (active/inactive) - Exposed ports - Configured domains (from NPM reverse proxy) - Running container count """ portainer = get_portainer_client() npm = get_npm_client() try: stacks = await portainer.get_stacks() proxy_hosts = await npm.get_proxy_hosts() # Build domain mapping (domain -> service name) domain_map = {} for proxy in proxy_hosts: for domain in proxy.get("domain_names", []): # Try to extract service name from forward_host forward_host = proxy.get("forward_host", "") domain_map[domain] = forward_host services = [] for stack in stacks: # Find domains for this stack stack_name = stack.get("Name", "") endpoint_id = stack.get("EndpointId") domains = [ domain for domain, host in domain_map.items() if stack_name in host or host in stack_name ] # Get container status for this stack containers_running = 0 containers_total = 0 try: all_containers = await portainer.get_containers(endpoint_id, all_containers=True) for container in all_containers: labels = container.get("Labels", {}) container_stack = labels.get("com.docker.compose.project", "") if container_stack.lower() == stack_name.lower(): containers_total += 1 if container.get("State", "") == "running": containers_running += 1 except Exception as e: logger.warning(f"Failed to get container status for {stack_name}: {e}") service_info = ServiceInfo( name=stack_name, stack_id=stack.get("Id"), status=stack.get("Status", "unknown"), endpoint_id=endpoint_id, ports=[], # TODO: Extract from stack file domains=domains, running=containers_running > 0, containers_running=containers_running, containers_total=containers_total ) services.append(service_info) return services except Exception as e: logger.error(f"Failed to list services: {e}") raise HTTPException(status_code=500, detail=str(e)) @router.get( "/services/{name}", response_model=ServiceInfo, summary="Get service details" ) async def get_service(name: str): """ Get detailed information about a specific service Args: name: Service/stack name Returns: Service details including status and configuration """ portainer = get_portainer_client() try: stacks = await portainer.get_stacks() # Find stack by name (case-insensitive) stack = next( (s for s in stacks if s.get("Name", "").lower() == name.lower()), None ) if not stack: raise HTTPException(status_code=404, detail=f"Service '{name}' not found") return ServiceInfo( name=stack.get("Name", ""), stack_id=stack.get("Id"), status=stack.get("Status", "unknown"), endpoint_id=stack.get("EndpointId"), ports=[], domains=[] ) except HTTPException: raise except Exception as e: logger.error(f"Failed to get service '{name}': {e}") raise HTTPException(status_code=500, detail=str(e)) @router.post( "/services/{service}/manage", response_model=Dict[str, Any], summary="Manage service lifecycle" ) async def manage_service(service: str, action: str, replicas: Optional[int] = None): """ Manage Docker Compose service/stack lifecycle. Args: service: Service/stack name action: One of: "start", "stop", "restart", "scale" replicas: Number of replicas (required for scale action) Returns: Success status and message """ from pydantic import BaseModel class ManageServiceRequest(BaseModel): action: str replicas: Optional[int] = None portainer = get_portainer_client() logger.info(f"Managing service '{service}': action={action}, replicas={replicas}") # Validate action valid_actions = ["start", "stop", "restart", "scale"] if action not in valid_actions: raise HTTPException( status_code=400, detail=f"Invalid action '{action}'. Must be one of: {', '.join(valid_actions)}" ) # Validate scale has replicas if action == "scale" and replicas is None: raise HTTPException( status_code=400, detail="'scale' action requires 'replicas' parameter" ) try: # Find the stack by name stacks = await portainer.get_stacks() stack = next( (s for s in stacks if s.get("Name", "").lower() == service.lower()), None ) if not stack: raise HTTPException( status_code=404, detail=f"Service '{service}' not found" ) stack_id = stack.get("Id") endpoint_id = stack.get("EndpointId") # Perform the action if action in ["start", "stop", "restart"]: # These actions need to be implemented # For now, return not implemented raise HTTPException( status_code=501, detail=f"Action '{action}' not yet implemented for services" ) elif action == "scale": # Scaling requires updating the stack's compose file raise HTTPException( status_code=501, detail="Scaling not yet implemented" ) return { "success": True, "action": action, "service": service, "message": f"Action '{action}' completed successfully" } except HTTPException: raise except Exception as e: logger.error(f"Failed to {action} service '{service}': {e}", exc_info=True) raise HTTPException(status_code=500, detail=str(e)) @router.get( "/services/{service}/status", response_model=Dict[str, Any], summary="Get detailed service status" ) async def get_service_status(service: str): """ Get comprehensive service status including containers, resources, and events. Args: service: Service/stack name Returns: Detailed status information """ portainer = get_portainer_client() logger.info(f"Getting status for service '{service}'") try: # Find the stack stacks = await portainer.get_stacks() stack = next( (s for s in stacks if s.get("Name", "").lower() == service.lower()), None ) if not stack: raise HTTPException( status_code=404, detail=f"Service '{service}' not found" ) stack_id = stack.get("Id") endpoint_id = stack.get("EndpointId") status_code = stack.get("Status", 0) # Get containers for this stack all_containers = await portainer.list_containers(all_containers=True) # Filter containers belonging to this stack # Stack name is usually in the container labels stack_containers = [] for container in all_containers: labels = container.get('Labels', {}) # Check if container belongs to this stack # Docker Compose adds labels like: com.docker.compose.project project = labels.get('com.docker.compose.project', '').lower() if project == service.lower(): stack_containers.append(container) # Format container info container_info = [] running_count = 0 for c in stack_containers: state = c.get('State', 'unknown') if state == 'running': running_count += 1 container_info.append({ 'name': c.get('Names', ['unknown'])[0].lstrip('/'), 'status': state, 'health': 'N/A', # Would need to inspect container for health 'uptime': c.get('Status', 'N/A') }) total_containers = len(stack_containers) replica_status = f"{running_count}/{total_containers} running" return { "name": stack.get("Name"), "status": "active" if status_code == 1 else "inactive", "stack_id": stack_id, "replica_status": replica_status, "containers": container_info, "resources": { "memory_total": "N/A", # Would need to aggregate container stats "cpu_usage": "N/A" }, "recent_events": [] # Would need to query Docker events API } except HTTPException: raise except Exception as e: logger.error(f"Failed to get status for service '{service}': {e}", exc_info=True) raise HTTPException(status_code=500, detail=str(e)) @router.get( "/ports", response_model=List[PortInfo], summary="List allocated ports" ) async def list_ports(): """ List all currently allocated ports Scans all running containers to extract: - Internal and external port mappings - Container internal hostnames and IPs - External domain names (from NPM proxy configuration) """ portainer = get_portainer_client() npm = get_npm_client() try: # Get all endpoints (Docker environments) endpoints = await portainer.get_endpoints() # Get proxy hosts for external domain mapping proxy_hosts = await npm.get_proxy_hosts() # Build mapping of forward_host:forward_port -> domains port_domain_map = {} for proxy in proxy_hosts: forward_host = proxy.get("forward_host", "") forward_port = proxy.get("forward_port", 0) domains = proxy.get("domain_names", []) key = f"{forward_host}:{forward_port}" if key not in port_domain_map: port_domain_map[key] = [] port_domain_map[key].extend(domains) ports = [] # Scan containers on each endpoint for endpoint in endpoints: endpoint_id = endpoint.get("Id") try: containers = await portainer.get_containers(endpoint_id, all_containers=False) for container in containers: container_name = container.get("Names", ["unknown"])[0].lstrip("/") state = container.get("State", "") # Skip non-running containers if state != "running": continue # Extract network information networks = container.get("NetworkSettings", {}).get("Networks", {}) internal_hostname = container_name internal_ip = None # Get first network IP for network_name, network_info in networks.items(): if network_info.get("IPAddress"): internal_ip = network_info.get("IPAddress") break # Extract port mappings port_mappings = container.get("Ports", []) for port_mapping in port_mappings: internal_port = port_mapping.get("PrivatePort") host_port = port_mapping.get("PublicPort") protocol = port_mapping.get("Type", "tcp") if not internal_port: continue # Find external domains for this port external_domains = [] # Try matching by container name and port key_by_name = f"{container_name}:{internal_port}" if key_by_name in port_domain_map: external_domains.extend(port_domain_map[key_by_name]) # Try matching by internal IP and port if internal_ip: key_by_ip = f"{internal_ip}:{internal_port}" if key_by_ip in port_domain_map: external_domains.extend(port_domain_map[key_by_ip]) # Try matching by localhost and host port if host_port: for localhost_variant in ["localhost", "127.0.0.1", "192.168.86.149"]: key_by_host = f"{localhost_variant}:{host_port}" if key_by_host in port_domain_map: external_domains.extend(port_domain_map[key_by_host]) # Deduplicate external domains external_domains = list(set(external_domains)) # Get service name from stack label or container name labels = container.get("Labels", {}) service_name = labels.get("com.docker.compose.service", container_name) port_info = PortInfo( port=internal_port, service=service_name, container_name=container_name, protocol=protocol, internal_hostname=internal_hostname, internal_ip=internal_ip, external_domains=external_domains, host_port=host_port, description=f"{container_name} on {endpoint.get('Name', 'unknown')}" ) ports.append(port_info) except Exception as e: logger.error(f"Failed to scan containers on endpoint {endpoint_id}: {e}") continue # Deduplicate ports based on (port, container_name, protocol) seen = set() unique_ports = [] for port_info in ports: key = (port_info.port, port_info.container_name, port_info.protocol) if key not in seen: seen.add(key) unique_ports.append(port_info) # Sort by port number unique_ports.sort(key=lambda p: p.port) return unique_ports except Exception as e: logger.error(f"Failed to list ports: {e}") raise HTTPException(status_code=500, detail=str(e)) @router.get( "/domains", response_model=List[DomainInfo], summary="List configured domains" ) async def list_domains(): """ List all configured domain names Returns domain-to-service mappings with SSL status. """ npm = get_npm_client() try: proxy_hosts = await npm.get_proxy_hosts() domains = [] for proxy in proxy_hosts: service_name = proxy.get("forward_host", "localhost") certificate_id = proxy.get("certificate_id", 0) for domain in proxy.get("domain_names", []): domain_info = DomainInfo( domain=domain, service=service_name, proxy_host_id=proxy.get("id"), ssl_enabled=certificate_id > 0, certificate_id=certificate_id if certificate_id > 0 else None ) domains.append(domain_info) return domains except Exception as e: logger.error(f"Failed to list domains: {e}") raise HTTPException(status_code=500, detail=str(e)) # Write endpoints @router.post( "/services", response_model=OperationResult, summary="Deploy a new Docker Compose stack", description="Deploy a new service by creating a Docker Compose stack in Portainer. Provide the stack name and compose file content. Requires admin authentication.", status_code=201 ) async def deploy_service( request: DeployServiceRequest, user: Dict = Depends(get_admin_user) ): """ Deploy a new Docker Compose stack via Portainer Creates a new Portainer stack from the provided Docker Compose content. This is equivalent to deploying a stack through the Portainer UI. Args: request: Stack deployment configuration containing: - name: Stack name (must be unique) - compose_content: Full docker-compose.yml content as string - endpoint_id: Portainer endpoint ID (default: 3 for local) Returns: Operation result with created stack details including stack ID Raises: 409: If a stack with the same name already exists 500: If deployment fails """ portainer = get_portainer_client() try: # Check if stack already exists stacks = await portainer.get_stacks() existing = next((s for s in stacks if s.get("Name") == request.name), None) if existing: raise HTTPException( status_code=409, detail=f"Service '{request.name}' already exists with ID {existing.get('Id')}" ) # Create new stack result = await portainer.create_stack( name=request.name, stack_file_content=request.compose_content, endpoint_id=request.endpoint_id ) logger.info(f"Deployed service '{request.name}' (stack ID: {result.get('Id')})") return OperationResult( success=True, message=f"Service '{request.name}' deployed successfully", details=result ) except HTTPException: raise except Exception as e: logger.error(f"Failed to deploy service '{request.name}': {e}") raise HTTPException(status_code=500, detail=str(e)) @router.put( "/services/{name}", response_model=OperationResult, summary="Update an existing Docker Compose stack", description="Update a deployed stack's Docker Compose configuration. This redeploys the stack with the new configuration. Requires admin authentication." ) async def update_service( name: str, request: UpdateServiceRequest, user: Dict = Depends(get_admin_user) ): """ Update an existing Docker Compose stack's configuration Updates the stack's compose file and redeploys it. This is equivalent to updating a stack through the Portainer UI. Args: name: Service/stack name request: Update configuration Returns: Operation result with updated stack details """ portainer = get_portainer_client() try: # Find stack by name stacks = await portainer.get_stacks() stack = next( (s for s in stacks if s.get("Name", "").lower() == name.lower()), None ) if not stack: raise HTTPException(status_code=404, detail=f"Service '{name}' not found") stack_id = stack.get("Id") endpoint_id = stack.get("EndpointId") # Update stack result = await portainer.update_stack( stack_id=stack_id, stack_file_content=request.compose_content, endpoint_id=endpoint_id, prune=request.prune, pull_image=request.pull_image ) logger.info(f"Updated service '{name}' (stack ID: {stack_id})") return OperationResult( success=True, message=f"Service '{name}' updated successfully", details=result ) except HTTPException: raise except Exception as e: logger.error(f"Failed to update service '{name}': {e}") raise HTTPException(status_code=500, detail=str(e)) @router.delete( "/services/{name}", response_model=OperationResult, summary="Delete a service", description="Delete a service and remove its stack. Requires admin authentication." ) async def delete_service( name: str, user: Dict = Depends(get_admin_user) ): """ Delete a service and remove its stack Args: name: Service/stack name Returns: Operation result confirmation """ portainer = get_portainer_client() try: # Find stack by name stacks = await portainer.get_stacks() stack = next( (s for s in stacks if s.get("Name", "").lower() == name.lower()), None ) if not stack: raise HTTPException(status_code=404, detail=f"Service '{name}' not found") stack_id = stack.get("Id") endpoint_id = stack.get("EndpointId") # Delete stack await portainer.delete_stack( stack_id=stack_id, endpoint_id=endpoint_id ) logger.info(f"Deleted service '{name}' (stack ID: {stack_id})") return OperationResult( success=True, message=f"Service '{name}' deleted successfully", details={"stack_id": stack_id} ) except HTTPException: raise except Exception as e: logger.error(f"Failed to delete service '{name}': {e}") raise HTTPException(status_code=500, detail=str(e)) @router.get( "/proxy/{proxy_id}", summary="Get proxy host details" ) async def get_proxy_host(proxy_id: int): """ Get detailed configuration of a specific proxy host Args: proxy_id: NPM proxy host ID Returns: Complete proxy host configuration including locations """ npm = get_npm_client() try: proxy_host = await npm.get_proxy_host(proxy_id) return proxy_host except Exception as e: logger.error(f"Failed to get proxy host {proxy_id}: {e}") raise HTTPException(status_code=500, detail=str(e)) @router.post( "/proxy", response_model=OperationResult, summary="Create a new proxy host", description="Create a new Nginx Proxy Manager proxy host with optional SSL certificate. Requires admin authentication.", status_code=201 ) async def create_proxy( request: CreateProxyRequest, user: Dict = Depends(get_admin_user) ): """ Create a new Nginx Proxy Manager proxy host Optionally request an SSL certificate from Let's Encrypt. Args: request: Proxy host configuration Returns: Operation result with proxy host details """ npm = get_npm_client() try: certificate_id = 0 # Request SSL certificate if requested if request.request_ssl_certificate: logger.info(f"Requesting SSL certificate for {request.domain_names}") cert_result = await npm.create_certificate( domain_names=request.domain_names ) certificate_id = cert_result.get("id", 0) logger.info(f"SSL certificate created: ID {certificate_id}") # Create proxy host proxy_result = await npm.create_proxy_host( domain_names=request.domain_names, forward_host=request.forward_host, forward_port=request.forward_port, forward_scheme=request.forward_scheme, certificate_id=certificate_id, ssl_forced=request.ssl_enabled, block_exploits=request.block_exploits, websocket_upgrade=request.websocket_upgrade, http2_support=request.http2_support ) logger.info(f"Created proxy host for {request.domain_names} → {request.forward_host}:{request.forward_port}") return OperationResult( success=True, message=f"Proxy host created for {', '.join(request.domain_names)}", details={ "proxy_host": proxy_result, "certificate_id": certificate_id if certificate_id > 0 else None } ) except Exception as e: logger.error(f"Failed to create proxy host: {e}") raise HTTPException(status_code=500, detail=str(e)) @router.put( "/proxy/{proxy_id}", response_model=OperationResult, summary="Update a proxy host", description="Update an existing Nginx Proxy Manager proxy host configuration. Requires admin authentication." ) async def update_proxy( proxy_id: int, config: Dict[str, Any], user: Dict = Depends(get_admin_user) ): """ Update an existing Nginx Proxy Manager proxy host Args: proxy_id: Proxy host ID to update config: Full proxy host configuration (get from get_proxy_host, modify, then update) Returns: Operation result with updated proxy host details """ npm = get_npm_client() try: result = await npm.update_proxy_host(proxy_id, config) logger.info(f"Updated proxy host {proxy_id}: {result.get('domain_names', [])}") return OperationResult( success=True, message=f"Proxy host {proxy_id} updated successfully", details={"proxy_host": result} ) except Exception as e: logger.error(f"Failed to update proxy host {proxy_id}: {e}") raise HTTPException(status_code=500, detail=str(e)) # Service Control Endpoints @router.get( "/service-groups", summary="List service groups" ) async def list_service_groups(): """ List all defined service groups Returns service groups with their member services and status. """ return { "groups": service_groups.list_service_groups(), "always_on": list(service_groups.ALWAYS_ON_SERVICES), "stoppable": service_groups.list_stoppable_services() } @router.post( "/services/{name}/stop", response_model=OperationResult, summary="Stop a service or service group", description="Stop a service or service group by stopping containers. Requires admin authentication when accessed externally via api.schweitz.net." ) async def stop_service( name: str, user: Dict = Depends(get_forward_auth_admin) ): """ Stop a service or service group This will: 1. Validate service can be stopped (not always-on) 2. Stop the Portainer stack containers Args: name: Service or group name Returns: Operation result with details """ portainer = get_portainer_client() try: # Get all services in the group services = service_groups.get_service_group(name) # Validate none are always-on is_valid, error_msg = service_groups.validate_stop_request(services) if not is_valid: raise HTTPException(status_code=403, detail=error_msg) results = { "stopped_services": [], "errors": [] } # Stop each service for service_name in services: try: # Stop Portainer stack containers stacks = await portainer.get_stacks() stack = next( (s for s in stacks if s.get("Name", "").lower() == service_name.lower()), None ) if stack: stack_id = stack.get("Id") endpoint_id = stack.get("EndpointId") logger.info(f"Stopping containers for stack: {service_name}") # Get containers for this stack containers = await portainer.get_containers(endpoint_id, all_containers=False) stopped_containers = [] for container in containers: labels = container.get("Labels", {}) container_stack = labels.get("com.docker.compose.project", "") if container_stack.lower() == service_name.lower(): container_id = container.get("Id") # Stop container via Portainer Docker API await portainer.stop_container(endpoint_id, container_id) stopped_containers.append(container.get("Names", ["unknown"])[0]) results["stopped_services"].append({ "service": service_name, "stack_id": stack_id, "containers": stopped_containers }) logger.info(f"Stopped service: {service_name}") else: results["errors"].append(f"Stack not found: {service_name}") except Exception as e: logger.error(f"Failed to stop service {service_name}: {e}") results["errors"].append(f"{service_name}: {str(e)}") success = len(results["stopped_services"]) > 0 message = f"Stopped {len(results['stopped_services'])} service(s)" if results["errors"]: message += f" with {len(results['errors'])} error(s)" return OperationResult( success=success, message=message, details=results ) except HTTPException: raise except Exception as e: logger.error(f"Failed to stop service group '{name}': {e}") raise HTTPException(status_code=500, detail=str(e)) @router.post( "/services/{name}/start", response_model=OperationResult, summary="Start a service or service group", description="Start a service or service group by starting containers. Requires admin authentication when accessed externally via api.schweitz.net." ) async def start_service( name: str, user: Dict = Depends(get_forward_auth_admin) ): """ Start a service or service group This will: 1. Start the Portainer stack containers Args: name: Service or group name Returns: Operation result with details """ portainer = get_portainer_client() try: # Get all services in the group services = service_groups.get_service_group(name) results = { "started_services": [], "errors": [] } # Start each service for service_name in services: try: # Start Portainer stack containers stacks = await portainer.get_stacks() stack = next( (s for s in stacks if s.get("Name", "").lower() == service_name.lower()), None ) if stack: stack_id = stack.get("Id") endpoint_id = stack.get("EndpointId") logger.info(f"Starting containers for stack: {service_name}") # Get containers for this stack containers = await portainer.get_containers(endpoint_id, all_containers=True) started_containers = [] for container in containers: labels = container.get("Labels", {}) container_stack = labels.get("com.docker.compose.project", "") if container_stack.lower() == service_name.lower(): container_id = container.get("Id") # Start container via Portainer Docker API await portainer.start_container(endpoint_id, container_id) started_containers.append(container.get("Names", ["unknown"])[0]) results["started_services"].append({ "service": service_name, "stack_id": stack_id, "containers": started_containers }) logger.info(f"Started service: {service_name}") else: results["errors"].append(f"Stack not found: {service_name}") except Exception as e: logger.error(f"Failed to start service {service_name}: {e}") results["errors"].append(f"{service_name}: {str(e)}") success = len(results["started_services"]) > 0 message = f"Started {len(results['started_services'])} service(s)" if results["errors"]: message += f" with {len(results['errors'])} error(s)" return OperationResult( success=success, message=message, details=results ) except HTTPException: raise except Exception as e: logger.error(f"Failed to start service group '{name}': {e}") raise HTTPException(status_code=500, detail=str(e)) @router.get( "/widget-data", summary="Get combined data for service control widget", response_model=Dict[str, Any] ) async def get_widget_data(): """ Get combined service data for the widget Returns all data needed by service-control widget in a single call: - Service list with status and container counts - Service groups and always-on list This endpoint is designed for browser-based widgets to avoid multiple API calls and cross-origin issues. """ try: portainer = get_portainer_client() npm = get_npm_client() # Fetch services (same logic as /services endpoint) stacks = await portainer.get_stacks() proxy_hosts = await npm.get_proxy_hosts() # Build domain mapping domain_map = {} for proxy in proxy_hosts: for domain in proxy.get("domain_names", []): forward_host = proxy.get("forward_host", "") domain_map[domain] = forward_host services = [] for stack in stacks: stack_name = stack.get("Name", "") endpoint_id = stack.get("EndpointId") domains = [ domain for domain, host in domain_map.items() if stack_name in host or host in stack_name ] # Get container status containers_running = 0 containers_total = 0 try: all_containers = await portainer.get_containers(endpoint_id, all_containers=True) for container in all_containers: labels = container.get("Labels", {}) container_stack = labels.get("com.docker.compose.project", "") if container_stack.lower() == stack_name.lower(): containers_total += 1 if container.get("State", "") == "running": containers_running += 1 except Exception as e: logger.warning(f"Failed to get container status for {stack_name}: {e}") services.append({ "name": stack_name, "stack_id": stack.get("Id"), "status": "active" if stack.get("Status") == 1 else "inactive", "endpoint_id": endpoint_id, "domains": domains, "running": containers_running > 0, "containers_running": containers_running, "containers_total": containers_total }) return { "success": True, "services": services, "service_groups": { "groups": service_groups.list_service_groups(), "always_on": list(service_groups.ALWAYS_ON_SERVICES), "stoppable": service_groups.list_stoppable_services() } } except Exception as e: logger.error(f"Failed to fetch widget data: {e}") raise HTTPException(status_code=500, detail=f"Failed to fetch widget data: {str(e)}") # ======================================================================== # Container Management Endpoints (for core-ai infrastructure tools) # ======================================================================== @router.get( "/containers", response_model=List[Dict[str, Any]], summary="List Docker containers" ) async def list_containers(status: Optional[str] = "running"): """ List Docker containers with optional status filter. Args: status: Filter by status - "all", "running", "stopped", "paused" (default: "running") Returns: List of containers in Docker API format """ portainer = get_portainer_client() logger.info(f"Listing containers (status filter: {status})") try: # Get all containers using Portainer with Docker socket fallback all_containers_flag = status in ["all", "stopped"] containers = await portainer.list_containers(all_containers=all_containers_flag) # Filter by status if needed if status == "running": containers = [c for c in containers if c.get('State') == 'running'] elif status == "stopped": containers = [c for c in containers if c.get('State') != 'running'] elif status == "paused": containers = [c for c in containers if c.get('State') == 'paused'] logger.info(f"Found {len(containers)} containers matching status '{status}'") return containers except Exception as e: logger.error(f"Failed to list containers: {e}", exc_info=True) raise HTTPException(status_code=500, detail=f"Failed to list containers: {str(e)}") @router.post( "/containers/{container}/{action}", response_model=Dict[str, Any], summary="Manage container state" ) async def manage_container(container: str, action: str): """ Perform lifecycle operation on a Docker container. Args: container: Container name or ID action: One of: "start", "stop", "restart", "pause", "unpause", "remove" Returns: Success status and message """ portainer = get_portainer_client() logger.info(f"Managing container '{container}': action={action}") # Validate action valid_actions = ["start", "stop", "restart", "pause", "unpause", "remove"] if action not in valid_actions: raise HTTPException( status_code=400, detail=f"Invalid action '{action}'. Must be one of: {', '.join(valid_actions)}" ) try: # Find the container all_containers = await portainer.list_containers(all_containers=True) matching_container = None for c in all_containers: names = c.get('Names', []) for name in names: clean_name = name.lstrip('/') if clean_name.lower() == container.lower(): matching_container = c break if matching_container: break if not matching_container: raise HTTPException( status_code=404, detail=f"Container '{container}' not found" ) container_id = matching_container['Id'] # Get endpoints endpoints = await portainer.get_endpoints() if not endpoints: raise HTTPException(status_code=500, detail="No Portainer endpoints available") endpoint_id = endpoints[0]['Id'] # Perform the action if action == "start": await portainer.start_container(endpoint_id, container_id) message = f"Started container '{container}' successfully" elif action == "stop": await portainer.stop_container(endpoint_id, container_id) message = f"Stopped container '{container}' successfully" elif action == "restart": await portainer.stop_container(endpoint_id, container_id) await portainer.start_container(endpoint_id, container_id) message = f"Restarted container '{container}' successfully" elif action in ["pause", "unpause", "remove"]: # These actions need to be added to Portainer client # For now, return error raise HTTPException( status_code=501, detail=f"Action '{action}' not yet implemented" ) logger.info(f"Successfully {action}ed container '{container}'") return { "success": True, "action": action, "container": container, "message": message } except HTTPException: raise except Exception as e: logger.error(f"Failed to {action} container '{container}': {e}", exc_info=True) # Provide helpful error messages error_str = str(e).lower() if "304" in error_str or "not modified" in error_str: raise HTTPException( status_code=304, detail=f"Container '{container}' is already in the target state" ) elif "conflict" in error_str: raise HTTPException( status_code=409, detail=f"Cannot {action} container '{container}': state conflict" ) else: raise HTTPException(status_code=500, detail=str(e)) @router.get( "/containers/{container}", response_model=Dict[str, Any], summary="Inspect container" ) async def inspect_container( container: str, details: Optional[str] = "summary" ): """ Get detailed information about a Docker container. Args: container: Container name or ID details: Level of detail - "summary", "full", or "resources" (not used server-side, for client formatting) Returns: Container details in Docker inspect format """ portainer = get_portainer_client() logger.info(f"Inspecting container '{container}' (details={details})") try: # Use Portainer's inspect_container which auto-detects endpoint and falls back to Docker socket info = await portainer.inspect_container(container) if not info: raise HTTPException( status_code=404, detail=f"Container '{container}' not found" ) logger.info(f"Successfully inspected container '{container}'") return info except HTTPException: raise except Exception as e: logger.error(f"Failed to inspect container '{container}': {e}", exc_info=True) raise HTTPException(status_code=500, detail=str(e)) @router.get( "/containers/{container}/logs", response_model=Dict[str, Any], summary="Get container logs" ) async def get_container_logs( container: str, lines: Optional[int] = 50, since: Optional[str] = None ): """ Retrieve logs from a Docker container. Args: container: Container name or ID lines: Number of log lines to retrieve (default: 50, max: 500) since: Time filter - "1h", "30m", or ISO timestamp (not yet implemented) Returns: Container logs with metadata """ import httpx logger.info(f"Retrieving logs for container '{container}' (lines={lines}, since={since})") # Clamp lines to reasonable limit lines = min(max(1, lines), 500) try: # Access Docker socket directly to get logs transport = httpx.AsyncHTTPTransport(uds="/var/run/docker.sock") async with httpx.AsyncClient(transport=transport, timeout=15.0) as client: # Get container logs via Docker API params = { "stdout": "true", "stderr": "true", "tail": lines, "timestamps": "true" } # TODO: Add 'since' parameter support # if since: # params["since"] = parse_since_to_timestamp(since) response = await client.get( f"http://localhost/v1.41/containers/{container}/logs", params=params ) if response.status_code == 404: raise HTTPException( status_code=404, detail=f"Container '{container}' not found" ) response.raise_for_status() logs_raw = response.text # Docker logs come with binary prefixes (8 bytes per line) # Strip these headers for cleaner output lines_list = logs_raw.split('\n') cleaned_lines = [] for line in lines_list: if len(line) > 8: # Skip the 8-byte Docker stream header if present cleaned_line = line[8:] if line[0:1] in [b'\x01', b'\x02', '\x01', '\x02'] else line cleaned_lines.append(cleaned_line) elif line: cleaned_lines.append(line) logs = '\n'.join(cleaned_lines).strip() logger.info(f"Successfully retrieved logs for container '{container}' ({len(cleaned_lines)} lines)") return { "container": container, "lines_requested": lines, "since": since, "logs": logs } except HTTPException: raise except Exception as e: logger.error(f"Failed to retrieve logs for container '{container}': {e}", exc_info=True) raise HTTPException(status_code=500, detail=str(e)) # ======================================================================== # Monitoring & Resource Endpoints (for core-ai infrastructure tools) # ======================================================================== @router.get( "/resources/system", response_model=Dict[str, Any], summary="Get system resource usage" ) async def get_system_resources(): """ Get overall system resource usage. Returns: System-level metrics including CPU, memory, disk, and network """ logger.info("Getting system resources") try: # Access Docker system info via socket import httpx transport = httpx.AsyncHTTPTransport(uds="/var/run/docker.sock") async with httpx.AsyncClient(transport=transport, timeout=10.0) as client: # Get system info info_response = await client.get("http://localhost/v1.41/info") info_response.raise_for_status() info = info_response.json() # Get system df (disk usage) df_response = await client.get("http://localhost/v1.41/system/df") df_response.raise_for_status() df = df_response.json() # Extract system metrics ncpu = info.get('NCPU', 0) mem_total = info.get('MemTotal', 0) # Calculate memory usage (rough estimate) # Docker doesn't directly provide used memory, so we estimate mem_used = mem_total * 0.5 # Placeholder - would need psutil or /proc mem_available = mem_total - mem_used mem_usage_pct = (mem_used / mem_total * 100) if mem_total > 0 else 0 # Disk usage from system/df layers_size = sum(img.get('Size', 0) for img in df.get('Images', [])) containers_size = sum(c.get('SizeRw', 0) for c in df.get('Containers', [])) volumes_size = sum(v.get('UsageData', {}).get('Size', 0) for v in df.get('Volumes', [])) total_disk_used = layers_size + containers_size + volumes_size return { "cpu": { "cores": ncpu, "usage_percent": None, # Would need to calculate from stats over time "load_average": [] # Not available from Docker API }, "memory": { "total_bytes": mem_total, "used_bytes": mem_used, "available_bytes": mem_available, "usage_percent": mem_usage_pct }, "disk": { "total_bytes": None, # Not available from Docker API "used_bytes": total_disk_used, "available_bytes": None, "usage_percent": None }, "network": { "interfaces": {} # Would need to parse network stats } } except Exception as e: logger.error(f"Failed to get system resources: {e}", exc_info=True) raise HTTPException(status_code=500, detail=str(e)) @router.get( "/resources/containers", response_model=List[Dict[str, Any]], summary="Get container resource usage" ) async def get_container_resources(container: Optional[str] = None): """ Get container-specific resource usage. Args: container: Optional specific container name/ID Returns: List of container resource stats """ logger.info(f"Getting container resources (container={container})") try: import httpx portainer = get_portainer_client() # Get list of containers if container: # Get specific container all_containers = await portainer.list_containers(all_containers=False) containers = [c for c in all_containers if container.lower() in c.get('Names', [''])[0].lower()] if not containers: raise HTTPException(status_code=404, detail=f"Container '{container}' not found") else: # Get all running containers containers = await portainer.list_containers(all_containers=False) # Get stats for each container stats_list = [] transport = httpx.AsyncHTTPTransport(uds="/var/run/docker.sock") async with httpx.AsyncClient(transport=transport, timeout=15.0) as client: for c in containers: container_id = c.get('Id') name = c.get('Names', ['unknown'])[0].lstrip('/') try: # Get container stats (one-time, not streaming) stats_response = await client.get( f"http://localhost/v1.41/containers/{container_id}/stats", params={"stream": "false"} ) stats_response.raise_for_status() stats = stats_response.json() # Parse stats cpu_stats = stats.get('cpu_stats', {}) precpu_stats = stats.get('precpu_stats', {}) memory_stats = stats.get('memory_stats', {}) networks = stats.get('networks', {}) blkio_stats = stats.get('blkio_stats', {}) # Calculate CPU percentage cpu_delta = cpu_stats.get('cpu_usage', {}).get('total_usage', 0) - \ precpu_stats.get('cpu_usage', {}).get('total_usage', 0) system_delta = cpu_stats.get('system_cpu_usage', 0) - \ precpu_stats.get('system_cpu_usage', 0) online_cpus = cpu_stats.get('online_cpus', 1) cpu_percent = 0.0 if system_delta > 0 and cpu_delta > 0: cpu_percent = (cpu_delta / system_delta) * online_cpus * 100.0 # Memory stats mem_usage = memory_stats.get('usage', 0) mem_limit = memory_stats.get('limit', 0) mem_percent = (mem_usage / mem_limit * 100) if mem_limit > 0 else 0 # Network stats net_rx = sum(net.get('rx_bytes', 0) for net in networks.values()) net_tx = sum(net.get('tx_bytes', 0) for net in networks.values()) # Block I/O stats io_service_bytes = blkio_stats.get('io_service_bytes_recursive', []) block_read = sum(entry.get('value', 0) for entry in io_service_bytes if entry.get('op') == 'Read') block_write = sum(entry.get('value', 0) for entry in io_service_bytes if entry.get('op') == 'Write') stats_list.append({ "name": name, "cpu_percent": cpu_percent, "memory_usage_bytes": mem_usage, "memory_limit_bytes": mem_limit, "memory_percent": mem_percent, "network_rx_bytes": net_rx, "network_tx_bytes": net_tx, "block_read_bytes": block_read, "block_write_bytes": block_write }) except Exception as e: logger.warning(f"Failed to get stats for container {name}: {e}") # Skip containers that fail to get stats continue return stats_list except HTTPException: raise except Exception as e: logger.error(f"Failed to get container resources: {e}", exc_info=True) raise HTTPException(status_code=500, detail=str(e)) return router # Create controller instance infrastructure_controller = InfrastructureController()