fix(sdk-python): report a series of failed exports once

sdks/python/README.md

@@ -142,7 +142,7 @@ the core SDK with provider integrations to enable the new metric.
 | `max_attribute_length` | `65536` | Non-negative integer per-content-attribute cap. Invalid values call `on_error`, do not raise into application code, and install a disabled client; truncated values use the ASCII `...[truncated]` marker, or its prefix when the cap is shorter than the marker. The cap also applies, with the marker, to string metadata and log attributes; any other string attribute is cut at the cap without a marker, except that a cap of `0` leaves those at 65536. |
 | `span_filter` | `None` | Export predicate `Callable[[ReadableSpan], bool]`. |
 | `sampler` | OTel environment configuration | OTel `Sampler` that overrides environment settings. Session roots use its root policy. Real parents keep their sampling decisions. |
-| `on_error` | `None` | Receives internal SDK errors; repeated instrumentation failures from one stream may be coalesced into one report. The SDK never raises them into application code. |
+| `on_error` | `None` | Receives internal SDK errors; repeated instrumentation failures from one stream may be coalesced into one report. The SDK never raises them into application code. A series of failed exports, for example during an ingest outage, is one report to `on_error` and the `telemetry_dev` logger. After an export reaches the ingest, the next failure is a new report. |
 | `disable_atexit` | `False` | Skip the automatic atexit shutdown. |
 | `timeout` | `10.0` | OTLP HTTP timeout in seconds. |
 | `span_exporter`, `log_exporter`, `metric_reader` | `None` | Test seams / offline mode; any of them enables the client without an API key. |
@@ -229,7 +229,7 @@ from telemetry_dev import TelemetrySpanProcessor
 | `metrics` | `True` | Auto-record GenAI duration/token histograms for exported spans when an API key is available. |
 | `service_name` | `OTEL_SERVICE_NAME` or `unknown_service` | `service.name` on the metrics resource. |
 | `environment` | `TELEMETRY_DEV_ENVIRONMENT` or `production` | `deployment.environment.name` on the metrics resource. |
-| `on_error` | `None` | Receives internal processor errors; errors are never raised into your code. |
+| `on_error` | `None` | Receives export errors of the telemetry.dev exporters and internal processor errors; errors are never raised into your code. A series of failed exports is one report. After an export reaches the ingest, the next failure is a new report. |
 | `span_exporter` | `None` | Advanced/test seam replacing the telemetry.dev OTLP trace exporter. |
 
 Without an API key and without `span_exporter`, `TelemetrySpanProcessor()` is an inert no-op:

sdks/python/src/telemetry_dev/_client.py

@@ -15,7 +15,6 @@
 from opentelemetry.sdk._logs.export import (
     BatchLogRecordProcessor,
     LogRecordExporter,
-    LogRecordExportResult,
     SimpleLogRecordProcessor,
 )
 from opentelemetry.sdk.metrics import Histogram, MeterProvider
@@ -26,7 +25,7 @@
 )
 from opentelemetry.sdk.resources import Resource
 from opentelemetry.sdk.trace import ReadableSpan, SpanLimits, TracerProvider
-from opentelemetry.sdk.trace.export import SpanExporter, SpanExportResult
+from opentelemetry.sdk.trace.export import SpanExporter
 from opentelemetry.sdk.trace.sampling import Sampler
 from opentelemetry.trace import Tracer
 
@@ -41,6 +40,7 @@
     resolve_config,
 )
 from ._context import SessionSampler
