Merge remote-tracking branch 'liustve/logs-mainline' into genesis-logs

Author: liustve

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

@@ -136,10 +136,33 @@ def _estimate_log_size(self, log: LogData, depth: int = 3) -> int:  # pylint: di
         """
         Estimates the size in bytes of a log by calculating the size of its body and its attributes
         and adding a buffer amount to account for other log metadata information.
-        Will process complex log structures up to the specified depth limit.
-        Includes cycle detection to prevent processing the log content more than once.
-        If the depth limit of the log structure is exceeded, returns the truncated calculation
-        to everything up to that point.
+
+        Features:
+        - Processes complex log structures up to the specified depth limit
+        - Includes cycle detection to prevent processing the same content more than once
+        - Returns truncated calculation if depth limit is exceeded
+
+        We set depth to 3 as this is the minimum required depth to estimate our consolidated Gen AI log events:
+
+        Example structure:
+        {
+            "output": {
+                "messages": [
+                    {
+                        "content": "Hello, World!",
+                        "role": "assistant"
+                    }
+                ]
+            },
+            "input": {
+                "messages": [
+                    {
+                        "content": "Say Hello, World!",
+                        "role": "user"
+                    }
+                ]
+            }
+        }
 
         Args:
             log: The Log object to calculate size for
@@ -175,12 +198,12 @@ def _estimate_log_size(self, log: LogData, depth: int = 3) -> int:  # pylint: di
                 if next_val is None:
                     continue
 
-                if isinstance(next_val, (str, bytes)):
+                if isinstance(next_val, bytes):
                     size += len(next_val)
                     continue
 
-                if isinstance(next_val, (float, int, bool)):
-                    size += len(str(next_val))
+                if isinstance(next_val, (str, float, int, bool)):
+                    size += AwsCloudWatchOtlpBatchLogRecordProcessor._estimate_utf8_size(str(next_val))
                     continue
 
                 # next_val must be Sequence["AnyValue"] or Mapping[str, "AnyValue"]
@@ -211,6 +234,20 @@ def _estimate_log_size(self, log: LogData, depth: int = 3) -> int:  # pylint: di
 
         return size
 
+    @staticmethod
+    def _estimate_utf8_size(s: str):
+        ascii_count = 0
+        non_ascii_count = 0
+
+        for char in s:
+            if ord(char) < 128:
+                ascii_count += 1
+            else:
+                non_ascii_count += 1
+
+        # Estimate: ASCII chars (1 byte) + upper bound of non-ASCII chars 4 bytes
+        return ascii_count + (non_ascii_count * 4)
+
     # Only export the logs once to avoid the race condition of the worker thread and force flush thread
     # https://github.com/open-telemetry/opentelemetry-python/issues/3193
     # https://github.com/open-telemetry/opentelemetry-python/blob/main/opentelemetry-sdk/src/opentelemetry/sdk/_shared_internal/__init__.py#L199

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

@@ -6,7 +6,8 @@
 import logging
 import random
 from io import BytesIO
-from time import sleep, time
+from threading import Event
+from time import time
 from typing import Dict, Optional, Sequence
 
 from botocore.session import Session
@@ -69,6 +70,7 @@ def __init__(
             compression=Compression.Gzip,
             session=AwsAuthSession(session=session, aws_region=self._aws_region, service="logs"),
         )
+        self._shutdown_event = Event()
 
     def export(self, batch: Sequence[LogData]) -> LogExportResult:
         """
@@ -123,9 +125,18 @@ def export(self, batch: Sequence[LogData]) -> LogExportResult:
                 resp.reason,
                 backoff_seconds,
             )
-            sleep(backoff_seconds)
+            # Use interruptible sleep that can be interrupted by shutdown
+            if self._shutdown_event.wait(backoff_seconds):
+                _logger.info("Export interrupted by shutdown")
+                return LogExportResult.FAILURE
+
             retry_num += 1
 
