Add tests, changelog entry

Author: DylanRussell

CHANGELOG.md

@@ -7,11 +7,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## Unreleased
 
-- Update OTLP gRPC/HTTP exporters: the export timeout is now inclusive of all retries and backoffs.
-  A +/-20% jitter was added to all backoffs. A pointless 32 second sleep that occurred after all retries
-  had completed/failed was removed.
-  ([#4564](https://github.com/open-telemetry/opentelemetry-python/pull/4564)).
 
+- Update OTLP gRPC/HTTP exporters: calling shutdown will now interrupt exporters that are sleeping
+  before a retry attempt, and cause them to return failure immediately.
+  Update BatchSpan/LogRecodProcessors: shutdown will now complete after 30 seconds of trying to finish
+  exporting any buffered telemetry, instead of continuing to export until all telemetry was exported.
+  ([#4564](https://github.com/open-telemetry/opentelemetry-python/pull/4564)).
 - Update OTLP gRPC/HTTP exporters: the export timeout is now inclusive of all retries and backoffs.
   A +/-20% jitter was added to all backoffs. A pointless 32 second sleep that occurred after all retries
   had completed/failed was removed.

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

@@ -20,7 +20,7 @@
 from collections.abc import Sequence  # noqa: F401
 from logging import getLogger
 from os import environ
-from time import sleep, time
+from time import time
 from typing import (  # noqa: F401
     Any,
     Callable,
@@ -293,54 +293,53 @@ 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.
-        with self._export_lock:
-            deadline_sec = time() + self._timeout
-            for retry_num in range(_MAX_RETRYS):
-                try:
-                    self._client.Export(
-                        request=self._translate_data(data),
-                        metadata=self._headers,
-                        timeout=deadline_sec - time(),
+        deadline_sec = time() + self._timeout
+        for retry_num in range(_MAX_RETRYS):
+            try:
+                self._client.Export(
+                    request=self._translate_data(data),
+                    metadata=self._headers,
+                    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
                     )
-                    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 + 1 == _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(),
-                            exc_info=error.code() == StatusCode.UNKNOWN,
-                        )
-                        return self._result.FAILURE
-                    logger.warning(
-                        "Transient error %s encountered while exporting %s to %s, retrying in %.2fs.",
-                        error.code(),
+                if (
+                    error.code() not in _RETRYABLE_ERROR_CODES
+                    or retry_num + 1 == _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,
-                        backoff_seconds,
+                        error.code(),
+                        exc_info=error.code() == StatusCode.UNKNOWN,
                     )
-                shutdown = self._shutdown_is_occuring.wait(backoff_seconds)
-                if shutdown:
-                    logger.warning("Shutdown in progress, aborting retry.")
-                    break
+                    return self._result.FAILURE
+                logger.warning(
+                    "Transient error %s encountered while exporting %s to %s, retrying in %.2fs.",
+                    error.code(),
+                    self._exporting,
+                    self._endpoint,
+                    backoff_seconds,
+                )
+            shutdown = self._shutdown_is_occuring.wait(backoff_seconds)
+            if shutdown:
+                logger.warning("Shutdown in progress, aborting retry.")
+                break
         # Not possible to reach here but the linter is complaining.
         return self._result.FAILURE
 

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

@@ -312,11 +312,10 @@ def test_shutdown(self):
                 "Exporter already shutdown, ignoring batch",
             )
 
-    def test_shutdown_interrupts_export_sleep(self):
-        # Returns unavailable and asks for a 20 second sleep before retry.
+    def test_shutdown_interrupts_export_retry_backoff(self):
         add_TraceServiceServicer_to_server(
             TraceServiceServicerWithExportParams(
-                StatusCode.UNAVAILABLE
+                StatusCode.UNAVAILABLE,
             ),
             self.server,
         )
@@ -325,24 +324,23 @@ def test_shutdown_interrupts_export_sleep(self):
             target=self.exporter.export, args=([self.span],)
         )
         with self.assertLogs(level=WARNING) as warning:
+            begin_wait = time.time()
             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()
+            time.sleep(0.05)
+            # The code should now be in a 1 second backoff.
             # 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()
+            end_wait = time.time()
             self.assertEqual(export_result, SpanExportResult.FAILURE)
             # Shutdown should have interrupted the sleep.
-            self.assertTrue((end_wait - begin_wait) / 1e9 <  .1)
-            print(warning.records)
+            self.assertTrue(end_wait - begin_wait < 0.2)
             self.assertEqual(
                 warning.records[1].message,
-                "Shutdown in progress, aborting retry."
+                "Shutdown in progress, aborting retry.",
             )
 
     def test_export_over_closed_grpc_channel(self):

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

@@ -15,10 +15,11 @@
 import gzip
 import logging
 import random
+import threading
 import zlib
 from io import BytesIO
 from os import environ
-from time import sleep, time
+from time import time
 from typing import Dict, Optional, Sequence
 
 import requests
@@ -174,7 +175,7 @@ def export(self, batch: Sequence[LogData]) -> LogExportResult:
                 not _is_retryable(resp)
                 or retry_num + 1 == _MAX_RETRYS
                 or backoff_seconds > (deadline_sec - time())
-                 or self._shutdown
+                or self._shutdown
             ):
                 _logger.error(
                     "Failed to export logs batch code: %s, reason: %s",
@@ -205,6 +206,7 @@ def shutdown(self):
         self._shutdown_is_occuring.set()
         self._session.close()
 
+
 def _compression_from_env() -> Compression:
     compression = (
         environ.get(

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

@@ -15,10 +15,11 @@
 import gzip
 import logging
 import random
+import threading
 import zlib
 from io import BytesIO
 from os import environ
-from time import sleep, time
+from time import time
 from typing import (  # noqa: F401
     Any,
     Callable,

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

@@ -15,11 +15,11 @@
 import gzip
 import logging
 import random
-import zlib
 import threading
+import zlib
 from io import BytesIO
 from os import environ
-from time import sleep, time
+from time import time
 from typing import Dict, Optional, Sequence
 
 import requests
@@ -35,9 +35,6 @@
 from opentelemetry.exporter.otlp.proto.http._common import (
     _is_retryable,
 )
-from opentelemetry.exporter.otlp.proto.http._common import (
-    _is_retryable,
-)
 from opentelemetry.sdk.environment_variables import (
     OTEL_EXPORTER_OTLP_CERTIFICATE,
     OTEL_EXPORTER_OTLP_CLIENT_CERTIFICATE,

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

@@ -12,6 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import threading
 import time
 from logging import WARNING
 from os import environ
@@ -541,3 +542,34 @@ def export_side_effect(*args, **kwargs):
         mock_post.side_effect = export_side_effect
         exporter = OTLPMetricExporter(timeout=0.4)
         exporter.export(self.metrics["sum_int"])
+
+    @patch.object(Session, "post")
+    def test_shutdown_interrupts_retry_backoff(self, mock_post):
+        exporter = OTLPMetricExporter(timeout=1.5)
+
+        resp = Response()
+        resp.status_code = 503
+        resp.reason = "UNAVAILABLE"
+        mock_post.return_value = resp
+        thread = threading.Thread(
+            target=exporter.export, args=(self.metrics["sum_int"],)
+        )
+        with self.assertLogs(level=WARNING) as warning:
+            before = time.time()
+            thread.start()
+            # Wait for the first attempt to fail, then enter a 1 second backoff.
+            time.sleep(0.05)
+            # Should cause export to wake up and return.
+            exporter.shutdown()
+            thread.join()
+            after = time.time()
+            self.assertIn(
+                "Transient error UNAVAILABLE encountered while exporting metrics batch, retrying in",
+                warning.records[0].message,
+            )
+            self.assertIn(
+                "Shutdown in progress, aborting retry.",
+                warning.records[1].message,
+            )
+
+            assert after - before < 0.2

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

@@ -14,6 +14,7 @@
 
 # pylint: disable=protected-access
 
+import threading
 import time
 import unittest
 from logging import WARNING
@@ -388,3 +389,34 @@ def export_side_effect(*args, **kwargs):
         mock_post.side_effect = export_side_effect
         exporter = OTLPLogExporter(timeout=0.4)
         exporter.export(self._get_sdk_log_data())
+
+    @patch.object(Session, "post")
+    def test_shutdown_interrupts_retry_backoff(self, mock_post):
+        exporter = OTLPLogExporter(timeout=1.5)
+
+        resp = Response()
+        resp.status_code = 503
+        resp.reason = "UNAVAILABLE"
+        mock_post.return_value = resp
+        thread = threading.Thread(
+            target=exporter.export, args=(self._get_sdk_log_data(),)
+        )
+        with self.assertLogs(level=WARNING) as warning:
+            before = time.time()
+            thread.start()
+            # Wait for the first attempt to fail, then enter a 1 second backoff.
+            time.sleep(0.05)
+            # Should cause export to wake up and return.
+            exporter.shutdown()
+            thread.join()
+            after = time.time()
+            self.assertIn(
+                "Transient error UNAVAILABLE encountered while exporting logs batch, retrying in",
+                warning.records[0].message,
+            )
+            self.assertIn(
+                "Shutdown in progress, aborting retry.",
+                warning.records[1].message,
+            )
+
+            assert after - before < 0.2

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

@@ -12,6 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import threading
 import time
 import unittest
 from logging import WARNING
@@ -288,3 +289,32 @@ def export_side_effect(*args, **kwargs):
         mock_post.side_effect = export_side_effect
         exporter = OTLPSpanExporter(timeout=0.4)
         exporter.export([BASIC_SPAN])
+
+    @patch.object(Session, "post")
+    def test_shutdown_interrupts_retry_backoff(self, mock_post):
+        exporter = OTLPSpanExporter(timeout=1.5)
+
+        resp = Response()
+        resp.status_code = 503
+        resp.reason = "UNAVAILABLE"
+        mock_post.return_value = resp
+        thread = threading.Thread(target=exporter.export, args=([BASIC_SPAN],))
+        with self.assertLogs(level=WARNING) as warning:
+            before = time.time()
+            thread.start()
+            # Wait for the first attempt to fail, then enter a 1 second backoff.
+            time.sleep(0.05)
+            # Should cause export to wake up and return.
+            exporter.shutdown()
+            thread.join()
+            after = time.time()
+            self.assertIn(
+                "Transient error UNAVAILABLE encountered while exporting span batch, retrying in",
+                warning.records[0].message,
+            )
+            self.assertIn(
+                "Shutdown in progress, aborting retry.",
+                warning.records[1].message,
+            )
+
+            assert after - before < 0.2

opentelemetry-sdk/src/opentelemetry/sdk/_shared_internal/__init__.py

@@ -182,7 +182,7 @@ def emit(self, data: Telemetry) -> None:
         self._queue.appendleft(data)
         if len(self._queue) >= self._max_export_batch_size:
             self._worker_awaken.set()
- 
+
     def shutdown(self, timeout_millis: int = 30000):
         if self._shutdown:
             return
@@ -220,4 +220,4 @@ def force_flush(self, timeout_millis: Optional[int] = None) -> bool:
             return False
         # Blocking call to export.
         self._export(BatchExportStrategy.EXPORT_ALL)
-        return True
+        return True