opentelemetry-exporter-otlp-proto-grpc: reconstruct client on first DEADLINE_EXCEEDED error

.changelog/5707.fixed

@@ -0,0 +1 @@
+`opentelemetry-exporter-otlp-proto-grpc`: reconstruct gRPC channel after 3 consecutive `UNAVAILABLE` or `DEADLINE_EXCEEDED` errors

exporter/opentelemetry-exporter-otlp-proto-grpc/src/opentelemetry/exporter/otlp/proto/grpc/exporter.py

@@ -120,6 +120,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())
@@ -351,6 +353,7 @@ def __init__(
 
         self._channel = None
         self._client = None
+        self._consecutive_reconnect_errors = 0
 
         self._shutdown_in_progress = threading.Event()
         self._shutdown = False
@@ -435,6 +438,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]
@@ -447,11 +451,18 @@ 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
+                    # 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 UNAVAILABLE error",
+                            "Reinitializing gRPC channel for %s exporter due to %s error",
                             self._exporting,
+                            error.code(),  # type: ignore [reportAttributeAccessIssue]
                         )
                         try:
                             if self._channel:

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,
@@ -609,28 +610,70 @@ 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 after consecutive UNAVAILABLE and DEADLINE_EXCEEDED errors"""
+        mock_trace_service = TraceServiceServicerWithExportParams(StatusCode.OK)
         add_TraceServiceServicer_to_server(
-            TraceServiceServicerWithExportParams(StatusCode.UNAVAILABLE),
+            mock_trace_service,
+            self.server,
+        )
+        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 per export.
+                exporter = OTLPSpanExporterForTesting(insecure=True, timeout=0.5)
+
+                # 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:
+                    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)
 
-        # 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)
+            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_not_called()
 
     def test_retryable_error_codes_initialization(self):
         # pylint: disable=protected-access