alex
aa25852091
- Inject explicit container_name into compose files at start/restart time via _ensure_container_name_in_compose() to prevent Docker Compose from generating UUID-based auto names that break backend network resolution. - Use instance.name.lower() directly instead of get_container_name() lookups which were unreliable with auto-generated names. - Apply compose sanitization, bind-address fix, and container-name injection on restart_instance as well so restarts pick up template fixes. - Add --force-recreate to docker compose up to ensure container_name changes take effect immediately. - Fix notification lifecycle tests to match current behavior (success severity, health_changed event for ownership test). Quality gates: ruff clean, pytest (7 notification lifecycle tests passed)
718 lines
21 KiB
Python
718 lines
21 KiB
Python
"""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", "--force-recreate"])
|
||
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 _check_app_binding(container_name: str, port: int) -> dict[str, str | bool]:
|
||
"""Diagnose whether the app is bound to 127.0.0.1 or 0.0.0.0.
|
||
|
||
Checks from both inside the container (localhost) and outside
|
||
(via Docker network) to detect binding issues.
|
||
|
||
Returns:
|
||
Dict with 'internal_ok', 'external_ok', 'internal_status',
|
||
'external_status', and 'diagnosis'.
|
||
"""
|
||
import subprocess
|
||
|
||
result: dict[str, Any] = {
|
||
"internal_ok": False,
|
||
"external_ok": False,
|
||
"internal_status": None,
|
||
"external_status": None,
|
||
"diagnosis": "unknown",
|
||
}
|
||
|
||
# Check from inside the container (loopback)
|
||
internal = subprocess.run(
|
||
[
|
||
"docker",
|
||
"exec",
|
||
container_name,
|
||
"sh",
|
||
"-c",
|
||
f"curl -s -o /dev/null -w '%{{http_code}}' http://localhost:{port}",
|
||
],
|
||
capture_output=True,
|
||
text=True,
|
||
timeout=5,
|
||
)
|
||
if internal.returncode == 0:
|
||
try:
|
||
result["internal_status"] = int(internal.stdout.strip())
|
||
result["internal_ok"] = result["internal_status"] > 0
|
||
except ValueError:
|
||
pass
|
||
|
||
# Check from outside the container (Docker network)
|
||
external = subprocess.run(
|
||
[
|
||
"curl",
|
||
"-s",
|
||
"-o",
|
||
"/dev/null",
|
||
"-w",
|
||
"%{http_code}",
|
||
f"http://{container_name}:{port}",
|
||
],
|
||
capture_output=True,
|
||
text=True,
|
||
timeout=5,
|
||
)
|
||
if external.returncode == 0:
|
||
try:
|
||
result["external_status"] = int(external.stdout.strip())
|
||
result["external_ok"] = result["external_status"] > 0
|
||
except ValueError:
|
||
pass
|
||
|
||
# Diagnose binding issue
|
||
if result["internal_ok"] and not result["external_ok"]:
|
||
result["diagnosis"] = (
|
||
f"App appears to be bound to 127.0.0.1:{port} inside the container. "
|
||
f"It must bind to 0.0.0.0:{port} to be accessible from the tunnel."
|
||
)
|
||
elif result["internal_ok"] and result["external_ok"]:
|
||
result["diagnosis"] = "App is accessible on both interfaces."
|
||
elif not result["internal_ok"] and not result["external_ok"]:
|
||
result["diagnosis"] = f"App is not responding on port {port} at all."
|
||
else:
|
||
result["diagnosis"] = "Unexpected binding state."
|
||
|
||
return result
|
||
|
||
|
||
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 from the Docker network
|
||
logger.info("Checking connectivity to %s:%d...", container_name, port)
|
||
accessible = False
|
||
last_status = None
|
||
for attempt in range(30): # 30 attempts × 1s = 30s max wait for app startup
|
||
check = subprocess.run(
|
||
[
|
||
"curl",
|
||
"-s",
|
||
"-o",
|
||
"/dev/null",
|
||
"-w",
|
||
"%{http_code}",
|
||
"--max-time",
|
||
"3",
|
||
f"http://{container_name}:{port}",
|
||
],
|
||
capture_output=True,
|
||
text=True,
|
||
timeout=5,
|
||
)
|
||
status_str = check.stdout.strip()
|
||
logger.info(
|
||
"Connectivity check %d/%d: http_code=%s (rc=%d)",
|
||
attempt + 1,
|
||
30,
|
||
status_str,
|
||
check.returncode,
|
||
)
|
||
try:
|
||
last_status = int(status_str)
|
||
# Accept 2xx, 3xx, 401, 403 as "app is listening"
|
||
if last_status in (401, 403) or 200 <= last_status < 400:
|
||
accessible = True
|
||
logger.info(
|
||
"App on %s:%d is ready (HTTP %d)",
|
||
container_name,
|
||
port,
|
||
last_status,
|
||
)
|
||
break
|
||
except ValueError:
|
||
pass
|
||
|
||
if check.returncode != 0:
|
||
logger.debug(
|
||
"curl failed: stderr=%s", check.stderr.strip() if check.stderr else ""
|
||
)
|
||
time.sleep(1)
|
||
|
||
if not accessible:
|
||
logger.warning(
|
||
"Container %s:%d not responding after 30s (last status: %s). "
|
||
"Running binding diagnostics...",
|
||
container_name,
|
||
port,
|
||
last_status,
|
||
)
|
||
diagnosis = _check_app_binding(container_name, port)
|
||
logger.warning(
|
||
"Binding diagnosis: internal=%s (HTTP %s), external=%s (HTTP %s). %s",
|
||
diagnosis["internal_ok"],
|
||
diagnosis["internal_status"],
|
||
diagnosis["external_ok"],
|
||
diagnosis["external_status"],
|
||
diagnosis["diagnosis"],
|
||
)
|
||
|
||
# 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),
|
||
}
|