"""Docker service for managing tool instances.""" import logging import os import re import subprocess import time from collections import Counter from pathlib import Path from typing import Any logger = logging.getLogger(__name__) def sort_volumes_by_specificity(volumes: list[str]) -> list[str]: """Sort volume strings so parent paths come before child paths. Docker Compose mounts volumes in array order. A later mount at a parent path hides earlier mounts at child paths. By sorting shallow paths first and deep paths last, deeper (more specific) mounts overlay correctly. Volume format: source:target or source:target:type Args: volumes: List of Docker volume mount strings. Returns: Sorted list with parent paths before child paths. """ def _target_depth(vol: str) -> int: parts = vol.split(":") if len(parts) < 2: return 0 target = parts[1].rstrip("/") if not target or target == "/": return 0 return target.count("/") # Detect duplicate targets and warn targets = [] for vol in volumes: parts = vol.split(":") targets.append(parts[1] if len(parts) > 1 else "") dupes = [t for t, c in Counter(targets).items() if c > 1] if dupes: logger.warning("Duplicate mount targets detected: %s", dupes) # Stable sort: parent paths first, child paths last return sorted(volumes, key=_target_depth) def render_compose_template(template: str, variables: dict[str, Any]) -> str: """Render a Docker Compose template with variable substitution. Args: template: The compose template string variables: Dictionary of variable names to values Returns: Rendered compose file content """ result = template for key, value in variables.items(): placeholder = f"{{{{{key}}}}}" result = result.replace(placeholder, str(value)) return result def ensure_instance_directory(instance_id: str, base_path: str | None = None) -> str: """Create and return the instance directory path. Args: instance_id: Unique instance identifier base_path: Base directory for all instances (defaults to Settings.instance_base_path) Returns: Absolute path to instance directory """ if base_path is None: from src.config import Settings base_path = Settings().instance_base_path instance_dir = Path(base_path) / instance_id instance_dir.mkdir(parents=True, exist_ok=True) return str(instance_dir.absolute()) def write_compose_file(instance_dir: str, content: str) -> str: """Write the rendered compose file to the instance directory. Args: instance_dir: Path to instance directory content: Rendered compose content Returns: Path to the compose file """ compose_path = Path(instance_dir) / "docker-compose.yml" compose_path.write_text(content) return str(compose_path) def write_env_file(instance_dir: str, env_vars: dict[str, str]) -> str: """Write environment variables to a .env file. Args: instance_dir: Path to instance directory env_vars: Dictionary of env var names to values Returns: Path to the env file """ env_path = Path(instance_dir) / ".env" lines = [f'{key}="{value}"' for key, value in env_vars.items()] env_path.write_text("\n".join(lines) + "\n") return str(env_path) def write_config_files(instance_dir: str, files: dict[str, str]) -> None: """Write config files to the instance directory. Args: instance_dir: Path to instance directory files: Dictionary of file paths (relative to instance dir) to content """ instance_path = Path(instance_dir) for file_path, content in files.items(): # Ensure the path is within the instance directory (security) full_path = instance_path / file_path try: full_path.resolve().relative_to(instance_path.resolve()) except ValueError: raise ValueError(f"File path '{file_path}' escapes instance directory") full_path.parent.mkdir(parents=True, exist_ok=True) full_path.write_text(content) def execute_compose_command( compose_path: str, action: str, timeout: int = 60, env_file: str | None = None ) -> tuple[int, str, str]: """Execute a docker compose command. Args: compose_path: Path to docker-compose.yml action: The compose action (up, down, start, stop, restart) timeout: Command timeout in seconds env_file: Optional path to .env file for environment variables Returns: Tuple of (returncode, stdout, stderr) """ instance_dir = Path(compose_path).parent cmd = ["docker", "compose", "-f", compose_path] if env_file: cmd.extend(["--env-file", env_file]) if action == "up": cmd.extend(["up", "-d"]) elif action == "down": cmd.extend(["down", "-v"]) elif action in ("start", "stop", "restart"): cmd.append(action) else: raise ValueError(f"Unknown compose action: {action}") result = subprocess.run( cmd, cwd=str(instance_dir), capture_output=True, text=True, timeout=timeout, ) return result.returncode, result.stdout, result.stderr def get_container_id(instance_name: str) -> str | None: """Get the container ID for a compose service. Searches all containers including stopped/exited ones. Args: instance_name: The service name in compose Returns: Container ID or None if not found """ # Docker container names are lowercase internally; normalize to ensure match result = subprocess.run( ["docker", "ps", "-a", "-q", "--filter", f"name={instance_name.lower()}"], capture_output=True, text=True, ) if result.returncode == 0 and result.stdout.strip(): return result.stdout.strip().split("\n")[0] return None def get_container_name(instance_name: str) -> str | None: """Get the full container name for a compose service. Searches all containers including stopped/exited ones. Args: instance_name: The service name in compose Returns: Container name or None if not found """ # Docker container names are lowercase internally; normalize to ensure match result = subprocess.run( [ "docker", "ps", "-a", "--format", "{{.Names}}", "--filter", f"name={instance_name.lower()}", ], capture_output=True, text=True, ) if result.returncode == 0 and result.stdout.strip(): return result.stdout.strip().split("\n")[0] return None def connect_container_to_network( container_name: str, network_name: str = "backend" ) -> bool: """Connect a Docker container to an existing network. Args: container_name: Name or ID of the container network_name: Name of the Docker network (default: backend) Returns: True if successful, False otherwise """ result = subprocess.run( ["docker", "network", "connect", network_name, container_name], capture_output=True, text=True, ) return result.returncode == 0 def get_container_status(container_id: str) -> dict[str, Any]: """Get the status of a Docker container. Args: container_id: Docker container ID Returns: Dict with 'status' (running, exited, restarting, not_found), 'exit_code' (int or None), and 'health' (health status or None) """ result = subprocess.run( [ "docker", "inspect", "-f", "{{.State.Status}}|{{.State.ExitCode}}|{{if .State.Health}}{{.State.Health.Status}}{{else}}none{{end}}", container_id, ], capture_output=True, text=True, ) if result.returncode != 0: return {"status": "not_found", "exit_code": None, "health": None} parts = result.stdout.strip().split("|") status = parts[0] if parts else "unknown" exit_code = int(parts[1]) if len(parts) > 1 and parts[1].isdigit() else None health = parts[2] if len(parts) > 2 and parts[2] != "none" else None return {"status": status, "exit_code": exit_code, "health": health} def wait_for_container_running( container_id: str, timeout: int = 30, interval: float = 2.0 ) -> dict[str, Any]: """Wait for a container to reach the running state. Polls docker inspect until the container status is "running" or timeout. Args: container_id: Docker container ID timeout: Maximum seconds to wait interval: Seconds between polls Returns: Dict with 'success' (bool), 'status' (str), 'exit_code' (int or None), and 'waited_seconds' (float) """ start_time = time.time() while time.time() - start_time < timeout: info = get_container_status(container_id) if info["status"] == "running": return { "success": True, "status": "running", "exit_code": None, "waited_seconds": time.time() - start_time, } if info["status"] == "exited": return { "success": False, "status": "exited", "exit_code": info["exit_code"], "waited_seconds": time.time() - start_time, } if info["status"] == "not_found": return { "success": False, "status": "not_found", "exit_code": None, "waited_seconds": time.time() - start_time, } time.sleep(interval) # Timeout reached info = get_container_status(container_id) return { "success": False, "status": info["status"], "exit_code": info["exit_code"], "waited_seconds": time.time() - start_time, } def get_container_logs(container_id: str, tail: int = 100) -> str: """Get the logs of a Docker container. Args: container_id: Docker container ID tail: Number of lines to return Returns: Container logs """ result = subprocess.run( ["docker", "logs", "--tail", str(tail), container_id], capture_output=True, text=True, ) if result.returncode == 0: return result.stdout return f"Failed to get logs: {result.stderr}" def find_free_port(start: int = 10000, end: int = 20000) -> int: """Find a free TCP port in the given range. Args: start: Start of port range end: End of port range Returns: Free port number """ import socket for port in range(start, end): with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: if s.connect_ex(("localhost", port)) != 0: return port raise RuntimeError(f"No free port found in range {start}-{end}") def start_cloudflared_tunnel( container_name: str, port: int, timeout: int = 30 ) -> dict[str, str]: """Start a temporary Cloudflare tunnel for a container. Uses 'cloudflared tunnel --url' to create a temporary tunnel with a random trycloudflare.com URL. Args: container_name: Name of the Docker container to tunnel to port: Port number the container listens on timeout: Maximum seconds to wait for tunnel URL Returns: Dict with 'url' (the public tunnel URL) and 'pid' (process ID) """ import subprocess import logging logger = logging.getLogger(__name__) # First verify the container is accessible logger.info("Checking connectivity to %s:%d...", container_name, port) for attempt in range(10): check = subprocess.run( [ "curl", "-s", "-o", "/dev/null", "-w", "%{http_code}", f"http://{container_name}:{port}", ], capture_output=True, text=True, timeout=5, ) logger.info( "Connectivity check %d: http_code=%s", attempt + 1, check.stdout.strip() ) if check.returncode == 0: break time.sleep(1) else: logger.warning( "Container %s:%d not responding to curl checks", container_name, port ) # Run cloudflared in background, capture output logger.info("Starting cloudflared tunnel to http://%s:%d", container_name, port) proc = subprocess.Popen( ["cloudflared", "tunnel", "--url", f"http://{container_name}:{port}"], stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, ) # Wait for the URL to appear in output url_pattern = re.compile(r"https://[a-z0-9-]+\.trycloudflare\.com") start_time = time.time() url = None if proc.stdout is None: proc.terminate() proc.wait(timeout=5) raise RuntimeError("Failed to capture cloudflared output") while time.time() - start_time < timeout: # Read available output import select readable, _, _ = select.select([proc.stdout], [], [], 1.0) if readable: line = proc.stdout.readline() if line: match = url_pattern.search(line) if match: url = match.group(0) break if not url: proc.terminate() proc.wait(timeout=5) raise RuntimeError( f"Failed to get tunnel URL within {timeout}s. " f"cloudflared output may contain errors." ) return {"url": url, "pid": str(proc.pid)} def stop_cloudflared_tunnel(pid: str) -> None: """Stop a cloudflared tunnel process. Args: pid: Process ID of the cloudflared tunnel """ import signal try: os.kill(int(pid), signal.SIGTERM) except ProcessLookupError: pass # Already stopped def recreate_tunnel( container_name: str, port: int, old_pid: str | None = None ) -> dict[str, str]: """Recreate a temporary Cloudflare tunnel. Stops the old tunnel (if pid provided) and starts a new one. Args: container_name: Name of the Docker container to tunnel to port: Port number the container listens on old_pid: Optional PID of the old tunnel process to stop Returns: Dict with 'url' and 'pid' for the new tunnel """ if old_pid: stop_cloudflared_tunnel(old_pid) return start_cloudflared_tunnel(container_name, port) def check_tunnel_health(url: str, timeout: int = 10) -> dict[str, Any]: """Check if a tunnel URL is healthy with smart error classification. Args: url: The tunnel URL to check timeout: Request timeout in seconds Returns: Dict with 'tunnel_status' (healthy, unreachable, error_response, not_applicable), 'status_code' (int or None), 'healthy' (bool), and 'error' (str or None) """ import subprocess try: result = subprocess.run( [ "curl", "-s", "-o", "/dev/null", "-w", "%{http_code}", "--max-time", str(timeout), url, ], capture_output=True, text=True, timeout=timeout + 5, ) status_code = int(result.stdout.strip()) if 200 <= status_code < 400: return { "tunnel_status": "healthy", "status_code": status_code, "healthy": True, "error": None, } elif status_code in (502, 503, 504): # Application error, not tunnel error return { "tunnel_status": "error_response", "status_code": status_code, "healthy": False, "error": f"Application returned HTTP {status_code}", } else: return { "tunnel_status": "error_response", "status_code": status_code, "healthy": False, "error": f"HTTP {status_code}", } except subprocess.TimeoutExpired: return { "tunnel_status": "unreachable", "status_code": None, "healthy": False, "error": "Tunnel request timed out", } except (ValueError, Exception) as e: error_str = str(e).lower() # Classify connection errors if any( err in error_str for err in [ "connection refused", "econnrefused", "could not resolve", "nodename", ] ): return { "tunnel_status": "unreachable", "status_code": None, "healthy": False, "error": f"Tunnel unreachable: {e}", } return { "tunnel_status": "unreachable", "status_code": None, "healthy": False, "error": str(e), }