Files
headquarter/apps/api/app/services/spawn.py
T
alex 51d93d9dc6 feat(FN-009): implement config and secrets management with runtime injection
- Add RuntimeInjectionService for scope-based config/secret resolution
- Mount configs as JSON files at /app/config/ with 0400 permissions
- Inject secrets as environment variables with uppercase keys
- Implement scope hierarchy: instance > project > user > global
- Create ConfigListPage and SecretListPage frontend components
- Mask secret values in API responses (never expose decrypted)
- Validate secrets exist before spawning containers
- Add comprehensive tests for runtime injection service
- Update documentation with config/secrets workflow
2026-05-15 16:44:26 +02:00

346 lines
11 KiB
Python

from __future__ import annotations
import json
import logging
import subprocess
from pathlib import Path
from typing import Any
from app.config import settings
from app.services.traefik import TraefikLabelGenerator
from app.tools.models import ToolManifest
logger = logging.getLogger(__name__)
class SpawnError(Exception):
pass
class SpawnService:
def __init__(
self,
compose_dir: Path | None = None,
network_name: str = "tools",
) -> None:
self.compose_dir = compose_dir or Path("/tmp/headquarter-compose")
self.network_name = network_name
self.compose_dir.mkdir(parents=True, exist_ok=True)
def _generate_compose_service(
self,
instance_id: str,
manifest: ToolManifest,
subdomain: str,
traefik_labels: dict[str, str],
project_slug: str,
user_slug: str,
workspace_path: Path | None = None,
config_path: Path | None = None,
ssh_key_path: Path | None = None,
config_mounts: list[str] | None = None,
secret_env_vars: dict[str, str] | None = None,
) -> dict[str, Any]:
service_name = f"tool-{instance_id[:8]}"
service: dict[str, Any] = {
"image": manifest.image,
"container_name": service_name,
"restart": "unless-stopped",
"labels": traefik_labels,
"networks": [self.network_name],
}
if manifest.runtime_command:
service["command"] = manifest.runtime_command
if manifest.runtime_entrypoint:
service["entrypoint"] = manifest.runtime_entrypoint
if manifest.runtime_user:
service["user"] = manifest.runtime_user
if manifest.runtime_working_dir:
service["working_dir"] = manifest.runtime_working_dir
ports = manifest.ports
if ports:
service["ports"] = [
f"{port.container_port}:{port.container_port}"
for port in ports
]
env = dict(manifest.env)
env.update({
"PROJECT_SLUG": project_slug,
"USER_SLUG": user_slug,
})
service["environment"] = env
volumes: list[str] = []
default_workspace = f"/data/workspaces/{user_slug}/{project_slug}"
for mount in manifest.workspace_mounts:
source = mount.source_pattern.format(
project_repo=str(workspace_path) if workspace_path else default_workspace,
)
ro_suffix = ":ro" if mount.read_only else ""
volumes.append(f"{source}:{mount.target}{ro_suffix}")
default_config = f"/data/configs/{user_slug}"
for mount in manifest.config_mounts:
source = mount.source_pattern.format(
user_config=str(config_path) if config_path else default_config,
)
ro_suffix = ":ro" if mount.read_only else ""
volumes.append(f"{source}:{mount.target}{ro_suffix}")
if ssh_key_path and ssh_key_path.exists():
volumes.append(f"{ssh_key_path}:/home/coder/.ssh:ro")
if config_mounts:
volumes.extend(config_mounts)
if volumes:
service["volumes"] = volumes
if secret_env_vars:
service["environment"].update(secret_env_vars)
if manifest.health_check:
hc = manifest.health_check
healthcheck: dict[str, Any] = {
"interval": f"{hc.interval_seconds}s",
"timeout": f"{hc.timeout_seconds}s",
"retries": hc.retries,
"start_period": f"{hc.start_period_seconds}s",
}
if hc.type == "http":
healthcheck["test"] = [
"CMD",
"curl",
"-f",
f"http://localhost:{hc.port}{hc.path}",
]
elif hc.type == "tcp":
healthcheck["test"] = [
"CMD",
"nc",
"-z",
"localhost",
str(hc.port),
]
elif hc.type == "command":
healthcheck["test"] = ["CMD"] + (hc.command or [])
service["healthcheck"] = healthcheck
if manifest.resource_limits:
rl = manifest.resource_limits
deploy: dict[str, Any] = {"resources": {"limits": {}}}
if rl.cpus:
deploy["resources"]["limits"]["cpus"] = str(rl.cpus)
if rl.memory_mb:
deploy["resources"]["limits"]["memory"] = f"{rl.memory_mb}M"
if rl.memory_swap_mb is not None and rl.memory_swap_mb >= 0:
deploy["resources"]["limits"]["swap"] = f"{rl.memory_swap_mb}M"
service["deploy"] = deploy
return service
def _write_compose_file(
self,
instance_id: str,
service: dict[str, Any],
) -> Path:
compose_path = self.compose_dir / f"{instance_id}.yml"
compose = {
"version": "3.8",
"services": {f"tool-{instance_id[:8]}": service},
"networks": {
self.network_name: {
"external": True,
},
},
}
compose_path.write_text(json.dumps(compose, indent=2))
return compose_path
def spawn(
self,
instance_id: str,
manifest: ToolManifest,
project_slug: str,
user_slug: str,
workspace_path: Path | None = None,
config_path: Path | None = None,
ssh_key_path: Path | None = None,
config_mounts: list[str] | None = None,
secret_env_vars: dict[str, str] | None = None,
) -> dict[str, Any]:
label_gen = TraefikLabelGenerator(domain=settings.root_domain)
primary_port = next(
(p.container_port for p in manifest.ports if p.primary),
manifest.ports[0].container_port if manifest.ports else 8080,
)
subdomain = label_gen.generate_subdomain(
tool_key=manifest.id,
project_slug=project_slug,
user_slug=user_slug,
)
traefik_labels = label_gen.generate_labels(
instance_id=instance_id,
tool_key=manifest.id,
project_slug=project_slug,
user_slug=user_slug,
container_port=primary_port,
network_name=self.network_name,
)
service = self._generate_compose_service(
instance_id=instance_id,
manifest=manifest,
subdomain=subdomain,
traefik_labels=traefik_labels,
project_slug=project_slug,
user_slug=user_slug,
workspace_path=workspace_path,
config_path=config_path,
ssh_key_path=ssh_key_path,
config_mounts=config_mounts,
secret_env_vars=secret_env_vars,
)
compose_path = self._write_compose_file(instance_id, service)
try:
result = subprocess.run(
[
"docker", "compose",
"-f", str(compose_path),
"-p", f"hq-tool-{instance_id[:8]}",
"up", "-d", "--remove-orphans",
],
capture_output=True,
text=True,
check=True,
)
logger.info("Spawned container for instance %s: %s", instance_id, result.stdout)
except subprocess.CalledProcessError as e:
logger.error("Failed to spawn container for instance %s: %s", instance_id, e.stderr)
raise SpawnError(f"Failed to spawn container: {e.stderr}") from e
container_id = self._get_container_id(instance_id)
return {
"container_id": container_id,
"subdomain": subdomain,
"traefik_labels": traefik_labels,
"compose_path": str(compose_path),
}
def stop(self, instance_id: str) -> None:
compose_path = self.compose_dir / f"{instance_id}.yml"
if not compose_path.exists():
logger.warning("Compose file not found for instance %s", instance_id)
return
try:
subprocess.run(
[
"docker", "compose",
"-f", str(compose_path),
"-p", f"hq-tool-{instance_id[:8]}",
"down",
],
capture_output=True,
text=True,
check=True,
)
logger.info("Stopped container for instance %s", instance_id)
except subprocess.CalledProcessError as e:
logger.error("Failed to stop container for instance %s: %s", instance_id, e.stderr)
raise SpawnError(f"Failed to stop container: {e.stderr}") from e
def get_status(self, instance_id: str) -> str:
container_id = self._get_container_id(instance_id)
if not container_id:
return "stopped"
try:
result = subprocess.run(
[
"docker", "inspect",
"-f", "{{.State.Status}}",
container_id,
],
capture_output=True,
text=True,
check=True,
)
status = result.stdout.strip()
if status == "running":
health = self._get_health_status(container_id)
if health == "healthy":
return "running"
elif health == "unhealthy":
return "error"
else:
return "creating"
elif status in ("exited", "dead"):
return "stopped"
elif status == "paused":
return "stopped"
else:
return "creating"
except subprocess.CalledProcessError:
return "stopped"
def _get_container_id(self, instance_id: str) -> str | None:
service_name = f"tool-{instance_id[:8]}"
project_name = f"hq-tool-{instance_id[:8]}"
try:
result = subprocess.run(
[
"docker", "compose",
"-p", project_name,
"ps", "-q", service_name,
],
capture_output=True,
text=True,
check=True,
)
container_id = result.stdout.strip()
return container_id if container_id else None
except subprocess.CalledProcessError:
return None
def _get_health_status(self, container_id: str) -> str | None:
try:
result = subprocess.run(
[
"docker", "inspect",
"-f", "{{.State.Health.Status}}",
container_id,
],
capture_output=True,
text=True,
check=True,
)
status = result.stdout.strip()
return status if status else None
except subprocess.CalledProcessError:
return None