diff --git a/benchmark/flask_diagnostic.py b/benchmark/flask_diagnostic.py new file mode 100644 index 0000000..6abd045 --- /dev/null +++ b/benchmark/flask_diagnostic.py @@ -0,0 +1,412 @@ +"""Counterbalanced Flask hypothesis probes; never an official performance result.""" + +import argparse +import copy +import hashlib +import json +import re +import signal +import sys +import uuid +from pathlib import Path + +from .contract_runner import protect_cleanup +from .contract_test import ContractFailure, load_cases, run_contract +from .environment import DockerEnvironment +from .healthcheck import EXTERNAL_READINESS +from .process import ROOT, execute +from .registry import implementation +from .results import BenchmarkFailure, atomic_json, require, select_run, strict_json, validate_run +from .run import now + +CACHE_ROOT = ROOT / ".cache" / "flask-diagnostic" +IMPLEMENTATION = "python-flask" +INITIAL_ARMS = ("baseline", "switch-1ms", "queue-quiet") +ARMS = (*INITIAL_ARMS, "affinity-one") + + +def diagnostic_plan(phase: str = "initial") -> list[dict]: + """Fix order before collecting data; reverse arms and endpoints in block two.""" + require(phase in ("initial", "scheduling"), "unknown diagnostic phase") + if phase == "scheduling": + return [ + { + "id": f"block-{index}-{arm}", + "block": index, + "arm": arm, + "endpoints": ["/json", "/db/42"], + } + for index, arm in enumerate(("baseline", "affinity-one", "baseline"), 1) + ] + return [ + { + "id": f"block-{block}-{arm}", + "block": block, + "arm": arm, + "endpoints": ["/json", "/db/42"] if block == 1 else ["/db/42", "/json"], + } + for block, arms in ((1, INITIAL_ARMS), (2, tuple(reversed(INITIAL_ARMS)))) + for arm in arms + ] + + +def validate_output_path(path: Path) -> Path: + candidate = path.absolute() + root = CACHE_ROOT.absolute() + require(candidate.is_relative_to(root), "output must stay below Flask diagnostic cache") + require(candidate.resolve().is_relative_to(root.resolve()), "output escapes diagnostic cache") + require(candidate.name == "diagnostic.json", "expected diagnostic.json output") + for component in (candidate, *candidate.parents): + require(not component.is_symlink(), "symlink in diagnostic output path") + return candidate + + +def hook_source(arm: str) -> str: + """Only fixed, reviewed interventions are allowed; baseline has no startup hook.""" + require(arm in ARMS, "unknown diagnostic arm") + if arm == "baseline": + return "" + intervention = { + "switch-1ms": "sys.setswitchinterval(0.001)", + "queue-quiet": 'logging.getLogger("waitress.queue").disabled = True', + "affinity-one": "os.sched_setaffinity(0, {allowed_before[0]})", + }[arm] + return ( + "import json, logging, os, sys\n" + "allowed_before = sorted(os.sched_getaffinity(0))\n" + + intervention + + "\nif os.getpid() == 1:\n" + + ' print("FLASK_DIAGNOSTIC " + json.dumps({"arm": ' + + repr(arm) + + ', "pid": os.getpid(), "interval": sys.getswitchinterval(), ' + + '"queue_disabled": logging.getLogger("waitress.queue").disabled, ' + + '"allowed_before": allowed_before, "allowed_after": sorted(os.sched_getaffinity(0)), ' + + '"gil_enabled": getattr(sys, "_is_gil_enabled", lambda: None)()}), flush=True)\n' + ) + + +def compose_override(arm: str, hook: Path) -> dict: + hook_source(arm) # Validate before producing any container configuration. + if arm == "baseline": + return {} + return { + "services": { + IMPLEMENTATION: { + "environment": {"PYTHONPATH": "/flask-diagnostic"}, + "volumes": [ + { + "type": "bind", + "source": str(hook.resolve()), + "target": "/flask-diagnostic", + "read_only": True, + } + ], + } + } + } + + +class FlaskEnvironment(DockerEnvironment): + """Keep production image/entrypoint intact and instrument only between windows.""" + + def __init__(self, oha: Path, directory: Path, arm: str): + super().__init__(oha, directory, health_policy=EXTERNAL_READINESS) + self.arm = arm + self.since = now() + self.hook_evidence = None + if arm != "baseline": + hook = self.artifacts / "hook" + hook.mkdir() + content = hook_source(arm) + (hook / "sitecustomize.py").write_text(content, encoding="utf-8") + override = self.artifacts / "diagnostic.compose.json" + atomic_json(override, compose_override(arm, hook)) + self.prefix = self.prefix[:-2] + ["-f", str(override), *self.prefix[-2:]] + self.hook_evidence = {"sha256": hashlib.sha256(content.encode()).hexdigest()} + + def start(self, identifier: str) -> dict: + result = super().start(identifier) + if self.arm != "baseline": + logs = execute(["docker", "logs", "--tail", "100", self.container], timeout=10) + markers = [ + line.removeprefix("FLASK_DIAGNOSTIC ") + for line in logs.splitlines() + if line.startswith("FLASK_DIAGNOSTIC ") + ] + require(len(markers) == 1, "diagnostic startup hook was not observed exactly once") + observed = strict_json(markers[0].encode()) + require( + observed.get("arm") == self.arm and observed.get("pid") == 1, + "hook belongs to wrong process/arm", + ) + require( + observed.get("queue_disabled") is (self.arm == "queue-quiet"), + "queue logger intervention mismatch", + ) + require( + observed.get("interval") == (0.001 if self.arm == "switch-1ms" else 0.005), + "thread interval intervention mismatch", + ) + before = observed.get("allowed_before") + require(type(before) is list and bool(before), "missing allowed CPU set") + expected = before[:1] if self.arm == "affinity-one" else before + require(observed.get("allowed_after") == expected, "CPU affinity intervention mismatch") + self.hook_evidence["observed"] = observed + result["diagnostic_hook"] = self.hook_evidence + return result + + def observe(self, label: str) -> dict: + """Snapshots include collection overhead; no exec/polling runs during oha.""" + self.check() + require(re.fullmatch(r"[a-z0-9-]+", label) is not None, "invalid observation label") + script = """import json, os, sys +from pathlib import Path +paths = ['/sys/fs/cgroup/cpu.stat', '/proc/1/stat', '/proc/1/status'] +result = {p: Path(p).read_text()[:16384] if Path(p).is_file() else None for p in paths} +threads = {} +vanished = [] +for path in sorted(Path('/proc/1/task').glob('[0-9]*'))[:64]: + try: + threads[path.name] = {name: (path / name).read_text()[:16384] + for name in ('comm', 'stat', 'schedstat', 'status')} + except FileNotFoundError: + vanished.append(path.name) +result['threads'] = threads +result['vanished_threads'] = vanished +result['helper_runtime_not_pid1'] = {'version': sys.version, + 'gil_enabled': getattr(sys, '_is_gil_enabled', lambda: None)(), + 'affinity': sorted(os.sched_getaffinity(0))} +print(json.dumps(result)) +""" + raw = execute(["docker", "exec", self.container, "python", "-c", script], timeout=10) + counters = strict_json(raw.encode()) + logs = execute( + [ + "docker", + "logs", + "--timestamps", + "--since", + self.since, + "--tail", + "10000", + self.container, + ], + timeout=10, + ) + encoded = logs.encode() + limited = encoded[-2 * 1024 * 1024 :] + log_path = self.artifacts / f"{label}.server.log" + log_path.write_bytes(limited) + snapshot = { + "at": now(), + "container_id": self.container, + "counters": counters, + "server_log": log_path.name, + "log_tail_limit_lines": 10000, + "log_tail_may_be_truncated": len(logs.splitlines()) >= 10000 + or len(limited) != len(encoded), + "queue_warning_lines_in_tail": limited.count(b"Task queue depth is"), + } + atomic_json(self.artifacts / f"{label}.observation.json", snapshot) + self.check() + return snapshot + + +def run_cell(environment, cell: dict, *, contract=run_contract, expected_image=None) -> dict: + failure = None + try: + require(environment.health_policy == EXTERNAL_READINESS, "diagnostic health policy differs") + require(environment.connections == 50, "diagnostic connections must be 50") + require(environment.request_timeout == 15, "diagnostic request timeout must be 15s") + environment.build(IMPLEMENTATION) + container = environment.start(IMPLEMENTATION) + if expected_image is not None: + require( + container.get("image_id") == expected_image, + "diagnostic image changed between cells", + ) + checks = contract("http://127.0.0.1:8080", implementation=IMPLEMENTATION) + require( + type(checks) is int and checks == 2 * len(load_cases()), "incomplete shared contract" + ) + environment.check() + endpoints = [] + for endpoint in cell["endpoints"]: + label = endpoint.strip("/").replace("/", "-") + before = environment.observe(label + "-before") + warmup = environment.measure(endpoint, 5, 0) + validate_run(warmup) + runs = [] + for index in (1, 2, 3): + observed = environment.measure(endpoint, 10, index) + validate_run(observed) + runs.append(observed) + after = environment.observe(label + "-after") + endpoints.append( + { + "endpoint": endpoint, + "warmup": warmup, + "runs": runs, + "selected": select_run(runs), + "before": before, + "after": after, + } + ) + environment.check() + readiness = copy.deepcopy(environment.readiness) + except BaseException as error: + failure = error + raise + finally: + try: + with protect_cleanup(): + environment.cleanup() + except (BenchmarkFailure, OSError) as error: + if failure is not None: + raise BenchmarkFailure(f"{failure}; cleanup also failed: {error}") from failure + raise + return { + **cell, + "container": container, + "readiness": readiness, + "contract_checks": checks, + "artifact_directory": str(environment.artifacts), + "endpoints": endpoints, + } + + +def run_diagnostic( + factory, output: Path, *, metadata: dict, contract=run_contract, phase="initial" +) -> dict: + output = validate_output_path(output) + for key in ("source_commit", "source_tree"): + require( + type(metadata.get(key)) is str and re.fullmatch(r"[0-9a-f]{40}", metadata[key]), + "diagnostic source metadata missing " + key, + ) + versions = metadata.get("versions", {}) + require( + set(versions) == {IMPLEMENTATION} + and set(implementation(IMPLEMENTATION)["version_fields"]) <= set(versions[IMPLEMENTATION]), + "diagnostic requires Flask version metadata", + ) + plan = diagnostic_plan(phase) + started_at = now() + cells = [] + progress = output.parent / "progress.json" + for cell in plan: + atomic_json( + progress, + { + "status": "running", + "official": False, + "publishable": False, + "metadata": metadata, + "plan": plan, + "active_cell": cell["id"], + "cells": cells, + }, + ) + print(f"Flask diagnostic {cell['id']}", flush=True) + try: + environment = factory(cell) + expected_image = cells[0]["container"]["image_id"] if cells else None + cells.append( + run_cell(environment, cell, contract=contract, expected_image=expected_image) + ) + except BaseException as error: + atomic_json( + progress, + { + "status": "failed", + "official": False, + "publishable": False, + "metadata": metadata, + "plan": plan, + "failed_cell": cell["id"], + "error_type": type(error).__name__, + "cells": cells, + }, + ) + raise + result = { + "schema_version": 1, + "mode": "flask-diagnostic", + "status": "verified", + "official": False, + "publishable": False, + "implementation": IMPLEMENTATION, + "started_at": started_at, + "completed_at": now(), + "metadata": copy.deepcopy(metadata), + "phase": phase, + "conditions": { + "connections": 50, + "warmup_seconds": 5, + "duration_seconds": 10, + "runs": 3, + "api_health_policy": EXTERNAL_READINESS, + }, + "limitations": [ + "Non-publishing diagnostic, not an official result or framework ranking.", + "CPU snapshots span warm-up and measurement plus observation overhead.", + "Server logs are bounded tails; warning counts may be lower bounds.", + "A bounded diagnostic matrix does not establish cross-host or universal performance effects.", + ], + "plan": plan, + "cells": cells, + } + atomic_json(output, result) + atomic_json( + progress, + { + "status": "completed", + "official": False, + "publishable": False, + "metadata": metadata, + "plan": plan, + "cells": cells, + }, + ) + return result + + +def main(argv: list[str] | None = None) -> int: + from .environment import provenance, registered_pinned_versions + from .install_oha import ensure_oha + + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--phase", choices=("initial", "scheduling"), default="initial") + args = parser.parse_args(argv) + + def interrupted(signum, _frame): + raise KeyboardInterrupt(f"received signal {signum}") + + previous = {sig: signal.signal(sig, interrupted) for sig in (signal.SIGINT, signal.SIGTERM)} + try: + directory = CACHE_ROOT / uuid.uuid4().hex + output = validate_output_path(directory / "diagnostic.json") + oha = ensure_oha() + metadata = provenance(oha) + metadata["versions"] = {IMPLEMENTATION: registered_pinned_versions()[IMPLEMENTATION]} + metadata["diagnostic_plan_sha256"] = hashlib.sha256( + json.dumps(diagnostic_plan(args.phase), sort_keys=True).encode() + ).hexdigest() + run_diagnostic( + lambda cell: FlaskEnvironment(oha, directory / cell["id"], cell["arm"]), + output, + metadata=metadata, + phase=args.phase, + ) + print(f"Flask diagnostic complete: {output}; no results published.") + return 0 + except (BenchmarkFailure, ContractFailure, OSError, ValueError, KeyboardInterrupt) as error: + print(f"Flask diagnostic failed: {error}", file=sys.stderr) + return 1 + finally: + for sig, handler in previous.items(): + signal.signal(sig, handler) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/docs/FLASK-DIAGNOSTIC.md b/docs/FLASK-DIAGNOSTIC.md new file mode 100644 index 0000000..9a1834c --- /dev/null +++ b/docs/FLASK-DIAGNOSTIC.md @@ -0,0 +1,189 @@ +# Flask JSON diagnostic + +## Conclusion and boundary + +Issue #70 found a reproducible **CPU-placement-sensitive scheduling effect** in +this repository's pinned CPython / Flask / Waitress stack under a one-CPU quota. +It did not establish that Flask's JSON handler is intrinsically slow, that the +published numbers were fabricated, or that a particular GIL/kernel function is +the root cause. + +In one same-job baseline / one-CPU-affinity / baseline diagnostic, JSON throughput +was approximately 102 / 1,883 / 103 requests/s. DB throughput stayed approximately +1,190 / 1,204 / 1,207 requests/s. The CPU quota, memory limit, production image, +Waitress worker count and DB pool limit were unchanged. Affinity selected one +already-allowed logical CPU; it did not reserve a dedicated physical core. + +These are **non-publishing diagnostic observations**, not replacement official +results, an approved production optimization, or a cross-framework ranking. +Existing applications, pins, official cohorts, methodology and `results/**` were +not changed. Neither #52's deferred publishing/protection work nor #59's next +justified official-run verification is completed by these experiments. + +## Retained evidence + +All sources below are immutable measured snapshots, not whichever branch is +currently checked out. The corresponding run artifacts were downloaded and their +ZIP SHA-256 values checked during the 2026-09-18 investigation. + +| Evidence | Source commit | Run / attempt | Artifact ID | +| --- | --- | --- | --- | +| Original eight-stack publication | `c8e033ea176366644f5bed5ded12cab79e6fcbeb` | [34925168324 / 1](https://github.com/tappe9/simple-api-benchmark/actions/runs/34925168324) | `10380696186` | +| Initial hypothesis matrix | `cbebd385b0f78503d797fcc73167f1ecce1b8611` | [35299848616 / 1](https://github.com/tappe9/simple-api-benchmark/actions/runs/35299848616) | `10529950752` | +| Scheduling probe | `a153a103f445af8cd571687fb095cc0d2f4cb9a3` | [35300974388 / 1](https://github.com/tappe9/simple-api-benchmark/actions/runs/35300974388) | `10529473210` | + +ZIP SHA-256 values, in the same order: + +```text +cc376a99e573d65c5cedc165ee00fda294795f45ed491a5586ef419f96b9989b +4df46b3f0843e1486d9fe3382c2962efe952dd414283c2f46ba731f6de0e9bf8 +ae16af92916f62f198ed509d57273831b5b118c8de4c2f054d57037e42123bec +``` + +The original selected JSON matched the published file byte-for-byte at audit +main `9b92c74cba34bd5a4bfc389075b9bef135131034`. All 72 original measured windows +passed `benchmark.report.audit_raw`. The original Flask JSON run 2 had mean +response time 315.451675 ms and mean first-byte time 315.450804 ms: downloading +the small body does not explain the observed latency. + +Both diagnostic archives were independently replayed against `parse_oha`, the +memory samples, selected whole-run values, observation files and source/tree +identities: **36 + 18 measured windows and 12 + 6 warm-ups**, with no transport +errors or non-200 responses. Every cell used a fresh container; all cells within +each job used one identical image ID. Image IDs between separate jobs need not +match and those jobs must not be pooled as paired measurements. The original +publication does not contain the later per-thread or throttling observations. + +Artifacts have finite retention; their availability was verified on the audit +date, not guaranteed forever. An expired original artifact is not recreated by +rerunning a diagnostic. See [Issue #70](https://github.com/tappe9/simple-api-benchmark/issues/70) +for the predeclared plans and review record. + +## Experiment 1: switch interval and queue logging + +The predefined arms were baseline, Python switch interval 1 ms, and disabled +`waitress.queue` warning output. The second block reversed both arm order and +JSON/DB endpoint order. Each cell recreated PostgreSQL and the API container. +Only the named intervention was applied through a read-only diagnostic startup +hook; baseline had no hook. + +Common settings were HTTP/1.1, 50 connections, external readiness stopped before +load, a five-second warm-up and three ten-second measured windows per endpoint. +The unchanged environment enforced one CPU, 512 MiB, one Waitress worker thread +and DB pool maximum 10. These ten-second windows are not the official thirty- +second profile. + +| Block | Arm | JSON requests/s | DB requests/s | +| --- | --- | ---: | ---: | +| 1 | baseline | 95.615 | 1,196.952 | +| 1 | switch-1ms | 92.682 | 1,215.155 | +| 1 | queue-quiet | 77.811 | 1,425.894 | +| 2 | queue-quiet | 81.356 | 1,388.971 | +| 2 | switch-1ms | 92.198 | 1,203.858 | +| 2 | baseline | 103.816 | 1,205.640 | + +The JSON/DB inversion reproduced. Neither tested intervention improved JSON in +this experiment. Suppressing warnings changed DB throughput, but did not explain +or fix the JSON anomaly. This does not exclude every possible GIL or logging +interaction; it rejects these two simple proposed remedies under the tested +conditions. + +## Experiment 2: bracketing CPU-affinity probe + +The follow-up plan was recorded before execution. On a new job it ran baseline, +`affinity-one`, then baseline, with fresh environments and JSON then DB in every +cell. Load settings stayed as above. A startup marker confirmed the intervention +in PID 1; `/proc/1/task/*/status` snapshots confirmed the observed threads' CPU +masks. The helper interpreter's metadata is separately labeled, not passed off +as PID 1's runtime configuration. + +| Arm | JSON requests/s | JSON mean ms | DB requests/s | +| --- | ---: | ---: | ---: | +| baseline before | 101.553 | 491.030 | 1,190.344 | +| affinity-one | 1,882.871 | 26.508 | 1,204.494 | +| baseline after | 102.701 | 485.357 | 1,206.691 | + +JSON throughput in the affinity arm was 18.54 and 18.33 times the two bracketing +baselines. These are within-diagnostic contrasts, not an improvement applied to +an existing official publication. The baselines could run on logical CPUs 0-3; +the affinity arm and its observed threads used CPU 0, still with the one-CPU +quota and identical production image. + +JSON observation-span counter deltas were: + +| Arm | Throttled periods / periods | Accumulated throttled seconds | Main-thread CPU seconds | Waitress-worker CPU seconds | Worker scheduler slices | +| --- | ---: | ---: | ---: | ---: | ---: | +| baseline before | 349 / 363 | 13.613507 | 26.358 | 9.030 | 1,081,401 | +| affinity-one | 25 / 360 | 0.003450 | 17.314 | 17.768 | 28,258 | +| baseline after | 350 / 363 | 13.393709 | 26.353 | 9.038 | 1,066,044 | + +These cumulative spans include warm-up, measured windows and observation overhead. +They are not per-request profiles. `schedstat` scheduler slices are not a count +of GIL handoffs, and accumulated throttling time is not a latency decomposition. +DB-pool helper threads showed no material CPU growth in these snapshots. + +## Interpretation and next decision + +[Waitress's documented design](https://docs.pylonsproject.org/projects/waitress/en/stable/design.html) +separates main-thread client I/O from worker-thread application execution. +[CPU bandwidth control](https://docs.kernel.org/scheduler/sched-bwc.html) limits +aggregate runtime using quota and period; a one-CPU quota is not the same setting +as restricting execution to one logical CPU. Those distinctions explain why +"one process / one worker / one CPU" does not specify every scheduling condition. + +The affinity intervention, bracketing baselines and thread/cgroup observations +support a scheduling/CPU-quota interaction in this particular stack. GIL +contention, I/O wakeups, lock contention and kernel scheduling are plausible +lower-level mechanisms, but these experiments do not isolate one of them. No +stack sampling or controlled quota-only experiment was performed. In particular, +it would overstate the evidence to say that the GIL alone, CPU throttling alone, +or a particular Waitress defect was proved responsible. Host noise, one treatment +cell and lack of independent-host replication limit generalization. + +The supported next action is to decide and document common CPU-allocation +semantics before changing any official comparison. Keeping the existing quota- +only baseline with an explanation is valid. Adopting affinity would require a +separately reviewed common-methodology decision across the complete cohort, not a +Flask-only boost or a silent overwrite of history. Any later official run still +requires explicit authorization for measurement, audited publication and Pages. + +## Reproducing the diagnostics + +Use a clean committed checkout, Python 3.10+, Git, curl, Make and a local Linux +Docker Compose v2 daemon reachable through a Unix socket. Leave port 8080 free +and do not run other benchmark services concurrently. No host Flask installation +is needed; the existing pinned production container is built and exercised. + +```bash +# No Docker or load: regression tests and CLI help. +python -m unittest discover -s tests -p test_benchmark_flask_diagnostic.py -v +python -m benchmark.flask_diagnostic --help + +# Explicit non-publishing invocations; do not run both concurrently. +python -m benchmark.flask_diagnostic --phase initial +python -m benchmark.flask_diagnostic --phase scheduling +``` + +The initial phase has 36 measured windows and 12 warm-ups; scheduling has 18 +measured windows and six warm-ups. Both retain all attempted samples without +selective retries under `.cache/flask-diagnostic//`. The final +`diagnostic.json` is written only after every cell and its scoped cleanup succeed; +`progress.json` and raw files retain failure evidence. Each invocation uses new +owned Compose projects. CLI help and invalid arguments exit before tool +installation or load. An image change between cells now fails before the next +cell's contract/load; the already measured archives had identical images and +are unchanged by this additional guard. + +The report records source/tree, versions, locks, conditions, plan digest, +container identities, readiness, interventions, raw run summaries, memory, +bounded server-log tails and outside-load observations. Tail warning counts may +be lower bounds. Collection introduces overhead outside individual oha windows. +SIGINT/SIGTERM attempt bounded cleanup; SIGKILL/host failure cannot guarantee it. +Inspect the printed owned project name before manually removing leftover +resources, and never use broad Docker cleanup commands for this investigation. + +There is no permanent diagnostic workflow or new automatic measurement. The +branch-only investigation workflow was removed after its evidence was retained. +Diagnostics cannot feed the official publisher or Pages. Investigation scratch +notes and Superpowers plans remain untracked under the paths in `AGENTS.md`; +this guide is the durable public explanation, not an implementation plan. diff --git a/docs/FLASK.md b/docs/FLASK.md index 26b5db6..54724fe 100644 --- a/docs/FLASK.md +++ b/docs/FLASK.md @@ -51,3 +51,10 @@ The gates cover unit behavior, dependency/hash locks, real PostgreSQL and Docker ## Publication boundary Python / Flask is registered and CI-covered, but Issue #32 does not activate a new official benchmark cohort. `active_cohort` remains the frozen `four-stack-v1`; expanded-cohort activation and new official measurements are separate work. + +## JSON performance diagnostic + +The [Flask diagnostic guide](FLASK-DIAGNOSTIC.md) records Issue #70's controlled +non-publishing experiments, CPU-placement sensitivity, evidence limits and +reproduction commands. Diagnostic values do not replace the official results or +approve a Flask-specific runtime/CPU tuning change. diff --git a/tests/test_benchmark_flask_diagnostic.py b/tests/test_benchmark_flask_diagnostic.py new file mode 100644 index 0000000..23b64d3 --- /dev/null +++ b/tests/test_benchmark_flask_diagnostic.py @@ -0,0 +1,325 @@ +"""Flask investigation must keep all arms non-publishing and clean every cell.""" + +import importlib +import importlib.util +import inspect +import json +import os +import shutil +import subprocess +import sys +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch + +from test_benchmark_runner import Environment + +from benchmark.environment import registered_pinned_versions +from benchmark.healthcheck import EXTERNAL_READINESS +from benchmark.report import validate_report +from benchmark.results import BenchmarkFailure + + +class DiagnosticEnvironment(Environment): + def __init__(self, root, cell, fail=None): + super().__init__(fail) + self.artifacts = root / cell["id"] + self.artifacts.mkdir() + self.health_policy = EXTERNAL_READINESS + self.connections = 50 + self.request_timeout = 15 + self.readiness = {"attempts": 1} + + def observe(self, label): + self.event("observe") + return {"label": label, "cpu_stat": "usage_usec 1"} + + +class FlaskDiagnosticTests(unittest.TestCase): + def setUp(self): + self.assertIsNotNone( + importlib.util.find_spec("benchmark.flask_diagnostic"), + "Flask needs the bounded non-publishing investigation runner", + ) + self.module = importlib.import_module("benchmark.flask_diagnostic") + directory = tempfile.TemporaryDirectory() + self.addCleanup(directory.cleanup) + self.root = Path(directory.name) + self.cache = self.root / ".cache" / "flask-diagnostic" + self.cache.mkdir(parents=True) + patcher = patch.object(self.module, "CACHE_ROOT", self.cache) + patcher.start() + self.addCleanup(patcher.stop) + self.output = self.cache / "diagnostic.json" + self.output.write_bytes(b"previous diagnostic") + self.published = self.root / "results" / "latest.json" + self.published.parent.mkdir() + self.published.write_bytes(b"previous official result") + self.metadata = { + "source_commit": "a" * 40, + "source_tree": "b" * 40, + "versions": {"python-flask": registered_pinned_versions()["python-flask"]}, + } + self.environments = [] + + def run_diagnostic(self, fail=None, alter=None, phase="initial"): + def factory(cell): + environment = DiagnosticEnvironment(self.cache, cell, fail) + if alter is not None: + alter(environment) + self.environments.append(environment) + return environment + + def contract(url, *, implementation): + return self.environments[-1].contract(url, implementation=implementation) + + return self.module.run_diagnostic( + factory, self.output, metadata=self.metadata, contract=contract, phase=phase + ) + + def test_fixed_counterbalanced_plan_and_every_sample_are_preserved(self): + result = self.run_diagnostic() + self.assertEqual( + [cell["arm"] for cell in result["cells"]], + ["baseline", "switch-1ms", "queue-quiet", "queue-quiet", "switch-1ms", "baseline"], + ) + self.assertEqual(len(self.environments), 6) + for position, (cell, environment) in enumerate(zip(result["cells"], self.environments)): + endpoints = ["/json", "/db/42"] if position < 3 else ["/db/42", "/json"] + self.assertEqual([item["endpoint"] for item in cell["endpoints"]], endpoints) + measured = [event for event in environment.events if isinstance(event, tuple)] + self.assertEqual( + measured, + [ + ("python-flask", endpoint, 5 if index == 0 else 10, index) + for endpoint in endpoints + for index in (0, 1, 2, 3) + ], + ) + self.assertEqual(environment.events.count("cleanup"), 1) + self.assertIsNone(environment.active) + self.assertEqual(cell["readiness"], {"attempts": 1}) + self.assertFalse(result["official"]) + self.assertFalse(result["publishable"]) + self.assertEqual(result["mode"], "flask-diagnostic") + self.assertEqual(json.loads(self.output.read_bytes()), result) + self.assertEqual(self.published.read_bytes(), b"previous official result") + with self.assertRaises(BenchmarkFailure): + validate_report(result) + + def test_all_failure_stages_clean_and_never_write_partial_success(self): + for stage in ( + "build", + "startup", + "contract", + "state", + "warmup", + "measurement", + "metric", + "observe", + "cleanup", + ): + for folder in self.cache.iterdir(): + if folder.is_dir(): + shutil.rmtree(folder) + with self.subTest(stage=stage), self.assertRaises(BenchmarkFailure): + self.run_diagnostic(stage) + environment = self.environments[-1] + self.assertEqual(environment.events.count("cleanup"), 1) + self.assertEqual(self.output.read_bytes(), b"previous diagnostic") + self.assertEqual(self.published.read_bytes(), b"previous official result") + + def test_cleanup_precedes_successful_output_and_retains_readiness(self): + def alter(environment): + original = environment.cleanup + + def cleanup(): + self.assertEqual(self.output.read_bytes(), b"previous diagnostic") + original() + environment.readiness = None + + environment.cleanup = cleanup + + self.run_diagnostic(alter=alter) + + def test_interruption_cleans_and_does_not_retry(self): + def alter(environment): + def interrupted(*args): + raise KeyboardInterrupt + + environment.measure = interrupted + + with self.assertRaises(KeyboardInterrupt): + self.run_diagnostic(alter=alter) + self.assertEqual(len(self.environments), 1) + self.assertEqual(self.environments[0].events.count("cleanup"), 1) + self.assertEqual(self.output.read_bytes(), b"previous diagnostic") + + def test_incomplete_contract_cannot_start_load(self): + def alter(environment): + environment.contract = lambda *args, **kwargs: True + + with self.assertRaisesRegex(BenchmarkFailure, "contract"): + self.run_diagnostic(alter=alter) + self.assertEqual(self.environments[0].measures, 0) + + def test_invalid_profile_or_metadata_prevents_load(self): + for field, value in ( + ("connections", 2), + ("request_timeout", 1), + ("health_policy", "container-healthcheck"), + ): + with self.subTest(field=field), self.assertRaises(BenchmarkFailure): + self.run_diagnostic(alter=lambda environment: setattr(environment, field, value)) + self.assertNotIn("build", self.environments[-1].events) + self.environments[-1].artifacts.rmdir() + self.metadata["source_commit"] = "unknown" + with self.assertRaises(BenchmarkFailure): + self.run_diagnostic() + + def test_later_cell_failure_preserves_completed_evidence_without_final_result(self): + count = 0 + + def alter(environment): + nonlocal count + count += 1 + if count == 2: + environment.fail = "measurement" + + with self.assertRaises(BenchmarkFailure): + self.run_diagnostic(alter=alter) + self.assertEqual(len(self.environments), 2) + self.assertTrue(all(e.events.count("cleanup") == 1 for e in self.environments)) + progress = json.loads((self.cache / "progress.json").read_bytes()) + self.assertEqual(progress["status"], "failed") + self.assertEqual(progress["failed_cell"], "block-1-switch-1ms") + self.assertEqual(len(progress["cells"]), 1) + self.assertEqual(self.output.read_bytes(), b"previous diagnostic") + + def test_image_drift_stops_before_loading_the_next_cell(self): + count = 0 + + def alter(environment): + nonlocal count + count += 1 + if count == 2: + original = environment.start + + def changed_image(identifier): + result = original(identifier) + result["image_id"] = "sha256:changed-test-image" + return result + + environment.start = changed_image + + with self.assertRaisesRegex(BenchmarkFailure, "image changed"): + self.run_diagnostic(alter=alter, phase="scheduling") + self.assertEqual(len(self.environments), 2) + self.assertEqual(self.environments[0].measures, 6) + self.assertEqual(self.environments[1].measures, 0) + self.assertTrue(all(e.events.count("cleanup") == 1 for e in self.environments)) + self.assertEqual(self.output.read_bytes(), b"previous diagnostic") + self.assertEqual(self.published.read_bytes(), b"previous official result") + progress = json.loads((self.cache / "progress.json").read_bytes()) + self.assertEqual(progress["status"], "failed") + self.assertEqual(len(progress["cells"]), 1) + + def test_invalid_measurement_is_not_retried_or_selected(self): + def alter(environment): + original = environment.measure + + def measure(*args): + value = original(*args) + value["requests_per_second"] = float("nan") + return value + + environment.measure = measure + + with self.assertRaises(BenchmarkFailure): + self.run_diagnostic(alter=alter) + self.assertEqual(len(self.environments), 1) + self.assertEqual(self.environments[0].events.count("cleanup"), 1) + self.assertEqual(self.output.read_bytes(), b"previous diagnostic") + + def test_output_and_symlinks_cannot_escape_cache(self): + for path in (self.published, self.cache / ".." / "diagnostic.json"): + with self.assertRaises(BenchmarkFailure): + self.module.validate_output_path(path) + linked = self.cache / "linked" + linked.symlink_to(self.published.parent, target_is_directory=True) + with self.assertRaises(BenchmarkFailure): + self.module.validate_output_path(linked / "diagnostic.json") + self.output.unlink() + self.output.symlink_to(self.published) + with self.assertRaises(BenchmarkFailure): + self.run_diagnostic() + self.assertEqual(self.published.read_bytes(), b"previous official result") + + def test_startup_hooks_change_only_the_declared_factor(self): + for arm in ("baseline", "switch-1ms", "queue-quiet"): + code = self.module.hook_source(arm) + probe = ( + "import sys, logging, json\n" + + code + + '\nprint(json.dumps({"interval":sys.getswitchinterval(),"queue":logging.getLogger("waitress.queue").disabled,"root":logging.getLogger().disabled}))' + ) + output = subprocess.check_output([sys.executable, "-c", probe], text=True, timeout=5) + observed = json.loads(output.splitlines()[-1]) + self.assertEqual( + observed["interval"], 0.001 if arm == "switch-1ms" else sys.getswitchinterval() + ) + self.assertEqual(observed["queue"], arm == "queue-quiet") + self.assertFalse(observed["root"]) + with self.assertRaises(BenchmarkFailure): + self.module.hook_source("arbitrary-code") + + def test_scheduling_phase_has_bracketing_baselines_and_unchanged_load(self): + self.assertIn("phase", inspect.signature(self.module.diagnostic_plan).parameters) + plan = self.module.diagnostic_plan("scheduling") + self.assertEqual([cell["arm"] for cell in plan], ["baseline", "affinity-one", "baseline"]) + self.assertTrue(all(cell["endpoints"] == ["/json", "/db/42"] for cell in plan)) + with self.assertRaises(BenchmarkFailure): + self.module.diagnostic_plan("arbitrary") + result = self.run_diagnostic(phase="scheduling") + self.assertEqual(len(result["cells"]), 3) + self.assertEqual(sum(e.measures for e in self.environments), 18) + self.assertEqual(result["phase"], "scheduling") + + def test_affinity_hook_changes_only_the_child_allowed_cpu_set(self): + self.assertIn("affinity-one", self.module.ARMS) + before = sorted(os.sched_getaffinity(0)) + probe = self.module.hook_source("affinity-one") + ( + '\nprint(json.dumps({"affinity":sorted(os.sched_getaffinity(0)), ' + '"interval":sys.getswitchinterval(),"queue":logging.getLogger("waitress.queue").disabled}))' + ) + output = subprocess.check_output([sys.executable, "-c", probe], text=True, timeout=5) + observed = json.loads(output.splitlines()[-1]) + self.assertEqual(observed["affinity"], [before[0]]) + self.assertEqual(observed["interval"], 0.005) + self.assertFalse(observed["queue"]) + self.assertEqual(sorted(os.sched_getaffinity(0)), before) + + def test_help_and_invalid_arguments_never_start_measurement(self): + for args, code in ((["--help"], 0), (["--unknown"], 2)): + with self.subTest(args=args), patch.object(sys, "argv", ["flask-diagnostic", *args]): + with patch( + "benchmark.install_oha.ensure_oha", + side_effect=AssertionError("load must not start"), + ): + with self.assertRaises(SystemExit) as stopped: + self.module.main() + self.assertEqual(stopped.exception.code, code) + + def test_baseline_override_does_not_change_command_or_environment(self): + self.assertEqual(self.module.compose_override("baseline", self.cache), {}) + for arm in ("switch-1ms", "queue-quiet"): + config = self.module.compose_override(arm, self.cache) + service = config["services"]["python-flask"] + self.assertEqual(set(service), {"environment", "volumes"}) + self.assertTrue(service["volumes"][0]["read_only"]) + self.assertEqual(service["environment"], {"PYTHONPATH": "/flask-diagnostic"}) + + +if __name__ == "__main__": + unittest.main()