+from ._export import ExportFailures, ReportingLogExporter, ReportingSpanExporter
 from ._metrics import (
     GuardedOTLPMetricExporter,
     MetricsRecorder,
@@ -60,68 +60,6 @@ def _default_scope_filter(span: ReadableSpan) -> bool:
     return scope is not None and scope.name == SCOPE_NAME
 
 
-class _ReportingSpanExporter(SpanExporter):
-    """Surfaces transport failures to on_error (TS parity: exporters route errors to onError,
-    the stock OTLP exporter only logs them)."""
-
-    def __init__(
-        self, inner: SpanExporter, on_error: Callable[[BaseException], None] | None
-    ) -> None:
-        self._inner = inner
-        self._on_error = on_error
-
-    def export(self, spans: Any) -> SpanExportResult:
-        try:
-            result = self._inner.export(spans)
-        except BaseException as exc:
-            report_error(self._on_error, "span export failed", exc)
-            return SpanExportResult.FAILURE
-        if result is SpanExportResult.FAILURE:
-            report_error(
-                self._on_error,
-                "span export failed",
-                RuntimeError("OTLP span export failed (check the API key and ingest URL)"),
-            )
-        return result
-
-    def force_flush(self, timeout_millis: int = 30000) -> bool:
-        return self._inner.force_flush(timeout_millis)
-
-    def shutdown(self) -> None:
-        self._inner.shutdown()
-
-
-class _ReportingLogExporter(LogRecordExporter):
-    """Log-exporter twin of _ReportingSpanExporter."""
-
-    def __init__(
-        self, inner: LogRecordExporter, on_error: Callable[[BaseException], None] | None
-    ) -> None:
-        self._inner = inner
-        self._on_error = on_error
-
-    def export(self, batch: Any) -> LogRecordExportResult:
-        try:
-            result = self._inner.export(batch)
-        except BaseException as exc:
-            report_error(self._on_error, "log export failed", exc)
-            return LogRecordExportResult.FAILURE
-        if result is LogRecordExportResult.FAILURE:
-            report_error(
-                self._on_error,
-                "log export failed",
-                RuntimeError("OTLP log export failed (check the API key and ingest URL)"),
-            )
-        return result
-
-    def force_flush(self, timeout_millis: int = 30000) -> bool:
-        inner_flush = getattr(self._inner, "force_flush", None)
-        return inner_flush(timeout_millis) if inner_flush is not None else True
-
-    def shutdown(self) -> None:
-        self._inner.shutdown()
-
-
 class Client:
     """Holds the telemetry pipelines for one init() call. Obtain via telemetry_dev.init()."""
 
@@ -195,6 +133,7 @@ def __init__(
             ),
         )
         self._tracer_provider.sampler = SessionSampler(self._tracer_provider.sampler)
+        failures = ExportFailures(on_error)
 
         # Without an api key (enabled via a test seam), never construct real network
         # exporters — they would POST to the ingest with a bogus Authorization header.
