From 0588bcfdbcf0c408323f7ea6fd6967d852ad4085 Mon Sep 17 00:00:00 2001 From: Lukas Hering Date: Mon, 28 Sep 2026 15:21:54 -0400 Subject: [PATCH 1/3] opentelemetry-exporter-otlp-proto-grpc: reconstruct client on first DEADLINE_EXCEEDED error --- .../exporter/otlp/proto/grpc/exporter.py | 12 +++-- .../tests/test_otlp_exporter_mixin.py | 51 ++++++++++++------- 2 files changed, 41 insertions(+), 22 deletions(-) diff --git a/exporter/opentelemetry-exporter-otlp-proto-grpc/src/opentelemetry/exporter/otlp/proto/grpc/exporter.py b/exporter/opentelemetry-exporter-otlp-proto-grpc/src/opentelemetry/exporter/otlp/proto/grpc/exporter.py index 175bd04d653..edb948ef6de 100644 --- a/exporter/opentelemetry-exporter-otlp-proto-grpc/src/opentelemetry/exporter/otlp/proto/grpc/exporter.py +++ b/exporter/opentelemetry-exporter-otlp-proto-grpc/src/opentelemetry/exporter/otlp/proto/grpc/exporter.py @@ -439,11 +439,17 @@ def _export( retry_info.ParseFromString(retry_info_bin) backoff_seconds = retry_info.retry_delay.seconds + retry_info.retry_delay.nanos / 1.0e9 - # For UNAVAILABLE errors, reinitialize the channel to force reconnection - if error.code() == StatusCode.UNAVAILABLE and retry_num == 0: # type: ignore + # For UNAVAILABLE and DEADLINE_EXCEEDED errors, reinitialize the channel + # to force reconnection. A DEADLINE_EXCEEDED can indicate a stalled stream + # on an otherwise healthy connection, which keepalive cannot detect. + if ( + error.code() in (StatusCode.UNAVAILABLE, StatusCode.DEADLINE_EXCEEDED) # type: ignore + and retry_num == 0 + ): logger.debug( - "Reinitializing gRPC channel for %s exporter due to UNAVAILABLE error", + "Reinitializing gRPC channel for %s exporter due to %s error", self._exporting, + error.code(), # type: ignore [reportAttributeAccessIssue] ) try: if self._channel: diff --git a/exporter/opentelemetry-exporter-otlp-proto-grpc/tests/test_otlp_exporter_mixin.py b/exporter/opentelemetry-exporter-otlp-proto-grpc/tests/test_otlp_exporter_mixin.py index ad2c19298d1..70edad2ea20 100644 --- a/exporter/opentelemetry-exporter-otlp-proto-grpc/tests/test_otlp_exporter_mixin.py +++ b/exporter/opentelemetry-exporter-otlp-proto-grpc/tests/test_otlp_exporter_mixin.py @@ -599,28 +599,41 @@ def test_permanent_failure(self): metrics[2].data.data_points[0].attributes, ) - def test_unavailable_reconnects(self): - """Test that the exporter reconnects on UNAVAILABLE error""" + def test_reconnects_on_transient_error(self): + """Test that the exporter reconnects on UNAVAILABLE and DEADLINE_EXCEEDED errors""" + mock_trace_service = TraceServiceServicerWithExportParams(StatusCode.OK) add_TraceServiceServicer_to_server( - TraceServiceServicerWithExportParams(StatusCode.UNAVAILABLE), + mock_trace_service, self.server, ) - - # Spy on grpc.insecure_channel to verify it's called for reconnection - with patch( - "opentelemetry.exporter.otlp.proto.grpc.exporter.insecure_channel", - side_effect=grpc.insecure_channel, - ) as mock_insecure_channel: - # Mock sleep to avoid waiting - with patch("time.sleep"): - # We expect FAILURE because the server keeps returning UNAVAILABLE - # but we want to verify reconnection attempts happened - self.exporter.export([self.span]) - - # Verify that we attempted to reinitialize the channel (called insecure_channel) - # Since the initial channel was created in setUp (unpatched), this call - # must be from the reconnection logic. - self.assertTrue(mock_insecure_channel.called) + cases = [ + # The server returns UNAVAILABLE immediately. + (StatusCode.UNAVAILABLE, None), + # The server stalls past the exporter timeout, so the client-side + # deadline expires + (StatusCode.OK, 1.0), + ] + for export_result, export_sleep in cases: + with self.subTest(export_result=export_result, export_sleep=export_sleep): + mock_trace_service.export_result = export_result + mock_trace_service.optional_export_sleep = export_sleep + # The timeout is shorter than the first backoff, so only a + # single attempt is made. + exporter = OTLPSpanExporterForTesting(insecure=True, timeout=0.5) + + # Spy on grpc.insecure_channel to verify it's called for reconnection. + # The initial channel was created above (unpatched), so this call + # must be from the reconnection logic. + with patch( + "opentelemetry.exporter.otlp.proto.grpc.exporter.insecure_channel", + side_effect=grpc.insecure_channel, + ) as mock_insecure_channel: + self.assertEqual( + exporter.export([self.span]), + SpanExportResult.FAILURE, + ) + + mock_insecure_channel.assert_called_once() def test_retryable_error_codes_initialization(self): # pylint: disable=protected-access From be6724993f12e34211ea992c16cfcf97e7dbb4ef Mon Sep 17 00:00:00 2001 From: Lukas Hering Date: Mon, 28 Sep 2026 15:24:39 -0400 Subject: [PATCH 2/3] add changelog fragment --- .changelog/5707.fixed | 1 + 1 file changed, 1 insertion(+) create mode 100644 .changelog/5707.fixed diff --git a/.changelog/5707.fixed b/.changelog/5707.fixed new file mode 100644 index 00000000000..83c2c0ec927 --- /dev/null +++ b/.changelog/5707.fixed @@ -0,0 +1 @@ +`opentelemetry-exporter-otlp-proto-grpc`: reconstruct client on first `DEADLINE_EXCEEDED` error From d6cff440b2ca5ab8eae1231b17b882edb47cdae2 Mon Sep 17 00:00:00 2001 From: Lukas Hering Date: Mon, 28 Sep 2026 21:22:25 -0400 Subject: [PATCH 3/3] update channel reconstruction policy --- .changelog/5707.fixed | 2 +- .../exporter/otlp/proto/grpc/exporter.py | 19 +++++---- .../tests/test_otlp_exporter_mixin.py | 40 ++++++++++++++++--- 3 files changed, 48 insertions(+), 13 deletions(-) diff --git a/.changelog/5707.fixed b/.changelog/5707.fixed index 83c2c0ec927..82e4d723e0b 100644 --- a/.changelog/5707.fixed +++ b/.changelog/5707.fixed @@ -1 +1 @@ -`opentelemetry-exporter-otlp-proto-grpc`: reconstruct client on first `DEADLINE_EXCEEDED` error +`opentelemetry-exporter-otlp-proto-grpc`: reconstruct gRPC channel after 3 consecutive `UNAVAILABLE` or `DEADLINE_EXCEEDED` errors diff --git a/exporter/opentelemetry-exporter-otlp-proto-grpc/src/opentelemetry/exporter/otlp/proto/grpc/exporter.py b/exporter/opentelemetry-exporter-otlp-proto-grpc/src/opentelemetry/exporter/otlp/proto/grpc/exporter.py index edb948ef6de..3c6f42090de 100644 --- a/exporter/opentelemetry-exporter-otlp-proto-grpc/src/opentelemetry/exporter/otlp/proto/grpc/exporter.py +++ b/exporter/opentelemetry-exporter-otlp-proto-grpc/src/opentelemetry/exporter/otlp/proto/grpc/exporter.py @@ -119,6 +119,8 @@ ] ) _MAX_RETRYS = 6 +_CHANNEL_RECONNECT_ERROR_CODES = frozenset([StatusCode.UNAVAILABLE, StatusCode.DEADLINE_EXCEEDED]) +_CHANNEL_RECONNECT_ERROR_THRESHOLD = 3 logger = getLogger(__name__) # This prevents logs generated when a log fails to be written to generate another log which fails to be written etc. etc. logger.addFilter(DuplicateFilter()) @@ -343,6 +345,7 @@ def __init__( self._channel = None self._client = None + self._consecutive_reconnect_errors = 0 self._shutdown_in_progress = threading.Event() self._shutdown = False @@ -427,6 +430,7 @@ def _export( metadata=self._headers, timeout=deadline_sec - time(), ) + self._consecutive_reconnect_errors = 0 return self._result.SUCCESS # type: ignore [reportReturnType] except RpcError as error: retry_info_bin = dict(error.trailing_metadata()).get( # type: ignore [reportAttributeAccessIssue] @@ -439,13 +443,14 @@ def _export( retry_info.ParseFromString(retry_info_bin) backoff_seconds = retry_info.retry_delay.seconds + retry_info.retry_delay.nanos / 1.0e9 - # For UNAVAILABLE and DEADLINE_EXCEEDED errors, reinitialize the channel - # to force reconnection. A DEADLINE_EXCEEDED can indicate a stalled stream - # on an otherwise healthy connection, which keepalive cannot detect. - if ( - error.code() in (StatusCode.UNAVAILABLE, StatusCode.DEADLINE_EXCEEDED) # type: ignore - and retry_num == 0 - ): + # After consecutive UNAVAILABLE or DEADLINE_EXCEEDED errors, reinitialize + # the channel to force reconnection. + if error.code() in _CHANNEL_RECONNECT_ERROR_CODES: # type: ignore [reportAttributeAccessIssue] + self._consecutive_reconnect_errors += 1 + else: + self._consecutive_reconnect_errors = 0 + if self._consecutive_reconnect_errors >= _CHANNEL_RECONNECT_ERROR_THRESHOLD: + self._consecutive_reconnect_errors = 0 logger.debug( "Reinitializing gRPC channel for %s exporter due to %s error", self._exporting, diff --git a/exporter/opentelemetry-exporter-otlp-proto-grpc/tests/test_otlp_exporter_mixin.py b/exporter/opentelemetry-exporter-otlp-proto-grpc/tests/test_otlp_exporter_mixin.py index 70edad2ea20..1a54575020e 100644 --- a/exporter/opentelemetry-exporter-otlp-proto-grpc/tests/test_otlp_exporter_mixin.py +++ b/exporter/opentelemetry-exporter-otlp-proto-grpc/tests/test_otlp_exporter_mixin.py @@ -27,6 +27,7 @@ encode_spans, ) from opentelemetry.exporter.otlp.proto.grpc.exporter import ( + _CHANNEL_RECONNECT_ERROR_THRESHOLD, _RETRYABLE_ERROR_CODES, InvalidCompressionValueException, OTLPExporterMixin, @@ -600,7 +601,7 @@ def test_permanent_failure(self): ) def test_reconnects_on_transient_error(self): - """Test that the exporter reconnects on UNAVAILABLE and DEADLINE_EXCEEDED errors""" + """Test that the exporter reconnects after consecutive UNAVAILABLE and DEADLINE_EXCEEDED errors""" mock_trace_service = TraceServiceServicerWithExportParams(StatusCode.OK) add_TraceServiceServicer_to_server( mock_trace_service, @@ -618,22 +619,51 @@ def test_reconnects_on_transient_error(self): mock_trace_service.export_result = export_result mock_trace_service.optional_export_sleep = export_sleep # The timeout is shorter than the first backoff, so only a - # single attempt is made. + # single attempt is made per export. exporter = OTLPSpanExporterForTesting(insecure=True, timeout=0.5) # Spy on grpc.insecure_channel to verify it's called for reconnection. - # The initial channel was created above (unpatched), so this call - # must be from the reconnection logic. with patch( "opentelemetry.exporter.otlp.proto.grpc.exporter.insecure_channel", side_effect=grpc.insecure_channel, ) as mock_insecure_channel: + for _ in range(_CHANNEL_RECONNECT_ERROR_THRESHOLD - 1): + self.assertEqual( + exporter.export([self.span]), + SpanExportResult.FAILURE, + ) + mock_insecure_channel.assert_not_called() + self.assertEqual( exporter.export([self.span]), SpanExportResult.FAILURE, ) + mock_insecure_channel.assert_called_once() + + def test_reconnect_error_count_resets_on_success(self): + """Test that a successful export resets the consecutive reconnect error count""" + mock_trace_service = TraceServiceServicerWithExportParams(StatusCode.UNAVAILABLE) + add_TraceServiceServicer_to_server( + mock_trace_service, + self.server, + ) + exporter = OTLPSpanExporterForTesting(insecure=True, timeout=0.5) + + with patch( + "opentelemetry.exporter.otlp.proto.grpc.exporter.insecure_channel", + side_effect=grpc.insecure_channel, + ) as mock_insecure_channel: + for _ in range(_CHANNEL_RECONNECT_ERROR_THRESHOLD - 1): + self.assertEqual(exporter.export([self.span]), SpanExportResult.FAILURE) + + mock_trace_service.export_result = StatusCode.OK + self.assertEqual(exporter.export([self.span]), SpanExportResult.SUCCESS) + + mock_trace_service.export_result = StatusCode.UNAVAILABLE + for _ in range(_CHANNEL_RECONNECT_ERROR_THRESHOLD - 1): + self.assertEqual(exporter.export([self.span]), SpanExportResult.FAILURE) - mock_insecure_channel.assert_called_once() + mock_insecure_channel.assert_not_called() def test_retryable_error_codes_initialization(self): # pylint: disable=protected-access