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 a9cd88511..2e3f705cb 100644 --- a/python/packages/jumpstarter-cli/jumpstarter_cli/run.py +++ b/python/packages/jumpstarter-cli/jumpstarter_cli/run.py @@ -11,8 +11,15 @@ 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__) +# 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.""" @@ -70,7 +77,14 @@ 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, +): """Handle child process with graceful shutdown.""" async def serve_with_graceful_shutdown(): # noqa: C901 received_signal = 0 @@ -99,46 +113,53 @@ async def signal_handler(): # Start signal handler immediately signal_tg.start_soon(signal_handler) - 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: - 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) + listen_addr, shutdown_metrics = start_metrics_server(_METRICS_BIND_ADDRESS) + logger.info("Serving metrics server at http://%s/metrics", listen_addr) - # Check if exporter set an exit code (e.g., from hook failure with on_failure='exit') - exporter_exit_code = exporter.exit_code + 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: + 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() @@ -204,7 +225,12 @@ 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, ): max_rapid_failures = config.failure_detection.max_rapid_failures rapid_failure_window = config.failure_detection.rapid_failure_window @@ -253,7 +279,14 @@ 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, + ) sys.exit(1) # should never happen @@ -295,7 +328,15 @@ def _serve_with_exc_handling( help="Exit after the current lease ends instead of waiting for a new one.", ) @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, +): """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 +354,11 @@ 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, + ) diff --git a/python/packages/jumpstarter/jumpstarter/driver/base.py b/python/packages/jumpstarter/jumpstarter/driver/base.py index fec600923..275bc0db8 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 @@ -16,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 @@ -38,12 +40,28 @@ ) 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, +) 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], ErrorType, 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), +) + SUPPORTED_CONTENT_ENCODINGS = ( {} if os.environ.get(JMP_DISABLE_COMPRESSION) == "1" @@ -113,11 +131,87 @@ def client(cls) -> str: def extra_labels(self) -> dict[str, str]: return {} + def _record_operation_metrics( + self, + *, + operation: str, + result: OperationResult, + duration_seconds: float, + 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. + 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 _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( + 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: """ op = request.method + started = time.perf_counter() self.logger.info( "Operation started", extra={"operation": op, "driver_type": self.driver_type}, @@ -132,62 +226,35 @@ async def DriverCall(self, request, context): else: result = await to_thread.run_sync(method, *args) - self.logger.info( - "Operation completed", - extra={"operation": op, "driver_type": self.driver_type, "result": "success"}, - ) - return jumpstarter_pb2.DriverCallResponse( + # 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), ) - except NotImplementedError as e: - 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.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.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.logger.warning( - "Operation failed", - extra={"operation": op, "driver_type": self.driver_type, - "result": "failure", "error_type": "connection_error"}, + self._record_operation_metrics( + operation=op, + result="success", + duration_seconds=time.perf_counter() - started, ) - await context.abort(StatusCode.UNAVAILABLE, str(e)) - except OSError as e: - self.logger.warning( - "Operation failed", - extra={"operation": op, "driver_type": self.driver_type, - "result": "failure", "error_type": "device_error"}, + self.logger.info( + "Operation completed", + extra={"operation": op, "driver_type": self.driver_type, "result": "success"}, ) - await context.abort(StatusCode.INTERNAL, str(e)) + return response + 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.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): """ :meta private: """ op = request.method + started = time.perf_counter() self.logger.info( "Operation started", extra={"operation": op, "driver_type": self.driver_type}, @@ -209,52 +276,21 @@ 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.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.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.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.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.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.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 f19add60e..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 @@ -59,9 +61,20 @@ class Session( def __contextmanager__(self) -> Generator[Self]: logging.getLogger().addHandler(self._logging_handler) self.root_device.reset() + set_log_context(exporter=self.name) + get_registry().adjust_active_sessions(exporter=self.name, delta=1.0) try: yield self finally: + try: + get_registry().adjust_active_sessions(exporter=self.name, delta=-1.0) + except Exception: + logger.warning( + "Failed to decrement active sessions metric for exporter %s", + self.name, + exc_info=True, + ) + unbind_log_context("exporter") try: self.root_device.close() except Exception as e: @@ -319,14 +332,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/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 diff --git a/python/packages/jumpstarter/jumpstarter/metrics/__init__.py b/python/packages/jumpstarter/jumpstarter/metrics/__init__.py new file mode 100644 index 000000000..56041b452 --- /dev/null +++ b/python/packages/jumpstarter/jumpstarter/metrics/__init__.py @@ -0,0 +1,15 @@ +"""Exporter-local Prometheus metrics.""" + +from .registry import ( + DEFAULT_EXEMPLAR_KEYS, + MetricsRegistry, + get_registry, +) +from .server import start_metrics_server + +__all__ = [ + "DEFAULT_EXEMPLAR_KEYS", + "MetricsRegistry", + "get_registry", + "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 new file mode 100644 index 000000000..9d6c0e5c1 --- /dev/null +++ b/python/packages/jumpstarter/jumpstarter/metrics/metrics_test.py @@ -0,0 +1,518 @@ +"""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, + 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, + filter_exemplars, +) +from jumpstarter.metrics.server import _parse_bind_addr + +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_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: + 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(":0") == ("127.0.0.1", 0) + 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"} + 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 '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 + + 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(): + 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() + 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(): + reg = get_registry() + reg.set_active_sessions(exporter="lab-01", value=2) + listen, shutdown = start_metrics_server("127.0.0.1:0", registry=reg) + assert listen, "expected non-empty listen address" + 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(): + 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, 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_fatal(): + """A taken fixed port must abort metrics startup.""" + 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: + 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_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): + 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 + + +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_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 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 + + 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 + + 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 + + +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 + + +@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 + diff --git a/python/packages/jumpstarter/jumpstarter/metrics/registry.py b/python/packages/jumpstarter/jumpstarter/metrics/registry.py new file mode 100644 index 000000000..e762796bf --- /dev/null +++ b/python/packages/jumpstarter/jumpstarter/metrics/registry.py @@ -0,0 +1,152 @@ +"""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", +] + + +def filter_exemplars(exemplars: dict[str, str] | None) -> dict[str, str] | None: + """Keep only the 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 exporter metric 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: OperationResult, + driver_type: str, + duration_seconds: float, + exemplars: dict[str, str] | None = None, + error_type: ErrorType | 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: 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( + 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 adjust_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) + + +# 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: + 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..0acc09e3f --- /dev/null +++ b/python/packages/jumpstarter/jumpstarter/metrics/server.py @@ -0,0 +1,96 @@ +"""HTTP GET /metrics server for exporter-local Prometheus scrape.""" + +from __future__ import annotations + +from collections.abc import Callable +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 + +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'. + + 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 = "127.0.0.1" + else: + host = "127.0.0.1" + port_s = addr + return host, int(port_s) + + +def start_metrics_server( + addr: str, registry: MetricsRegistry | None = None +) -> tuple[str, ShutdownFunc | None]: + """Start an HTTP server exposing GET /metrics. + + Returns ``(listen_address, shutdown)``. When metrics are disabled + (addr \"0\" or empty), returns ``(\"\", None)``. + + 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) 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). + """ + if addr == "" or addr == "0": + return "", None + + 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) + 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, format: str, *args) -> None: # noqa: A002 + return + + 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() + + 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}", shutdown diff --git a/python/packages/jumpstarter/jumpstarter/streams/common.py b/python/packages/jumpstarter/jumpstarter/streams/common.py index 40e650ca5..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, @@ -11,12 +12,38 @@ 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__) -async def copy_stream(dst: AnyByteStream, src: AnyByteStream): +async def copy_stream( + dst: AnyByteStream, + src: AnyByteStream, + *, + metrics_direction: StreamDirection | None = None, + 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: + 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=metrics_exemplars, + ) await dst.send(v) with suppress( AttributeError, @@ -37,11 +64,31 @@ 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): 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"