+    def shutdown(self) -> None:
+        """Shutdown the exporter and interrupt any ongoing waits."""
+        self._shutdown_event.set()
+        return super().shutdown()
+
     def _send(self, serialized_data: bytes, timeout_sec: float):
         try:
             response = self._session.post(

aws-opentelemetry-distro/tests/amazon/opentelemetry/distro/exporter/otlp/aws/common/test_aws_auth_session.py

@@ -46,18 +46,3 @@ def test_aws_auth_session(self, _, __):
         self.assertIn(AUTHORIZATION_HEADER, actual_headers)
         self.assertIn(X_AMZ_DATE_HEADER, actual_headers)
         self.assertIn(X_AMZ_SECURITY_TOKEN_HEADER, actual_headers)
-
-    @patch("requests.Session.request", return_value=requests.Response())
-    @patch("botocore.session.Session.get_credentials", return_value=mock_credentials)
-    @patch("botocore.auth.SigV4Auth.add_auth", side_effect=Exception("Signing failed"))
-    def test_aws_auth_session_signing_error(self, mock_add_auth, mock_get_credentials, mock_request):
-        """Tests that aws_auth_session does not any Sigv4 headers if signing errors."""
-
-        session = AwsAuthSession("us-east-1", "xray")
-        actual_headers = {"test": "test"}
-
-        session.request("POST", AWS_OTLP_TRACES_ENDPOINT, data="", headers=actual_headers)
-
-        self.assertNotIn(AUTHORIZATION_HEADER, actual_headers)
-        self.assertNotIn(X_AMZ_DATE_HEADER, actual_headers)
-        self.assertNotIn(X_AMZ_SECURITY_TOKEN_HEADER, actual_headers)

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,
 )
@@ -116,8 +116,8 @@ def test_process_log_data_nested_structure_size_exceeds_max_log_size(self):
 
     def test_process_log_data_primitive(self):
 
-        primitives: List[AnyValue] = ["test", b"test", 1, 1.2, True, False, None]
-        expected_sizes = [4, 4, 1, 3, 4, 5, 0]
+        primitives: List[AnyValue] = ["test", b"test", 1, 1.2, True, False, None, "深入 Python", "calfé"]
+        expected_sizes = [4, 4, 1, 3, 4, 5, 0, 2 * 4 + len(" Python"), 1 * 4 + len("calf")]
 
         for index, primitive in enumerate(primitives):
             log = self.generate_test_log_data(log_body=primitive, count=1)
@@ -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)
@@ -241,11 +241,11 @@ def test_force_flush_returns_false_when_shutdown(self):
         self.mock_exporter.export.assert_not_called()
 
     @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_force_flush_exports_only_one_batch(self, _, __, ___):
         """Tests that force_flush should try to at least export one batch of logs. Rest of the logs will be dropped"""
         # Set max_export_batch_size to 5 to limit batch size

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

@@ -74,18 +74,19 @@ def test_should_not_export_again_if_not_retryable(self, mock_request):
         self.assertEqual(result, LogExportResult.FAILURE)
 
     @patch(
-        "amazon.opentelemetry.distro.exporter.otlp.aws.logs.otlp_aws_logs_exporter.sleep", side_effect=lambda x: None
+        "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_sleep):
+    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.
         But there is no Retry-After header."""
         self.exporter._timeout = 10000  # Large timeout to avoid early exit
         result = self.exporter.export(self.logs)
 
-        self.assertEqual(mock_sleep.call_count, _MAX_RETRYS - 1)
+        self.assertEqual(mock_wait.call_count, _MAX_RETRYS - 1)
 
-        delays = mock_sleep.call_args_list
+        delays = mock_wait.call_args_list
 
         for index, delay in enumerate(delays):
             expected_base = 2**index
@@ -98,28 +99,31 @@ def test_should_export_again_with_backoff_if_retryable_and_no_retry_after_header
         self.assertEqual(result, LogExportResult.FAILURE)
 
     @patch(
-        "amazon.opentelemetry.distro.exporter.otlp.aws.logs.otlp_aws_logs_exporter.sleep", side_effect=lambda x: None
+        "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],
     )
-    def test_should_export_again_with_server_delay_if_retryable_and_retry_after_header(self, mock_request, mock_sleep):
+    def test_should_export_again_with_server_delay_if_retryable_and_retry_after_header(self, mock_request, mock_wait):
         """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)
-        delays = mock_sleep.call_args_list
+
+        delays = mock_wait.call_args_list
 
         for delay in delays:
             self.assertEqual(delay[0][0], 10)
 
