From ea2ab3255f3e607233259ebdef214172017d9bcd Mon Sep 17 00:00:00 2001 From: Roddie Kieley Date: Wed, 29 Jul 2026 11:53:56 -0230 Subject: [PATCH 01/16] feat: exporter prometheus_client /metrics (JEP-0013 Phase 2) - Add exporter-local `prometheus_client` registry with JEP-named series: `jumpstarter_operations_total`, `jumpstarter_operation_duration_seconds`, `jumpstarter_operation_errors_total`, `jumpstarter_stream_bytes_total`, `jumpstarter_active_sessions`, plus exemplars (`client`, `lease_id`). - Expose HTTP `GET /metrics` on the exporter process for lab/dev scrape (same registry Phase 3 will later reverse-scrape via MetricsStream). - Minimal core-path wiring so series increment under test; full per-driver telemetry architecture remains Phase 4. --- .../jumpstarter-cli/jumpstarter_cli/run.py | 62 +++++++- .../jumpstarter/jumpstarter/driver/base.py | 109 +++++++++++++ .../jumpstarter/exporter/session.py | 14 +- .../jumpstarter/metrics/__init__.py | 17 +++ .../jumpstarter/metrics/metrics_test.py | 133 ++++++++++++++++ .../jumpstarter/metrics/registry.py | 144 ++++++++++++++++++ .../jumpstarter/jumpstarter/metrics/server.py | 61 ++++++++ .../jumpstarter/jumpstarter/streams/common.py | 50 +++++- python/packages/jumpstarter/pyproject.toml | 1 + python/uv.lock | 11 ++ 10 files changed, 591 insertions(+), 11 deletions(-) create mode 100644 python/packages/jumpstarter/jumpstarter/metrics/__init__.py create mode 100644 python/packages/jumpstarter/jumpstarter/metrics/metrics_test.py create mode 100644 python/packages/jumpstarter/jumpstarter/metrics/registry.py create mode 100644 python/packages/jumpstarter/jumpstarter/metrics/server.py diff --git a/python/packages/jumpstarter-cli/jumpstarter_cli/run.py b/python/packages/jumpstarter-cli/jumpstarter_cli/run.py index a9cd88511..5da4ccb0b 100644 --- a/python/packages/jumpstarter-cli/jumpstarter_cli/run.py +++ b/python/packages/jumpstarter-cli/jumpstarter_cli/run.py @@ -70,7 +70,15 @@ def _reap_zombie_processes(capture_child=None): logger.warning(f"PARENT: Error during zombie reaping: {e}") -def _handle_child(config, parsed_bind=None, tls_insecure=False, tls_cert=None, tls_key=None, passphrase=None): # noqa: C901 +def _handle_child( # noqa: C901 + config, + parsed_bind=None, + tls_insecure=False, + tls_cert=None, + tls_key=None, + passphrase=None, + metrics_bind_address=":8080", +): """Handle child process with graceful shutdown.""" async def serve_with_graceful_shutdown(): # noqa: C901 received_signal = 0 @@ -99,6 +107,12 @@ async def signal_handler(): # Start signal handler immediately signal_tg.start_soon(signal_handler) + from jumpstarter.metrics import start_metrics_server + + listen_addr = start_metrics_server(metrics_bind_address) + if listen_addr: + logger.info("Serving metrics server at http://%s/metrics", listen_addr) + if parsed_bind is not None: host, port = parsed_bind tls_credentials = None @@ -204,7 +218,13 @@ def parent_signal_handler(signum, _): def _serve_with_exc_handling( - config, parsed_bind=None, tls_insecure=False, tls_cert=None, tls_key=None, passphrase=None + config, + parsed_bind=None, + tls_insecure=False, + tls_cert=None, + tls_key=None, + passphrase=None, + metrics_bind_address=":8080", ): max_rapid_failures = config.failure_detection.max_rapid_failures rapid_failure_window = config.failure_detection.rapid_failure_window @@ -253,7 +273,15 @@ def _serve_with_exc_handling( rapid_failure_count = 0 else: os.setsid() # Become group leader so all spawned subprocesses are reached by parent's signals - _handle_child(config, parsed_bind, tls_insecure, tls_cert, tls_key, passphrase) + _handle_child( + config, + parsed_bind, + tls_insecure, + tls_cert, + tls_key, + passphrase, + metrics_bind_address, + ) sys.exit(1) # should never happen @@ -294,8 +322,24 @@ def _serve_with_exc_handling( default=False, help="Exit after the current lease ends instead of waiting for a new one.", ) +@click.option( + "--metrics-bind-address", + "metrics_bind_address", + default=":8080", + show_default=True, + help="Address for HTTP GET /metrics (Prometheus/OpenMetrics). Use 0 to disable.", +) @handle_exceptions -def run(config, listener_bind, tls_insecure, tls_cert, tls_key, passphrase, exit_on_lease_end): +def run( + config, + listener_bind, + tls_insecure, + tls_cert, + tls_key, + passphrase, + exit_on_lease_end, + metrics_bind_address, +): """Run an exporter locally.""" if listener_bind is not None and config is None: raise click.UsageError("--exporter-config (or --exporter) is required when using --tls-grpc-listener") @@ -313,4 +357,12 @@ def run(config, listener_bind, tls_insecure, tls_cert, tls_key, passphrase, exit if exit_on_lease_end: config.exit_on_lease_end = True parsed_bind = _parse_listener_bind(listener_bind) if listener_bind is not None else None - return _serve_with_exc_handling(config, parsed_bind, tls_insecure, tls_cert, tls_key, passphrase) + return _serve_with_exc_handling( + config, + parsed_bind, + tls_insecure, + tls_cert, + tls_key, + passphrase, + metrics_bind_address, + ) diff --git a/python/packages/jumpstarter/jumpstarter/driver/base.py b/python/packages/jumpstarter/jumpstarter/driver/base.py index fec600923..769d3ce01 100644 --- a/python/packages/jumpstarter/jumpstarter/driver/base.py +++ b/python/packages/jumpstarter/jumpstarter/driver/base.py @@ -6,6 +6,7 @@ import logging import os +import time from abc import ABCMeta, abstractmethod from contextlib import asynccontextmanager from dataclasses import field @@ -113,11 +114,36 @@ def client(cls) -> str: def extra_labels(self) -> dict[str, str]: return {} + def _record_operation_metrics( + self, + *, + operation: str, + result: str, + duration_seconds: float, + error_type: str | None = None, + ) -> None: + from jumpstarter.metrics.registry import ( + exemplars_from_log_context, + exporter_from_log_context, + get_registry, + ) + + get_registry().record_operation( + exporter=exporter_from_log_context(default=self.name if hasattr(self, "name") else "unknown"), + operation=operation, + result=result, + driver_type=self.driver_type, + duration_seconds=duration_seconds, + exemplars=exemplars_from_log_context(), + error_type=error_type, + ) + async def DriverCall(self, request, context): """ :meta private: """ op = request.method + started = time.perf_counter() self.logger.info( "Operation started", extra={"operation": op, "driver_type": self.driver_type}, @@ -132,6 +158,11 @@ async def DriverCall(self, request, context): else: result = await to_thread.run_sync(method, *args) + self._record_operation_metrics( + operation=op, + result="success", + duration_seconds=time.perf_counter() - started, + ) self.logger.info( "Operation completed", extra={"operation": op, "driver_type": self.driver_type, "result": "success"}, @@ -141,6 +172,12 @@ async def DriverCall(self, request, context): result=encode_value(result), ) except NotImplementedError as e: + self._record_operation_metrics( + operation=op, + result="failure", + duration_seconds=time.perf_counter() - started, + error_type="not_implemented", + ) self.logger.warning( "Operation failed", extra={"operation": op, "driver_type": self.driver_type, @@ -148,6 +185,12 @@ async def DriverCall(self, request, context): ) await context.abort(StatusCode.UNIMPLEMENTED, str(e)) except ValueError as e: + self._record_operation_metrics( + operation=op, + result="failure", + duration_seconds=time.perf_counter() - started, + error_type="validation_error", + ) self.logger.warning( "Operation failed", extra={"operation": op, "driver_type": self.driver_type, @@ -155,6 +198,12 @@ async def DriverCall(self, request, context): ) await context.abort(StatusCode.INVALID_ARGUMENT, str(e)) except TimeoutError as e: + self._record_operation_metrics( + operation=op, + result="failure", + duration_seconds=time.perf_counter() - started, + error_type="timeout", + ) self.logger.warning( "Operation failed", extra={"operation": op, "driver_type": self.driver_type, @@ -162,6 +211,12 @@ async def DriverCall(self, request, context): ) await context.abort(StatusCode.DEADLINE_EXCEEDED, str(e)) except ConnectionError as e: + self._record_operation_metrics( + operation=op, + result="failure", + duration_seconds=time.perf_counter() - started, + error_type="connection_error", + ) self.logger.warning( "Operation failed", extra={"operation": op, "driver_type": self.driver_type, @@ -169,6 +224,12 @@ async def DriverCall(self, request, context): ) await context.abort(StatusCode.UNAVAILABLE, str(e)) except OSError as e: + self._record_operation_metrics( + operation=op, + result="failure", + duration_seconds=time.perf_counter() - started, + error_type="device_error", + ) self.logger.warning( "Operation failed", extra={"operation": op, "driver_type": self.driver_type, @@ -176,6 +237,12 @@ async def DriverCall(self, request, context): ) await context.abort(StatusCode.INTERNAL, str(e)) except Exception as e: + self._record_operation_metrics( + operation=op, + result="failure", + duration_seconds=time.perf_counter() - started, + error_type="internal_error", + ) self.logger.warning( "Operation failed", extra={"operation": op, "driver_type": self.driver_type, @@ -188,6 +255,7 @@ async def StreamingDriverCall(self, request, context): :meta private: """ op = request.method + started = time.perf_counter() self.logger.info( "Operation started", extra={"operation": op, "driver_type": self.driver_type}, @@ -209,11 +277,22 @@ async def StreamingDriverCall(self, request, context): uuid=str(uuid4()), result=encode_value(result), ) + self._record_operation_metrics( + operation=op, + result="success", + duration_seconds=time.perf_counter() - started, + ) self.logger.info( "Operation completed", extra={"operation": op, "driver_type": self.driver_type, "result": "success"}, ) except NotImplementedError as e: + self._record_operation_metrics( + operation=op, + result="failure", + duration_seconds=time.perf_counter() - started, + error_type="not_implemented", + ) self.logger.warning( "Operation failed", extra={"operation": op, "driver_type": self.driver_type, @@ -221,6 +300,12 @@ async def StreamingDriverCall(self, request, context): ) await context.abort(StatusCode.UNIMPLEMENTED, str(e)) except ValueError as e: + self._record_operation_metrics( + operation=op, + result="failure", + duration_seconds=time.perf_counter() - started, + error_type="validation_error", + ) self.logger.warning( "Operation failed", extra={"operation": op, "driver_type": self.driver_type, @@ -228,6 +313,12 @@ async def StreamingDriverCall(self, request, context): ) await context.abort(StatusCode.INVALID_ARGUMENT, str(e)) except TimeoutError as e: + self._record_operation_metrics( + operation=op, + result="failure", + duration_seconds=time.perf_counter() - started, + error_type="timeout", + ) self.logger.warning( "Operation failed", extra={"operation": op, "driver_type": self.driver_type, @@ -235,6 +326,12 @@ async def StreamingDriverCall(self, request, context): ) await context.abort(StatusCode.DEADLINE_EXCEEDED, str(e)) except ConnectionError as e: + self._record_operation_metrics( + operation=op, + result="failure", + duration_seconds=time.perf_counter() - started, + error_type="connection_error", + ) self.logger.warning( "Operation failed", extra={"operation": op, "driver_type": self.driver_type, @@ -242,6 +339,12 @@ async def StreamingDriverCall(self, request, context): ) await context.abort(StatusCode.UNAVAILABLE, str(e)) except OSError as e: + self._record_operation_metrics( + operation=op, + result="failure", + duration_seconds=time.perf_counter() - started, + error_type="device_error", + ) self.logger.warning( "Operation failed", extra={"operation": op, "driver_type": self.driver_type, @@ -249,6 +352,12 @@ async def StreamingDriverCall(self, request, context): ) await context.abort(StatusCode.INTERNAL, str(e)) except Exception as e: + self._record_operation_metrics( + operation=op, + result="failure", + duration_seconds=time.perf_counter() - started, + error_type="internal_error", + ) self.logger.warning( "Operation failed", extra={"operation": op, "driver_type": self.driver_type, diff --git a/python/packages/jumpstarter/jumpstarter/exporter/session.py b/python/packages/jumpstarter/jumpstarter/exporter/session.py index f19add60e..bde94e05c 100644 --- a/python/packages/jumpstarter/jumpstarter/exporter/session.py +++ b/python/packages/jumpstarter/jumpstarter/exporter/session.py @@ -57,11 +57,20 @@ class Session( @contextmanager def __contextmanager__(self) -> Generator[Self]: + from jumpstarter.logging import set_log_context + from jumpstarter.metrics import get_registry + logging.getLogger().addHandler(self._logging_handler) self.root_device.reset() + set_log_context(exporter=self.name) + get_registry().inc_active_sessions(exporter=self.name, delta=1.0) try: yield self finally: + try: + get_registry().inc_active_sessions(exporter=self.name, delta=-1.0) + except Exception: + pass try: self.root_device.close() except Exception as e: @@ -319,14 +328,15 @@ async def StreamingDriverCall(self, request, context): async def Stream(self, _request_iterator, context): request = StreamRequestMetadata(**dict(list(context.invocation_metadata()))).request logger.debug("Streaming(%s)", request) - async with self[request.uuid].Stream(request, context) as stream: + driver = self[request.uuid] + async with driver.Stream(request, context) as stream: metadata = [] with suppress(TypedAttributeLookupError): metadata.extend(stream.extra(MetadataStreamAttributes.metadata).items()) await context.send_initial_metadata(metadata) async with RouterStream(context=context) as remote: - async with forward_stream(remote, stream): + async with forward_stream(remote, stream, metrics_driver_type=driver.driver_type): event = Event() context.add_done_callback(lambda _: event.set()) await event.wait() diff --git a/python/packages/jumpstarter/jumpstarter/metrics/__init__.py b/python/packages/jumpstarter/jumpstarter/metrics/__init__.py new file mode 100644 index 000000000..bec5b06f3 --- /dev/null +++ b/python/packages/jumpstarter/jumpstarter/metrics/__init__.py @@ -0,0 +1,17 @@ +"""Exporter-local Prometheus metrics (JEP-0013 Phase 2).""" + +from .registry import ( + DEFAULT_EXEMPLAR_KEYS, + MetricsRegistry, + get_registry, + reset_registry_for_tests, +) +from .server import start_metrics_server + +__all__ = [ + "DEFAULT_EXEMPLAR_KEYS", + "MetricsRegistry", + "get_registry", + "reset_registry_for_tests", + "start_metrics_server", +] diff --git a/python/packages/jumpstarter/jumpstarter/metrics/metrics_test.py b/python/packages/jumpstarter/jumpstarter/metrics/metrics_test.py new file mode 100644 index 000000000..c68bd84e2 --- /dev/null +++ b/python/packages/jumpstarter/jumpstarter/metrics/metrics_test.py @@ -0,0 +1,133 @@ +"""JEP-0013 Phase 2 exporter metrics tests.""" + +from __future__ import annotations + +import re +import urllib.error +import urllib.request + +import pytest + +from jumpstarter.metrics import ( + DEFAULT_EXEMPLAR_KEYS, + get_registry, + reset_registry_for_tests, + start_metrics_server, +) + +SERIES = ( + "jumpstarter_operations_total", + "jumpstarter_operation_duration_seconds", + "jumpstarter_operation_errors_total", + "jumpstarter_stream_bytes_total", + "jumpstarter_active_sessions", +) + + +@pytest.fixture(autouse=True) +def _fresh_registry(): + reset_registry_for_tests() + yield + reset_registry_for_tests() + + +def test_default_exemplar_keys(): + assert DEFAULT_EXEMPLAR_KEYS == ("client", "lease_id") + + +def test_generate_latest_contains_named_series_after_increments(): + reg = get_registry() + exemplars = {"client": "ci-bot", "lease_id": "lease-abc"} + reg.record_operation( + exporter="lab-01", + operation="on", + result="success", + driver_type="power", + duration_seconds=0.05, + exemplars=exemplars, + ) + reg.record_operation( + exporter="lab-01", + operation="flash", + result="failure", + driver_type="storage", + duration_seconds=1.2, + exemplars=exemplars, + error_type="timeout", + ) + reg.add_stream_bytes( + exporter="lab-01", + driver_type="serial", + direction="tx", + nbytes=128, + exemplars=exemplars, + ) + reg.set_active_sessions(exporter="lab-01", value=1) + + body = reg.generate_latest().decode() + for name in SERIES: + assert name in body, f"expected series {name} in exposition" + + assert 'jumpstarter_operations_total{' in body or "jumpstarter_operations_total{" in body + assert 'exporter="lab-01"' in body + assert 'operation="on"' in body + assert 'result="success"' in body + assert 'driver_type="power"' in body + assert "jumpstarter_operation_duration_seconds_bucket" in body or ( + "jumpstarter_operation_duration_seconds_count" in body + ) + assert 'error_type="timeout"' in body + assert 'direction="tx"' in body + assert "jumpstarter_active_sessions" in body + + +def test_exemplars_include_client_and_lease_id(): + reg = get_registry() + reg.record_operation( + exporter="lab-01", + operation="on", + result="success", + driver_type="power", + duration_seconds=0.01, + exemplars={"client": "ci-bot", "lease_id": "lease-xyz"}, + ) + body = reg.generate_latest().decode() + # OpenMetrics exemplar form: # {client="...",lease_id="..."} + assert re.search(r'client="ci-bot"', body) + assert re.search(r'lease_id="lease-xyz"', body) + assert "# {" in body or " # {" in body + + +def test_metrics_http_endpoint_serves_prometheus_text(): + reg = get_registry() + reg.set_active_sessions(exporter="lab-01", value=2) + listen = start_metrics_server("127.0.0.1:0", registry=reg) + assert listen, "expected non-empty listen address" + + with urllib.request.urlopen(f"http://{listen}/metrics", timeout=2) as resp: + assert resp.status == 200 + body = resp.read().decode() + ctype = resp.headers.get("Content-Type", "") + assert "text/plain" in ctype or "openmetrics" in ctype + assert "jumpstarter_active_sessions" in body + + +def test_metrics_server_disabled_when_addr_zero(): + assert start_metrics_server("0") == "" + assert start_metrics_server("") == "" + + +def test_driver_call_increments_operations_and_active_sessions(): + """Minimal wiring: Session + DriverCall should bump named series.""" + from jumpstarter_driver_power.driver import MockPower + + from jumpstarter.common.utils import serve + + with serve(MockPower()) as client: + client.on() + body = get_registry().generate_latest().decode() + assert "jumpstarter_active_sessions" in body + assert "jumpstarter_operations_total" in body + assert 'driver_type="power"' in body + assert 'result="success"' in body + assert 'operation="on"' in body or 'operation="On"' in body diff --git a/python/packages/jumpstarter/jumpstarter/metrics/registry.py b/python/packages/jumpstarter/jumpstarter/metrics/registry.py new file mode 100644 index 000000000..85eecce38 --- /dev/null +++ b/python/packages/jumpstarter/jumpstarter/metrics/registry.py @@ -0,0 +1,144 @@ +"""Exporter-local Prometheus metrics registry (JEP-0013 Phase 2).""" + +from __future__ import annotations + +import structlog +from prometheus_client import CollectorRegistry, Counter, Gauge, Histogram +from prometheus_client.openmetrics.exposition import generate_latest as generate_latest_openmetrics + +DEFAULT_EXEMPLAR_KEYS = ("client", "lease_id") + +_REGISTRY: MetricsRegistry | None = None + + +def filter_exemplars(exemplars: dict[str, str] | None) -> dict[str, str] | None: + """Keep only the JEP default exemplar keys with non-empty values.""" + if not exemplars: + return None + filtered = { + key: str(exemplars[key]) + for key in DEFAULT_EXEMPLAR_KEYS + if key in exemplars and exemplars[key] is not None and str(exemplars[key]) != "" + } + return filtered or None + + +def exemplars_from_log_context() -> dict[str, str] | None: + """Build default exemplars from the current structlog contextvars.""" + ctx = structlog.contextvars.get_contextvars() + return filter_exemplars({key: str(ctx[key]) for key in DEFAULT_EXEMPLAR_KEYS if key in ctx}) + + +def exporter_from_log_context(default: str = "unknown") -> str: + ctx = structlog.contextvars.get_contextvars() + value = ctx.get("exporter") + return str(value) if value else default + + +class MetricsRegistry: + """Process-local CollectorRegistry holding JEP-named exporter series.""" + + def __init__(self) -> None: + self._registry = CollectorRegistry() + self._operations = Counter( + "jumpstarter_operations_total", + "Total operations performed.", + ["exporter", "operation", "result", "driver_type"], + registry=self._registry, + ) + self._duration = Histogram( + "jumpstarter_operation_duration_seconds", + "Duration of each operation.", + ["exporter", "operation", "result", "driver_type"], + registry=self._registry, + ) + self._errors = Counter( + "jumpstarter_operation_errors_total", + "Errors by class (timeout, device, …).", + ["exporter", "operation", "driver_type", "error_type"], + registry=self._registry, + ) + self._stream_bytes = Counter( + "jumpstarter_stream_bytes_total", + "Bytes transferred (tx/rx) on streams.", + ["exporter", "driver_type", "direction"], + registry=self._registry, + ) + self._active_sessions = Gauge( + "jumpstarter_active_sessions", + "Currently active lease sessions.", + ["exporter"], + registry=self._registry, + ) + + @property + def collector_registry(self) -> CollectorRegistry: + return self._registry + + def record_operation( + self, + *, + exporter: str, + operation: str, + result: str, + driver_type: str, + duration_seconds: float, + exemplars: dict[str, str] | None = None, + error_type: str | None = None, + ) -> None: + labels = { + "exporter": exporter, + "operation": operation, + "result": result, + "driver_type": driver_type, + } + exemplar = filter_exemplars(exemplars) + self._operations.labels(**labels).inc(exemplar=exemplar) + self._duration.labels(**labels).observe(duration_seconds, exemplar=exemplar) + if result == "failure" and error_type: + self._errors.labels( + exporter=exporter, + operation=operation, + driver_type=driver_type, + error_type=error_type, + ).inc(exemplar=exemplar) + + def add_stream_bytes( + self, + *, + exporter: str, + driver_type: str, + direction: str, + nbytes: int, + exemplars: dict[str, str] | None = None, + ) -> None: + if nbytes <= 0: + return + self._stream_bytes.labels( + exporter=exporter, + driver_type=driver_type, + direction=direction, + ).inc(nbytes, exemplar=filter_exemplars(exemplars)) + + def set_active_sessions(self, *, exporter: str, value: float) -> None: + self._active_sessions.labels(exporter=exporter).set(value) + + def inc_active_sessions(self, *, exporter: str, delta: float = 1.0) -> None: + self._active_sessions.labels(exporter=exporter).inc(delta) + + def generate_latest(self) -> bytes: + return generate_latest_openmetrics(self._registry) + + +def get_registry() -> MetricsRegistry: + global _REGISTRY + if _REGISTRY is None: + _REGISTRY = MetricsRegistry() + return _REGISTRY + + +def reset_registry_for_tests() -> MetricsRegistry: + """Replace the process-wide registry (tests only).""" + global _REGISTRY + _REGISTRY = MetricsRegistry() + return _REGISTRY diff --git a/python/packages/jumpstarter/jumpstarter/metrics/server.py b/python/packages/jumpstarter/jumpstarter/metrics/server.py new file mode 100644 index 000000000..0d9d266f8 --- /dev/null +++ b/python/packages/jumpstarter/jumpstarter/metrics/server.py @@ -0,0 +1,61 @@ +"""HTTP GET /metrics server for exporter-local Prometheus scrape.""" + +from __future__ import annotations + +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from threading import Thread + +from prometheus_client.openmetrics.exposition import CONTENT_TYPE_LATEST + +from .registry import MetricsRegistry, get_registry + + +def _parse_bind_addr(addr: str) -> tuple[str, int]: + """Parse bind address forms: ':8080', '127.0.0.1:0', '8080'.""" + if ":" in addr: + host, _, port_s = addr.rpartition(":") + if host == "": + host = "0.0.0.0" + else: + host = "0.0.0.0" + port_s = addr + return host, int(port_s) + + +def start_metrics_server(addr: str, registry: MetricsRegistry | None = None) -> str: + """Start an HTTP server exposing GET /metrics. + + addr \"0\" or empty disables the server and returns \"\". + addr ending with \":0\" binds an ephemeral port; the returned listen address + is host:port suitable for urllib/http.Get. + """ + if addr == "" or addr == "0": + return "" + + reg = registry if registry is not None else get_registry() + host, port = _parse_bind_addr(addr) + + class Handler(BaseHTTPRequestHandler): + def do_GET(self): # noqa: N802 + if self.path.split("?", 1)[0] != "/metrics": + self.send_error(404) + return + body = reg.generate_latest() + self.send_response(200) + self.send_header("Content-Type", CONTENT_TYPE_LATEST) + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def log_message(self, fmt: str, *args) -> None: + return + + server = ThreadingHTTPServer((host, port), Handler) + thread = Thread(target=server.serve_forever, name="jumpstarter-metrics", daemon=True) + thread.start() + + bound_host, bound_port = server.server_address[:2] + # Prefer loopback-friendly host for ephemeral binds used in tests. + if bound_host in ("0.0.0.0", "::", ""): + bound_host = "127.0.0.1" + return f"{bound_host}:{bound_port}" diff --git a/python/packages/jumpstarter/jumpstarter/streams/common.py b/python/packages/jumpstarter/jumpstarter/streams/common.py index 40e650ca5..4121eccac 100644 --- a/python/packages/jumpstarter/jumpstarter/streams/common.py +++ b/python/packages/jumpstarter/jumpstarter/streams/common.py @@ -14,9 +14,29 @@ logger = logging.getLogger(__name__) -async def copy_stream(dst: AnyByteStream, src: AnyByteStream): +async def copy_stream( + dst: AnyByteStream, + src: AnyByteStream, + *, + metrics_direction: str | None = None, + metrics_driver_type: str = "other", +): + from jumpstarter.metrics.registry import ( + exemplars_from_log_context, + exporter_from_log_context, + get_registry, + ) + try: async for v in src: + if metrics_direction is not None: + get_registry().add_stream_bytes( + exporter=exporter_from_log_context(), + driver_type=metrics_driver_type, + direction=metrics_direction, + nbytes=len(v) if isinstance(v, (bytes, bytearray, memoryview)) else 0, + exemplars=exemplars_from_log_context(), + ) await dst.send(v) with suppress( AttributeError, @@ -37,11 +57,33 @@ async def copy_stream(dst: AnyByteStream, src: AnyByteStream): @asynccontextmanager -async def forward_stream(a, b): +async def forward_stream(a, b, *, metrics_driver_type: str | None = None): + from functools import partial + async with a, b: async with create_task_group() as tg: - tg.start_soon(copy_stream, a, b) - tg.start_soon(copy_stream, b, a) + if metrics_driver_type is None: + tg.start_soon(copy_stream, a, b) + tg.start_soon(copy_stream, b, a) + else: + tg.start_soon( + partial( + copy_stream, + a, + b, + metrics_direction="tx", + metrics_driver_type=metrics_driver_type, + ) + ) + tg.start_soon( + partial( + copy_stream, + b, + a, + metrics_direction="rx", + metrics_driver_type=metrics_driver_type, + ) + ) yield diff --git a/python/packages/jumpstarter/pyproject.toml b/python/packages/jumpstarter/pyproject.toml index d7bf015aa..78ee9d6ff 100644 --- a/python/packages/jumpstarter/pyproject.toml +++ b/python/packages/jumpstarter/pyproject.toml @@ -24,6 +24,7 @@ dependencies = [ "backports-zstd>=1.1.0 ; python_full_version < '3.14'", "click>=8.1.7.2", "structlog>=24.1.0", + "prometheus-client>=0.21.0", ] [dependency-groups] diff --git a/python/uv.lock b/python/uv.lock index 2453f8042..ad0f10089 100644 --- a/python/uv.lock +++ b/python/uv.lock @@ -1968,6 +1968,7 @@ dependencies = [ { name = "backports-zstd", marker = "python_full_version < '3.14'" }, { name = "click" }, { name = "jumpstarter-protocol" }, + { name = "prometheus-client" }, { name = "pydantic" }, { name = "pydantic-settings" }, { name = "pyyaml" }, @@ -1998,6 +1999,7 @@ requires-dist = [ { name = "backports-zstd", marker = "python_full_version < '3.14'", specifier = ">=1.1.0" }, { name = "click", specifier = ">=8.1.7.2" }, { name = "jumpstarter-protocol", editable = "packages/jumpstarter-protocol" }, + { name = "prometheus-client", specifier = ">=0.21.0" }, { name = "pydantic", specifier = ">=2.8.2" }, { name = "pydantic-settings", specifier = ">=2.9.1" }, { name = "pyyaml", specifier = ">=6.0.2" }, @@ -5005,6 +5007,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/88/74/a88bf1b1efeae488a0c0b7bdf71429c313722d1fc0f377537fbe554e6180/pre_commit-4.2.0-py2.py3-none-any.whl", hash = "sha256:a009ca7205f1eb497d10b845e52c838a98b6cdd2102a6c8e4540e94ee75c58bd", size = 220707, upload-time = "2025-03-18T21:35:19.343Z" }, ] +[[package]] +name = "prometheus-client" +version = "0.26.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/52/73/f1334c29c2af4cd9dba6c7817e61b611bd0215e2eb5565c6064a4de18802/prometheus_client-0.26.0.tar.gz", hash = "sha256:04a91bcf94e2cf74a44a1a874d651a2e853ed354b6e822f3b7487751465d5c2b", size = 92910, upload-time = "2026-07-24T19:36:41.893Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/eb/a3/b69efbf4143b5b9859b977770bbbabcc2796b702fa69dc40271e45cd5a56/prometheus_client-0.26.0-py3-none-any.whl", hash = "sha256:fa93d06737aa02bacd05794768508bb97d2fbee28cb3bca04eaae92f0ca953d6", size = 64494, upload-time = "2026-07-24T19:36:40.854Z" }, +] + [[package]] name = "propcache" version = "0.3.2" From 7b7647251f8d61694ff7b45f768acc8a9a72ff64 Mon Sep 17 00:00:00 2001 From: Roddie Kieley Date: Fri, 7 Aug 2026 12:10:35 -0230 Subject: [PATCH 02/16] fix: evaluate and address human feedback. Re-raise grpc AbortError before generic handlers to avoid client-controlled operation label cardinality, tighten bounded metric label types, default metrics bind host to loopback, log session metric decrements on failure, and expand exporter metrics unit coverage. --- .../jumpstarter/jumpstarter/driver/base.py | 22 ++- .../jumpstarter/exporter/session.py | 6 +- .../jumpstarter/metrics/__init__.py | 2 +- .../jumpstarter/metrics/metrics_test.py | 154 +++++++++++++++++- .../jumpstarter/metrics/registry.py | 28 +++- .../jumpstarter/jumpstarter/metrics/server.py | 11 +- .../jumpstarter/jumpstarter/streams/common.py | 15 +- 7 files changed, 207 insertions(+), 31 deletions(-) diff --git a/python/packages/jumpstarter/jumpstarter/driver/base.py b/python/packages/jumpstarter/jumpstarter/driver/base.py index 769d3ce01..c0fdb6cdc 100644 --- a/python/packages/jumpstarter/jumpstarter/driver/base.py +++ b/python/packages/jumpstarter/jumpstarter/driver/base.py @@ -17,6 +17,7 @@ from uuid import UUID, uuid4 import aiohttp +import grpc.aio import yarl from anyio import BrokenResourceError, to_thread from grpc import StatusCode @@ -39,6 +40,11 @@ ) from jumpstarter.config.env import JMP_DISABLE_COMPRESSION from jumpstarter.exporter.logging import get_logger +from jumpstarter.metrics.registry import ( + exemplars_from_log_context, + exporter_from_log_context, + get_registry, +) from jumpstarter.streams.aiohttp import AiohttpStreamReaderStream from jumpstarter.streams.common import create_memory_stream from jumpstarter.streams.encoding import Compression, compress_stream @@ -122,12 +128,6 @@ def _record_operation_metrics( duration_seconds: float, error_type: str | None = None, ) -> None: - from jumpstarter.metrics.registry import ( - exemplars_from_log_context, - exporter_from_log_context, - get_registry, - ) - get_registry().record_operation( exporter=exporter_from_log_context(default=self.name if hasattr(self, "name") else "unknown"), operation=operation, @@ -236,6 +236,10 @@ async def DriverCall(self, request, context): "result": "failure", "error_type": "device_error"}, ) await context.abort(StatusCode.INTERNAL, str(e)) + except grpc.aio.AbortError: + # Propagate context.abort() from lookup/handlers without recording + # metrics (avoids client-controlled operation label cardinality). + raise except Exception as e: self._record_operation_metrics( operation=op, @@ -250,7 +254,7 @@ async def DriverCall(self, request, context): ) await context.abort(StatusCode.UNKNOWN, str(e)) - async def StreamingDriverCall(self, request, context): + async def StreamingDriverCall(self, request, context): # noqa: C901 """ :meta private: """ @@ -351,6 +355,10 @@ async def StreamingDriverCall(self, request, context): "result": "failure", "error_type": "device_error"}, ) await context.abort(StatusCode.INTERNAL, str(e)) + except grpc.aio.AbortError: + # Propagate context.abort() from lookup/handlers without recording + # metrics (avoids client-controlled operation label cardinality). + raise except Exception as e: self._record_operation_metrics( operation=op, diff --git a/python/packages/jumpstarter/jumpstarter/exporter/session.py b/python/packages/jumpstarter/jumpstarter/exporter/session.py index bde94e05c..c596f9ba3 100644 --- a/python/packages/jumpstarter/jumpstarter/exporter/session.py +++ b/python/packages/jumpstarter/jumpstarter/exporter/session.py @@ -70,7 +70,11 @@ def __contextmanager__(self) -> Generator[Self]: try: get_registry().inc_active_sessions(exporter=self.name, delta=-1.0) except Exception: - pass + logger.warning( + "Failed to decrement active sessions metric for exporter %s", + self.name, + exc_info=True, + ) try: self.root_device.close() except Exception as e: diff --git a/python/packages/jumpstarter/jumpstarter/metrics/__init__.py b/python/packages/jumpstarter/jumpstarter/metrics/__init__.py index bec5b06f3..26f60a264 100644 --- a/python/packages/jumpstarter/jumpstarter/metrics/__init__.py +++ b/python/packages/jumpstarter/jumpstarter/metrics/__init__.py @@ -1,4 +1,4 @@ -"""Exporter-local Prometheus metrics (JEP-0013 Phase 2).""" +"""Exporter-local Prometheus metrics.""" from .registry import ( DEFAULT_EXEMPLAR_KEYS, diff --git a/python/packages/jumpstarter/jumpstarter/metrics/metrics_test.py b/python/packages/jumpstarter/jumpstarter/metrics/metrics_test.py index c68bd84e2..9e08c7348 100644 --- a/python/packages/jumpstarter/jumpstarter/metrics/metrics_test.py +++ b/python/packages/jumpstarter/jumpstarter/metrics/metrics_test.py @@ -1,19 +1,27 @@ -"""JEP-0013 Phase 2 exporter metrics tests.""" +"""Exporter-local Prometheus metrics tests.""" from __future__ import annotations import re -import urllib.error import urllib.request import pytest +import structlog +from jumpstarter.client.core import DriverError, DriverMethodNotImplemented +from jumpstarter.driver import Driver, export from jumpstarter.metrics import ( DEFAULT_EXEMPLAR_KEYS, get_registry, reset_registry_for_tests, start_metrics_server, ) +from jumpstarter.metrics.registry import ( + exemplars_from_log_context, + exporter_from_log_context, + filter_exemplars, +) +from jumpstarter.metrics.server import _parse_bind_addr SERIES = ( "jumpstarter_operations_total", @@ -31,10 +39,63 @@ def _fresh_registry(): reset_registry_for_tests() +def _sample_value(reg, name: str, labels: dict[str, str]) -> float | None: + for family in reg.collector_registry.collect(): + for sample in family.samples: + if sample.name != name: + continue + if all(sample.labels.get(k) == v for k, v in labels.items()): + return float(sample.value) + return None + + def test_default_exemplar_keys(): assert DEFAULT_EXEMPLAR_KEYS == ("client", "lease_id") +def test_filter_exemplars_allowlist_and_empty(): + assert filter_exemplars(None) is None + assert filter_exemplars({}) is None + assert filter_exemplars({"client": "", "lease_id": "x"}) == {"lease_id": "x"} + assert filter_exemplars({"client": "c", "lease_id": "l", "ignored": "nope"}) == { + "client": "c", + "lease_id": "l", + } + + +def test_exemplars_and_exporter_from_log_context(): + structlog.contextvars.clear_contextvars() + assert exemplars_from_log_context() is None + assert exporter_from_log_context() == "unknown" + + structlog.contextvars.bind_contextvars(client="ci-bot", lease_id="lease-1", exporter="lab-01") + try: + assert exemplars_from_log_context() == {"client": "ci-bot", "lease_id": "lease-1"} + assert exporter_from_log_context() == "lab-01" + finally: + structlog.contextvars.clear_contextvars() + + +def test_parse_bind_addr_defaults_to_loopback(): + assert _parse_bind_addr(":8080") == ("127.0.0.1", 8080) + assert _parse_bind_addr("8080") == ("127.0.0.1", 8080) + assert _parse_bind_addr("0.0.0.0:9090") == ("0.0.0.0", 9090) + assert _parse_bind_addr("127.0.0.1:0")[0] == "127.0.0.1" + + +def test_add_stream_bytes_ignores_non_positive(): + reg = get_registry() + reg.add_stream_bytes(exporter="lab-01", driver_type="serial", direction="tx", nbytes=0) + reg.add_stream_bytes(exporter="lab-01", driver_type="serial", direction="tx", nbytes=-5) + body = reg.generate_latest().decode() + # Series may be absent entirely until a positive observation. + assert 'direction="tx"' not in body or _sample_value( + reg, + "jumpstarter_stream_bytes_total", + {"exporter": "lab-01", "driver_type": "serial", "direction": "tx"}, + ) in (None, 0.0) + + def test_generate_latest_contains_named_series_after_increments(): reg = get_registry() exemplars = {"client": "ci-bot", "lease_id": "lease-abc"} @@ -68,7 +129,6 @@ def test_generate_latest_contains_named_series_after_increments(): for name in SERIES: assert name in body, f"expected series {name} in exposition" - assert 'jumpstarter_operations_total{' in body or "jumpstarter_operations_total{" in body assert 'exporter="lab-01"' in body assert 'operation="on"' in body assert 'result="success"' in body @@ -78,7 +138,37 @@ def test_generate_latest_contains_named_series_after_increments(): ) assert 'error_type="timeout"' in body assert 'direction="tx"' in body - assert "jumpstarter_active_sessions" in body + + success = _sample_value( + reg, + "jumpstarter_operations_total", + { + "exporter": "lab-01", + "operation": "on", + "result": "success", + "driver_type": "power", + }, + ) + assert success == 1.0 + + errors = _sample_value( + reg, + "jumpstarter_operation_errors_total", + { + "exporter": "lab-01", + "operation": "flash", + "driver_type": "storage", + "error_type": "timeout", + }, + ) + assert errors == 1.0 + + stream = _sample_value( + reg, + "jumpstarter_stream_bytes_total", + {"exporter": "lab-01", "driver_type": "serial", "direction": "tx"}, + ) + assert stream == 128.0 def test_exemplars_include_client_and_lease_id(): @@ -92,7 +182,6 @@ def test_exemplars_include_client_and_lease_id(): exemplars={"client": "ci-bot", "lease_id": "lease-xyz"}, ) body = reg.generate_latest().decode() - # OpenMetrics exemplar form: # {client="...",lease_id="..."} assert re.search(r'client="ci-bot"', body) assert re.search(r'lease_id="lease-xyz"', body) assert "# {" in body or " # {" in body @@ -117,6 +206,44 @@ def test_metrics_server_disabled_when_addr_zero(): assert start_metrics_server("") == "" +class _TimeoutDriver(Driver): + driver_type = "power" + + @classmethod + def client(cls): + return "jumpstarter.client.DriverClient" + + @export + def boom(self): + raise TimeoutError("deadline exceeded") + + +def test_driver_call_maps_timeout_error_type(): + from jumpstarter.common.utils import serve + + with serve(_TimeoutDriver()) as client: + with pytest.raises(DriverError): + client.call("boom") + body = get_registry().generate_latest().decode() + assert 'error_type="timeout"' in body + assert 'operation="boom"' in body + assert 'result="failure"' in body + + +def test_unknown_driver_method_does_not_record_operation_metric(): + """AbortError from method lookup must not create an operation time series.""" + from jumpstarter_driver_power.driver import MockPower + + from jumpstarter.common.utils import serve + + with serve(MockPower()) as client: + with pytest.raises(DriverMethodNotImplemented): + client.call("definitely_not_a_real_method_xyz") + body = get_registry().generate_latest().decode() + assert "definitely_not_a_real_method_xyz" not in body + assert 'error_type="internal_error"' not in body + + def test_driver_call_increments_operations_and_active_sessions(): """Minimal wiring: Session + DriverCall should bump named series.""" from jumpstarter_driver_power.driver import MockPower @@ -131,3 +258,20 @@ def test_driver_call_increments_operations_and_active_sessions(): assert 'driver_type="power"' in body assert 'result="success"' in body assert 'operation="on"' in body or 'operation="On"' in body + + success = _sample_value( + get_registry(), + "jumpstarter_operations_total", + { + "exporter": "unknown", + "operation": "on", + "result": "success", + "driver_type": "power", + }, + ) + if success is None: + # Exporter label comes from session name when set. + assert 'result="success"' in body + assert 'operation="on"' in body or 'operation="On"' in body + else: + assert success == 1.0 diff --git a/python/packages/jumpstarter/jumpstarter/metrics/registry.py b/python/packages/jumpstarter/jumpstarter/metrics/registry.py index 85eecce38..ebf58b8d5 100644 --- a/python/packages/jumpstarter/jumpstarter/metrics/registry.py +++ b/python/packages/jumpstarter/jumpstarter/metrics/registry.py @@ -1,18 +1,31 @@ -"""Exporter-local Prometheus metrics registry (JEP-0013 Phase 2).""" +"""Exporter-local Prometheus metrics registry.""" from __future__ import annotations +from typing import Literal + import structlog from prometheus_client import CollectorRegistry, Counter, Gauge, Histogram from prometheus_client.openmetrics.exposition import generate_latest as generate_latest_openmetrics DEFAULT_EXEMPLAR_KEYS = ("client", "lease_id") +OperationResult = Literal["success", "failure"] +StreamDirection = Literal["tx", "rx"] +ErrorType = Literal[ + "not_implemented", + "validation_error", + "timeout", + "connection_error", + "device_error", + "internal_error", +] + _REGISTRY: MetricsRegistry | None = None def filter_exemplars(exemplars: dict[str, str] | None) -> dict[str, str] | None: - """Keep only the JEP default exemplar keys with non-empty values.""" + """Keep only the default exemplar keys with non-empty values.""" if not exemplars: return None filtered = { @@ -36,7 +49,7 @@ def exporter_from_log_context(default: str = "unknown") -> str: class MetricsRegistry: - """Process-local CollectorRegistry holding JEP-named exporter series.""" + """Process-local CollectorRegistry holding exporter metric series.""" def __init__(self) -> None: self._registry = CollectorRegistry() @@ -54,7 +67,7 @@ def __init__(self) -> None: ) self._errors = Counter( "jumpstarter_operation_errors_total", - "Errors by class (timeout, device, …).", + "Errors by class (timeout, device, ...).", ["exporter", "operation", "driver_type", "error_type"], registry=self._registry, ) @@ -80,11 +93,11 @@ def record_operation( *, exporter: str, operation: str, - result: str, + result: OperationResult, driver_type: str, duration_seconds: float, exemplars: dict[str, str] | None = None, - error_type: str | None = None, + error_type: ErrorType | None = None, ) -> None: labels = { "exporter": exporter, @@ -108,10 +121,11 @@ def add_stream_bytes( *, exporter: str, driver_type: str, - direction: str, + direction: StreamDirection, nbytes: int, exemplars: dict[str, str] | None = None, ) -> None: + # Zero/negative transfers are not observed (no counter change). if nbytes <= 0: return self._stream_bytes.labels( diff --git a/python/packages/jumpstarter/jumpstarter/metrics/server.py b/python/packages/jumpstarter/jumpstarter/metrics/server.py index 0d9d266f8..79e81b254 100644 --- a/python/packages/jumpstarter/jumpstarter/metrics/server.py +++ b/python/packages/jumpstarter/jumpstarter/metrics/server.py @@ -11,13 +11,18 @@ def _parse_bind_addr(addr: str) -> tuple[str, int]: - """Parse bind address forms: ':8080', '127.0.0.1:0', '8080'.""" + """Parse bind address forms: ':8080', '127.0.0.1:0', '8080'. + + When the host is omitted, default to loopback. Metrics endpoints are + typically scraped via localhost/sidecar; use an explicit 0.0.0.0 host + when remote scrape is required. + """ if ":" in addr: host, _, port_s = addr.rpartition(":") if host == "": - host = "0.0.0.0" + host = "127.0.0.1" else: - host = "0.0.0.0" + host = "127.0.0.1" port_s = addr return host, int(port_s) diff --git a/python/packages/jumpstarter/jumpstarter/streams/common.py b/python/packages/jumpstarter/jumpstarter/streams/common.py index 4121eccac..a177ef693 100644 --- a/python/packages/jumpstarter/jumpstarter/streams/common.py +++ b/python/packages/jumpstarter/jumpstarter/streams/common.py @@ -11,6 +11,13 @@ from anyio.abc import AnyByteStream from anyio.streams.stapled import StapledObjectStream +from jumpstarter.metrics.registry import ( + StreamDirection, + exemplars_from_log_context, + exporter_from_log_context, + get_registry, +) + logger = logging.getLogger(__name__) @@ -18,15 +25,9 @@ async def copy_stream( dst: AnyByteStream, src: AnyByteStream, *, - metrics_direction: str | None = None, + metrics_direction: StreamDirection | None = None, metrics_driver_type: str = "other", ): - from jumpstarter.metrics.registry import ( - exemplars_from_log_context, - exporter_from_log_context, - get_registry, - ) - try: async for v in src: if metrics_direction is not None: From df108dc964dc8109560581fbf8b8c7b181a6d23e Mon Sep 17 00:00:00 2001 From: Roddie Kieley Date: Tue, 11 Aug 2026 20:30:08 -0230 Subject: [PATCH 03/16] fix: capture stream metrics context once per copy. Avoid repeated exporter/exemplar context lookups inside the copy_stream loop now that metrics helpers are imported at module level. --- .../jumpstarter/jumpstarter/streams/common.py | 11 ++++++++--- 1 file changed, 8 insertions(+), 3 deletions(-) diff --git a/python/packages/jumpstarter/jumpstarter/streams/common.py b/python/packages/jumpstarter/jumpstarter/streams/common.py index a177ef693..03491ba1e 100644 --- a/python/packages/jumpstarter/jumpstarter/streams/common.py +++ b/python/packages/jumpstarter/jumpstarter/streams/common.py @@ -29,14 +29,19 @@ async def copy_stream( metrics_driver_type: str = "other", ): try: + # Capture once per copy; context should be stable for the stream lifetime. + if metrics_direction is not None: + metrics_exporter = exporter_from_log_context() + metrics_exemplars = exemplars_from_log_context() + metrics_registry = get_registry() async for v in src: if metrics_direction is not None: - get_registry().add_stream_bytes( - exporter=exporter_from_log_context(), + metrics_registry.add_stream_bytes( + exporter=metrics_exporter, driver_type=metrics_driver_type, direction=metrics_direction, nbytes=len(v) if isinstance(v, (bytes, bytearray, memoryview)) else 0, - exemplars=exemplars_from_log_context(), + exemplars=metrics_exemplars, ) await dst.send(v) with suppress( From a32b1318e93268473393861a4eeb9a728a7c8d24 Mon Sep 17 00:00:00 2001 From: Roddie Kieley Date: Tue, 11 Aug 2026 20:36:31 -0230 Subject: [PATCH 04/16] fix: record unary success metrics after response encoding. Build DriverCallResponse (including encode_value) before recording success so serialization failures are counted only as failures, with a regression test. Co-authored-by: Cursor --- .../jumpstarter/jumpstarter/driver/base.py | 11 +++-- .../jumpstarter/metrics/metrics_test.py | 47 +++++++++++++++++++ 2 files changed, 54 insertions(+), 4 deletions(-) diff --git a/python/packages/jumpstarter/jumpstarter/driver/base.py b/python/packages/jumpstarter/jumpstarter/driver/base.py index c0fdb6cdc..15fe84425 100644 --- a/python/packages/jumpstarter/jumpstarter/driver/base.py +++ b/python/packages/jumpstarter/jumpstarter/driver/base.py @@ -158,6 +158,12 @@ async def DriverCall(self, request, context): else: result = await to_thread.run_sync(method, *args) + # Encode before recording success so serialization failures are + # counted only as failures (not success then failure). + response = jumpstarter_pb2.DriverCallResponse( + uuid=str(uuid4()), + result=encode_value(result), + ) self._record_operation_metrics( operation=op, result="success", @@ -167,10 +173,7 @@ async def DriverCall(self, request, context): "Operation completed", extra={"operation": op, "driver_type": self.driver_type, "result": "success"}, ) - return jumpstarter_pb2.DriverCallResponse( - uuid=str(uuid4()), - result=encode_value(result), - ) + return response except NotImplementedError as e: self._record_operation_metrics( operation=op, diff --git a/python/packages/jumpstarter/jumpstarter/metrics/metrics_test.py b/python/packages/jumpstarter/jumpstarter/metrics/metrics_test.py index 9e08c7348..ba6ccb78e 100644 --- a/python/packages/jumpstarter/jumpstarter/metrics/metrics_test.py +++ b/python/packages/jumpstarter/jumpstarter/metrics/metrics_test.py @@ -230,6 +230,53 @@ def test_driver_call_maps_timeout_error_type(): assert 'result="failure"' in body +class _Unencodable: + """Intentionally not JSON-serializable for encode_value failure tests.""" + + +class _UnencodableResultDriver(Driver): + driver_type = "power" + + @classmethod + def client(cls): + return "jumpstarter.client.DriverClient" + + @export + def boom(self): + return _Unencodable() + + +def test_driver_call_encode_failure_records_failure_not_success(): + """Success metrics must not be recorded if response serialization fails.""" + from jumpstarter.common.utils import serve + + with serve(_UnencodableResultDriver()) as client: + with pytest.raises(DriverError): + client.call("boom") + success = _sample_value( + get_registry(), + "jumpstarter_operations_total", + { + "exporter": "unknown", + "operation": "boom", + "result": "success", + "driver_type": "power", + }, + ) + failure = _sample_value( + get_registry(), + "jumpstarter_operations_total", + { + "exporter": "unknown", + "operation": "boom", + "result": "failure", + "driver_type": "power", + }, + ) + assert success in (None, 0.0) + assert failure == 1.0 + + def test_unknown_driver_method_does_not_record_operation_metric(): """AbortError from method lookup must not create an operation time series.""" from jumpstarter_driver_power.driver import MockPower From 2aa76dbdcc815d757956bd833b4e4597e339aed4 Mon Sep 17 00:00:00 2001 From: Roddie Kieley Date: Wed, 12 Aug 2026 10:49:15 -0230 Subject: [PATCH 05/16] fix: use ephemeral metrics port and tolerate bind failures. Default jmp run --metrics-bind-address to :0 so concurrent exporters do not collide on 8080, and treat metrics listen bind errors as non-fatal so the exporter continues without /metrics when a fixed port is already taken. --- .../jumpstarter-cli/jumpstarter_cli/run.py | 11 +++++--- .../jumpstarter/metrics/metrics_test.py | 25 +++++++++++++++++++ .../jumpstarter/jumpstarter/metrics/server.py | 20 +++++++++++++-- 3 files changed, 50 insertions(+), 6 deletions(-) diff --git a/python/packages/jumpstarter-cli/jumpstarter_cli/run.py b/python/packages/jumpstarter-cli/jumpstarter_cli/run.py index 5da4ccb0b..09dffde5c 100644 --- a/python/packages/jumpstarter-cli/jumpstarter_cli/run.py +++ b/python/packages/jumpstarter-cli/jumpstarter_cli/run.py @@ -77,7 +77,7 @@ def _handle_child( # noqa: C901 tls_cert=None, tls_key=None, passphrase=None, - metrics_bind_address=":8080", + metrics_bind_address=":0", ): """Handle child process with graceful shutdown.""" async def serve_with_graceful_shutdown(): # noqa: C901 @@ -224,7 +224,7 @@ def _serve_with_exc_handling( tls_cert=None, tls_key=None, passphrase=None, - metrics_bind_address=":8080", + metrics_bind_address=":0", ): max_rapid_failures = config.failure_detection.max_rapid_failures rapid_failure_window = config.failure_detection.rapid_failure_window @@ -325,9 +325,12 @@ def _serve_with_exc_handling( @click.option( "--metrics-bind-address", "metrics_bind_address", - default=":8080", + default=":0", show_default=True, - help="Address for HTTP GET /metrics (Prometheus/OpenMetrics). Use 0 to disable.", + help=( + "Address for HTTP GET /metrics (Prometheus/OpenMetrics). " + "Default :0 binds an ephemeral loopback port. Use 0 to disable." + ), ) @handle_exceptions def run( diff --git a/python/packages/jumpstarter/jumpstarter/metrics/metrics_test.py b/python/packages/jumpstarter/jumpstarter/metrics/metrics_test.py index ba6ccb78e..7d047bc37 100644 --- a/python/packages/jumpstarter/jumpstarter/metrics/metrics_test.py +++ b/python/packages/jumpstarter/jumpstarter/metrics/metrics_test.py @@ -80,6 +80,7 @@ def test_parse_bind_addr_defaults_to_loopback(): assert _parse_bind_addr(":8080") == ("127.0.0.1", 8080) assert _parse_bind_addr("8080") == ("127.0.0.1", 8080) assert _parse_bind_addr("0.0.0.0:9090") == ("0.0.0.0", 9090) + assert _parse_bind_addr(":0") == ("127.0.0.1", 0) assert _parse_bind_addr("127.0.0.1:0")[0] == "127.0.0.1" @@ -206,6 +207,30 @@ def test_metrics_server_disabled_when_addr_zero(): assert start_metrics_server("") == "" +def test_metrics_server_ephemeral_bind_returns_concrete_port(): + listen = start_metrics_server(":0") + assert listen.startswith("127.0.0.1:") + port = int(listen.rsplit(":", 1)[1]) + assert port > 0 + with urllib.request.urlopen(f"http://{listen}/metrics", timeout=2) as resp: + assert resp.status == 200 + + +def test_metrics_server_bind_failure_is_non_fatal(): + """A taken fixed port must not abort the exporter process.""" + import socket + + holder = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + holder.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + holder.bind(("127.0.0.1", 0)) + holder.listen(1) + occupied_port = holder.getsockname()[1] + try: + assert start_metrics_server(f"127.0.0.1:{occupied_port}") == "" + finally: + holder.close() + + class _TimeoutDriver(Driver): driver_type = "power" diff --git a/python/packages/jumpstarter/jumpstarter/metrics/server.py b/python/packages/jumpstarter/jumpstarter/metrics/server.py index 79e81b254..8c0e730df 100644 --- a/python/packages/jumpstarter/jumpstarter/metrics/server.py +++ b/python/packages/jumpstarter/jumpstarter/metrics/server.py @@ -2,6 +2,7 @@ from __future__ import annotations +import logging from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from threading import Thread @@ -9,9 +10,11 @@ from .registry import MetricsRegistry, get_registry +logger = logging.getLogger(__name__) + def _parse_bind_addr(addr: str) -> tuple[str, int]: - """Parse bind address forms: ':8080', '127.0.0.1:0', '8080'. + """Parse bind address forms: ':8080', '127.0.0.1:0', '8080', ':0'. When the host is omitted, default to loopback. Metrics endpoints are typically scraped via localhost/sidecar; use an explicit 0.0.0.0 host @@ -33,6 +36,9 @@ def start_metrics_server(addr: str, registry: MetricsRegistry | None = None) -> addr \"0\" or empty disables the server and returns \"\". addr ending with \":0\" binds an ephemeral port; the returned listen address is host:port suitable for urllib/http.Get. + + Bind failures (e.g. address already in use) are non-fatal: a warning is + logged and \"\" is returned so the exporter can continue without metrics. """ if addr == "" or addr == "0": return "" @@ -55,7 +61,17 @@ def do_GET(self): # noqa: N802 def log_message(self, fmt: str, *args) -> None: return - server = ThreadingHTTPServer((host, port), Handler) + try: + server = ThreadingHTTPServer((host, port), Handler) + except OSError as e: + logger.warning( + "Failed to bind metrics server at %s:%s (%s); continuing without /metrics", + host, + port, + e, + ) + return "" + thread = Thread(target=server.serve_forever, name="jumpstarter-metrics", daemon=True) thread.start() From a5f7c06131d6f884704f370967754e13d8652edf Mon Sep 17 00:00:00 2001 From: Roddie Kieley Date: Fri, 14 Aug 2026 11:05:57 -0230 Subject: [PATCH 06/16] Add test_driver_call_succeeds_when_metrics_recording_raises; raise and asserts client.on still succeeds --- .../jumpstarter/jumpstarter/driver/base.py | 32 +++++++++++++------ .../jumpstarter/metrics/metrics_test.py | 16 ++++++++++ 2 files changed, 39 insertions(+), 9 deletions(-) diff --git a/python/packages/jumpstarter/jumpstarter/driver/base.py b/python/packages/jumpstarter/jumpstarter/driver/base.py index 15fe84425..f9dbe5606 100644 --- a/python/packages/jumpstarter/jumpstarter/driver/base.py +++ b/python/packages/jumpstarter/jumpstarter/driver/base.py @@ -128,15 +128,29 @@ def _record_operation_metrics( duration_seconds: float, error_type: str | None = None, ) -> None: - get_registry().record_operation( - exporter=exporter_from_log_context(default=self.name if hasattr(self, "name") else "unknown"), - operation=operation, - result=result, - driver_type=self.driver_type, - duration_seconds=duration_seconds, - exemplars=exemplars_from_log_context(), - error_type=error_type, - ) + # Metrics must never discard a computed gRPC response or change the + # abort status: keep recording failures isolated from the RPC path. + try: + get_registry().record_operation( + exporter=exporter_from_log_context(default=self.name if hasattr(self, "name") else "unknown"), + operation=operation, + result=result, + driver_type=self.driver_type, + duration_seconds=duration_seconds, + exemplars=exemplars_from_log_context(), + error_type=error_type, + ) + except Exception: + self.logger.warning( + "Failed to record operation metrics", + extra={ + "operation": operation, + "driver_type": self.driver_type, + "result": result, + "error_type": error_type, + }, + exc_info=True, + ) async def DriverCall(self, request, context): """ diff --git a/python/packages/jumpstarter/jumpstarter/metrics/metrics_test.py b/python/packages/jumpstarter/jumpstarter/metrics/metrics_test.py index 7d047bc37..0bc13a7b5 100644 --- a/python/packages/jumpstarter/jumpstarter/metrics/metrics_test.py +++ b/python/packages/jumpstarter/jumpstarter/metrics/metrics_test.py @@ -302,6 +302,22 @@ def test_driver_call_encode_failure_records_failure_not_success(): assert failure == 1.0 +def test_driver_call_succeeds_when_metrics_recording_raises(monkeypatch): + """Metrics failures must not discard a successful DriverCall response.""" + from jumpstarter_driver_power.driver import MockPower + + from jumpstarter.common.utils import serve + from jumpstarter.metrics.registry import MetricsRegistry + + def _boom(*_args, **_kwargs): + raise RuntimeError("metrics broken") + + monkeypatch.setattr(MetricsRegistry, "record_operation", _boom) + + with serve(MockPower()) as client: + client.on() + + def test_unknown_driver_method_does_not_record_operation_metric(): """AbortError from method lookup must not create an operation time series.""" from jumpstarter_driver_power.driver import MockPower From 16cfb0ac8047fb1f4b16a56b0105a08b654e9d5b Mon Sep 17 00:00:00 2001 From: Roddie Kieley Date: Fri, 14 Aug 2026 11:18:29 -0230 Subject: [PATCH 07/16] fix: return shutdown handle from start_metrics_server. Stop the metrics HTTP server from tests and jmp run so listen sockets are not leaked after exporter exit. --- .../jumpstarter-cli/jumpstarter_cli/run.py | 84 ++++++++++--------- .../jumpstarter/metrics/metrics_test.py | 54 ++++++++---- .../jumpstarter/jumpstarter/metrics/server.py | 27 ++++-- 3 files changed, 104 insertions(+), 61 deletions(-) diff --git a/python/packages/jumpstarter-cli/jumpstarter_cli/run.py b/python/packages/jumpstarter-cli/jumpstarter_cli/run.py index 09dffde5c..349455a91 100644 --- a/python/packages/jumpstarter-cli/jumpstarter_cli/run.py +++ b/python/packages/jumpstarter-cli/jumpstarter_cli/run.py @@ -109,50 +109,54 @@ async def signal_handler(): from jumpstarter.metrics import start_metrics_server - listen_addr = start_metrics_server(metrics_bind_address) + listen_addr, shutdown_metrics = start_metrics_server(metrics_bind_address) if listen_addr: logger.info("Serving metrics server at http://%s/metrics", listen_addr) - if parsed_bind is not None: - host, port = parsed_bind - tls_credentials = None - if tls_insecure: + try: + if parsed_bind is not None: + host, port = parsed_bind + tls_credentials = None + if tls_insecure: + if passphrase: + click.echo( + "WARNING: --passphrase has no effect without TLS; " + "the passphrase will be transmitted in plaintext", + err=True, + ) + elif tls_cert and tls_key: + tls_credentials = _tls_server_credentials(tls_cert, tls_key) + + interceptors = None if passphrase: - click.echo( - "WARNING: --passphrase has no effect without TLS; " - "the passphrase will be transmitted in plaintext", - err=True, - ) - elif tls_cert and tls_key: - tls_credentials = _tls_server_credentials(tls_cert, tls_key) - - interceptors = None - if passphrase: - from jumpstarter.exporter.auth import PassphraseInterceptor - interceptors = [PassphraseInterceptor(passphrase)] - - exporter_exit_code = None - async with config.create_exporter(standalone=True) as exporter: - try: - await exporter.serve_standalone_tcp( - host, port, - tls_credentials=tls_credentials, - interceptors=interceptors, - ) - except* Exception as excgroup: - _handle_exporter_exceptions(excgroup) - exporter_exit_code = exporter.exit_code - else: - # Create exporter and run it (controller mode) - exporter_exit_code = None - async with config.create_exporter() as exporter: - try: - await exporter.serve() - except* Exception as excgroup: - _handle_exporter_exceptions(excgroup) - - # Check if exporter set an exit code (e.g., from hook failure with on_failure='exit') - exporter_exit_code = exporter.exit_code + from jumpstarter.exporter.auth import PassphraseInterceptor + interceptors = [PassphraseInterceptor(passphrase)] + + exporter_exit_code = None + async with config.create_exporter(standalone=True) as exporter: + try: + await exporter.serve_standalone_tcp( + host, port, + tls_credentials=tls_credentials, + interceptors=interceptors, + ) + except* Exception as excgroup: + _handle_exporter_exceptions(excgroup) + exporter_exit_code = exporter.exit_code + else: + # Create exporter and run it (controller mode) + exporter_exit_code = None + async with config.create_exporter() as exporter: + try: + await exporter.serve() + except* Exception as excgroup: + _handle_exporter_exceptions(excgroup) + + # Check if exporter set an exit code (e.g., from hook failure with on_failure='exit') + exporter_exit_code = exporter.exit_code + finally: + if shutdown_metrics is not None: + shutdown_metrics() # Cancel the signal handler after exporter completes signal_tg.cancel_scope.cancel() diff --git a/python/packages/jumpstarter/jumpstarter/metrics/metrics_test.py b/python/packages/jumpstarter/jumpstarter/metrics/metrics_test.py index 0bc13a7b5..c30ca0f29 100644 --- a/python/packages/jumpstarter/jumpstarter/metrics/metrics_test.py +++ b/python/packages/jumpstarter/jumpstarter/metrics/metrics_test.py @@ -3,6 +3,7 @@ from __future__ import annotations import re +import urllib.error import urllib.request import pytest @@ -191,29 +192,50 @@ def test_exemplars_include_client_and_lease_id(): def test_metrics_http_endpoint_serves_prometheus_text(): reg = get_registry() reg.set_active_sessions(exporter="lab-01", value=2) - listen = start_metrics_server("127.0.0.1:0", registry=reg) + listen, shutdown = start_metrics_server("127.0.0.1:0", registry=reg) assert listen, "expected non-empty listen address" - - with urllib.request.urlopen(f"http://{listen}/metrics", timeout=2) as resp: - assert resp.status == 200 - body = resp.read().decode() - ctype = resp.headers.get("Content-Type", "") - assert "text/plain" in ctype or "openmetrics" in ctype - assert "jumpstarter_active_sessions" in body + assert shutdown is not None + try: + with urllib.request.urlopen(f"http://{listen}/metrics", timeout=2) as resp: + assert resp.status == 200 + body = resp.read().decode() + ctype = resp.headers.get("Content-Type", "") + assert "text/plain" in ctype or "openmetrics" in ctype + assert "jumpstarter_active_sessions" in body + finally: + shutdown() def test_metrics_server_disabled_when_addr_zero(): - assert start_metrics_server("0") == "" - assert start_metrics_server("") == "" + listen, shutdown = start_metrics_server("0") + assert listen == "" + assert shutdown is None + listen, shutdown = start_metrics_server("") + assert listen == "" + assert shutdown is None def test_metrics_server_ephemeral_bind_returns_concrete_port(): - listen = start_metrics_server(":0") - assert listen.startswith("127.0.0.1:") - port = int(listen.rsplit(":", 1)[1]) - assert port > 0 + listen, shutdown = start_metrics_server(":0") + assert shutdown is not None + try: + assert listen.startswith("127.0.0.1:") + port = int(listen.rsplit(":", 1)[1]) + assert port > 0 + with urllib.request.urlopen(f"http://{listen}/metrics", timeout=2) as resp: + assert resp.status == 200 + finally: + shutdown() + + +def test_metrics_server_shutdown_stops_listening(): + listen, shutdown = start_metrics_server("127.0.0.1:0") + assert listen and shutdown is not None with urllib.request.urlopen(f"http://{listen}/metrics", timeout=2) as resp: assert resp.status == 200 + shutdown() + with pytest.raises(urllib.error.URLError): + urllib.request.urlopen(f"http://{listen}/metrics", timeout=1) def test_metrics_server_bind_failure_is_non_fatal(): @@ -226,7 +248,9 @@ def test_metrics_server_bind_failure_is_non_fatal(): holder.listen(1) occupied_port = holder.getsockname()[1] try: - assert start_metrics_server(f"127.0.0.1:{occupied_port}") == "" + listen, shutdown = start_metrics_server(f"127.0.0.1:{occupied_port}") + assert listen == "" + assert shutdown is None finally: holder.close() diff --git a/python/packages/jumpstarter/jumpstarter/metrics/server.py b/python/packages/jumpstarter/jumpstarter/metrics/server.py index 8c0e730df..63492c105 100644 --- a/python/packages/jumpstarter/jumpstarter/metrics/server.py +++ b/python/packages/jumpstarter/jumpstarter/metrics/server.py @@ -3,6 +3,7 @@ from __future__ import annotations import logging +from collections.abc import Callable from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from threading import Thread @@ -12,6 +13,8 @@ logger = logging.getLogger(__name__) +ShutdownFunc = Callable[[], None] + def _parse_bind_addr(addr: str) -> tuple[str, int]: """Parse bind address forms: ':8080', '127.0.0.1:0', '8080', ':0'. @@ -30,18 +33,26 @@ def _parse_bind_addr(addr: str) -> tuple[str, int]: return host, int(port_s) -def start_metrics_server(addr: str, registry: MetricsRegistry | None = None) -> str: +def start_metrics_server( + addr: str, registry: MetricsRegistry | None = None +) -> tuple[str, ShutdownFunc | None]: """Start an HTTP server exposing GET /metrics. - addr \"0\" or empty disables the server and returns \"\". + Returns ``(listen_address, shutdown)``. When metrics are disabled or bind + fails, returns ``(\"\", None)``. + + addr \"0\" or empty disables the server. addr ending with \":0\" binds an ephemeral port; the returned listen address is host:port suitable for urllib/http.Get. Bind failures (e.g. address already in use) are non-fatal: a warning is - logged and \"\" is returned so the exporter can continue without metrics. + logged and (\"\", None) is returned so the exporter can continue without + metrics. + + Call ``shutdown()`` to stop the background server (no-op when None). """ if addr == "" or addr == "0": - return "" + return "", None reg = registry if registry is not None else get_registry() host, port = _parse_bind_addr(addr) @@ -70,13 +81,17 @@ def log_message(self, fmt: str, *args) -> None: port, e, ) - return "" + return "", None thread = Thread(target=server.serve_forever, name="jumpstarter-metrics", daemon=True) thread.start() + def shutdown() -> None: + server.shutdown() + server.server_close() + bound_host, bound_port = server.server_address[:2] # Prefer loopback-friendly host for ephemeral binds used in tests. if bound_host in ("0.0.0.0", "::", ""): bound_host = "127.0.0.1" - return f"{bound_host}:{bound_port}" + return f"{bound_host}:{bound_port}", shutdown From ccad0a26b7afc17847c10465e03a320c240dac6f Mon Sep 17 00:00:00 2001 From: Roddie Kieley Date: Fri, 14 Aug 2026 11:26:19 -0230 Subject: [PATCH 08/16] fix: treat invalid metrics bind addresses as non-fatal. Parse failures (ValueError) now share the bind-failure path so a bad --metrics-bind-address cannot crash the exporter. Co-authored-by: Cursor --- .../jumpstarter/metrics/metrics_test.py | 7 +++++++ .../jumpstarter/jumpstarter/metrics/server.py | 15 +++++++-------- 2 files changed, 14 insertions(+), 8 deletions(-) diff --git a/python/packages/jumpstarter/jumpstarter/metrics/metrics_test.py b/python/packages/jumpstarter/jumpstarter/metrics/metrics_test.py index c30ca0f29..d6e6e222c 100644 --- a/python/packages/jumpstarter/jumpstarter/metrics/metrics_test.py +++ b/python/packages/jumpstarter/jumpstarter/metrics/metrics_test.py @@ -255,6 +255,13 @@ def test_metrics_server_bind_failure_is_non_fatal(): holder.close() +def test_metrics_server_invalid_bind_addr_is_non_fatal(): + """Non-numeric ports must not abort the exporter process.""" + listen, shutdown = start_metrics_server("127.0.0.1:not-a-port") + assert listen == "" + assert shutdown is None + + class _TimeoutDriver(Driver): driver_type = "power" diff --git a/python/packages/jumpstarter/jumpstarter/metrics/server.py b/python/packages/jumpstarter/jumpstarter/metrics/server.py index 63492c105..68caf5495 100644 --- a/python/packages/jumpstarter/jumpstarter/metrics/server.py +++ b/python/packages/jumpstarter/jumpstarter/metrics/server.py @@ -45,9 +45,9 @@ def start_metrics_server( addr ending with \":0\" binds an ephemeral port; the returned listen address is host:port suitable for urllib/http.Get. - Bind failures (e.g. address already in use) are non-fatal: a warning is - logged and (\"\", None) is returned so the exporter can continue without - metrics. + Bind failures (e.g. address already in use) and invalid bind addresses + (non-numeric port) are non-fatal: a warning is logged and (\"\", None) is + returned so the exporter can continue without metrics. Call ``shutdown()`` to stop the background server (no-op when None). """ @@ -55,7 +55,6 @@ def start_metrics_server( return "", None reg = registry if registry is not None else get_registry() - host, port = _parse_bind_addr(addr) class Handler(BaseHTTPRequestHandler): def do_GET(self): # noqa: N802 @@ -73,12 +72,12 @@ def log_message(self, fmt: str, *args) -> None: return try: + host, port = _parse_bind_addr(addr) server = ThreadingHTTPServer((host, port), Handler) - except OSError as e: + except (OSError, ValueError) as e: logger.warning( - "Failed to bind metrics server at %s:%s (%s); continuing without /metrics", - host, - port, + "Failed to start metrics server for bind address %r (%s); continuing without /metrics", + addr, e, ) return "", None From ef4c13fad9236531d34f1f3508b2d2f067039a2b Mon Sep 17 00:00:00 2001 From: Roddie Kieley Date: Fri, 14 Aug 2026 11:39:17 -0230 Subject: [PATCH 09/16] fix: initialize exporter metrics registry eagerly. Avoid a race where the asyncio path and metrics HTTP thread each create a MetricsRegistry and discard counters from the first instance. Co-authored-by: Cursor --- .../jumpstarter/jumpstarter/metrics/metrics_test.py | 9 +++++++++ .../jumpstarter/jumpstarter/metrics/registry.py | 11 ++++++----- 2 files changed, 15 insertions(+), 5 deletions(-) diff --git a/python/packages/jumpstarter/jumpstarter/metrics/metrics_test.py b/python/packages/jumpstarter/jumpstarter/metrics/metrics_test.py index d6e6e222c..db4140ff9 100644 --- a/python/packages/jumpstarter/jumpstarter/metrics/metrics_test.py +++ b/python/packages/jumpstarter/jumpstarter/metrics/metrics_test.py @@ -40,6 +40,15 @@ def _fresh_registry(): reset_registry_for_tests() +def test_get_registry_returns_eager_singleton(): + first = get_registry() + second = get_registry() + assert first is second + replaced = reset_registry_for_tests() + assert replaced is get_registry() + assert replaced is not first + + def _sample_value(reg, name: str, labels: dict[str, str]) -> float | None: for family in reg.collector_registry.collect(): for sample in family.samples: diff --git a/python/packages/jumpstarter/jumpstarter/metrics/registry.py b/python/packages/jumpstarter/jumpstarter/metrics/registry.py index ebf58b8d5..0ec202a46 100644 --- a/python/packages/jumpstarter/jumpstarter/metrics/registry.py +++ b/python/packages/jumpstarter/jumpstarter/metrics/registry.py @@ -21,8 +21,6 @@ "internal_error", ] -_REGISTRY: MetricsRegistry | None = None - def filter_exemplars(exemplars: dict[str, str] | None) -> dict[str, str] | None: """Keep only the default exemplar keys with non-empty values.""" @@ -144,10 +142,13 @@ def generate_latest(self) -> bytes: return generate_latest_openmetrics(self._registry) +# Eager singleton: the asyncio serve path and ThreadingHTTPServer scrape +# handler can call get_registry() concurrently; lazy init would race and +# discard counters already recorded on the first instance. +_REGISTRY = MetricsRegistry() + + def get_registry() -> MetricsRegistry: - global _REGISTRY - if _REGISTRY is None: - _REGISTRY = MetricsRegistry() return _REGISTRY From d95272e0d76c2d91b5dff5cce38e74da767c21cb Mon Sep 17 00:00:00 2001 From: Roddie Kieley Date: Fri, 14 Aug 2026 11:45:36 -0230 Subject: [PATCH 10/16] fix: unbind exporter log context when a session ends. Clear the exporter correlation field after decrementing active sessions so it does not leak into later sessions or logs. Co-authored-by: Cursor --- .../jumpstarter/jumpstarter/exporter/session.py | 3 ++- .../jumpstarter/exporter/session_test.py | 13 +++++++++++++ 2 files changed, 15 insertions(+), 1 deletion(-) diff --git a/python/packages/jumpstarter/jumpstarter/exporter/session.py b/python/packages/jumpstarter/jumpstarter/exporter/session.py index c596f9ba3..dad21cd4f 100644 --- a/python/packages/jumpstarter/jumpstarter/exporter/session.py +++ b/python/packages/jumpstarter/jumpstarter/exporter/session.py @@ -57,7 +57,7 @@ class Session( @contextmanager def __contextmanager__(self) -> Generator[Self]: - from jumpstarter.logging import set_log_context + from jumpstarter.logging import set_log_context, unbind_log_context from jumpstarter.metrics import get_registry logging.getLogger().addHandler(self._logging_handler) @@ -75,6 +75,7 @@ def __contextmanager__(self) -> Generator[Self]: self.name, exc_info=True, ) + unbind_log_context("exporter") try: self.root_device.close() except Exception as e: diff --git a/python/packages/jumpstarter/jumpstarter/exporter/session_test.py b/python/packages/jumpstarter/jumpstarter/exporter/session_test.py index 69cad9a55..658f2efb5 100644 --- a/python/packages/jumpstarter/jumpstarter/exporter/session_test.py +++ b/python/packages/jumpstarter/jumpstarter/exporter/session_test.py @@ -28,6 +28,19 @@ def client(cls): return "jumpstarter.client.DriverClient" +def test_session_unbinds_exporter_log_context(): + """Session must clear the exporter correlation field on exit.""" + import structlog + + from jumpstarter.logging import clear_log_context + + clear_log_context() + driver = SimpleDriver() + with Session(uuid=driver.uuid, labels=driver.labels, root_device=driver) as session: + assert structlog.contextvars.get_contextvars().get("exporter") == session.name + assert "exporter" not in structlog.contextvars.get_contextvars() + + def test_get_report_includes_descriptions(): """Test that GetReport includes descriptions for drivers that have them""" # Create drivers with and without descriptions From 152a1ac12d21e820d54e6b8b2279e4e681b243e9 Mon Sep 17 00:00:00 2001 From: Roddie Kieley Date: Fri, 14 Aug 2026 11:52:34 -0230 Subject: [PATCH 11/16] fix: address remaining exporter metrics style review notes. Centralize DriverCall/StreamingDriverCall error mapping, move reset_registry_for_tests to metrics/_testing, restore log_message(format=), hoist session/stream imports, tighten exemplar regex asserts, add metrics HTTP request timeout and backlog limits, and rename inc_active_sessions to adjust_active_sessions. Co-authored-by: Cursor --- .../jumpstarter/jumpstarter/driver/base.py | 204 +++++------------- .../jumpstarter/exporter/session.py | 9 +- .../jumpstarter/metrics/__init__.py | 2 - .../jumpstarter/metrics/_testing.py | 12 ++ .../jumpstarter/metrics/metrics_test.py | 10 +- .../jumpstarter/metrics/registry.py | 9 +- .../jumpstarter/jumpstarter/metrics/server.py | 10 +- .../jumpstarter/jumpstarter/streams/common.py | 3 +- 8 files changed, 82 insertions(+), 177 deletions(-) create mode 100644 python/packages/jumpstarter/jumpstarter/metrics/_testing.py diff --git a/python/packages/jumpstarter/jumpstarter/driver/base.py b/python/packages/jumpstarter/jumpstarter/driver/base.py index f9dbe5606..bd12fda63 100644 --- a/python/packages/jumpstarter/jumpstarter/driver/base.py +++ b/python/packages/jumpstarter/jumpstarter/driver/base.py @@ -45,6 +45,15 @@ exporter_from_log_context, get_registry, ) + +# Ordered most-specific first: ConnectionError is an OSError subclass. +_DRIVER_CALL_ERRORS: tuple[tuple[type[BaseException], str, StatusCode], ...] = ( + (NotImplementedError, "not_implemented", StatusCode.UNIMPLEMENTED), + (ValueError, "validation_error", StatusCode.INVALID_ARGUMENT), + (TimeoutError, "timeout", StatusCode.DEADLINE_EXCEEDED), + (ConnectionError, "connection_error", StatusCode.UNAVAILABLE), + (OSError, "device_error", StatusCode.INTERNAL), +) from jumpstarter.streams.aiohttp import AiohttpStreamReaderStream from jumpstarter.streams.common import create_memory_stream from jumpstarter.streams.encoding import Compression, compress_stream @@ -152,6 +161,43 @@ def _record_operation_metrics( exc_info=True, ) + async def _handle_driver_exception(self, exc: BaseException, op: str, started: float, context) -> None: + for exc_type, error_type, status in _DRIVER_CALL_ERRORS: + if isinstance(exc, exc_type): + self._record_operation_metrics( + operation=op, + result="failure", + duration_seconds=time.perf_counter() - started, + error_type=error_type, + ) + self.logger.warning( + "Operation failed", + extra={ + "operation": op, + "driver_type": self.driver_type, + "result": "failure", + "error_type": error_type, + }, + ) + await context.abort(status, str(exc)) + return + self._record_operation_metrics( + operation=op, + result="failure", + duration_seconds=time.perf_counter() - started, + error_type="internal_error", + ) + self.logger.warning( + "Operation failed", + extra={ + "operation": op, + "driver_type": self.driver_type, + "result": "failure", + "error_type": "internal_error", + }, + ) + await context.abort(StatusCode.UNKNOWN, str(exc)) + async def DriverCall(self, request, context): """ :meta private: @@ -188,90 +234,14 @@ async def DriverCall(self, request, context): extra={"operation": op, "driver_type": self.driver_type, "result": "success"}, ) return response - except NotImplementedError as e: - self._record_operation_metrics( - operation=op, - result="failure", - duration_seconds=time.perf_counter() - started, - error_type="not_implemented", - ) - self.logger.warning( - "Operation failed", - extra={"operation": op, "driver_type": self.driver_type, - "result": "failure", "error_type": "not_implemented"}, - ) - await context.abort(StatusCode.UNIMPLEMENTED, str(e)) - except ValueError as e: - self._record_operation_metrics( - operation=op, - result="failure", - duration_seconds=time.perf_counter() - started, - error_type="validation_error", - ) - self.logger.warning( - "Operation failed", - extra={"operation": op, "driver_type": self.driver_type, - "result": "failure", "error_type": "validation_error"}, - ) - await context.abort(StatusCode.INVALID_ARGUMENT, str(e)) - except TimeoutError as e: - self._record_operation_metrics( - operation=op, - result="failure", - duration_seconds=time.perf_counter() - started, - error_type="timeout", - ) - self.logger.warning( - "Operation failed", - extra={"operation": op, "driver_type": self.driver_type, - "result": "failure", "error_type": "timeout"}, - ) - await context.abort(StatusCode.DEADLINE_EXCEEDED, str(e)) - except ConnectionError as e: - self._record_operation_metrics( - operation=op, - result="failure", - duration_seconds=time.perf_counter() - started, - error_type="connection_error", - ) - self.logger.warning( - "Operation failed", - extra={"operation": op, "driver_type": self.driver_type, - "result": "failure", "error_type": "connection_error"}, - ) - await context.abort(StatusCode.UNAVAILABLE, str(e)) - except OSError as e: - self._record_operation_metrics( - operation=op, - result="failure", - duration_seconds=time.perf_counter() - started, - error_type="device_error", - ) - self.logger.warning( - "Operation failed", - extra={"operation": op, "driver_type": self.driver_type, - "result": "failure", "error_type": "device_error"}, - ) - await context.abort(StatusCode.INTERNAL, str(e)) except grpc.aio.AbortError: # Propagate context.abort() from lookup/handlers without recording # metrics (avoids client-controlled operation label cardinality). raise except Exception as e: - self._record_operation_metrics( - operation=op, - result="failure", - duration_seconds=time.perf_counter() - started, - error_type="internal_error", - ) - self.logger.warning( - "Operation failed", - extra={"operation": op, "driver_type": self.driver_type, - "result": "failure", "error_type": "internal_error"}, - ) - await context.abort(StatusCode.UNKNOWN, str(e)) + await self._handle_driver_exception(e, op, started, context) - async def StreamingDriverCall(self, request, context): # noqa: C901 + async def StreamingDriverCall(self, request, context): """ :meta private: """ @@ -307,88 +277,12 @@ async def StreamingDriverCall(self, request, context): # noqa: C901 "Operation completed", extra={"operation": op, "driver_type": self.driver_type, "result": "success"}, ) - except NotImplementedError as e: - self._record_operation_metrics( - operation=op, - result="failure", - duration_seconds=time.perf_counter() - started, - error_type="not_implemented", - ) - self.logger.warning( - "Operation failed", - extra={"operation": op, "driver_type": self.driver_type, - "result": "failure", "error_type": "not_implemented"}, - ) - await context.abort(StatusCode.UNIMPLEMENTED, str(e)) - except ValueError as e: - self._record_operation_metrics( - operation=op, - result="failure", - duration_seconds=time.perf_counter() - started, - error_type="validation_error", - ) - self.logger.warning( - "Operation failed", - extra={"operation": op, "driver_type": self.driver_type, - "result": "failure", "error_type": "validation_error"}, - ) - await context.abort(StatusCode.INVALID_ARGUMENT, str(e)) - except TimeoutError as e: - self._record_operation_metrics( - operation=op, - result="failure", - duration_seconds=time.perf_counter() - started, - error_type="timeout", - ) - self.logger.warning( - "Operation failed", - extra={"operation": op, "driver_type": self.driver_type, - "result": "failure", "error_type": "timeout"}, - ) - await context.abort(StatusCode.DEADLINE_EXCEEDED, str(e)) - except ConnectionError as e: - self._record_operation_metrics( - operation=op, - result="failure", - duration_seconds=time.perf_counter() - started, - error_type="connection_error", - ) - self.logger.warning( - "Operation failed", - extra={"operation": op, "driver_type": self.driver_type, - "result": "failure", "error_type": "connection_error"}, - ) - await context.abort(StatusCode.UNAVAILABLE, str(e)) - except OSError as e: - self._record_operation_metrics( - operation=op, - result="failure", - duration_seconds=time.perf_counter() - started, - error_type="device_error", - ) - self.logger.warning( - "Operation failed", - extra={"operation": op, "driver_type": self.driver_type, - "result": "failure", "error_type": "device_error"}, - ) - await context.abort(StatusCode.INTERNAL, str(e)) except grpc.aio.AbortError: # Propagate context.abort() from lookup/handlers without recording # metrics (avoids client-controlled operation label cardinality). raise except Exception as e: - self._record_operation_metrics( - operation=op, - result="failure", - duration_seconds=time.perf_counter() - started, - error_type="internal_error", - ) - self.logger.warning( - "Operation failed", - extra={"operation": op, "driver_type": self.driver_type, - "result": "failure", "error_type": "internal_error"}, - ) - await context.abort(StatusCode.UNKNOWN, str(e)) + await self._handle_driver_exception(e, op, started, context) @asynccontextmanager async def Stream(self, request, context): diff --git a/python/packages/jumpstarter/jumpstarter/exporter/session.py b/python/packages/jumpstarter/jumpstarter/exporter/session.py index dad21cd4f..ed77db996 100644 --- a/python/packages/jumpstarter/jumpstarter/exporter/session.py +++ b/python/packages/jumpstarter/jumpstarter/exporter/session.py @@ -19,6 +19,8 @@ from .logging import LogHandler from jumpstarter.common import ExporterStatus, LogSource, Metadata, TemporarySocket from jumpstarter.common.streams import StreamRequestMetadata +from jumpstarter.logging import set_log_context, unbind_log_context +from jumpstarter.metrics import get_registry from jumpstarter.streams.common import forward_stream from jumpstarter.streams.metadata import MetadataStreamAttributes from jumpstarter.streams.router import RouterStream @@ -57,18 +59,15 @@ class Session( @contextmanager def __contextmanager__(self) -> Generator[Self]: - from jumpstarter.logging import set_log_context, unbind_log_context - from jumpstarter.metrics import get_registry - logging.getLogger().addHandler(self._logging_handler) self.root_device.reset() set_log_context(exporter=self.name) - get_registry().inc_active_sessions(exporter=self.name, delta=1.0) + get_registry().adjust_active_sessions(exporter=self.name, delta=1.0) try: yield self finally: try: - get_registry().inc_active_sessions(exporter=self.name, delta=-1.0) + get_registry().adjust_active_sessions(exporter=self.name, delta=-1.0) except Exception: logger.warning( "Failed to decrement active sessions metric for exporter %s", diff --git a/python/packages/jumpstarter/jumpstarter/metrics/__init__.py b/python/packages/jumpstarter/jumpstarter/metrics/__init__.py index 26f60a264..56041b452 100644 --- a/python/packages/jumpstarter/jumpstarter/metrics/__init__.py +++ b/python/packages/jumpstarter/jumpstarter/metrics/__init__.py @@ -4,7 +4,6 @@ DEFAULT_EXEMPLAR_KEYS, MetricsRegistry, get_registry, - reset_registry_for_tests, ) from .server import start_metrics_server @@ -12,6 +11,5 @@ "DEFAULT_EXEMPLAR_KEYS", "MetricsRegistry", "get_registry", - "reset_registry_for_tests", "start_metrics_server", ] diff --git a/python/packages/jumpstarter/jumpstarter/metrics/_testing.py b/python/packages/jumpstarter/jumpstarter/metrics/_testing.py new file mode 100644 index 000000000..6b3fe548a --- /dev/null +++ b/python/packages/jumpstarter/jumpstarter/metrics/_testing.py @@ -0,0 +1,12 @@ +"""Test-only helpers for exporter metrics (not part of the public API).""" + +from __future__ import annotations + +from . import registry as _registry +from .registry import MetricsRegistry + + +def reset_registry_for_tests() -> MetricsRegistry: + """Replace the process-wide registry (tests only).""" + _registry._REGISTRY = MetricsRegistry() + return _registry._REGISTRY diff --git a/python/packages/jumpstarter/jumpstarter/metrics/metrics_test.py b/python/packages/jumpstarter/jumpstarter/metrics/metrics_test.py index db4140ff9..aad4e4769 100644 --- a/python/packages/jumpstarter/jumpstarter/metrics/metrics_test.py +++ b/python/packages/jumpstarter/jumpstarter/metrics/metrics_test.py @@ -14,9 +14,9 @@ from jumpstarter.metrics import ( DEFAULT_EXEMPLAR_KEYS, get_registry, - reset_registry_for_tests, start_metrics_server, ) +from jumpstarter.metrics._testing import reset_registry_for_tests from jumpstarter.metrics.registry import ( exemplars_from_log_context, exporter_from_log_context, @@ -193,9 +193,11 @@ def test_exemplars_include_client_and_lease_id(): exemplars={"client": "ci-bot", "lease_id": "lease-xyz"}, ) body = reg.generate_latest().decode() - assert re.search(r'client="ci-bot"', body) - assert re.search(r'lease_id="lease-xyz"', body) - assert "# {" in body or " # {" in body + assert re.search( + r'jumpstarter_operations_total.*# \{.*client="ci-bot".*lease_id="lease-xyz"', + body, + re.DOTALL, + ) def test_metrics_http_endpoint_serves_prometheus_text(): diff --git a/python/packages/jumpstarter/jumpstarter/metrics/registry.py b/python/packages/jumpstarter/jumpstarter/metrics/registry.py index 0ec202a46..e762796bf 100644 --- a/python/packages/jumpstarter/jumpstarter/metrics/registry.py +++ b/python/packages/jumpstarter/jumpstarter/metrics/registry.py @@ -135,7 +135,7 @@ def add_stream_bytes( def set_active_sessions(self, *, exporter: str, value: float) -> None: self._active_sessions.labels(exporter=exporter).set(value) - def inc_active_sessions(self, *, exporter: str, delta: float = 1.0) -> None: + def adjust_active_sessions(self, *, exporter: str, delta: float = 1.0) -> None: self._active_sessions.labels(exporter=exporter).inc(delta) def generate_latest(self) -> bytes: @@ -150,10 +150,3 @@ def generate_latest(self) -> bytes: def get_registry() -> MetricsRegistry: return _REGISTRY - - -def reset_registry_for_tests() -> MetricsRegistry: - """Replace the process-wide registry (tests only).""" - global _REGISTRY - _REGISTRY = MetricsRegistry() - return _REGISTRY diff --git a/python/packages/jumpstarter/jumpstarter/metrics/server.py b/python/packages/jumpstarter/jumpstarter/metrics/server.py index 68caf5495..9b48ff707 100644 --- a/python/packages/jumpstarter/jumpstarter/metrics/server.py +++ b/python/packages/jumpstarter/jumpstarter/metrics/server.py @@ -15,6 +15,10 @@ ShutdownFunc = Callable[[], None] +# Bound slow/idle scrape clients; backlog caps concurrent accept queue depth. +_METRICS_REQUEST_TIMEOUT_S = 30 +_METRICS_REQUEST_QUEUE_SIZE = 16 + def _parse_bind_addr(addr: str) -> tuple[str, int]: """Parse bind address forms: ':8080', '127.0.0.1:0', '8080', ':0'. @@ -57,6 +61,8 @@ def start_metrics_server( reg = registry if registry is not None else get_registry() class Handler(BaseHTTPRequestHandler): + timeout = _METRICS_REQUEST_TIMEOUT_S + def do_GET(self): # noqa: N802 if self.path.split("?", 1)[0] != "/metrics": self.send_error(404) @@ -68,12 +74,14 @@ def do_GET(self): # noqa: N802 self.end_headers() self.wfile.write(body) - def log_message(self, fmt: str, *args) -> None: + def log_message(self, format: str, *args) -> None: # noqa: A002 return try: host, port = _parse_bind_addr(addr) server = ThreadingHTTPServer((host, port), Handler) + server.request_queue_size = _METRICS_REQUEST_QUEUE_SIZE + server.timeout = _METRICS_REQUEST_TIMEOUT_S except (OSError, ValueError) as e: logger.warning( "Failed to start metrics server for bind address %r (%s); continuing without /metrics", diff --git a/python/packages/jumpstarter/jumpstarter/streams/common.py b/python/packages/jumpstarter/jumpstarter/streams/common.py index 03491ba1e..90a643106 100644 --- a/python/packages/jumpstarter/jumpstarter/streams/common.py +++ b/python/packages/jumpstarter/jumpstarter/streams/common.py @@ -1,6 +1,7 @@ import asyncio import logging from contextlib import asynccontextmanager, suppress +from functools import partial from anyio import ( BrokenResourceError, @@ -64,8 +65,6 @@ async def copy_stream( @asynccontextmanager async def forward_stream(a, b, *, metrics_driver_type: str | None = None): - from functools import partial - async with a, b: async with create_task_group() as tg: if metrics_driver_type is None: From 35674c9e1adf2c644c925cbfe108abf37e04af9d Mon Sep 17 00:00:00 2001 From: Roddie Kieley Date: Mon, 17 Aug 2026 07:21:11 -0230 Subject: [PATCH 12/16] fix: move driver error map below imports to satisfy E402. Co-authored-by: Cursor --- python/packages/jumpstarter/jumpstarter/driver/base.py | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/python/packages/jumpstarter/jumpstarter/driver/base.py b/python/packages/jumpstarter/jumpstarter/driver/base.py index bd12fda63..c0df00181 100644 --- a/python/packages/jumpstarter/jumpstarter/driver/base.py +++ b/python/packages/jumpstarter/jumpstarter/driver/base.py @@ -45,6 +45,11 @@ exporter_from_log_context, get_registry, ) +from jumpstarter.streams.aiohttp import AiohttpStreamReaderStream +from jumpstarter.streams.common import create_memory_stream +from jumpstarter.streams.encoding import Compression, compress_stream +from jumpstarter.streams.metadata import MetadataStream +from jumpstarter.streams.progress import ProgressStream # Ordered most-specific first: ConnectionError is an OSError subclass. _DRIVER_CALL_ERRORS: tuple[tuple[type[BaseException], str, StatusCode], ...] = ( @@ -54,11 +59,6 @@ (ConnectionError, "connection_error", StatusCode.UNAVAILABLE), (OSError, "device_error", StatusCode.INTERNAL), ) -from jumpstarter.streams.aiohttp import AiohttpStreamReaderStream -from jumpstarter.streams.common import create_memory_stream -from jumpstarter.streams.encoding import Compression, compress_stream -from jumpstarter.streams.metadata import MetadataStream -from jumpstarter.streams.progress import ProgressStream SUPPORTED_CONTENT_ENCODINGS = ( {} From 1bb49a2b0d00d50f0cb5a735c99415ca1be6fb0b Mon Sep 17 00:00:00 2001 From: Roddie Kieley Date: Wed, 19 Aug 2026 18:53:00 -0230 Subject: [PATCH 13/16] docs: record Phase 2 exporter metric cardinality risks. Co-authored-by: Cursor --- .../JEP-0013-observability-telemetry-logs.md | 16 ++++++++++++++++ .../jumpstarter-cli/jumpstarter_cli/run.py | 4 ++-- 2 files changed, 18 insertions(+), 2 deletions(-) diff --git a/docs/source/contributing/jeps/JEP-0013-observability-telemetry-logs.md b/docs/source/contributing/jeps/JEP-0013-observability-telemetry-logs.md index 5d5debbec..435936a40 100644 --- a/docs/source/contributing/jeps/JEP-0013-observability-telemetry-logs.md +++ b/docs/source/contributing/jeps/JEP-0013-observability-telemetry-logs.md @@ -1601,6 +1601,17 @@ all subsequent phases have E2E coverage from the start. - document tested pairs; W3C Trace Context in gRPC remains best-effort across Python and Go (no OTel SDK requirement to propagate `traceparent` where needed). +- **`operation` as a Prometheus label (Phase 2):** exporter series use + `@export` method names (`on`, `off`, `flash`, …) as the `operation` + label. The set is finite per process (loaded drivers), and unknown + methods are not recorded. If cardinality grows in the field, evaluate + moving `operation` from a series label to an exemplar key (see + *Cardinality guidelines*). +- **Per-chunk `jumpstarter_stream_bytes_total` increments (Phase 2):** + `copy_stream` calls `Counter.inc()` once per chunk. That is in-process + and uses independent tx/rx label sets, so Phase 2 accepts it. If flash + or storage throughput regresses, batch byte counts and flush + periodically instead of incrementing on every chunk. ## Rejected Alternatives @@ -1669,6 +1680,11 @@ all subsequent phases have E2E coverage from the start. - Event retention: Loki retention policy (per-tenant, per-stream retention classes) for annotated log events (**DD-2**); whether Jumpstarter should document recommended retention defaults or leave this to operators. +- Whether the exporter `operation` metric label should stay a bounded + series dimension or move to exemplars if cardinality grows (see + *Risks*). +- Whether `jumpstarter_stream_bytes_total` should batch per-chunk + increments if flash performance degrades (see *Risks*). ## Future Possibilities diff --git a/python/packages/jumpstarter-cli/jumpstarter_cli/run.py b/python/packages/jumpstarter-cli/jumpstarter_cli/run.py index 349455a91..ba8b8ffa7 100644 --- a/python/packages/jumpstarter-cli/jumpstarter_cli/run.py +++ b/python/packages/jumpstarter-cli/jumpstarter_cli/run.py @@ -11,6 +11,8 @@ from jumpstarter_cli_common.config import opt_config from jumpstarter_cli_common.exceptions import handle_exceptions +from jumpstarter.metrics import start_metrics_server + logger = logging.getLogger(__name__) @@ -107,8 +109,6 @@ async def signal_handler(): # Start signal handler immediately signal_tg.start_soon(signal_handler) - from jumpstarter.metrics import start_metrics_server - listen_addr, shutdown_metrics = start_metrics_server(metrics_bind_address) if listen_addr: logger.info("Serving metrics server at http://%s/metrics", listen_addr) From 8574c11786674c8861ed3c67d2d2e84a14395ca3 Mon Sep 17 00:00:00 2001 From: Roddie Kieley Date: Sun, 23 Aug 2026 15:57:50 -0230 Subject: [PATCH 14/16] fix: address raballew review on exporter metrics typing and cancel. Narrow _record_operation_metrics to OperationResult/ErrorType, set the metrics HTTP backlog before listen(), annotate ServicerContext on _handle_driver_exception, and cover unary client cancel so hang does not create operation or error series. Co-authored-by: Cursor --- .../jumpstarter/jumpstarter/driver/base.py | 16 +++++-- .../jumpstarter/metrics/metrics_test.py | 46 +++++++++++++++++++ .../jumpstarter/jumpstarter/metrics/server.py | 5 +- 3 files changed, 62 insertions(+), 5 deletions(-) diff --git a/python/packages/jumpstarter/jumpstarter/driver/base.py b/python/packages/jumpstarter/jumpstarter/driver/base.py index c0df00181..275bc0db8 100644 --- a/python/packages/jumpstarter/jumpstarter/driver/base.py +++ b/python/packages/jumpstarter/jumpstarter/driver/base.py @@ -41,6 +41,8 @@ from jumpstarter.config.env import JMP_DISABLE_COMPRESSION from jumpstarter.exporter.logging import get_logger from jumpstarter.metrics.registry import ( + ErrorType, + OperationResult, exemplars_from_log_context, exporter_from_log_context, get_registry, @@ -52,7 +54,7 @@ from jumpstarter.streams.progress import ProgressStream # Ordered most-specific first: ConnectionError is an OSError subclass. -_DRIVER_CALL_ERRORS: tuple[tuple[type[BaseException], str, StatusCode], ...] = ( +_DRIVER_CALL_ERRORS: tuple[tuple[type[BaseException], ErrorType, StatusCode], ...] = ( (NotImplementedError, "not_implemented", StatusCode.UNIMPLEMENTED), (ValueError, "validation_error", StatusCode.INVALID_ARGUMENT), (TimeoutError, "timeout", StatusCode.DEADLINE_EXCEEDED), @@ -133,9 +135,9 @@ def _record_operation_metrics( self, *, operation: str, - result: str, + result: OperationResult, duration_seconds: float, - error_type: str | None = None, + error_type: ErrorType | None = None, ) -> None: # Metrics must never discard a computed gRPC response or change the # abort status: keep recording failures isolated from the RPC path. @@ -161,7 +163,13 @@ def _record_operation_metrics( exc_info=True, ) - async def _handle_driver_exception(self, exc: BaseException, op: str, started: float, context) -> None: + async def _handle_driver_exception( + self, + exc: BaseException, + op: str, + started: float, + context: grpc.aio.ServicerContext, + ) -> None: for exc_type, error_type, status in _DRIVER_CALL_ERRORS: if isinstance(exc, exc_type): self._record_operation_metrics( diff --git a/python/packages/jumpstarter/jumpstarter/metrics/metrics_test.py b/python/packages/jumpstarter/jumpstarter/metrics/metrics_test.py index aad4e4769..a156ae7a6 100644 --- a/python/packages/jumpstarter/jumpstarter/metrics/metrics_test.py +++ b/python/packages/jumpstarter/jumpstarter/metrics/metrics_test.py @@ -405,3 +405,49 @@ def test_driver_call_increments_operations_and_active_sessions(): assert 'operation="on"' in body or 'operation="On"' in body else: assert success == 1.0 + + +class _SlowHangDriver(Driver): + driver_type = "testing" + + @classmethod + def client(cls): + return "jumpstarter.client.DriverClient" + + def __post_init__(self): + super().__post_init__() + import threading + + self._ready = threading.Event() + + @export + async def hang(self): + import anyio + + self._ready.set() + await anyio.sleep(30) + return "done" + + +def test_client_cancelled_driver_call_does_not_record_operation_metric(): + """Client-initiated cancel must not create operation or error series.""" + import concurrent.futures + import time + + from jumpstarter.common.utils import serve + + driver = _SlowHangDriver() + with serve(driver) as client: + fut = client.portal.start_task_soon(client.call_async, "hang") + assert driver._ready.wait(timeout=5), "driver hang() never started" + assert fut.cancel(), "expected in-flight DriverCall future to cancel" + try: + fut.result(timeout=5) + except (concurrent.futures.CancelledError, Exception): + pass + + # Allow any late server-side cleanup before asserting the registry. + time.sleep(0.1) + body = get_registry().generate_latest().decode() + assert 'operation="hang"' not in body + assert "jumpstarter_operation_errors_total{" not in body diff --git a/python/packages/jumpstarter/jumpstarter/metrics/server.py b/python/packages/jumpstarter/jumpstarter/metrics/server.py index 9b48ff707..8463f6ca7 100644 --- a/python/packages/jumpstarter/jumpstarter/metrics/server.py +++ b/python/packages/jumpstarter/jumpstarter/metrics/server.py @@ -79,9 +79,12 @@ def log_message(self, format: str, *args) -> None: # noqa: A002 try: host, port = _parse_bind_addr(addr) - server = ThreadingHTTPServer((host, port), Handler) + # bind_and_activate=False so request_queue_size is applied before listen(). + server = ThreadingHTTPServer((host, port), Handler, bind_and_activate=False) server.request_queue_size = _METRICS_REQUEST_QUEUE_SIZE server.timeout = _METRICS_REQUEST_TIMEOUT_S + server.server_bind() + server.server_activate() except (OSError, ValueError) as e: logger.warning( "Failed to start metrics server for bind address %r (%s); continuing without /metrics", From 697d81c46b325e316ef0aa5477e5b7e780f7f706 Mon Sep 17 00:00:00 2001 From: Roddie Kieley Date: Sun, 23 Aug 2026 16:27:31 -0230 Subject: [PATCH 15/16] fix: make exporter metrics bind fatal and always-on. Address mangelajo review items on PR #934: bind/parse failures now raise instead of soft-failing without /metrics, and jmp run always starts the local HTTP metrics server on ephemeral loopback (:0) with no public --metrics-bind-address option (Phase 3 will gate/reverse-scrape). Co-authored-by: Cursor --- .../jumpstarter-cli/jumpstarter_cli/run.py | 25 ++++--------- .../jumpstarter/metrics/metrics_test.py | 18 +++++----- .../jumpstarter/jumpstarter/metrics/server.py | 35 +++++++------------ 3 files changed, 27 insertions(+), 51 deletions(-) diff --git a/python/packages/jumpstarter-cli/jumpstarter_cli/run.py b/python/packages/jumpstarter-cli/jumpstarter_cli/run.py index ba8b8ffa7..2e3f705cb 100644 --- a/python/packages/jumpstarter-cli/jumpstarter_cli/run.py +++ b/python/packages/jumpstarter-cli/jumpstarter_cli/run.py @@ -15,6 +15,11 @@ logger = logging.getLogger(__name__) +# Phase 2 interim: always expose local HTTP /metrics on ephemeral loopback. +# Phase 3 replaces this with Telemetry reverse-scrape (unix/memory or in-process); +# bind address is intentionally not a user-facing CLI option. +_METRICS_BIND_ADDRESS = ":0" + def _parse_listener_bind(value: str) -> tuple[str, int]: """Parse '[host:]port' into (host, port). Default host is 0.0.0.0.""" @@ -79,7 +84,6 @@ def _handle_child( # noqa: C901 tls_cert=None, tls_key=None, passphrase=None, - metrics_bind_address=":0", ): """Handle child process with graceful shutdown.""" async def serve_with_graceful_shutdown(): # noqa: C901 @@ -109,9 +113,8 @@ async def signal_handler(): # Start signal handler immediately signal_tg.start_soon(signal_handler) - listen_addr, shutdown_metrics = start_metrics_server(metrics_bind_address) - if listen_addr: - logger.info("Serving metrics server at http://%s/metrics", listen_addr) + listen_addr, shutdown_metrics = start_metrics_server(_METRICS_BIND_ADDRESS) + logger.info("Serving metrics server at http://%s/metrics", listen_addr) try: if parsed_bind is not None: @@ -228,7 +231,6 @@ def _serve_with_exc_handling( tls_cert=None, tls_key=None, passphrase=None, - metrics_bind_address=":0", ): max_rapid_failures = config.failure_detection.max_rapid_failures rapid_failure_window = config.failure_detection.rapid_failure_window @@ -284,7 +286,6 @@ def _serve_with_exc_handling( tls_cert, tls_key, passphrase, - metrics_bind_address, ) sys.exit(1) # should never happen @@ -326,16 +327,6 @@ def _serve_with_exc_handling( default=False, help="Exit after the current lease ends instead of waiting for a new one.", ) -@click.option( - "--metrics-bind-address", - "metrics_bind_address", - default=":0", - show_default=True, - help=( - "Address for HTTP GET /metrics (Prometheus/OpenMetrics). " - "Default :0 binds an ephemeral loopback port. Use 0 to disable." - ), -) @handle_exceptions def run( config, @@ -345,7 +336,6 @@ def run( tls_key, passphrase, exit_on_lease_end, - metrics_bind_address, ): """Run an exporter locally.""" if listener_bind is not None and config is None: @@ -371,5 +361,4 @@ def run( tls_cert, tls_key, passphrase, - metrics_bind_address, ) diff --git a/python/packages/jumpstarter/jumpstarter/metrics/metrics_test.py b/python/packages/jumpstarter/jumpstarter/metrics/metrics_test.py index a156ae7a6..d5ccb783b 100644 --- a/python/packages/jumpstarter/jumpstarter/metrics/metrics_test.py +++ b/python/packages/jumpstarter/jumpstarter/metrics/metrics_test.py @@ -249,8 +249,8 @@ def test_metrics_server_shutdown_stops_listening(): urllib.request.urlopen(f"http://{listen}/metrics", timeout=1) -def test_metrics_server_bind_failure_is_non_fatal(): - """A taken fixed port must not abort the exporter process.""" +def test_metrics_server_bind_failure_is_fatal(): + """A taken fixed port must abort metrics startup.""" import socket holder = socket.socket(socket.AF_INET, socket.SOCK_STREAM) @@ -259,18 +259,16 @@ def test_metrics_server_bind_failure_is_non_fatal(): holder.listen(1) occupied_port = holder.getsockname()[1] try: - listen, shutdown = start_metrics_server(f"127.0.0.1:{occupied_port}") - assert listen == "" - assert shutdown is None + with pytest.raises(OSError): + start_metrics_server(f"127.0.0.1:{occupied_port}") finally: holder.close() -def test_metrics_server_invalid_bind_addr_is_non_fatal(): - """Non-numeric ports must not abort the exporter process.""" - listen, shutdown = start_metrics_server("127.0.0.1:not-a-port") - assert listen == "" - assert shutdown is None +def test_metrics_server_invalid_bind_addr_is_fatal(): + """Non-numeric ports must abort metrics startup.""" + with pytest.raises(ValueError): + start_metrics_server("127.0.0.1:not-a-port") class _TimeoutDriver(Driver): diff --git a/python/packages/jumpstarter/jumpstarter/metrics/server.py b/python/packages/jumpstarter/jumpstarter/metrics/server.py index 8463f6ca7..0acc09e3f 100644 --- a/python/packages/jumpstarter/jumpstarter/metrics/server.py +++ b/python/packages/jumpstarter/jumpstarter/metrics/server.py @@ -2,7 +2,6 @@ from __future__ import annotations -import logging from collections.abc import Callable from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from threading import Thread @@ -11,8 +10,6 @@ from .registry import MetricsRegistry, get_registry -logger = logging.getLogger(__name__) - ShutdownFunc = Callable[[], None] # Bound slow/idle scrape clients; backlog caps concurrent accept queue depth. @@ -42,16 +39,16 @@ def start_metrics_server( ) -> tuple[str, ShutdownFunc | None]: """Start an HTTP server exposing GET /metrics. - Returns ``(listen_address, shutdown)``. When metrics are disabled or bind - fails, returns ``(\"\", None)``. + Returns ``(listen_address, shutdown)``. When metrics are disabled + (addr \"0\" or empty), returns ``(\"\", None)``. - addr \"0\" or empty disables the server. addr ending with \":0\" binds an ephemeral port; the returned listen address is host:port suitable for urllib/http.Get. Bind failures (e.g. address already in use) and invalid bind addresses - (non-numeric port) are non-fatal: a warning is logged and (\"\", None) is - returned so the exporter can continue without metrics. + (non-numeric port) raise so a missing ``/metrics`` endpoint is fatal and + visible to administrators. Explicit disable via \"0\" / \"\" remains the only + quiet off path. Call ``shutdown()`` to stop the background server (no-op when None). """ @@ -77,21 +74,13 @@ def do_GET(self): # noqa: N802 def log_message(self, format: str, *args) -> None: # noqa: A002 return - try: - host, port = _parse_bind_addr(addr) - # bind_and_activate=False so request_queue_size is applied before listen(). - server = ThreadingHTTPServer((host, port), Handler, bind_and_activate=False) - server.request_queue_size = _METRICS_REQUEST_QUEUE_SIZE - server.timeout = _METRICS_REQUEST_TIMEOUT_S - server.server_bind() - server.server_activate() - except (OSError, ValueError) as e: - logger.warning( - "Failed to start metrics server for bind address %r (%s); continuing without /metrics", - addr, - e, - ) - return "", None + host, port = _parse_bind_addr(addr) + # bind_and_activate=False so request_queue_size is applied before listen(). + server = ThreadingHTTPServer((host, port), Handler, bind_and_activate=False) + server.request_queue_size = _METRICS_REQUEST_QUEUE_SIZE + server.timeout = _METRICS_REQUEST_TIMEOUT_S + server.server_bind() + server.server_activate() thread = Thread(target=server.serve_forever, name="jumpstarter-metrics", daemon=True) thread.start() From c7df5c9f38694bdc0a24d2b0417542f90ed9cb59 Mon Sep 17 00:00:00 2001 From: Roddie Kieley Date: Mon, 24 Aug 2026 15:01:34 -0230 Subject: [PATCH 16/16] test: assert copy_stream records stream-byte metrics. Cover the chunk-copy path so jumpstarter_stream_bytes_total is no longer only incremented via registry helpers, addressing the remaining #934 nit. Co-authored-by: Cursor --- .../jumpstarter/metrics/metrics_test.py | 67 +++++++++++++++++++ 1 file changed, 67 insertions(+) diff --git a/python/packages/jumpstarter/jumpstarter/metrics/metrics_test.py b/python/packages/jumpstarter/jumpstarter/metrics/metrics_test.py index d5ccb783b..9d6c0e5c1 100644 --- a/python/packages/jumpstarter/jumpstarter/metrics/metrics_test.py +++ b/python/packages/jumpstarter/jumpstarter/metrics/metrics_test.py @@ -449,3 +449,70 @@ def test_client_cancelled_driver_call_does_not_record_operation_metric(): body = get_registry().generate_latest().decode() assert 'operation="hang"' not in body assert "jumpstarter_operation_errors_total{" not in body + + +@pytest.mark.anyio +async def test_copy_stream_records_stream_bytes_from_chunks(): + """copy_stream with metrics_direction must observe chunk lengths, not registry helpers.""" + from anyio import create_memory_object_stream + + from jumpstarter.streams.common import copy_stream + + chunks = (b"abc", b"defg") + src_tx, src_rx = create_memory_object_stream[bytes](8) + dst_tx, dst_rx = create_memory_object_stream[bytes](8) + + structlog.contextvars.bind_contextvars( + client="ci-bot", + lease_id="lease-stream", + exporter="lab-01", + ) + try: + for chunk in chunks: + await src_tx.send(chunk) + await src_tx.aclose() + await copy_stream( + dst_tx, + src_rx, + metrics_direction="tx", + metrics_driver_type="serial", + ) + received = b"".join([await dst_rx.receive() for _ in chunks]) + await dst_tx.aclose() + await dst_rx.aclose() + finally: + structlog.contextvars.clear_contextvars() + + assert received == b"".join(chunks) + assert ( + _sample_value( + get_registry(), + "jumpstarter_stream_bytes_total", + { + "exporter": "lab-01", + "driver_type": "serial", + "direction": "tx", + }, + ) + == float(sum(len(c) for c in chunks)) + ) + + +@pytest.mark.anyio +async def test_copy_stream_without_metrics_direction_does_not_record(): + from anyio import create_memory_object_stream + + from jumpstarter.streams.common import copy_stream + + src_tx, src_rx = create_memory_object_stream[bytes](8) + dst_tx, dst_rx = create_memory_object_stream[bytes](8) + await src_tx.send(b"no-metrics") + await src_tx.aclose() + await copy_stream(dst_tx, src_rx) + assert await dst_rx.receive() == b"no-metrics" + await dst_tx.aclose() + await dst_rx.aclose() + + body = get_registry().generate_latest().decode() + assert "jumpstarter_stream_bytes_total{" not in body +