From 9b4c30b75c48e494a971660ae3f368daaacf111a Mon Sep 17 00:00:00 2001 From: Antoine Vandeplanque Date: Thu, 1 Oct 2026 17:43:58 +0200 Subject: [PATCH] configure collection on host and warn on intentional coverage limits --- docs/SSH_RUN.md | 40 +++++++++++++- ops/app/compose.runtime.yaml | 6 ++ ops/gitops/reconcile.py | 7 ++- ops/ssh/host_run.py | 104 ++++++++++++++++++++++++++++++++++- tests/test_ssh_run.py | 83 +++++++++++++++++++++++++++- 5 files changed, 233 insertions(+), 7 deletions(-) diff --git a/docs/SSH_RUN.md b/docs/SSH_RUN.md index 27edca2..22a78f9 100644 --- a/docs/SSH_RUN.md +++ b/docs/SSH_RUN.md @@ -5,8 +5,11 @@ resolves a published full SHA or Git tag to an immutable image digest and invoke a fixed SSH command on `tonio@192.168.1.56:22`. The build VM must reach that LAN SSH listener (not Gitea's port 2222). No timer or Git request branch is used. -Actions waits for the pipeline's exit status. Exit 0 is green; any nonzero pipeline -exit fails the job. Green does not guarantee complete source coverage: see +Actions waits for the pipeline's outcome. Exit 0 is green. Exit 1 is also green +with an explanatory warning **only** when the saved manifest and pipeline sidecar +confirm successful import, zero rejected records, and partial coverage caused +solely by configured page/detail limits. Other nonzero exits fail the job. Missing +or ambiguous evidence never suppresses failure. Green does not guarantee complete source coverage: see [pipeline outcomes](DEPLOYMENT.md#outcomes-and-recovery). Scraper output stays in private host logs; CI gets only request/revision/exit summaries. Migrations are manual, DB credentials stay on the homelab, and there are no automatic retries. @@ -112,12 +115,43 @@ one-shot collection/import immediately, not just a queue request. ## State, failures and recovery +### Host-side collection policy + +Set these in `~/containers/car-tracker/app.env` (not Actions secrets): + +```dotenv +COLLECT_FAMILY=saab-900 +COLLECT_MAX_PAGES=1 +COLLECT_MAX_DETAILS=all +COLLECT_DELAY=5 +``` + +The SSH wrapper reads resolved Compose settings and supplies a validated, fixed +pipeline command, overriding the image's smoke-test CMD without rebuilding it. +`COLLECT_MAX_DETAILS=all` omits `--max-details`: every unique ad discovered on the +fetched search pages gets a detail attempt. It does not remove the per-search page +limit or guarantee detail success. A positive integer sets a detail cap per run. +`COLLECT_FAMILY=all` includes every search in the image's searches.toml; otherwise +it filters by family. Pages must be a positive integer; delay must be finite and +at least 1 second. Missing settings retain defaults: saab-900, 1 page, 1 detail, +5 seconds. Invalid settings fail before launching the collector. + +Install both the updated host_run.py/reconcile.py and compose.runtime.yaml when +upgrading. Settings affect the next new request; rerunning a completed Actions run +returns its previous result instead of collecting with new settings. More details +increase runtime; the Actions timeout remains 45 minutes. The pipeline and local +state still retain exit 1 for partial coverage; the SSH wrapper converts verified +limit-only outcomes into an Actions warning/success. No run is launched by editing app.env. + +### Execution outcomes + Local outcomes: `~/containers/car-tracker/state/requests/.json`. Private runtime logs: the neighboring `.log`. Host failures before launch return 70; SSH connection failures return 255. Inspect host state/logs locally. The request UUID is derived from the Actions run ID. Retrying the same workflow -run does not repeat a consumed request; it returns the recorded exit status. A +run does not repeat a consumed request; it evaluates the recorded outcome using +the same limit-only warning policy (the original pipeline exit remains saved). A new manual dispatch creates a new request, even for the same release. Parallel runs are refused by the host lock, not queued. diff --git a/ops/app/compose.runtime.yaml b/ops/app/compose.runtime.yaml index 29c944c..6edc513 100644 --- a/ops/app/compose.runtime.yaml +++ b/ops/app/compose.runtime.yaml @@ -28,6 +28,12 @@ services: PGSSLMODE: "${PGSSLMODE:-require}" PGPASSFILE: /run/db/pgpass PGSSLROOTCERT: /run/db/root.crt + # Host-owned collection policy. The restricted SSH wrapper validates these + # and builds a fixed argv; all omits the corresponding CLI limit/filter. + COLLECT_FAMILY: "${COLLECT_FAMILY:-saab-900}" + COLLECT_MAX_PAGES: "${COLLECT_MAX_PAGES:-1}" + COLLECT_MAX_DETAILS: "${COLLECT_MAX_DETAILS:-1}" + COLLECT_DELAY: "${COLLECT_DELAY:-5}" extra_hosts: - "host.docker.internal:host-gateway" volumes: diff --git a/ops/gitops/reconcile.py b/ops/gitops/reconcile.py index 57841fb..4ba4766 100644 --- a/ops/gitops/reconcile.py +++ b/ops/gitops/reconcile.py @@ -60,11 +60,14 @@ def compose_environment(image): "PGHOST", "PGPORT", "PGDATABASE", "PGUSER", "PGSSLMODE", "COMPOSE_FILE", "COMPOSE_PROJECT_NAME", "COMPOSE_PROFILES", "COMPOSE_ENV_FILES"): env.pop(key, None) + for key in list(env): + if key.startswith("COLLECT_"): + env.pop(key) env["CAR_TRACKER_IMAGE"] = image return env -def attempt(request, state, compose_file, env_file, private_log=False): +def attempt(request, state, compose_file, env_file, private_log=False, app_command=None): outcome = state / f"{request['request_id']}.json" if outcome.exists(): recorded = json.loads(outcome.read_text()) @@ -93,6 +96,8 @@ def attempt(request, state, compose_file, env_file, private_log=False): try: args = [*compose, "run", "--rm", "--no-deps", "--pull", "never", "--name", f"car-tracker-run-{request['request_id']}", "app"] + if app_command is not None: + args.extend(app_command) if private_log: log_path = state / f"{request['request_id']}.log" descriptor = os.open(log_path, os.O_WRONLY | os.O_CREAT | os.O_APPEND, 0o600) diff --git a/ops/ssh/host_run.py b/ops/ssh/host_run.py index 6ad0b41..f2716ca 100644 --- a/ops/ssh/host_run.py +++ b/ops/ssh/host_run.py @@ -2,12 +2,101 @@ import fcntl import json +import math import os from pathlib import Path +import re import subprocess import sys -from reconcile import attempt, validate_request +from reconcile import attempt, command, compose_environment, validate_request + + +def limit_reasons(manifest, pipeline): + """Recognize only verified intentional limits, never generic partial runs.""" + if (manifest.get("status") != "partial" or "error_type" in manifest or + pipeline.get("import_status") != "succeeded" or + pipeline.get("collection_status") != "partial" or + "error_type" in pipeline): + return [] + summary = pipeline.get("import_summary", {}) + if summary.get("rejected") != 0 or summary.get("run_id") != manifest.get("run_id"): + return [] + searches = manifest.get("searches", []) + if not searches or len(searches) != len(manifest.get("planned_searches", [])): + return [] + reasons = set() + for search in searches: + if search.get("status") == "complete": + continue + if search.get("status") != "partial" or search.get("reason") != "page_limit": + return [] + reasons.add("page_limit") + details = manifest.get("details", {}) + if details.get("status") not in ("complete", "partial", "disabled"): + return [] + for item in details.get("listings", {}).values(): + if item.get("status") == "complete": + continue + if item.get("status") != "not_attempted" or item.get("reason") != "detail_limit": + return [] + reasons.add("detail_limit") + return sorted(reasons) + + +def actions_outcome(code, request, state, raw_root): + if code != 1: + return code + try: + paths = [] + with (state / f"{request['request_id']}.log").open() as log: + for line in log: + if line.startswith("Manifest: "): + paths.append(line.removeprefix("Manifest: ").strip()) + if len(paths) != 1 or not re.fullmatch( + r"/data/[0-9]{4}-[0-9]{2}-[0-9]{2}/[0-9a-f-]{36}/manifest\.json", paths[0]): + return code + raw_root = raw_root.resolve() + path = (raw_root / paths[0].removeprefix("/data/")).resolve() + if not path.is_relative_to(raw_root): + return code + manifest = json.loads(path.read_text()) + pipeline = json.loads(path.with_name("pipeline.json").read_text()) + reasons = limit_reasons(manifest, pipeline) + if not reasons: + return code + explanations = {"page_limit": "the configured page limit left additional search pages unfetched", + "detail_limit": "the configured detail limit left some discovered ads without detail attempts"} + message = "; ".join(explanations[reason] for reason in reasons) + print(f"::warning::Import succeeded with no rejected records, but coverage is intentionally partial: {message}. " + "This is not full source coverage. Host state retains pipeline exit 1.", flush=True) + return 0 + except (OSError, ValueError, KeyError, TypeError, AttributeError): + # Missing/malformed evidence must never turn a failure green. + return code + + +def collection_command(settings): + family = settings.get("COLLECT_FAMILY", "saab-900") + pages = settings.get("COLLECT_MAX_PAGES", "1") + details = settings.get("COLLECT_MAX_DETAILS", "1") + delay = settings.get("COLLECT_DELAY", "5") + if not isinstance(family, str) or not family.strip(): + raise ValueError("A collection family or all is required") + if not isinstance(pages, str) or not pages.isdecimal() or int(pages) < 1: + raise ValueError("Page limit must be a positive integer") + if not isinstance(details, str) or (details != "all" and + (not details.isdecimal() or int(details) < 1)): + raise ValueError("Detail limit must be all or a positive integer") + if not isinstance(delay, str) or not math.isfinite(float(delay)) or float(delay) < 1: + raise ValueError("Delay must be finite and at least one second") + args = ["car-tracker-pipeline", "--headed", "--config", "/app/searches.toml", + "--output", "/data", "--max-pages", pages, "--delay", delay] + if family != "all": + args.extend(["--family", family]) + if details != "all": + args.extend(["--max-details", details]) + return args def parse_command(value): @@ -29,7 +118,18 @@ def run(root, request): for path in state.glob("*.json"): if json.loads(path.read_text()).get("status") == "attempted": raise ValueError("Unknown previous outcome requires investigation") - return attempt(request, state, root / "compose.runtime.yaml", root / "app.env", private_log=True) + config = json.loads(command( + "docker", "compose", "--env-file", str(root / "app.env"), + "-f", str(root / "compose.runtime.yaml"), "config", "--format", "json", + env=compose_environment(request["image"]))) + args = collection_command(config["services"]["app"]["environment"]) + code = attempt(request, state, root / "compose.runtime.yaml", root / "app.env", + private_log=True, app_command=args) + raw_mounts = [volume["source"] for volume in config["services"]["app"].get("volumes", []) + if volume.get("target") == "/data" and volume.get("type") == "bind"] + if len(raw_mounts) != 1: + return code + return actions_outcome(code, request, state, Path(raw_mounts[0])) def main(): diff --git a/tests/test_ssh_run.py b/tests/test_ssh_run.py index 3f4cacb..c9c9595 100644 --- a/tests/test_ssh_run.py +++ b/tests/test_ssh_run.py @@ -1,4 +1,6 @@ import importlib.util +import copy +import io import json from pathlib import Path import subprocess @@ -40,16 +42,21 @@ class SSHRunTests(unittest.TestCase): def test_host_returns_pipeline_exit_and_keeps_output_private(self): def command(*args, **kwargs): + if "--format" in args: + return json.dumps({"services": {"app": {"environment": {"COLLECT_MAX_DETAILS": "all"}}}}) if "inspect" in args: return json.dumps([{"Config": {"Labels": {"org.opencontainers.image.revision": "a" * 40}}}]) return "" def launch(args, **kwargs): + self.assertIn("car-tracker-pipeline", args) + self.assertNotIn("--max-details", args) self.assertEqual(kwargs["stderr"], subprocess.STDOUT) kwargs["stdout"].write("Private collection output\n") return subprocess.CompletedProcess(args, 3) - with patch.object(agent, "command", side_effect=command), \ + with patch.object(host, "command", side_effect=command), \ + patch.object(agent, "command", side_effect=command), \ patch.object(agent.subprocess, "run", side_effect=launch) as child: self.assertEqual(host.run(self.root, self.request), 3) self.assertEqual(host.run(self.root, self.request), 3) @@ -74,3 +81,77 @@ class SSHRunTests(unittest.TestCase): fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB) with self.assertRaises(ValueError): host.run(self.root, self.request) + + def test_collection_all_omits_detail_limit_but_keeps_page_limit(self): + args = host.collection_command({"COLLECT_MAX_DETAILS": "all"}) + self.assertNotIn("--max-details", args) + self.assertEqual(args[args.index("--max-pages") + 1], "1") + self.assertEqual(args[args.index("--family") + 1], "saab-900") + + def test_explicit_settings_and_all_families(self): + args = host.collection_command({"COLLECT_FAMILY": "all", "COLLECT_MAX_PAGES": "3", + "COLLECT_MAX_DETAILS": "20", "COLLECT_DELAY": "6"}) + self.assertNotIn("--family", args) + self.assertEqual(args[args.index("--max-details") + 1], "20") + self.assertEqual(args[args.index("--max-pages") + 1], "3") + + def test_invalid_settings_are_refused(self): + for key, value in [("COLLECT_MAX_DETAILS", "0"), ("COLLECT_MAX_DETAILS", "-1"), + ("COLLECT_MAX_DETAILS", "all; id"), ("COLLECT_MAX_PAGES", "all"), + ("COLLECT_MAX_PAGES", "0"), ("COLLECT_DELAY", "nan"), + ("COLLECT_DELAY", "inf"), ("COLLECT_DELAY", "0.5"), + ("COLLECT_FAMILY", "")]: + with self.subTest(key=key, value=value), self.assertRaises(ValueError): + host.collection_command({key: value}) + + def bounded_outcome(self): + manifest = {"run_id": "a" * 36, "status": "partial", + "planned_searches": [{"family": "saab-900"}], + "searches": [{"status": "partial", "reason": "page_limit"}], + "details": {"status": "complete", "listings": {"1": {"status": "complete"}}}} + pipeline = {"collection_status": "partial", "import_status": "succeeded", + "import_summary": {"run_id": "a" * 36, "rejected": 0}} + return manifest, pipeline + + def test_only_intentional_limits_are_warnings(self): + manifest, pipeline = self.bounded_outcome() + self.assertEqual(host.limit_reasons(manifest, pipeline), ["page_limit"]) + manifest["details"] = {"status": "partial", "listings": { + "1": {"status": "not_attempted", "reason": "detail_limit"}}} + self.assertEqual(host.limit_reasons(manifest, pipeline), ["detail_limit", "page_limit"]) + for mutate in ( + lambda m, p: m.update(error_type="SourceAccessError"), + lambda m, p: m["details"]["listings"]["1"].update(reason="run_stopped"), + lambda m, p: m["details"]["listings"]["1"].update(status="unavailable"), + lambda m, p: m["searches"][0].update(reason="TimeoutError"), + lambda m, p: m["planned_searches"].append({"family": "another"}), + lambda m, p: p.update(import_status="unknown"), + lambda m, p: p["import_summary"].update(rejected=1), + lambda m, p: p["import_summary"].update(run_id="different"), + ): + m, p = copy.deepcopy(manifest), copy.deepcopy(pipeline) + mutate(m, p) + self.assertEqual(host.limit_reasons(m, p), []) + + def test_verified_limit_outcome_warns_and_returns_success(self): + raw = self.root / "raw" + path = raw / "2026-10-01" / ("a" * 36) + path.mkdir(parents=True) + manifest, pipeline = self.bounded_outcome() + (path / "manifest.json").write_text(json.dumps(manifest)) + (path / "pipeline.json").write_text(json.dumps(pipeline)) + log = self.root / f"{self.request['request_id']}.log" + log.write_text(f"Manifest: /data/2026-10-01/{'a' * 36}/manifest.json\n") + with patch("sys.stdout", new_callable=io.StringIO) as output: + self.assertEqual(host.actions_outcome(1, self.request, self.root, raw), 0) + self.assertIn("::warning::", output.getvalue()) + self.assertIn("page limit", output.getvalue()) + (path / "pipeline.json").unlink() + self.assertEqual(host.actions_outcome(1, self.request, self.root, raw), 1) + self.assertEqual(host.actions_outcome(3, self.request, self.root, raw), 3) + + def test_unknown_or_unsafe_manifest_path_remains_failure(self): + log = self.root / f"{self.request['request_id']}.log" + for value in ("", "Manifest: /data/../../secret\n", "Manifest: /data/unknown/manifest.json\n"): + log.write_text(value) + self.assertEqual(host.actions_outcome(1, self.request, self.root, self.root), 1)