vergabe-teilnahme/scripts/synthetic_pilot.py

#!/usr/bin/env python3
"""Provisional loopback HTTP workload using Gunicorn and disposable PostgreSQL."""

import argparse
import atexit
import concurrent.futures
import hashlib
import http.cookiejar
import json
import math
import os
import platform
import socket
import subprocess
import sys
import tempfile
import threading
import time
import urllib.error
import urllib.parse
import urllib.request
import uuid
from datetime import UTC, datetime
from pathlib import Path


def summarize(rows):
    durations = sorted(row["seconds"] for row in rows)
    return {
        "requests": len(rows),
        "errors": sum(not r["ok"] for r in rows),
        "p95_seconds": durations[math.ceil(len(durations) * 0.95) - 1] if durations else None,
    }


def validate(config):
    for key in (
        "users",
        "iterations",
        "tenders",
        "document_bytes",
        "workers",
        "timeout_seconds",
    ):
        if type(config[key]) is not int or config[key] < 1:
            raise ValueError(f"{key} must be a positive integer")
    for key in (
        "think_seconds",
        "ordinary_p95_seconds",
        "duration_seconds",
        "idle_seconds",
    ):
        if not isinstance(config[key], (int, float)) or not math.isfinite(config[key]):
            raise ValueError(f"{key} must be finite")
    if config["think_seconds"] < 0 or config["ordinary_p95_seconds"] <= 0:
        raise ValueError("invalid timing parameter")
    if config["duration_seconds"] <= 0 or config["idle_seconds"] < 0:
        raise ValueError("duration must be positive and idle nonnegative")
    if config["document_bytes"] < 8:
        raise ValueError("document_bytes must be at least 8")
    return config


