Make many changes

Author: DylanRussell

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

@@ -273,7 +273,7 @@ def __init__(
             )
         self._client = self._stub(self._channel)
 
-        self._export_lock = threading.Lock()
+        self._shutdown_is_occuring = threading.Event()
         self._shutdown = False
 
     @abstractmethod
@@ -294,60 +294,60 @@ def _export(
         # 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
+        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
+            )
+            try:
+                self._client.Export(
+                    request=self._translate_data(data),
+                    metadata=self._headers,
+                    timeout=self._timeout,
                 )
-                try:
-                    self._client.Export(
-                        request=self._translate_data(data),
-                        metadata=self._headers,
-                        timeout=self._timeout,
-                    )
-                    return self._result.SUCCESS
-                except RpcError as error:
-                    retry_info_bin = dict(error.trailing_metadata()).get(
-                        "google.rpc.retryinfo-bin"
+                return self._result.SUCCESS
+            except RpcError as error:
+                retry_info_bin = dict(error.trailing_metadata()).get(
+                    "google.rpc.retryinfo-bin"
+                )
+                if retry_info_bin is not None:
+                    retry_info.ParseFromString(retry_info_bin)
+                    backoff_seconds = (
+                        retry_info.retry_delay.seconds
+                        + retry_info.retry_delay.nanos / 1.0e9
                     )
-                    if retry_info_bin is not None:
-                        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 backoff_seconds > (deadline_sec - time())
-                    ):
-                        logger.error(
-                            "Failed to export %s to %s, error code: %s",
-                            self._exporting,
-                            self._endpoint,
-                            error.code(),
-                            exc_info=error.code() == StatusCode.UNKNOWN,
-                        )
-                        return self._result.FAILURE
-                    logger.warning(
-                        "Transient error %s encountered while exporting logs batch, retrying in %.2fs.",
+                if (
+                    error.code() not in _RETRYABLE_ERROR_CODES
+                    or retry_num == _MAX_RETRYS
+                    or backoff_seconds > (deadline_sec - time())
+                    or self._shutdown
+                ):
+                    logger.error(
+                        "Failed to export %s to %s, error code: %s",
+                        self._exporting,
+                        self._endpoint,
                         error.code(),
-                        backoff_seconds,
+                        exc_info=error.code() == StatusCode.UNKNOWN,
                     )
-                    sleep(backoff_seconds)
-        # Not possible to reach here but the linter is complaining.
+                    return self._result.FAILURE
+                logger.warning(
+                    "Transient error %s encountered while exporting logs batch, retrying in %.2fs.",
+                    error.code(),
+                    backoff_seconds,
+                )
+                shutdown = self._shutdown_is_occuring.wait(backoff_seconds)
+                if shutdown:
+                    logger.warning("Shutdown in progress, aborting retry.")
+                    break
         return self._result.FAILURE
 
     def shutdown(self, timeout_millis: float = 30_000, **kwargs) -> None:
         if self._shutdown:
             logger.warning("Exporter already shutdown, ignoring call")
             return
-        # wait for the last export if any
-        self._export_lock.acquire(timeout=timeout_millis / 1e3)
+        self._shutdown_is_occuring.set()
         self._shutdown = True
         self._channel.close()
-        self._export_lock.release()
 
     @property
     @abstractmethod

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

@@ -311,48 +311,38 @@ def test_shutdown(self):
                 "Exporter already shutdown, ignoring batch",
             )
 
-    def test_shutdown_wait_last_export(self):
+    def test_shutdown_interrupts_export_sleep(self):
+        # Returns unavailable and asks for a 20 second sleep before retry.
         add_TraceServiceServicer_to_server(
             TraceServiceServicerWithExportParams(
-                StatusCode.OK, optional_export_sleep=1
+                StatusCode.UNAVAILABLE
             ),
             self.server,
         )
 
         export_thread = ThreadWithReturnValue(
             target=self.exporter.export, args=([self.span],)
         )
-        export_thread.start()
-        # Wait a bit for exporter to get lock and make export call.
-        time.sleep(0.25)
-        # pylint: disable=protected-access
-        self.assertTrue(self.exporter._export_lock.locked())
-        self.exporter.shutdown(timeout_millis=3000)
-        # pylint: disable=protected-access
-        self.assertTrue(self.exporter._shutdown)
-        self.assertEqual(export_thread.join(), SpanExportResult.SUCCESS)
-
-    def test_shutdown_doesnot_wait_last_export(self):
-        add_TraceServiceServicer_to_server(
-            TraceServiceServicerWithExportParams(
-                StatusCode.OK, optional_export_sleep=3
-            ),
-            self.server,
-        )
-
-        export_thread = ThreadWithReturnValue(
-            target=self.exporter.export, args=([self.span],)
-        )
-        export_thread.start()
-        # Wait for exporter to get lock and make export call.
-        time.sleep(0.25)
-        # pylint: disable=protected-access
-        self.assertTrue(self.exporter._export_lock.locked())
-        # Set to 1 seconds, so the 3 second server-side delay will not be reached.
-        self.exporter.shutdown(timeout_millis=1000)
-        # pylint: disable=protected-access
-        self.assertTrue(self.exporter._shutdown)
-        self.assertEqual(export_thread.join(), None)
+        with self.assertLogs(level=WARNING) as warning:
+            export_thread.start()
+            # Wait a bit for export to fail and the backoff sleep to start
+            time.sleep(.1)
+            # The code should now be in a sleep that's between .8 and 1.2 seconds.
+            begin_wait = time.time_ns()
+            # pylint: disable=protected-access
+            self.assertFalse(self.exporter._shutdown_is_occuring.is_set())
+            self.exporter.shutdown()
+            self.assertTrue(self.exporter._shutdown_is_occuring.is_set())
+            export_result = export_thread.join()
+            end_wait = time.time_ns()
+            self.assertEqual(export_result, SpanExportResult.FAILURE)
+            # Shutdown should have interrupted the sleep.
+            self.assertTrue((end_wait - begin_wait) / 1e9 <  .1)
+            print(warning.records)
+            self.assertEqual(
+                warning.records[1].message,
+                "Shutdown in progress, aborting retry."
+            )
 
     def test_export_over_closed_grpc_channel(self):
         # pylint: disable=protected-access

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

