From dbe14ca07db8c1033e78b622437ca1a36d18fd7a Mon Sep 17 00:00:00 2001 From: nhobin219 <92616895+nhobin219@users.noreply.github.com> Date: Sat, 3 Oct 2026 22:37:31 +0000 Subject: [PATCH] fix(examples): reconnect the otel exporters after a dropped connection Each exporter held one publication for the life of the process, so a broker restart silently ended a service's telemetry. Exporters now take a common.Publisher holding the stream's URI: it connects on the first batch, drops a connection that fails, and the next batch reconnects. A failed batch is still dropped, never raised, and one line is printed when telemetry starts being dropped and one when it recovers. Printed, not logged, so a record about a failing exporter cannot loop back through it. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01FsSDkeb5rVAxA1FSmKKfQi --- CHANGELOG.md | 8 ++++ examples/otel/common.py | 88 +++++++++++++++++++++++++++++--------- examples/otel/logs.py | 12 ++---- examples/otel/metrics.py | 12 ++---- examples/otel/services.py | 22 +++++----- examples/otel/spans.py | 12 ++---- tests/test_otel_example.py | 55 +++++++++++++++++++++++- 7 files changed, 150 insertions(+), 59 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 968935c..ba6004a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -30,6 +30,14 @@ there was nothing to have changed from. Everything above it is ordinary. does. Its maintainer processes were still opening the logs when the run ended, and cost it several seconds: `demo.main` took 3–11 s with them and about 1 s without. +- **The OTel exporters reconnect after a dropped connection.** They used to + hold one publication for the life of the process, so a broker restart + ended a service's telemetry for good, silently. Each exporter now takes a + `common.Publisher` with the stream's URI: it connects on the first batch, + drops a connection that fails, and the next batch connects again. A + failed batch is still dropped rather than raised, so telemetry never fails + the application. One line is printed when telemetry starts being dropped + and one when it is published again. ## 0.13.1 — 2026-10-03 diff --git a/examples/otel/common.py b/examples/otel/common.py index db4cb08..a7e3e54 100644 --- a/examples/otel/common.py +++ b/examples/otel/common.py @@ -11,11 +11,11 @@ import math from typing import TYPE_CHECKING, Any +import streamcast + if TYPE_CHECKING: from collections.abc import Sequence - import streamcast - # -- the schema pieces -------------------------------------------------------------- ANY_VALUE: dict[str, Any] = { @@ -119,27 +119,73 @@ def scope(value: Any) -> dict[str, object] | None: # noqa: ANN401 — Instrumen # -- publishing from OTel's export thread ----------------------------------------- -def publish( - publication: streamcast.Publication, - loop: asyncio.AbstractEventLoop, - rows: Sequence[dict[str, object]], -) -> bool: - """Publish `rows` from an OTel exporter; True once the stream has them. +class Publisher: + """One stream's publisher for OTel's exporters, reconnecting after a drop. OTel's batch processors call `export` on their own worker thread, and a - streamcast publisher lives on an event loop — so the batch is handed to + streamcast publisher lives on an event loop — so each batch is handed to the loop and the thread waits for the stream's acknowledgement. That makes a flush mean the rows are DURABLE, not merely sent. + + It connects on the first batch, and a batch that fails on the connection + drops it, so the next batch connects again: a broker restart costs the + batches sent while it was down, not every batch after. The SDK calls + `export` every half second or so, which is all the retrying there is. """ - if loop.is_closed(): - # A provider flushing at interpreter exit, after the loop has gone: - # nothing can be published, and the SDK wants a result, not a raise. - return False - - sent = asyncio.run_coroutine_threadsafe(publication.send_many(rows), loop) - try: - sent.result(timeout=10) - except Exception: # noqa: BLE001 — the SDK wants a result, not an exception - return False - - return True + + def __init__(self, uri: str, loop: asyncio.AbstractEventLoop) -> None: + self.uri = uri + self._loop = loop + self._publication: streamcast.Publication | None = None + self._failing = False + # Two services' processors can export at once; one connection each time. + self._connecting = asyncio.Lock() + + async def _send(self, rows: Sequence[dict[str, object]]) -> None: + async with self._connecting: + if self._publication is None: + self._publication = await streamcast.publish(self.uri) + + publication = self._publication + + try: + await publication.send_many(rows) + except streamcast.Rejected: + raise # a bad row, on a connection that is fine + + except BaseException: + if self._publication is publication: + self._publication = None # the next batch reconnects + + raise + + def send(self, rows: Sequence[dict[str, object]]) -> bool: + """Publish `rows`; True once the stream has them. Never raises: failing + to send telemetry must not fail the application that made it.""" + if self._loop.is_closed(): + # A provider flushing at interpreter exit, after the loop has gone. + return False + + sent = asyncio.run_coroutine_threadsafe(self._send(rows), self._loop) + try: + sent.result(timeout=10) + except Exception as exc: # noqa: BLE001 — the SDK wants a result, not an exception + sent.cancel() + # Printed, not logged: a `logging` record would be exported + # through this same failing publisher. Once per outage. + if not self._failing: + print(f"telemetry to {self.uri} is being dropped: {exc!r}", flush=True) + + self._failing = True + return False + + if self._failing: + print(f"telemetry to {self.uri} is being published again", flush=True) + + self._failing = False + return True + + async def aclose(self) -> None: + if self._publication is not None: + await self._publication.close() + self._publication = None diff --git a/examples/otel/logs.py b/examples/otel/logs.py index 1eb4708..cd969b2 100644 --- a/examples/otel/logs.py +++ b/examples/otel/logs.py @@ -22,13 +22,10 @@ from examples.otel import common if TYPE_CHECKING: - import asyncio from collections.abc import Sequence from opentelemetry.sdk._logs import ReadableLogRecord - import streamcast - SCHEMA: dict[str, Any] = { "type": "object", "properties": { @@ -79,15 +76,12 @@ def row(record: ReadableLogRecord) -> dict[str, object]: class StreamLogExporter(LogRecordExporter): """Publishes each batch of OTel log records to a stream.""" - def __init__( - self, publication: streamcast.Publication, loop: asyncio.AbstractEventLoop - ) -> None: - self._publication = publication - self._loop = loop + def __init__(self, publisher: common.Publisher) -> None: + self._publisher = publisher def export(self, batch: Sequence[ReadableLogRecord]) -> LogRecordExportResult: rows = [row(record) for record in batch] - if common.publish(self._publication, self._loop, rows): + if self._publisher.send(rows): return LogRecordExportResult.SUCCESS return LogRecordExportResult.FAILURE diff --git a/examples/otel/metrics.py b/examples/otel/metrics.py index df8d85a..9a883c0 100644 --- a/examples/otel/metrics.py +++ b/examples/otel/metrics.py @@ -46,13 +46,9 @@ from examples.otel import common if TYPE_CHECKING: - import asyncio - from opentelemetry.sdk.metrics import Exemplar from opentelemetry.sdk.metrics.export import MetricsData - import streamcast - def _list_of(items: dict[str, Any]) -> dict[str, Any]: return {"type": ["array", "null"], "items": items} @@ -252,15 +248,13 @@ class StreamMetricExporter(MetricExporter): def __init__( self, - publication: streamcast.Publication, - loop: asyncio.AbstractEventLoop, + publisher: common.Publisher, preferred_temporality: dict[type, AggregationTemporality] | None = None, ) -> None: super().__init__( preferred_temporality=preferred_temporality or PREFERRED_TEMPORALITY ) - self._publication = publication - self._loop = loop + self._publisher = publisher def export( self, @@ -269,7 +263,7 @@ def export( **kwargs: object, # noqa: ARG002 ) -> MetricExportResult: batch = rows(metrics_data) - if not batch or common.publish(self._publication, self._loop, batch): + if not batch or self._publisher.send(batch): return MetricExportResult.SUCCESS return MetricExportResult.FAILURE diff --git a/examples/otel/services.py b/examples/otel/services.py index 6b87f1c..7288514 100644 --- a/examples/otel/services.py +++ b/examples/otel/services.py @@ -36,8 +36,7 @@ from opentelemetry.sdk.trace.export import BatchSpanProcessor from opentelemetry.trace import SpanKind, Status, StatusCode -import streamcast -from examples.otel import logs, metrics, spans +from examples.otel import common, logs, metrics, spans if TYPE_CHECKING: from opentelemetry.metrics import Counter, Histogram @@ -248,16 +247,19 @@ def orders_forever(exporters: Exporters, stop: threading.Event) -> None: async def exporting(base: str): # noqa: ANN201 """A publisher on each stream, wrapped as the exporters OTel's SDK calls.""" loop = asyncio.get_running_loop() - async with ( - streamcast.publish(f"{base}/logs") as log_publication, - streamcast.publish(f"{base}/spans") as span_publication, - streamcast.publish(f"{base}/metrics") as metric_publication, - ): + publishers = [ + common.Publisher(f"{base}/{name}", loop) + for name in ("logs", "spans", "metrics") + ] + try: yield Exporters( - logs.StreamLogExporter(log_publication, loop), - spans.StreamSpanExporter(span_publication, loop), - metrics.StreamMetricExporter(metric_publication, loop), + logs.StreamLogExporter(publishers[0]), + spans.StreamSpanExporter(publishers[1]), + metrics.StreamMetricExporter(publishers[2]), ) + finally: + for publisher in publishers: + await publisher.aclose() async def run(broker: str) -> None: diff --git a/examples/otel/spans.py b/examples/otel/spans.py index bfb7c24..26cc925 100644 --- a/examples/otel/spans.py +++ b/examples/otel/spans.py @@ -20,13 +20,10 @@ from examples.otel import common if TYPE_CHECKING: - import asyncio from collections.abc import Sequence from opentelemetry.sdk.trace import ReadableSpan - import streamcast - def _struct(properties: dict[str, Any]) -> dict[str, Any]: return { @@ -137,15 +134,12 @@ def row(span: ReadableSpan) -> dict[str, object]: class StreamSpanExporter(SpanExporter): """Publishes each batch of finished OTel spans to a stream.""" - def __init__( - self, publication: streamcast.Publication, loop: asyncio.AbstractEventLoop - ) -> None: - self._publication = publication - self._loop = loop + def __init__(self, publisher: common.Publisher) -> None: + self._publisher = publisher def export(self, spans: Sequence[ReadableSpan]) -> SpanExportResult: rows = [row(span) for span in spans] - if common.publish(self._publication, self._loop, rows): + if self._publisher.send(rows): return SpanExportResult.SUCCESS return SpanExportResult.FAILURE diff --git a/tests/test_otel_example.py b/tests/test_otel_example.py index 2c05ed3..9b3d91e 100644 --- a/tests/test_otel_example.py +++ b/tests/test_otel_example.py @@ -9,10 +9,12 @@ from __future__ import annotations +import asyncio import math import re from typing import Any +import litelink import pytest pytest.importorskip("opentelemetry.sdk", reason="the OTel example's dev dependency") @@ -134,6 +136,57 @@ def test_each_kind(self, value, stored): assert common.any_value(value) == stored +class TestThePublisher: + """What every exporter publishes through: telemetry must never fail the app.""" + + async def test_a_broker_restart_costs_only_the_batches_sent_while_it_was_down( + self, tmp_path, capsys + ): + schema = { + "type": "object", + "properties": {"i": {"type": "integer"}}, + "required": ["i"], + } + + async def broker(port: int = 0): # noqa: ANN202 + stream = streamcast.Stream.new("t", root=tmp_path, schema=schema) + return await streamcast.serve( + stream, "127.0.0.1", port, publish=True, maintain=False + ) + + server = await broker() + port = server.sockets[0].getsockname()[1] + publisher = common.Publisher( + f"ws://127.0.0.1:{port}/t", asyncio.get_running_loop() + ) + # `send` blocks on the loop, as OTel's export thread does. + assert await asyncio.to_thread(publisher.send, [{"i": 1}]) + + server.close() + await server.wait_closed() + # Down: dropped, not raised. Twice, but reported once. + assert not await asyncio.to_thread(publisher.send, [{"i": 2}]) + assert not await asyncio.to_thread(publisher.send, [{"i": 3}]) + + server = await broker(port) + try: + assert await asyncio.to_thread(publisher.send, [{"i": 4}]) + await publisher.aclose() + finally: + server.close() + await server.wait_closed() + + # One line when the outage starts, one when it ends: not one a batch. + printed = capsys.readouterr().out.splitlines() + assert len(printed) == 2 + assert "is being dropped" in printed[0] + assert "is being published again" in printed[1] + with litelink.open(tmp_path, "t", read_only=True) as log: + stored = log.sql("SELECT i FROM log ORDER BY litelink_offset").read_all() + + assert stored.column("i").to_pylist() == [1, 4] + + def emitted() -> list: """Real SDK log records: `logging` through OTel's handler, captured in memory.""" import logging @@ -482,7 +535,7 @@ def test_the_exporter_asks_for_delta_counters_and_histograms(self): from opentelemetry.sdk.metrics import Counter, Histogram, UpDownCounter from opentelemetry.sdk.metrics.export import AggregationTemporality - exporter = metrics.StreamMetricExporter(None, None) # ty: ignore[invalid-argument-type] + exporter = metrics.StreamMetricExporter(None) # ty: ignore[invalid-argument-type] preferred = exporter._preferred_temporality # noqa: SLF001 assert preferred is not None assert preferred[Counter] == AggregationTemporality.DELTA