devx/src/devx/tools/check_docker_init.py

"""Check that Docker Compose services with healthchecks have ``init: true``.

This prevents zombie process accumulation on production VMs. Without
``init: true``, Docker uses the container's PID 1 process to reap
child processes. Many images (especially those using CMD-SHELL
healthchecks with ``wget``) don't call ``wait()`` on children, causing
zombies to accumulate.

The check scans all Jinja2 docker-compose templates for services that
have a ``healthcheck:`` key but no ``init: true`` key. Since the
templates use Jinja2 syntax (not pure YAML), the check uses text-based
parsing to identify service blocks and their properties.

Usage::

    python -m devx.tools.check_docker_init
    python -m devx.tools.check_docker_init --path ansible/roles/observability/templates/docker-compose.yml.j2

Exit code 0 if all services with healthchecks have init: true, 1 otherwise.
"""

from __future__ import annotations

import re
import sys
from pathlib import Path

import click

REPO_ROOT = Path.cwd()
DEFAULT_TEMPLATES_DIR = REPO_ROOT / "ansible" / "roles"


def _find_compose_templates(base: Path) -> list[Path]:
    """Find all Jinja2 docker-compose templates under a base directory."""
    if base.is_file():
        return [base]
    if not base.is_dir():
        return []
    results: list[Path] = []
    for pattern in ("*docker-compose*", "*compose*"):
        results.extend(base.rglob(f"{pattern}.yml.j2"))
        results.extend(base.rglob(f"{pattern}.yaml.j2"))
    # Also check exporters-compose
    results.extend(base.rglob("exporters-compose*.j2"))
    # Deduplicate while preserving order
    seen: set[Path] = set()
    unique: list[Path] = []
    for p in sorted(results):
        if p not in seen:
            seen.add(p)
            unique.append(p)
    return unique


def _parse_services(content: str) -> dict[str, list[str]]:
    """Parse service blocks from a docker-compose Jinja2 template.

    Returns a mapping of service_name → list of lines in that service block.
    """
    lines = content.splitlines()
    in_services = False
    services: dict[str, list[str]] = {}
    current_svc: str | None = None
    current_lines: list[str] = []

    for line in lines:
        if line.startswith("services:"):
            in_services = True
            continue
        if not in_services:
            continue
        # Top-level keys (networks:, volumes:) end the services section
        if re.match(r"^(networks|volumes):\s*$", line):
            if current_svc is not None:
                services[current_svc] = current_lines
            current_svc = None
            in_services = False
            continue
        # Service definition: exactly 2-space indent, ends with :
        # Service names can contain Jinja2 variables like {{ app_name }}
        # or {{ app_name }}-db. Match: 2-space indent + non-whitespace
        # chars (including {{ }}, -, _, .) + optional spaces inside {{ }} + :
        m = re.match(r"^  (\{\{.*?\}\}[a-zA-Z0-9_-]*|[a-zA-Z0-9_().-]+):\s*$", line)
        if m:
            if current_svc is not None:
                services[current_svc] = current_lines
            current_svc = m.group(1)
            current_lines = []
        elif current_svc is not None:
            current_lines.append(line)

    if current_svc is not None:
        services[current_svc] = current_lines

    return services


def _check_template(filepath: Path, repo_root: Path) -> list[str]:
    """Check a single docker-compose template for missing init: true.

    Returns a list of error messages (empty if all OK).
    """
    errors: list[str] = []
    content = filepath.read_text(encoding="utf-8")

    if "services:" not in content:
        return errors

    services = _parse_services(content)

    for svc_name, svc_lines in services.items():
        svc_text = "\n".join(svc_lines)
        has_init = "init: true" in svc_text
        has_healthcheck = "healthcheck:" in svc_text
        # Skip services that are conditionally included (Jinja2 if blocks)
        # but still check them — the healthcheck is inside the conditional
        if has_healthcheck and not has_init:
            try:
                display_path = filepath.relative_to(repo_root)
            except ValueError:
                display_path = filepath
            errors.append(
                f"{display_path}: service '{svc_name}' has a healthcheck "
                f"but no 'init: true'. Without init: true, CMD-SHELL "
                f"healthchecks (wget, pgrep) spawn children that become "
                f"zombies when PID 1 doesn't reap them. Add 'init: true' "
                f"to enable Docker's built-in tini as PID 1."
            )

    return errors


@click.command()
@click.option(
    "--path",
    type=click.Path(exists=True, path_type=Path),
    help="Check a specific file or directory (default: ansible/roles/).",
)
@click.option(
    "--templates-dir",
    type=click.Path(exists=True, path_type=Path),
    default=None,
    help="Override the default templates directory (default: ansible/roles/).",
)
def main(path: Path | None, templates_dir: Path | None) -> None:
    """Check that Docker Compose services with healthchecks have init: true."""
    tdir = templates_dir or DEFAULT_TEMPLATES_DIR
    files = _find_compose_templates(path) if path else _find_compose_templates(tdir)

    all_errors: list[str] = []
    for f in files:
        errors = _check_template(f, tdir)
        all_errors.extend(errors)

    if all_errors:
        click.echo("[check-docker-init] FAIL: services with healthchecks missing init: true:")
        for err in all_errors:
            click.echo(f"  - {err}")
        sys.exit(1)
    else:
        click.echo("[check-docker-init] OK: all services with healthchecks have init: true.")


if __name__ == "__main__":  # pragma: no cover
    main()