diff --git a/.github/workflows/model-smoke.yml b/.github/workflows/model-smoke.yml index bcb0bc0..a0277ea 100644 --- a/.github/workflows/model-smoke.yml +++ b/.github/workflows/model-smoke.yml @@ -18,8 +18,9 @@ on: - worker/ai_bridge/jobs.py # The worker's HTTP handling, which every model test and the MCP workers' health check go through. - worker/ai_bridge/server.py - # Imported by both: the counters every model job passes through. + # Imported by both: the counters and the log lines every model job passes through. - worker/ai_bridge/metrics.py + - worker/ai_bridge/log.py - scripts/benchmark.py - scripts/ner_measure.py - scripts/public_data.py diff --git a/CHANGELOG.md b/CHANGELOG.md index d49ed16..39b87e4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,6 +10,7 @@ All notable changes to this project are documented in this file. The format foll - A graceful drain: on SIGTERM or SIGINT the worker answers `POST /v1/jobs` with 503 `draining` and `Retry-After: 1`, fails queued jobs as `service_restarted` at once, keeps answering `GET`, and lets running jobs finish for up to `BRIDGE_DRAIN_SECONDS` (integer 0 to 300, default 30) before it stops; a second signal stops it at once. - `GET /readyz`, behind any configured token like `/healthz`: 200 when the worker is not draining and a child is idle with its backend's models loaded, otherwise 503 with a reason (`draining`, `no_idle_child`, `models_loading`). A child now loads its backend's models as it starts instead of on its first job. Compose's MCP workers report healthy only once ready. +- Optional structured logging: `BRIDGE_LOG=json` writes one JSON object per line on stderr, best effort (a full queue or the exit can drop lines), a `request` line after each reply (method, route template, status, duration, the token's name), a `job` line when a job ends (job ID, task, owner, terminal status, queue and run times), and `start`, `drain` and `stop` lines. A fixed schema checked value by value keeps inputs, results, headers, query strings, tokens and exception text out; a stderr that blocks or fails never holds a request. `off`, the default, writes nothing, and any other value stops the worker at start naming `BRIDGE_LOG`. - A full connection budget (32) is answered 503 `service_busy` with `Retry-After: 1` and `Connection: close` instead of a closed socket, and a request line over 8 KiB or a header block over 16 KiB is refused with 431 `headers_too_large`. ### Changed diff --git a/README.md b/README.md index 53b875f..4126f9e 100644 --- a/README.md +++ b/README.md @@ -13,6 +13,7 @@ Includes a FrankenPHP worker-mode example, deterministic fault tests, and an opt - Bounded concurrency, queue/result capacity, retention, request sizes and execution deadlines. - Cancellation, progress polling, crash detection and generic errors that omit internal exception text. - `GET /metrics` on the worker: aggregate job counts, queue depth, pool readiness and duration histograms in the Prometheus text format, behind the service token and carrying no job content; counters reset on restart ([the protocol](docs/protocol.md#metrics)). +- Optional structured logging with `BRIDGE_LOG=json`: best effort, one JSON line on stderr per request and per finished job, plus start, drain and stop, with a fixed schema that never holds inputs, results, headers, query strings, tokens or exception text; off by default ([the protocol](docs/protocol.md#logging)). - A FrankenPHP example that submits work and returns a job ID immediately. - A stdio MCP server with one tool per task, a second client of the same protocol beside the PHP client, plus an index builder over the protocol and a `bridge_search` tool that searches what it built, returning offsets and snippets rather than vectors, a compose service that puts the pinned ONNX models behind it, and a ten-question evaluation for LLM hosts with a runner for the Claude Code CLI; see [the MCP guide](docs/mcp.md). - CPU inference with `cross-encoder/ms-marco-TinyBERT-L2-v2`, `sentence-transformers/all-MiniLM-L6-v2` and int8 `dslim/bert-base-NER` and `dslim/bert-base-NER-uncased` through ONNX Runtime, using pinned model files and wheel hashes. diff --git a/docs/protocol.md b/docs/protocol.md index 2e17844..6dfb5f8 100644 --- a/docs/protocol.md +++ b/docs/protocol.md @@ -81,3 +81,17 @@ On SIGTERM or SIGINT the worker drains: every `POST /v1/jobs` is answered 503 `d | `bridge_info` | gauge | `protocol="1"`, `backend` | Always 1 | Both histograms have the buckets 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30, 60 and 300 seconds and `+Inf`, cumulative as the format requires, with `_sum` and `_count`. A scrape copies the numbers under the store's lock and formats them without it, so it never waits for a job. An invalid submission (400) creates no job and counts nothing. + +## Logging + +The worker writes no log by default: `BRIDGE_LOG` unset, empty or `off` writes nothing at all. With `BRIDGE_LOG=json` it writes one JSON object per line on stderr; any other value stops the worker at start with a message naming the variable. Every line has `ts` (UTC, ISO 8601 with milliseconds and `Z`), `level` (`info` or `warning`) and `event`, then exactly the keys of its event, `null` where a value is unknown or does not apply. Values come only from fixed sets, numbers, a job ID or a token's name: no input, result, header, query string, path, idempotency key, token, error message or exception text is ever written, and a key outside the table is dropped by the formatter whatever a caller passes. Lines are written by a thread of their own behind a queue of 1024: a stderr that blocks or fails never holds or fails a request or a job, and lines that do not fit are dropped. Logging is best effort: lines may be dropped when the queue is full or at exit. A stop waits about 2 s at most for the queued lines to be written, and a process that ends without that stop (killed, for instance) loses what was still queued. + +| `event` | When | Keys | +| --- | --- | --- | +| `request` | After the reply to a request is sent, a 401, 408, 413, 431 and 503 included; a connection that ends before any reply writes none | `method` (`GET`, `POST`, `other`, or `null` when the request line could not be read or parsed), `route` (`/healthz`, `/readyz`, `/metrics`, `/v1/jobs`, `/v1/jobs/{id}`, `/v1/jobs/{id}/cancel`, or `other` for any other path, a query string included), `status`, `duration_ms` (from the start of reading the request to the reply sent), `owner` (the token's name, `null` when the request was not authenticated) | +| `job` | A job reaches `succeeded`, `failed`, `timed_out` or `cancelled` | `job_id`, `task`, `owner`, `job_status`, `queue_ms` (submission to `running`, `null` if no child started it), `run_ms` (`running` to the terminal status, `null` likewise) | +| `start` | The port is bound and the children are starting | `backend`, `concurrency`, `capacity`, `retention_seconds`, `max_jobs`, `drain_seconds`, `tokens` (how many tokens are configured; never a name or a token) | +| `drain` | A drain ends, before the worker stops serving | `duration_ms` (from the signal), `running` (jobs still running when it ended, which the stop then cancels), `interrupted` (a second signal ended it) | +| `stop` | The worker has stopped its children and closed the job table | `cancelled` (unfinished jobs the stop ended as `cancelled`) | + +`level` is `warning` for a request answered 5xx, a job that ended `failed` or `timed_out`, a drain that left jobs running and a stop that cancelled any; otherwise `info`. A 503 for a full connection budget is written without reading anything, so its `method` is `null`, its `route` `other` and its `duration_ms` 0. The timings are measured by the parent process, which reads a child's messages every 10 ms or so, so `queue_ms` and `run_ms` can each be off by that much. A job read back from `BRIDGE_STATE_DIR` ended in an earlier process and writes no line. diff --git a/docs/security.md b/docs/security.md index 6542b82..2d6d327 100644 --- a/docs/security.md +++ b/docs/security.md @@ -13,7 +13,7 @@ This experimental service is designed for trusted application-to-service traffic - Redaction is assistive, not a compliance control. The rules find only the structured identifiers they recognise, and the optional named-entity models were trained on English news, so their recall on names from other domains or languages is lower and unmeasured here; review redacted output before treating it as free of personal data. The model files are third-party ONNX conversions of the original weights, pinned by revision and hash but not signed by the original author. - The MCP server is a client of this service for an AI agent over stdio, where there is no authentication: the host that launches it inherits its trust, and it holds the token from its environment and never returns it. Running it over HTTP or SSE would be a new authorisation surface that the bearer token does not cover; it is out of scope. An agent can be prompt-injected and can then fill a worker's capacity, so point the MCP server at a dedicated worker, never at one serving a production application; compose's `mcp` and `mcp-model` each have their own. It reads files only under `BRIDGE_MCP_ROOT` and writes none; snippets it returns are the agent's own file content and enter the agent's context like any file it reads. Index files built for `bridge_search` are derived data: `index.json` holds every chunk's text verbatim and `vectors.f32` holds vectors that can reveal it, so keep them with the same care as their sources. The builder runs with the service token like the PHP client and is the only component that writes, into the directory the operator names. - `GET /metrics` reveals aggregate numbers only: how many jobs of each task were submitted, refused for capacity and ended in each terminal status, how many are queued and running, the pool's size and readiness, the concurrency and capacity settings, the backend name, and histograms of job and queue durations per task. It never reveals a job ID, idempotency key, input, result, error message, model path or the token; its label values come from fixed sets, never from a request. Counts and timings still describe the traffic (how busy the service is, which tasks run, how long they take), so the endpoint is behind the same service token as every route: scrape it from inside the private network with that token, never through a public route. How often it is scraped is the operator's concern; a scrape copies a few counters under the store's lock and formats them outside it. -- Access logs are disabled to avoid accidentally recording identifiers and inputs. Internal exceptions become generic task errors. This is not a full observability or compliance solution. +- The worker writes no log unless `BRIDGE_LOG=json` is set. Then it writes JSON lines on stderr for each request (method, route template, status, duration, the token's name), each job that ends (job ID, task, the token's name, terminal status, queue and run times) and the start, drain and stop (the backend, the numeric limits and how many tokens are configured, how many jobs a drain or stop cut short). It never writes an input, result, header, path or query string, idempotency key, token, error message or exception text: the formatter keeps only its fixed keys and checks each value against a fixed set, a number, the job ID's 32 hex characters or the token-name pattern, writing `null` otherwise. Job IDs and token names still link requests to clients, and timings still describe the traffic, so keep the log with the same care as the metrics. Internal exceptions become generic task errors. This is not a full observability or compliance solution; [the protocol](protocol.md#logging) has the schema. - Cancel/timeout terminates the task process, but cannot roll back previously completed external effects. No task in the prototype modifies an external system. - `/healthz` proves HTTP service availability, not model integrity, ability to spawn processes or successful inference. `/readyz` adds that a child process started and loaded the configured backend's models without an error; it does not prove that they infer correctly. An ONNX worker checks before it starts that each model directory holds its tokenizer and model files and refuses to start otherwise; that proves presence, not integrity, which the fetcher's hashes establish. Use a controlled model smoke test for inference validation. - A drain (the first SIGTERM or SIGINT) keeps the inputs of running jobs in their children until they finish, the deadline passes or a second signal ends it, and keeps every finished result until the process exits. It drops the input of every job still queued at once: the job is failed as `service_restarted` and nothing of it is replayed, so a caller resubmits. Results and inputs are never written anywhere for a later process; with `BRIDGE_STATE_DIR` the job table holds what it always holds (never an input). The process ends with its memory, so after it nothing of a job survives outside that table. diff --git a/docs/testing.md b/docs/testing.md index b673d43..302a0c7 100644 --- a/docs/testing.md +++ b/docs/testing.md @@ -58,7 +58,7 @@ PHP branch support was verified locally on 2026-09-17 on PHP 8.2.33, 8.3.33, 8.4 | Layer | Scope and counting | | --- | --- | -| Python worker | 356 tests: lifecycle, validation and HTTP, the drain (submissions refused with 503, queued jobs failed at once, running ones finished within `BRIDGE_DRAIN_SECONDS`, a second signal), `/readyz` and its reasons for any configured token, the 503 for a full connection budget and the 431 for an oversized request line or header block, per-client tokens (each source and every way it is refused at start, a job and its idempotency key scoped to the submitting token, through the job table and a table from before owners, the `owner` metrics series, and no token in an error, the metrics, the table, a child's environment or a start message), the job table in SQLite (a store reopened on its file fails unfinished jobs as `service_restarted`, serves retained results, deletes expired rows from the file and never holds an input; a submission the table refuses is a 429, a later write it refuses leaves the job served, and a key whose row outlived a failed delete runs again) and idempotency keys, process crashes, cancellation, timeouts, retention, the Retry-After a full or closing store gives and its header on every 429, allocation/exit races, the warm pool (that successive jobs share a child in which no live container or frame holds a reference to a finished job's input or result, after a success, a failure and real task payloads, with the detector checked against a planted needle; that a cancelled, timed-out or crashed child is replaced; that a child is recycled after its share of jobs; that reads, submissions and cancellations answer at once while a child is being started, that a child whose start ends after the store closed does not outlive it, and that a job assigned to or waiting for a live child does not fail when no replacement can start; that a job whose child died before taking it is tried once more on another, only once, that a job sent but not yet taken can still be cancelled or time out, and that a job whose child crashed after taking it is not; that a child, idle or running a job, ignores the SIGINT a terminal's Ctrl+C sends the whole process group; that the service token is absent from a child's environment and from its environment block when the service started it; that a message for an earlier job lands nowhere; that a child that never reports ready is replaced and that children dying at start are not restarted in a loop; and that a cancellation does not hold up reads), redaction rules including IPv6, lower-case IBANs and email addresses in any script and their speed on hostile input, NER aggregation, name joining, the second title-cased reading, the uncased model's reading of the whole text and the merge driven by stubs, the fetcher's pins and the model directories an ONNX worker checks for before it starts, batched cross-encoder scoring, the NER measurement's sample, scoring and WNUT-17 reader, the load measurement's settings, percentiles, error counting and table, the ranking measurement's metrics, BM25, dataset reader and fit to the worker's limits, the public datasets' pins and licence gate, the workflows' runners, action pins and checkout credentials, both Claude Code hooks (the post-edit rules, and the pre-call push guard through its own cases), progress bounds, the service's reading of its BRIDGE_* settings (defaults, every variable, malformed and out-of-range values, a state directory that cannot hold the table), and the model-name table every result reports from | +| Python worker | 395 tests: lifecycle, validation and HTTP, the optional JSON log (every line parses and holds only its event's keys, an input, token, header, query string, idempotency key and a failing task's exception text never reach it, `off` writes nothing and a bad `BRIDGE_LOG` stops the start, the route templates, the 401 without an owner, consistent job timings, and a stderr that fails or blocks costing a request nothing), the drain (submissions refused with 503, queued jobs failed at once, running ones finished within `BRIDGE_DRAIN_SECONDS`, a second signal), `/readyz` and its reasons for any configured token, the 503 for a full connection budget and the 431 for an oversized request line or header block, per-client tokens (each source and every way it is refused at start, a job and its idempotency key scoped to the submitting token, through the job table and a table from before owners, the `owner` metrics series, and no token in an error, the metrics, the table, a child's environment or a start message), the job table in SQLite (a store reopened on its file fails unfinished jobs as `service_restarted`, serves retained results, deletes expired rows from the file and never holds an input; a submission the table refuses is a 429, a later write it refuses leaves the job served, and a key whose row outlived a failed delete runs again) and idempotency keys, process crashes, cancellation, timeouts, retention, the Retry-After a full or closing store gives and its header on every 429, allocation/exit races, the warm pool (that successive jobs share a child in which no live container or frame holds a reference to a finished job's input or result, after a success, a failure and real task payloads, with the detector checked against a planted needle; that a cancelled, timed-out or crashed child is replaced; that a child is recycled after its share of jobs; that reads, submissions and cancellations answer at once while a child is being started, that a child whose start ends after the store closed does not outlive it, and that a job assigned to or waiting for a live child does not fail when no replacement can start; that a job whose child died before taking it is tried once more on another, only once, that a job sent but not yet taken can still be cancelled or time out, and that a job whose child crashed after taking it is not; that a child, idle or running a job, ignores the SIGINT a terminal's Ctrl+C sends the whole process group; that the service token is absent from a child's environment and from its environment block when the service started it; that a message for an earlier job lands nowhere; that a child that never reports ready is replaced and that children dying at start are not restarted in a loop; and that a cancellation does not hold up reads), redaction rules including IPv6, lower-case IBANs and email addresses in any script and their speed on hostile input, NER aggregation, name joining, the second title-cased reading, the uncased model's reading of the whole text and the merge driven by stubs, the fetcher's pins and the model directories an ONNX worker checks for before it starts, batched cross-encoder scoring, the NER measurement's sample, scoring and WNUT-17 reader, the load measurement's settings, percentiles, error counting and table, the ranking measurement's metrics, BM25, dataset reader and fit to the worker's limits, the public datasets' pins and licence gate, the workflows' runners, action pins and checkout credentials, both Claude Code hooks (the post-edit rules, and the pre-call push guard through its own cases), progress bounds, the service's reading of its BRIDGE_* settings (defaults, every variable, malformed and out-of-range values, a state directory that cannot hold the table), and the model-name table every result reports from | | MCP server | 125 tests, offline with hash-pinned wheels: the protocol client against an in-process worker, every tool through the SDK's in-memory client with test tasks enabled, the entry point as a subprocess, result validators on synthetic data, the index builder (chunking, the format and each corruption it refuses, building against the worker, caching, its command line) `bridge_search` (the model gate, path confinement, reloads, rerank on and off), the reachability of every answer in the LLM evaluation, that a default search sees only part of its corpus and that its two-step question takes two steps, that the demo worker reaches none of the semantic evaluation's answers, that a tool waits out a full worker for its Retry-After within the call's budget or with its own backoff without one, before each of a search's jobs and within one budget, and says so when it stays full, the evaluation runner's command, isolation checks, scoring and per-backend launcher and index check without starting `claude`, a check that the server's model names match the worker's, that every path parameter says where it resolves and every position in a result says what it counts, and a check that every timing claim in a tool description is a measured median | | PHP contracts + HTTP | 192 checks per PHP version, a full fault-worker's Retry-After read through the client among them and a two-token `tokens-worker` where another token's read and cancel are 404 and the same idempotency key is two jobs; this **includes** the 148 contract checks (Retry-After from a loopback peer, present, absent, malformed and repeated, and a header block over 16 KiB refused), not 192 + 148; the contract checks include a fake transport's view of each request, the wait deadline across a transport failure, and that no file in `src/` outside `Laravel/` and `Symfony/` names a framework | | PSR-18 transport | The 148 contract checks plus 30 through `Psr18Transport` and in-repo fakes of the PSR-7, PSR-17 and PSR-18 interfaces (`tests/php/psr18_fakes.php`): a round trip with its method, URL, headers and body, a body at the 262144-byte cap and one over it read one byte past the cap and no further, five redirect statuses refused unread, a non-JSON or missing content type, malformed JSON, a 429's Retry-After from one header, a repeated one and none, a client exception and a body read failure as `transport_error` without their text; 178 checks by `php-psr18-tests` with `.cache/php-vendor` on the include path, once, in the MCP server job; `php-tests` skips them with one printed line; no real Symfony HttpClient or Guzzle runs | diff --git a/tests/python/test_jobs.py b/tests/python/test_jobs.py index 79718eb..6931067 100644 --- a/tests/python/test_jobs.py +++ b/tests/python/test_jobs.py @@ -1,5 +1,6 @@ import contextlib import hashlib +import io import json import math import multiprocessing @@ -16,11 +17,12 @@ from types import SimpleNamespace from unittest.mock import Mock, patch -from ai_bridge import jobs, tasks +from ai_bridge import jobs, log, tasks from ai_bridge import server as service from ai_bridge.jobs import CapacityError, JobStore, Settings, TERMINAL from ai_bridge.server import BridgeServer from ai_bridge.tasks import EMBED_DIMENSIONS, MAX_TEXTS, InvalidInput, _entities, _merge, execute, validate +from test_log import parse as parse_log class FailingDatabase: @@ -2195,5 +2197,105 @@ def record(directory, max_length, stride=0): for name, length, stride in jobs.WARM_UP["onnx"])) +class JobLogTests(unittest.TestCase): + """A job line when a job reaches a terminal status: its task, owner and status, and where its time went.""" + + def setUp(self): + self.stream = io.StringIO() + log.configure("json", self.stream) + self.addCleanup(log.close) + self.store = JobStore(Settings(concurrency=1, test_tasks=True)) + self.addCleanup(self.store.close) + + def wait(self, job_id, states=TERMINAL, timeout=5): + deadline = time.monotonic() + timeout + while self.store.get(job_id)["status"] not in states: + self.assertLess(time.monotonic(), deadline, f"job did not reach {states}") + time.sleep(0.01) + + def lines(self): + log.close() + self.text = self.stream.getvalue() + parsed = parse_log(self.text) + # Exactly one line per finished job: a second terminal line would vanish in the dict below. + ids = [line["job_id"] for line in parsed] + self.assertEqual(len(ids), len(set(ids)), ids) + return {line["job_id"]: line for line in parsed} + + def test_the_timings_add_up_to_the_job_s_life(self): + started = time.monotonic() + job = self.store.submit("test.delay", {"seconds": 0.3, "value": "sentinel-input"}) + self.wait(job["id"]) + elapsed_ms = (time.monotonic() - started) * 1000 + line = self.lines()[job["id"]] + self.assertEqual((line["task"], line["owner"], line["job_status"], line["level"]), + ("test.delay", "default", "succeeded", "info")) + self.assertGreaterEqual(line["queue_ms"], 0) + # The parent sees the start and the result on its 10 ms poll of the child, so either may read a little late. + self.assertGreaterEqual(line["run_ms"], 270) + self.assertLessEqual(line["queue_ms"] + line["run_ms"], elapsed_ms) + self.assertNotIn("sentinel-input", self.text) + + def test_a_job_that_never_started_has_no_queue_or_run_time(self): + running = self.store.submit("test.delay", {"seconds": 2}) + self.wait(running["id"], {"running"}) + queued = self.store.submit("test.delay", {"seconds": 0}) + self.store.cancel(queued["id"]) + self.store.cancel(running["id"]) + lines = self.lines() + self.assertEqual((lines[queued["id"]]["job_status"], lines[queued["id"]]["queue_ms"], + lines[queued["id"]]["run_ms"]), ("cancelled", None, None)) + self.assertEqual(lines[running["id"]]["job_status"], "cancelled") + self.assertIsNotNone(lines[running["id"]]["run_ms"]) + + def test_failures_and_timeouts_are_warnings(self): + # A warm child first, so the 300 ms deadline is spent running rather than waiting for one to start. + deadline = time.monotonic() + 10 + while self.store.readiness() is not None: + self.assertLess(time.monotonic(), deadline, "no child became ready") + time.sleep(0.01) + late = self.store.submit("test.delay", {"seconds": 5}, timeout_ms=300) + self.wait(late["id"]) + failing = self.store.submit("test.error", {}) + self.wait(failing["id"]) + crashing = self.store.submit("test.crash", {}) + self.wait(crashing["id"]) + lines = self.lines() + self.assertEqual([(lines[job["id"]]["job_status"], lines[job["id"]]["level"]) for job in (failing, crashing, late)], + [("failed", "warning")] * 2 + [("timed_out", "warning")]) + # The deadline counts from submission, so the queue and the run share its 300 ms. + self.assertGreaterEqual(lines[late["id"]]["queue_ms"] + lines[late["id"]]["run_ms"], 290) + for value in ("internal exception", "task_failed", "worker_crashed", "deadline_exceeded"): + self.assertNotIn(value, self.text) + + def test_a_drain_and_a_close_log_the_jobs_they_end_and_close_counts_them(self): + running = self.store.submit("test.delay", {"seconds": 5}) + self.wait(running["id"], {"running"}) + queued = self.store.submit("test.delay", {"seconds": 0}) + self.store.drain() + self.assertEqual(self.store.close(), 1) + lines = self.lines() + self.assertEqual((lines[queued["id"]]["job_status"], lines[queued["id"]]["queue_ms"]), ("failed", None)) + self.assertEqual(lines[running["id"]]["job_status"], "cancelled") + self.assertNotIn("service_restarted", self.text) + + def test_a_job_read_back_from_the_table_writes_nothing(self): + with tempfile.TemporaryDirectory() as state_dir: + first = JobStore(Settings(test_tasks=True, state_dir=state_dir)) + try: + job = first.submit("test.delay", {"seconds": 0}) + deadline = time.monotonic() + 5 + while first.get(job["id"])["status"] not in TERMINAL: + self.assertLess(time.monotonic(), deadline) + time.sleep(0.01) + finally: + first.close() + log.close() + self.stream = io.StringIO() + log.configure("json", self.stream) + JobStore(Settings(test_tasks=True, state_dir=state_dir)).close() + self.assertEqual(self.lines(), {}) + + if __name__ == "__main__": unittest.main() diff --git a/tests/python/test_log.py b/tests/python/test_log.py new file mode 100644 index 0000000..9350fc1 --- /dev/null +++ b/tests/python/test_log.py @@ -0,0 +1,239 @@ +import contextlib +import io +import json +import logging +import re +import threading +import time +import unittest + +from ai_bridge import log, server + +TIMESTAMP = re.compile(r"\d{4}-\d\d-\d\dT\d\d:\d\d:\d\d\.\d{3}Z") + + +@contextlib.contextmanager +def capture(): + """JSON logging into a buffer for the duration; the list it yields fills with the parsed lines once the block ends, + when the writer has written everything and logging is off again.""" + stream = io.StringIO() + lines = [] + log.configure("json", stream) + try: + yield lines + finally: + log.close() + # The raw text too, for searches through everything written. + capture.text = stream.getvalue() + lines.extend(parse(capture.text)) + + +def parse(text): + """Every line as JSON, each holding exactly the keys of its event; fails on anything else.""" + parsed = [] + for line in text.splitlines(): + record = json.loads(line) + assert isinstance(record, dict), line + assert set(record) == {"ts", "level", "event", *log.EVENTS[record["event"]]}, line + assert TIMESTAMP.fullmatch(record["ts"]), line + assert record["level"] in ("info", "warning"), line + parsed.append(record) + return parsed + + +class BlockedStream: + """A stderr whose writes wait until released, as a full pipe nobody reads would.""" + + def __init__(self): + self.released = threading.Event() + self.written = [] + + def write(self, text): + self.released.wait() + self.written.append(text) + + def flush(self): + pass + + +class BrokenStream: + def write(self, text): + raise OSError("stderr is gone") + + def flush(self): + raise OSError("stderr is gone") + + +class FormatterTests(unittest.TestCase): + def setUp(self): + self.addCleanup(log.close) + + def test_each_event_writes_exactly_its_keys_with_null_for_a_missing_one(self): + with capture() as lines: + log.event("request", method="GET", route="/healthz", status=200, duration_ms=1.5, owner="default") + log.event("job", job_id="a" * 32, task="rerank", owner="default", job_status="succeeded") + log.event("start", backend="lexical", concurrency=2, capacity=128, retention_seconds=300.0, + max_jobs=1000, drain_seconds=30, tokens=1) + log.event("drain", duration_ms=0.0, running=0, interrupted=False) + log.event("stop", cancelled=0) + self.assertEqual([line["event"] for line in lines], ["request", "job", "start", "drain", "stop"]) + self.assertEqual(lines[0], {**lines[0], "method": "GET", "route": "/healthz", "status": 200, + "duration_ms": 1.5, "owner": "default", "level": "info"}) + self.assertEqual((lines[1]["queue_ms"], lines[1]["run_ms"]), (None, None)) + self.assertEqual(lines[2]["tokens"], 1) + + def test_the_schema_is_the_documented_one(self): + # Written out here, not read from log.EVENTS, so a change to the schema has to change this test too. + expected = { + "request": ["method", "route", "status", "duration_ms", "owner"], + "job": ["job_id", "task", "owner", "job_status", "queue_ms", "run_ms"], + "start": ["backend", "concurrency", "capacity", "retention_seconds", "max_jobs", "drain_seconds", "tokens"], + "stop": ["cancelled"], + "drain": ["duration_ms", "running", "interrupted"], + } + with capture() as lines: + for name in expected: + log.event(name) + self.assertEqual({line["event"]: list(line) for line in lines}, + {name: ["ts", "level", "event", *keys] for name, keys in expected.items()}) + + def test_a_key_outside_the_event_s_schema_is_dropped(self): + with capture() as lines: + log.event("request", method="GET", route="/healthz", status=200, duration_ms=1, owner=None, + path="/healthz?token=sentinel", input="sentinel", error="sentinel", job_id="a" * 32) + log.event("stop", cancelled=0, message="sentinel") + self.assertNotIn("sentinel", capture.text) + self.assertNotIn("job_id", lines[0]) + + def test_a_value_outside_its_fixed_set_or_type_is_null(self): + with capture() as lines: + log.event("request", method="sentinel", route="/healthz?sentinel", status="200", duration_ms=float("nan"), + owner="sentinel token with spaces") + log.event("request", method="GET", route="/v1/jobs", status=True, duration_ms=-1, owner="x" * 65) + log.event("job", job_id="sentinel" * 4, task="sentinel", owner=["default"], job_status="sentinel", + queue_ms="1", run_ms=float("inf")) + log.event("start", backend="sentinel", concurrency=1.5, capacity=-1, retention_seconds="300", + max_jobs=True, drain_seconds=None, tokens="2") + log.event("drain", duration_ms="1", running=-1, interrupted=1) + self.assertNotIn("sentinel", capture.text) + # A route that is not a template and a method outside the fixed set are null rather than echoed. + self.assertEqual({key: lines[1][key] for key in log.EVENTS["request"]}, + {"method": "GET", "route": "/v1/jobs", "status": None, "duration_ms": None, "owner": None}) + for line in [lines[0]] + lines[2:]: + with self.subTest(event=line["event"]): + self.assertEqual({line[key] for key in log.EVENTS[line["event"]]}, {None}) + + def test_an_unknown_event_or_a_plain_message_writes_nothing(self): + with capture() as lines: + log.event("sentinel", status=200) + logging.getLogger("ai_bridge").warning("sentinel %s", "argument") + logging.getLogger("ai_bridge.jobs").error("sentinel", exc_info=RuntimeError("sentinel")) + try: + raise RuntimeError("sentinel") + except RuntimeError: + logging.getLogger("ai_bridge").exception("sentinel") + self.assertEqual(capture.text, "") + self.assertEqual(lines, []) + + def test_the_message_and_exception_of_an_event_never_reach_the_line(self): + with capture() as lines: + logging.getLogger("ai_bridge").error("sentinel", exc_info=RuntimeError("sentinel"), stack_info=True, + extra={"bridge": ("stop", {"cancelled": 1})}) + self.assertNotIn("sentinel", capture.text) + self.assertNotIn("Traceback", capture.text) + self.assertEqual(lines[0]["level"], "warning") + + def test_the_level_is_info_or_warning(self): + with capture() as lines: + log.event("stop", cancelled=0) + log.event("stop", warning=True, cancelled=1) + self.assertEqual([line["level"] for line in lines], ["info", "warning"]) + + def test_the_timestamp_is_utc_with_milliseconds(self): + before = time.time() + with capture() as lines: + log.event("stop", cancelled=0) + stamp = time.mktime(time.strptime(lines[0]["ts"][:19], "%Y-%m-%dT%H:%M:%S")) - time.timezone + self.assertLess(abs(stamp - before), 2) + + def test_the_owner_pattern_is_the_token_name_pattern(self): + self.assertEqual(log._NAME.pattern, server.TOKEN_NAME.pattern) + + +class ModeTests(unittest.TestCase): + def setUp(self): + self.addCleanup(log.close) + + def test_off_writes_nothing(self): + stream = io.StringIO() + log.configure("off", stream) + log.event("stop", cancelled=0) + log.close() + self.assertEqual(stream.getvalue(), "") + + def test_nothing_is_written_before_configure_or_after_close(self): + stream = io.StringIO() + log.event("stop", cancelled=0) + log.configure("json", stream) + log.close() + log.event("stop", cancelled=0) + self.assertEqual(stream.getvalue(), "") + + def test_an_unknown_mode_is_refused(self): + for mode in ("JSON", "on", "text", "1"): + with self.subTest(mode=mode), self.assertRaises(ValueError): + log.configure(mode, io.StringIO()) + + def test_configure_twice_keeps_one_writer(self): + first, second = io.StringIO(), io.StringIO() + log.configure("json", first) + log.configure("json", second) + log.event("stop", cancelled=0) + log.close() + self.assertEqual((first.getvalue(), len(second.getvalue().splitlines())), ("", 1)) + + +class BrokenStderrTests(unittest.TestCase): + def setUp(self): + self.addCleanup(log.close) + + def test_a_stream_that_raises_is_swallowed(self): + log.configure("json", BrokenStream()) + log.event("stop", cancelled=0) + log.close() + + def test_a_stream_that_blocks_never_holds_the_caller(self): + stream = BlockedStream() + log.configure("json", stream) + started = time.monotonic() + for _ in range(5000): + log.event("stop", cancelled=0) + self.assertLess(time.monotonic() - started, 5) + # A writer stuck on the stream does not hold close() for long either. + started = time.monotonic() + log.close() + self.assertLess(time.monotonic() - started, 3) + stream.released.set() + + def test_a_writer_left_with_a_full_queue_ends_once_its_stream_is_released(self): + stream = BlockedStream() + self.addCleanup(stream.released.set) + log.configure("json", stream) + _, lines, writer, _ = log._active + # One line held by the blocked write and a full queue behind it, so close() cannot queue its end. + log.event("stop", cancelled=0) + deadline = time.monotonic() + 5 + while lines.qsize(): + self.assertLess(time.monotonic(), deadline, "the writer never took the first line") + time.sleep(0.01) + for _ in range(log.QUEUE_LINES + 1): + log.event("stop", cancelled=0) + log.close() + stream.released.set() + writer.join(5) + self.assertFalse(writer.is_alive()) + self.assertEqual(len(stream.written), log.QUEUE_LINES + 1) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/python/test_server.py b/tests/python/test_server.py index 335ba83..599fcb8 100644 --- a/tests/python/test_server.py +++ b/tests/python/test_server.py @@ -1,11 +1,13 @@ import contextlib import http.client +import io import json import os import re import signal import socket import sqlite3 +import struct import subprocess import sys import tempfile @@ -15,9 +17,10 @@ import uuid from unittest.mock import MagicMock, patch -from ai_bridge import jobs, metrics, server, tasks +from ai_bridge import jobs, log, metrics, server, tasks from ai_bridge.jobs import Settings from ai_bridge.server import BridgeServer +from test_log import BlockedStream, BrokenStream, parse as parse_log TOKEN = "test-only-bridge-token-never-use-in-production" @@ -1361,3 +1364,349 @@ def test_no_token_value_reaches_the_job_table_and_a_restored_job_keeps_its_owner self.assertEqual(self.call("GET", "/v1/jobs/" + job["id"], OTHER)[0], 404) self.assertEqual(self.call("GET", "/v1/jobs/" + job["id"], APP)[1]["status"], "succeeded") self.assertEqual(self.submit(APP, "table-key")[1]["id"], job["id"]) + + +SENTINEL = "sentinel-7f3a-must-never-be-logged" + + +def settle(bridge, timeout=2): + """Until every connection slot is free again: a slot is released after its request line is queued.""" + deadline = time.monotonic() + timeout + while bridge.slots._value != 32: + if time.monotonic() > deadline: + raise AssertionError("a connection was still being handled") + time.sleep(0.01) + + +class LoggingHttpTests(unittest.TestCase): + """A real server on a loopback port with BRIDGE_LOG=json's writer capturing what it would put on stderr.""" + + def setUp(self): + self.stream = io.StringIO() + log.configure("json", self.stream) + self.addCleanup(log.close) + self.server = BridgeServer(("127.0.0.1", 0), {"app": APP, "other": OTHER}, + Settings(concurrency=1, capacity=16, test_tasks=True), drain_seconds=0) + self.thread = threading.Thread(target=self.server.serve_forever, kwargs={"poll_interval": 0.05}, daemon=True) + self.thread.start() + self.addCleanup(self.server.server_close) + self.addCleanup(self.thread.join) + self.addCleanup(self.server.shutdown) + + def lines(self): + """Every line written so far, each checked against its event's schema; logging stops here.""" + settle(self.server) + log.close() + self.text = self.stream.getvalue() + return parse_log(self.text) + + def call(self, method, path, body=None, token=APP, headers=None): + connection = http.client.HTTPConnection(*self.server.server_address, timeout=2) + try: + sent = {"Content-Type": "application/json", **({"Authorization": "Bearer " + token} if token else {}), + **(headers or {})} + connection.request(method, path, body, sent) + response = connection.getresponse() + return response.status, json.loads(response.read()) + finally: + connection.close() + + def ended(self, job_id): + deadline = time.monotonic() + 5 + while self.server.jobs.get(job_id, "app")["status"] not in jobs.TERMINAL: + self.assertLess(time.monotonic(), deadline, "the job never ended") + time.sleep(0.01) + + def test_a_request_line_per_route_with_its_template_and_never_the_path(self): + job_id = "b" * 32 + calls = [("GET", "/healthz", None, 200), ("GET", "/readyz", None, None), ("GET", "/metrics", None, 200), + ("GET", "/v1/jobs/" + job_id, None, 404), ("POST", "/v1/jobs/" + job_id + "/cancel", "{}", 404), + ("POST", "/v1/jobs", "{}", 400), ("GET", "/" + SENTINEL, None, 404), + ("GET", "/healthz?" + SENTINEL, None, 404), ("GET", "/v1/jobs/" + SENTINEL, None, 404), + ("DELETE", "/v1/jobs", None, 501)] + for method, path, body, status in calls: + connection = http.client.HTTPConnection(*self.server.server_address, timeout=2) + try: + connection.request(method, path, body, {"Authorization": "Bearer " + APP, + "Content-Type": "application/json"}) + response = connection.getresponse() + response.read() + if status is not None: + self.assertEqual(response.status, status) + finally: + connection.close() + requests = [line for line in self.lines() if line["event"] == "request"] + self.assertEqual([(line["method"], line["route"]) for line in requests], + [("GET", "/healthz"), ("GET", "/readyz"), ("GET", "/metrics"), ("GET", "/v1/jobs/{id}"), + ("POST", "/v1/jobs/{id}/cancel"), ("POST", "/v1/jobs"), ("GET", "other"), ("GET", "other"), + ("GET", "other"), ("other", "/v1/jobs")]) + self.assertEqual([line["status"] for line in requests][3:], [404, 404, 400, 404, 404, 404, 501]) + # The unsupported method is refused before authentication, so it has no owner. + self.assertEqual([line["owner"] for line in requests], ["app"] * 9 + [None]) + self.assertEqual([line["level"] == "warning" for line in requests], [line["status"] >= 500 for line in requests]) + for line in requests: + self.assertGreaterEqual(line["duration_ms"], 0) + self.assertNotIn(SENTINEL, self.text) + self.assertNotIn(job_id, self.text) + + def test_a_401_is_logged_with_no_owner(self): + self.assertEqual(self.call("GET", "/healthz", token=None)[0], 401) + self.assertEqual(self.call("GET", "/healthz", token=SENTINEL + "x" * 32)[0], 401) + lines = self.lines() + self.assertEqual([(line["status"], line["owner"], line["route"]) for line in lines], + [(401, None, "/healthz")] * 2) + self.assertNotIn(SENTINEL, self.text) + + def test_each_token_s_requests_carry_its_name(self): + self.call("GET", "/healthz", token=OTHER) + self.call("GET", "/healthz", token=APP) + self.assertEqual([line["owner"] for line in self.lines()], ["other", "app"]) + + def test_no_input_result_token_header_query_key_or_exception_text_is_written(self): + body = json.dumps({"task": "test.inspect", "input": {"value": [SENTINEL]}, "idempotency_key": SENTINEL + "-k"}) + headers = {"X-Sentinel": SENTINEL, "Cookie": SENTINEL} + # A query string is not the route, so this is read as an unknown route's body, not a submission. + self.assertEqual(self.call("POST", "/v1/jobs?" + SENTINEL, body, headers=headers)[0], 400) + status, inspected = self.call("POST", "/v1/jobs", body, headers=headers) + self.assertEqual(status, 202) + status, failing = self.call("POST", "/v1/jobs", json.dumps({"task": "test.error", "input": {}})) + self.assertEqual(status, 202) + status, ranked = self.call("POST", "/v1/jobs", json.dumps( + {"task": "rerank", "input": {"query": SENTINEL, "documents": [SENTINEL, "other text"]}})) + self.assertEqual(status, 202) + for job in (inspected, failing, ranked): + self.ended(job["id"]) + self.assertEqual(self.call("GET", "/v1/jobs/" + failing["id"])[1]["status"], "failed") + result = self.call("GET", "/v1/jobs/" + ranked["id"])[1]["result"] + self.assertEqual(self.call("POST", "/v1/jobs", "{" + SENTINEL)[0], 400) + lines = self.lines() + for value in (SENTINEL, APP, OTHER, "Bearer", "internal exception", "Traceback", "RuntimeError", + "Task did not complete", "task_failed", result["model"], "test-only", "?"): + with self.subTest(value=value): + self.assertNotIn(value, self.text) + ended = {line["job_id"]: line for line in lines if line["event"] == "job"} + self.assertEqual({job_id: (line["task"], line["job_status"], line["owner"]) for job_id, line in ended.items()}, + {inspected["id"]: ("test.inspect", "succeeded", "app"), + failing["id"]: ("test.error", "failed", "app"), ranked["id"]: ("rerank", "succeeded", "app")}) + self.assertEqual(ended[failing["id"]]["level"], "warning") + self.assertEqual(ended[ranked["id"]]["level"], "info") + + def test_refusals_before_and_around_the_handler_are_logged(self): + # 413 for a body over the limit, 431 for a header block or a request line over its own. + self.assertEqual(self.call("POST", "/v1/jobs", " " * 262145)[0], 413) + head = b"GET /healthz HTTP/1.1\r\nHost: x\r\nX-Pad: " + SENTINEL.encode() * 600 + b"\r\n\r\n" + self.assertEqual(exchange(self.server.server_address, head)[0], 431) + long_line = b"GET /" + SENTINEL.encode() * 300 + b" HTTP/1.1\r\n\r\n" + self.assertEqual(exchange(self.server.server_address, long_line)[0], 431) + # 408: a body announced and never sent, so the five-second read deadline passes. + self.assertEqual(exchange(self.server.server_address, + b"POST /v1/jobs HTTP/1.1\r\nHost: x\r\nAuthorization: Bearer " + APP.encode() + + b"\r\nContent-Type: application/json\r\nContent-Length: 10\r\n\r\n", wait=8)[0], 408) + lines = self.lines() + self.assertEqual([(line["method"], line["route"], line["status"], line["owner"]) for line in lines], + [("POST", "/v1/jobs", 413, "app"), ("GET", "/healthz", 431, None), (None, "other", 431, None), + ("POST", "/v1/jobs", 408, "app")]) + self.assertGreaterEqual(lines[3]["duration_ms"], 4000) + self.assertNotIn(SENTINEL, self.text) + + def test_a_busy_503_is_logged_without_anything_read(self): + idle = [socket.create_connection(self.server.server_address, timeout=2) for _ in range(32)] + try: + deadline = time.monotonic() + 5 + while self.server.slots._value: + self.assertLess(time.monotonic(), deadline, "the idle connections were not all taken") + time.sleep(0.01) + self.assertEqual(exchange(self.server.server_address, b"GET /" + SENTINEL.encode())[0], 503) + finally: + for connection in idle: + connection.close() + busy = [line for line in self.lines() if line["status"] == 503] + self.assertEqual([(line["method"], line["route"], line["owner"], line["level"]) for line in busy], + [(None, "other", None, "warning")]) + self.assertNotIn(SENTINEL, self.text) + + def test_the_drain_says_how_many_jobs_the_deadline_cut_short(self): + job = self.call("POST", "/v1/jobs", json.dumps({"task": "test.delay", "input": {"seconds": 5}}))[1] + deadline = time.monotonic() + 5 + while self.server.jobs.get(job["id"], "app")["status"] != "running": + self.assertLess(time.monotonic(), deadline) + time.sleep(0.01) + self.server.stop() + self.thread.join(5) + drained = [line for line in self.lines() if line["event"] == "drain"] + self.assertEqual([(line["running"], line["interrupted"], line["level"]) for line in drained], + [(1, False, "warning")]) + self.assertLess(drained[0]["duration_ms"], 2000) + + def test_a_drain_with_nothing_running_is_info(self): + self.server.stop() + self.thread.join(5) + drained = [line for line in self.lines() if line["event"] == "drain"] + self.assertEqual([(line["running"], line["interrupted"], line["level"]) for line in drained], + [(0, False, "info")]) + + def test_a_connection_that_times_out_before_its_head_is_read_writes_no_line(self): + silent = socket.create_connection(self.server.server_address, timeout=10) + half = socket.create_connection(self.server.server_address, timeout=10) + try: + half.sendall(b"GET /healthz HTTP/1.1\r\nHost: x\r\n") + # The handler's five-second idle timeout ends both without a reply. + for connection in (silent, half): + self.assertEqual(connection.recv(1), b"") + settle(self.server, 8) + finally: + silent.close() + half.close() + self.assertEqual(self.lines(), []) + + +class PeerResetTests(unittest.TestCase): + """A peer that resets mid-request puts nothing but JSON lines on stderr, and nothing at all with logging off.""" + + def reset_peer(self, mode): + stream = io.StringIO() + with patch.object(sys, "stderr", stream): + log.configure(mode, stream) + self.addCleanup(log.close) + bridge = BridgeServer(("127.0.0.1", 0), TOKEN, Settings()) + thread = threading.Thread(target=bridge.serve_forever, kwargs={"poll_interval": 0.05}, daemon=True) + thread.start() + try: + for head in (b"GET /healthz HTTP/1.1\r\nHost: x\r\n", + b"GET /healthz HTTP/1.1\r\nHost: x\r\nAuthorization: Bearer " + TOKEN.encode() + b"\r\n\r\n"): + connection = socket.create_connection(bridge.server_address, timeout=2) + connection.sendall(head) + time.sleep(0.1) # lets the server read what was sent, so the reset lands mid-request + # Linger 0: close() sends a reset instead of a FIN. + connection.setsockopt(socket.SOL_SOCKET, socket.SO_LINGER, struct.pack("ii", 1, 0)) + connection.close() + settle(bridge) + finally: + bridge.shutdown() + thread.join() + bridge.server_close() + log.close() + return stream.getvalue() + + def test_json_writes_json_lines_only(self): + text = self.reset_peer("json") + parse_log(text) + self.assertNotIn("Traceback", text) + self.assertNotIn("127.0.0.1", text) + + def test_off_writes_nothing(self): + self.assertEqual(self.reset_peer("off"), "") + + +class BrokenLogStreamHttpTests(unittest.TestCase): + """A stderr that raises or never drains costs the requests nothing.""" + + def serve(self, stream): + log.configure("json", stream) + self.addCleanup(log.close) + bridge = BridgeServer(("127.0.0.1", 0), TOKEN, Settings(capacity=64)) + thread = threading.Thread(target=bridge.serve_forever, kwargs={"poll_interval": 0.05}, daemon=True) + thread.start() + self.addCleanup(bridge.server_close) + self.addCleanup(thread.join) + self.addCleanup(bridge.shutdown) + started = time.monotonic() + for _ in range(20): + connection = http.client.HTTPConnection(*bridge.server_address, timeout=2) + try: + connection.request("POST", "/v1/jobs", json.dumps({"task": "embed", "input": {"texts": ["x"]}}), + {"Authorization": "Bearer " + TOKEN, "Content-Type": "application/json"}) + self.assertEqual(connection.getresponse().status, 202) + finally: + connection.close() + self.assertLess(time.monotonic() - started, 5) + + def test_a_stderr_that_raises(self): + self.serve(BrokenStream()) + + def test_a_stderr_that_blocks(self): + stream = BlockedStream() + self.addCleanup(stream.released.set) + self.serve(stream) + + +class LoggingStartTests(unittest.TestCase): + """main() reads BRIDGE_LOG, writes start once the port is bound and stop once its children are gone.""" + + def main(self, environment, serve=None): + stream = io.StringIO() + self.addCleanup(log.close) + with patch.dict(os.environ, environment), patch.object(sys, "argv", ["ai_bridge.server", "--port", "0"]), \ + patch.object(server.signal, "signal"), patch.object(sys, "stderr", stream), \ + patch.object(server.BridgeServer, "serve_forever", serve or MagicMock()): + for name in [name for name in os.environ if name.startswith("BRIDGE_") and name not in environment]: + del os.environ[name] + server.main() + return stream.getvalue() + + def test_json_writes_start_and_stop_with_safe_settings_only(self): + text = self.main({"BRIDGE_LOG": "json", "BRIDGE_TOKENS": f"app={APP},{SENTINEL}={OTHER}", + "BRIDGE_CAPACITY": "9", "BRIDGE_DRAIN_SECONDS": "7"}) + lines = parse_log(text) + self.assertEqual([line["event"] for line in lines], ["start", "stop"]) + self.assertEqual({key: lines[0][key] for key in log.EVENTS["start"]}, + {"backend": "lexical", "concurrency": 2, "capacity": 9, "retention_seconds": 300.0, + "max_jobs": 1000, "drain_seconds": 7, "tokens": 2}) + self.assertEqual((lines[1]["cancelled"], lines[1]["level"]), (0, "info")) + for value in (APP, OTHER, SENTINEL, "app"): + self.assertNotIn(value, text) + + def serve_with_a_running_job(self, signals): + """A serve_forever that first starts a long job, then signals the server and serves until it stops.""" + serve_forever = server.BridgeServer.serve_forever + + def serve(bridge, poll_interval=0.5): + job = bridge.jobs.submit("test.delay", {"seconds": 10}) + deadline = time.monotonic() + 10 + while bridge.jobs.get(job["id"])["status"] != "running": + self.assertLess(time.monotonic(), deadline, "the job never started") + time.sleep(0.01) + for _ in range(signals): + bridge.stop() + serve_forever(bridge, poll_interval) + + return serve + + def test_a_job_running_at_shutdown_makes_stop_a_warning(self): + text = self.main({"BRIDGE_LOG": "json", "BRIDGE_TOKEN": TOKEN, "BRIDGE_TEST_TASKS": "1", + "BRIDGE_DRAIN_SECONDS": "0"}, self.serve_with_a_running_job(1)) + lines = parse_log(text) + self.assertEqual([line["event"] for line in lines], ["start", "drain", "job", "stop"]) + self.assertEqual((lines[1]["running"], lines[1]["interrupted"]), (1, False)) + self.assertGreaterEqual(lines[3]["cancelled"], 1) + self.assertEqual(lines[3]["level"], "warning") + + def test_a_second_signal_writes_one_interrupted_drain_line_before_stop(self): + snapshot = jobs.JobStore.snapshot + + def slow_snapshot(store): + # Widens the moment between the interruption and the drain line, which the stop must wait out. + time.sleep(0.3) + return snapshot(store) + + with patch.object(jobs.JobStore, "snapshot", slow_snapshot): + text = self.main({"BRIDGE_LOG": "json", "BRIDGE_TOKEN": TOKEN, "BRIDGE_TEST_TASKS": "1", + "BRIDGE_DRAIN_SECONDS": "30"}, self.serve_with_a_running_job(2)) + lines = parse_log(text) + self.assertEqual([line["event"] for line in lines], ["start", "drain", "job", "stop"]) + self.assertEqual((lines[1]["interrupted"], lines[1]["running"]), (True, 1)) + self.assertEqual((lines[3]["cancelled"], lines[3]["level"]), (1, "warning")) + + def test_off_or_unset_writes_nothing(self): + for environment in ({}, {"BRIDGE_LOG": "off"}, {"BRIDGE_LOG": ""}): + with self.subTest(environment=environment): + self.assertEqual(self.main({"BRIDGE_TOKEN": TOKEN, **environment}), "") + + def test_any_other_value_stops_the_start_naming_the_variable(self): + for value in ("JSON", "on", "text", SENTINEL): + with self.subTest(value=value), patch.object(server, "BridgeServer") as bridge: + with self.assertRaises(SystemExit) as raised: + self.main({"BRIDGE_TOKEN": TOKEN, "BRIDGE_LOG": value}) + self.assertIn("BRIDGE_LOG", str(raised.exception.code)) + self.assertNotIn(SENTINEL, str(raised.exception.code)) + bridge.assert_not_called() diff --git a/worker/ai_bridge/jobs.py b/worker/ai_bridge/jobs.py index f3806ad..cfad449 100644 --- a/worker/ai_bridge/jobs.py +++ b/worker/ai_bridge/jobs.py @@ -21,7 +21,7 @@ from dataclasses import dataclass from pathlib import Path -from . import metrics, tasks +from . import log, metrics, tasks from .tasks import MODEL_NAMES, TEST_TASKS, execute, validate STATUSES = ("succeeded", "failed", "timed_out", "cancelled") @@ -405,6 +405,11 @@ def _finish(self, job, status, result=None, code=None): # A job read back from the table finished in an earlier process and never ends here. self._ended[job["_owner"], job["task"], status] += 1 metrics.observe(self._durations[job["task"]], job["_finished"] - job["_submitted_at"]) + started = job.get("_started_at") + log.event("job", warning=status in ("failed", "timed_out"), job_id=job["id"], task=job["task"], + owner=job["_owner"], job_status=status, + queue_ms=None if started is None else round((started - job["_submitted_at"]) * 1000, 3), + run_ms=None if started is None else round((job["_finished"] - started) * 1000, 3)) def _abort(self, job, status, code=None): """End a job before its child has: the child is retired, since it may be halfway through the input.""" @@ -621,7 +626,8 @@ def _drain_message(self, worker, message): if kind == "started": # Not written through: a restart fails a running job exactly as it fails a queued one. job["status"] = "running" - metrics.observe(self._waits[job["task"]], time.monotonic() - job["_submitted_at"]) + job["_started_at"] = time.monotonic() + metrics.observe(self._waits[job["task"]], job["_started_at"] - job["_submitted_at"]) return False if kind == "progress": job["progress"] = value @@ -656,6 +662,7 @@ def _reap(self, retiring): self._retiring.remove(worker) def close(self): + """Stop every child; returns how many unfinished jobs this ended as cancelled.""" self._stopping.set() self._thread.join(timeout=2) with self._lock: @@ -664,9 +671,9 @@ def close(self): if self._db is not None: self._db.close() self._db = None - for job in self._jobs.values(): - if job["status"] not in TERMINAL: - self._abort(job, "cancelled") + unfinished = [job for job in self._jobs.values() if job["status"] not in TERMINAL] + for job in unfinished: + self._abort(job, "cancelled") for worker in list(self._pool): self._retire(worker, stop=True) retiring = list(self._retiring) @@ -684,3 +691,4 @@ def close(self): self._thread.join() with self._lock: self._retiring.clear() + return len(unfinished) diff --git a/worker/ai_bridge/log.py b/worker/ai_bridge/log.py new file mode 100644 index 0000000..26c5c91 --- /dev/null +++ b/worker/ai_bridge/log.py @@ -0,0 +1,181 @@ +"""Optional structured logging: JSON lines on stderr with BRIDGE_LOG=json, nothing at all with off (the default). +A line holds only its event's schema keys, each value from a fixed set, a number, a job ID or a token's name, so no +input, result, header, query string, token, error message or exception text can reach it, whatever a caller passes. +Writing happens on a thread of its own behind a bounded queue: a stderr that blocks or fails never holds or fails a +request; its lines are dropped instead.""" + +import json +import logging +import math +import queue +import re +import sys +import threading +import time + +from .tasks import MODEL_NAMES, TEST_TASKS + +MODES = ("off", "json") +# Every key a line of each event carries, in this order, null when not given or not valid. +EVENTS = { + "request": ("method", "route", "status", "duration_ms", "owner"), + "job": ("job_id", "task", "owner", "job_status", "queue_ms", "run_ms"), + "start": ("backend", "concurrency", "capacity", "retention_seconds", "max_jobs", "drain_seconds", "tokens"), + "stop": ("cancelled",), + "drain": ("duration_ms", "running", "interrupted"), +} +METHODS = ("GET", "POST", "other") +ROUTES = ("/healthz", "/readyz", "/metrics", "/v1/jobs", "/v1/jobs/{id}", "/v1/jobs/{id}/cancel", "other") +STATUSES = ("succeeded", "failed", "timed_out", "cancelled") +TASKS = frozenset(task for names in MODEL_NAMES.values() for task in names) | frozenset(TEST_TASKS) +# server.TOKEN_NAME, which this module cannot import; a test keeps the two equal. +_NAME = re.compile(r"[A-Za-z0-9_.-]{1,64}") +_JOB_ID = re.compile(r"[a-f0-9]{32}") +QUEUE_LINES = 1024 + + +def _count(value): + return type(value) is int and value >= 0 + + +def _duration(value): + return type(value) in (int, float) and math.isfinite(value) and value >= 0 + + +def _matches(pattern): + return lambda value: type(value) is str and pattern.fullmatch(value) is not None + + +def _among(choices): + return lambda value: type(value) is str and value in choices + + +# What each key's value must be; anything else is written as null. +_VALID = { + "method": _among(METHODS), "route": _among(ROUTES), + "status": lambda value: type(value) is int and 100 <= value <= 599, + "duration_ms": _duration, "queue_ms": _duration, "run_ms": _duration, + "owner": _matches(_NAME), "job_id": _matches(_JOB_ID), "task": _among(TASKS), "job_status": _among(STATUSES), + "backend": _among(MODEL_NAMES), "retention_seconds": _duration, + "concurrency": _count, "capacity": _count, "max_jobs": _count, "drain_seconds": _count, "tokens": _count, + "cancelled": _count, "running": _count, "interrupted": lambda value: type(value) is bool, +} + + +def _event(record): + """The (event, fields) a record carries, or None for a record event() did not make.""" + carried = getattr(record, "bridge", None) + if type(carried) is tuple and len(carried) == 2 and carried[0] in EVENTS and type(carried[1]) is dict: + return carried + return None + + +class JsonFormatter(logging.Formatter): + """One JSON object from the record's event and fields alone: the message, its arguments, an exception and a stack + are never read.""" + + def format(self, record): + name, fields = _event(record) + moment = time.gmtime(record.created) + line = { + "ts": time.strftime("%Y-%m-%dT%H:%M:%S", moment) + ".%03dZ" % int(record.msecs), + "level": "warning" if record.levelno >= logging.WARNING else "info", + "event": name, + } + for key in EVENTS[name]: + value = fields.get(key) + line[key] = value if value is not None and _VALID[key](value) else None + return json.dumps(line, separators=(",", ":")) + + +class _Enqueue(logging.Handler): + """Formats in the caller and hands the line over without waiting; a full queue drops it.""" + + def __init__(self, lines): + super().__init__() + self._lines = lines + self.setFormatter(JsonFormatter()) + self.addFilter(lambda record: _event(record) is not None) + + def emit(self, record): + try: + self._lines.put_nowait(self.format(record)) + except Exception: + pass + + def handleError(self, record): + # The standard handler prints the exception to stderr, which is what must never happen here. + pass + + +def _write(lines, stream, stopped): + while True: + try: + line = lines.get(timeout=0.1) + except queue.Empty: + # close() could not queue the end when the queue was full: it ends here once the queue is empty. + if stopped.is_set(): + return + continue + if line is None: + return + try: + stream.write(line + "\n") + stream.flush() + except Exception: + pass + + +_logger = logging.getLogger("ai_bridge") +_logger.propagate = False +_logger.setLevel(logging.INFO) +# Without a handler of its own the standard library's last resort would print warnings to stderr. +_logger.addHandler(logging.NullHandler()) +_lock = threading.Lock() +_active = None # (handler, queue, writer thread, stop event) while json is on + + +def configure(mode, stream=None): + """Start writing JSON lines to stream (stderr when None) for json; off writes nothing. Replaces an earlier one.""" + global _active + if mode not in MODES: + raise ValueError("log mode must be off or json") + close() + if mode == "off": + return + lines = queue.Queue(QUEUE_LINES) + stopped = threading.Event() + writer = threading.Thread(target=_write, args=(lines, sys.stderr if stream is None else stream, stopped), + daemon=True) + writer.start() + handler = _Enqueue(lines) + with _lock: + _active = (handler, lines, writer, stopped) + _logger.addHandler(handler) + + +def close(): + """Stop logging and let the writer finish what is queued, waiting a moment at most for a stream that blocks.""" + global _active + with _lock: + active, _active = _active, None + if active is None: + return + handler, lines, writer, stopped = active + _logger.removeHandler(handler) + stopped.set() + try: + lines.put(None, timeout=1) + except queue.Full: + return + writer.join(1) + + +def event(name, warning=False, **fields): + """Write one line of this event; a key outside its schema is dropped and a value of the wrong kind is null.""" + if _active is None: + return + try: + _logger.log(logging.WARNING if warning else logging.INFO, name, extra={"bridge": (name, fields)}) + except Exception: + pass diff --git a/worker/ai_bridge/server.py b/worker/ai_bridge/server.py index 7fea9e5..cb667f4 100644 --- a/worker/ai_bridge/server.py +++ b/worker/ai_bridge/server.py @@ -14,6 +14,7 @@ from dataclasses import replace from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from . import log from .jobs import DEFAULT_OWNER, CapacityError, DrainingError, JobStore, Settings from .metrics import CONTENT_TYPE, render from .tasks import missing_models @@ -21,6 +22,10 @@ MAX_BODY = 262144 MAX_REQUEST_LINE = 8192 MAX_HEADER_BLOCK = 16384 +# What a request line names as the route: the template a path matches, never the path itself. +_ROUTES = {"/healthz": "/healthz", "/readyz": "/readyz", "/metrics": "/metrics", "/v1/jobs": "/v1/jobs"} +_JOB_ROUTES = ((re.compile(r"/v1/jobs/[a-f0-9]{32}"), "/v1/jobs/{id}"), + (re.compile(r"/v1/jobs/[a-f0-9]{32}/cancel"), "/v1/jobs/{id}/cancel")) _BUSY_BODY = b'{"error":{"code":"service_busy","message":"Every connection slot is taken"}}' BUSY = (b"HTTP/1.0 503 Service Unavailable\r\nContent-Type: application/json\r\nContent-Length: %d\r\n" b"Cache-Control: no-store\r\nRetry-After: 1\r\nConnection: close\r\n\r\n%s" % (len(_BUSY_BODY), _BUSY_BODY)) @@ -158,8 +163,10 @@ def __init__(self, address, tokens, settings=None, drain_seconds=30): # The header each name is presented with; every request is compared with all of them. self.credentials = tuple((name, ("Bearer " + token).encode("ascii")) for name, token in tokens.items()) self.drain_seconds = drain_seconds + self.cancelled = 0 # unfinished jobs the stop ended, for the stop line self._signalled_at = None self._interrupted = threading.Event() + self._drainer = None self.slots = threading.BoundedSemaphore(32) # The children the store starts inherit this process's environment block as it is at that moment, and # removing a variable later leaves it in /proc//environ; so the tokens leave before any child exists. @@ -181,6 +188,7 @@ def process_request(self, request, address): except OSError: pass self.shutdown_request(request) + log.event("request", warning=True, method=None, route="other", status=503, duration_ms=0, owner=None) return try: super().process_request(request, address) @@ -194,9 +202,16 @@ def process_request_thread(self, request, address): finally: self.slots.release() + def handle_error(self, request, client_address): + # The default prints the peer's address and a traceback to stderr, which takes only JSON lines or nothing. + pass + def server_close(self): super().server_close() - self.jobs.close() + if self._drainer is not None: + # A second signal stops serving before the drain thread has written its line, which goes before stop's. + self._drainer.join(1) + self.cancelled = self.jobs.close() def stop(self): """A signal's handler: the first starts the drain, a second ends it at once.""" @@ -206,7 +221,8 @@ def stop(self): threading.Thread(target=self.shutdown, daemon=True).start() return self._signalled_at = time.monotonic() - threading.Thread(target=self._drain_then_stop, daemon=True).start() + self._drainer = threading.Thread(target=self._drain_then_stop, daemon=True) + self._drainer.start() def _drain_then_stop(self): """Until running jobs end or the deadline passes, then stop serving; server_close ends what still runs.""" @@ -216,6 +232,9 @@ def _drain_then_stop(self): left = deadline - time.monotonic() if left <= 0 or self._interrupted.wait(min(left, 0.05)): break + running = self.jobs.snapshot()["running"] + log.event("drain", warning=running > 0, duration_ms=round((time.monotonic() - self._signalled_at) * 1000, 3), + running=running, interrupted=self._interrupted.is_set()) self.shutdown() @@ -229,12 +248,39 @@ def setup(self): self.rfile = _Head(self.rfile) def handle_one_request(self): + started = time.monotonic() + self.command = self.path = self._owner = self._status = None try: super().handle_one_request() except HeadTooLarge: # A long request line is refused before the request is parsed, so these may not be set yet. self.requestline, self.request_version = "", "HTTP/1.0" self._error(431, "headers_too_large", "Request line or header block exceeds its limit") + finally: + # After the reply is written; a connection that ended before any reply has no line. + if self._status is not None: + log.event("request", warning=self._status >= 500, method=self._method(), route=self._route(), + status=self._status, duration_ms=round((time.monotonic() - started) * 1000, 3), + owner=self._owner) + + def _method(self): + if self.command is None: + return None + return self.command if self.command in ("GET", "POST") else "other" + + def _route(self): + if self.path in _ROUTES: + return _ROUTES[self.path] + if self.path is not None: + for pattern, route in _JOB_ROUTES: + if pattern.fullmatch(self.path): + return route + return "other" + + def send_response(self, code, message=None): + # The standard library's own refusals pass an HTTPStatus. + self._status = int(code) + super().send_response(code, message) def log_message(self, *_): # Do not log URLs, credentials, inputs or model results. @@ -273,6 +319,7 @@ def _authorized(self): if len(values) != 1 or owner is None: self._error(401, "unauthorized", "Service authentication required") return None + self._owner = owner return owner def _body(self): @@ -396,6 +443,9 @@ def main(): drain_seconds = _number("BRIDGE_DRAIN_SECONDS", "30", int) if not 0 <= drain_seconds <= 300: raise SystemExit("BRIDGE_DRAIN_SECONDS must be an integer from 0 to 300") + log_mode = os.environ.get("BRIDGE_LOG") or "off" + if log_mode not in log.MODES: + raise SystemExit("BRIDGE_LOG must be off or json") # Without its files every job needing a model would fail as task_failed, which cannot say why: a cache fetched # before a model was added has no directory for it. missing = missing_models(settings.model_dir) if settings.backend == "onnx" else [] @@ -407,13 +457,21 @@ def main(): except sqlite3.Error: raise SystemExit(f"BRIDGE_STATE_DIR {settings.state_dir} must be a directory the service can write " "jobs.sqlite in") from None - with bridge as server: - def stop(*_): - server.stop() - - signal.signal(signal.SIGTERM, stop) - signal.signal(signal.SIGINT, stop) - server.serve_forever(poll_interval=0.1) + log.configure(log_mode) + try: + log.event("start", backend=settings.backend, concurrency=settings.concurrency, capacity=settings.capacity, + retention_seconds=settings.retention_seconds, max_jobs=settings.max_jobs, + drain_seconds=drain_seconds, tokens=len(tokens)) + with bridge as server: + def stop(*_): + server.stop() + + signal.signal(signal.SIGTERM, stop) + signal.signal(signal.SIGINT, stop) + server.serve_forever(poll_interval=0.1) + log.event("stop", warning=bool(bridge.cancelled), cancelled=bridge.cancelled) + finally: + log.close() if __name__ == "__main__":