refactor: rewrite tunnel system with host-network cloudflared containers

Replace the subprocess-based tunnel implementation with Docker containers
running on the host network. This eliminates all container name resolution
bugs that caused tunnel 502 errors.

New design:
- Each tunnel is a docker run --network host cloudflare/cloudflared container
- cloudflared connects to localhost:{published_port} (Docker port forwarding)
- No dependency on container names, backend network DNS, or binding diagnostics
- Tunnels named predictably: tunnel-{instance_name}
- Start/stop/recreate use container names instead of PIDs

Files changed:
- NEW: apps/api/src/services/tunnel.py — clean tunnel module (start/stop/recreate/health)
- apps/api/src/services/docker.py — removed 250 lines of old tunnel code
- apps/api/src/api/tool_instances.py — use new tunnel module, store container_name
- apps/api/src/services/health_monitor.py — updated import
- apps/web/src/components/session-card.tsx — Recreate Tunnel button always visible

Quality gates: ruff clean, 13 tests passed (health_monitor + notifications)
This commit is contained in:
2026-05-30 12:28:54 +02:00
parent 401ad2e65d
commit 6cf06d2380
5 changed files with 249 additions and 375 deletions
+33 -41
View File
@@ -45,7 +45,6 @@ from src.services.config_profile_resolver import (
resolve_profile, resolve_profile,
) )
from src.services.docker import ( from src.services.docker import (
check_tunnel_health,
connect_container_to_network, connect_container_to_network,
ensure_instance_directory, ensure_instance_directory,
execute_compose_command, execute_compose_command,
@@ -53,16 +52,19 @@ from src.services.docker import (
get_container_id, get_container_id,
get_container_logs, get_container_logs,
get_container_status, get_container_status,
recreate_tunnel,
render_compose_template, render_compose_template,
sort_volumes_by_specificity, sort_volumes_by_specificity,
start_cloudflared_tunnel,
stop_cloudflared_tunnel,
wait_for_container_running, wait_for_container_running,
write_compose_file, write_compose_file,
write_config_files, write_config_files,
write_env_file, write_env_file,
) )
from src.services.tunnel import (
check_tunnel_health,
recreate_tunnel,
start_tunnel,
stop_tunnel,
)
from src.services.docker_build import build_image from src.services.docker_build import build_image
from src.services.manifest_compiler import ( from src.services.manifest_compiler import (
compile_compose, compile_compose,
@@ -1960,12 +1962,12 @@ async def start_instance(
"error": f"Tool type '{instance.tool_type_id}' not found", "error": f"Tool type '{instance.tool_type_id}' not found",
} }
instance_port = tool_type.default_port or 0 published_port = instance.port or 0
logger.debug( logger.debug(
"Tool type for instance %s: name=%s, default_port=%s, interface_type=%s", "Tool type for instance %s: name=%s, published_port=%s, interface_type=%s",
instance.id, instance.id,
tool_type.name, tool_type.name,
instance_port, published_port,
tool_type.interface_type, tool_type.interface_type,
) )
@@ -1974,23 +1976,22 @@ async def start_instance(
# Create temporary Cloudflare tunnel for public access # Create temporary Cloudflare tunnel for public access
try: try:
logger.debug( logger.debug(
"Creating temporary tunnel for instance %s (container=%s, port=%d)", "Creating tunnel for instance %s (port=%d)",
instance.id, instance.id,
instance.container_name, published_port,
instance_port,
) )
tunnel_info = start_cloudflared_tunnel( tunnel_info = start_tunnel(
container_name=instance.container_name or instance.name, instance_name=instance.name,
port=instance_port, published_port=published_port,
) )
instance.tunnel_id = tunnel_info["pid"] instance.tunnel_id = tunnel_info["container_name"]
instance.public_url = tunnel_info["url"] instance.public_url = tunnel_info["url"]
instance.url = tunnel_info["url"] instance.url = tunnel_info["url"]
await session.commit() await session.commit()
logger.debug( logger.debug(
"Created temporary tunnel for instance %s: pid=%s, url=%s", "Created tunnel for instance %s: container=%s, url=%s",
instance.id, instance.id,
tunnel_info["pid"], tunnel_info["container_name"],
tunnel_info["url"], tunnel_info["url"],
) )
except Exception as exc: except Exception as exc:
@@ -2060,9 +2061,9 @@ async def stop_instance(
# Stop Cloudflare tunnel if exists # Stop Cloudflare tunnel if exists
if instance.tunnel_id: if instance.tunnel_id:
try: try:
stop_cloudflared_tunnel(instance.tunnel_id) stop_tunnel(instance.name)
logger.debug( logger.debug(
"Stopped tunnel for instance %s (pid=%s)", "Stopped tunnel for instance %s (container=%s)",
instance.id, instance.id,
instance.tunnel_id, instance.tunnel_id,
) )
@@ -2129,9 +2130,9 @@ async def restart_instance(
# Stop old tunnel if exists # Stop old tunnel if exists
if instance.tunnel_id: if instance.tunnel_id:
try: try:
stop_cloudflared_tunnel(instance.tunnel_id) stop_tunnel(instance.name)
logger.debug( logger.debug(
"Stopped old tunnel for instance %s (pid=%s)", "Stopped old tunnel for instance %s (container=%s)",
instance.id, instance.id,
instance.tunnel_id, instance.tunnel_id,
) )
@@ -2197,17 +2198,15 @@ async def restart_instance(
"error": f"Tool type '{tool_type.name if tool_type else 'unknown'}' has no port configured", "error": f"Tool type '{tool_type.name if tool_type else 'unknown'}' has no port configured",
} }
instance_port = tool_type.default_port
# Only create tunnel for web-enabled tools # Only create tunnel for web-enabled tools
if tool_type.interface_type == "web": if tool_type.interface_type == "web":
# Create new temporary tunnel # Create new tunnel
try: try:
tunnel_info = start_cloudflared_tunnel( tunnel_info = start_tunnel(
container_name=instance.name.lower(), instance_name=instance.name,
port=instance_port, published_port=instance.port or 0,
) )
instance.tunnel_id = tunnel_info["pid"] instance.tunnel_id = tunnel_info["container_name"]
instance.public_url = tunnel_info["url"] instance.public_url = tunnel_info["url"]
instance.url = tunnel_info["url"] instance.url = tunnel_info["url"]
logger.debug( logger.debug(
@@ -2306,9 +2305,9 @@ async def delete_instance(
# Stop Cloudflare tunnel if exists # Stop Cloudflare tunnel if exists
if instance.tunnel_id: if instance.tunnel_id:
try: try:
stop_cloudflared_tunnel(instance.tunnel_id) stop_tunnel(instance.name)
logger.debug( logger.debug(
"Stopped tunnel for instance %s (pid=%s)", "Stopped tunnel for instance %s (container=%s)",
instance.id, instance.id,
instance.tunnel_id, instance.tunnel_id,
) )
@@ -2438,26 +2437,19 @@ async def recreate_tunnel_endpoint(
"message": "Tunnel is already healthy", "message": "Tunnel is already healthy",
} }
# Get tool type for default port
tool_type = await session.get(ToolType, instance.tool_type_id)
instance_port = (
tool_type.default_port if tool_type and tool_type.default_port else 8080
)
try: try:
tunnel_info = recreate_tunnel( tunnel_info = recreate_tunnel(
container_name=instance.container_name or instance.name, instance_name=instance.name,
port=instance_port, published_port=instance.port or 0,
old_pid=instance.tunnel_id,
) )
instance.tunnel_id = tunnel_info["pid"] instance.tunnel_id = tunnel_info["container_name"]
instance.public_url = tunnel_info["url"] instance.public_url = tunnel_info["url"]
instance.url = tunnel_info["url"] instance.url = tunnel_info["url"]
await session.commit() await session.commit()
logger.debug( logger.debug(
"Recreated tunnel for instance %s: pid=%s, url=%s", "Recreated tunnel for instance %s: container=%s, url=%s",
instance.id, instance.id,
tunnel_info["pid"], tunnel_info["container_name"],
tunnel_info["url"], tunnel_info["url"],
) )
return {"status": "healthy", "url": instance.url} return {"status": "healthy", "url": instance.url}
-332
View File
@@ -1,8 +1,6 @@
"""Docker service for managing tool instances.""" """Docker service for managing tool instances."""
import logging import logging
import os
import re
import subprocess import subprocess
import time import time
from collections import Counter from collections import Counter
@@ -384,334 +382,4 @@ def find_free_port(start: int = 10000, end: int = 20000) -> int:
raise RuntimeError(f"No free port found in range {start}-{end}") 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),
}
+2 -1
View File
@@ -13,7 +13,8 @@ from src.database import SessionLocal
from src.models.health_check import HealthCheck from src.models.health_check import HealthCheck
from src.models.tool_instance import ToolInstance from src.models.tool_instance import ToolInstance
from src.services.correlation import get_correlation_id from src.services.correlation import get_correlation_id
from src.services.docker import check_tunnel_health, get_container_status from src.services.docker import get_container_status
from src.services.tunnel import check_tunnel_health
from src.services.event_bus import InstanceEventBus, InstanceEventPayload from src.services.event_bus import InstanceEventBus, InstanceEventPayload
from src.services.notification_service import notification_service from src.services.notification_service import notification_service
+213
View File
@@ -0,0 +1,213 @@
"""Clean tunnel service using host-network cloudflared containers.
Design:
- Each tunnel runs as a Docker container on the host network.
- cloudflared connects to localhost:{published_port}, leveraging Docker's
port forwarding. No container name resolution is required.
- Tunnels are named predictably (tunnel-{instance_name}) for start/stop.
"""
import logging
import re
import subprocess
from typing import Any
logger = logging.getLogger(__name__)
TUNNEL_IMAGE = "cloudflare/cloudflared:latest"
def _tunnel_container_name(instance_name: str) -> str:
return f"tunnel-{instance_name.lower()}"
def _ensure_image() -> None:
"""Pull cloudflared image if not already present."""
result = subprocess.run(
["docker", "images", "-q", TUNNEL_IMAGE],
capture_output=True,
text=True,
)
if not result.stdout.strip():
logger.info("Pulling %s ...", TUNNEL_IMAGE)
pull = subprocess.run(
["docker", "pull", TUNNEL_IMAGE],
capture_output=True,
text=True,
)
if pull.returncode != 0:
logger.warning("Failed to pull %s: %s", TUNNEL_IMAGE, pull.stderr)
def start_tunnel(instance_name: str, published_port: int, timeout: int = 30) -> dict[str, str]:
"""Start a temporary Cloudflare tunnel for an instance.
Args:
instance_name: The tool instance name (used to derive tunnel container name).
published_port: The host port Docker forwards to the container.
timeout: Seconds to wait for the tunnel URL.
Returns:
Dict with 'url' and 'container_name'.
"""
_ensure_image()
tunnel_name = _tunnel_container_name(instance_name)
# Clean up any stale tunnel container with this name
subprocess.run(
["docker", "stop", "-t", "3", tunnel_name],
capture_output=True,
text=True,
)
subprocess.run(
["docker", "rm", tunnel_name],
capture_output=True,
text=True,
)
cmd = [
"docker", "run", "-d", "--rm",
"--network", "host",
"--name", tunnel_name,
TUNNEL_IMAGE,
"tunnel", "--url", f"http://localhost:{published_port}",
]
proc = subprocess.run(cmd, capture_output=True, text=True)
if proc.returncode != 0:
raise RuntimeError(
f"Failed to start tunnel container {tunnel_name}: {proc.stderr}"
)
# Wait for URL to appear in logs
url_pattern = re.compile(r"https://[a-z0-9-]+\.trycloudflare\.com")
start_time = __import__("time").time()
url: str | None = None
logs_result = subprocess.run(
["docker", "logs", tunnel_name],
capture_output=True,
text=True,
)
while __import__("time").time() - start_time < timeout:
match = url_pattern.search(logs_result.stdout)
if match:
url = match.group(0)
break
__import__("time").sleep(0.5)
logs_result = subprocess.run(
["docker", "logs", tunnel_name],
capture_output=True,
text=True,
)
if not url:
subprocess.run(["docker", "stop", "-t", "3", tunnel_name], capture_output=True)
raise RuntimeError(
f"Tunnel {tunnel_name} did not produce a URL within {timeout}s. "
f"cloudflared logs:\n{logs_result.stdout[-2000:]}"
)
logger.info(
"Tunnel %s started for %s on port %d%s",
tunnel_name,
instance_name,
published_port,
url,
)
return {"url": url, "container_name": tunnel_name}
def stop_tunnel(instance_name: str) -> None:
"""Stop the tunnel container for an instance."""
tunnel_name = _tunnel_container_name(instance_name)
subprocess.run(
["docker", "stop", "-t", "5", tunnel_name],
capture_output=True,
text=True,
)
logger.debug("Stopped tunnel container %s", tunnel_name)
def recreate_tunnel(instance_name: str, published_port: int) -> dict[str, str]:
"""Recreate a tunnel for an instance."""
stop_tunnel(instance_name)
return start_tunnel(instance_name, published_port)
def check_tunnel_health(url: str, timeout: int = 10) -> dict[str, Any]:
"""Check if a tunnel URL is healthy.
Returns:
Dict with 'tunnel_status', 'status_code', 'healthy', 'error'.
"""
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,
}
if status_code in (502, 503, 504):
return {
"tunnel_status": "error_response",
"status_code": status_code,
"healthy": False,
"error": f"Application returned HTTP {status_code}",
}
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 exc:
error_str = str(exc).lower()
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: {exc}",
}
return {
"tunnel_status": "unreachable",
"status_code": None,
"healthy": False,
"error": str(exc),
}
+1 -1
View File
@@ -333,7 +333,7 @@ export function SessionCard({
onClick: () => onRecreateTunnel(session), onClick: () => onRecreateTunnel(session),
}, },
] ]
: []), : []),
...(isActive && onStop ...(isActive && onStop
? [ ? [
{ {