@@ -206,7 +145,7 @@ def __init__(
                 compression=Compression.Gzip,
                 timeout=timeout,
                 preferred_temporality={Histogram: AggregationTemporality.DELTA},
-                on_error=on_error,
+                failures=failures,
             )
             metric_exporter.output_chunks = self._output_chunks
             metric_reader = PeriodicExportingMetricReader(
@@ -221,14 +160,14 @@ def __init__(
             )
 
         if span_exporter is None:
-            span_exporter = _ReportingSpanExporter(
+            span_exporter = ReportingSpanExporter(
                 OTLPSpanExporter(
                     endpoint=f"{config.base_url}/v1/traces",
                     headers=headers,
                     compression=Compression.Gzip,
                     timeout=timeout,
                 ),
-                on_error,
+                failures,
             )
         if span_filter is None and register_global:
             span_filter = _default_scope_filter
@@ -243,14 +182,14 @@ def __init__(
         self.tracer = self._tracer_provider.get_tracer(SCOPE_NAME, SDK_VERSION)
 
         if log_exporter is None and config.api_key is not None:
-            log_exporter = _ReportingLogExporter(
+            log_exporter = ReportingLogExporter(
                 OTLPLogExporter(
                     endpoint=f"{config.base_url}/v1/logs",
                     headers=headers,
                     compression=Compression.Gzip,
                     timeout=timeout,
                 ),
-                on_error,
+                failures,
             )
         if log_exporter is not None:
             self._logger_provider = LoggerProvider(resource=resource, shutdown_on_exit=False)

sdks/python/src/telemetry_dev/_export.py

@@ -0,0 +1,100 @@
+from __future__ import annotations
+
+import threading
+from collections.abc import Callable
+from typing import Any
+
+from opentelemetry.sdk._logs.export import LogRecordExporter, LogRecordExportResult
+from opentelemetry.sdk.trace.export import SpanExporter, SpanExportResult
+
+from ._config import report_error
+
+
+class ExportFailures:
+    """Reports the first failed export of a series to on_error and the SDK logger.
+
+    The series stops when an export reaches the ingest, and the next failed export starts a new
+    series. During an outage, each batch is a failed export, and one report for each batch is too
+    many. The exporters of one client or TelemetrySpanProcessor use one instance, so an outage
+    that stops traces, logs, and metrics is one report. Exporters run on background threads, thus
+    report() and clear() hold a lock.
+    """
+
+    def __init__(self, on_error: Callable[[BaseException], None] | None) -> None:
+        self._on_error = on_error
+        self._lock = threading.Lock()
+        self._reported = False
+
+    def report(self, message: str, exc: BaseException) -> None:
+        with self._lock:
+            if self._reported:
+                return
+            self._reported = True
+        # report_error runs outside the lock. Thus a slow on_error does not stop other exporters,
+        # and an on_error that sends telemetry does not deadlock.
+        report_error(self._on_error, message, exc)
+
+    def clear(self) -> None:
+        """Records that an export reached the ingest, so that report() sends the next failure."""
+        with self._lock:
+            self._reported = False
+
+
+class ReportingSpanExporter(SpanExporter):
+    """Reports export failures through its ExportFailures (TS parity). The OTLP exporter of
+    OpenTelemetry only logs them."""
+
+    def __init__(self, inner: SpanExporter, failures: ExportFailures) -> None:
+        self._inner = inner
+        self._failures = failures
+
+    def export(self, spans: Any) -> SpanExportResult:
+        try:
+            result = self._inner.export(spans)
+        except BaseException as exc:
+            self._failures.report("span export failed", exc)
+            return SpanExportResult.FAILURE
+        if result is SpanExportResult.FAILURE:
+            self._failures.report(
+                "span export failed",
+                RuntimeError("OTLP span export failed (check the API key and ingest URL)"),
+            )
+        else:
+            self._failures.clear()
+        return result
+
+    def force_flush(self, timeout_millis: int = 30000) -> bool:
+        return self._inner.force_flush(timeout_millis)
+
+    def shutdown(self) -> None:
+        self._inner.shutdown()
+
+
+class ReportingLogExporter(LogRecordExporter):
+    """Log-exporter twin of ReportingSpanExporter."""
+
+    def __init__(self, inner: LogRecordExporter, failures: ExportFailures) -> None:
+        self._inner = inner
+        self._failures = failures
+
+    def export(self, batch: Any) -> LogRecordExportResult:
+        try:
+            result = self._inner.export(batch)
+        except BaseException as exc:
+            self._failures.report("log export failed", exc)
+            return LogRecordExportResult.FAILURE
+        if result is LogRecordExportResult.FAILURE:
+            self._failures.report(
+                "log export failed",
+                RuntimeError("OTLP log export failed (check the API key and ingest URL)"),
+            )
+        else:
+            self._failures.clear()
+        return result
+
+    def force_flush(self, timeout_millis: int = 30000) -> bool:
+        inner_flush = getattr(self._inner, "force_flush", None)
+        return inner_flush(timeout_millis) if inner_flush is not None else True
+
+    def shutdown(self) -> None:
+        self._inner.shutdown()

sdks/python/src/telemetry_dev/_metrics.py

@@ -27,6 +27,7 @@
 from opentelemetry.trace import INVALID_SPAN_CONTEXT, NonRecordingSpan, set_span_in_context
 
 from ._config import report_error
+from ._export import ExportFailures
 from ._semconv import (
     ATTR_ERROR_TYPE,
     ATTR_TIME_TO_FIRST_CHUNK,
@@ -166,16 +167,17 @@ def _has_data_points(metrics_data: MetricsData) -> bool:
 
 class GuardedOTLPMetricExporter(OTLPMetricExporter):
     """OTLP metric exporter that skips POSTs for collections with zero data points
-    (mirrors the TS emitter's empty-datapoint guard) and surfaces failures to on_error."""
+    (mirrors the TS emitter's empty-datapoint guard) and reports failures through its
+    ExportFailures."""
 
     def __init__(
         self,
         *args: Any,
-        on_error: Callable[[BaseException], None] | None = None,
+        failures: ExportFailures,
         **kwargs: Any,
     ) -> None:
         super().__init__(*args, **kwargs)  # pyright: ignore[reportUnknownMemberType]
-        self._td_on_error = on_error
+        self._td_failures = failures
         self.output_chunks: OutputChunkAggregation | None = None
 
     def export(
@@ -187,20 +189,22 @@ def export(
         if self.output_chunks is not None:
             metrics_data = _with_output_chunks(metrics_data, self.output_chunks)
         if not _has_data_points(metrics_data):
+            # No request reaches the ingest, so a series of failed exports continues.
             return MetricExportResult.SUCCESS
         try:
             result = super().export(  # pyright: ignore[reportUnknownMemberType]
                 metrics_data, timeout_millis=timeout_millis, **kwargs
             )
         except BaseException as exc:
-            report_error(self._td_on_error, "metric export failed", exc)
+            self._td_failures.report("metric export failed", exc)
             return MetricExportResult.FAILURE
         if result is MetricExportResult.FAILURE:
-            report_error(
-                self._td_on_error,
+            self._td_failures.report(
                 "metric export failed",
                 RuntimeError("OTLP metric export failed (check the API key and ingest URL)"),
             )
+        else:
+            self._td_failures.clear()
         return result
 
 

sdks/python/src/telemetry_dev/otel.py

@@ -22,6 +22,7 @@
 from opentelemetry.sdk.trace.export import SpanExporter, SpanExportResult
 
 from ._config import SDK_VERSION, logger, report_error, resolve_config
+from ._export import ExportFailures, ReportingSpanExporter
 from ._metrics import GuardedOTLPMetricExporter, MetricsRecorder, sdk_meter_provider
 from ._processor import ExportMode, StampingSpanProcessor
 from ._semconv import SCOPE_NAME
@@ -104,13 +105,17 @@ def __init__(
             service_name=service_name,
         )
 
+        # The span and metric exporters use one failure state, as the exporters of a client do.
+        failures = ExportFailures(on_error)
+
         if span_exporter is None:
             span_exporter = create_telemetry_span_exporter(
                 api_key=config.api_key,
                 base_url=config.base_url,
             )
             if isinstance(span_exporter, _NoOpSpanExporter):
                 return
+            span_exporter = ReportingSpanExporter(span_exporter, failures)
 
         if metrics_recorder is None and metrics and config.api_key is not None:
             resource = Resource.create(
@@ -128,6 +133,7 @@ def __init__(
                 compression=Compression.Gzip,
                 timeout=_EXPORTER_TIMEOUT_S,
                 preferred_temporality={Histogram: AggregationTemporality.DELTA},
+                failures=failures,
             )
             reader = PeriodicExportingMetricReader(
                 metric_exporter,

sdks/python/tests/otlp_capture.py

@@ -26,6 +26,8 @@ def decompressed(self) -> bytes:
 class CaptureServer:
     url: str
     requests: list[CapturedRequest]
+    # The HTTP status of each response. To make an ingest outage, set an error status such as 503.
+    status: int = 200
 
     def requests_for(self, path: str) -> list[CapturedRequest]:
         return [request for request in self.requests if request.path == path]
@@ -47,7 +49,7 @@ def do_POST(self) -> None:
                     body=body,
                 )
             )
-            self.send_response(200)
+            self.send_response(capture.status)
             self.send_header("content-type", "application/x-protobuf")
             self.send_header("content-length", "0")
             self.end_headers()
@@ -59,10 +61,11 @@ def log_message(self, format: str, *args: object) -> None:
     thread = threading.Thread(
         target=server.serve_forever, kwargs={"poll_interval": 0.01}, daemon=True
     )
+    host, port = server.server_address[0], server.server_address[1]
+    capture = CaptureServer(url=f"http://{host}:{port}", requests=captured)
     thread.start()
     try:
-        host, port = server.server_address[0], server.server_address[1]
-        yield CaptureServer(url=f"http://{host}:{port}", requests=captured)
+        yield capture
     finally:
         server.shutdown()
         server.server_close()