[WIP] Support for handling LLO in OTLP AWS Logs Exporter

aws-opentelemetry-distro/pyproject.toml

@@ -25,61 +25,61 @@ classifiers = [
 ]
 
 dependencies = [
-  "opentelemetry-api == 1.27.0",
-  "opentelemetry-sdk == 1.27.0",
-  "opentelemetry-exporter-otlp-proto-grpc == 1.27.0",
-  "opentelemetry-exporter-otlp-proto-http == 1.27.0",
-  "opentelemetry-propagator-b3 == 1.27.0",
-  "opentelemetry-propagator-jaeger == 1.27.0",
-  "opentelemetry-exporter-otlp-proto-common == 1.27.0",
-  "opentelemetry-sdk-extension-aws == 2.0.2",
-  "opentelemetry-propagator-aws-xray == 1.0.1",
-  "opentelemetry-distro == 0.48b0",
-  "opentelemetry-propagator-ot-trace == 0.48b0",
-  "opentelemetry-instrumentation == 0.48b0",
-  "opentelemetry-instrumentation-aws-lambda == 0.48b0",
-  "opentelemetry-instrumentation-aio-pika == 0.48b0",
-  "opentelemetry-instrumentation-aiohttp-client == 0.48b0",
-  "opentelemetry-instrumentation-aiopg == 0.48b0",
-  "opentelemetry-instrumentation-asgi == 0.48b0",
-  "opentelemetry-instrumentation-asyncpg == 0.48b0",
-  "opentelemetry-instrumentation-boto == 0.48b0",
-  "opentelemetry-instrumentation-boto3sqs == 0.48b0",
-  "opentelemetry-instrumentation-botocore == 0.48b0",
-  "opentelemetry-instrumentation-celery == 0.48b0",
-  "opentelemetry-instrumentation-confluent-kafka == 0.48b0",
-  "opentelemetry-instrumentation-dbapi == 0.48b0",
-  "opentelemetry-instrumentation-django == 0.48b0",
-  "opentelemetry-instrumentation-elasticsearch == 0.48b0",
-  "opentelemetry-instrumentation-falcon == 0.48b0",
-  "opentelemetry-instrumentation-fastapi == 0.48b0",
-  "opentelemetry-instrumentation-flask == 0.48b0",
-  "opentelemetry-instrumentation-grpc == 0.48b0",
-  "opentelemetry-instrumentation-httpx == 0.48b0",
-  "opentelemetry-instrumentation-jinja2 == 0.48b0",
-  "opentelemetry-instrumentation-kafka-python == 0.48b0",
-  "opentelemetry-instrumentation-logging == 0.48b0",
-  "opentelemetry-instrumentation-mysql == 0.48b0",
-  "opentelemetry-instrumentation-mysqlclient == 0.48b0",
-  "opentelemetry-instrumentation-pika == 0.48b0",
-  "opentelemetry-instrumentation-psycopg2 == 0.48b0",
-  "opentelemetry-instrumentation-pymemcache == 0.48b0",
-  "opentelemetry-instrumentation-pymongo == 0.48b0",
-  "opentelemetry-instrumentation-pymysql == 0.48b0",
-  "opentelemetry-instrumentation-pyramid == 0.48b0",
-  "opentelemetry-instrumentation-redis == 0.48b0",
-  "opentelemetry-instrumentation-remoulade == 0.48b0",
-  "opentelemetry-instrumentation-requests == 0.48b0",
-  "opentelemetry-instrumentation-sqlalchemy == 0.48b0",
-  "opentelemetry-instrumentation-sqlite3 == 0.48b0",
-  "opentelemetry-instrumentation-starlette == 0.48b0",
-  "opentelemetry-instrumentation-system-metrics == 0.48b0",
-  "opentelemetry-instrumentation-tornado == 0.48b0",
-  "opentelemetry-instrumentation-tortoiseorm == 0.48b0",
-  "opentelemetry-instrumentation-urllib == 0.48b0",
-  "opentelemetry-instrumentation-urllib3 == 0.48b0",
-  "opentelemetry-instrumentation-wsgi == 0.48b0",
-  "opentelemetry-instrumentation-cassandra == 0.48b0",
+  "opentelemetry-api >= 1.29.0",
+  "opentelemetry-sdk >= 1.29.0",
+  "opentelemetry-exporter-otlp-proto-grpc >= 1.29.0",
+  "opentelemetry-exporter-otlp-proto-http >= 1.29.0",
+  "opentelemetry-propagator-b3 >= 1.29.0",
+  "opentelemetry-propagator-jaeger >= 1.29.0",
+  "opentelemetry-exporter-otlp-proto-common >= 1.29.0",
+  "opentelemetry-sdk-extension-aws >= 2.0.2",
+  "opentelemetry-propagator-aws-xray >= 1.0.1",
+  "opentelemetry-distro >= 0.50b0",
+  "opentelemetry-propagator-ot-trace >= 0.50b0",
+  "opentelemetry-instrumentation >= 0.50b0",
+  "opentelemetry-instrumentation-aws-lambda >= 0.50b0",
+  "opentelemetry-instrumentation-aio-pika >= 0.50b0",
+  "opentelemetry-instrumentation-aiohttp-client >= 0.50b0",
+  "opentelemetry-instrumentation-aiopg >= 0.50b0",
+  "opentelemetry-instrumentation-asgi >= 0.50b0",
+  "opentelemetry-instrumentation-asyncpg >= 0.50b0",
+  "opentelemetry-instrumentation-boto >= 0.50b0",
+  "opentelemetry-instrumentation-boto3sqs >= 0.50b0",
+  "opentelemetry-instrumentation-botocore >= 0.50b0",
+  "opentelemetry-instrumentation-celery >= 0.50b0",
+  "opentelemetry-instrumentation-confluent-kafka >= 0.50b0",
+  "opentelemetry-instrumentation-dbapi >= 0.50b0",
+  "opentelemetry-instrumentation-django >= 0.50b0",
+  "opentelemetry-instrumentation-elasticsearch >= 0.50b0",
+  "opentelemetry-instrumentation-falcon >= 0.50b0",
+  "opentelemetry-instrumentation-fastapi >= 0.50b0",
+  "opentelemetry-instrumentation-flask >= 0.50b0",
+  "opentelemetry-instrumentation-grpc >= 0.50b0",
+  "opentelemetry-instrumentation-httpx >= 0.50b0",
+  "opentelemetry-instrumentation-jinja2 >= 0.50b0",
+  "opentelemetry-instrumentation-kafka-python >= 0.50b0",
+  "opentelemetry-instrumentation-logging >= 0.50b0",
+  "opentelemetry-instrumentation-mysql >= 0.50b0",
+  "opentelemetry-instrumentation-mysqlclient >= 0.50b0",
+  "opentelemetry-instrumentation-pika >= 0.50b0",
+  "opentelemetry-instrumentation-psycopg2 >= 0.50b0",
+  "opentelemetry-instrumentation-pymemcache >= 0.50b0",
+  "opentelemetry-instrumentation-pymongo >= 0.50b0",
+  "opentelemetry-instrumentation-pymysql >= 0.50b0",
+  "opentelemetry-instrumentation-pyramid >= 0.50b0",
+  "opentelemetry-instrumentation-redis >= 0.50b0",
+  "opentelemetry-instrumentation-remoulade >= 0.50b0",
+  "opentelemetry-instrumentation-requests >= 0.50b0",
+  "opentelemetry-instrumentation-sqlalchemy >= 0.50b0",
+  "opentelemetry-instrumentation-sqlite3 >= 0.50b0",
+  "opentelemetry-instrumentation-starlette >= 0.50b0",
+  "opentelemetry-instrumentation-system-metrics >= 0.50b0",
+  "opentelemetry-instrumentation-tornado >= 0.50b0",
+  "opentelemetry-instrumentation-tortoiseorm >= 0.50b0",
+  "opentelemetry-instrumentation-urllib >= 0.50b0",
+  "opentelemetry-instrumentation-urllib3 >= 0.50b0",
+  "opentelemetry-instrumentation-wsgi >= 0.50b0",
+  "opentelemetry-instrumentation-cassandra >= 0.50b0",
 ]
 
 [project.optional-dependencies]

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

