Tussilago/control_plane/runtime_plans.py
2026-04-27 20:43:26 +02:00

366 lines
11 KiB
Python

from __future__ import annotations
from dataclasses import dataclass
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from collections.abc import Sequence
from pathlib import Path
@dataclass(frozen=True, slots=True)
class PostgresContainerConfig:
"""Input required to build a Podman command for a tenant PostgreSQL service."""
container_name: str
network_name: str
hostname: str
username: str
database_name: str
data_directory: Path
password_file: Path
pod_name: str | None = None
image_reference: str = "docker.io/library/postgres:17-alpine"
memory_limit_mib: int = 512
cpu_limit: float = 1.0
@dataclass(frozen=True, slots=True)
class RedisContainerConfig:
"""Input required to build a Podman command for a tenant Redis service."""
container_name: str
network_name: str
hostname: str
data_directory: Path
password_file: Path
pod_name: str | None = None
image_reference: str = "docker.io/library/redis:7.4-alpine"
memory_limit_mib: int = 256
cpu_limit: float = 0.5
@dataclass(frozen=True, slots=True)
class DjangoApplicationLaunchConfig:
"""Input required to build a uv-driven Gunicorn command for a Django app."""
wsgi_module: str
port: int = 8000
bind_host: str = "127.0.0.1"
workers: int = 2
python_executable: Path | None = None
uv_project_path: Path | None = None
@dataclass(frozen=True, slots=True)
class DjangoContainerImageBuildConfig:
"""Input required to build the reusable local Django test image."""
image_reference: str
containerfile_path: Path
context_directory: Path
@dataclass(frozen=True, slots=True)
class DjangoContainerRuntimeConfig:
"""Input required to run a local Django test container with Podman."""
container_name: str
network_name: str
hostname: str
image_reference: str
application_directory: Path
pod_name: str | None = None
host_port: int | None = None
guest_port: int = 8000
working_directory: str = "/srv/test-app"
environment: tuple[tuple[str, str], ...] = ()
secret_mounts: tuple[tuple[Path, str], ...] = ()
labels: tuple[tuple[str, str], ...] = ()
memory_limit_mib: int = 256
cpu_limit: float = 1.0
def build_postgres_container_command(
config: PostgresContainerConfig,
) -> tuple[str, ...]:
"""Build a hardened Podman command for a deployment-scoped PostgreSQL service.
Returns:
Tuple of Podman arguments ready for subprocess execution.
"""
command = [
"podman",
"run",
"--detach",
"--replace",
"--name",
config.container_name,
]
if config.pod_name is None:
command.extend(("--network", config.network_name, "--hostname", config.hostname))
else:
command.extend(("--pod", config.pod_name))
command.extend(
[
"--cap-drop=all",
"--cap-add=CHOWN",
"--cap-add=FOWNER",
"--cap-add=SETUID",
"--cap-add=SETGID",
"--cap-add=DAC_OVERRIDE",
"--security-opt=no-new-privileges",
"--pids-limit=256",
"--memory",
f"{config.memory_limit_mib}m",
"--cpus",
str(config.cpu_limit),
"--read-only",
"--tmpfs",
"/tmp:rw,nosuid,nodev,noexec,size=64m", # noqa: S108
"--tmpfs",
"/var/run/postgresql:rw,nosuid,nodev,noexec,size=16m",
"--volume",
f"{config.data_directory}:/var/lib/postgresql/data:Z,rw",
"--volume",
f"{config.password_file}:/run/secrets/postgres-password:Z,ro",
"--env",
f"POSTGRES_USER={config.username}",
"--env",
f"POSTGRES_DB={config.database_name}",
"--env",
"POSTGRES_PASSWORD_FILE=/run/secrets/postgres-password",
"--health-cmd",
f"pg_isready -U {config.username} -d {config.database_name}",
"--health-interval",
"10s",
"--health-retries",
"5",
config.image_reference,
"postgres",
"-c",
"listen_addresses=*",
"-c",
"password_encryption=scram-sha-256",
],
)
return tuple(command)
def build_redis_container_command(config: RedisContainerConfig) -> tuple[str, ...]:
"""Build a hardened Podman command for a deployment-scoped Redis service.
Returns:
Tuple of Podman arguments ready for subprocess execution.
"""
command = [
"podman",
"run",
"--detach",
"--replace",
"--name",
config.container_name,
]
if config.pod_name is None:
command.extend(("--network", config.network_name, "--hostname", config.hostname))
else:
command.extend(("--pod", config.pod_name))
command.extend(
[
"--cap-drop=all",
"--security-opt=no-new-privileges",
"--pids-limit=128",
"--memory",
f"{config.memory_limit_mib}m",
"--cpus",
str(config.cpu_limit),
"--read-only",
"--tmpfs",
"/tmp:rw,nosuid,nodev,noexec,size=32m", # noqa: S108
"--volume",
f"{config.data_directory}:/data:Z,rw",
"--volume",
f"{config.password_file}:/run/secrets/redis-password:Z,ro",
"--health-cmd",
"sh -eu -c 'redis-cli --no-auth-warning -a \"$(cat /run/secrets/redis-password)\" ping'",
"--health-interval",
"10s",
"--health-retries",
"5",
config.image_reference,
"sh",
"-eu",
"-c",
'redis_password=$(cat /run/secrets/redis-password) && exec redis-server --appendonly yes --protected-mode yes --requirepass "${redis_password}"',
],
)
return tuple(command)
def build_django_server_command(
config: DjangoApplicationLaunchConfig,
) -> tuple[str, ...]:
"""Build a uv-driven Gunicorn command for a hosted Django deployment.
Returns:
Tuple of command arguments ready for subprocess execution.
Raises:
ValueError: If both direct-python and uv-project execution modes are requested.
"""
if config.python_executable is not None and config.uv_project_path is not None:
msg = "python_executable and uv_project_path are mutually exclusive"
raise ValueError(msg)
if config.python_executable is not None:
command = [str(config.python_executable), "-m", "gunicorn"]
else:
command = ["uv", "run"]
if config.uv_project_path is not None:
command.extend(["--project", str(config.uv_project_path)])
command.append("gunicorn")
command.extend(
[
"--bind",
f"{config.bind_host}:{config.port}",
"--workers",
str(config.workers),
"--access-logfile",
"-",
"--error-logfile",
"-",
"--capture-output",
"--graceful-timeout",
"30",
"--timeout",
"60",
config.wsgi_module,
],
)
return tuple(command)
def build_django_container_image_command(
config: DjangoContainerImageBuildConfig,
) -> tuple[str, ...]:
"""Build a Podman image command for the reusable Django test runtime.
Returns:
Tuple of Podman arguments ready for subprocess execution.
"""
return (
"podman",
"build",
"--pull=missing",
"--tag",
config.image_reference,
"--file",
str(config.containerfile_path),
str(config.context_directory),
)
def build_django_container_run_command(
config: DjangoContainerRuntimeConfig,
*,
command: Sequence[str],
detach: bool,
remove: bool = False,
) -> tuple[str, ...]:
"""Build a hardened Podman command for a local Django test container.
Returns:
Tuple of Podman arguments ready for subprocess execution.
Raises:
ValueError: If the command sequence is empty.
"""
if not command:
msg = "command must not be empty"
raise ValueError(msg)
podman_command = ["podman", "run"]
if detach:
podman_command.extend(("--detach", "--replace"))
if remove:
podman_command.append("--rm")
podman_command.extend(
[
"--name",
config.container_name,
],
)
if config.pod_name is None:
podman_command.extend(("--network", config.network_name, "--hostname", config.hostname))
else:
podman_command.extend(("--pod", config.pod_name))
podman_command.extend(
[
"--workdir",
config.working_directory,
"--cap-drop=all",
"--security-opt=no-new-privileges",
"--pids-limit=256",
"--memory",
f"{config.memory_limit_mib}m",
"--cpus",
str(config.cpu_limit),
"--read-only",
"--tmpfs",
"/tmp:rw,nosuid,nodev,noexec,size=64m", # noqa: S108
"--tmpfs",
"/run:rw,nosuid,nodev,noexec,size=16m",
"--volume",
f"{config.application_directory}:{config.working_directory}:Z,ro",
],
)
if config.host_port is not None and config.pod_name is None:
podman_command.extend(("--publish", f"127.0.0.1:{config.host_port}:{config.guest_port}"))
for mount_source, mount_target in config.secret_mounts:
podman_command.extend(("--volume", f"{mount_source}:{mount_target}:Z,ro"))
for key, value in config.environment:
podman_command.extend(("--env", f"{key}={value}"))
for key, value in config.labels:
podman_command.extend(("--label", f"{key}={value}"))
podman_command.append(config.image_reference)
podman_command.extend(command)
return tuple(podman_command)
def build_django_migrate_command(
uv_project_path: Path | None = None,
*,
python_executable: Path | None = None,
) -> tuple[str, ...]:
"""Build a uv-driven migration command for a hosted Django deployment.
Returns:
Tuple of command arguments ready for subprocess execution.
Raises:
ValueError: If direct-python and uv-project execution modes are mixed.
"""
if python_executable is not None and uv_project_path is not None:
msg = "python_executable and uv_project_path are mutually exclusive"
raise ValueError(msg)
if python_executable is not None:
return (str(python_executable), "manage.py", "migrate", "--noinput")
command = ["uv", "run"]
if uv_project_path is not None:
command.extend(["--project", str(uv_project_path)])
command.extend(["python", "manage.py", "migrate", "--noinput"])
return tuple(command)