Refactor tests, HTTP exporters a bit

Author: DylanRussell

exporter/opentelemetry-exporter-otlp-proto-common/pyproject.toml

@@ -29,6 +29,7 @@ classifiers = [
 ]
 dependencies = [
   "opentelemetry-proto == 1.34.0.dev",
+  "requests ~= 2.7",
 ]
 
 [project.urls]

exporter/opentelemetry-exporter-otlp-proto-common/src/opentelemetry/exporter/otlp/proto/common/_internal/__init__.py

@@ -27,6 +27,8 @@
     TypeVar,
 )
 
+import requests
+
 from opentelemetry.proto.common.v1.common_pb2 import AnyValue as PB2AnyValue
 from opentelemetry.proto.common.v1.common_pb2 import (
     ArrayValue as PB2ArrayValue,
@@ -108,6 +110,14 @@ def _encode_key_value(
     )
 
 
+def _is_retryable(resp: requests.Response) -> bool:
+    if resp.status_code == 408:
+        return True
+    if resp.status_code >= 500 and resp.status_code <= 599:
+        return True
+    return False
+
+
 def _encode_array(
     array: Sequence[Any], allow_null: bool = False
 ) -> Sequence[PB2AnyValue]:

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

@@ -2,6 +2,7 @@ asgiref==3.7.2
 Deprecated==1.2.14
 importlib-metadata==6.11.0
 iniconfig==2.0.0
+requests == 2.7.0
 packaging==24.0
 pluggy==1.5.0
 protobuf==5.26.1

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

@@ -72,17 +72,18 @@
 from opentelemetry.sdk.trace import ReadableSpan
 from opentelemetry.util.re import parse_env_headers
 
-_JSON_CONFIG = json.dumps(
+# 5 is the maximum allowable attempts allowed by grpc retry policy.
+# This policy results in backoffs of 1s, 2s, 4s, and then 8s after the initial failed attempt,
+# plus or minus a 20% jitter. Timeout set on the RPC call encompasses the retry backoffs AND time spent waiting
+# for a response. DEADLINE_EXCEEDED is returned if all the attempts cannot complete within the
+# timeout, and all fail. A header `grpc-retry-pushback-ms` when set by the server will override
+# and take precedence over this backoff. See https://grpc.io/docs/guides/retry/ for more details.
+_GRPC_RETRY_POLICY = json.dumps(
     {
         "methodConfig": [
             {
                 "name": [dict()],
                 "retryPolicy": {
-                    # 5 is the maximum allowable attempts allowed by grpc retry policy.
-                    # This policy results in backoffs of 1s, 2s, 4s, and then 8s after the initial failed attempt.
-                    # Timeout set on the RPC call encompasses the retry backoffs AND time spent waiting
-                    # for a response. DEADLINE_EXCEEDED is returned if all the attempts cannot complete within the
-                    # timeout, and all fail. See https://grpc.io/docs/guides/retry/ for more details.
                     "maxAttempts": 5,
                     "initialBackoff": "1s",
                     "maxBackoff": "9s",
@@ -213,8 +214,6 @@ class OTLPExporterMixin(
         compression: gRPC compression method to use
     """
 
-    _MAX_RETRY_TIMEOUT = 64
-
     def __init__(
         self,
         endpoint: Optional[str] = None,
@@ -276,7 +275,7 @@ def __init__(
                 self._endpoint,
                 compression=compression,
                 options=[
-                    ("grpc.service_config", _JSON_CONFIG),
+                    ("grpc.service_config", _GRPC_RETRY_POLICY),
                 ],
             )
         else:
@@ -291,7 +290,7 @@ def __init__(
                 credentials,
                 compression=compression,
                 options=[
-                    ("grpc.service_config", _JSON_CONFIG),
+                    ("grpc.service_config", _GRPC_RETRY_POLICY),
                 ],
             )
         self._client = self._stub(self._channel)

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

@@ -20,14 +20,7 @@
 from unittest import TestCase
 from unittest.mock import ANY, Mock, patch
 
-from google.rpc.code_pb2 import (  # pylint: disable=no-name-in-module
-    ALREADY_EXISTS,
-    OK,
-    UNAVAILABLE,
-)
-from google.rpc.status_pb2 import Status  # pylint: disable=no-name-in-module
-from grpc import Compression, server
-from grpc_status import rpc_status
+from grpc import Compression, StatusCode, server
 
 from opentelemetry.exporter.otlp.proto.common.trace_encoder import (
     encode_spans,
@@ -90,25 +83,36 @@ def shutdown(self, timeout_millis=30_000):
 class TraceServiceServicerWithExportParams(TraceServiceServicer):
     def __init__(
         self,
-        export_result: int,
+        export_result: StatusCode,
         optional_export_sleep: Optional[float] = None,
+        optional_first_time_retry_millis: Optional[int] = None,
     ):
         self.export_result = export_result
         self.optional_export_sleep = optional_export_sleep
+        self.optional_first_time_retry_millis = (
+            optional_first_time_retry_millis
+        )
+        self.first_attempt = True
+        self.num_requests = 0
+        self.now = time.time()
 
     # pylint: disable=invalid-name,unused-argument
     def Export(self, request, context):
-        logger.warning("Export Request Received")
+        self.num_requests += 1
         if self.optional_export_sleep:
             time.sleep(self.optional_export_sleep)
-        if self.export_result != OK:
-            context.abort_with_status(
-                rpc_status.to_status(
-                    Status(
-                        code=self.export_result,
+        if self.export_result != StatusCode.OK:
+            if self.optional_first_time_retry_millis and self.first_attempt:
+                self.first_attempt = False
+                context.set_trailing_metadata(
+                    (
+                        (
+                            "grpc-retry-pushback-ms",
+                            str(self.optional_first_time_retry_millis),
+                        ),
                     )
                 )
-            )
+            context.abort(self.export_result, "")
 
         return ExportTraceServiceResponse()
 
@@ -287,7 +291,7 @@ def test_otlp_exporter_otlp_compression_envvar(
 
     def test_shutdown(self):
         add_TraceServiceServicer_to_server(
-            TraceServiceServicerWithExportParams(OK),
+            TraceServiceServicerWithExportParams(StatusCode.OK),
             self.server,
         )
         self.assertEqual(
@@ -305,7 +309,9 @@ def test_shutdown(self):
 
     def test_shutdown_wait_last_export(self):
         add_TraceServiceServicer_to_server(
-            TraceServiceServicerWithExportParams(OK, optional_export_sleep=1),
+            TraceServiceServicerWithExportParams(
+                StatusCode.OK, optional_export_sleep=1
+            ),
             self.server,
         )
 
@@ -324,7 +330,9 @@ def test_shutdown_wait_last_export(self):
 
     def test_shutdown_doesnot_wait_last_export(self):
         add_TraceServiceServicer_to_server(
-            TraceServiceServicerWithExportParams(OK, optional_export_sleep=3),
+            TraceServiceServicerWithExportParams(
+                StatusCode.OK, optional_export_sleep=3
+            ),
             self.server,
         )
 
@@ -346,7 +354,7 @@ def test_export_over_closed_grpc_channel(self):
         # pylint: disable=protected-access
 
         add_TraceServiceServicer_to_server(
-            TraceServiceServicerWithExportParams(OK),
+            TraceServiceServicerWithExportParams(StatusCode.OK),
             self.server,
         )
         self.exporter.export([self.span])
@@ -358,61 +366,56 @@ def test_export_over_closed_grpc_channel(self):
             str(err.exception), "Cannot invoke RPC on closed channel!"
         )
 
+    def test_retry_with_server_pushback(self):
+        mock_trace_service = TraceServiceServicerWithExportParams(
+            StatusCode.UNAVAILABLE, optional_first_time_retry_millis=200
+        )
+        add_TraceServiceServicer_to_server(
+            mock_trace_service,
+            self.server,
+        )
+        exporter = OTLPSpanExporterForTesting(insecure=True, timeout=10)
+        before = time.time()
+        self.assertEqual(
+            exporter.export([self.span]),
+            SpanExportResult.FAILURE,
+        )
+        after = time.time()
+        # We set the `grpc-retry-pushback-ms` header to 200 millis on the first server response.
+        # So we expect the first request at time 0, second at time 0.2,
+        # third at 1.2 (start of backoff policy), fourth at time 3.2, last at time 7.2.
+        self.assertEqual(mock_trace_service.num_requests, 5)
+        # The backoffs have a jitter +- 20%, so we have to put a higher bound than 7.2.
+        self.assertTrue(after - before < 8.8)
+
     def test_retry_timeout(self):
+        mock_trace_service = TraceServiceServicerWithExportParams(
+            StatusCode.UNAVAILABLE
+        )
         add_TraceServiceServicer_to_server(
-            TraceServiceServicerWithExportParams(UNAVAILABLE),
+            mock_trace_service,
             self.server,
         )
         # Set timeout to 1.5 seconds.
         exporter = OTLPSpanExporterForTesting(insecure=True, timeout=1.5)
-        with self.assertLogs(level=WARNING) as warning:
-            self.assertEqual(
-                exporter.export([self.span]),
-                SpanExportResult.FAILURE,
-            )
-            # Our GRPC retry policy starts with a 1 second backoff then doubles.
-            # So we expect just two calls: one at time 0, one at time 1.
-            # The final log is from when export fails.
-            self.assertEqual(len(warning.records), 3)
-            for idx, log in enumerate(warning.records):
-                if idx != 2:
-                    self.assertEqual(
-                        "Export Request Received",
-                        log.message,
-                    )
-                else:
-                    self.assertEqual(
-                        "Failed to export traces to localhost:4317, error code: StatusCode.DEADLINE_EXCEEDED",
-                        log.message,
-                    )
-        with self.assertLogs(level=WARNING) as warning:
-            exporter = OTLPSpanExporterForTesting(insecure=True, timeout=5)
-            # pylint: disable=protected-access
-            self.assertEqual(exporter._timeout, 5)
-            self.assertEqual(
-                exporter.export([self.span]),
-                SpanExportResult.FAILURE,
-            )
-            # We expect 3 calls: time 0, time 1, time 3, but not time 7.
-            # The final log is from when export fails.
-            self.assertEqual(len(warning.records), 4)
-            for idx, log in enumerate(warning.records):
-                if idx != 3:
-                    self.assertEqual(
-                        "Export Request Received",
-                        log.message,
-                    )
-                else:
-                    self.assertEqual(
-                        "Failed to export traces to localhost:4317, error code: StatusCode.DEADLINE_EXCEEDED",
-                        log.message,
-                    )
+        before = time.time()
+        self.assertEqual(
+            exporter.export([self.span]),
+            SpanExportResult.FAILURE,
+        )
+        after = time.time()
+        # Our retry starts with a 1 second backoff then doubles.
+        # So we expect just two calls: one at time 0, one at time 1.
+        self.assertEqual(mock_trace_service.num_requests, 2)
+        # gRPC retry config waits for the timeout (1.5) before cancelling the request.
+        self.assertTrue(after - before < 1.6)
 
     def test_timeout_set_correctly(self):
+        mock_trace_service = TraceServiceServicerWithExportParams(
+            StatusCode.OK, optional_export_sleep=0.5
+        )
         add_TraceServiceServicer_to_server(
-            TraceServiceServicerWithExportParams(
-                OK, optional_export_sleep=0.5
-            ),
+            mock_trace_service,
             self.server,
         )
         exporter = OTLPSpanExporterForTesting(insecure=True, timeout=0.4)
@@ -427,6 +430,7 @@ def test_timeout_set_correctly(self):
                 "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]),
@@ -444,7 +448,9 @@ def test_otlp_headers_from_env(self):
     def test_permanent_failure(self):
         with self.assertLogs(level=WARNING) as warning:
             add_TraceServiceServicer_to_server(
-                TraceServiceServicerWithExportParams(ALREADY_EXISTS),
+                TraceServiceServicerWithExportParams(
+                    StatusCode.ALREADY_EXISTS
+                ),
                 self.server,
             )
             self.assertEqual(