From 89673426f723d56a2714ffd69faecc0c19a272fb Mon Sep 17 00:00:00 2001 From: Mike Langmayr Date: Wed, 28 Oct 2026 14:07:01 -0700 Subject: [PATCH 1/3] Add optional OpenTelemetry tracing and request metrics --- README.md | 3 +- docs/source/api/index.md | 8 + docs/source/index.md | 3 + docs/source/installation.md | 6 + docs/source/telemetry.md | 57 ++++++ libby/cli/libby_cli.py | 19 +- libby/daemon.py | 16 +- libby/libby.py | 31 ++-- libby/telemetry.py | 345 ++++++++++++++++++++++++++++++++++++ pyproject.toml | 5 + tests/test_telemetry.py | 297 +++++++++++++++++++++++++++++++ tox.ini | 4 +- 12 files changed, 778 insertions(+), 16 deletions(-) create mode 100644 docs/source/telemetry.md create mode 100644 libby/telemetry.py create mode 100644 tests/test_telemetry.py diff --git a/README.md b/README.md index eb32903..932ad2e 100644 --- a/README.md +++ b/README.md @@ -5,7 +5,8 @@ Libby: a tiny messaging library which uses Bamboo with pluggable transports (ZMQ ## Documentation Full docs (installation, keywords, the `Client` library, the `libby` CLI, how -to build a `LibbyDaemon` peer, the keygrabber, plus the generated API +to build a `LibbyDaemon` peer, the keygrabber, optional OpenTelemetry export, +plus the generated API reference) are built with Sphinx + the Shibuya theme and published to GitHub Pages: diff --git a/docs/source/api/index.md b/docs/source/api/index.md index 55bbb98..7503ceb 100644 --- a/docs/source/api/index.md +++ b/docs/source/api/index.md @@ -97,6 +97,14 @@ the docs build does not install the optional `influxdb` extra. :show-inheritance: ``` +## Telemetry + +```{eval-rst} +.. automodule:: libby.telemetry + :members: + :show-inheritance: +``` + ## Responses and errors ```{eval-rst} diff --git a/docs/source/index.md b/docs/source/index.md index 1d0f0fd..85bd5d4 100644 --- a/docs/source/index.md +++ b/docs/source/index.md @@ -13,6 +13,7 @@ with pluggable transports (ZMQ or RabbitMQ). It gives you: `wait_for` (libby's `ktl.waitFor`). - **Keygrabber**: a daemon that polls keywords from other peers and writes them to a time-series database for dashboarding. +- **Telemetry**: optional OpenTelemetry traces and metrics for every request. - **`libby` CLI** — a command-line front end for keyword peers (`show` / `modify` / `list` / `describe` / `waitfor`), with TAB completion and wildcard listing to find which peers are up. @@ -27,6 +28,7 @@ client cli daemon keygrabber +telemetry api/index ``` @@ -37,4 +39,5 @@ api/index - Writing a script or tool that talks to peers? Read {doc}`client`. - Poking at peers interactively? Read {doc}`cli`. - Recording keywords for Grafana? Read {doc}`keygrabber`. +- Tracing requests across peers? Read {doc}`telemetry`. - Looking for a specific class or function? See {doc}`api/index`. diff --git a/docs/source/installation.md b/docs/source/installation.md index 14ac3ef..f4126af 100644 --- a/docs/source/installation.md +++ b/docs/source/installation.md @@ -41,6 +41,12 @@ To also pull in the docs toolchain (Sphinx + the Shibuya theme): pip install -e ".[docs]" ``` +For OpenTelemetry export (see {doc}`telemetry`): + +```bash +pip install -e ".[otel]" +``` + ## Testing ```bash diff --git a/docs/source/telemetry.md b/docs/source/telemetry.md new file mode 100644 index 0000000..aabe3c9 --- /dev/null +++ b/docs/source/telemetry.md @@ -0,0 +1,57 @@ +# Telemetry + +libby can export traces and metrics over OTLP to an +[OpenTelemetry](https://opentelemetry.io/) Collector. It is optional and off +by default: without the `otel` extra libby never imports OpenTelemetry, and +with it nothing is exported until an OTLP endpoint is set. + +## Turning it on + +```bash +pip install "libby[otel]" +export OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:4318 +``` + +Every `LibbyDaemon` then exports from `start()` until `stop()`, and the `libby` +CLI exports each command's trace. The CLI exports no metrics, since every run +would start new series. Exports use OTLP over HTTP and honour the standard +`OTEL_*` variables, e.g. `OTEL_TRACES_SAMPLER` or `OTEL_SDK_DISABLED=true`. + +| Process | `service.name` | `service.namespace` | +|---|---|---| +| `LibbyDaemon` | wire id, e.g. `hsfei.adc` | `group_id` | +| `libby` CLI | `libby-cli` | | + +An application that installs its own OpenTelemetry SDK can call +`libby.telemetry.use_current_providers()` to have libby record into it. + +## What is recorded + +**Traces.** Each request is a client span on the sender and a server span on +the handling peer, joined into one trace through the W3C `traceparent` that +bamboo carries in the envelope. A publish is a producer span, and each +subscriber's handling is a consumer span under it. A CLI command is one span +over all the requests it makes. Span names are the key or keyword, e.g. +`positionvalue`. + +**Metrics.** Two histograms in seconds, which give request rate, errors and +duration per key: + +| Metric | Attributes | +|---|---| +| `rpc.client.call.duration` | `rpc.method`, `server.address`, `error.type` | +| `rpc.server.call.duration` | `rpc.method`, `error.type` | + +`error.type` is set only on failures: the bamboo status (`timeout`, +`no_key`, ...) for a request that got no answer, `error_response` for an +answer with `ok: false`, or the exception type. + +**Logs.** libby does not export logs. A line logged while a span is active +gets ` trace_id=... span_id=...` appended, in the daemon's configured log +handler and the CLI's log file, so a Collector reading journald or the files +can link it to its trace. + +## Cost + +Off, each hook costs about 0.1 µs. On, a traced request adds roughly 15 µs on +the sender and 12 µs on the handler. diff --git a/libby/cli/libby_cli.py b/libby/cli/libby_cli.py index de2d792..6c95e0c 100644 --- a/libby/cli/libby_cli.py +++ b/libby/cli/libby_cli.py @@ -17,6 +17,7 @@ import argcomplete from argcomplete.shell_integration import shellcode +from libby import telemetry from libby.cli.completion import ( CompletionCache, Listings, @@ -42,6 +43,9 @@ from libby.response import unwrap DEFAULT_SELF_ID = "cli" +CLI_SERVICE_NAME = "libby-cli" +# Bound the exit flush, so a down collector cannot stall every command +CLI_EXPORT_TIMEOUT_S = 1.0 DEFAULT_TIMEOUT_S = 3.0 DEFAULT_LOG_FILE = Path.home() / ".libby" / "cli.log" @@ -115,7 +119,7 @@ def _setup_logging(namespace: argparse.Namespace, config: Dict[str, Any]) -> Non ) except OSError: file_handler = logging.NullHandler() - file_handler.setFormatter(logging.Formatter( + file_handler.setFormatter(telemetry.TraceContextFormatter( "%(asctime)s %(levelname)s %(name)s: %(message)s" )) setattr(file_handler, "_libby_cli_handler", True) @@ -791,9 +795,20 @@ def main(argv: Optional[List[str]] = None) -> int: argv_display = shlex.join(_redact_argv(argv if argv is not None else sys.argv[1:])) logger.info("invoke cmd=%s argv=%s", namespace.cmd, argv_display) + handle = telemetry.configure(CLI_SERVICE_NAME, export_timeout_s=CLI_EXPORT_TIMEOUT_S, + export_metrics=False) + try: + return _run_command(namespace) + finally: + if handle is not None: + handle.shutdown() + + +def _run_command(namespace: argparse.Namespace) -> int: start = time.monotonic() try: - rc = namespace.func(namespace) + with telemetry.operation(f"libby {namespace.cmd}"): + rc = namespace.func(namespace) except LibbyError as ex: # Core raises LibbyError; the CLI is the boundary that exits logger.warning("cmd=%s failed: %s", namespace.cmd, ex) diff --git a/libby/daemon.py b/libby/daemon.py index 4ab9e2b..c29f291 100644 --- a/libby/daemon.py +++ b/libby/daemon.py @@ -9,9 +9,11 @@ from dataclasses import asdict, is_dataclass from typing import Any, Callable, Dict, Iterable, List, Mapping, Optional, Type, TypeVar +from . import telemetry from .config import ConfigError, DaemonConfigLoader, with_env_overrides from .keyword import Keyword from .libby import Libby +from .naming import qualified_peer_id Payload = Dict[str, Any] @@ -102,6 +104,7 @@ def __init__(self) -> None: self.libby: Optional[Libby] = None self._last_error: Optional[str] = None self._start_time: Optional[float] = None + self._telemetry: Optional[telemetry.Telemetry] = None self.logger = logging.getLogger(type(self).__name__) self._ensure_last_error_handler() @@ -222,7 +225,7 @@ def _setup_logging(self) -> None: handler.setLevel(level) handler.setFormatter( - logging.Formatter( + telemetry.TraceContextFormatter( str( raw.get( "format", @@ -392,6 +395,10 @@ def start(self) -> None: self._stop_event.clear() self._start_time = time.monotonic() try: + self._telemetry = telemetry.configure( + qualified_peer_id(self.config_peer_id(), self.config_group_id()), + self.config_group_id(), + ) self.libby = self.build_libby() self._register_services(self.services) self._register_topics(self.topics) @@ -437,6 +444,7 @@ def start(self) -> None: ) except Exception: self._started = False + self._close_telemetry() raise def request_stop(self) -> None: @@ -459,6 +467,7 @@ def stop(self) -> None: self._started = False self._stop_event.set() self.logger.info("stopped") + self._close_telemetry() def serve(self) -> None: """Start the daemon and block until a signal or stop request.""" @@ -558,6 +567,11 @@ def _close_libby(self) -> None: except Exception: self.logger.exception("Libby transport shutdown failed") + def _close_telemetry(self) -> None: + handle, self._telemetry = self._telemetry, None + if handle is not None: + handle.shutdown() + def payload(self, value: Any = None, /, **extra: Any) -> dict: """Normalize a user result into a JSON-serializable dictionary.""" diff --git a/libby/libby.py b/libby/libby.py index 946ad10..cf2a7e3 100644 --- a/libby/libby.py +++ b/libby/libby.py @@ -10,6 +10,7 @@ from bamboo.protocol import Protocol from bamboo.discovery import Discovery +from . import telemetry from .keyword import Keyword, match_pattern from .keyword_registry import KeywordRegistry from .naming import qualified_peer_id @@ -48,8 +49,7 @@ def __init__( # bamboo owns the semantics if keys: - for k in keys: - self.proto.serve(k, callback if callback else (lambda *_: None)) + self.serve_keys(keys, callback if callback else (lambda *_: None)) # Periodic discovery (provided by bamboo) self._disco: Optional[Discovery] = None @@ -186,7 +186,13 @@ def __exit__(self, exc_type, exc, tb): self.stop() # thin passthroughs to bamboo.Protocol def request(self, peer_id: str, key: str, payload: Dict[str, Any], ttl_ms: int = 8000): - return self.proto.request_peer(peer_id, key, payload, timeout_s=ttl_ms / 1000.0) + with telemetry.client_call(key, peer_id) as call: + result = self.proto.request_peer( + peer_id, key, payload, timeout_s=ttl_ms / 1000.0, + traceparent=call.traceparent, tracestate=call.tracestate, + ) + call.check_request(result) + return result def rpc(self, peer_id: str, key: str, payload: Dict[str, Any], ttl_ms: int = 8000): return self.request(peer_id, key, payload, ttl_ms) @@ -204,10 +210,12 @@ def broadcast_request( address book. Our own reply is dropped: on RabbitMQ our queue sits on the fanout exchange like everyone else's. """ - msg = MessageBuilder(sourceid=self.self_id).req(key, payload).to(None).build() - with self._reply_queue(msg.env.transid) as replies: - self.proto.send(msg) - return list(self._drain_replies(replies, timeout_s)) + with telemetry.client_call(key, None) as call: + msg = (MessageBuilder(sourceid=self.self_id).req(key, payload).to(None) + .trace(call.traceparent, call.tracestate).build()) + with self._reply_queue(msg.env.transid) as replies: + self.proto.send(msg) + return list(self._drain_replies(replies, timeout_s)) @contextmanager def _reply_queue(self, transid: str) -> Iterator["queue.Queue[Any]"]: @@ -234,8 +242,9 @@ def _drain_replies(self, replies: "queue.Queue[Any]", yield BroadcastReply(resp.env.sourceid, dict(resp.env.payload or {})) def serve_keys(self, keys: List[str], callback: Callable[[dict, dict], Optional[dict]]) -> None: + traced = telemetry.traced_handler(callback) for k in keys: - self.proto.serve(k, callback) + self.proto.serve(k, traced) def register_keyword(self, keyword: Keyword) -> None: if keyword.name in self._keywords: @@ -333,14 +342,16 @@ def _keys_read(self, payload: dict, _ctx: dict) -> dict: return {"ok": True, "values": values} def listen(self, topic: str, handler: Callable[[Any], None]) -> None: - self.proto.listen(topic, handler) + self.proto.listen(topic, telemetry.traced_listener(topic, handler)) def listen_many(self, handlers: Dict[str, Callable[[Any], None]]) -> None: for topic, h in handlers.items(): self.listen(topic, h) def publish(self, topic: str, payload: Dict[str, Any]) -> int: - return self.proto.publish(topic, payload) + with telemetry.publish_call(topic) as call: + return self.proto.publish(topic, payload, traceparent=call.traceparent, + tracestate=call.tracestate) def emit(self, topic: str, payload: Dict[str, Any]) -> int: return self.publish(topic, payload) diff --git a/libby/telemetry.py b/libby/telemetry.py new file mode 100644 index 0000000..3cb391a --- /dev/null +++ b/libby/telemetry.py @@ -0,0 +1,345 @@ +"""Optional OpenTelemetry traces and metrics for libby peers. + +Everything here is a no-op until :func:`configure` succeeds, which needs the +``libby[otel]`` extra and an OTLP endpoint set through the standard ``OTEL_*`` +environment variables. Without both, libby never imports OpenTelemetry. +""" +from __future__ import annotations + +import logging +import os +import socket +import time +from contextlib import ExitStack +from typing import Any, Callable, Dict, Optional + +logger = logging.getLogger(__name__) + +RPC_SYSTEM = "libby" +ERROR_RESPONSE = "error_response" +DURATION_BUCKETS_S = ( + 0.0001, 0.00025, 0.0005, 0.001, 0.0025, 0.005, 0.01, + 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0, +) +_ENDPOINT_VARIABLES = ( + "OTEL_EXPORTER_OTLP_ENDPOINT", + "OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", + "OTEL_EXPORTER_OTLP_METRICS_ENDPOINT", +) + +Handler = Callable[[dict, dict], Optional[dict]] +Listener = Callable[[Any], None] + + +class _Instruments: + """Tracer, histograms and OTel API modules, bound once telemetry is on.""" + + def __init__(self) -> None: + # Deferred: OpenTelemetry is the optional libby[otel] extra + # pylint: disable=import-outside-toplevel + from opentelemetry import metrics, propagate, trace + + self.trace = trace + self.propagate = propagate + self.tracer = trace.get_tracer("libby") + meter = metrics.get_meter("libby") + self.client_duration = meter.create_histogram( + "rpc.client.call.duration", unit="s", + description="Duration of libby requests this peer sent", + explicit_bucket_boundaries_advisory=DURATION_BUCKETS_S, + ) + self.server_duration = meter.create_histogram( + "rpc.server.call.duration", unit="s", + description="Duration of libby requests this peer handled", + explicit_bucket_boundaries_advisory=DURATION_BUCKETS_S, + ) + + def incoming(self, traceparent: Optional[str], tracestate: Optional[str]) -> Any: + """Return the context a received envelope's trace fields describe.""" + if not traceparent: + return None + carrier = {"traceparent": traceparent} + if tracestate: + carrier["tracestate"] = tracestate + return self.propagate.extract(carrier) + + def finish(self, span: Any, error_type: Optional[str], attributes: Dict[str, Any], + histogram: Any, started: float) -> None: + """Mark a failed span and record the call's duration, if it is timed.""" + if error_type is not None: + span.set_status(self.trace.StatusCode.ERROR) + span.set_attribute("error.type", error_type) + attributes = {**attributes, "error.type": error_type} + if histogram is not None: + histogram.record(time.perf_counter() - started, attributes) + + +# OTel providers are process-global, so the switch for using them is too +_instruments: Optional[_Instruments] = None # pylint: disable=invalid-name + + +class Telemetry: # pylint: disable=too-few-public-methods + """SDK providers installed by :func:`configure`; shut down to flush them.""" + + def __init__(self, tracer_provider: Any, meter_provider: Optional[Any]) -> None: + self._tracer_provider = tracer_provider + self._meter_provider = meter_provider + + def shutdown(self) -> None: + """Stop recording, export what is pending and stop the exporters.""" + disable() + self._tracer_provider.shutdown() + if self._meter_provider is not None: + self._meter_provider.shutdown() + + +def export_requested() -> bool: + """Return whether the environment names an OTLP endpoint to export to.""" + if os.environ.get("OTEL_SDK_DISABLED", "").strip().lower() == "true": + return False + return any(os.environ.get(name) for name in _ENDPOINT_VARIABLES) + + +def enabled() -> bool: + """Return whether libby is recording spans and metrics.""" + return _instruments is not None + + +def configure(service_name: str, service_namespace: Optional[str] = None, *, + export_timeout_s: Optional[float] = None, + export_metrics: bool = True) -> Optional[Telemetry]: + """Turn on tracing and metrics for this process if export is requested. + + Install the OTLP SDK unless the application already installed its own + providers, in which case libby records into those. Return a + :class:`Telemetry` only to the caller that installed the SDK, since it + owns the shutdown. Short-lived processes pass ``export_metrics=False``: + each run is a new ``service.instance.id``, so its metrics are new series. + """ + if _instruments is not None or not export_requested(): + return None + try: + # Deferred: OpenTelemetry is the optional libby[otel] extra + # pylint: disable=import-outside-toplevel + from opentelemetry import trace + handle = None + if isinstance(trace.get_tracer_provider(), trace.ProxyTracerProvider): + handle = _install_sdk(service_name, service_namespace, export_timeout_s, + export_metrics) + except ImportError: + logger.warning("an OTLP endpoint is set but libby[otel] is not installed; " + "telemetry stays off") + return None + use_current_providers() + return handle + + +def _install_sdk(service_name: str, service_namespace: Optional[str], + export_timeout_s: Optional[float], export_metrics: bool) -> Telemetry: + # pylint: disable=import-outside-toplevel + from opentelemetry import trace + from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter + from opentelemetry.sdk.trace import TracerProvider + from opentelemetry.sdk.trace.export import BatchSpanProcessor + + resource = _resource(service_name, service_namespace) + timeout = {} if export_timeout_s is None else {"timeout": export_timeout_s} + + tracer_provider = TracerProvider(resource=resource) + tracer_provider.add_span_processor(BatchSpanProcessor(OTLPSpanExporter(**timeout))) + trace.set_tracer_provider(tracer_provider) + meter_provider = _install_meter(resource, timeout) if export_metrics else None + return Telemetry(tracer_provider, meter_provider) + + +def _install_meter(resource: Any, timeout: Dict[str, float]) -> Any: + # pylint: disable=import-outside-toplevel + from opentelemetry import metrics + from opentelemetry.exporter.otlp.proto.http.metric_exporter import OTLPMetricExporter + from opentelemetry.sdk.metrics import MeterProvider + from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader + + meter_provider = MeterProvider( + resource=resource, + metric_readers=[PeriodicExportingMetricReader(OTLPMetricExporter(**timeout))], + ) + metrics.set_meter_provider(meter_provider) + return meter_provider + + +def _resource(service_name: str, service_namespace: Optional[str]) -> Any: + from opentelemetry.sdk.resources import Resource # pylint: disable=import-outside-toplevel + + attributes = {"service.name": service_name, "host.name": socket.gethostname()} + if service_namespace: + attributes["service.namespace"] = service_namespace + return Resource.create(attributes) + + +def use_current_providers() -> None: + """Record into the providers already set on the OTel API, e.g. an app's own SDK.""" + global _instruments # pylint: disable=global-statement + _instruments = _Instruments() + + +def disable() -> None: + """Stop recording; spans and metrics become no-ops again.""" + global _instruments # pylint: disable=global-statement + _instruments = None + + +class Call: + """Trace context for one outgoing message, and the call's outcome.""" + + __slots__ = ("traceparent", "tracestate", "error_type", "_span", "_stack", "_finish") + + def __init__(self) -> None: + self.traceparent: Optional[str] = None + self.tracestate: Optional[str] = None + self.error_type: Optional[str] = None + self._span: Any = None + self._stack: Optional[ExitStack] = None + self._finish: Callable[[Any, Optional[str]], None] = lambda *_: None + + @classmethod + def start(cls, instruments: _Instruments, name: str, kind: Any, + attributes: Dict[str, Any], histogram: Any = None) -> "Call": + """Open a current span named ``name`` and capture its trace context.""" + call = cls() + started = time.perf_counter() + call._stack = ExitStack() + call._span = call._stack.enter_context(instruments.tracer.start_as_current_span( + name, kind=kind, attributes=attributes, + record_exception=False, set_status_on_exception=False, + )) + call._finish = lambda span, error_type: instruments.finish( + span, error_type, attributes, histogram, started) + carrier: Dict[str, str] = {} + instruments.propagate.inject(carrier) + call.traceparent = carrier.get("traceparent") + call.tracestate = carrier.get("tracestate") + return call + + def __enter__(self) -> "Call": + return self + + def __exit__(self, exc_type: Any, exc: Any, _tb: Any) -> bool: + if self._stack is None: + return False + if exc is not None: + self.error_type = exc_type.__qualname__ + self._span.record_exception(exc) + self._finish(self._span, self.error_type) + self._stack.close() + return False + + def check_request(self, result: Dict[str, Any]) -> None: + """Classify a bamboo ``request_peer`` result as success or error.""" + if self._stack is None: + return + status = result.get("status") + if status != "delivered": + self.error_type = str(status) + elif not (result.get("resp") or {}).get("ok", True): + self.error_type = ERROR_RESPONSE + + +_UNTRACED = Call() + + +def client_call(key: str, peer_id: Optional[str]) -> Call: + """Trace and time a request; ``peer_id`` ``None`` means a broadcast.""" + instruments = _instruments + if instruments is None: + return _UNTRACED + attributes: Dict[str, Any] = {"rpc.system": RPC_SYSTEM, "rpc.method": key} + if peer_id is not None: + attributes["server.address"] = peer_id + return Call.start(instruments, key, instruments.trace.SpanKind.CLIENT, + attributes, instruments.client_duration) + + +def publish_call(topic: str) -> Call: + """Trace a publish so subscribers can link their handling to it.""" + instruments = _instruments + if instruments is None: + return _UNTRACED + return Call.start( + instruments, f"publish {topic}", instruments.trace.SpanKind.PRODUCER, + {"messaging.system": RPC_SYSTEM, "messaging.destination.name": topic}, + ) + + +def operation(name: str) -> Call: + """Trace a unit of work spanning several calls, e.g. one CLI command.""" + instruments = _instruments + if instruments is None: + return _UNTRACED + return Call.start(instruments, name, instruments.trace.SpanKind.INTERNAL, {}) + + +def traced_handler(callback: Handler) -> Handler: + """Wrap a ``serve`` callback in a server span and a duration metric.""" + + def handle(payload: dict, ctx: dict) -> Optional[dict]: + instruments = _instruments + if instruments is None: + return callback(payload, ctx) + key = ctx.get("key", "") + attributes: Dict[str, Any] = {"rpc.system": RPC_SYSTEM, "rpc.method": key} + started = time.perf_counter() + error_type: Optional[str] = None + with instruments.tracer.start_as_current_span( + key, context=instruments.incoming(ctx.get("traceparent"), ctx.get("tracestate")), + kind=instruments.trace.SpanKind.SERVER, + # client.address stays off the metric: CLI peer ids are unbounded + attributes={**attributes, "client.address": ctx.get("source", "")}, + ) as span: + try: + result = callback(payload, ctx) + if isinstance(result, dict) and not result.get("ok", True): + error_type = ERROR_RESPONSE + return result + except Exception as ex: + error_type = type(ex).__qualname__ + raise + finally: + instruments.finish(span, error_type, attributes, + instruments.server_duration, started) + + return handle + + +def traced_listener(topic: str, handler: Listener) -> Listener: + """Wrap a topic handler in a consumer span linked to the publisher's.""" + + def handle(msg: Any) -> None: + instruments = _instruments + if instruments is None: + handler(msg) + return + env = msg.env + with instruments.tracer.start_as_current_span( + f"process {topic}", + context=instruments.incoming(env.traceparent, env.tracestate), + kind=instruments.trace.SpanKind.CONSUMER, + attributes={"messaging.system": RPC_SYSTEM, "messaging.destination.name": topic}, + ): + handler(msg) + + return handle + + +class TraceContextFormatter(logging.Formatter): + """Append the active trace and span ids, so a log line links to its trace.""" + + def formatMessage(self, record: logging.LogRecord) -> str: + line = super().formatMessage(record) + instruments = _instruments + if instruments is None: + return line + span_context = instruments.trace.get_current_span().get_span_context() + if not span_context.is_valid: + return line + return (f"{line} trace_id={span_context.trace_id:032x}" + f" span_id={span_context.span_id:016x}") diff --git a/pyproject.toml b/pyproject.toml index 17cc390..bd9d07a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -26,6 +26,11 @@ dependencies = [ influxdb = [ "influxdb-client>=1.40", ] +otel = [ + "opentelemetry-api>=1.27", + "opentelemetry-sdk>=1.27", + "opentelemetry-exporter-otlp-proto-http>=1.27", +] docs = [ "sphinx>=7.2", "shibuya>=2024.1.21", diff --git a/tests/test_telemetry.py b/tests/test_telemetry.py new file mode 100644 index 0000000..51c4bc0 --- /dev/null +++ b/tests/test_telemetry.py @@ -0,0 +1,297 @@ +"""Tests for libby's optional OpenTelemetry instrumentation. + +The disabled cases need nothing beyond libby. The traced cases need the +``otel`` extra (tox installs it) and run per transport, recording into +in-memory exporters installed once for the whole module, since the OTel API +allows one global provider per process. +""" +from __future__ import annotations + +import importlib.util +import logging +import os +import threading +import unittest +from typing import Any, Dict, List, Optional +from unittest import mock + +from libby import Client, KeywordError, telemetry +from libby.libby import Libby + +from tests.test_client_integration import (GROUP_ID, RABBITMQ_URL, RPC_TIMEOUT_S, + _broker_available, _free_endpoint, + _RabbitFixtureDaemon, _ZmqFixtureDaemon) + +# find_spec on the submodule raises when the opentelemetry package itself is absent +HAS_OTEL = (importlib.util.find_spec("opentelemetry") is not None + and importlib.util.find_spec("opentelemetry.sdk") is not None) +PUBLISH_WAIT_S = 5.0 +UNREACHABLE_TTL_MS = 300 +TRACEPARENT = "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01" + +# Every traced class is skipUnless-guarded on HAS_OTEL, which pylint cannot +# see through +# pylint: disable=possibly-used-before-assignment + +if HAS_OTEL: + from opentelemetry import metrics, trace + from opentelemetry.sdk.metrics import MeterProvider + from opentelemetry.sdk.metrics.export import InMemoryMetricReader + from opentelemetry.sdk.trace import TracerProvider + from opentelemetry.sdk.trace.export import SimpleSpanProcessor + from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter + from opentelemetry.trace import SpanKind + + SPANS = InMemorySpanExporter() + METRICS = InMemoryMetricReader() + + +def setUpModule() -> None: # pylint: disable=invalid-name + """Install in-memory providers once; the OTel API refuses to replace them.""" + if not HAS_OTEL: + return + tracer_provider = TracerProvider() + tracer_provider.add_span_processor(SimpleSpanProcessor(SPANS)) + trace.set_tracer_provider(tracer_provider) + metrics.set_meter_provider(MeterProvider(metric_readers=[METRICS])) + + +class DisabledTests(unittest.TestCase): + """With telemetry off, every hook passes straight through.""" + + def setUp(self) -> None: # pylint: disable=invalid-name + telemetry.disable() + + def test_client_call_carries_no_trace_context(self): + """Send requests untraced.""" + with telemetry.client_call("positionvalue", "hsfei.adc") as call: + call.check_request({"status": "timeout"}) + self.assertIsNone(call.traceparent) + self.assertIsNone(call.tracestate) + self.assertIsNone(call.error_type) + + def test_handler_is_called_unchanged(self): + """Hand payload and ctx to the callback and return its result as is.""" + seen: List[Any] = [] + + def callback(payload: dict, ctx: dict) -> dict: + seen.append((payload, ctx)) + return {"ok": True} + + ctx = {"key": "k", "traceparent": TRACEPARENT} + self.assertEqual(telemetry.traced_handler(callback)({"a": 1}, ctx), {"ok": True}) + self.assertEqual(seen, [({"a": 1}, ctx)]) + + def test_log_lines_are_unchanged(self): + """Append nothing to a log line.""" + formatter = telemetry.TraceContextFormatter("%(message)s") + record = logging.LogRecord("t", logging.INFO, __file__, 1, "hello", None, None) + self.assertEqual(formatter.format(record), "hello") + + def test_configure_needs_an_endpoint(self): + """Stay off when no OTLP endpoint is set.""" + variables = {name: "" for name in ( + "OTEL_EXPORTER_OTLP_ENDPOINT", + "OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", + "OTEL_EXPORTER_OTLP_METRICS_ENDPOINT", + )} + with mock.patch.dict(os.environ, variables): + self.assertIsNone(telemetry.configure("hsfei.adc")) + self.assertFalse(telemetry.enabled()) + + def test_sdk_disabled_wins_over_an_endpoint(self): + """Honour OTEL_SDK_DISABLED even when an endpoint is set.""" + with mock.patch.dict(os.environ, {"OTEL_EXPORTER_OTLP_ENDPOINT": "http://localhost:4318", + "OTEL_SDK_DISABLED": "true"}): + self.assertFalse(telemetry.export_requested()) + + +def _by_kind(spans: List[Any], kind: Any) -> List[Any]: + return [span for span in spans if span.kind == kind] + + +def _duration_points(name: str) -> List[Any]: + data = METRICS.get_metrics_data() + return [ + point + for resource in (data.resource_metrics if data else []) + for scope in resource.scope_metrics + for metric in scope.metrics if metric.name == name + for point in metric.data.data_points + ] + + +def _attributes(points: List[Any]) -> List[Dict[str, Any]]: + return [dict(point.attributes) for point in points] + + +@unittest.skipUnless(HAS_OTEL, "needs the otel extra") +class TracedUnitTests(unittest.TestCase): + """Span shape for calls that need no transport.""" + + def setUp(self) -> None: # pylint: disable=invalid-name + SPANS.clear() + telemetry.use_current_providers() + + def tearDown(self) -> None: # pylint: disable=invalid-name + telemetry.disable() + + def test_handler_continues_the_callers_trace(self): + """Parent the server span on the traceparent from the envelope.""" + handler = telemetry.traced_handler(lambda payload, ctx: {"ok": True}) + handler({}, {"key": "positionvalue", "source": "cli", + "traceparent": TRACEPARENT, "tracestate": None}) + (span,) = SPANS.get_finished_spans() + self.assertEqual(span.kind, SpanKind.SERVER) + self.assertEqual(span.name, "positionvalue") + self.assertEqual(f"{span.context.trace_id:032x}", TRACEPARENT.split("-")[1]) + self.assertEqual(f"{span.parent.span_id:016x}", TRACEPARENT.split("-")[2]) + + def test_handler_exception_is_recorded_and_raised(self): + """Mark the span failed with the exception type, and still raise.""" + + def callback(_payload: dict, _ctx: dict) -> dict: + raise ValueError("bad") + + with self.assertRaises(ValueError): + telemetry.traced_handler(callback)({}, {"key": "k"}) + (span,) = SPANS.get_finished_spans() + self.assertEqual(span.attributes["error.type"], "ValueError") + self.assertFalse(span.status.is_ok) + + def test_log_lines_inside_a_span_carry_its_ids(self): + """Append trace_id and span_id while a span is active.""" + formatter = telemetry.TraceContextFormatter("%(message)s") + record = logging.LogRecord("t", logging.INFO, __file__, 1, "hello", None, None) + with telemetry.operation("work"): + line = formatter.format(record) + (span,) = SPANS.get_finished_spans() + self.assertEqual(line, f"hello trace_id={span.context.trace_id:032x}" + f" span_id={span.context.span_id:016x}") + + def test_unanswered_request_is_a_timeout_error(self): + """Record a request nobody answers as error.type timeout.""" + libby = Libby.zmq("telemetryprobe", _free_endpoint(), + {"nobody": _free_endpoint()}) + try: + libby.request("nobody", "positionvalue", {}, ttl_ms=UNREACHABLE_TTL_MS) + finally: + libby.stop() + (span,) = _by_kind(SPANS.get_finished_spans(), SpanKind.CLIENT) + self.assertEqual(span.attributes["error.type"], "timeout") + self.assertIn({"rpc.system": "libby", "rpc.method": "positionvalue", + "server.address": "nobody", "error.type": "timeout"}, + _attributes(_duration_points("rpc.client.call.duration"))) + + +class _Bases: # pylint: disable=too-few-public-methods + """Namespace for the transport-agnostic traced cases.""" + + @unittest.skipUnless(HAS_OTEL, "needs the otel extra") + class TracedCases(unittest.TestCase): + """Tracing that must hold identically on every transport. + + Concrete subclasses supply ``client`` and ``peer`` from ``setUpClass``. + """ + + client: Client + peer: str + + def setUp(self) -> None: # pylint: disable=invalid-name + telemetry.use_current_providers() + SPANS.clear() + + def tearDown(self) -> None: # pylint: disable=invalid-name + telemetry.disable() + + def test_request_and_handler_share_one_trace(self): + """Parent the daemon's server span on the client's request span.""" + self.client.get(f"{self.peer}.positionvalue", timeout_s=RPC_TIMEOUT_S) + spans = SPANS.get_finished_spans() + (client_span,) = _by_kind(spans, SpanKind.CLIENT) + (server_span,) = _by_kind(spans, SpanKind.SERVER) + self.assertEqual(client_span.name, "positionvalue") + self.assertEqual(client_span.attributes["server.address"], self.peer) + self.assertEqual(server_span.context.trace_id, client_span.context.trace_id) + self.assertEqual(server_span.parent.span_id, client_span.context.span_id) + self.assertIn({"rpc.system": "libby", "rpc.method": "positionvalue"}, + _attributes(_duration_points("rpc.server.call.duration"))) + + def test_error_response_marks_both_spans(self): + """Record a refused write as error_response on client and server.""" + with self.assertRaises(KeywordError): + self.client.set(f"{self.peer}.isreferenced", False, timeout_s=RPC_TIMEOUT_S) + for span in SPANS.get_finished_spans(): + self.assertEqual(span.attributes["error.type"], telemetry.ERROR_RESPONSE) + + +@unittest.skipUnless(_broker_available(), "no RabbitMQ broker reachable at amqp://localhost") +class RabbitMQTracedTests(_Bases.TracedCases): + """Traced cases over RabbitMQ.""" + + @classmethod + def setUpClass(cls): + cls.daemon = _RabbitFixtureDaemon() + cls.daemon.peer_id = "telemetrytestrmq" + cls.daemon.start() + cls.peer = f"{GROUP_ID}.telemetrytestrmq" + cls.client = Client.rabbitmq(rabbitmq_url=RABBITMQ_URL) + + @classmethod + def tearDownClass(cls): + cls.client.close() + cls.daemon.stop() + + +class ZmqTracedTests(_Bases.TracedCases): + """Traced cases over ZMQ.""" + + @classmethod + def setUpClass(cls): + endpoint = _free_endpoint() + cls.daemon = _ZmqFixtureDaemon() + cls.daemon.peer_id = "telemetrytestzmq" + cls.daemon.bind = endpoint + cls.daemon.start() + cls.peer = f"{GROUP_ID}.telemetrytestzmq" + cls.client = Client.zmq(bind=_free_endpoint(), address_book={cls.peer: endpoint}) + + @classmethod + def tearDownClass(cls): + cls.client.close() + cls.daemon.stop() + + +@unittest.skipUnless(HAS_OTEL, "needs the otel extra") +class ZmqPublishTests(unittest.TestCase): + """A subscriber's handling joins the publisher's trace.""" + + def setUp(self) -> None: # pylint: disable=invalid-name + telemetry.use_current_providers() + SPANS.clear() + + def tearDown(self) -> None: # pylint: disable=invalid-name + telemetry.disable() + + def test_consumer_span_follows_the_publish(self): + """Parent the subscriber's process span on the publish span.""" + publisher_bind, subscriber_bind = _free_endpoint(), _free_endpoint() + publisher = Libby.zmq("telemetrypub", publisher_bind, {"telemetrysub": subscriber_bind}) + subscriber = Libby.zmq("telemetrysub", subscriber_bind, {"telemetrypub": publisher_bind}) + received = threading.Event() + try: + subscriber.listen("telemetry.topic", lambda _msg: received.set()) + publisher.publish("telemetry.topic", {"n": 1}) + self.assertTrue(received.wait(PUBLISH_WAIT_S)) + finally: + publisher.stop() + subscriber.stop() + spans = SPANS.get_finished_spans() + (producer,) = _by_kind(spans, SpanKind.PRODUCER) + consumer: Optional[Any] = next(iter(_by_kind(spans, SpanKind.CONSUMER)), None) + self.assertIsNotNone(consumer) + self.assertEqual(consumer.parent.span_id, producer.context.span_id) + + +if __name__ == "__main__": + unittest.main() diff --git a/tox.ini b/tox.ini index 6e2fa84..7784b73 100644 --- a/tox.ini +++ b/tox.ini @@ -5,8 +5,8 @@ isolated_build = true [testenv] description = Run the test suite -# Installed so the InfluxDB sink's tests run rather than skip themselves -extras = influxdb +# Installed so the InfluxDB sink and telemetry tests run rather than skip themselves +extras = influxdb, otel deps = commands = python -m unittest discover -s tests From 8a2bc2e27232d1b6ac5d68e122a597f80771755b Mon Sep 17 00:00:00 2001 From: Mike Langmayr Date: Thu, 29 Oct 2026 15:29:13 -0700 Subject: [PATCH 2/3] Keep the OTel SDK for the process lifetime and write envelope trace context as W3C --- libby/cli/libby_cli.py | 7 ++-- libby/daemon.py | 11 +----- libby/telemetry.py | 78 ++++++++++++++++++------------------ tests/test_telemetry.py | 88 ++++++++++++++++++++++++++++++++++++++++- 4 files changed, 130 insertions(+), 54 deletions(-) diff --git a/libby/cli/libby_cli.py b/libby/cli/libby_cli.py index 6c95e0c..d7daa6c 100644 --- a/libby/cli/libby_cli.py +++ b/libby/cli/libby_cli.py @@ -795,13 +795,12 @@ def main(argv: Optional[List[str]] = None) -> int: argv_display = shlex.join(_redact_argv(argv if argv is not None else sys.argv[1:])) logger.info("invoke cmd=%s argv=%s", namespace.cmd, argv_display) - handle = telemetry.configure(CLI_SERVICE_NAME, export_timeout_s=CLI_EXPORT_TIMEOUT_S, - export_metrics=False) + telemetry.configure(CLI_SERVICE_NAME, export_timeout_s=CLI_EXPORT_TIMEOUT_S, + export_metrics=False) try: return _run_command(namespace) finally: - if handle is not None: - handle.shutdown() + telemetry.flush() def _run_command(namespace: argparse.Namespace) -> int: diff --git a/libby/daemon.py b/libby/daemon.py index c29f291..b6e2dae 100644 --- a/libby/daemon.py +++ b/libby/daemon.py @@ -104,7 +104,6 @@ def __init__(self) -> None: self.libby: Optional[Libby] = None self._last_error: Optional[str] = None self._start_time: Optional[float] = None - self._telemetry: Optional[telemetry.Telemetry] = None self.logger = logging.getLogger(type(self).__name__) self._ensure_last_error_handler() @@ -395,7 +394,7 @@ def start(self) -> None: self._stop_event.clear() self._start_time = time.monotonic() try: - self._telemetry = telemetry.configure( + telemetry.configure( qualified_peer_id(self.config_peer_id(), self.config_group_id()), self.config_group_id(), ) @@ -444,7 +443,6 @@ def start(self) -> None: ) except Exception: self._started = False - self._close_telemetry() raise def request_stop(self) -> None: @@ -467,7 +465,7 @@ def stop(self) -> None: self._started = False self._stop_event.set() self.logger.info("stopped") - self._close_telemetry() + telemetry.flush() def serve(self) -> None: """Start the daemon and block until a signal or stop request.""" @@ -567,11 +565,6 @@ def _close_libby(self) -> None: except Exception: self.logger.exception("Libby transport shutdown failed") - def _close_telemetry(self) -> None: - handle, self._telemetry = self._telemetry, None - if handle is not None: - handle.shutdown() - def payload(self, value: Any = None, /, **extra: Any) -> dict: """Normalize a user result into a JSON-serializable dictionary.""" diff --git a/libby/telemetry.py b/libby/telemetry.py index 3cb391a..fb8c1a8 100644 --- a/libby/telemetry.py +++ b/libby/telemetry.py @@ -37,10 +37,13 @@ class _Instruments: def __init__(self) -> None: # Deferred: OpenTelemetry is the optional libby[otel] extra # pylint: disable=import-outside-toplevel - from opentelemetry import metrics, propagate, trace + from opentelemetry import metrics, trace + from opentelemetry.trace.propagation.tracecontext import TraceContextTextMapPropagator self.trace = trace - self.propagate = propagate + self.metrics = metrics + # bamboo's envelope fields are W3C Trace Context whatever OTEL_PROPAGATORS says + self.propagator = TraceContextTextMapPropagator() self.tracer = trace.get_tracer("libby") meter = metrics.get_meter("libby") self.client_duration = meter.create_histogram( @@ -61,7 +64,7 @@ def incoming(self, traceparent: Optional[str], tracestate: Optional[str]) -> Any carrier = {"traceparent": traceparent} if tracestate: carrier["tracestate"] = tracestate - return self.propagate.extract(carrier) + return self.propagator.extract(carrier) def finish(self, span: Any, error_type: Optional[str], attributes: Dict[str, Any], histogram: Any, started: float) -> None: @@ -78,21 +81,6 @@ def finish(self, span: Any, error_type: Optional[str], attributes: Dict[str, Any _instruments: Optional[_Instruments] = None # pylint: disable=invalid-name -class Telemetry: # pylint: disable=too-few-public-methods - """SDK providers installed by :func:`configure`; shut down to flush them.""" - - def __init__(self, tracer_provider: Any, meter_provider: Optional[Any]) -> None: - self._tracer_provider = tracer_provider - self._meter_provider = meter_provider - - def shutdown(self) -> None: - """Stop recording, export what is pending and stop the exporters.""" - disable() - self._tracer_provider.shutdown() - if self._meter_provider is not None: - self._meter_provider.shutdown() - - def export_requested() -> bool: """Return whether the environment names an OTLP endpoint to export to.""" if os.environ.get("OTEL_SDK_DISABLED", "").strip().lower() == "true": @@ -107,35 +95,49 @@ def enabled() -> bool: def configure(service_name: str, service_namespace: Optional[str] = None, *, export_timeout_s: Optional[float] = None, - export_metrics: bool = True) -> Optional[Telemetry]: + export_metrics: bool = True) -> bool: """Turn on tracing and metrics for this process if export is requested. Install the OTLP SDK unless the application already installed its own - providers, in which case libby records into those. Return a - :class:`Telemetry` only to the caller that installed the SDK, since it - owns the shutdown. Short-lived processes pass ``export_metrics=False``: - each run is a new ``service.instance.id``, so its metrics are new series. + providers, in which case libby records into those. The OTel API allows one + provider per process, so the SDK stays installed until the process exits + and later calls only report whether telemetry is on. Short-lived processes + pass ``export_metrics=False``: each run is a new ``service.instance.id``, + so its metrics are new series. """ - if _instruments is not None or not export_requested(): - return None + if _instruments is not None: + return True + if not export_requested(): + return False try: # Deferred: OpenTelemetry is the optional libby[otel] extra # pylint: disable=import-outside-toplevel from opentelemetry import trace - handle = None if isinstance(trace.get_tracer_provider(), trace.ProxyTracerProvider): - handle = _install_sdk(service_name, service_namespace, export_timeout_s, - export_metrics) + _install_sdk(service_name, service_namespace, export_timeout_s, export_metrics) except ImportError: logger.warning("an OTLP endpoint is set but libby[otel] is not installed; " "telemetry stays off") - return None + return False use_current_providers() - return handle + return True + + +def flush() -> None: + """Export pending spans and metrics now, e.g. before a daemon stops.""" + instruments = _instruments + if instruments is None: + return + for provider in (instruments.trace.get_tracer_provider(), + instruments.metrics.get_meter_provider()): + # API-only providers have nothing to flush + force_flush = getattr(provider, "force_flush", None) + if force_flush is not None: + force_flush() def _install_sdk(service_name: str, service_namespace: Optional[str], - export_timeout_s: Optional[float], export_metrics: bool) -> Telemetry: + export_timeout_s: Optional[float], export_metrics: bool) -> None: # pylint: disable=import-outside-toplevel from opentelemetry import trace from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter @@ -148,23 +150,21 @@ def _install_sdk(service_name: str, service_namespace: Optional[str], tracer_provider = TracerProvider(resource=resource) tracer_provider.add_span_processor(BatchSpanProcessor(OTLPSpanExporter(**timeout))) trace.set_tracer_provider(tracer_provider) - meter_provider = _install_meter(resource, timeout) if export_metrics else None - return Telemetry(tracer_provider, meter_provider) + if export_metrics: + _install_meter(resource, timeout) -def _install_meter(resource: Any, timeout: Dict[str, float]) -> Any: +def _install_meter(resource: Any, timeout: Dict[str, float]) -> None: # pylint: disable=import-outside-toplevel from opentelemetry import metrics from opentelemetry.exporter.otlp.proto.http.metric_exporter import OTLPMetricExporter from opentelemetry.sdk.metrics import MeterProvider from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader - meter_provider = MeterProvider( + metrics.set_meter_provider(MeterProvider( resource=resource, metric_readers=[PeriodicExportingMetricReader(OTLPMetricExporter(**timeout))], - ) - metrics.set_meter_provider(meter_provider) - return meter_provider + )) def _resource(service_name: str, service_namespace: Optional[str]) -> Any: @@ -215,7 +215,7 @@ def start(cls, instruments: _Instruments, name: str, kind: Any, call._finish = lambda span, error_type: instruments.finish( span, error_type, attributes, histogram, started) carrier: Dict[str, str] = {} - instruments.propagate.inject(carrier) + instruments.propagator.inject(carrier) call.traceparent = carrier.get("traceparent") call.tracestate = carrier.get("tracestate") return call diff --git a/tests/test_telemetry.py b/tests/test_telemetry.py index 51c4bc0..67d22f2 100644 --- a/tests/test_telemetry.py +++ b/tests/test_telemetry.py @@ -7,9 +7,13 @@ """ from __future__ import annotations +import http.server import importlib.util import logging import os +import subprocess +import sys +import textwrap import threading import unittest from typing import Any, Dict, List, Optional @@ -26,6 +30,7 @@ HAS_OTEL = (importlib.util.find_spec("opentelemetry") is not None and importlib.util.find_spec("opentelemetry.sdk") is not None) PUBLISH_WAIT_S = 5.0 +SUBPROCESS_TIMEOUT_S = 60.0 UNREACHABLE_TTL_MS = 300 TRACEPARENT = "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01" @@ -34,7 +39,11 @@ # pylint: disable=possibly-used-before-assignment if HAS_OTEL: - from opentelemetry import metrics, trace + from opentelemetry import metrics, propagate, trace + from opentelemetry.propagators.composite import CompositePropagator + # Generated protobuf module, which pylint cannot introspect + from opentelemetry.proto.collector.trace.v1.trace_service_pb2 import ( # pylint: disable=no-name-in-module + ExportTraceServiceRequest) from opentelemetry.sdk.metrics import MeterProvider from opentelemetry.sdk.metrics.export import InMemoryMetricReader from opentelemetry.sdk.trace import TracerProvider @@ -96,7 +105,7 @@ def test_configure_needs_an_endpoint(self): "OTEL_EXPORTER_OTLP_METRICS_ENDPOINT", )} with mock.patch.dict(os.environ, variables): - self.assertIsNone(telemetry.configure("hsfei.adc")) + self.assertFalse(telemetry.configure("hsfei.adc")) self.assertFalse(telemetry.enabled()) def test_sdk_disabled_wins_over_an_endpoint(self): @@ -184,6 +193,81 @@ def test_unanswered_request_is_a_timeout_error(self): _attributes(_duration_points("rpc.client.call.duration"))) + def test_envelope_context_ignores_the_global_propagator(self): + """Write and read W3C trace context even when OTEL_PROPAGATORS names none.""" + original = propagate.get_global_textmap() + propagate.set_global_textmap(CompositePropagator([])) + try: + with telemetry.client_call("positionvalue", "hsfei.adc") as call: + pass + telemetry.traced_handler(lambda payload, ctx: {"ok": True})( + {}, {"key": "positionvalue", "traceparent": call.traceparent, + "tracestate": call.tracestate}) + finally: + propagate.set_global_textmap(original) + self.assertIsNotNone(call.traceparent) + (client_span,) = _by_kind(SPANS.get_finished_spans(), SpanKind.CLIENT) + (server_span,) = _by_kind(SPANS.get_finished_spans(), SpanKind.SERVER) + self.assertEqual(server_span.parent.span_id, client_span.context.span_id) + + +class _SpanSink(http.server.BaseHTTPRequestHandler): + """OTLP/HTTP endpoint recording the names of the spans posted to it.""" + + names: List[str] = [] + + def do_POST(self) -> None: # pylint: disable=invalid-name + """Record span names from a trace export and accept every export.""" + body = self.rfile.read(int(self.headers["Content-Length"])) + if self.path == "/v1/traces": + request = ExportTraceServiceRequest() + request.ParseFromString(body) + self.names.extend(span.name for resource in request.resource_spans + for scope in resource.scope_spans for span in scope.spans) + self.send_response(200) + self.end_headers() + + def log_message(self, *_args: Any) -> None: # pylint: disable=arguments-differ + """Keep the test output quiet.""" + + +# A fresh process, since the SDK install under test is once per process +_RESTART_SCRIPT = textwrap.dedent(""" + import sys + from libby import telemetry + from libby.daemon import LibbyDaemon + + daemon = LibbyDaemon.from_config({"peer_id": "restart", "transport": "zmq", + "bind": sys.argv[1], "discovery_enabled": False}) + for run in ("first run", "second run"): + daemon.start() + with telemetry.operation(run): + pass + daemon.stop() +""") + + +@unittest.skipUnless(HAS_OTEL, "needs the otel extra") +class SdkLifecycleTests(unittest.TestCase): + """The SDK libby installs survives a daemon stopping and starting again.""" + + def test_restarted_daemon_keeps_exporting(self): + """Export spans from both runs of a daemon restarted in one process.""" + _SpanSink.names = [] + server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), _SpanSink) + threading.Thread(target=server.serve_forever, daemon=True).start() + env = {**os.environ, + "OTEL_EXPORTER_OTLP_ENDPOINT": f"http://127.0.0.1:{server.server_address[1]}"} + try: + subprocess.run([sys.executable, "-c", _RESTART_SCRIPT, _free_endpoint()], + env=env, check=True, timeout=SUBPROCESS_TIMEOUT_S, + capture_output=True) + finally: + server.shutdown() + server.server_close() + self.assertEqual(sorted(_SpanSink.names), ["first run", "second run"]) + + class _Bases: # pylint: disable=too-few-public-methods """Namespace for the transport-agnostic traced cases.""" From 4a859fe4db085e12b1c8c4bde0d1d54c0fda84d1 Mon Sep 17 00:00:00 2001 From: Mike Langmayr Date: Thu, 29 Oct 2026 15:30:11 -0700 Subject: [PATCH 3/3] Describe the telemetry lifecycle in the docs --- docs/source/telemetry.md | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/docs/source/telemetry.md b/docs/source/telemetry.md index aabe3c9..06e2279 100644 --- a/docs/source/telemetry.md +++ b/docs/source/telemetry.md @@ -12,8 +12,9 @@ pip install "libby[otel]" export OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:4318 ``` -Every `LibbyDaemon` then exports from `start()` until `stop()`, and the `libby` -CLI exports each command's trace. The CLI exports no metrics, since every run +Every `LibbyDaemon` turns export on in `start()` and flushes in `stop()`, and +the `libby` CLI exports each command's trace. The SDK stays installed until the +process exits, so a daemon restarted in the same process keeps exporting. The CLI exports no metrics, since every run would start new series. Exports use OTLP over HTTP and honour the standard `OTEL_*` variables, e.g. `OTEL_TRACES_SAMPLER` or `OTEL_SDK_DISABLED=true`.