Respond to comments on PR

Author: DylanRussell

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

@@ -293,33 +293,32 @@ def _export(
         # FIXME remove this check if the export type for traces
         # gets updated to a class that represents the proto
         # TracesData and use the code below instead.
-        retry_info = RetryInfo()
         with self._export_lock:
             deadline_sec = time() + self._timeout
-            for retry_num in range(1, _MAX_RETRYS + 1):
-                backoff_seconds = 2 ** (retry_num - 1) * random.uniform(
-                    0.8, 1.2
-                )
+            for retry_num in range(_MAX_RETRYS):
                 try:
                     self._client.Export(
                         request=self._translate_data(data),
                         metadata=self._headers,
-                        timeout=self._timeout,
+                        timeout=deadline_sec - time(),
                     )
                     return self._result.SUCCESS
                 except RpcError as error:
                     retry_info_bin = dict(error.trailing_metadata()).get(
                         "google.rpc.retryinfo-bin"
                     )
+                    # multiplying by a random number between .8 and 1.2 introduces a +/20% jitter to each backoff.
+                    backoff_seconds = 2**retry_num * random.uniform(0.8, 1.2)
                     if retry_info_bin is not None:
+                        retry_info = RetryInfo()
                         retry_info.ParseFromString(retry_info_bin)
                         backoff_seconds = (
                             retry_info.retry_delay.seconds
                             + retry_info.retry_delay.nanos / 1.0e9
                         )
                     if (
                         error.code() not in _RETRYABLE_ERROR_CODES
-                        or retry_num == _MAX_RETRYS
+                        or retry_num + 1 == _MAX_RETRYS
                         or backoff_seconds > (deadline_sec - time())
                     ):
                         logger.error(

exporter/opentelemetry-exporter-otlp-proto-grpc/test-requirements.txt

@@ -11,6 +11,7 @@ tomli==2.0.1
 typing_extensions==4.10.0
 wrapt==1.16.0
 zipp==3.19.2
+googleapis-common-protos==1.63.2
 -e opentelemetry-api
 -e tests/opentelemetry-test-utils
 -e exporter/opentelemetry-exporter-otlp-proto-common

exporter/opentelemetry-exporter-otlp-proto-grpc/tests/test_otlp_exporter_mixin.py

@@ -90,33 +90,32 @@ class TraceServiceServicerWithExportParams(TraceServiceServicer):
     def __init__(
         self,
         export_result: StatusCode,
-        optional_retry_millis: Optional[int] = None,
+        optional_retry_nanos: Optional[int] = None,
         optional_export_sleep: Optional[float] = None,
     ):
         self.export_result = export_result
         self.optional_export_sleep = optional_export_sleep
-        self.optional_retry_millis = optional_retry_millis
+        self.optional_retry_nanos = optional_retry_nanos
         self.num_requests = 0
 
     # pylint: disable=invalid-name,unused-argument
     def Export(self, request, context):
         self.num_requests += 1
         if self.optional_export_sleep:
             time.sleep(self.optional_export_sleep)
-        if self.export_result != StatusCode.OK:
-            if self.optional_retry_millis:
-                context.set_trailing_metadata(
+        if self.export_result != StatusCode.OK and self.optional_retry_nanos:
+            context.set_trailing_metadata(
+                (
                     (
-                        (
-                            "google.rpc.retryinfo-bin",
-                            RetryInfo(
-                                retry_delay=Duration(
-                                    nanos=self.optional_retry_millis
-                                )
-                            ).SerializeToString(),
-                        ),
-                    )
+                        "google.rpc.retryinfo-bin",
+                        RetryInfo(
+                            retry_delay=Duration(
+                                nanos=self.optional_retry_nanos
+                            )
+                        ).SerializeToString(),
+                    ),
                 )
+            )
         context.set_code(self.export_result)
 
         return ExportTraceServiceResponse()
@@ -372,7 +371,8 @@ def test_export_over_closed_grpc_channel(self):
 
     def test_retry_info_is_respected(self):
         mock_trace_service = TraceServiceServicerWithExportParams(
-            StatusCode.UNAVAILABLE, optional_retry_millis=200
+            StatusCode.UNAVAILABLE,
+            optional_retry_nanos=200000000,  # .2 seconds
         )
         add_TraceServiceServicer_to_server(
             mock_trace_service,
@@ -386,8 +386,8 @@ def test_retry_info_is_respected(self):
         )
         after = time.time()
         self.assertEqual(mock_trace_service.num_requests, 6)
-        # 200 * 5 = 1 second, plus some wiggle room so the test passes consistently.
-        self.assertTrue(after - before < 1.1)
+        # 1 second plus wiggle room so the test passes consistently.
+        self.assertAlmostEqual(after - before, 1, 1)
 
     def test_retry_not_made_if_would_exceed_timeout(self):
         mock_trace_service = TraceServiceServicerWithExportParams(
@@ -408,34 +408,35 @@ def test_retry_not_made_if_would_exceed_timeout(self):
         # First call at time 0, second at time 1, third at time 3, fourth would exceed timeout.
         self.assertEqual(mock_trace_service.num_requests, 3)
         # There's a +/-20% jitter on each backoff.
-        self.assertTrue(after - before < 3.7)
+        self.assertTrue(2.35 < after - before < 3.65)
 
     def test_timeout_set_correctly(self):
         mock_trace_service = TraceServiceServicerWithExportParams(
-            StatusCode.OK, optional_export_sleep=0.5
+            StatusCode.UNAVAILABLE, optional_export_sleep=0.25
         )
         add_TraceServiceServicer_to_server(
             mock_trace_service,
             self.server,
         )
-        exporter = OTLPSpanExporterForTesting(insecure=True, timeout=0.4)
-        # Should timeout. Deadline should be set to now + timeout.
-        # That is 400 millis from now, and export sleeps for 500 millis.
+        exporter = OTLPSpanExporterForTesting(insecure=True, timeout=1.4)
+        # Should timeout after 1.4 seconds. First attempt takes .25 seconds
+        # Then a 1 second sleep, then deadline exceeded after .15 seconds,
+        # mid way through second call.
         with self.assertLogs(level=WARNING) as warning:
-            self.assertEqual(
-                exporter.export([self.span]),
-                SpanExportResult.FAILURE,
-            )
+            before = time.time()
+            # Eliminate the jitter.
+            with patch("random.uniform", return_value=1):
+                self.assertEqual(
+                    exporter.export([self.span]),
+                    SpanExportResult.FAILURE,
+                )
+            after = time.time()
             self.assertEqual(
                 "Failed to export traces to localhost:4317, error code: StatusCode.DEADLINE_EXCEEDED",
                 warning.records[-1].message,
             )
-            self.assertEqual(mock_trace_service.num_requests, 1)
-        exporter = OTLPSpanExporterForTesting(insecure=True, timeout=0.8)
-        self.assertEqual(
-            exporter.export([self.span]),
-            SpanExportResult.SUCCESS,
-        )
+            self.assertEqual(mock_trace_service.num_requests, 2)
+            self.assertAlmostEqual(after - before, 1.4, 1)
 
     def test_otlp_headers_from_env(self):
         # pylint: disable=protected-access

exporter/opentelemetry-exporter-otlp-proto-http/src/opentelemetry/exporter/otlp/proto/http/_log_exporter/__init__.py

@@ -163,14 +163,15 @@ def export(self, batch: Sequence[LogData]) -> LogExportResult:
 
         serialized_data = encode_logs(batch).SerializeToString()
         deadline_sec = time() + self._timeout
-        for retry_num in range(1, _MAX_RETRYS + 1):
-            backoff_seconds = 2 ** (retry_num - 1) * random.uniform(0.8, 1.2)
+        for retry_num in range(_MAX_RETRYS):
             resp = self._export(serialized_data, deadline_sec - time())
             if resp.ok:
                 return LogExportResult.SUCCESS
+            # multiplying by a random number between .8 and 1.2 introduces a +/20% jitter to each backoff.
+            backoff_seconds = 2**retry_num * random.uniform(0.8, 1.2)
             if (
                 not _is_retryable(resp)
-                or retry_num == _MAX_RETRYS
+                or retry_num + 1 == _MAX_RETRYS
                 or backoff_seconds > (deadline_sec - time())
             ):
                 _logger.error(

exporter/opentelemetry-exporter-otlp-proto-http/src/opentelemetry/exporter/otlp/proto/http/metric_exporter/__init__.py

@@ -215,14 +215,15 @@ def export(
         deadline_sec = time() + (
             timeout_millis / 1e3 if timeout_millis else self._timeout
         )
-        for retry_num in range(1, _MAX_RETRYS + 1):
-            backoff_seconds = 2 ** (retry_num - 1) * random.uniform(0.8, 1.2)
+        for retry_num in range(_MAX_RETRYS):
             resp = self._export(serialized_data, deadline_sec - time())
             if resp.ok:
                 return MetricExportResult.SUCCESS
+            # multiplying by a random number between .8 and 1.2 introduces a +/20% jitter to each backoff.
+            backoff_seconds = 2 ** retry_num * random.uniform(0.8, 1.2)
             if (
                 not _is_retryable(resp)
-                or retry_num == _MAX_RETRYS
+                or retry_num + 1 == _MAX_RETRYS
                 or backoff_seconds > (deadline_sec - time())
             ):
                 _logger.error(

exporter/opentelemetry-exporter-otlp-proto-http/src/opentelemetry/exporter/otlp/proto/http/trace_exporter/__init__.py

@@ -161,14 +161,15 @@ def export(self, spans: Sequence[ReadableSpan]) -> SpanExportResult:
 
         serialized_data = encode_spans(spans).SerializePartialToString()
         deadline_sec = time() + self._timeout
-        for retry_num in range(1, _MAX_RETRYS + 1):
-            backoff_seconds = 2 ** (retry_num - 1) * random.uniform(0.8, 1.2)
+        for retry_num in range(_MAX_RETRYS):
             resp = self._export(serialized_data, deadline_sec - time())
             if resp.ok:
                 return SpanExportResult.SUCCESS
+            # multiplying by a random number between .8 and 1.2 introduces a +/20% jitter to each backoff.
+            backoff_seconds = 2**retry_num * random.uniform(0.8, 1.2)
             if (
                 not _is_retryable(resp)
-                or retry_num == _MAX_RETRYS
+                or retry_num + 1 == _MAX_RETRYS
                 or backoff_seconds > (deadline_sec - time())
             ):
                 _logger.error(

exporter/opentelemetry-exporter-otlp-proto-http/tests/metrics/test_otlp_metrics_exporter.py

@@ -523,7 +523,7 @@ def test_retry_timeout(self, mock_post):
             # First call at time 0, second at time 1, then an early return before the second backoff sleep b/c it would exceed timeout.
             self.assertEqual(mock_post.call_count, 2)
             # There's a +/-20% jitter on each backoff.
-            self.assertTrue(after - before < 1.3)
+            self.assertTrue(0.75 < after - before < 1.25)
             self.assertIn(
                 "Transient error UNAVAILABLE encountered while exporting metrics batch, retrying in",
                 warning.records[0].message,
@@ -540,7 +540,7 @@ def test_retry_timeout(self, mock_post):
         # First call at time 0, second at time 1, third at time 3.
         self.assertEqual(mock_post.call_count, 3)
         # There's a +/-20% jitter on each backoff.
-        self.assertTrue(after - before < 3.7)
+        self.assertTrue(2.35 < after - before < 3.65)
         self.assertIn(
             "Transient error UNAVAILABLE encountered while exporting metrics batch, retrying in",
             warning.records[0].message,
@@ -553,7 +553,7 @@ def test_timeout_set_correctly(self, mock_post):
 
         def export_side_effect(*args, **kwargs):
             # Timeout should be set to something slightly less than 400 milliseconds depending on how much time has passed.
-            self.assertTrue(0.4 - kwargs["timeout"] < 0.0005)
+            self.assertAlmostEqual(0.4, kwargs["timeout"], 2)
             return resp
 
         mock_post.side_effect = export_side_effect

exporter/opentelemetry-exporter-otlp-proto-http/tests/test_proto_log_exporter.py

@@ -369,7 +369,7 @@ def test_retry_timeout(self, mock_post):
             # First call at time 0, second at time 1, then an early return before the second backoff sleep b/c it would exceed timeout.
             self.assertEqual(mock_post.call_count, 2)
             # There's a +/-20% jitter on each backoff.
-            self.assertTrue(after - before < 1.3)
+            self.assertTrue(0.75 < after - before < 1.25)
             self.assertIn(
                 "Transient error UNAVAILABLE encountered while exporting logs batch, retrying in",
                 warning.records[0].message,
@@ -382,7 +382,7 @@ def test_timeout_set_correctly(self, mock_post):
 
         def export_side_effect(*args, **kwargs):
             # Timeout should be set to something slightly less than 400 milliseconds depending on how much time has passed.
-            self.assertTrue(0.4 - kwargs["timeout"] < 0.0005)
+            self.assertAlmostEqual(0.4, kwargs["timeout"], 2)
             return resp
 
         mock_post.side_effect = export_side_effect

exporter/opentelemetry-exporter-otlp-proto-http/tests/test_proto_span_exporter.py

@@ -269,7 +269,7 @@ def test_retry_timeout(self, mock_post):
             # First call at time 0, second at time 1, then an early return before the second backoff sleep b/c it would exceed timeout.
             self.assertEqual(mock_post.call_count, 2)
             # There's a +/-20% jitter on each backoff.
-            self.assertTrue(after - before < 1.3)
+            self.assertTrue(0.75 < after - before < 1.25)
             self.assertIn(
                 "Transient error UNAVAILABLE encountered while exporting span batch, retrying in",
                 warning.records[0].message,
@@ -282,7 +282,7 @@ def test_timeout_set_correctly(self, mock_post):
 
         def export_side_effect(*args, **kwargs):
             # Timeout should be set to something slightly less than 400 milliseconds depending on how much time has passed.
-            self.assertTrue(0.4 - kwargs["timeout"] < 0.0005)
+            self.assertAlmostEqual(0.4, kwargs["timeout"], 2)
             return resp
 
         mock_post.side_effect = export_side_effect