From 67ca471326bda49334f1443adc700c6ff216bb9b Mon Sep 17 00:00:00 2001 From: Derek Date: Sun, 11 Oct 2026 13:14:53 +1100 Subject: [PATCH 1/5] fix: e2e harness restores the stack, runs via uv `make test-e2e`, `test-resilience` and `test-flows` run through `uv run --script`. The scripts declare PyYAML 6.0.3 and Playwright 1.63.0 inline (PEP 723), so a host needs uv and nothing else from PyPI. The suite reads the compose project before it stops anything, then on every way out (a failure and a SIGTERM included) starts each service that was running or had run to completion again, from the compose files, project and environment it came from. A last verdict fails when one comes back on a different config hash. `make test-e2e` no longer leaves only ClickHouse running. The data file goes to the ingest edge 16 requests at a time. The receiver answers once the loader has flushed the record, so one request at a time paid a flush per event, and 1000 events took 323 s. post.py deletes its three ingest rows once every claim has run, the console claim that reads them back included. A table still carrying the `_timestamp_optimized` projection refuses a lightweight DELETE with Code 344, so it falls back to an `ALTER TABLE ... DELETE` mutation, polled for up to 60 s and reported rather than failed. --- Makefile | 9 +- docs/developing.md | 10 +- docs/observability.md | 2 + docs/troubleshooting.md | 9 +- scripts/_pipeline.py | 42 +++++++ scripts/_search_rule.py | 6 +- scripts/_stack_state.py | 165 +++++++++++++++++++++++++ scripts/post.py | 135 +++++++++++++++----- scripts/test_e2e.py | 196 ++++++++++++++++++++++++------ scripts/test_flows.py | 7 ++ scripts/tests/test_ingest_send.py | 83 +++++++++++++ scripts/tests/test_post.py | 120 ++++++++++++++++++ scripts/tests/test_stack_state.py | 155 +++++++++++++++++++++++ 13 files changed, 855 insertions(+), 84 deletions(-) create mode 100644 scripts/_stack_state.py create mode 100644 scripts/tests/test_ingest_send.py create mode 100644 scripts/tests/test_stack_state.py diff --git a/Makefile b/Makefile index 353da34..7078fbd 100644 --- a/Makefile +++ b/Makefile @@ -646,19 +646,22 @@ check-proxy: ## Validate the Envoy proxy configs, plain and TLS, on the pinned E # The suite starts each test's stack from its own -f list. Under console TLS that stack needs the TLS fragment and the network it pins, so the suite takes every fragment `make ci` chains, the exposure opt-outs and container logs included; with TLS off it keeps its own list. DFE_E2E_COMPOSE_FILES follows either way, and the suite skips the empty entries the colons leave. E2E_COMPOSE_FILES = $(if $(filter true,$(strip $(DFE_PROXY_TLS))),$(UI_CHAIN)):$(strip $(DFE_E2E_COMPOSE_FILES)) +# The scripts that import packages beyond the standard library declare them inline (PEP 723), so uv installs them and a host needs only uv. +UV ?= uv + .PHONY: test-e2e test-e2e: ## End-to-end test executor (pass test names via E2E_TESTS) - @DFE_E2E_COMPOSE_FILES='$(E2E_COMPOSE_FILES)' python3 ./scripts/test_e2e.py $(E2E_TESTS) + @DFE_E2E_COMPOSE_FILES='$(E2E_COMPOSE_FILES)' $(UV) run --script ./scripts/test_e2e.py $(E2E_TESTS) # Opt-in, never part of test-e2e: each test stops a backing service or a DFE app # under load. .PHONY: test-resilience test-resilience: ## Outage tests: stop a service under load and prove the rest survives (pass test names via E2E_TESTS) - @DFE_E2E_COMPOSE_FILES='$(E2E_COMPOSE_FILES)' python3 ./scripts/test_e2e.py --outages $(E2E_TESTS) + @DFE_E2E_COMPOSE_FILES='$(E2E_COMPOSE_FILES)' $(UV) run --script ./scripts/test_e2e.py --outages $(E2E_TESTS) .PHONY: test-flows test-flows: ## Flow shapes against a running stack (needs DFE_ENGINE_REPO; FLOW_ARGS passes flags) - @python3 ./scripts/test_flows.py $(FLOW_ARGS) + @$(UV) run --script ./scripts/test_flows.py $(FLOW_ARGS) # The post-deploy source test: add a source through the console and prove every # hop. The runner is dfe-infra's, the same one `dfe-ops acceptance --suite diff --git a/docs/developing.md b/docs/developing.md index 4b6d4e8..c0d9415 100644 --- a/docs/developing.md +++ b/docs/developing.md @@ -27,7 +27,7 @@ exist, and how a profile decides which of them run. |---|---| | Docker and Docker Compose v2 | everything | | Python 3 | the helper scripts under `scripts/` | -| PyYAML | `make test-e2e` and `make test-resilience` only (`pip install pyyaml`, or `uv run --with pyyaml`) | +| uv | `make test-e2e`, `make test-resilience` and `make test-flows` only. It installs the PyYAML and Playwright those scripts declare inline | | ruff | `make check-python` only | | pytest | `make check-tests` only (`uvx` fetches the pin if you have it) | | git credentials for the hyperi-io repos (or `DFE_SRC_ROOT` checkouts) | `make dev` | @@ -324,6 +324,10 @@ An archiver that subscribes to the test's landing topic and archives to a since a baseline taken before the send. The archive volume outlives the stack, so files from an earlier run never count. +The data file goes to the ingest edge 16 requests at a time. The receiver answers each request once the loader has flushed it, so one at a time costs a flush per event. + +The suite stops whatever stack the project is running, so it reads the project first. On every way out, a failure or a SIGTERM included, it starts each service that was running or had run to completion again, from the compose files it was started from. A last verdict fails if one comes back on a different config hash. Run it as the user that started the stack: `make` writes that user's UID and GID into the proxies' and the engine's config. + Under [console TLS](configuration.md#console-tls-opt-in) each test's stack chains the fragments `make ci` does, the exposure opt-outs and container logs included. The console checks go to `https://:` and verify its certificate by chain and name, against `DFE_PROXY_CA_BUNDLE` when a private CA signed it. A certificate that does not verify fails the check with the variable to change. Two things about the runner worth knowing before you debug it: @@ -355,7 +359,7 @@ The `core` footprint starts `dfe-hunt-runner`, as `make dev` does. On a profile ### Search to rule to hunt -- the console path ```bash -uv run --with pyyaml --with playwright==1.63.0 python3 scripts/test_e2e.py search-rule-hunt +make test-e2e E2E_TESTS=search-rule-hunt ``` A `detection:` block carrying `from_search` makes its rule the way an analyst does, in a browser. It adds HyperDX to the stack, signs in to the console as the stack's admin and opens Observe search on `main`. The search bar gets the block's `search`, held to the run's marker, and the side panel picks its `filter`. Then Create Rule, and the rule page it opens. @@ -364,7 +368,7 @@ The stored rule has to scan `dfe.main` with no SQL errors and a WHERE that carri The side panel only offers values rows already hold, so both event sets are sent once before the search. The hunt's first window can reach those rows, so their near misses are judged too, and their matches are not. -The console holds every page behind its setup wizard, so the test first makes the `e2e` organisation and one user of the deployment's own, as the wizard would. It needs an interpreter with Playwright and Google Chrome (`playwright install chrome` where Chrome is absent). `DFE_E2E_HEADED=1` shows the browser, and `DFE_E2E_SHOTS_DIR` keeps the step screenshots, which otherwise go to a new temporary directory named in the log. +The console holds every page behind its setup wizard, so the test first makes the `e2e` organisation and one user of the deployment's own, as the wizard would. uv installs Playwright from the script's inline metadata, and the test drives Google Chrome (`uvx playwright install chrome` where it is absent). `DFE_E2E_HEADED=1` shows the browser, and `DFE_E2E_SHOTS_DIR` keeps the step screenshots, which otherwise go to a new temporary directory named in the log. Under console TLS the suite first verifies the console certificate itself, then pins Chrome to that certificate's public key, because Chrome takes no CA bundle. diff --git a/docs/observability.md b/docs/observability.md index 61595f3..d1afdda 100644 --- a/docs/observability.md +++ b/docs/observability.md @@ -209,6 +209,8 @@ Node metrics are still not collected. through the API reaches the running receiver, and a record matching its rule lands in a table of its own. +Once every claim has run, the three ingest rows are deleted: a lightweight `DELETE`, or a mutation (`ALTER TABLE ... DELETE`) on a table whose projections refuse one. A delete that does not finish within 60s is reported, never failed. + The freshness window is what makes the second claim mean "streaming now" rather than "streamed once". The first two are the pair the Kubernetes bootstrap smoke asserts as CORE 1 and CORE 2. diff --git a/docs/troubleshooting.md b/docs/troubleshooting.md index e0af306..141f3c1 100644 --- a/docs/troubleshooting.md +++ b/docs/troubleshooting.md @@ -367,14 +367,13 @@ nothing set and requiring that it fails for the right reason. A green run proves the stack refuses to resolve with nothing set; it does not prove every single image is pinned, because compose aborts on the first missing variable. -### `PyYAML is required and not installed` +### `PyYAML is not installed in this interpreter` -PyYAML is the e2e harness's only third-party dependency, and it is not in the -standard library: +The e2e harness declares PyYAML and Playwright inline (PEP 723), and `make test-e2e` runs it through uv, which installs them. A bare `python3 scripts/test_e2e.py` skips that step: ```bash -pip install pyyaml -uv run --with pyyaml python3 scripts/test_e2e.py # or run it this way +make test-e2e +uv run --script scripts/test_e2e.py # the same, without make ``` The power-on self test (`scripts/post.py`) has no such dependency -- it is diff --git a/scripts/_pipeline.py b/scripts/_pipeline.py index 994ea87..fbf384b 100644 --- a/scripts/_pipeline.py +++ b/scripts/_pipeline.py @@ -26,6 +26,8 @@ import re import time import typing +from concurrent.futures import ThreadPoolExecutor +from http.client import HTTPException from urllib.error import URLError from urllib.request import Request, urlopen @@ -169,6 +171,26 @@ def http_post( )[0] +def post_all( + url: str, bodies: typing.Sequence[str], *, workers: int, timeout: int = 10 +) -> list[int]: + """POST every body from `workers` concurrent senders, returning each status in order. + + The receiver answers once the next hop holds the record, so one request at a + time waits out the loader's flush on every event. A request that got no HTTP + answer at all reports status 0. + """ + + def _send(body: str) -> int: + try: + return http_post(url, body, timeout=timeout) + except (HTTPException, OSError): + return 0 + + with ThreadPoolExecutor(max_workers=max(1, workers)) as pool: + return list(pool.map(_send, bodies)) + + def http_delete(url: str, token: str = "", timeout: int = 10) -> int: """DELETE a resource and return the HTTP status. @@ -200,6 +222,26 @@ def ch_query(sql: str, *, debug: typing.Callable[[str], None] | None = None) -> return "" +def ch_execute(sql: str, *, timeout: int = 10) -> tuple[bool, str]: + """Run a statement over ClickHouse's HTTP interface, keeping the server's answer. + + Returns (True, body) on success and (False, the error text) otherwise, for a + caller that has to act on WHY a statement was refused. + """ + request = Request(clickhouse_url(), data=sql.encode(), method="POST") + request.add_header("X-ClickHouse-User", env_or("CLICKHOUSE_USERNAME", "default")) + request.add_header("X-ClickHouse-Key", os.environ.get("CLICKHOUSE_PASSWORD", "")) + try: + with urlopen(request, timeout=timeout) as response: + return True, response.read().decode().strip() + except URLError as error: + if hasattr(error, "read"): + return False, error.read().decode(errors="replace").strip() + return False, str(error.reason) + except OSError as error: + return False, str(error) + + def ch_count(database: str, table: str, *, debug=None) -> int: """Return the row count of a table, or 0 if it cannot be read.""" count = ch_int(f"SELECT count() FROM {database}.{table}", debug=debug) diff --git a/scripts/_search_rule.py b/scripts/_search_rule.py index fbfeb11..c52a99d 100644 --- a/scripts/_search_rule.py +++ b/scripts/_search_rule.py @@ -59,9 +59,9 @@ # An option in whichever console select is open. OPEN_OPTION = ".ant-select-dropdown:not(.ant-select-dropdown-hidden) .ant-select-item-option-content" PLAYWRIGHT_HINT = ( - "run the suite under an interpreter carrying Playwright, e.g. " - "`uv run --with pyyaml --with playwright python3 scripts/test_e2e.py`, and " - "`playwright install chrome` where Google Chrome is not installed" + "run the suite through uv, which installs the Playwright it declares " + "(`make test-e2e`, or `uv run --script scripts/test_e2e.py`), and install " + "Google Chrome where it is absent (`uvx playwright install chrome`)" ) diff --git a/scripts/_stack_state.py b/scripts/_stack_state.py new file mode 100644 index 0000000..91f1a71 --- /dev/null +++ b/scripts/_stack_state.py @@ -0,0 +1,165 @@ +# Project: dfe-docker +# File: _stack_state.py +# Purpose: The containers a compose project holds, and the commands that start them again +# Language: Python +# +# License: BUSL-1.1 +# Copyright: (c) 2026 HYPERI PTY LIMITED + +"""What a compose project was running before the e2e suite, read off `docker inspect`. + +Internal support module - imported by the e2e suite, not executed directly. The +suite stops every service to run one stack per test, so it reads the project +first and starts the same services from the same compose files at the end. Compose +labels each container with the files, directory and project it came from, and +with a hash of its resolved config, which says whether the container that comes +back is configured as the one that went. +""" + +from collections import defaultdict +from collections.abc import Iterable, Mapping +from dataclasses import dataclass + +SERVICE_LABEL = "com.docker.compose.service" +PROJECT_LABEL = "com.docker.compose.project" +CONFIG_FILES_LABEL = "com.docker.compose.project.config_files" +WORKING_DIR_LABEL = "com.docker.compose.project.working_dir" +ENVIRONMENT_FILE_LABEL = "com.docker.compose.project.environment_file" +CONFIG_HASH_LABEL = "com.docker.compose.config-hash" +ONEOFF_LABEL = "com.docker.compose.oneoff" + + +@dataclass(frozen=True, slots=True) +class Container: + """One service container, as compose labelled it and docker reports it. + + Attributes: + service: The compose service it runs. + running: Whether its process is running. + exit_code: The code its process last exited with. + restart: Its restart policy, empty when none was set. + config_hash: The hash compose took of the service's resolved config. + project: The compose project it belongs to. + working_dir: The project directory compose resolved relative paths against. + config_files: The compose files it was started from, in order. + environment_files: The `--env-file` arguments it was started with. + """ + + service: str + running: bool + exit_code: int + restart: str + config_hash: str + project: str + working_dir: str + config_files: tuple[str, ...] + environment_files: tuple[str, ...] + + @property + def run_to_completion(self) -> bool: + """A one-shot that finished cleanly, which `up` runs to completion again.""" + return not (self.running) and self.exit_code == 0 and self.restart in ("", "no") + + @property + def restored(self) -> bool: + """Started again at the end: running, or a one-shot that finished cleanly.""" + return self.running or self.run_to_completion + + +def _split(value: str) -> tuple[str, ...]: + return tuple(part for part in value.split(",") if part) + + +def containers(documents: Iterable[Mapping]) -> list[Container]: + """Every service container in `docker inspect` output, `docker compose run` ones left out.""" + found = [] + for document in documents: + labels = (document.get("Config") or {}).get("Labels") or {} + if not (labels.get(SERVICE_LABEL)) or labels.get(ONEOFF_LABEL) == "True": + continue + state = document.get("State") or {} + policy = (document.get("HostConfig") or {}).get("RestartPolicy") or {} + found.append( + Container( + service=labels[SERVICE_LABEL], + running=bool(state.get("Running")), + exit_code=int(state.get("ExitCode") or 0), + restart=policy.get("Name") or "", + config_hash=labels.get(CONFIG_HASH_LABEL, ""), + project=labels.get(PROJECT_LABEL, ""), + working_dir=labels.get(WORKING_DIR_LABEL, ""), + config_files=_split(labels.get(CONFIG_FILES_LABEL, "")), + environment_files=_split(labels.get(ENVIRONMENT_FILE_LABEL, "")), + ) + ) + return sorted(found, key=lambda container: container.service) + + +@dataclass(frozen=True, slots=True) +class Restore: + """One `docker compose up` that starts services from the files they came from.""" + + project: str + working_dir: str + config_files: tuple[str, ...] + environment_files: tuple[str, ...] + services: tuple[str, ...] + + def command(self) -> list[str]: + """The compose command, every profile enabled so each service resolves.""" + command = ["docker", "compose", "--project-name", self.project] + command += ["--project-directory", self.working_dir] + for path in self.environment_files: + command += ["--env-file", path] + for path in self.config_files: + command += ["-f", path] + return command + ["--profile", "*", "up", "-d", *self.services] + + +def restores(found: Iterable[Container]) -> list[Restore]: + """One `up` per set of compose files the restored containers were started from.""" + groups: dict[tuple, list[str]] = defaultdict(list) + for container in found: + if container.restored: + key = ( + container.project, + container.working_dir, + container.config_files, + container.environment_files, + ) + groups[key].append(container.service) + return [ + Restore(project, working_dir, files, env_files, tuple(sorted(services))) + for (project, working_dir, files, env_files), services in sorted(groups.items()) + ] + + +def differences(before: Iterable[Container], after: Iterable[Container]) -> list[str]: + """How the project differs from how it was found, empty when it does not. + + Each restored service must have a container configured as before, still + running if it was. Nothing may be running that was not. + """ + before = list(before) + now = {container.service: container for container in after} + problems = [] + for was in before: + if not (was.restored): + continue + current = now.get(was.service) + if current is None: + problems.append(f"'{was.service}' has no container") + elif was.running and not (current.running): + problems.append(f"'{was.service}' is not running") + elif current.config_hash != was.config_hash: + problems.append( + f"'{was.service}' is configured differently (config hash " + f"{was.config_hash[:12]} became {current.config_hash[:12]})" + ) + was_running = {container.service for container in before if container.running} + problems += [ + f"'{service}' is running and was not" + for service, current in sorted(now.items()) + if current.running and service not in was_running + ] + return problems diff --git a/scripts/post.py b/scripts/post.py index d730271..d684202 100644 --- a/scripts/post.py +++ b/scripts/post.py @@ -89,6 +89,7 @@ MARKER_EXPRESSIONS, _decoded, ch_count, + ch_execute, ch_int, ch_marker_count, ch_query, @@ -116,6 +117,15 @@ LAND_TIMEOUT_SECONDS = 60.0 LAND_INTERVAL_SECONDS = 3.0 +# Both deletes run asynchronously and the rows are polled until gone, so a large +# table cannot hold a request past its timeout. +CLEANUP_TIMEOUT_SECONDS = 60.0 +CLEANUP_INTERVAL_SECONDS = 2.0 +CLEANUP_STATEMENT_TIMEOUT_SECONDS = 30 +# How ClickHouse refuses a lightweight DELETE on a table with projections, under a +# table-level lightweight_mutation_projection_mode that no query setting overrides. +PROJECTION_REFUSAL = ("Code: 344", "projection") + # How long to wait for the ingest component to report ready. `make dev`/`make ci` # do not pass `--wait` to `docker compose up`, so this runs while containers are # still warming; a single probe would decide "not ready" a second after boot. @@ -749,28 +759,89 @@ def _run_id() -> str: return f"post-{os.getpid()}-{os.urandom(4).hex()}" -def _cleanup(database: str, table: str, marker: str) -> None: - """Report that this run's synthetic rows remain. It cannot remove them. +def _marker_predicate(database: str, table: str, marker: str) -> str | None: + """The WHERE clause that finds this run's rows, or None when none can be found.""" + escaped = escape_literal(marker) + for expression in MARKER_EXPRESSIONS: + where = f"{expression} = '{escaped}'" + if (ch_int(f"SELECT count() FROM {database}.{table} WHERE {where}") or 0) > 0: + return where + return None + + +def _refused_for_projections(answer: str) -> bool: + """Whether ClickHouse refused a lightweight DELETE because the table has projections.""" + return all(part in answer for part in PROJECTION_REFUSAL) + + +def _mutation_failure(database: str, table: str, marker: str) -> str: + """The last failure of this run's unfinished delete mutation, or empty.""" + return ch_query( + "SELECT latest_fail_reason FROM system.mutations " + f"WHERE database = '{escape_literal(database)}' " + f"AND table = '{escape_literal(table)}' AND NOT is_done " + f"AND position(command, '{escape_literal(marker)}') > 0 LIMIT 1" + ) - We tried deleting them and it does not work: the engine-provisioned - `dfe.main` carries PROJECTIONS, and ClickHouse refuses a lightweight - DELETE on such a table unless `lightweight_mutation_projection_mode` is - changed: - Code: 344. DELETE query is not allowed for table dfe.main because - as it has projections and setting lightweight_mutation_projection_mode - is set to THROW. +def _cleanup(database: str, table: str, marker: str) -> None: + """Delete this run's synthetic rows, by mutation where a lightweight DELETE is refused. - Changing a server-level mutation setting to tidy up after a self test is a - worse trade than leaving three rows, so it says so instead. The rows are - identifiable (`post--` in `_tags.marker`) if you want them gone. + A table the engine created before it stopped adding the `_timestamp_optimized` + projection still carries it, and ClickHouse refuses a lightweight DELETE there + (Code 344) while the table's `lightweight_mutation_projection_mode` is `throw`. + A mutation (`ALTER TABLE ... DELETE`) rebuilds the projection with the parts it + rewrites, so it is the fallback, and the table's setting is left alone. - Writing to the real landing table is deliberate: a dedicated table would - prove a different pipeline than the one being asserted. + Tidy-up only: what it could not remove is reported, never failed. Writing to + the real landing table is deliberate, because a dedicated table would prove a + different pipeline than the one being asserted. """ + where = _marker_predicate(database, table, marker) + if where is None: + _print( + msg=f" note: no row tagged {marker} in {database}.{table} to remove" + ) + return + + how = "lightweight DELETE" + deleted, answer = ch_execute( + f"DELETE FROM {database}.{table} WHERE {where} " + "SETTINGS lightweight_deletes_sync = 0", + timeout=CLEANUP_STATEMENT_TIMEOUT_SECONDS, + ) + if not (deleted) and _refused_for_projections(answer): + how = "mutation, as the table's projections refuse a lightweight DELETE" + deleted, answer = ch_execute( + f"ALTER TABLE {database}.{table} DELETE WHERE {where} " + "SETTINGS mutations_sync = 0", + timeout=CLEANUP_STATEMENT_TIMEOUT_SECONDS, + ) + if not (deleted): + refusal = answer.splitlines()[0] if answer else "no answer" + _print( + msg=f" note: {EVENT_COUNT} synthetic row(s) remain in {database}.{table} " + f"tagged {marker} -- the {how} was refused: {refusal}" + ) + return + + left = poll_until( + lambda: ch_int(f"SELECT count() FROM {database}.{table} WHERE {where}"), + timeout=CLEANUP_TIMEOUT_SECONDS, + interval=CLEANUP_INTERVAL_SECONDS, + done=lambda count: count == 0, + ) + if left == 0: + _print( + msg=f" removed the synthetic rows tagged {marker} from " + f"{database}.{table} by {how}" + ) + return + failure = _mutation_failure(database, table, marker) _print( - msg=f" note: {EVENT_COUNT} synthetic row(s) remain in {database}.{table} " - f"tagged {marker} -- lightweight DELETE is refused on a table with projections" + msg=f" note: {left} synthetic row(s) tagged {marker} remain in " + f"{database}.{table} {CLEANUP_TIMEOUT_SECONDS:.0f}s after the {how}" + + (f": {failure}" if failure else " -- system.mutations carries its progress") ) @@ -2270,25 +2341,23 @@ def _report(attempt, result): _print( msg=f"PASS {matched}/{EVENT_COUNT} marked event(s) landed in {database}.{table}" ) - _cleanup(database, table, marker) # Every remaining claim runs even when an earlier one fails, so one boot # reports every broken pipeline rather than the first one. - return _report_claims( - claims={ - "ingest": HELD, - "hunts": _verify_hunt( - database=database, ingest_url=ingest_url, marker=marker, table=table - ), - "subscription": _verify_loader_subscription(), - "self-monitoring": _verify_self_monitoring(), - "observability": _verify_hyperdx(), - "console": _verify_ui_query( - database=database, marker=marker, table=table - ), - "idle apps": _verify_idle_apps(), - "routing": _verify_routing_applied(database=database), - } - ) + claims = { + "ingest": HELD, + "hunts": _verify_hunt( + database=database, ingest_url=ingest_url, marker=marker, table=table + ), + "subscription": _verify_loader_subscription(), + "self-monitoring": _verify_self_monitoring(), + "observability": _verify_hyperdx(), + "console": _verify_ui_query(database=database, marker=marker, table=table), + "idle apps": _verify_idle_apps(), + "routing": _verify_routing_applied(database=database), + } + # After the console claim, which reads these same rows back. + _cleanup(database, table, marker) + return _report_claims(claims=claims) _print( msg=f"FAIL only {matched}/{EVENT_COUNT} marked event(s) reached " diff --git a/scripts/test_e2e.py b/scripts/test_e2e.py index c593906..b54a6fd 100755 --- a/scripts/test_e2e.py +++ b/scripts/test_e2e.py @@ -1,4 +1,4 @@ -#!/usr/bin/env python3 +#!/usr/bin/env -S uv run --script # Project: dfe-docker # File: scripts/test-e2e.py @@ -16,12 +16,21 @@ # TEST_CONFIG=tests/e2e/e2e-tests.yaml ./scripts/test-e2e.py # Custom config # LOG_LEVEL=debug ./scripts/test-e2e.py # Verbose service logs +# /// script +# requires-python = ">=3.12" +# dependencies = [ +# "playwright==1.63.0", +# "pyyaml==6.0.3", +# ] +# /// + import argparse import functools import json import logging import os import secrets +import signal import ssl import subprocess import sys @@ -31,15 +40,14 @@ try: import yaml except ModuleNotFoundError: # pragma: no cover - environment guard - # PyYAML is the harness's only third-party dependency and it is NOT in the - # stdlib, so a plain `make test-e2e` on a fresh machine dies with a bare - # ModuleNotFoundError that says nothing about how to proceed. Every other - # missing prerequisite here (docker) gets an actionable message; this one - # should too. + # An interpreter that skipped the inline metadata above, such as a bare + # `python3 scripts/test_e2e.py`, otherwise dies with a ModuleNotFoundError + # that does not say how to proceed. sys.exit( - "test_e2e: PyYAML is required and not installed.\n" - " Install it: pip install pyyaml\n" - " Or run via uv: uv run --with pyyaml python3 scripts/test_e2e.py" + "test_e2e: PyYAML is not installed in this interpreter.\n" + " Run it through uv, which installs what the script declares:\n" + " make test-e2e\n" + " uv run --script scripts/test_e2e.py" ) from collections import Counter @@ -55,6 +63,7 @@ import _detection import _outage import _search_rule +import _stack_state import post from _common import FALSY, _config_argument, _load_dotenv, _use_mounted_configs from _pipeline import MARKER_EXPRESSIONS # noqa: F401 - re-exported for callers @@ -67,6 +76,7 @@ expand_env, otel_fresh_counts, poll_until, + post_all, ) @@ -151,6 +161,11 @@ ARCHIVE_INTERVAL_SECONDS = 10 TARGET_DB = "dfe" TARGET_TABLE = "main" +# The receiver answers each request once the loader has flushed it, which one +# sender at a time waits out per event, so a data file goes in this many at once. +SEND_WORKERS = 16 +# How long the services the run found have, together, to come back healthy. +RESTORE_TIMEOUT_SECONDS = 300 # Marker lookup lives in _pipeline, shared with the power-on self test. @@ -163,7 +178,10 @@ # - Every rendered-config variable is then blanked. `make` exports the engine's # render paths, and compose inherits them unless each service is sent back to # the config this suite mounts for it. +# - FOUND_ENVIRON keeps the environment as the run received it, the one the +# services it found are started again under. # ------------------------------------------------------------------------------ +FOUND_ENVIRON = dict(os.environ) _load_dotenv() _use_mounted_configs(environ=os.environ) @@ -1031,6 +1049,101 @@ def container_id(service): return ids[0] if (result.returncode == 0 and ids) else "" +# ------------------------------------------------------------------------------ +# Project Containers +# - Every service container in THIS compose project, stopped or not, as +# _stack_state reads them, or None when docker could not say +# ------------------------------------------------------------------------------ +def project_containers(): + listed = run_cmd( + ["docker", "compose", "--profile", "*", "ps", "--all", "--quiet"], capture=True + ) + if listed.returncode != 0: + LOGGER.error(f"Could not list this project's containers: {listed.stderr}") + return None + ids = (listed.stdout or "").split() + if not (ids): + return [] + inspected = run_cmd(["docker", "inspect", *ids], capture=True) + try: + return _stack_state.containers(json.loads(inspected.stdout or "[]")) + except ValueError: + LOGGER.error(f"Could not inspect this project's containers: {inspected.stderr}") + return None + + +# ------------------------------------------------------------------------------ +# Wait Settled +# - Polls until one service is back as it was found: running and healthy, or a +# one-shot that has finished, or the deadline passes +# ------------------------------------------------------------------------------ +def wait_settled(container, deadline): + while True: + state = container_state(container.service) + if state and container.running: + if state.status == "running" and state.health in ("healthy", ""): + return + elif state and state.status != "running": + return + if time.monotonic() >= deadline: + return + time.sleep(2) + + +# ------------------------------------------------------------------------------ +# Put Back +# - Leaves the project as the run found it: the test stack down, then each service +# that was running, or had run to completion, started again from the compose +# files, project and environment it was started with, and checked against the +# config hash compose recorded for it +# ------------------------------------------------------------------------------ +def put_back(ctx, found): + print() + stack_down() + + for restore in _stack_state.restores(found): + LOGGER.info( + f"Starting the {len(restore.services)} service(s) this run found, from " + f"{len(restore.config_files)} compose file(s)..." + ) + result = subprocess.run( + restore.command(), + cwd=restore.working_dir or PROJECT_DIR, + env=FOUND_ENVIRON, + capture_output=True, + text=True, + encoding="utf-8", + errors="replace", + ) + if result.returncode != 0: + detail = (result.stderr or result.stdout or "").strip() + mark_fail(ctx, f"the services this run found did not start again: {detail}") + + deadline = time.monotonic() + RESTORE_TIMEOUT_SECONDS + for container in found: + if container.restored: + wait_settled(container, deadline) + + after = project_containers() + problems = ( + ["the project could not be read back"] + if (after is None) + else _stack_state.differences(found, after) + ) + running = sum(1 for container in found if container.running) + completed = sum(1 for container in found if container.run_to_completion) + if problems: + mark_fail(ctx, f"the stack is not as this run found it: {'; '.join(problems)}") + elif running or completed: + mark_pass( + ctx, + f"the stack is as this run found it: {running} service(s) running and " + f"{completed} one-shot(s) completed, each on the config it had", + ) + else: + mark_pass(ctx, "this run found no stack running and leaves none") + + # ------------------------------------------------------------------------------ # Report Unready Service # - Dumps what a timed-out service was actually doing, before teardown removes it @@ -1189,28 +1302,26 @@ def send_events(ctx, test_name, marker, data_file_name, database, table, ingest_ events = data_file_path.read_text(encoding="utf-8").splitlines() num_events = len(events) - LOGGER.info(f"Sending {num_events} events to '{ingest_url}'...") + LOGGER.info( + f"Sending {num_events} events to '{ingest_url}' ({SEND_WORKERS} at a time)..." + ) LOGGER.debug(f"Marker: '{marker}'") - ctx.total_sent = 0 - send_errors = 0 - + bodies = [] for line in events: event = json.loads(line) event["_source"] = table event["_tags"] = {"marker": marker, "test_name": test_name} - enriched = json.dumps(event, separators=(",", ":")) - - try: - status = http_post(ingest_url, enriched) - except Exception: - status = 0 + bodies.append(json.dumps(event, separators=(",", ":"))) + send_errors = 0 + for body, status in zip( + bodies, post_all(ingest_url, bodies, workers=SEND_WORKERS), strict=True + ): if status < 200 or status >= 300: - LOGGER.debug(f"Response code '{status}' received for '{enriched}'") + LOGGER.debug(f"Response code '{status}' received for '{body}'") send_errors += 1 - - ctx.total_sent += 1 + ctx.total_sent = len(bodies) LOGGER.info(f"Sent {ctx.total_sent} events (errors = {send_errors})") @@ -3346,6 +3457,30 @@ def run_test(ctx, mode, test, persistent_services): print("------------------------------------------------------------") +# ------------------------------------------------------------------------------ +# Run Suite +# - Reads the project, stops it, builds and runs every test, then puts the project +# back as it was found on every way out, a SIGTERM included +# ------------------------------------------------------------------------------ +def run_suite(ctx, mode, tests, persistent_services): + found = project_containers() + if found is None: + error("the running stack could not be read, so nothing was stopped") + signal.signal(signal.SIGTERM, lambda signum, frame: sys.exit(128 + signum)) + try: + if stack_is_up(): + print() + stack_down(persistent_services) + + # Build only the DFE services the selected tests actually need. + build_images(mode, {service for test in tests for service in test.services}) + + for test in tests: + run_test(ctx, mode, test, persistent_services) + finally: + put_back(ctx, found) + + # ============================================================================== # Miscellaneous Functions # ============================================================================== @@ -3462,12 +3597,7 @@ def main(): global_config = config.get("global", {}).copy() global_config["clickhouse"] = config.get("clickhouse", {}) - # Tear down any existing stack before starting persistent_services = get_config("persistent_services", {}, global_config) - if stack_is_up(): - print() - stack_down(persistent_services) - mode = get_config("mode", {}, global_config) # Resolve which tests to run - CLI args filter, otherwise every test of the @@ -3496,16 +3626,8 @@ def main(): if bool(test_config.get("outage")) == args.outages: tests_to_run.append(resolve_test_case(test_config, global_config)) - # Build only the DFE services the selected tests actually need. - all_services = set() - for test in tests_to_run: - all_services.update(test.services.keys()) - - build_images(mode, all_services) - - # Run resolved tests - for test in tests_to_run: - run_test(ctx, mode, test, persistent_services) + if tests_to_run: + run_suite(ctx, mode, tests_to_run, persistent_services) # Post test execution cleanup cleanup(global_config.get("clickhouse", {})) diff --git a/scripts/test_flows.py b/scripts/test_flows.py index 1f47cfd..d46a713 100644 --- a/scripts/test_flows.py +++ b/scripts/test_flows.py @@ -14,6 +14,13 @@ # ./scripts/test_flows.py --engine-repo ../dfe-engine # DFE_ENGINE_REPO=../dfe-engine ./scripts/test_flows.py +# /// script +# requires-python = ">=3.12" +# dependencies = [ +# "pyyaml==6.0.3", +# ] +# /// + """The flow shapes, on the compose stack, over localhost ports. The fixtures and the assertions live in dfe-engine (``tests/e2e/flows``) and are diff --git a/scripts/tests/test_ingest_send.py b/scripts/tests/test_ingest_send.py new file mode 100644 index 0000000..c782060 --- /dev/null +++ b/scripts/tests/test_ingest_send.py @@ -0,0 +1,83 @@ +# Project: dfe-docker +# File: tests/test_ingest_send.py +# Purpose: Assert the e2e suite's data file is sent concurrently, each answer kept +# Language: Python +# +# License: BUSL-1.1 +# Copyright: (c) 2026 HYPERI PTY LIMITED + +"""The receiver holds each answer until the next hop has the record. + +A stand-in receiver that holds every request for a fixed time shows the cost: +one sender at a time pays the hold per event, concurrent senders share it. +""" + +import threading +import time +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer + +import pytest + +import _pipeline + +HOLD_SECONDS = 0.2 +EVENTS = 10 + + +class _HoldingReceiver(BaseHTTPRequestHandler): + """Answers each POST after HOLD_SECONDS, 503 for a body asking to be refused.""" + + def do_POST(self) -> None: + body = self.rfile.read(int(self.headers["Content-Length"])) + time.sleep(HOLD_SECONDS) + self.send_response(503 if b"refuse" in body else 200) + self.end_headers() + + def log_message(self, *args) -> None: + return + + +@pytest.fixture +def receiver_url(): + server = ThreadingHTTPServer(("127.0.0.1", 0), _HoldingReceiver) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + yield f"http://127.0.0.1:{server.server_address[1]}/ingest" + finally: + server.shutdown() + server.server_close() + + +def _timed(url: str, workers: int) -> tuple[list[int], float]: + bodies = [f'{{"n": {n}}}' for n in range(EVENTS)] + started = time.monotonic() + statuses = _pipeline.post_all(url, bodies, workers=workers) + return statuses, time.monotonic() - started + + +def test_one_sender_at_a_time_pays_the_hold_per_event(receiver_url: str) -> None: + statuses, elapsed = _timed(receiver_url, workers=1) + + assert statuses == [200] * EVENTS + assert elapsed >= EVENTS * HOLD_SECONDS + + +def test_concurrent_senders_share_the_hold(receiver_url: str) -> None: + statuses, elapsed = _timed(receiver_url, workers=EVENTS) + + assert statuses == [200] * EVENTS + assert elapsed < EVENTS * HOLD_SECONDS / 2 + + +def test_each_answer_stays_with_its_own_request(receiver_url: str) -> None: + bodies = ['{"n": 0}', '{"refuse": 1}', '{"n": 2}'] + + assert _pipeline.post_all(receiver_url, bodies, workers=3) == [200, 503, 200] + + +def test_a_request_nothing_answers_reports_status_zero() -> None: + with ThreadingHTTPServer(("127.0.0.1", 0), _HoldingReceiver) as closed: + port = closed.server_address[1] + + assert _pipeline.post_all(f"http://127.0.0.1:{port}/", ["{}"], workers=1) == [0] diff --git a/scripts/tests/test_post.py b/scripts/tests/test_post.py index 6fd7e12..b90c770 100644 --- a/scripts/tests/test_post.py +++ b/scripts/tests/test_post.py @@ -938,3 +938,123 @@ def test_a_stack_without_the_runner_needs_no_runner_password( monkeypatch.delenv(_RUNNER_PASSWORD_KEY, raising=False) assert post._weak_secrets() == [] + + +# ClickHouse 26.3's answer to a lightweight DELETE on a table with a projection. +_PROJECTION_REFUSAL = ( + "Code: 344. DB::Exception: DELETE query is not allowed for table dfe.main " + "because as it has projections and setting lightweight_mutation_projection_mode " + "is set to THROW. User should change lightweight_mutation_projection_mode OR " + "drop all the projections manually before running the query. " + "(SUPPORT_IS_DISABLED) (version 26.3.42.3 (official build))" +) + + +class _ClickHouse: + """A stand-in ClickHouse whose marked rows go once a delete is accepted and lands.""" + + def __init__( + self, *, refuse_lightweight: str = "", rows: int = 3, deletes_land: bool = True + ) -> None: + self.refuse_lightweight = refuse_lightweight + self.rows = rows + self.deletes_land = deletes_land + self.statements: list[str] = [] + + def execute(self, sql: str, *, timeout: int) -> tuple[bool, str]: + self.statements.append(sql) + if sql.startswith("DELETE") and self.refuse_lightweight: + return False, self.refuse_lightweight + if self.deletes_land: + self.rows = 0 + return True, "" + + def count(self, sql: str, **kwargs) -> int: + return self.rows + + +def _stub_clickhouse(monkeypatch: pytest.MonkeyPatch, clickhouse: _ClickHouse) -> None: + monkeypatch.setattr(post, "ch_execute", clickhouse.execute) + monkeypatch.setattr(post, "ch_int", clickhouse.count) + monkeypatch.setattr(post, "ch_query", lambda sql: "") + monkeypatch.setattr(post, "CLEANUP_TIMEOUT_SECONDS", 0.0) + + +def test_a_table_with_projections_is_cleaned_by_a_mutation( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture +) -> None: + clickhouse = _ClickHouse(refuse_lightweight=_PROJECTION_REFUSAL) + _stub_clickhouse(monkeypatch, clickhouse) + + post._cleanup("dfe", "main", "post-1-abcd") + + assert [sql.split()[0] for sql in clickhouse.statements] == ["DELETE", "ALTER"] + assert clickhouse.statements[1].startswith("ALTER TABLE dfe.main DELETE WHERE ") + assert "'post-1-abcd'" in clickhouse.statements[1] + assert "by mutation" in capsys.readouterr().err + + +def test_a_table_without_projections_takes_the_lightweight_delete( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture +) -> None: + clickhouse = _ClickHouse() + _stub_clickhouse(monkeypatch, clickhouse) + + post._cleanup("dfe", "main", "post-1-abcd") + + assert len(clickhouse.statements) == 1 + assert clickhouse.statements[0].startswith("DELETE FROM dfe.main WHERE ") + assert "by lightweight DELETE" in capsys.readouterr().err + + +def test_any_other_refusal_is_reported_and_not_retried_as_a_mutation( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture +) -> None: + clickhouse = _ClickHouse( + refuse_lightweight="Code: 497. DB::Exception: default: Not enough privileges." + ) + _stub_clickhouse(monkeypatch, clickhouse) + + post._cleanup("dfe", "main", "post-1-abcd") + + assert len(clickhouse.statements) == 1 + assert "Not enough privileges" in capsys.readouterr().err + + +def test_rows_still_there_at_the_deadline_are_reported( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture +) -> None: + clickhouse = _ClickHouse(refuse_lightweight=_PROJECTION_REFUSAL, deletes_land=False) + _stub_clickhouse(monkeypatch, clickhouse) + + post._cleanup("dfe", "main", "post-1-abcd") + + assert "3 synthetic row(s) tagged post-1-abcd remain" in capsys.readouterr().err + + +def test_no_marked_row_means_nothing_is_deleted( + monkeypatch: pytest.MonkeyPatch, +) -> None: + clickhouse = _ClickHouse(rows=0) + _stub_clickhouse(monkeypatch, clickhouse) + + post._cleanup("dfe", "main", "post-1-abcd") + + assert clickhouse.statements == [] + + +def test_the_rows_are_deleted_after_the_console_reads_them_back( + monkeypatch: pytest.MonkeyPatch, +) -> None: + calls = _stub_claims(monkeypatch) + _stub_main(monkeypatch) + monkeypatch.setattr( + post, + "_ingest_target", + lambda *, table: ("dfe-receiver", "http://ingest/ingest"), + ) + monkeypatch.setattr(post, "_cleanup", lambda *args: calls.append("_cleanup")) + + assert post.main() == 0 + assert calls[-1] == "_cleanup" + assert calls.index("_verify_ui_query") < calls.index("_cleanup") diff --git a/scripts/tests/test_stack_state.py b/scripts/tests/test_stack_state.py new file mode 100644 index 0000000..fbc4c78 --- /dev/null +++ b/scripts/tests/test_stack_state.py @@ -0,0 +1,155 @@ +# Project: dfe-docker +# File: tests/test_stack_state.py +# Purpose: Assert how the e2e suite reads a stack and plans to start it again +# Language: Python +# +# License: BUSL-1.1 +# Copyright: (c) 2026 HYPERI PTY LIMITED + +"""The e2e suite stops every service, so it must put back exactly what it found. + +Driven from `docker inspect` documents shaped as docker writes them: which +containers come back, from which compose files, and when the project counts as +different from how it was found. +""" + +import pytest + +import _stack_state + +_FILES = "/opt/dfe/docker-compose.yml,/opt/dfe/docker-compose.tls.yml" + + +def _document( + service: str, + *, + running: bool = True, + exit_code: int = 0, + restart: str = "unless-stopped", + config_hash: str = "aaaa", + files: str = _FILES, + oneoff: str = "False", +) -> dict: + return { + "Config": { + "Labels": { + _stack_state.SERVICE_LABEL: service, + _stack_state.PROJECT_LABEL: "dfe-docker", + _stack_state.WORKING_DIR_LABEL: "/opt/dfe", + _stack_state.CONFIG_FILES_LABEL: files, + _stack_state.CONFIG_HASH_LABEL: config_hash, + _stack_state.ONEOFF_LABEL: oneoff, + } + }, + "State": {"Running": running, "ExitCode": exit_code}, + "HostConfig": {"RestartPolicy": {"Name": restart}}, + } + + +def test_a_running_service_and_a_finished_one_shot_both_come_back() -> None: + found = _stack_state.containers( + [ + _document("dfe-engine"), + _document("dlq-init", running=False, restart="no"), + ] + ) + + [restore] = _stack_state.restores(found) + + assert restore.services == ("dfe-engine", "dlq-init") + + +@pytest.mark.parametrize( + "document", + [ + _document("dfe-loader", running=False), + _document("dlq-init", running=False, exit_code=1, restart="no"), + _document("dfe-engine", oneoff="True"), + ], + ids=["stopped-service", "failed-one-shot", "compose-run"], +) +def test_what_was_not_running_or_finished_cleanly_stays_down(document: dict) -> None: + assert _stack_state.restores(_stack_state.containers([document])) == [] + + +def test_the_restore_names_the_project_directory_and_files_it_came_from() -> None: + [restore] = _stack_state.restores( + _stack_state.containers([_document("dfe-engine")]) + ) + + assert restore.command() == [ + "docker", + "compose", + "--project-name", + "dfe-docker", + "--project-directory", + "/opt/dfe", + "-f", + "/opt/dfe/docker-compose.yml", + "-f", + "/opt/dfe/docker-compose.tls.yml", + "--profile", + "*", + "up", + "-d", + "dfe-engine", + ] + + +def test_services_started_from_different_files_get_one_up_each() -> None: + found = _stack_state.containers( + [ + _document("dfe-engine"), + _document("dfe-ui", files="/opt/dfe/docker-compose.yml"), + ] + ) + + assert [restore.services for restore in _stack_state.restores(found)] == [ + ("dfe-ui",), + ("dfe-engine",), + ] + + +def test_a_project_put_back_as_found_shows_no_difference() -> None: + found = _stack_state.containers( + [_document("dfe-engine"), _document("dlq-init", running=False, restart="no")] + ) + + assert _stack_state.differences(found, found) == [] + + +def test_found_nothing_and_left_nothing_shows_no_difference() -> None: + assert _stack_state.differences([], []) == [] + + +@pytest.mark.parametrize( + ("after", "problem"), + [ + ([], "'dfe-engine' has no container"), + ([_document("dfe-engine", running=False)], "'dfe-engine' is not running"), + ( + [_document("dfe-engine", config_hash="bbbb")], + "'dfe-engine' is configured differently", + ), + ( + [_document("dfe-engine"), _document("clickhouse")], + "'clickhouse' is running and was not", + ), + ], + ids=["gone", "stopped", "reconfigured", "extra"], +) +def test_each_way_a_project_can_differ_is_named(after: list, problem: str) -> None: + found = _stack_state.containers([_document("dfe-engine")]) + + problems = _stack_state.differences(found, _stack_state.containers(after)) + + assert len(problems) == 1 + assert problems[0].startswith(problem) + + +def test_a_stack_left_up_after_finding_none_is_a_difference() -> None: + problems = _stack_state.differences( + [], _stack_state.containers([_document("clickhouse")]) + ) + + assert problems == ["'clickhouse' is running and was not"] From eff92542270ac4ef74582659f1e7b24a30f2a123 Mon Sep 17 00:00:00 2001 From: Derek Date: Sun, 11 Oct 2026 13:20:13 +1100 Subject: [PATCH 2/5] fix: restore without compose files since deleted A container an older e2e run started is labelled with that run's override under `.tmp/`, which the run deletes when it finishes. Naming it in the restore made compose refuse the whole `up`, so the next run failed to put that container back. A file that no longer exists is now left out and named in the log, and the config hash check still decides whether the container came back as it was. --- scripts/_stack_state.py | 30 +++++++++++++++++++++++++----- scripts/test_e2e.py | 6 +++++- scripts/tests/test_stack_state.py | 12 ++++++++++++ 3 files changed, 42 insertions(+), 6 deletions(-) diff --git a/scripts/_stack_state.py b/scripts/_stack_state.py index 91f1a71..5e9bace 100644 --- a/scripts/_stack_state.py +++ b/scripts/_stack_state.py @@ -17,7 +17,7 @@ """ from collections import defaultdict -from collections.abc import Iterable, Mapping +from collections.abc import Callable, Iterable, Mapping from dataclasses import dataclass SERVICE_LABEL = "com.docker.compose.service" @@ -97,13 +97,20 @@ def containers(documents: Iterable[Mapping]) -> list[Container]: @dataclass(frozen=True, slots=True) class Restore: - """One `docker compose up` that starts services from the files they came from.""" + """One `docker compose up` that starts services from the files they came from. + + Attributes: + missing: Compose files the services were started from that no longer exist, + left out of the command, and the config hash check judges what that + changed. + """ project: str working_dir: str config_files: tuple[str, ...] environment_files: tuple[str, ...] services: tuple[str, ...] + missing: tuple[str, ...] = () def command(self) -> list[str]: """The compose command, every profile enabled so each service resolves.""" @@ -116,8 +123,14 @@ def command(self) -> list[str]: return command + ["--profile", "*", "up", "-d", *self.services] -def restores(found: Iterable[Container]) -> list[Restore]: - """One `up` per set of compose files the restored containers were started from.""" +def restores( + found: Iterable[Container], *, exists: Callable[[str], bool] = lambda path: True +) -> list[Restore]: + """One `up` per set of compose files the restored containers were started from. + + A file `exists` rejects is left out, as an e2e run's own override is once the + run has cleaned up after itself. + """ groups: dict[tuple, list[str]] = defaultdict(list) for container in found: if container.restored: @@ -129,7 +142,14 @@ def restores(found: Iterable[Container]) -> list[Restore]: ) groups[key].append(container.service) return [ - Restore(project, working_dir, files, env_files, tuple(sorted(services))) + Restore( + project, + working_dir, + tuple(path for path in files if exists(path)), + env_files, + tuple(sorted(services)), + missing=tuple(path for path in files if not (exists(path))), + ) for (project, working_dir, files, env_files), services in sorted(groups.items()) ] diff --git a/scripts/test_e2e.py b/scripts/test_e2e.py index b54a6fd..4370a64 100755 --- a/scripts/test_e2e.py +++ b/scripts/test_e2e.py @@ -1101,11 +1101,15 @@ def put_back(ctx, found): print() stack_down() - for restore in _stack_state.restores(found): + for restore in _stack_state.restores(found, exists=os.path.exists): LOGGER.info( f"Starting the {len(restore.services)} service(s) this run found, from " f"{len(restore.config_files)} compose file(s)..." ) + if restore.missing: + LOGGER.warning( + f" Left out, as they no longer exist: {', '.join(restore.missing)}" + ) result = subprocess.run( restore.command(), cwd=restore.working_dir or PROJECT_DIR, diff --git a/scripts/tests/test_stack_state.py b/scripts/tests/test_stack_state.py index fbc4c78..95ba9e5 100644 --- a/scripts/tests/test_stack_state.py +++ b/scripts/tests/test_stack_state.py @@ -96,6 +96,18 @@ def test_the_restore_names_the_project_directory_and_files_it_came_from() -> Non ] +def test_a_compose_file_that_is_gone_is_left_out_and_named() -> None: + found = _stack_state.containers([_document("clickhouse")]) + + [restore] = _stack_state.restores( + found, exists=lambda path: not (path.endswith("tls.yml")) + ) + + assert restore.config_files == ("/opt/dfe/docker-compose.yml",) + assert restore.missing == ("/opt/dfe/docker-compose.tls.yml",) + assert "/opt/dfe/docker-compose.tls.yml" not in restore.command() + + def test_services_started_from_different_files_get_one_up_each() -> None: found = _stack_state.containers( [ From cbccc7b615d932dff5b984ed839d63d89e80b1b9 Mon Sep 17 00:00:00 2001 From: Derek Date: Sun, 11 Oct 2026 13:21:15 +1100 Subject: [PATCH 3/5] docs: the e2e owner's ids reach the proxies on TLS --- docs/developing.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/developing.md b/docs/developing.md index c0d9415..3902ade 100644 --- a/docs/developing.md +++ b/docs/developing.md @@ -326,7 +326,7 @@ files from an earlier run never count. The data file goes to the ingest edge 16 requests at a time. The receiver answers each request once the loader has flushed it, so one at a time costs a flush per event. -The suite stops whatever stack the project is running, so it reads the project first. On every way out, a failure or a SIGTERM included, it starts each service that was running or had run to completion again, from the compose files it was started from. A last verdict fails if one comes back on a different config hash. Run it as the user that started the stack: `make` writes that user's UID and GID into the proxies' and the engine's config. +The suite stops whatever stack the project is running, so it reads the project first. On every way out, a failure or a SIGTERM included, it starts each service that was running or had run to completion again, from the compose files it was started from. A last verdict fails if one comes back on a different config hash. Run it as the user that started the stack: `make` writes that user's GID into the engine's config, and under console TLS its UID and GID into both proxies'. Under [console TLS](configuration.md#console-tls-opt-in) each test's stack chains the fragments `make ci` does, the exposure opt-outs and container logs included. The console checks go to `https://:` and verify its certificate by chain and name, against `DFE_PROXY_CA_BUNDLE` when a private CA signed it. A certificate that does not verify fails the check with the variable to change. From f41338f6a0205c98a5b6286015751030170bf836 Mon Sep 17 00:00:00 2001 From: Derek Date: Sun, 11 Oct 2026 13:39:38 +1100 Subject: [PATCH 4/5] fix: restore with only the found services' profiles dfe-engine depends on both brokers, each optional and gated by its own compose profile, so the restore's `--profile "*"` started both and the second failed on port 9092. The restore now asks compose for every service those files define and enables only the profiles of the services it starts, which is the set the original `make ci` enabled. --- scripts/_stack_state.py | 31 +++++++++++++++++++++--- scripts/test_e2e.py | 17 +++++++++++-- scripts/tests/test_stack_state.py | 40 ++++++++++++++++++++++++++++--- 3 files changed, 80 insertions(+), 8 deletions(-) diff --git a/scripts/_stack_state.py b/scripts/_stack_state.py index 5e9bace..25c811c 100644 --- a/scripts/_stack_state.py +++ b/scripts/_stack_state.py @@ -112,15 +112,40 @@ class Restore: services: tuple[str, ...] missing: tuple[str, ...] = () - def command(self) -> list[str]: - """The compose command, every profile enabled so each service resolves.""" + def compose(self) -> list[str]: + """`docker compose` on the project, directory and files the services came from.""" command = ["docker", "compose", "--project-name", self.project] command += ["--project-directory", self.working_dir] for path in self.environment_files: command += ["--env-file", path] for path in self.config_files: command += ["-f", path] - return command + ["--profile", "*", "up", "-d", *self.services] + return command + + def config(self) -> list[str]: + """The command that prints every service those files define, as JSON.""" + return self.compose() + ["--profile", "*", "config", "--format", "json"] + + def command(self, profiles: Iterable[str] = ()) -> list[str]: + """The `up`, with only the given profiles enabled. + + Every profile at once is wrong: both brokers are optional dependencies + of dfe-engine, and the profile is what picks which one starts. + """ + flags = [ + part for profile in sorted(set(profiles)) for part in ("--profile", profile) + ] + return self.compose() + flags + ["up", "-d", *self.services] + + +def profiles_of(model: Mapping, services: Iterable[str]) -> set[str]: + """The profiles the named services belong to, from `docker compose config` JSON.""" + defined = model.get("services") or {} + return { + profile + for service in services + for profile in (defined.get(service) or {}).get("profiles") or () + } def restores( diff --git a/scripts/test_e2e.py b/scripts/test_e2e.py index 4370a64..41eb1cc 100755 --- a/scripts/test_e2e.py +++ b/scripts/test_e2e.py @@ -1110,8 +1110,8 @@ def put_back(ctx, found): LOGGER.warning( f" Left out, as they no longer exist: {', '.join(restore.missing)}" ) - result = subprocess.run( - restore.command(), + run = functools.partial( + subprocess.run, cwd=restore.working_dir or PROJECT_DIR, env=FOUND_ENVIRON, capture_output=True, @@ -1119,6 +1119,19 @@ def put_back(ctx, found): encoding="utf-8", errors="replace", ) + config = run(restore.config()) + try: + model = json.loads(config.stdout) if (config.returncode == 0) else None + except ValueError: + model = None + if model is None: + mark_fail( + ctx, + "the services this run found are not started again: compose could " + f"not resolve the files they came from: {config.stderr.strip()}", + ) + continue + result = run(restore.command(_stack_state.profiles_of(model, restore.services))) if result.returncode != 0: detail = (result.stderr or result.stdout or "").strip() mark_fail(ctx, f"the services this run found did not start again: {detail}") diff --git a/scripts/tests/test_stack_state.py b/scripts/tests/test_stack_state.py index 95ba9e5..ba55739 100644 --- a/scripts/tests/test_stack_state.py +++ b/scripts/tests/test_stack_state.py @@ -72,12 +72,12 @@ def test_what_was_not_running_or_finished_cleanly_stays_down(document: dict) -> assert _stack_state.restores(_stack_state.containers([document])) == [] -def test_the_restore_names_the_project_directory_and_files_it_came_from() -> None: +def test_the_restore_names_the_project_directory_files_and_profiles() -> None: [restore] = _stack_state.restores( _stack_state.containers([_document("dfe-engine")]) ) - assert restore.command() == [ + assert restore.command(["core", "clickhouse", "core"]) == [ "docker", "compose", "--project-name", @@ -89,13 +89,47 @@ def test_the_restore_names_the_project_directory_and_files_it_came_from() -> Non "-f", "/opt/dfe/docker-compose.tls.yml", "--profile", - "*", + "clickhouse", + "--profile", + "core", "up", "-d", "dfe-engine", ] +# The shape that matters: dfe-engine depends on both brokers, each optional and +# gated by its own profile, so enabling every profile starts both. +_MODEL = { + "services": { + "clickhouse": {"profiles": ["clickhouse"]}, + "dfe-engine": {"profiles": ["core"]}, + "kafka-apache": {"profiles": ["kafka-apache"]}, + "kafka-redpanda": {"profiles": ["kafka-redpanda"]}, + "dfe-instance": {}, + } +} + + +def test_only_the_restored_services_profiles_are_enabled() -> None: + profiles = _stack_state.profiles_of(_MODEL, ["dfe-engine", "clickhouse"]) + + assert profiles == {"core", "clickhouse"} + + +def test_a_service_with_no_profile_or_unknown_to_the_model_adds_none() -> None: + assert _stack_state.profiles_of(_MODEL, ["dfe-instance", "gone"]) == set() + + +def test_the_config_command_reads_every_service_the_files_define() -> None: + [restore] = _stack_state.restores( + _stack_state.containers([_document("dfe-engine")]) + ) + + assert restore.config()[-5:] == ["--profile", "*", "config", "--format", "json"] + assert "*" not in restore.command(["core"]) + + def test_a_compose_file_that_is_gone_is_left_out_and_named() -> None: found = _stack_state.containers([_document("clickhouse")]) From 411ec4eb2cbbc8a7c5ee9acb5f91be9acb3290b8 Mon Sep 17 00:00:00 2001 From: Derek Date: Sun, 11 Oct 2026 13:59:25 +1100 Subject: [PATCH 5/5] fix: lock the e2e scripts' packages by hash `scripts/test_e2e.py.lock` and `scripts/test_flows.py.lock` (from `uv lock --script`) record every package with a sha256 per file, the transitive greenlet, pyee and typing-extensions included. The make targets and the shebang run `uv run --locked`, so a lock that no longer matches its script stops the run instead of resolving something new. Renovate's pep723 manager is opt-in (`managerFilePatterns: []`) and has no artifact step, so it can move an inline pin but never rewrites the lock beside it. `make check-locks` runs `uv lock --check` on each script with uv 0.13.0, and CI runs it after the unit tests. A stdlib test asserts every script with inline metadata is pinned exactly, has a lock recording those pins, and is named by `check-locks`. --- .github/workflows/ci.yml | 9 ++ Makefile | 26 +++-- docs/developing.md | 9 +- docs/troubleshooting.md | 6 +- scripts/_search_rule.py | 4 +- scripts/test_e2e.py | 6 +- scripts/test_e2e.py.lock | 162 +++++++++++++++++++++++++++++ scripts/test_flows.py.lock | 52 +++++++++ scripts/tests/test_script_locks.py | 98 +++++++++++++++++ 9 files changed, 355 insertions(+), 17 deletions(-) create mode 100644 scripts/test_e2e.py.lock create mode 100644 scripts/test_flows.py.lock create mode 100644 scripts/tests/test_script_locks.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index e5e51f0..87979c2 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -118,6 +118,15 @@ jobs: - name: Tests run: make check-tests PYTEST=pytest + # `make test-e2e` runs its scripts with `uv run --locked`, and Renovate's + # pep723 manager can bump a script's pin but has no step that rewrites its + # lock, so a pin and lock that disagree are caught here, not mid-suite. + - name: Install uv + run: pipx install --python "${{ steps.python.outputs.python-path }}" "uv==$(make -s print-uv-version)" + + - name: Script locks + run: make check-locks LOCK_UV=uv + # The slim and single profiles are projections of the Kubernetes tiers of the # same name, so they go stale when those tiers move and nothing here notices. # dfe-infra is PRIVATE, so the checkout reads it with a short-lived GitHub App diff --git a/Makefile b/Makefile index 7078fbd..60e9102 100644 --- a/Makefile +++ b/Makefile @@ -52,7 +52,7 @@ endif # any more (both sweep every profile), and requiring a resolvable profile to STOP # a stack is the same lockout the compose secret comments argue against -- set # DFE_PROFILE to something that does not exist and you could not tear down. -BOOTSTRAP_GOALS := init env-files creds dev-posture help login stack dial modes down clean limits check check-compose check-hardfail check-dockerfile check-docs check-proxy check-python check-tests +BOOTSTRAP_GOALS := init env-files creds dev-posture help login stack dial modes down clean limits check check-compose check-hardfail check-dockerfile check-docs check-locks check-proxy check-python check-tests # Resolve the active profile only when a goal actually needs the compose stack ifneq (,$(filter-out $(BOOTSTRAP_GOALS),$(or $(MAKECMDGOALS),help))) @@ -536,7 +536,7 @@ infra: .env storage-dirs ## Start infrastructure services # --------------------------------------------------------------------------- .PHONY: check -check: check-compose check-hardfail check-dockerfile check-docs check-proxy check-python check-tests ## Run every static check CI runs +check: check-compose check-hardfail check-dockerfile check-docs check-proxy check-python check-tests check-locks ## Run every static check CI runs .PHONY: limits limits: ## Show the resource limits and totals, computed from the resolved config @@ -571,6 +571,20 @@ print-pytest-version: check-tests: ## Run the helper-script unit tests (scripts/tests) $(PYTEST) -q scripts/tests +# Each inline-metadata script runs with `uv run --locked`, and Renovate's pep723 manager rewrites a pin but never its lock, so this fails a pin and lock that disagree. +# Pinned and installed in CI the way RUFF is (LOCK_UV=uv). +UV_VERSION := 0.13.0 +LOCK_UV ?= uvx uv@$(UV_VERSION) + +.PHONY: print-uv-version +print-uv-version: + @echo $(UV_VERSION) + +.PHONY: check-locks +check-locks: ## Assert each inline-metadata script's lock matches its pins (relock with `uv lock --script