From e31282c02d6a01cbfb261a9dc2e7b53cc712fcd0 Mon Sep 17 00:00:00 2001 From: kert Date: Sun, 25 Oct 2026 14:40:31 -0400 Subject: [PATCH] =?UTF-8?q?feat(bcda):=20/metrics=20for=20the=20BCDA=20man?= =?UTF-8?q?ager=20=E2=80=94=20OTel=20instruments=20on=20a=20second=20port,?= =?UTF-8?q?=20scraped=20by=20Prometheus?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit bcda.ui.metrics follows the #579 convention: perf.meter hands back an OTel meter + PrometheusMetricReader when STACK_TELEMETRY is on and a no-op MockMeter otherwise. Streamlit cannot mount a route and only executes the app script when a browser session opens, so the registry is published from a second port (BCDA_METRICS_PORT, 8502) by a new `python -m bcda.ui` entrypoint that starts the metrics server first and then boots Streamlit's CLI in the same process; the image CMD and `stack bcda ui` use it. Prometheus scrapes bcda:8502/metrics directly (infra/prometheus/targets/services.yml) like api and llm. Instruments: stack_bcda_runs_total{kind,state}, stack_bcda_runs_inflight {kind}, stack_bcda_run_seconds{kind} (recorded by the runner, which now labels runs export / adopt / load / vendor_check / other), stack_bcda_page_views_total{page}, and two observable gauges read at scrape time, stack_bcda_store_jobs{status} and stack_bcda_store_bytes. Attribute values are closed sets. The image installs the perf extra; the compose service joins the observability network with OTEL_SERVICE_NAME=bcda. polars joins the bcda extra: the views import it, and it had only been a dev dependency, so the first metrics image crash-looped on `No module named polars` the moment the entrypoint imported the app at process start. Verified in the stack: bcda:8502/metrics serves stack_bcda_* series, the prometheus target is up and stack_bcda_store_jobs answers a query. refs #1139 --- compose.yml | 7 + .../specs/2026-10-25-bcda-v3-design.md | 4 + infra/images/bcda.Dockerfile | 11 +- infra/prometheus/targets/services.yml | 8 + pyproject.toml | 1 + src/bcda/ui/__main__.py | 36 ++++ src/bcda/ui/app.py | 3 +- src/bcda/ui/metrics.py | 188 ++++++++++++++++++ src/bcda/ui/runner.py | 20 +- src/bcda/ui/views.py | 14 +- src/cli/bcda.py | 9 +- tests/bcda/test_ui_main.py | 41 ++++ tests/bcda/test_ui_metrics.py | 127 ++++++++++++ tests/cli/test_cli_bcda.py | 2 +- uv.lock | 3 + 15 files changed, 457 insertions(+), 17 deletions(-) create mode 100644 src/bcda/ui/__main__.py create mode 100644 src/bcda/ui/metrics.py create mode 100644 tests/bcda/test_ui_main.py create mode 100644 tests/bcda/test_ui_metrics.py diff --git a/compose.yml b/compose.yml index 8b13537..c2998b7 100644 --- a/compose.yml +++ b/compose.yml @@ -685,10 +685,17 @@ services: container_name: bcda networks: - gateway + - observability environment: - BCDA_CLIENT_ID=${BCDA_CLIENT_ID:-} - BCDA_CLIENT_SECRET=${BCDA_CLIENT_SECRET:-} - BCDA_BASE_URL=${BCDA_BASE_URL:-https://sandbox.bcda.cms.gov} + # /metrics on :8502 scraped by prometheus (infra/prometheus/targets); + # traces to the collector like api and llm. + - STACK_TELEMETRY=true + - BCDA_METRICS_PORT=8502 + - OTEL_SERVICE_NAME=bcda + - OTEL_EXPORTER_OTLP_ENDPOINT=http://otel-collector:4317 volumes: - ./data:/app/data - ./dev:/app/dev:ro diff --git a/docs/superpowers/specs/2026-10-25-bcda-v3-design.md b/docs/superpowers/specs/2026-10-25-bcda-v3-design.md index b278cd4..170ac9e 100644 --- a/docs/superpowers/specs/2026-10-25-bcda-v3-design.md +++ b/docs/superpowers/specs/2026-10-25-bcda-v3-design.md @@ -82,3 +82,7 @@ Tests: service with `httpx.MockTransport`, runner for thread-routed logs and sta ### Containerised (2026-10-25) `infra/images/bcda.Dockerfile` (same base and install pattern as the api image; `uv sync --extra cli --extra bcda`, Streamlit settings in `infra/images/bcda/streamlit.toml`, health at `/_stcore/health`), compose service `bcda` on the gateway network with `./data` read-write (exports, aco.duckdb, replica publish) and `./dev` read-only (vendored upstream, crosswalk seeds, vendor script), Traefik route `bcda.` behind git-sso, a dashboard tile, the `bcda` subdomain in stack.toml (CoreDNS hosts and the SSO cert list follow) and an `[images.bcda]` entry so the deploy workflow builds, scans and pushes it on changes to `src/**`, `pyproject.toml` or the image files. Public sandbox credentials are built in; `BCDA_CLIENT_ID` / `BCDA_CLIENT_SECRET` / `BCDA_BASE_URL` in `.env` override. + +### `/metrics` (2026-10-25) + +`bcda.ui.metrics` follows the #579 convention (`perf.meter` → OTel meter + Prometheus reader when `STACK_TELEMETRY` is on, a no-op `MockMeter` otherwise). Streamlit cannot mount a route, so the manager publishes the process's Prometheus registry from a second port (`BCDA_METRICS_PORT`, 8502) via `prometheus_client.start_http_server`, and Prometheus scrapes `bcda:8502/metrics` directly like `api` and `llm`. Instruments: `stack_bcda_runs_total{kind,state}`, `stack_bcda_runs_inflight{kind}`, `stack_bcda_run_seconds{kind}`, `stack_bcda_page_views_total{page}`, and two observable gauges read at scrape time, `stack_bcda_store_jobs{status}` and `stack_bcda_store_bytes`. Run kinds are a closed set (export, adopt, load, vendor_check, other). The image installs the `perf` extra; the compose service joins the observability network and sets `OTEL_SERVICE_NAME=bcda`. diff --git a/infra/images/bcda.Dockerfile b/infra/images/bcda.Dockerfile index ce76e5a..ff0eb65 100644 --- a/infra/images/bcda.Dockerfile +++ b/infra/images/bcda.Dockerfile @@ -3,7 +3,8 @@ # connector pipeline, crosswalk and vendored-upstream tooling # (src/bcda/ui, `stack bcda ui`). Same base and install pattern as # api.Dockerfile; the bcda extra brings streamlit, the cli bundle brings -# aco (flatten + load + pipeline) and conf. +# aco (flatten + load + pipeline) and conf, perf brings the OTel meter +# and Prometheus reader behind /metrics. FROM ghcr.io/astral-sh/uv:python3.13-bookworm-slim WORKDIR /app @@ -19,7 +20,7 @@ ENV UV_PYTHON_PREFERENCE=only-system \ UV_LINK_MODE=copy \ UV_PROJECT_ENVIRONMENT=.venv \ UV_INDEX_URL=${PYPI_INDEX_URL} -RUN uv sync --no-dev --extra cli --extra bcda && uv pip install -e . +RUN uv sync --no-dev --extra cli --extra bcda --extra perf && uv pip install -e . COPY stack.toml ./ @@ -27,10 +28,12 @@ COPY stack.toml ./ # stats, bound to all interfaces, websocket-friendly behind Traefik). COPY infra/images/bcda/streamlit.toml /app/.streamlit/config.toml -EXPOSE 8501 +# 8501 Streamlit · 8502 Prometheus registry (/metrics, perf meter via STACK_TELEMETRY) +EXPOSE 8501 8502 # /_stcore/health answers "ok" once the Streamlit server is up. HEALTHCHECK --interval=30s --timeout=5s --retries=3 \ CMD curl -sf --max-time 4 http://localhost:8501/_stcore/health || exit 1 -CMD ["uv", "run", "--no-sync", "streamlit", "run", "src/bcda/ui/app.py"] +# python -m bcda.ui starts the /metrics server, then streamlit in-process. +CMD ["uv", "run", "--no-sync", "python", "-m", "bcda.ui"] diff --git a/infra/prometheus/targets/services.yml b/infra/prometheus/targets/services.yml index 53308aa..85dd706 100644 --- a/infra/prometheus/targets/services.yml +++ b/infra/prometheus/targets/services.yml @@ -58,3 +58,11 @@ labels: job: llm __metrics_path__: /metrics + +# BCDA manager — Streamlit cannot mount a route, so bcda.ui.metrics serves +# the same OTel Prometheus registry from a second port: stack_bcda_runs_*, +# stack_bcda_page_views_total, stack_bcda_store_*. +- targets: ['bcda:8502'] + labels: + job: bcda + __metrics_path__: /metrics diff --git a/pyproject.toml b/pyproject.toml index 19da8d0..e2b7fe6 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -42,6 +42,7 @@ bcda = [ "narwhals>=2.17.0", "pyarrow>=23.0.1", "streamlit>=1.40", + "polars>=1.0", ] bib = [ "stack[conf]", diff --git a/src/bcda/ui/__main__.py b/src/bcda/ui/__main__.py new file mode 100644 index 0000000..8ab643d --- /dev/null +++ b/src/bcda/ui/__main__.py @@ -0,0 +1,36 @@ +"""``python -m bcda.ui`` — start the BCDA manager with its metrics endpoint. + +Streamlit executes ``app.py`` only when a browser session opens, so a +``metrics.serve()`` call inside the page never runs on a fresh +container and Prometheus sees a closed port. This entrypoint starts the +``/metrics`` server first (a no-op when ``STACK_TELEMETRY`` is off), then +boots Streamlit's CLI in the same process so page views and background +runs land in the same registry. + + python -m bcda.ui # :8501, metrics on BCDA_METRICS_PORT + python -m bcda.ui --server.port 9000 # extra args go to `streamlit run` +""" + +from __future__ import annotations + +import sys +from pathlib import Path + + +def main(argv: list[str] | None = None) -> int: + from streamlit.web import cli as stcli + + from bcda.ui import app, metrics + + metrics.serve() + args = list(sys.argv[1:] if argv is None else argv) + sys.argv = ["streamlit", "run", str(Path(app.__file__)), *args] + try: + stcli.main() + except SystemExit as exc: # click exits through SystemExit + return int(exc.code or 0) + return 0 + + +if __name__ == "__main__": # pragma: no cover — module entrypoint + sys.exit(main()) diff --git a/src/bcda/ui/app.py b/src/bcda/ui/app.py index 3c755d5..929bb5f 100644 --- a/src/bcda/ui/app.py +++ b/src/bcda/ui/app.py @@ -8,7 +8,7 @@ from __future__ import annotations import streamlit as st -from bcda.ui import views +from bcda.ui import metrics, views PAGES = [ st.Page(views.connect, title="Connect", icon="🔌", default=True), @@ -23,6 +23,7 @@ PAGES = [ def main() -> None: st.set_page_config(page_title="BCDA manager", page_icon="🏥", layout="wide") + metrics.serve() # /metrics on BCDA_METRICS_PORT when STACK_TELEMETRY is on; no-op otherwise conn = st.session_state.get("conn") with st.sidebar: st.markdown("### BCDA manager") diff --git a/src/bcda/ui/metrics.py b/src/bcda/ui/metrics.py new file mode 100644 index 0000000..5c74b17 --- /dev/null +++ b/src/bcda/ui/metrics.py @@ -0,0 +1,188 @@ +"""OpenTelemetry instruments for the BCDA manager, served on ``/metrics``. + +Follows the P26/#579 convention (``perf.meter`` returns a real OTel meter +when ``STACK_TELEMETRY`` is on and a zero-cost ``MockMeter`` otherwise; +the ``PrometheusMetricReader`` fills the in-process ``prometheus_client`` +registry). Streamlit cannot mount a route, so :func:`serve` publishes +that registry from a second port in the same process — Prometheus +scrapes ``bcda:8502/metrics`` directly, like ``api`` and ``llm``. + +Instruments (all ``stack_bcda_…``):: + + stack_bcda_runs_total{kind,state} background runs finished: done / failed + stack_bcda_runs_inflight{kind} runs executing right now + stack_bcda_run_seconds{kind} wall time of a run + stack_bcda_page_views_total{page} page renders + stack_bcda_store_jobs{status} store jobs by current status (gauge) + stack_bcda_store_bytes bytes held in exports/ (gauge) + +Attribute values are closed sets — run kind (export, adopt, load, +vendor_check, other), state, page name, store status — never job keys +or MBIs, so the series count stays bounded. +""" + +from __future__ import annotations + +import logging +import os +import threading +from typing import Any + +METER_NAME = "stack.bcda" +DEFAULT_PORT = 8502 +KINDS = ("export", "adopt", "load", "vendor_check", "other") + +log = logging.getLogger(__name__) + +_instruments: _Instruments | None = None +_build_lock = threading.Lock() +_server_port: int | None = None + + +class _Instruments: + def __init__(self) -> None: + from perf import meter + + m: Any = meter(METER_NAME) + self.runs_total = m.create_counter( + "stack_bcda_runs_total", + unit="1", + description="Background runs finished, by kind and state", + ) + self.runs_inflight = m.create_up_down_counter( + "stack_bcda_runs_inflight", + unit="1", + description="Background runs executing right now", + ) + self.run_seconds = m.create_histogram( + "stack_bcda_run_seconds", + unit="s", + description="Wall time of a background run", + ) + self.page_views = m.create_counter( + "stack_bcda_page_views_total", unit="1", description="Manager page renders" + ) + # Observable gauges read the store on each scrape; the callback is + # installed by observe_store() once a store path is known. + self._store_callbacks: list[Any] = [] + self.store_jobs = m.create_observable_gauge( + "stack_bcda_store_jobs", + callbacks=[self._observe_jobs], + unit="1", + description="Store jobs by current status", + ) + self.store_bytes = m.create_observable_gauge( + "stack_bcda_store_bytes", + callbacks=[self._observe_bytes], + unit="By", + description="Bytes held in the store's exports directory", + ) + self._store_path: Any = None + + # ── gauge callbacks (run at scrape time) ───────────────── + + def _observe_jobs(self, options: Any = None) -> list[Any]: + from opentelemetry.metrics import Observation + + return [ + Observation(n, {"status": status}) + for status, n in store_job_counts(self._store_path).items() + ] + + def _observe_bytes(self, options: Any = None) -> list[Any]: + from opentelemetry.metrics import Observation + + return [Observation(store_bytes(self._store_path))] + + +def _get() -> _Instruments: + global _instruments + if _instruments is None: + with _build_lock: + if _instruments is None: + _instruments = _Instruments() + return _instruments + + +def _kind(kind: str) -> str: + return kind if kind in KINDS else "other" + + +# ── store observation helpers (pure, used by the callbacks and tests) ── + + +def store_job_counts(path: Any = None) -> dict[str, int]: + """``{status: count}`` over the store's current job states.""" + try: + from bcda.ui import service as svc + + store = svc.make_store(path) + counts: dict[str, int] = {} + for job in store.list_jobs(): + counts[job["status"]] = counts.get(job["status"], 0) + 1 + return counts + except Exception: # noqa: BLE001 — a scrape must never raise + return {} + + +def store_bytes(path: Any = None) -> int: + try: + from bcda.ui import service as svc + + return sum(d["bytes"] for d in svc.export_dirs(path)) + except Exception: # noqa: BLE001 + return 0 + + +# ── recording API (called by runner and views) ────────────── + + +def observe_store(path: Any) -> None: + """Point the store gauges at *path* (``None`` = the configured store).""" + _get()._store_path = path + + +def run_started(kind: str) -> None: + _get().runs_inflight.add(1, {"kind": _kind(kind)}) + + +def run_finished(kind: str, state: str, seconds: float) -> None: + inst = _get() + k = _kind(kind) + inst.runs_inflight.add(-1, {"kind": k}) + inst.runs_total.add(1, {"kind": k, "state": state}) + inst.run_seconds.record(max(0.0, seconds), {"kind": k}) + + +def page_view(page: str) -> None: + _get().page_views.add(1, {"page": page}) + + +# ── the /metrics server ────────────────────────────────────── + + +def serve(port: int | None = None) -> int | None: + """Publish the process's Prometheus registry on ``:/metrics``. + + No-op (returns ``None``) when telemetry is off. Idempotent: the + first call starts the server, later calls return its port. Reads + ``BCDA_METRICS_PORT`` when *port* is not given. + """ + global _server_port + from perf import _enabled, init + + if not _enabled(): + return None + if _server_port is not None: + return _server_port + init() # meter provider + Prometheus reader (idempotent) + from prometheus_client import start_http_server + + port = int( + port if port is not None else os.environ.get("BCDA_METRICS_PORT", DEFAULT_PORT) + ) + server, _thread = start_http_server(port) + _server_port = server.server_address[1] + _get() # build the instrument set so the gauges exist before the first scrape + log.info("bcda metrics on :%d/metrics", _server_port) + return _server_port diff --git a/src/bcda/ui/runner.py b/src/bcda/ui/runner.py index 4153159..854310e 100644 --- a/src/bcda/ui/runner.py +++ b/src/bcda/ui/runner.py @@ -34,6 +34,7 @@ class Run: id: str name: str + kind: str = "other" state: str = "queued" started: float = field(default_factory=time.time) finished: float | None = None @@ -91,11 +92,22 @@ class Runner: # ── API ────────────────────────────────────────────────── def submit( - self, name: str, fn: Callable[..., Any], *args: Any, **kwargs: Any + self, + name: str, + fn: Callable[..., Any], + *args: Any, + kind: str = "other", + **kwargs: Any, ) -> Run: - """Start *fn* on a daemon thread; return its Run immediately.""" + """Start *fn* on a daemon thread; return its Run immediately. + + ``kind`` labels the run for metrics (export, adopt, load, + vendor_check, other) — see :mod:`bcda.ui.metrics`. + """ + from bcda.ui import metrics + self._ensure_handler() - run = Run(id=uuid.uuid4().hex[:8], name=name) + run = Run(id=uuid.uuid4().hex[:8], name=name, kind=kind) with self._lock: self._runs[run.id] = run @@ -105,6 +117,7 @@ class Runner: self._by_thread[run.thread_id] = run run.state = "running" run.append(f"started: {name}") + metrics.run_started(kind) # The state flips last: ``wait()`` and the page read ``state`` # without a lock, so result, log and finish time must already be # in place when they see done/failed. @@ -120,6 +133,7 @@ class Runner: run.finished = time.time() run.state = "failed" finally: + metrics.run_finished(kind, run.state, run.elapsed) with self._lock: self._by_thread.pop(run.thread_id, None) diff --git a/src/bcda/ui/views.py b/src/bcda/ui/views.py index 8f05adf..acfe1e4 100644 --- a/src/bcda/ui/views.py +++ b/src/bcda/ui/views.py @@ -14,6 +14,7 @@ from typing import Any import polars as pl import streamlit as st +from bcda.ui import metrics from bcda.ui import service as svc from bcda.ui.runner import runner @@ -81,6 +82,7 @@ def runs_panel() -> None: def connect() -> None: + metrics.page_view("connect") st.title("Connect") st.caption( "BCDA v3 (v1/v2 end 2027-07-30). `/auth` and `/data` are shared by every version; " @@ -169,6 +171,7 @@ def connect() -> None: def exports() -> None: + metrics.page_view("exports") st.title("Exports") conn = _require_conn() if conn is None: @@ -228,6 +231,7 @@ def exports() -> None: svc.export_per_type, store, client, + kind="export", types=types, endpoint=endpoint, since=since, @@ -240,6 +244,7 @@ def exports() -> None: svc.start_export, store, client, + kind="export", endpoint=endpoint, types=types, since=since, @@ -290,6 +295,7 @@ def exports() -> None: _store(), _client(), job["job_url"], + kind="adopt", types=[t for t in job["types"].split(",") if t and t != "(all)"] or None, ) @@ -317,6 +323,7 @@ def exports() -> None: def store() -> None: + metrics.page_view("store") st.title("Store") path = svc.store_path() st.caption( @@ -365,6 +372,7 @@ def store() -> None: def load() -> None: + metrics.page_view("load") st.title("Flatten + load") st.caption( "Newest export per resource type → `bcda.*` in aco.duckdb (v2 or v3 dialect per resource, " @@ -381,6 +389,7 @@ def load() -> None: run = runner.submit( "flatten + load", svc.load_and_publish, + kind="load", run_pipeline=run_pipeline, skip_flatten=skip_flatten, ) @@ -392,6 +401,7 @@ def load() -> None: def data() -> None: + metrics.page_view("data") st.title("Data") try: con = svc.replica_connect() @@ -439,6 +449,7 @@ def data() -> None: def dictionary() -> None: + metrics.page_view("dictionary") st.title("Data dictionary + CCLF crosswalk") try: con = svc.replica_connect() @@ -515,6 +526,7 @@ def dictionary() -> None: def upstream() -> None: + metrics.page_view("upstream") st.title("Upstream: CMSgov/bcda-app + bcda.cms.gov") status = svc.vendor_status() if not status.get("available"): @@ -535,7 +547,7 @@ def upstream() -> None: with st.expander("Files"): st.dataframe(pl.DataFrame(status["files"]), hide_index=True, width="stretch") if st.button("Check upstream for drift (network)", key="btn_vendor_check"): - run = runner.submit("vendor drift check", svc.vendor_check) + run = runner.submit("vendor drift check", svc.vendor_check, kind="vendor_check") st.toast(f"Checking upstream as run {run.id}") st.caption( "A weekly workflow runs the same check and files an issue on drift; refresh with `vendor_bcda_app.py --sync` then `gen_bcda_v3_crosswalk.py`." diff --git a/src/cli/bcda.py b/src/cli/bcda.py index 33a7dda..53a57f4 100644 --- a/src/cli/bcda.py +++ b/src/cli/bcda.py @@ -13,7 +13,6 @@ sandbox entities (default *Large Model Entity*). from __future__ import annotations import json -from pathlib import Path import typer @@ -50,18 +49,14 @@ def ui( port: int = typer.Option(8501, help="Port for the Streamlit server."), headless: bool = typer.Option(True, help="Do not open a browser."), ) -> None: - """Launch the BCDA manager (Streamlit).""" + """Launch the BCDA manager (Streamlit; /metrics when STACK_TELEMETRY is on).""" import subprocess import sys - from bcda.ui import app as ui_app - cmd = [ sys.executable, "-m", - "streamlit", - "run", - str(Path(ui_app.__file__)), + "bcda.ui", "--server.port", str(port), "--server.headless", diff --git a/tests/bcda/test_ui_main.py b/tests/bcda/test_ui_main.py new file mode 100644 index 0000000..4da5f68 --- /dev/null +++ b/tests/bcda/test_ui_main.py @@ -0,0 +1,41 @@ +"""python -m bcda.ui — metrics first, then Streamlit in-process.""" + +from __future__ import annotations + +import sys + +from bcda.ui import __main__ as entry +from bcda.ui import metrics + + +def test_main_serves_metrics_then_boots_streamlit(monkeypatch) -> None: + calls: list = [] + monkeypatch.setattr(metrics, "serve", lambda: calls.append("serve") or 8502) + + class FakeCli: + @staticmethod + def main() -> None: + calls.append(("argv", list(sys.argv))) + raise SystemExit(0) + + monkeypatch.setattr("streamlit.web.cli.main", FakeCli.main) + assert entry.main(["--server.port", "9000"]) == 0 + assert calls[0] == "serve" + argv = calls[1][1] + assert argv[:2] == ["streamlit", "run"] and argv[2].endswith("bcda/ui/app.py") + assert argv[3:] == ["--server.port", "9000"] + + +def test_main_passes_exit_code_and_default_argv(monkeypatch) -> None: + monkeypatch.setattr(metrics, "serve", lambda: None) + monkeypatch.setattr(sys, "argv", ["bcda.ui", "--server.headless", "true"]) + + def exits() -> None: + raise SystemExit(3) + + monkeypatch.setattr("streamlit.web.cli.main", exits) + assert entry.main() == 3 + assert sys.argv[3:] == ["--server.headless", "true"] + + monkeypatch.setattr("streamlit.web.cli.main", lambda: None) + assert entry.main([]) == 0 diff --git a/tests/bcda/test_ui_metrics.py b/tests/bcda/test_ui_metrics.py new file mode 100644 index 0000000..38a7543 --- /dev/null +++ b/tests/bcda/test_ui_metrics.py @@ -0,0 +1,127 @@ +"""bcda.ui.metrics — instruments, store gauges and the /metrics server.""" + +from __future__ import annotations + +import socket +import urllib.request + +import pytest + +from bcda.store import Store +from bcda.ui import metrics +from bcda.ui.runner import Runner + + +@pytest.fixture(autouse=True) +def _fresh_instruments(monkeypatch): + """Rebuild the instrument set per test; keep any server from leaking state.""" + monkeypatch.setattr(metrics, "_instruments", None) + yield + monkeypatch.setattr(metrics, "_instruments", None) + + +def test_mock_meter_path_is_free(monkeypatch) -> None: + monkeypatch.setenv("STACK_TELEMETRY", "false") + metrics.run_started("export") + metrics.run_finished("export", "done", 1.5) + metrics.run_finished("bogus kind", "failed", -1) # clamps and folds to "other" + metrics.page_view("connect") + metrics.observe_store(None) + assert metrics.serve(port=0) is None + assert metrics._kind("load") == "load" and metrics._kind("nope") == "other" + + +def test_store_observations(tmp_path, monkeypatch) -> None: + assert ( + metrics.store_job_counts(tmp_path) == {} and metrics.store_bytes(tmp_path) == 0 + ) + store = Store(path=str(tmp_path)) + store._append_job(key="A", status="complete") + store._append_job(key="B", status="failed") + store._append_job(key="C", status="complete") + d = tmp_path / "exports" / "A" + d.mkdir(parents=True) + (d / "Patient.ndjson").write_bytes(b"x" * 10) + assert metrics.store_job_counts(tmp_path) == {"complete": 2, "failed": 1} + assert metrics.store_bytes(tmp_path) == 10 + # errors never propagate to a scrape + from bcda.ui import service as svc + + def boom(*a, **k): # noqa: ANN002, ANN003, ANN202 + raise OSError("disk gone") + + monkeypatch.setattr(svc, "make_store", boom) + monkeypatch.setattr(svc, "export_dirs", boom) + assert ( + metrics.store_job_counts(tmp_path) == {} and metrics.store_bytes(tmp_path) == 0 + ) + + +def test_runner_records_runs(monkeypatch) -> None: + seen: list = [] + monkeypatch.setattr( + metrics, "run_started", lambda kind: seen.append(("start", kind)) + ) + monkeypatch.setattr( + metrics, + "run_finished", + lambda kind, state, secs: seen.append(("end", kind, state, secs >= 0)), + ) + r = Runner() + r.wait(r.submit("ok", lambda: 1, kind="load").id) + + def boom() -> None: + raise RuntimeError("x") + + r.wait(r.submit("bad", boom).id) + assert seen == [ + ("start", "load"), + ("end", "load", "done", True), + ("start", "other"), + ("end", "other", "failed", True), + ] + assert r.all()[0].kind == "other" and r.all()[1].kind == "load" + + +def _free_port() -> int: + with socket.socket() as s: + s.bind(("127.0.0.1", 0)) + return s.getsockname()[1] + + +def test_serve_publishes_registry(monkeypatch, tmp_path) -> None: + """With telemetry on, /metrics carries stack_bcda_* series incl. the store gauges.""" + import perf + from perf import _meter + + monkeypatch.setenv("STACK_TELEMETRY", "true") + monkeypatch.setattr(perf, "_initialised", False) + monkeypatch.setattr(_meter, "_provider_ready", False) + monkeypatch.setattr(metrics, "_server_port", None) + store = Store(path=str(tmp_path)) + store._append_job(key="A", status="complete") + + port = metrics.serve(port=_free_port()) + assert port and metrics.serve() == port # idempotent + metrics.observe_store(tmp_path) + metrics.run_started("export") + metrics.run_finished("export", "done", 2.0) + metrics.page_view("exports") + + body = ( + urllib.request.urlopen(f"http://127.0.0.1:{port}/metrics", timeout=5) + .read() + .decode() + ) + assert 'stack_bcda_runs_total{kind="export",state="done"} 1.0' in body + assert 'stack_bcda_page_views_total{page="exports"} 1.0' in body + assert 'stack_bcda_store_jobs{status="complete"} 1.0' in body + assert "stack_bcda_store_bytes 0.0" in body + assert "stack_bcda_run_seconds" in body + + # the default port comes from the environment + monkeypatch.setattr(metrics, "_server_port", None) + env_port = _free_port() + monkeypatch.setenv("BCDA_METRICS_PORT", str(env_port)) + assert metrics.serve() == env_port + perf.shutdown() diff --git a/tests/cli/test_cli_bcda.py b/tests/cli/test_cli_bcda.py index 940d77d..17544d3 100644 --- a/tests/cli/test_cli_bcda.py +++ b/tests/cli/test_cli_bcda.py @@ -247,5 +247,5 @@ def test_ui_launches_streamlit(monkeypatch) -> None: result = runner.invoke(app, ["bcda", "ui", "--port", "9999", "--no-headless"]) assert result.exit_code == 0 and "http://localhost:9999" in result.output cmd = call.call_args.args[0] - assert cmd[1:4] == ["-m", "streamlit", "run"] and cmd[4].endswith("bcda/ui/app.py") + assert cmd[1:3] == ["-m", "bcda.ui"] assert cmd[cmd.index("--server.headless") + 1] == "false" diff --git a/uv.lock b/uv.lock index 4919f68..4c4af6f 100644 --- a/uv.lock +++ b/uv.lock @@ -3959,6 +3959,7 @@ all = [ { name = "opentelemetry-instrumentation-fastapi" }, { name = "opentelemetry-sdk" }, { name = "opentelemetry-semantic-conventions" }, + { name = "polars" }, { name = "psutil" }, { name = "psycopg", extra = ["binary"] }, { name = "pyarrow" }, @@ -3995,6 +3996,7 @@ bcda = [ { name = "fsspec" }, { name = "httpx" }, { name = "narwhals" }, + { name = "polars" }, { name = "pyarrow" }, { name = "pydantic" }, { name = "streamlit" }, @@ -4209,6 +4211,7 @@ requires-dist = [ { name = "opentelemetry-sdk", marker = "extra == 'perf'", specifier = ">=1.25.0" }, { name = "opentelemetry-semantic-conventions", marker = "extra == 'perf'", specifier = ">=0.46b0" }, { name = "pdfminer-six", marker = "extra == 'prisma'", specifier = ">=20221105" }, + { name = "polars", marker = "extra == 'bcda'", specifier = ">=1.0" }, { name = "psutil", marker = "extra == 'perf'", specifier = ">=5.9.0" }, { name = "psycopg", extras = ["binary"], marker = "extra == 'llm'", specifier = ">=3.2.0" }, { name = "pyarrow", marker = "extra == 'aco'", specifier = ">=23.0.1" },