@@ -1,13 +1,16 @@
 # Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
 # SPDX-License-Identifier: Apache-2.0
 
+import os
 import sys
 from logging import Logger, getLogger
 
 import pkg_resources
 
 _logger: Logger = getLogger(__name__)
 
+AGENT_OBSERVABILITY_ENABLED = "AGENT_OBSERVABILITY_ENABLED"
+
 
 def is_installed(req: str) -> bool:
     """Is the given required package installed?"""
@@ -21,3 +24,7 @@ def is_installed(req: str) -> bool:
         _logger.debug("Skipping instrumentation patch: package %s, exception: %s", req, exc)
         return False
     return True
+
+
+def is_agent_observability_enabled() -> bool:
+    return os.environ.get(AGENT_OBSERVABILITY_ENABLED, "false").lower() == "true"

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

@@ -11,6 +11,7 @@
 
 from amazon.opentelemetry.distro._aws_attribute_keys import AWS_LOCAL_SERVICE
 from amazon.opentelemetry.distro._aws_resource_attribute_configurator import get_service_attribute
+from amazon.opentelemetry.distro._utils import is_agent_observability_enabled
 from amazon.opentelemetry.distro.always_record_sampler import AlwaysRecordSampler
 from amazon.opentelemetry.distro.attribute_propagating_span_processor_builder import (
     AttributePropagatingSpanProcessorBuilder,
@@ -21,13 +22,14 @@
     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 AwsBatchLogRecordProcessor
 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
 from amazon.opentelemetry.distro.sampler.aws_xray_remote_sampler import AwsXRayRemoteSampler
 from amazon.opentelemetry.distro.scope_based_exporter import ScopeBasedPeriodicExportingMetricReader
 from amazon.opentelemetry.distro.scope_based_filtering_view import ScopeBasedRetainingView
-from opentelemetry._logs import set_logger_provider
+from opentelemetry._logs import get_logger_provider, set_logger_provider
 from opentelemetry.exporter.otlp.proto.http._log_exporter import OTLPLogExporter
 from opentelemetry.exporter.otlp.proto.http.metric_exporter import OTLPMetricExporter as OTLPHttpOTLPMetricExporter
 from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
@@ -82,6 +84,7 @@
 DEPRECATED_APP_SIGNALS_EXPORTER_ENDPOINT_CONFIG = "OTEL_AWS_APP_SIGNALS_EXPORTER_ENDPOINT"
 APPLICATION_SIGNALS_EXPORTER_ENDPOINT_CONFIG = "OTEL_AWS_APPLICATION_SIGNALS_EXPORTER_ENDPOINT"
 METRIC_EXPORT_INTERVAL_CONFIG = "OTEL_METRIC_EXPORT_INTERVAL"
+OTEL_LOGS_EXPORTER = "OTEL_LOGS_EXPORTER"
 DEFAULT_METRIC_EXPORT_INTERVAL = 60000.0
 AWS_LAMBDA_FUNCTION_NAME_CONFIG = "AWS_LAMBDA_FUNCTION_NAME"
 AWS_XRAY_DAEMON_ADDRESS_CONFIG = "AWS_XRAY_DAEMON_ADDRESS"
@@ -91,6 +94,8 @@
 OTEL_EXPORTER_OTLP_LOGS_ENDPOINT = "OTEL_EXPORTER_OTLP_LOGS_ENDPOINT"
 OTEL_EXPORTER_OTLP_LOGS_HEADERS = "OTEL_EXPORTER_OTLP_LOGS_HEADERS"
 
+AGENT_OBSERVABILITY_ENABLED = "AGENT_OBSERVABILITY_ENABLED"
+
 AWS_TRACES_OTLP_ENDPOINT_PATTERN = r"https://xray\.([a-z0-9-]+)\.amazonaws\.com/v1/traces$"
 AWS_LOGS_OTLP_ENDPOINT_PATTERN = r"https://logs\.([a-z0-9-]+)\.amazonaws\.com/v1/logs$"
 
@@ -160,26 +165,27 @@ def _initialize_components():
     sampler_name = _get_sampler()
     sampler = _custom_import_sampler(sampler_name, resource)
 
+    logging_enabled = os.getenv(_OTEL_PYTHON_LOGGING_AUTO_INSTRUMENTATION_ENABLED, "false")
+    if logging_enabled.strip().lower() == "true":
+        _init_logging(log_exporters, resource)
+
     _init_tracing(
         exporters=trace_exporters,
         id_generator=id_generator,
         sampler=sampler,
         resource=resource,
     )
     _init_metrics(metric_exporters, resource)
-    logging_enabled = os.getenv(_OTEL_PYTHON_LOGGING_AUTO_INSTRUMENTATION_ENABLED, "false")
-    if logging_enabled.strip().lower() == "true":
-        _init_logging(log_exporters, resource)
 
 
 def _init_logging(
     exporters: Dict[str, Type[LogExporter]],
     resource: Resource = None,
 ):
 
-    # Provides a default OTLP log exporter when none is specified.
+    # Provides a default OTLP log exporter when the environment is not set.
     # This is the behavior for the logs exporters for other languages.
-    if not exporters:
+    if not exporters and os.environ.get(OTEL_LOGS_EXPORTER) is None:
         exporters = {"otlp": OTLPLogExporter}
 
     provider = LoggerProvider(resource=resource)
@@ -188,7 +194,11 @@ def _init_logging(
     for _, exporter_class in exporters.items():
         exporter_args: Dict[str, any] = {}
         log_exporter = _customize_logs_exporter(exporter_class(**exporter_args), resource)
-        provider.add_log_record_processor(BatchLogRecordProcessor(exporter=log_exporter))
+
+        if isinstance(log_exporter, OTLPAwsLogExporter):
+            provider.add_log_record_processor(AwsBatchLogRecordProcessor(exporter=log_exporter))
+        else:
+            provider.add_log_record_processor(BatchLogRecordProcessor(exporter=log_exporter))
 
     handler = LoggingHandler(level=NOTSET, logger_provider=provider)
 
@@ -359,7 +369,10 @@ def _customize_span_exporter(span_exporter: SpanExporter, resource: Resource) ->
         _logger.info("Detected using AWS OTLP Traces Endpoint.")
 
         if isinstance(span_exporter, OTLPSpanExporter):
-            span_exporter = OTLPAwsSpanExporter(endpoint=traces_endpoint)
+            if is_agent_observability_enabled():
+                span_exporter = OTLPAwsSpanExporter(endpoint=traces_endpoint, logger_provider=get_logger_provider())
+            else:
+                span_exporter = OTLPAwsSpanExporter(endpoint=traces_endpoint)
 
         else:
             _logger.warning(

aws-opentelemetry-distro/src/amazon/opentelemetry/distro/exporter/otlp/aws/common/constants.py

@@ -0,0 +1,4 @@
+BASE_LOG_BUFFER_BYTE_SIZE = 2000
+MAX_LOG_REQUEST_BYTE_SIZE = (
+    1048576  # https://docs.aws.amazon.com/AmazonCloudWatch/latest/monitoring/CloudWatch-OTLPEndpoint.html
+)

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

@@ -0,0 +1,130 @@
+import logging
+from typing import Mapping, Sequence
+
+from amazon.opentelemetry.distro.exporter.otlp.aws.common.constants import (
+    BASE_LOG_BUFFER_BYTE_SIZE,
+    MAX_LOG_REQUEST_BYTE_SIZE,
+)
+from amazon.opentelemetry.distro.exporter.otlp.aws.logs.otlp_aws_logs_exporter import OTLPAwsLogExporter
+from opentelemetry.sdk._logs import LogData
+from opentelemetry.sdk._logs._internal.export import (
+    _SUPPRESS_INSTRUMENTATION_KEY,
+    BatchLogExportStrategy,
+    attach,
+    detach,
+    set_value,
+)
+from opentelemetry.sdk._logs.export import BatchLogRecordProcessor
+from opentelemetry.util.types import AnyValue
+
+_logger = logging.getLogger(__name__)
+
+
+class AwsBatchLogRecordProcessor(BatchLogRecordProcessor):
+
+    def __init__(
+        self,
+        exporter: OTLPAwsLogExporter,
+        schedule_delay_millis: float | None = None,
+        max_export_batch_size: int | None = None,
+        export_timeout_millis: float | None = None,
+        max_queue_size: int | None = None,
+    ):
+
+        super().__init__(
+            exporter=exporter,
+            schedule_delay_millis=schedule_delay_millis,
+            max_export_batch_size=max_export_batch_size,
+            export_timeout_millis=export_timeout_millis,
+            max_queue_size=max_queue_size,
+        )
+
+    # https://github.com/open-telemetry/opentelemetry-python/blob/main/opentelemetry-sdk/src/opentelemetry/sdk/_shared_internal/__init__.py#L143
+    def _export(self, batch_strategy: BatchLogExportStrategy) -> None:
+        """
+        Preserves existing batching behavior but will intermediarly export small log batches if the size of the data in the batch is at or
+        above AWS CloudWatch's maximum request size limit of 1 MB.
+
+        - Data size of exported batches will ALWAYS be <= 1 MB except for the case below:
+        - If the data size of an exported batch is ever > 1 MB then the batch size is guaranteed to be 1
+        """
+
+        with self._export_lock:
+            iteration = 0
+            # We could see concurrent export calls from worker and force_flush. We call _should_export_batch
+            # once the lock is obtained to see if we still need to make the requested export.
+            while self._should_export_batch(batch_strategy, iteration):
+
+                iteration += 1
+                token = attach(set_value(_SUPPRESS_INSTRUMENTATION_KEY, True))
+                try:
+                    batch_length = min(self._max_export_batch_size, len(self._queue))
+                    batch_data_size = 0
+                    batch = []
+
+                    for _ in range(batch_length):
+
+                        log_data = self._queue.pop()
+                        log_size = self._get_size_of_log(log_data)
+
+                        if batch and (batch_data_size + log_size > MAX_LOG_REQUEST_BYTE_SIZE):
+                            # if batch_data_size > MAX_LOG_REQUEST_BYTE_SIZE then len(batch) == 1
+                            if batch_data_size > MAX_LOG_REQUEST_BYTE_SIZE:
+                                self._exporter.set_gen_ai_flag()
+
+                            self._exporter.export(batch)
+                            batch_data_size = 0
+                            batch = []
+
+                        batch_data_size += log_size
+                        batch.append(log_data)
+
+                    if batch:
+                        # if batch_data_size > MAX_LOG_REQUEST_BYTE_SIZE then len(batch) == 1
+                        if batch_data_size > MAX_LOG_REQUEST_BYTE_SIZE:
+                            self._exporter.set_gen_ai_flag()
+
+                        self._exporter.export(batch)
+                except Exception:  # pylint: disable=broad-exception-caught
+                    _logger.exception("Exception while exporting logs.")
+                detach(token)
+
+    @staticmethod
+    def _get_size_of_log(log_data: LogData) -> int:
+        """
+        Estimates the size of a given LogData based on the size of the body + a buffer
+        amount representing a rough guess of other data present in the log.
+        """
+        size = BASE_LOG_BUFFER_BYTE_SIZE
+        body = log_data.log_record.body
+
+        if body:
+            size += AwsBatchLogRecordProcessor._get_size_of_any_value(body)
+
+        return size
+
+    @staticmethod
+    def _get_size_of_any_value(val: AnyValue) -> int:
+        """
+        Recursively calculates the size of an AnyValue type in bytes.
+        """
+        size = 0
+
+        if isinstance(val, str) or isinstance(val, bytes):
+            return len(val)
+
+        if isinstance(val, bool):
+            return 4 if val else 5
+
+        if isinstance(val, int) or isinstance(val, float):
+            return len(str(val))
+
+        if isinstance(val, Sequence):
+            for content in val:
+                size += AwsBatchLogRecordProcessor._get_size_of_any_value(content)
+
+        if isinstance(val, Mapping):
+            for _, content in val.items():
+                size += AwsBatchLogRecordProcessor._get_size_of_any_value(content)
+
+        return size