def run(config, root):
    # Never consume a deployment database, media tree, session or credential.
    os.environ["DATABASE_URL"] = os.environ["SYNTHETIC_DATABASE_URL"]
    os.environ["SECRET_KEY"] = "synthetic-disposable-not-a-deployment-key"
    os.environ["DJANGO_SETTINGS_MODULE"] = "synthetic_settings"
    (root / "synthetic_settings.py").write_text(
        "from vergabe_teilnahme.settings.base import *\n"
        'DEBUG = False\nALLOWED_HOSTS = ["127.0.0.1"]\nNETKINGDOM_ENABLED = False\n'
        f"MEDIA_ROOT = {str(root / 'media')!r}\n"
        f"ISSUE_FACADE_LOCAL_DB = {str(root / 'issues.db')!r}\n"
        'ISSUE_FACADE_GITEA = None\nFORCE_SCRIPT_NAME = None\nMEDIA_URL = "/media/"\n'
        'EMAIL_BACKEND = "django.core.mail.backends.locmem.EmailBackend"\n'
        "SESSION_COOKIE_SECURE = False\nCSRF_COOKIE_SECURE = False\n"
    )
    sys.path.insert(0, str(root))
    import django

    django.setup()
    from django.core.management import call_command
    from django.db import connections

    from vergabe_teilnahme.apps.accounts.models import Mitarbeiter
    from vergabe_teilnahme.apps.ausschreibungen.models import Ausschreibung
    from vergabe_teilnahme.apps.dokumente.models import Dokument

    call_command("migrate", verbosity=0)
    for i in range(config["tenders"]):
        Ausschreibung.objects.create(titel=f"Synthetic tender {i}", ausschreiber="Synthetic")
    users = []
    password = uuid.uuid4().hex
    for i in range(config["users"]):
        user = Mitarbeiter.objects.create_user(username=f"synthetic-{i}", password=password)
        users.append(user.username)
    with socket.socket() as sock:
        sock.bind(("127.0.0.1", 0))
        port = sock.getsockname()[1]
    base_url = f"http://127.0.0.1:{port}"
    env = dict(os.environ, PYTHONPATH=str(root) + os.pathsep + str(Path.cwd()))
    server_log = (root / "server.log").open("w")
    server = subprocess.Popen(
        [
            sys.executable,
            "-m",
            "gunicorn",
            "vergabe_teilnahme.wsgi:application",
            "--bind",
            f"127.0.0.1:{port}",
            "--workers",
            str(config["workers"]),
            "--timeout",
            "30",
        ],
        env=env,
        stdout=server_log,
        stderr=server_log,
    )
    atexit.register(lambda: server.terminate() if server.poll() is None else None)
    for _ in range(100):
        try:
            with urllib.request.urlopen(base_url + "/health/", timeout=1) as response:
                if response.status == 200:
                    break
        except OSError:
            if server.poll() is not None:
                raise RuntimeError("Gunicorn startup failed; see isolated server log")
            time.sleep(0.1)
    else:
        server.terminate()
        raise RuntimeError("Gunicorn readiness timeout")
    samples = []
    db_samples = []
    sampling_done = threading.Event()
    phase = {"name": "idle"}

    def sample():
        ticks_per_second = os.sysconf("SC_CLK_TCK")
        while not sampling_done.is_set():
            pids = [server.pid]
            try:
                pids += [
                    int(x)
                    for x in Path(f"/proc/{server.pid}/task/{server.pid}/children")
                    .read_text()
                    .split()
                ]
                cpu, rss = 0, 0
                for pid in pids:
                    fields = Path(f"/proc/{pid}/stat").read_text().split(") ", 1)[1].split()
                    cpu += (int(fields[11]) + int(fields[12])) / ticks_per_second
                    rss += int(fields[21]) * os.sysconf("SC_PAGE_SIZE")
                samples.append(
                    {
                        "at": datetime.now(UTC).isoformat(),
                        "monotonic_seconds": time.monotonic(),
                        "phase": phase["name"],
                        "cpu_seconds": cpu,
                        "rss_bytes": rss,
                        "pids": pids,
                    }
                )
            except (OSError, ValueError):
                samples.append({"at": datetime.now(UTC).isoformat(), "sample_error": True})
            sampling_done.wait(0.5)

    def sample_db():
        while not sampling_done.is_set():
            value = subprocess.check_output(
                [
                    "docker",
                    "stats",
                    "--no-stream",
                    "--format",
                    "{{json .}}",
                    os.environ["SYNTHETIC_CONTAINER"],
                ],
                text=True,
            )
            db_samples.append(
                {
                    "at": datetime.now(UTC).isoformat(),
                    "phase": phase["name"],
                    "docker_stats": json.loads(value),
                }
            )
            sampling_done.wait(1)

    sampler = threading.Thread(target=sample, daemon=True)
    db_sampler = threading.Thread(target=sample_db, daemon=True)
    sampler.start()
    db_sampler.start()
    time.sleep(config["idle_seconds"])
    phase["name"] = "ordinary"
    deadline = time.monotonic() + config["duration_seconds"]
    completed = [0] * config["users"]
    barrier = threading.Barrier(config["users"])

    def worker(i):
        jar = http.cookiejar.CookieJar()

        class NoRedirect(urllib.request.HTTPRedirectHandler):
            def redirect_request(self, *args, **kwargs):
                return None

        opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(jar), NoRedirect())
        rows = []

        def request(operation, method, path, data=None, expected=200, transfer=False):
            started = time.perf_counter()
            try:
                headers = {}
                body = None
                if method == "post":
                    token = next((c.value for c in jar if c.name == "csrftoken"), "")
                    fields = dict(data or {}, csrfmiddlewaretoken=token)
                    if transfer:
                        boundary = "synthetic-" + uuid.uuid4().hex
                        body = b""
                        for name, value in fields.items():
                            body += (f'--{boundary}\r\nContent-Disposition: form-data; '
                                     f'name="{name}"\r\n\r\n{value}\r\n').encode()
                        body += (
                            (f'--{boundary}\r\nContent-Disposition: form-data; '
                                 'name="datei"; filename="synthetic.pdf"\r\n'
                                 'Content-Type: application/pdf\r\n\r\n').encode()
                            + b"%PDF-1.4"
                            + b"x" * (config["document_bytes"] - 8)
                            + f"\r\n--{boundary}--\r\n".encode()
                        )
                        headers["Content-Type"] = f"multipart/form-data; boundary={boundary}"
                    else:
                        body = urllib.parse.urlencode(fields).encode()
                    headers["Referer"] = base_url + path
                req = urllib.request.Request(
                    base_url + path, data=body, headers=headers, method=method.upper()
                )
                try:
                    response = opener.open(req, timeout=config["timeout_seconds"])
                except urllib.error.HTTPError as exc:
                    response = exc
                received = len(response.read())
                status = response.status
                ok = status == expected
                response.close()
            except Exception:
                status, ok, received = None, False, 0
            rows.append(
                {
                    "user": i,
                    "operation": operation,
                    "transfer": transfer,
                    "seconds": time.perf_counter() - started,
                    "status": status,
                    "ok": ok,
                    "response_bytes": received,
                }
            )
            return ok

        request("login-form", "get", "/accounts/login/")
        barrier.wait(timeout=30)
        request(
            "login",
            "post",
            "/accounts/login/",
            {"username": users[i], "password": password},
            302,
        )
        for n in range(config["iterations"]):
            if time.monotonic() >= deadline:
                break
            request("list", "get", "/ausschreibungen/")
            request("search", "get", "/ausschreibungen/?q=Synthetic")
            title = f"Workload-{i}-{n}"
            created = request(
                "tender-create",
                "post",
                "/ausschreibungen/neu/",
                {"titel": title, "ausschreiber": "Synthetic"},
                302,
            )
            tender = Ausschreibung.objects.filter(titel=title).first() if created else None
            if tender:
                base = f"/ausschreibungen/{tender.pk}/"
                request("detail", "get", base)
                request(
                    "lot-create",
                    "post",
                    base + "lose/neu/",
                    {"losnummer": "1", "lostitel": "Synthetic lot"},
                    302,
                )
                request(
                    "task-create",
                    "post",
                    base + "aufgaben/neu/",
                    {"titel": "Synthetic task", "typ": "fachlich", "prioritaet": 2},
                    302,
                )
                request(
                    "document-upload",
                    "post",
                    base + "dokumente/hochladen/",
                    {"kategorie": "intern", "version": "1.0"},
                    302,
                    True,
                )
                document = Dokument.objects.filter(ausschreibung=tender).first()
                if document:
                    request("document-download", "get", document.datei.url, transfer=True)
            completed[i] += 1
            time.sleep(config["think_seconds"])
        connections.close_all()
        return rows

    started = datetime.now(UTC).isoformat()
    before = time.perf_counter()
    with concurrent.futures.ThreadPoolExecutor(max_workers=config["users"]) as pool:
        rows = [r for group in pool.map(worker, range(config["users"])) for r in group]
    ordinary = summarize([r for r in rows if not r["transfer"]])
    transfers = summarize([r for r in rows if r["transfer"]])
    complete = all(completed) and len(rows) == sum(completed) * 8 + config["users"] * 2
    result = {
        "schema_version": 1,
        "started_at": started,
        "finished_at": datetime.now(UTC).isoformat(),
        "duration_seconds": time.perf_counter() - before,
        "config": config,
        "environment": {
            "python": platform.python_version(),
            "django": django.get_version(),
            "platform": platform.platform(),
            "database": "disposable PostgreSQL 16 container",
            "transport": "loopback HTTP Gunicorn",
            "workers": config["workers"],
        },
        "ordinary": ordinary,
        "document_transfer": transfers,
        "complete": complete,
        "provisional_pass": complete
        and ordinary["errors"] == 0
        and transfers["errors"] == 0
        and ordinary["p95_seconds"] <= config["ordinary_p95_seconds"],
        "limitations": [
            "No browser, ingress, TLS, SSO or application container quota",
            "No CPU/memory allocation recommendation or human acceptance",
            "No timed startup/restart or distinct burst; configured idle and ordinary phases only",
        ],
        "operations": rows,
        "runtime_samples": samples,
        "database_samples": db_samples,
        "completed_iterations": completed,
    }
    sampling_done.set()
    sampler.join()
    db_sampler.join()
    server.terminate()
    server.wait(timeout=10)
    server_log.close()
    connections.close_all()
    return result


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--config", type=Path, required=True)
    parser.add_argument("--output", type=Path, required=True)
    args = parser.parse_args()
    config = validate(json.loads(args.config.read_text()))
    sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
    name = "vergabe-synthetic-" + uuid.uuid4().hex[:12]
    # Cached image only, loopback random host port, no existing volume or database.
    subprocess.run(
        [
            "docker",
            "run",
            "--detach",
            "--pull=never",
            "--name",
            name,
            "--publish",
            "127.0.0.1::5432",
            "--env",
            "POSTGRES_DB=synthetic",
            "--env",
            "POSTGRES_USER=synthetic",
            "--env",
            "POSTGRES_PASSWORD=synthetic-local-only",
            "postgres:16-alpine",
        ],
        check=True,
        stdout=subprocess.DEVNULL,
    )
    try:
        for _ in range(100):
            ready = subprocess.run(
                ["docker", "exec", name, "pg_isready", "-U", "synthetic"],
                stdout=subprocess.DEVNULL,
                stderr=subprocess.DEVNULL,
            )
            if ready.returncode == 0:
                break
            time.sleep(0.1)
        else:
            raise RuntimeError("isolated PostgreSQL startup failed")
        port = (
            subprocess.check_output(["docker", "port", name, "5432/tcp"], text=True)
            .strip()
            .split(":")[-1]
        )
        os.environ["SYNTHETIC_CONTAINER"] = name
        os.environ["SYNTHETIC_DATABASE_URL"] = (
            f"postgres://synthetic:synthetic-local-only@127.0.0.1:{port}/synthetic"
        )
        with tempfile.TemporaryDirectory(prefix="vergabe-synthetic-") as directory:
            result = run(config, Path(directory))
        result["postgres_image_id"] = subprocess.check_output(
            ["docker", "inspect", "--format", "{{.Image}}", name], text=True
        ).strip()
        result["postgres_final_stats"] = json.loads(
            subprocess.check_output(
                ["docker", "stats", "--no-stream", "--format", "{{json .}}", name],
                text=True,
            )
        )
    finally:
        subprocess.run(
            ["docker", "rm", "--force", "--volumes", name],
            check=True,
            stdout=subprocess.DEVNULL,
        )
    result["source_revision"] = subprocess.check_output(
        ["git", "rev-parse", "HEAD"], text=True
    ).strip()
    result["driver_sha256"] = hashlib.sha256(Path(__file__).read_bytes()).hexdigest()
    app_files = sorted(Path("vergabe_teilnahme").rglob("*.py")) + [
        Path("uv.lock"),
        Path("pyproject.toml"),
    ]
    result["app_source_sha256"] = hashlib.sha256(
        b"".join(str(p).encode() + b"\0" + p.read_bytes() for p in app_files)
    ).hexdigest()
    result["host"] = {
        "logical_cpus": os.cpu_count(),
        "affinity_cpus": len(os.sched_getaffinity(0)),
        "cpu_model": next(
            (
                x.split(":", 1)[1].strip()
                for x in Path("/proc/cpuinfo").read_text().splitlines()
                if x.startswith("model name")
            ),
            "unknown",
        ),
        "memory_total": next(
            x for x in Path("/proc/meminfo").read_text().splitlines() if x.startswith("MemTotal:")
        ),
        "cgroup_cpu_max": Path("/sys/fs/cgroup/cpu.max").read_text().strip()
        if Path("/sys/fs/cgroup/cpu.max").exists()
        else "unavailable",
        "app_cpu_quota": "not imposed by harness",
    }
    result["config_sha256"] = hashlib.sha256(args.config.read_bytes()).hexdigest()
    args.output.write_text(json.dumps(result, indent=2) + "\n")
    print(
        json.dumps(
            {
                k: result[k]
                for k in (
                    "ordinary",
                    "document_transfer",
                    "complete",
                    "provisional_pass",
                )
            }
        )
    )
    raise SystemExit(0 if result["provisional_pass"] else 1)