Fix flaky jitter tests and add explicit timeout deadline check in OTLPAwsLogsExporter (#501)

*Description of changes:*
Unit tests would occasionally fail because of a transitive timing issue
using Python's `time()` library, see:

https://github.com/aws-observability/aws-otel-python-instrumentation/actions/runs/18681954286/job/53290674826

- Add timeout check before each export attempt to fail fast when
deadline is exceeded and fixes flaky failing retry test by mocking time
to prevent unexpected timeout errors in the test.

- Updates `tox.ini` to point the upstream repos to the one currently in
`pyproject.toml`

By submitting this pull request, I confirm that you can use, modify,
copy, and redistribute this contribution, under the terms of your
choice.

---------

Co-authored-by: ADOT Patch workflow <[email protected]>

Author: liustve

CHANGELOG.md

@@ -13,3 +13,5 @@ If your change does not need a CHANGELOG entry, add the "skip changelog" label t
 ## Unreleased
 - Add botocore instrumentation extension for Bedrock AgentCore services with span attributes
   ([#490](https://github.com/aws-observability/aws-otel-python-instrumentation/pull/490))
+- Fix timeout handling for exceeded deadline in retry logic in OTLPAwsLogsExporter
+  ([#501](https://github.com/aws-observability/aws-otel-python-instrumentation/pull/501))

aws-opentelemetry-distro/src/amazon/opentelemetry/distro/exporter/otlp/aws/logs/otlp_aws_logs_exporter.py

@@ -104,7 +104,14 @@ def export(self, batch: Sequence[LogData]) -> LogExportResult:
         # 3) Deadline timeout will be exceeded
         # 4) Non-retryable errors (4xx except 429) immediately exit the loop
         while True:
-            resp = self._send(data, deadline_sec - time())
+            # Check timeout before each attempt. The retry logic below also checks
+            # timeout, but this provides an additional safeguard against flaky timing edge cases.
+            remaining_time = deadline_sec - time()
+            if remaining_time <= 0:
+                _logger.error("Timeout deadline exceeded before attempt.")
+                return LogExportResult.FAILURE
+
+            resp = self._send(data, remaining_time)
 
             if resp.ok:
                 return LogExportResult.SUCCESS
@@ -172,8 +179,10 @@ def _get_retry_delay_sec(self, headers: CaseInsensitiveDict, retry_num: int) ->
         """
         # Check for Retry-After header first, then use exponential backoff with jitter
         retry_after_delay = self._parse_retryable_header(headers.get(self._RETRY_AFTER_HEADER))
+
         if retry_after_delay > -1:
             return retry_after_delay
+
         # multiplying by a random number between .8 and 1.2 introduces a +/-20% jitter to each backoff.
         return 2**retry_num * random.uniform(0.8, 1.2)
 

aws-opentelemetry-distro/tests/amazon/opentelemetry/distro/exporter/otlp/aws/logs/test_otlp_aws_logs_exporter.py

@@ -18,29 +18,31 @@
 
 class TestOTLPAwsLogsExporter(TestCase):
     _ENDPOINT = "https://logs.us-west-2.amazonaws.com/v1/logs"
-    good_response = requests.Response()
-    good_response.status_code = 200
 
-    non_retryable_response = requests.Response()
-    non_retryable_response.status_code = 404
+    def setUp(self):
+        self.logs = self.generate_test_log_data()
+        self.exporter = OTLPAwsLogExporter(session=get_aws_session(), aws_region="us-east-1", endpoint=self._ENDPOINT)
 
-    retryable_response_no_header = requests.Response()
-    retryable_response_no_header.status_code = 429
+        self.good_response = requests.Response()
+        self.good_response.status_code = 200
 
-    retryable_response_header = requests.Response()
-    retryable_response_header.headers = CaseInsensitiveDict({"Retry-After": "10"})
-    retryable_response_header.status_code = 503
+        self.non_retryable_response = requests.Response()
+        self.non_retryable_response.status_code = 404
 
-    retryable_response_bad_header = requests.Response()
-    retryable_response_bad_header.headers = CaseInsensitiveDict({"Retry-After": "-12"})
-    retryable_response_bad_header.status_code = 503
+        self.retryable_response_no_header = requests.Response()
+        self.retryable_response_no_header.status_code = 429
 
-    def setUp(self):
-        self.logs = self.generate_test_log_data()
-        self.exporter = OTLPAwsLogExporter(session=get_aws_session(), aws_region="us-east-1", endpoint=self._ENDPOINT)
+        self.retryable_response_header = requests.Response()
+        self.retryable_response_header.headers = CaseInsensitiveDict({"Retry-After": "10"})
+        self.retryable_response_header.status_code = 503
 
-    @patch("requests.Session.post", return_value=good_response)
+        self.retryable_response_bad_header = requests.Response()
+        self.retryable_response_bad_header.headers = CaseInsensitiveDict({"Retry-After": "-12"})
+        self.retryable_response_bad_header.status_code = 503
+
+    @patch("requests.Session.post")
     def test_export_success(self, mock_request):
+        mock_request.return_value = self.good_response
         """Tests that the exporter always compresses the serialized logs with gzip before exporting."""
         result = self.exporter.export(self.logs)
 
@@ -57,17 +59,19 @@ def test_export_success(self, mock_request):
         self.assertTrue(len(data) >= 10)
         self.assertEqual(data[0:2], b"\x1f\x8b")
 
-    @patch("requests.Session.post", return_value=good_response)
+    @patch("requests.Session.post")
     def test_should_not_export_if_shutdown(self, mock_request):
+        mock_request.return_value = self.good_response
         """Tests that no export request is made if the exporter is shutdown."""
         self.exporter.shutdown()
         result = self.exporter.export(self.logs)
 
         mock_request.assert_not_called()
         self.assertEqual(result, LogExportResult.FAILURE)
 
-    @patch("requests.Session.post", return_value=non_retryable_response)
+    @patch("requests.Session.post")
     def test_should_not_export_again_if_not_retryable(self, mock_request):
+        mock_request.return_value = self.non_retryable_response
         """Tests that only one export request is made if the response status code is non-retryable."""
         result = self.exporter.export(self.logs)
         mock_request.assert_called_once()
@@ -78,12 +82,16 @@ def test_should_not_export_again_if_not_retryable(self, mock_request):
         "amazon.opentelemetry.distro.exporter.otlp.aws.logs.otlp_aws_logs_exporter.Event.wait",
         side_effect=lambda x: False,
     )
-    @patch("requests.Session.post", return_value=retryable_response_no_header)
+    @patch("requests.Session.post")
     def test_should_export_again_with_backoff_if_retryable_and_no_retry_after_header(self, mock_request, mock_wait):
+        mock_request.return_value = self.retryable_response_no_header
         """Tests that multiple export requests are made with exponential delay if the response status code is retryable.
         But there is no Retry-After header."""
         self.exporter._timeout = 10000  # Large timeout to avoid early exit
-        result = self.exporter.export(self.logs)
+
+        with patch("amazon.opentelemetry.distro.exporter.otlp.aws.logs.otlp_aws_logs_exporter.time") as mock_time:
+            mock_time.return_value = 0  # Keep time constant to avoid timeout issues
+            result = self.exporter.export(self.logs)
 
         self.assertEqual(mock_wait.call_count, _MAX_RETRYS - 1)
 
@@ -103,15 +111,21 @@ def test_should_export_again_with_backoff_if_retryable_and_no_retry_after_header
         "amazon.opentelemetry.distro.exporter.otlp.aws.logs.otlp_aws_logs_exporter.Event.wait",
         side_effect=lambda x: False,
     )
-    @patch(
-        "requests.Session.post",
-        side_effect=[retryable_response_header, retryable_response_header, retryable_response_header, good_response],
-    )
+    @patch("requests.Session.post")
     def test_should_export_again_with_server_delay_if_retryable_and_retry_after_header(self, mock_request, mock_wait):
+        mock_request.side_effect = [
+            self.retryable_response_header,
+            self.retryable_response_header,
+            self.retryable_response_header,
+            self.good_response,
+        ]
         """Tests that multiple export requests are made with the server's suggested
         delay if the response status code is retryable and there is a Retry-After header."""
         self.exporter._timeout = 10000  # Large timeout to avoid early exit
-        result = self.exporter.export(self.logs)
+
+        with patch("amazon.opentelemetry.distro.exporter.otlp.aws.logs.otlp_aws_logs_exporter.time") as mock_time:
+            mock_time.return_value = 0  # Keep time constant to avoid timeout issues
+            result = self.exporter.export(self.logs)
 
         delays = mock_wait.call_args_list
 
@@ -126,22 +140,23 @@ def test_should_export_again_with_server_delay_if_retryable_and_retry_after_head
         "amazon.opentelemetry.distro.exporter.otlp.aws.logs.otlp_aws_logs_exporter.Event.wait",
         side_effect=lambda x: False,
     )
-    @patch(
-        "requests.Session.post",
-        side_effect=[
-            retryable_response_bad_header,
-            retryable_response_bad_header,
-            retryable_response_bad_header,
-            good_response,
-        ],
-    )
+    @patch("requests.Session.post")
     def test_should_export_again_with_backoff_delay_if_retryable_and_bad_retry_after_header(
         self, mock_request, mock_wait
     ):
+        mock_request.side_effect = [
+            self.retryable_response_bad_header,
+            self.retryable_response_bad_header,
+            self.retryable_response_bad_header,
+            self.good_response,
+        ]
         """Tests that multiple export requests are made with exponential delay if the response status code is retryable.
         but the Retry-After header is invalid or malformed."""
         self.exporter._timeout = 10000  # Large timeout to avoid early exit
-        result = self.exporter.export(self.logs)
+
+        with patch("amazon.opentelemetry.distro.exporter.otlp.aws.logs.otlp_aws_logs_exporter.time") as mock_time:
+            mock_time.return_value = 0  # Keep time constant to avoid timeout issues
+            result = self.exporter.export(self.logs)
 
         delays = mock_wait.call_args_list
 
@@ -156,8 +171,9 @@ def test_should_export_again_with_backoff_delay_if_retryable_and_bad_retry_after
         self.assertEqual(mock_request.call_count, 4)
         self.assertEqual(result, LogExportResult.SUCCESS)
 
-    @patch("requests.Session.post", side_effect=[requests.exceptions.ConnectionError(), good_response])
+    @patch("requests.Session.post")
     def test_export_connection_error_retry(self, mock_request):
+        mock_request.side_effect = [requests.exceptions.ConnectionError(), self.good_response]
         """Tests that the exporter retries on ConnectionError."""
         result = self.exporter.export(self.logs)
 
@@ -168,8 +184,9 @@ def test_export_connection_error_retry(self, mock_request):
         "amazon.opentelemetry.distro.exporter.otlp.aws.logs.otlp_aws_logs_exporter.Event.wait",
         side_effect=lambda x: False,
     )
-    @patch("requests.Session.post", return_value=retryable_response_no_header)
+    @patch("requests.Session.post")
     def test_should_stop_retrying_when_deadline_exceeded(self, mock_request, mock_wait):
+        mock_request.return_value = self.retryable_response_no_header
         """Tests that the exporter stops retrying when the deadline is exceeded."""
         self.exporter._timeout = 5  # Short timeout to trigger deadline check
 
@@ -191,19 +208,23 @@ def test_should_stop_retrying_when_deadline_exceeded(self, mock_request, mock_wa
         "amazon.opentelemetry.distro.exporter.otlp.aws.logs.otlp_aws_logs_exporter.Event.wait",
         side_effect=lambda x: True,
     )
-    @patch("requests.Session.post", return_value=retryable_response_no_header)
+    @patch("requests.Session.post")
     def test_export_interrupted_by_shutdown(self, mock_request, mock_wait):
+        mock_request.return_value = self.retryable_response_no_header
         """Tests that export can be interrupted by shutdown during retry wait."""
         self.exporter._timeout = 10000
 
-        result = self.exporter.export(self.logs)
+        with patch("amazon.opentelemetry.distro.exporter.otlp.aws.logs.otlp_aws_logs_exporter.time") as mock_time:
+            mock_time.return_value = 0  # Keep time constant to avoid timeout issues
+            result = self.exporter.export(self.logs)
 
         # Should make one request, then get interrupted during retry wait
         self.assertEqual(mock_request.call_count, 1)
         self.assertEqual(result, LogExportResult.FAILURE)
 
-    @patch("requests.Session.post", return_value=good_response)
+    @patch("requests.Session.post")
     def test_export_with_log_group_and_stream_headers(self, mock_request):
+        mock_request.return_value = self.good_response
         """Tests that log_group and log_stream are properly set as headers when provided."""
         log_group = "test-log-group"
         log_stream = "test-log-stream"
@@ -228,6 +249,17 @@ def test_export_with_log_group_and_stream_headers(self, mock_request):
         self.assertEqual(session_headers["x-aws-log-group"], log_group)
         self.assertEqual(session_headers["x-aws-log-stream"], log_stream)
 
+    @patch("requests.Session.post")
+    def test_should_not_make_request_when_deadline_exceeded_immediately(self, mock_request):
+        """Tests that no HTTP request is made when deadline is already exceeded before first attempt."""
+        self.exporter._timeout = 0  # Deadline already exceeded
+
+        result = self.exporter.export(self.logs)
+
+        # Should not make any HTTP requests
+        mock_request.assert_not_called()
+        self.assertEqual(result, LogExportResult.FAILURE)
+
     @staticmethod
     def generate_test_log_data(count=5):
         logs = []

tox.ini

@@ -22,8 +22,8 @@ deps =
 
 setenv =
   ; TODO: The two repos branches need manual updated over time, need to figure out a more sustainable solution.
-  CORE_REPO="git+https://github.com/open-telemetry/opentelemetry-python.git@release/v1.25.x-0.46bx"
-  CONTRIB_REPO="git+https://github.com/open-telemetry/opentelemetry-python-contrib.git@release/v1.25.x-0.46bx"
+  CORE_REPO="git+https://github.com/open-telemetry/opentelemetry-python.git@release/v1.33.x-0.54bx"
+  CONTRIB_REPO="git+https://github.com/open-telemetry/opentelemetry-python-contrib.git@release/v1.33.x-0.54bx"
 
 changedir =
   test-aws-opentelemetry-distro: aws-opentelemetry-distro/tests