merge

Author: liustve

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

@@ -11,6 +11,7 @@
 
 AGENT_OBSERVABILITY_ENABLED = "AGENT_OBSERVABILITY_ENABLED"
 
+
 def is_installed(req: str) -> bool:
     """Is the given required package installed?"""
 
@@ -24,5 +25,6 @@ def is_installed(req: str) -> bool:
         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

@@ -3,15 +3,15 @@
 # Modifications Copyright The OpenTelemetry Authors. Licensed under the Apache License 2.0 License.
 import os
 import re
-from logging import NOTSET, CRITICAL, Logger, getLogger
+from logging import NOTSET, Logger, getLogger
 from typing import ClassVar, Dict, List, Type, Union
 
 from importlib_metadata import version
 from typing_extensions import override
 
 from amazon.opentelemetry.distro._aws_attribute_keys import AWS_LOCAL_SERVICE
-from amazon.opentelemetry.distro._utils import is_agent_observability_enabled
 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,
@@ -22,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, get_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
@@ -123,24 +124,6 @@ class AwsOpenTelemetryConfigurator(_OTelSDKConfigurator):
     # pylint: disable=no-self-use
     @override
     def _configure(self, **kwargs):
-
-        print(f"OTEL_EXPORTER_OTLP_LOGS_HEADERS: {os.environ.get('OTEL_EXPORTER_OTLP_LOGS_HEADERS', 'Not set')}")
-        print(f"OTEL_PYTHON_LOGGING_AUTO_INSTRUMENTATION_ENABLED: {os.environ.get('OTEL_PYTHON_LOGGING_AUTO_INSTRUMENTATION_ENABLED', 'Not set')}")
-        print(f"OTEL_METRICS_EXPORTER: {os.environ.get('OTEL_METRICS_EXPORTER', 'Not set')}")
-        print(f"OTEL_TRACES_EXPORTER: {os.environ.get('OTEL_TRACES_EXPORTER', 'Not set')}")
-        print(f"OTEL_LOGS_EXPORTER: {os.environ.get('OTEL_LOGS_EXPORTER', 'Not set')}")
-        print(f"OTEL_PYTHON_DISTRO: {os.environ.get('OTEL_PYTHON_DISTRO', 'Not set')}")
-        print(f"OTEL_PYTHON_CONFIGURATOR: {os.environ.get('OTEL_PYTHON_CONFIGURATOR', 'Not set')}")
-        print(f"OTEL_EXPORTER_OTLP_PROTOCOL: {os.environ.get('OTEL_EXPORTER_OTLP_PROTOCOL', 'Not set')}")
-        print(f"OTEL_EXPORTER_OTLP_TRACES_ENDPOINT: {os.environ.get('OTEL_EXPORTER_OTLP_TRACES_ENDPOINT', 'Not set')}")
-        print(f"OTEL_EXPORTER_OTLP_LOGS_ENDPOINT: {os.environ.get('OTEL_EXPORTER_OTLP_LOGS_ENDPOINT', 'Not set')}")
-        print(f"OTEL_RESOURCE_ATTRIBUTES: {os.environ.get('OTEL_RESOURCE_ATTRIBUTES', 'Not set')}")
-        print(f"AGENT_OBSERVABILITY_ENABLED: {os.environ.get('AGENT_OBSERVABILITY_ENABLED', 'Not set')}")
-        print(f"AWS_CLOUDWATCH_LOG_GROUP: {os.environ.get('AWS_CLOUDWATCH_LOG_GROUP', 'Not set')}")
-        print(f"AWS_CLOUDWATCH_LOG_STREAM: {os.environ.get('AWS_CLOUDWATCH_LOG_STREAM', 'Not set')}")
-        print(f"OTEL_PYTHON_DISABLED_INSTRUMENTATIONS: {os.environ.get('OTEL_PYTHON_DISABLED_INSTRUMENTATIONS', 'Not set')}")
-        print(f"PYTHONPATH: {os.environ.get('PYTHONPATH', 'Not set')}")
-
         if _is_defer_to_workers_enabled() and _is_wsgi_master_process():
             _logger.info(
                 "Skipping ADOT initialization since deferral to worker is enabled, and this is a master process."
@@ -193,9 +176,6 @@ def _initialize_components():
         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(
@@ -205,7 +185,7 @@ def _init_logging(
 
     # 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 and os.environ.get(OTEL_LOGS_EXPORTER) == None:
+    if not exporters and os.environ.get(OTEL_LOGS_EXPORTER) is None:
         exporters = {"otlp": OTLPLogExporter}
 
     provider = LoggerProvider(resource=resource)
@@ -214,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(AwsBatchLogRecordProcessor(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)
 
@@ -385,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(
@@ -650,4 +637,4 @@ def create_exporter(self):
                 endpoint=application_signals_endpoint, preferred_temporality=temporality_dict
             )
 
-        raise RuntimeError(f"Unsupported AWS Application Signals export protocol: {protocol} ")
+        raise RuntimeError(f"Unsupported AWS Application Signals export protocol: {protocol} ")

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

@@ -1,2 +1,4 @@
-BASE_LOG_BUFFER_BYTE_SIZE = 450000
-MAX_LOG_REQUEST_BYTE_SIZE = 1048576 # https://docs.aws.amazon.com/AmazonCloudWatch/latest/monitoring/CloudWatch-OTLPEndpoint.html
+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

@@ -1,47 +1,51 @@
-from time import sleep
-import json
 import logging
-import os
-import threading
 from typing import Mapping, Sequence
-from opentelemetry.sdk._logs.export import BatchLogRecordProcessor
-from amazon.opentelemetry.distro.exporter.otlp.aws.common.constants import MAX_LOG_REQUEST_BYTE_SIZE, BASE_LOG_BUFFER_BYTE_SIZE
-from amazon.opentelemetry.distro.exporter.otlp.aws.logs.otlp_aws_logs_exporter import OTLPAwsLogExporter
-from opentelemetry.sdk._logs._internal.export import BatchLogExportStrategy, detach, attach, set_value, _SUPPRESS_INSTRUMENTATION_KEY
 
+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__(        
+    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
-        ):      
+        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
+            max_queue_size=max_queue_size,
         )
 
-    # Code based off of:
     # 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:
         """
-        Overrides the batching behavior of upstream's export method. 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.
+        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: 
+        - 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
         """
 
@@ -57,78 +61,70 @@ def _export(self, batch_strategy: BatchLogExportStrategy) -> None:
                     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)        
-        
-    def _get_size_of_log(self, log_data: LogData) -> int:
+                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.
+        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 += self._get_size_of_any_value(body)
+            size += AwsBatchLogRecordProcessor._get_size_of_any_value(body)
 
         return size
 
-    def _get_size_of_any_value(self, val: AnyValue) -> int:
+    @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):
-            if val:
-                return 4 #len(True) = 4
-            return 5 #len(False) = 5
-        
+            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 += self._get_size_of_any_value(content)
-        
+                size += AwsBatchLogRecordProcessor._get_size_of_any_value(content)
+
         if isinstance(val, Mapping):
             for _, content in val.items():
-                size += self._get_size_of_any_value(content)
-        
-        return size
-
-
+                size += AwsBatchLogRecordProcessor._get_size_of_any_value(content)
 
-
-
-        
-        
-    
+        return size