add compressed console log exporter for lambda environment

Author: jj22ee

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

@@ -23,6 +23,9 @@
     AwsMetricAttributesSpanExporterBuilder,
 )
 from amazon.opentelemetry.distro.aws_span_metrics_processor_builder import AwsSpanMetricsProcessorBuilder
+from amazon.opentelemetry.distro.exporter.console.logs.compressed_console_log_exporter import (
+    CompressedConsoleLogExporter,
+)
 from amazon.opentelemetry.distro.otlp_udp_exporter import OTLPUdpSpanExporter
 from amazon.opentelemetry.distro.sampler.aws_xray_remote_sampler import AwsXRayRemoteSampler
 from amazon.opentelemetry.distro.scope_based_exporter import ScopeBasedPeriodicExportingMetricReader
@@ -46,7 +49,7 @@
 )
 from opentelemetry.sdk._events import EventLoggerProvider
 from opentelemetry.sdk._logs import LoggerProvider, LoggingHandler
-from opentelemetry.sdk._logs.export import BatchLogRecordProcessor, LogExporter
+from opentelemetry.sdk._logs.export import BatchLogRecordProcessor, ConsoleLogExporter, LogExporter
 from opentelemetry.sdk.environment_variables import (
     _OTEL_PYTHON_LOGGING_AUTO_INSTRUMENTATION_ENABLED,
     OTEL_EXPORTER_OTLP_METRICS_PROTOCOL,
@@ -209,6 +212,9 @@ def _init_logging(
     set_logger_provider(provider)
 
     for _, exporter_class in exporters.items():
+        if exporter_class is ConsoleLogExporter and _is_lambda_environment():
+            exporter_class = CompressedConsoleLogExporter
+            _logger.debug("Lambda environment detected, using CompressedConsoleLogExporter")
         exporter_args = {}
         _customize_log_record_processor(
             logger_provider=provider, log_exporter=_customize_logs_exporter(exporter_class(**exporter_args))

aws-opentelemetry-distro/src/amazon/opentelemetry/distro/exporter/console/logs/compressed_console_log_exporter.py

@@ -0,0 +1,16 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+import re
+from typing import Sequence
+
+from opentelemetry.sdk._logs import LogData
+from opentelemetry.sdk._logs.export import ConsoleLogExporter, LogExportResult
+
+
+class CompressedConsoleLogExporter(ConsoleLogExporter):
+    def export(self, batch: Sequence[LogData]):
+        for data in batch:
+            formatted_json = self.formatter(data.log_record)
+            print(re.sub(r"\s*([{}[\]:,])\s*", r"\1", formatted_json), flush=True)
+
+        return LogExportResult.SUCCESS

aws-opentelemetry-distro/tests/amazon/opentelemetry/distro/exporter/console/logs/test_compressed_console_log_exporter.py

@@ -0,0 +1,72 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+import unittest
+from unittest.mock import Mock, patch
+
+from amazon.opentelemetry.distro.exporter.console.logs.compressed_console_log_exporter import (
+    CompressedConsoleLogExporter,
+)
+from opentelemetry.sdk._logs.export import LogExportResult
+
+
+class TestCompressedConsoleLogExporter(unittest.TestCase):
+
+    def setUp(self):
+        self.exporter = CompressedConsoleLogExporter()
+
+    @patch("builtins.print")
+    def test_export_compresses_json(self, mock_print):
+        # Mock log data
+        mock_log_data = Mock()
+        mock_log_record = Mock()
+        mock_log_data.log_record = mock_log_record
+
+        # Mock formatted JSON with whitespace
+        formatted_json = '{\n    "body": "test message",\n    "severity_number": 9,\n    "attributes": {\n        "key": "value"\n    }\n}'  # noqa: E501
+        self.exporter.formatter = Mock(return_value=formatted_json)
+
+        # Call export
+        result = self.exporter.export([mock_log_data])
+
+        # Verify result
+        self.assertEqual(result, LogExportResult.SUCCESS)
+
+        # Verify print calls
+        self.assertEqual(mock_print.call_count, 1)
+        mock_print.assert_called_with(
+            '{"body":"test message","severity_number":9,"attributes":{"key":"value"}}', flush=True
+        )
+
+    @patch("builtins.print")
+    def test_export_multiple_records(self, mock_print):
+        # Mock multiple log data
+        mock_log_data1 = Mock()
+        mock_log_data2 = Mock()
+        mock_log_data1.log_record = Mock()
+        mock_log_data2.log_record = Mock()
+
+        formatted_json = '{\n    "body": "test"\n}'
+        self.exporter.formatter = Mock(return_value=formatted_json)
+
+        # Call export
+        result = self.exporter.export([mock_log_data1, mock_log_data2])
+
+        # Verify result
+        self.assertEqual(result, LogExportResult.SUCCESS)
+
+        # Verify print calls
+        self.assertEqual(mock_print.call_count, 2)  # 2 records
+        # Each record should print compressed JSON
+        expected_calls = [unittest.mock.call('{"body":"test"}', flush=True)] * 2
+        mock_print.assert_has_calls(expected_calls)
+
+    @patch("builtins.print")
+    def test_export_empty_batch(self, mock_print):
+        # Call export with empty batch
+        result = self.exporter.export([])
+
+        # Verify result
+        self.assertEqual(result, LogExportResult.SUCCESS)
+
+        # Verify print calls
+        mock_print.assert_not_called()  # No records, no prints

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

@@ -47,6 +47,9 @@
 from amazon.opentelemetry.distro.aws_opentelemetry_distro import AwsOpenTelemetryDistro
 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.console.logs.compressed_console_log_exporter import (
+    CompressedConsoleLogExporter,
+)
 from amazon.opentelemetry.distro.exporter.otlp.aws.common.aws_auth_session import AwsAuthSession
 
 # pylint: disable=line-too-long
@@ -70,7 +73,7 @@
 from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
 from opentelemetry.metrics import get_meter_provider
 from opentelemetry.processor.baggage import BaggageSpanProcessor
-from opentelemetry.sdk._logs.export import BatchLogRecordProcessor
+from opentelemetry.sdk._logs.export import BatchLogRecordProcessor, ConsoleLogExporter
 from opentelemetry.sdk.environment_variables import OTEL_TRACES_SAMPLER, OTEL_TRACES_SAMPLER_ARG
 from opentelemetry.sdk.metrics._internal.export import PeriodicExportingMetricReader
 from opentelemetry.sdk.resources import Resource
@@ -676,6 +679,58 @@ def capture_exporter(*args, **kwargs):
 
         os.environ.pop(OTEL_EXPORTER_OTLP_LOGS_ENDPOINT)
 
+    @patch("amazon.opentelemetry.distro.aws_opentelemetry_configurator.LoggingHandler", return_value=MagicMock())
+    @patch("logging.getLogger", return_value=MagicMock())
+    @patch("amazon.opentelemetry.distro.aws_opentelemetry_configurator._customize_logs_exporter")
+    @patch("amazon.opentelemetry.distro.aws_opentelemetry_configurator.LoggerProvider", return_value=MagicMock())
+    @patch("amazon.opentelemetry.distro.aws_opentelemetry_configurator._customize_log_record_processor")
+    def test_init_logging_console_exporter_replacement(
+        self,
+        mock_customize_processor,
+        mock_logger_provider,
+        mock_customize_logs_exporter,
+        mock_get_logger,
+        mock_logging_handler,
+    ):
+        """Test that ConsoleLogExporter is replaced with CompressedConsoleLogExporter when in Lambda"""
+
+        # Mock _is_lambda_environment to return True
+        with patch(
+            "amazon.opentelemetry.distro.aws_opentelemetry_configurator._is_lambda_environment", return_value=True
+        ):
+            # Test with ConsoleLogExporter
+            exporters = {"console": ConsoleLogExporter}
+            _init_logging(exporters, Resource.get_empty())
+
+            # Verify that _customize_log_record_processor was called
+            mock_customize_processor.assert_called_once()
+
+            # Get the exporter that was passed to _customize_logs_exporter
+            call_args = mock_customize_logs_exporter.call_args
+            exporter_instance = call_args[0][0]
+
+            # Verify it's a CompressedConsoleLogExporter instance
+            self.assertIsInstance(exporter_instance, CompressedConsoleLogExporter)
+
+        # Reset mocks
+        mock_customize_processor.reset_mock()
+        mock_customize_logs_exporter.reset_mock()
+
+        # Test when not in Lambda environment - should not replace
+        with patch(
+            "amazon.opentelemetry.distro.aws_opentelemetry_configurator._is_lambda_environment", return_value=False
+        ):
+            exporters = {"console": ConsoleLogExporter}
+            _init_logging(exporters, Resource.get_empty())
+
+            # Get the exporter that was passed to _customize_logs_exporter
+            call_args = mock_customize_logs_exporter.call_args
+            exporter_instance = call_args[0][0]
+
+            # Verify it's still a regular ConsoleLogExporter
+            self.assertIsInstance(exporter_instance, ConsoleLogExporter)
+            self.assertNotIsInstance(exporter_instance, CompressedConsoleLogExporter)
+
     def test_customize_span_processors(self):
         mock_tracer_provider: TracerProvider = MagicMock()
         os.environ.pop("AGENT_OBSERVABILITY_ENABLED", None)