-        self.assertEqual(mock_sleep.call_count, 3)
+        self.assertEqual(mock_wait.call_count, 3)
         self.assertEqual(mock_request.call_count, 4)
         self.assertEqual(result, LogExportResult.SUCCESS)
 
     @patch(
-        "amazon.opentelemetry.distro.exporter.otlp.aws.logs.otlp_aws_logs_exporter.sleep", side_effect=lambda x: None
+        "amazon.opentelemetry.distro.exporter.otlp.aws.logs.otlp_aws_logs_exporter.Event.wait",
+        side_effect=lambda x: False,
     )
     @patch(
         "requests.Session.post",
@@ -131,13 +135,14 @@ def test_should_export_again_with_server_delay_if_retryable_and_retry_after_head
         ],
     )
     def test_should_export_again_with_backoff_delay_if_retryable_and_bad_retry_after_header(
-        self, mock_request, mock_sleep
+        self, mock_request, mock_wait
     ):
         """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)
-        delays = mock_sleep.call_args_list
+
+        delays = mock_wait.call_args_list
 
         for index, delay in enumerate(delays):
             expected_base = 2**index
@@ -146,7 +151,7 @@ def test_should_export_again_with_backoff_delay_if_retryable_and_bad_retry_after
             self.assertGreaterEqual(actual_delay, expected_base * 0.8)
             self.assertLessEqual(actual_delay, expected_base * 1.2)
 
-        self.assertEqual(mock_sleep.call_count, 3)
+        self.assertEqual(mock_wait.call_count, 3)
         self.assertEqual(mock_request.call_count, 4)
         self.assertEqual(result, LogExportResult.SUCCESS)
 
@@ -159,28 +164,44 @@ def test_export_connection_error_retry(self, mock_request):
         self.assertEqual(result, LogExportResult.SUCCESS)
 
     @patch(
-        "amazon.opentelemetry.distro.exporter.otlp.aws.logs.otlp_aws_logs_exporter.sleep", side_effect=lambda x: None
+        "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_sleep):
+    def test_should_stop_retrying_when_deadline_exceeded(self, mock_request, mock_wait):
         """Tests that the exporter stops retrying when the deadline is exceeded."""
         self.exporter._timeout = 5  # Short timeout to trigger deadline check
 
-        # Mock time to simulate time passing
         with patch("amazon.opentelemetry.distro.exporter.otlp.aws.logs.otlp_aws_logs_exporter.time") as mock_time:
             # First call returns start time, subsequent calls simulate time passing
             mock_time.side_effect = [0, 0, 1, 2, 4, 8]  # Exponential backoff would be 1, 2, 4 seconds
 
             result = self.exporter.export(self.logs)
 
             # Should stop before max retries due to deadline
-            self.assertLess(mock_sleep.call_count, _MAX_RETRYS)
+            self.assertLess(mock_wait.call_count, _MAX_RETRYS)
             self.assertLess(mock_request.call_count, _MAX_RETRYS + 1)
             self.assertEqual(result, LogExportResult.FAILURE)
 
             # 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, mock_wait):
+        """Tests that export can be interrupted by shutdown during retry wait."""
+        self.exporter._timeout = 10000
+
+        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):
         logs = []

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

@@ -26,6 +26,7 @@
     OtlpLogHeaderSetting,
     _check_emf_exporter_enabled,
     _custom_import_sampler,
+    _customize_log_record_processor,
     _customize_logs_exporter,
     _customize_metric_exporters,
     _customize_resource,
@@ -46,6 +47,11 @@
 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
+
+# pylint: disable=line-too-long
+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
 from amazon.opentelemetry.distro.exporter.otlp.aws.traces.otlp_aws_span_exporter import OTLPAwsSpanExporter
 from amazon.opentelemetry.distro.otlp_udp_exporter import OTLPUdpSpanExporter
@@ -1010,6 +1016,18 @@ def test_validate_and_fetch_logs_header(self):
         # Clean up
         os.environ.pop(OTEL_EXPORTER_OTLP_LOGS_HEADERS, None)
 
+    @patch("amazon.opentelemetry.distro.aws_opentelemetry_configurator.is_agent_observability_enabled")
+    @patch("amazon.opentelemetry.distro.aws_opentelemetry_configurator._is_aws_otlp_endpoint")
+    def test_customize_log_record_processor_with_agent_observability(self, mock_is_aws_endpoint, mock_is_agent_enabled):
+        """Test that AwsCloudWatchOtlpBatchLogRecordProcessor is used when agent observability is enabled"""
+        mock_exporter = MagicMock(spec=OTLPAwsLogExporter)
+        mock_is_agent_enabled.return_value = True
+        mock_is_aws_endpoint.return_value = True
+
+        processor = _customize_log_record_processor(mock_exporter)
+
+        self.assertIsInstance(processor, AwsCloudWatchOtlpBatchLogRecordProcessor)
+
     @patch("amazon.opentelemetry.distro.aws_opentelemetry_configurator._validate_and_fetch_logs_header")
     @patch("amazon.opentelemetry.distro.aws_opentelemetry_configurator.get_aws_session")
     def test_create_emf_exporter(self, mock_get_session, mock_validate):