Consolidate botocore usage in EMF and Sigv4 Exporters (#419)

*Description of changes:*
This PR cleans up botocore usages in EMF and Sigv4 exporters and
migrates them to the `_utils` class.
- Added get_aws_session() function in _utils.py to handle botocore
session creation

- Refactored get_aws_region() to properly handle AWS_REGION and
AWS_DEFAULT_REGION environment variables

- Added IS_BOTOCORE_INSTALLED flag to prevent runtime errors for
customers not using botocore-dependent features

- Added _create_aws_otlp_exporter() function to prevent runtime errors
from instantiating AWS OTLP exporter creation if the user does not have
botocore installed

- Changed OTLP log exporter to allow optional constructor parameters for
`x-aws-log-group` and `x-aws-log-stream` headers to consolidate it with
EMF exporter

Force flush race condition: Added force flush implementation to prevent
race conditions in `AwsBatchLogRecordProcessor` as here:
#407


By submitting this pull request, I confirm that you can use, modify,
copy, and redistribute this contribution, under the terms of your
choice.

Author: liustve

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

@@ -4,6 +4,7 @@
 import os
 from importlib.metadata import PackageNotFoundError, version
 from logging import Logger, getLogger
+from typing import Optional
 
 from packaging.requirements import Requirement
 
@@ -37,30 +38,40 @@ def is_agent_observability_enabled() -> bool:
     return os.environ.get(AGENT_OBSERVABILITY_ENABLED, "false").lower() == "true"
 
 
-def get_aws_region() -> str:
-    """Get AWS region using botocore session.
+IS_BOTOCORE_INSTALLED: bool = is_installed("botocore")
 
-    botocore automatically checks in the following priority order:
-    1. AWS_REGION environment variable
-    2. AWS_DEFAULT_REGION environment variable
-    3. AWS CLI config file (~/.aws/config)
-    4. EC2 instance metadata service
 
-    Returns:
-        The AWS region if found, None otherwise.
+def get_aws_session():
     """
-    if is_installed("botocore"):
-        try:
-            from botocore import session  # pylint: disable=import-outside-toplevel
-
-            botocore_session = session.Session()
-            if botocore_session.region_name:
-                return botocore_session.region_name
-        except (ImportError, AttributeError):
-            # botocore failed to determine region
-            pass
-
-    _logger.warning(
-        "AWS region not found. Please set AWS_REGION environment variable or configure AWS CLI with 'aws configure'."
-    )
+    Returns a botocore session only if botocore is installed, otherwise None.
+    If AWS Region is defined in `AWS_REGION` or `AWS_DEFAULT_REGION` environment variables,
+    then the region is set in the botocore session before returning.
+
+    We do this to prevent runtime errors for ADOT customers that do not need
+    any features that require botocore.
+    """
+    if IS_BOTOCORE_INSTALLED:
+        # pylint: disable=import-outside-toplevel
+        from botocore.session import Session
+
+        session = Session()
+        # Botocore only looks up AWS_DEFAULT_REGION when creating a session/client
+        # See: https://docs.aws.amazon.com/sdkref/latest/guide/feature-region.html#feature-region-sdk-compat
+        region = os.environ.get("AWS_REGION") or os.environ.get("AWS_DEFAULT_REGION")
+        if region:
+            session.set_config_variable("region", region)
+        return session
     return None
+
+
+def get_aws_region() -> Optional[str]:
+    """Get AWS region from environment or botocore session.
+
+    Returns the AWS region in the following priority order:
+    1. AWS_REGION environment variable
+    2. AWS_DEFAULT_REGION environment variable
+    3. botocore session's region (if botocore is available)
+    4. None if no region can be determined
+    """
+    botocore_session = get_aws_session()
+    return botocore_session.get_config_variable("region") if botocore_session else None

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

@@ -12,7 +12,7 @@
 
 from amazon.opentelemetry.distro._aws_attribute_keys import AWS_LOCAL_SERVICE, AWS_SERVICE_TYPE
 from amazon.opentelemetry.distro._aws_resource_attribute_configurator import get_service_attribute
-from amazon.opentelemetry.distro._utils import is_agent_observability_enabled, is_installed
+from amazon.opentelemetry.distro._utils import get_aws_session, is_agent_observability_enabled
 from amazon.opentelemetry.distro.always_record_sampler import AlwaysRecordSampler
 from amazon.opentelemetry.distro.attribute_propagating_span_processor_builder import (
     AttributePropagatingSpanProcessorBuilder,
@@ -23,13 +23,6 @@
     AwsMetricAttributesSpanExporterBuilder,
 )
 from amazon.opentelemetry.distro.aws_span_metrics_processor_builder import AwsSpanMetricsProcessorBuilder
-
-# 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
 from amazon.opentelemetry.distro.sampler.aws_xray_remote_sampler import AwsXRayRemoteSampler
 from amazon.opentelemetry.distro.scope_based_exporter import ScopeBasedPeriodicExportingMetricReader
@@ -102,6 +95,8 @@
 OTEL_EXPORTER_OTLP_LOGS_ENDPOINT = "OTEL_EXPORTER_OTLP_LOGS_ENDPOINT"
 OTEL_EXPORTER_OTLP_LOGS_HEADERS = "OTEL_EXPORTER_OTLP_LOGS_HEADERS"
 
+XRAY_SERVICE = "xray"
+LOGS_SERIVCE = "logs"
 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$"
 
@@ -215,9 +210,9 @@ def _init_logging(
 
     for _, exporter_class in exporters.items():
         exporter_args = {}
-        log_exporter: LogExporter = _customize_logs_exporter(exporter_class(**exporter_args))
-        log_processor = _customize_log_record_processor(log_exporter)
-        provider.add_log_record_processor(log_processor)
+        _customize_log_record_processor(
+            logger_provider=provider, log_exporter=_customize_logs_exporter(exporter_class(**exporter_args))
+        )
 
     event_logger_provider = EventLoggerProvider(logger_provider=provider)
     set_event_logger_provider(event_logger_provider)
@@ -304,10 +299,11 @@ def _export_unsampled_span_for_agent_observability(trace_provider: TracerProvide
         return
 
     traces_endpoint = os.environ.get(OTEL_EXPORTER_OTLP_TRACES_ENDPOINT)
+    if traces_endpoint and _is_aws_otlp_endpoint(traces_endpoint, XRAY_SERVICE):
+        endpoint, region = _extract_endpoint_and_region_from_otlp_endpoint(traces_endpoint)
+        span_exporter = _create_aws_otlp_exporter(endpoint=endpoint, service=XRAY_SERVICE, region=region)
 
-    span_exporter = OTLPAwsSpanExporter(endpoint=traces_endpoint, logger_provider=get_logger_provider())
-
-    trace_provider.add_span_processor(BatchUnsampledSpanProcessor(span_exporter=span_exporter))
+        trace_provider.add_span_processor(BatchUnsampledSpanProcessor(span_exporter=span_exporter))
 
 
 def _is_defer_to_workers_enabled():
@@ -356,7 +352,7 @@ def _custom_import_sampler(sampler_name: str, resource: Resource) -> Sampler:
     if sampler_name is None:
         sampler_name = "parentbased_always_on"
 
-    if sampler_name == "xray":
+    if sampler_name == XRAY_SERVICE:
         # Example env var value
         # OTEL_TRACES_SAMPLER_ARG=endpoint=http://localhost:2000,polling_interval=360
         sampler_argument_env: str = os.getenv(OTEL_TRACES_SAMPLER_ARG, None)
@@ -402,50 +398,52 @@ def _customize_span_exporter(span_exporter: SpanExporter, resource: Resource) ->
             traces_endpoint = os.environ.get(AWS_XRAY_DAEMON_ADDRESS_CONFIG, "127.0.0.1:2000")
             span_exporter = OTLPUdpSpanExporter(endpoint=traces_endpoint)
 
-    if _is_aws_otlp_endpoint(traces_endpoint, "xray"):
+    if traces_endpoint and _is_aws_otlp_endpoint(traces_endpoint, XRAY_SERVICE):
         _logger.info("Detected using AWS OTLP Traces Endpoint.")
 
         if isinstance(span_exporter, OTLPSpanExporter):
-            if is_agent_observability_enabled():
-                # Span exporter needs an instance of logger provider in ai agent
-                # observability case because we need to split input/output prompts
-                # from span attributes and send them to the logs pipeline per
-                # the new Gen AI semantic convention from OTel
-                # ref: https://opentelemetry.io/docs/specs/semconv/gen-ai/gen-ai-events/
-                span_exporter = OTLPAwsSpanExporter(endpoint=traces_endpoint, logger_provider=get_logger_provider())
-            else:
-                span_exporter = OTLPAwsSpanExporter(endpoint=traces_endpoint)
+            endpoint, region = _extract_endpoint_and_region_from_otlp_endpoint(traces_endpoint)
+            return _create_aws_otlp_exporter(endpoint=endpoint, service=XRAY_SERVICE, region=region)
 
-        else:
-            _logger.warning(
-                "Improper configuration see: please export/set "
-                "OTEL_EXPORTER_OTLP_TRACES_PROTOCOL=http/protobuf and OTEL_TRACES_EXPORTER=otlp"
-            )
+        _logger.warning(
+            "Improper configuration see: please export/set "
+            "OTEL_EXPORTER_OTLP_TRACES_PROTOCOL=http/protobuf and OTEL_TRACES_EXPORTER=otlp"
+        )
 
     if not _is_application_signals_enabled():
         return span_exporter
 
     return AwsMetricAttributesSpanExporterBuilder(span_exporter, resource).build()
 
 
-def _customize_log_record_processor(log_exporter: LogExporter):
-    if isinstance(log_exporter, OTLPAwsLogExporter) and is_agent_observability_enabled():
-        return AwsCloudWatchOtlpBatchLogRecordProcessor(exporter=log_exporter)
+def _customize_log_record_processor(logger_provider: LoggerProvider, log_exporter: Optional[LogExporter]) -> None:
+    if not log_exporter:
+        return
+
+    if is_agent_observability_enabled():
+        # pylint: disable=import-outside-toplevel
+        from amazon.opentelemetry.distro.exporter.otlp.aws.logs._aws_cw_otlp_batch_log_record_processor import (
+            AwsCloudWatchOtlpBatchLogRecordProcessor,
+        )
 
-    return BatchLogRecordProcessor(exporter=log_exporter)
+        logger_provider.add_log_record_processor(AwsCloudWatchOtlpBatchLogRecordProcessor(exporter=log_exporter))
+    else:
+        logger_provider.add_log_record_processor(BatchLogRecordProcessor(exporter=log_exporter))
 
 
 def _customize_logs_exporter(log_exporter: LogExporter) -> LogExporter:
     logs_endpoint = os.environ.get(OTEL_EXPORTER_OTLP_LOGS_ENDPOINT)
 
-    if _is_aws_otlp_endpoint(logs_endpoint, "logs"):
+    if logs_endpoint and _is_aws_otlp_endpoint(logs_endpoint, LOGS_SERIVCE):
+
         _logger.info("Detected using AWS OTLP Logs Endpoint.")
 
         if isinstance(log_exporter, OTLPLogExporter) and _validate_and_fetch_logs_header().is_valid:
+            endpoint, region = _extract_endpoint_and_region_from_otlp_endpoint(logs_endpoint)
             # Setting default compression mode to Gzip as this is the behavior in upstream's
             # collector otlp http exporter:
             # https://github.com/open-telemetry/opentelemetry-collector/tree/main/exporter/otlphttpexporter
-            return OTLPAwsLogExporter(endpoint=logs_endpoint)
+            return _create_aws_otlp_exporter(endpoint=endpoint, service=LOGS_SERIVCE, region=region)
 
         _logger.warning(
             "Improper configuration see: please export/set "
@@ -514,7 +512,7 @@ def _customize_metric_exporters(
         metric_readers.append(scope_based_periodic_exporting_metric_reader)
 
     if is_emf_enabled:
-        emf_exporter = create_emf_exporter()
+        emf_exporter = _create_emf_exporter()
         if emf_exporter:
             metric_readers.append(PeriodicExportingMetricReader(emf_exporter))
 
@@ -604,17 +602,24 @@ def _is_lambda_environment():
     return AWS_LAMBDA_FUNCTION_NAME_CONFIG in os.environ
 
 
-def _is_aws_otlp_endpoint(otlp_endpoint: Optional[str] = None, service: str = "xray") -> bool:
+def _is_aws_otlp_endpoint(otlp_endpoint: Optional[str], service: str) -> bool:
     """Is the given endpoint an AWS OTLP endpoint?"""
 
-    pattern = AWS_TRACES_OTLP_ENDPOINT_PATTERN if service == "xray" else AWS_LOGS_OTLP_ENDPOINT_PATTERN
-
     if not otlp_endpoint:
         return False
 
+    pattern = AWS_TRACES_OTLP_ENDPOINT_PATTERN if service == XRAY_SERVICE else AWS_LOGS_OTLP_ENDPOINT_PATTERN
+
     return bool(re.match(pattern, otlp_endpoint.lower()))
 
 
+def _extract_endpoint_and_region_from_otlp_endpoint(endpoint: str):
+    endpoint = endpoint.lower()
+    region = endpoint.split(".")[1]
+
+    return endpoint, region
+
+
 def _validate_and_fetch_logs_header() -> OtlpLogHeaderSetting:
     """Checks if x-aws-log-group and x-aws-log-stream are present in the headers in order to send logs to
     AWS OTLP Logs endpoint."""
@@ -631,7 +636,6 @@ def _validate_and_fetch_logs_header() -> OtlpLogHeaderSetting:
     log_group = None
     log_stream = None
     namespace = None
-    filtered_log_headers_count = 0
 
     for pair in logs_headers.split(","):
         if "=" in pair:
@@ -640,14 +644,12 @@ def _validate_and_fetch_logs_header() -> OtlpLogHeaderSetting:
             value = split[1]
             if key == AWS_OTLP_LOGS_GROUP_HEADER and value:
                 log_group = value
-                filtered_log_headers_count += 1
             elif key == AWS_OTLP_LOGS_STREAM_HEADER and value:
                 log_stream = value
-                filtered_log_headers_count += 1
             elif key == AWS_EMF_METRICS_NAMESPACE and value:
                 namespace = value
 
-    is_valid = filtered_log_headers_count == 2 and log_group is not None and log_stream is not None
+    is_valid = log_group is not None and log_stream is not None
 
     if not is_valid:
         _logger.warning(
@@ -769,11 +771,12 @@ def _check_emf_exporter_enabled() -> bool:
     return True
 
 
-def create_emf_exporter():
+def _create_emf_exporter():
     """Create and configure the CloudWatch EMF exporter."""
     try:
+        session = get_aws_session()
         # Check if botocore is available before importing the EMF exporter
-        if not is_installed("botocore"):
+        if not session:
             _logger.warning("botocore is not installed. EMF exporter requires botocore")
             return None
 
@@ -788,6 +791,7 @@ def create_emf_exporter():
             return None
 
         return AwsCloudWatchEmfExporter(
+            session=session,
             namespace=log_header_setting.namespace,
             log_group_name=log_header_setting.log_group,
             log_stream_name=log_header_setting.log_stream,
@@ -796,3 +800,39 @@ def create_emf_exporter():
     except Exception as errors:
         _logger.error("Failed to create EMF exporter: %s", errors)
         return None
+
+
+def _create_aws_otlp_exporter(endpoint: str, service: str, region: str):
+    """Create and configure the AWS OTLP exporters."""
+    try:
+        session = get_aws_session()
+        # Check if botocore is available before importing the AWS exporter
+        if not session:
+            _logger.warning("Sigv4 Auth requires botocore to be enabled")
+            return None
+
+        # pylint: disable=import-outside-toplevel
+        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
+
+        if service == XRAY_SERVICE:
+            if is_agent_observability_enabled():
+                # Span exporter needs an instance of logger provider in ai agent
+                # observability case because we need to split input/output prompts
+                # from span attributes and send them to the logs pipeline per
+                # the new Gen AI semantic convention from OTel
+                # ref: https://opentelemetry.io/docs/specs/semconv/gen-ai/gen-ai-events/
+                return OTLPAwsSpanExporter(
+                    session=session, endpoint=endpoint, aws_region=region, logger_provider=get_logger_provider()
+                )
+
+            return OTLPAwsSpanExporter(session=session, endpoint=endpoint, aws_region=region)
+
+        if service == LOGS_SERIVCE:
+            return OTLPAwsLogExporter(session=session, aws_region=region)
+
+        return None
+    # pylint: disable=broad-exception-caught
+    except Exception as errors:
+        _logger.error("Failed to create AWS OTLP exporter: %s", errors)
+        return None

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

@@ -86,8 +86,9 @@ def _configure(self, **kwargs):
             # Set GenAI capture content default
             os.environ.setdefault(OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT, "true")
 
-            # Set OTLP endpoints with AWS region if not already set
             region = get_aws_region()
+
+            # Set OTLP endpoints with AWS region if not already set
             if region:
                 os.environ.setdefault(
                     OTEL_EXPORTER_OTLP_TRACES_ENDPOINT, f"https://xray.{region}.amazonaws.com/v1/traces"

aws-opentelemetry-distro/src/amazon/opentelemetry/distro/exporter/aws/metrics/_cloudwatch_log_client.py

@@ -8,8 +8,8 @@
 import uuid
 from typing import Any, Dict, List, Optional
 
-import botocore.session
 from botocore.exceptions import ClientError
+from botocore.session import Session
 
 logger = logging.getLogger(__name__)
 
@@ -90,6 +90,7 @@ class CloudWatchLogClient:
     def __init__(
         self,
         log_group_name: str,
+        session: Session,
         log_stream_name: Optional[str] = None,
         aws_region: Optional[str] = None,
         **kwargs,
@@ -105,8 +106,6 @@ def __init__(
         """
         self.log_group_name = log_group_name
         self.log_stream_name = log_stream_name or self._generate_log_stream_name()
-
-        session = botocore.session.Session()
         self.logs_client = session.create_client("logs", region_name=aws_region, **kwargs)
 
         # Event batch to store logs before sending to CloudWatch