fix sleep unit tests + renaming aws batch log processor

Author: liustve

aws-opentelemetry-distro/src/amazon/opentelemetry/distro/aws_opentelemetry_configurator.py

@@ -23,7 +23,7 @@
     AwsMetricAttributesSpanExporterBuilder,
 )
 from amazon.opentelemetry.distro.aws_span_metrics_processor_builder import AwsSpanMetricsProcessorBuilder
-from amazon.opentelemetry.distro.exporter.otlp.aws.logs.aws_batch_log_record_processor import (
+from amazon.opentelemetry.distro.exporter.otlp.aws.logs._aws_cw_otlp_batch_log_record_processor import (
     AwsCloudWatchOtlpBatchLogRecordProcessor,
 )
 from amazon.opentelemetry.distro.exporter.otlp.aws.logs.otlp_aws_logs_exporter import OTLPAwsLogExporter

aws-opentelemetry-distro/src/amazon/opentelemetry/distro/exporter/otlp/aws/logs/aws_batch_log_record_processor.py renamed to aws-opentelemetry-distro/src/amazon/opentelemetry/distro/exporter/otlp/aws/logs/_aws_cw_otlp_batch_log_record_processor.py

@@ -1,6 +1,6 @@
-# Copyright The OpenTelemetry Authors
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
 # SPDX-License-Identifier: Apache-2.0
-# Modifications Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+# Modifications Copyright The OpenTelemetry Authors. Licensed under the Apache License 2.0 License.
 
 import logging
 from typing import Mapping, Optional, Sequence, cast

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

@@ -1,6 +1,6 @@
-# Copyright The OpenTelemetry Authors
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
 # SPDX-License-Identifier: Apache-2.0
-# Modifications Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+# Modifications Copyright The OpenTelemetry Authors. Licensed under the Apache License 2.0 License.
 
 import gzip
 import logging
@@ -120,7 +120,7 @@ def export(self, batch: Sequence[LogData]) -> LogExportResult:
             if self._shutdown_event.wait(backoff_seconds):
                 _logger.info("Export interrupted by shutdown")
                 return LogExportResult.FAILURE
-            
+
             retry_num += 1
 
     def shutdown(self) -> None:

aws-opentelemetry-distro/tests/amazon/opentelemetry/distro/exporter/otlp/aws/logs/test_aws_batch_log_record_processor.py renamed to aws-opentelemetry-distro/tests/amazon/opentelemetry/distro/exporter/otlp/aws/logs/test_aws_cw_otlp_batch_log_record_processor.py

@@ -5,7 +5,7 @@
 from typing import List
 from unittest.mock import MagicMock, patch
 
-from amazon.opentelemetry.distro.exporter.otlp.aws.logs.aws_batch_log_record_processor import (
+from amazon.opentelemetry.distro.exporter.otlp.aws.logs._aws_cw_otlp_batch_log_record_processor import (
     AwsCloudWatchOtlpBatchLogRecordProcessor,
     BatchLogExportStrategy,
 )
@@ -136,11 +136,11 @@ def test_process_log_data_with_cycle(self):
         self.assertEqual(actual_size, expected_size)
 
     @patch(
-        "amazon.opentelemetry.distro.exporter.otlp.aws.logs.aws_batch_log_record_processor.attach",
+        "amazon.opentelemetry.distro.exporter.otlp.aws.logs._aws_cw_otlp_batch_log_record_processor.attach",
         return_value=MagicMock(),
     )
-    @patch("amazon.opentelemetry.distro.exporter.otlp.aws.logs.aws_batch_log_record_processor.detach")
-    @patch("amazon.opentelemetry.distro.exporter.otlp.aws.logs.aws_batch_log_record_processor.set_value")
+    @patch("amazon.opentelemetry.distro.exporter.otlp.aws.logs._aws_cw_otlp_batch_log_record_processor.detach")
+    @patch("amazon.opentelemetry.distro.exporter.otlp.aws.logs._aws_cw_otlp_batch_log_record_processor.set_value")
     def test_export_single_batch_under_size_limit(self, _, __, ___):
         """Tests that export is only called once if a single batch is under the size limit"""
         log_count = 10
@@ -163,11 +163,11 @@ def test_export_single_batch_under_size_limit(self, _, __, ___):
         self.mock_exporter.export.assert_called_once()
 
     @patch(
-        "amazon.opentelemetry.distro.exporter.otlp.aws.logs.aws_batch_log_record_processor.attach",
+        "amazon.opentelemetry.distro.exporter.otlp.aws.logs._aws_cw_otlp_batch_log_record_processor.attach",
         return_value=MagicMock(),
     )
-    @patch("amazon.opentelemetry.distro.exporter.otlp.aws.logs.aws_batch_log_record_processor.detach")
-    @patch("amazon.opentelemetry.distro.exporter.otlp.aws.logs.aws_batch_log_record_processor.set_value")
+    @patch("amazon.opentelemetry.distro.exporter.otlp.aws.logs._aws_cw_otlp_batch_log_record_processor.detach")
+    @patch("amazon.opentelemetry.distro.exporter.otlp.aws.logs._aws_cw_otlp_batch_log_record_processor.set_value")
     def test_export_single_batch_all_logs_over_size_limit(self, _, __, ___):
         """Should make multiple export calls of batch size 1 to export logs of size > 1 MB."""
 
@@ -188,11 +188,11 @@ def test_export_single_batch_all_logs_over_size_limit(self, _, __, ___):
             self.assertEqual(len(batch[0]), 1)
 
     @patch(
-        "amazon.opentelemetry.distro.exporter.otlp.aws.logs.aws_batch_log_record_processor.attach",
+        "amazon.opentelemetry.distro.exporter.otlp.aws.logs._aws_cw_otlp_batch_log_record_processor.attach",
         return_value=MagicMock(),
     )
-    @patch("amazon.opentelemetry.distro.exporter.otlp.aws.logs.aws_batch_log_record_processor.detach")
-    @patch("amazon.opentelemetry.distro.exporter.otlp.aws.logs.aws_batch_log_record_processor.set_value")
+    @patch("amazon.opentelemetry.distro.exporter.otlp.aws.logs._aws_cw_otlp_batch_log_record_processor.detach")
+    @patch("amazon.opentelemetry.distro.exporter.otlp.aws.logs._aws_cw_otlp_batch_log_record_processor.set_value")
     def test_export_single_batch_some_logs_over_size_limit(self, _, __, ___):
         """Should make calls to export smaller sub-batch logs"""
         large_log_body = "X" * (self.processor._MAX_LOG_REQUEST_BYTE_SIZE + 1)

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

@@ -73,7 +73,10 @@ def test_should_not_export_again_if_not_retryable(self, mock_request):
 
         self.assertEqual(result, LogExportResult.FAILURE)
 
-    @patch("threading.Event.wait", side_effect=lambda x: False)
+    @patch(
+        "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)
     def test_should_export_again_with_backoff_if_retryable_and_no_retry_after_header(self, mock_request, mock_wait):
         """Tests that multiple export requests are made with exponential delay if the response status code is retryable.
@@ -95,7 +98,10 @@ def test_should_export_again_with_backoff_if_retryable_and_no_retry_after_header
         self.assertEqual(mock_request.call_count, _MAX_RETRYS)
         self.assertEqual(result, LogExportResult.FAILURE)
 
-    @patch("threading.Event.wait", side_effect=lambda x: False)
+    @patch(
+        "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],
@@ -105,6 +111,7 @@ def test_should_export_again_with_server_delay_if_retryable_and_retry_after_head
         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)
+
         delays = mock_wait.call_args_list
 
         for delay in delays:
@@ -114,7 +121,10 @@ def test_should_export_again_with_server_delay_if_retryable_and_retry_after_head
         self.assertEqual(mock_request.call_count, 4)
         self.assertEqual(result, LogExportResult.SUCCESS)
 
-    @patch("threading.Event.wait", side_effect=lambda x: False)
+    @patch(
+        "amazon.opentelemetry.distro.exporter.otlp.aws.logs.otlp_aws_logs_exporter.Event.wait",
+        side_effect=lambda x: False,
+    )
     @patch(
         "requests.Session.post",
         side_effect=[
@@ -131,6 +141,7 @@ def test_should_export_again_with_backoff_delay_if_retryable_and_bad_retry_after
         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)
+
         delays = mock_wait.call_args_list
 
         for index, delay in enumerate(delays):
@@ -152,7 +163,10 @@ def test_export_connection_error_retry(self, mock_request):
         self.assertEqual(mock_request.call_count, 2)
         self.assertEqual(result, LogExportResult.SUCCESS)
 
-    @patch("threading.Event.wait", side_effect=lambda x: False)
+    @patch(
+        "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)
     def test_should_stop_retrying_when_deadline_exceeded(self, mock_request, mock_wait):
         """Tests that the exporter stops retrying when the deadline is exceeded."""
@@ -172,23 +186,21 @@ def test_should_stop_retrying_when_deadline_exceeded(self, mock_request, mock_wa
             # Verify total time passed is at the timeout limit
             self.assertGreaterEqual(5, self.exporter._timeout)
 
+    @patch(
+        "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)
-    def test_export_interrupted_by_shutdown(self, mock_request):
+    def test_export_interrupted_by_shutdown(self, mock_request, mock_wait):
         """Tests that export can be interrupted by shutdown during retry wait."""
         self.exporter._timeout = 10000
-        
-        # Mock Event.wait to call shutdown on first call, then return True (interrupted)
-        # We cannot call shutdown() at the beginning since the exporter would just automatically return a FAILURE result without even attempting the export.
-        def mock_wait_with_shutdown(timeout):
-            self.exporter.shutdown()  
-            return True
-            
-        with patch.object(self.exporter._shutdown_event, 'wait', side_effect=mock_wait_with_shutdown):
-            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)
+
+        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)
+        mock_wait.assert_called_once()
 
     @staticmethod
     def generate_test_log_data(count=5):

aws-opentelemetry-distro/tests/amazon/opentelemetry/distro/test_aws_opentelementry_configurator.py

@@ -47,7 +47,7 @@
 from amazon.opentelemetry.distro.aws_span_metrics_processor import AwsSpanMetricsProcessor
 from amazon.opentelemetry.distro.exporter.aws.metrics.aws_cloudwatch_emf_exporter import AwsCloudWatchEmfExporter
 from amazon.opentelemetry.distro.exporter.otlp.aws.common.aws_auth_session import AwsAuthSession
-from amazon.opentelemetry.distro.exporter.otlp.aws.logs.aws_batch_log_record_processor import (
+from amazon.opentelemetry.distro.exporter.otlp.aws.logs._aws_cw_otlp_batch_log_record_processor import (
     AwsCloudWatchOtlpBatchLogRecordProcessor,
 )
 from amazon.opentelemetry.distro.exporter.otlp.aws.logs.otlp_aws_logs_exporter import OTLPAwsLogExporter