@@ -14,6 +14,7 @@
 
 import gzip
 import logging
+import threading
 import random
 import zlib
 from io import BytesIO
@@ -77,6 +78,7 @@ def __init__(
         compression: Optional[Compression] = None,
         session: Optional[requests.Session] = None,
     ):
+        self._shutdown_is_occuring = threading.Event()
         self._endpoint = endpoint or environ.get(
             OTEL_EXPORTER_OTLP_LOGS_ENDPOINT,
             _append_logs_path(
@@ -164,14 +166,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)
             resp = self._export(serialized_data, deadline_sec - time())
             if resp.ok:
                 return LogExportResult.SUCCESS
+            backoff_seconds = 2 ** (retry_num - 1) * random.uniform(0.8, 1.2)
             if (
                 not _is_retryable(resp)
                 or retry_num == _MAX_RETRYS
                 or backoff_seconds > (deadline_sec - time())
+                or self._shutdown
             ):
                 _logger.error(
                     "Failed to export logs batch code: %s, reason: %s",
@@ -184,8 +187,10 @@ def export(self, batch: Sequence[LogData]) -> LogExportResult:
                 resp.reason,
                 backoff_seconds,
             )
-            sleep(backoff_seconds)
-        # Not possible to reach here but the linter is complaining.
+            shutdown = self._shutdown_is_occuring.wait(backoff_seconds)
+            if shutdown:
+                _logger.warning("Shutdown in progress, aborting retry.")
+                break
         return LogExportResult.FAILURE
 
     def force_flush(self, timeout_millis: float = 10_000) -> bool:
@@ -196,6 +201,7 @@ def shutdown(self):
         if self._shutdown:
             _logger.warning("Exporter already shutdown, ignoring call")
             return
+        self._shutdown_is_occuring.set()
         self._session.close()
         self._shutdown = True
 

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

@@ -14,6 +14,7 @@
 
 import gzip
 import logging
+import threading
 import random
 import zlib
 from io import BytesIO
@@ -120,6 +121,7 @@ def __init__(
         | None = None,
         preferred_aggregation: dict[type, Aggregation] | None = None,
     ):
+        self._shutdown_is_occuring = threading.Event()
         self._endpoint = endpoint or environ.get(
             OTEL_EXPORTER_OTLP_METRICS_ENDPOINT,
             _append_metrics_path(
@@ -224,6 +226,7 @@ def export(
                 not _is_retryable(resp)
                 or retry_num == _MAX_RETRYS
                 or backoff_seconds > (deadline_sec - time())
+                or self._shutdown
             ):
                 _logger.error(
                     "Failed to export metrics batch code: %s, reason: %s",
@@ -236,14 +239,17 @@ def export(
                 resp.reason,
                 backoff_seconds,
             )
-            sleep(backoff_seconds)
-        # Not possible to reach here but the linter is complaining.
+            shutdown = self._shutdown_is_occuring.wait(backoff_seconds)
+            if shutdown:
+                _logger.warning("Shutdown in progress, aborting retry.")
+                break
         return MetricExportResult.FAILURE
 
     def shutdown(self, timeout_millis: float = 30_000, **kwargs) -> None:
         if self._shutdown:
             _logger.warning("Exporter already shutdown, ignoring call")
             return
+        self._shutdown_is_occuring.set()
         self._session.close()
         self._shutdown = True
 

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

@@ -16,6 +16,7 @@
 import logging
 import random
 import zlib
+import threading
 from io import BytesIO
 from os import environ
 from time import sleep, time
@@ -76,6 +77,7 @@ def __init__(
         compression: Optional[Compression] = None,
         session: Optional[requests.Session] = None,
     ):
+        self._shutdown_is_occuring = threading.Event()
         self._endpoint = endpoint or environ.get(
             OTEL_EXPORTER_OTLP_TRACES_ENDPOINT,
             _append_trace_path(
@@ -162,14 +164,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)
             resp = self._export(serialized_data, deadline_sec - time())
             if resp.ok:
                 return SpanExportResult.SUCCESS
+            backoff_seconds = 2 ** (retry_num - 1) * random.uniform(0.8, 1.2)
             if (
                 not _is_retryable(resp)
                 or retry_num == _MAX_RETRYS
                 or backoff_seconds > (deadline_sec - time())
+                or self._shutdown
             ):
                 _logger.error(
                     "Failed to export span batch code: %s, reason: %s",
@@ -182,14 +185,17 @@ def export(self, spans: Sequence[ReadableSpan]) -> SpanExportResult:
                 resp.reason,
                 backoff_seconds,
             )
-            sleep(backoff_seconds)
-        # Not possible to reach here but the linter is complaining.
+            shutdown = self._shutdown_is_occuring.wait(backoff_seconds)
+            if shutdown:
+                _logger.warning("Shutdown in progress, aborting retry.")
+                break
         return SpanExportResult.FAILURE
 
     def shutdown(self):
         if self._shutdown:
             _logger.warning("Exporter already shutdown, ignoring call")
             return
+        self._shutdown_is_occuring.set()
         self._session.close()
         self._shutdown = True