From bdf01dd6ef2e345227a4e3f0ceb818bb6066076d Mon Sep 17 00:00:00 2001 From: ephraimduncan Date: Tue, 27 Oct 2026 10:59:28 +0000 Subject: [PATCH] fix(sdk-python): report a series of failed exports once During an ingest outage, each failed batch sent one more report to on_error and to the telemetry_dev logger. The trace, log, and metric exporters of a client use one failure state. The first failed export goes to on_error and the logger. The SDK sends no report for subsequent failures until an export reaches the ingest again. A metric export with no data sends no request, thus it does not stop the series. The state holds a lock, because the exporters run on background threads. TelemetrySpanProcessor uses one state for its span and metric exporters. Its span exporter also sends export failures to on_error, as the TypeScript processor does. Flush and shutdown errors go to report_error as before. OpenTelemetry catches export failures in force_flush and shutdown, thus a flush or shutdown error is different from an export failure. --- sdks/python/README.md | 4 +- sdks/python/src/telemetry_dev/_client.py | 77 ++--------- sdks/python/src/telemetry_dev/_export.py | 100 ++++++++++++++ sdks/python/src/telemetry_dev/_metrics.py | 16 ++- sdks/python/src/telemetry_dev/otel.py | 6 + sdks/python/tests/otlp_capture.py | 9 +- sdks/python/tests/test_export_on_error.py | 157 +++++++++++++++++++--- sdks/python/tests/test_metrics.py | 5 +- 8 files changed, 273 insertions(+), 101 deletions(-) create mode 100644 sdks/python/src/telemetry_dev/_export.py diff --git a/sdks/python/README.md b/sdks/python/README.md index de4a461..398df3a 100644 --- a/sdks/python/README.md +++ b/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: diff --git a/sdks/python/src/telemetry_dev/_client.py b/sdks/python/src/telemetry_dev/_client.py index b58be4a..4498d98 100644 --- a/sdks/python/src/telemetry_dev/_client.py +++ b/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) diff --git a/sdks/python/src/telemetry_dev/_export.py b/sdks/python/src/telemetry_dev/_export.py new file mode 100644 index 0000000..ab96d61 --- /dev/null +++ b/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() diff --git a/sdks/python/src/telemetry_dev/_metrics.py b/sdks/python/src/telemetry_dev/_metrics.py index 6aebff4..bf6e660 100644 --- a/sdks/python/src/telemetry_dev/_metrics.py +++ b/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 diff --git a/sdks/python/src/telemetry_dev/otel.py b/sdks/python/src/telemetry_dev/otel.py index d38537b..c21f102 100644 --- a/sdks/python/src/telemetry_dev/otel.py +++ b/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,6 +105,9 @@ 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, @@ -111,6 +115,7 @@ def __init__( ) 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, diff --git a/sdks/python/tests/otlp_capture.py b/sdks/python/tests/otlp_capture.py index 7a88773..27c4528 100644 --- a/sdks/python/tests/otlp_capture.py +++ b/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() diff --git a/sdks/python/tests/test_export_on_error.py b/sdks/python/tests/test_export_on_error.py index aa791e7..c059f48 100644 --- a/sdks/python/tests/test_export_on_error.py +++ b/sdks/python/tests/test_export_on_error.py @@ -1,6 +1,7 @@ from __future__ import annotations # pyright: reportPrivateUsage=false +import threading from collections.abc import Mapping, Sequence from typing import Any @@ -18,13 +19,16 @@ ScopeMetrics, ) from opentelemetry.sdk.resources import Resource -from opentelemetry.sdk.trace import ReadableSpan +from opentelemetry.sdk.trace import ReadableSpan, TracerProvider from opentelemetry.sdk.trace.export import SpanExporter, SpanExportResult from opentelemetry.sdk.util.instrumentation import InstrumentationScope from opentelemetry.util.types import AttributeValue -from telemetry_dev._client import _ReportingLogExporter, _ReportingSpanExporter +import telemetry_dev +from telemetry_dev._export import ExportFailures, ReportingLogExporter, ReportingSpanExporter from telemetry_dev._metrics import GuardedOTLPMetricExporter +from telemetry_dev.otel import TelemetrySpanProcessor +from tests.otlp_capture import otlp_capture_server _EMPTY_ATTRIBUTES: Mapping[str, AttributeValue] = {} @@ -104,7 +108,7 @@ def test_span_export_failure_result_reports_error_and_passes_result_through() -> errors: list[BaseException] = [] inner = FakeSpanExporter(SpanExportResult.FAILURE) - result = _ReportingSpanExporter(inner, errors.append).export(()) + result = ReportingSpanExporter(inner, ExportFailures(errors.append)).export(()) assert result is SpanExportResult.FAILURE assert inner.export_calls == 1 @@ -118,7 +122,7 @@ def test_span_export_exception_reports_original_error_and_returns_failure() -> N boom = ConnectionError("span transport down") inner = FakeSpanExporter(boom) - result = _ReportingSpanExporter(inner, errors.append).export(()) + result = ReportingSpanExporter(inner, ExportFailures(errors.append)).export(()) assert result is SpanExportResult.FAILURE assert inner.export_calls == 1 @@ -129,7 +133,7 @@ def test_span_export_success_does_not_report_error() -> None: errors: list[BaseException] = [] inner = FakeSpanExporter(SpanExportResult.SUCCESS) - result = _ReportingSpanExporter(inner, errors.append).export(()) + result = ReportingSpanExporter(inner, ExportFailures(errors.append)).export(()) assert result is SpanExportResult.SUCCESS assert inner.export_calls == 1 @@ -140,7 +144,7 @@ def test_log_export_failure_result_reports_error_and_passes_result_through() -> errors: list[BaseException] = [] inner = FakeLogExporter(LogRecordExportResult.FAILURE) - result = _ReportingLogExporter(inner, errors.append).export(()) + result = ReportingLogExporter(inner, ExportFailures(errors.append)).export(()) assert result is LogRecordExportResult.FAILURE assert inner.export_calls == 1 @@ -154,7 +158,7 @@ def test_log_export_exception_reports_original_error_and_returns_failure() -> No boom = ConnectionError("log transport down") inner = FakeLogExporter(boom) - result = _ReportingLogExporter(inner, errors.append).export(()) + result = ReportingLogExporter(inner, ExportFailures(errors.append)).export(()) assert result is LogRecordExportResult.FAILURE assert inner.export_calls == 1 @@ -165,7 +169,7 @@ def test_log_export_success_does_not_report_error() -> None: errors: list[BaseException] = [] inner = FakeLogExporter(LogRecordExportResult.SUCCESS) - result = _ReportingLogExporter(inner, errors.append).export(()) + result = ReportingLogExporter(inner, ExportFailures(errors.append)).export(()) assert result is LogRecordExportResult.SUCCESS assert inner.export_calls == 1 @@ -188,7 +192,9 @@ def export_failure( return MetricExportResult.FAILURE monkeypatch.setattr(OTLPMetricExporter, "export", export_failure) - exporter = GuardedOTLPMetricExporter("http://example.test/v1/metrics", on_error=errors.append) + exporter = GuardedOTLPMetricExporter( + "http://example.test/v1/metrics", failures=ExportFailures(errors.append) + ) metrics_data = _metrics_data_with_data_points(1) result = exporter.export(metrics_data) @@ -217,7 +223,9 @@ def export_raises( raise boom monkeypatch.setattr(OTLPMetricExporter, "export", export_raises) - exporter = GuardedOTLPMetricExporter("http://example.test/v1/metrics", on_error=errors.append) + exporter = GuardedOTLPMetricExporter( + "http://example.test/v1/metrics", failures=ExportFailures(errors.append) + ) metrics_data = _metrics_data_with_data_points(1) result = exporter.export(metrics_data) @@ -241,7 +249,9 @@ def export_success( return MetricExportResult.SUCCESS monkeypatch.setattr(OTLPMetricExporter, "export", export_success) - exporter = GuardedOTLPMetricExporter("http://example.test/v1/metrics", on_error=errors.append) + exporter = GuardedOTLPMetricExporter( + "http://example.test/v1/metrics", failures=ExportFailures(errors.append) + ) metrics_data = _metrics_data_with_data_points(1) result = exporter.export(metrics_data) @@ -251,26 +261,133 @@ def export_success( assert errors == [] -def test_metric_export_skips_empty_collections_without_reporting_error( +def test_metric_export_skips_empty_collections_without_reporting_or_ending_a_failure_series( monkeypatch: pytest.MonkeyPatch, ) -> None: errors: list[BaseException] = [] calls: list[MetricsData] = [] - def export_unexpected( + def export_failure( self: OTLPMetricExporter, metrics_data: MetricsData, timeout_millis: float | None = 10_000, **kwargs: Any, ) -> MetricExportResult: calls.append(metrics_data) - raise AssertionError("empty metrics should not be exported") - - monkeypatch.setattr(OTLPMetricExporter, "export", export_unexpected) - exporter = GuardedOTLPMetricExporter("http://example.test/v1/metrics", on_error=errors.append) + return MetricExportResult.FAILURE - result = exporter.export(_metrics_data_with_data_points(0)) + monkeypatch.setattr(OTLPMetricExporter, "export", export_failure) + exporter = GuardedOTLPMetricExporter( + "http://example.test/v1/metrics", failures=ExportFailures(errors.append) + ) + empty = _metrics_data_with_data_points(0) + failing = _metrics_data_with_data_points(1) - assert result is MetricExportResult.SUCCESS - assert calls == [] + assert exporter.export(empty) is MetricExportResult.SUCCESS assert errors == [] + assert exporter.export(failing) is MetricExportResult.FAILURE + # The empty collection sends no request, so the series of failures continues. + assert exporter.export(empty) is MetricExportResult.SUCCESS + assert exporter.export(failing) is MetricExportResult.FAILURE + + assert calls == [failing, failing] + assert len(errors) == 1 + + +def test_an_ingest_outage_is_one_report_until_an_export_reaches_the_ingest_again( + caplog: pytest.LogCaptureFixture, +) -> None: + errors: list[BaseException] = [] + + with otlp_capture_server() as ingest: + ingest.status = 503 + telemetry_dev.init( + api_key="td_live_test", + base_url=ingest.url, + export_mode="immediate", + log_level="error", + disable_atexit=True, + # Shorter than the first retry delay, so the exporter does not retry a 503. + timeout=0.5, + on_error=errors.append, + ) + + # One outage stops traces, logs, and metrics. + telemetry_dev.start_span("chat", type="generation").end() + telemetry_dev.log("first outage") + telemetry_dev.flush() + assert sorted(request.path for request in ingest.requests) == [ + "/v1/logs", + "/v1/metrics", + "/v1/traces", + ] + # A flush with no new metrics sends no metric request, and the series continues. + telemetry_dev.flush() + assert len(ingest.requests_for("/v1/metrics")) == 1 + telemetry_dev.log("still in the first outage") + assert len(errors) == 1 + + ingest.status = 200 + telemetry_dev.log("the ingest is back") + ingest.status = 503 + telemetry_dev.log("second outage") + telemetry_dev.log("still in the second outage") + telemetry_dev.shutdown() + + assert len(errors) == 2 + assert "span export failed" in str(errors[0]) + assert "log export failed" in str(errors[1]) + logged = [record for record in caplog.records if record.name == "telemetry_dev"] + assert len(logged) == 2 + + +def test_a_failure_during_a_slow_report_on_another_thread_is_quiet_and_does_not_wait() -> None: + errors: list[BaseException] = [] + first_report_started = threading.Event() + second_failure_done = threading.Event() + + def slow_on_error(error: BaseException) -> None: + errors.append(error) + if len(errors) == 1: + first_report_started.set() + second_failure_done.wait(timeout=5) + + failures = ExportFailures(slow_on_error) + spans = ReportingSpanExporter(FakeSpanExporter(SpanExportResult.FAILURE), failures) + logs = ReportingLogExporter(FakeLogExporter(LogRecordExportResult.FAILURE), failures) + + span_thread = threading.Thread(target=spans.export, args=((),)) + span_thread.start() + assert first_report_started.wait(timeout=5) + assert logs.export(()) is LogRecordExportResult.FAILURE + # The second export did not wait for the first on_error call. + assert span_thread.is_alive() + second_failure_done.set() + span_thread.join(timeout=5) + + assert len(errors) == 1 + + +def test_an_outage_that_stops_byo_traces_and_metrics_is_one_report() -> None: + errors: list[BaseException] = [] + + with otlp_capture_server() as ingest: + # The exporters do not retry a 400, thus each export stops after one request. + ingest.status = 400 + processor = TelemetrySpanProcessor( + api_key="td_live_test", + base_url=ingest.url, + export_mode="immediate", + on_error=errors.append, + ) + provider = TracerProvider(shutdown_on_exit=False) + provider.add_span_processor(processor) + span = provider.get_tracer("user-app").start_span("chat") + span.set_attribute("gen_ai.operation.name", "chat") + span.end() + # The shutdown sends the metrics. + provider.shutdown() + + assert sorted(request.path for request in ingest.requests) == ["/v1/metrics", "/v1/traces"] + assert len(errors) == 1 + assert "span export failed" in str(errors[0]) diff --git a/sdks/python/tests/test_metrics.py b/sdks/python/tests/test_metrics.py index 9eb6df8..5846b59 100644 --- a/sdks/python/tests/test_metrics.py +++ b/sdks/python/tests/test_metrics.py @@ -23,6 +23,7 @@ from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter import telemetry_dev +from telemetry_dev._export import ExportFailures from telemetry_dev._metrics import ( GuardedOTLPMetricExporter, OutputChunkAggregation, @@ -386,7 +387,9 @@ def test_quiet_interval_emits_zero_metric_posts( if internal_metrics: monkeypatch.setenv("OTEL_PYTHON_SDK_INTERNAL_METRICS_ENABLED", "true") with otlp_capture_server() as server: - exporter = GuardedOTLPMetricExporter(endpoint=f"{server.url}/v1/metrics") + exporter = GuardedOTLPMetricExporter( + endpoint=f"{server.url}/v1/metrics", failures=ExportFailures(None) + ) reader = PeriodicExportingMetricReader(exporter, export_interval_millis=2**31 - 1) provider = sdk_meter_provider(reader, Resource.create()) meter = provider.get_meter("quiet")