SigV4 Authentication Support for OTLP HTTP Logs Exporter

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

@@ -3,7 +3,7 @@
 # Modifications Copyright The OpenTelemetry Authors. Licensed under the Apache License 2.0 License.
 import os
 import re
-from logging import Logger, getLogger
+from logging import NOTSET, Logger, getLogger
 from typing import ClassVar, Dict, List, Type, Union
 
 from importlib_metadata import version
@@ -21,11 +21,14 @@
     AwsMetricAttributesSpanExporterBuilder,
 )
 from amazon.opentelemetry.distro.aws_span_metrics_processor_builder import AwsSpanMetricsProcessorBuilder
-from amazon.opentelemetry.distro.otlp_aws_span_exporter import OTLPAwsSpanExporter
+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.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
 from opentelemetry.metrics import set_meter_provider
@@ -36,9 +39,10 @@
     _import_exporters,
     _import_id_generator,
     _import_sampler,
-    _init_logging,
     _OTelSDKConfigurator,
 )
+from opentelemetry.sdk._logs import LoggerProvider, LoggingHandler
+from opentelemetry.sdk._logs.export import BatchLogRecordProcessor, LogExporter
 from opentelemetry.sdk.environment_variables import (
     _OTEL_PYTHON_LOGGING_AUTO_INSTRUMENTATION_ENABLED,
     OTEL_EXPORTER_OTLP_METRICS_PROTOCOL,
@@ -84,7 +88,11 @@
 OTEL_AWS_PYTHON_DEFER_TO_WORKERS_ENABLED_CONFIG = "OTEL_AWS_PYTHON_DEFER_TO_WORKERS_ENABLED"
 SYSTEM_METRICS_INSTRUMENTATION_SCOPE_NAME = "opentelemetry.instrumentation.system_metrics"
 OTEL_EXPORTER_OTLP_TRACES_ENDPOINT = "OTEL_EXPORTER_OTLP_TRACES_ENDPOINT"
-XRAY_OTLP_ENDPOINT_PATTERN = r"https://xray\.([a-z0-9-]+)\.amazonaws\.com/v1/traces$"
+OTEL_EXPORTER_OTLP_LOGS_ENDPOINT = "OTEL_EXPORTER_OTLP_LOGS_ENDPOINT"
+
+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$"
+
 # UDP package size is not larger than 64KB
 LAMBDA_SPAN_EXPORT_BATCH_SIZE = 10
 
@@ -160,6 +168,24 @@ def _initialize_components():
         _init_logging(log_exporters, resource)
 
 
+def _init_logging(
+    exporters: Dict[str, Type[LogExporter]],
+    resource: Resource = None,
+):
+
+    provider = LoggerProvider(resource=resource)
+    set_logger_provider(provider)
+
+    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))
+
+    handler = LoggingHandler(level=NOTSET, logger_provider=provider)
+
+    getLogger().addHandler(handler)
+
+
 def _init_tracing(
     exporters: Dict[str, Type[SpanExporter]],
     id_generator: IdGenerator = None,
@@ -177,7 +203,7 @@ def _init_tracing(
     for _, exporter_class in exporters.items():
         exporter_args: Dict[str, any] = {}
         span_exporter: SpanExporter = exporter_class(**exporter_args)
-        span_exporter = _customize_exporter(span_exporter, resource)
+        span_exporter = _customize_span_exporter(span_exporter, resource)
         trace_provider.add_span_processor(
             BatchSpanProcessor(span_exporter=span_exporter, max_export_batch_size=_span_export_batch_size())
         )
@@ -312,15 +338,14 @@ def _customize_sampler(sampler: Sampler) -> Sampler:
     return AlwaysRecordSampler(sampler)
 
 
-def _customize_exporter(span_exporter: SpanExporter, resource: Resource) -> SpanExporter:
+def _customize_span_exporter(span_exporter: SpanExporter, resource: Resource) -> SpanExporter:
     if _is_lambda_environment():
         # Override OTLP http default endpoint to UDP
         if isinstance(span_exporter, OTLPSpanExporter) and os.getenv(OTEL_EXPORTER_OTLP_TRACES_ENDPOINT) is None:
             traces_endpoint = os.environ.get(AWS_XRAY_DAEMON_ADDRESS_CONFIG, "127.0.0.1:2000")
             span_exporter = OTLPUdpSpanExporter(endpoint=traces_endpoint)
 
-    if is_xray_otlp_endpoint(os.environ.get(OTEL_EXPORTER_OTLP_TRACES_ENDPOINT)):
-        # TODO: Change this url once doc writer has added a section for using SigV4 without collector
+    if is_aws_otlp_endpoint(os.environ.get(OTEL_EXPORTER_OTLP_TRACES_ENDPOINT), "xray"):
         _logger.info("Detected using AWS OTLP XRay Endpoint.")
 
         if isinstance(span_exporter, OTLPSpanExporter):
@@ -338,6 +363,16 @@ def _customize_exporter(span_exporter: SpanExporter, resource: Resource) -> Span
     return AwsMetricAttributesSpanExporterBuilder(span_exporter, resource).build()
 
 
+def _customize_logs_exporter(log_exporter: LogExporter, resource: Resource) -> LogExporter:
+    if is_aws_otlp_endpoint(os.environ.get(OTEL_EXPORTER_OTLP_LOGS_ENDPOINT), "logs"):
+        _logger.info("Detected using AWS OTLP Logs Endpoint.")
+
+        if isinstance(log_exporter, OTLPLogExporter):
+            return OTLPAwsLogExporter(endpoint=os.getenv(OTEL_EXPORTER_OTLP_LOGS_ENDPOINT))
+
+    return log_exporter
+
+
 def _customize_span_processors(provider: TracerProvider, resource: Resource) -> None:
     # Add LambdaSpanProcessor to list of processors regardless of application signals.
     if _is_lambda_environment():
@@ -458,12 +493,15 @@ def _is_lambda_environment():
     return AWS_LAMBDA_FUNCTION_NAME_CONFIG in os.environ
 
 
-def is_xray_otlp_endpoint(otlp_endpoint: str = None) -> bool:
-    """Is the given endpoint the XRay OTLP endpoint?"""
+def is_aws_otlp_endpoint(otlp_endpoint: str = None, service: str = "xray") -> 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
 
-    return bool(re.match(XRAY_OTLP_ENDPOINT_PATTERN, otlp_endpoint.lower()))
+    return bool(re.match(pattern, otlp_endpoint.lower()))
 
 
 def _get_metric_export_interval():

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

@@ -0,0 +1,68 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+import logging
+
+import requests
+
+from amazon.opentelemetry.distro._utils import is_installed
+
+_logger = logging.getLogger(__name__)
+
+
+class AwsAuthSession(requests.Session):
+
+    def __init__(self, aws_region, service):
+
+        self._has_required_dependencies = False
+
+        # Requires botocore to be installed to sign the headers. However,
+        # some users might not need to use this exporter. In order not conflict
+        # with existing behavior, we check for botocore before initializing this exporter.
+
+        if aws_region and service and is_installed("botocore"):
+            # pylint: disable=import-outside-toplevel
+            from botocore import auth, awsrequest, session
+
+            self._boto_auth = auth
+            self._boto_aws_request = awsrequest
+            self._boto_session = session.Session()
+
+            self._aws_region = aws_region
+            self._service = service
+            self._has_required_dependencies = True
+
+        else:
+            _logger.error(
+                "botocore is required to enable SigV4 Authentication. Please install it using `pip install botocore`",
+            )
+
+        super().__init__()
+
+    def request(self, method, url, *args, data=None, headers=None, **kwargs):
+        if self._has_required_dependencies:
+
+            credentials = self._boto_session.get_credentials()
+
+            if credentials is not None:
+                signer = self._boto_auth.SigV4Auth(credentials, self._service, self._aws_region)
+
+                request = self._boto_aws_request.AWSRequest(
+                    method="POST",
+                    url=url,
+                    data=data,
+                    headers={"Content-Type": "application/x-protobuf"},
+                )
+
+                try:
+                    signer.add_auth(request)
+
+                    if headers is None:
+                        headers = {}
+
+                    headers.update(dict(request.headers))
+
+                except Exception as signing_error:  # pylint: disable=broad-except
+                    _logger.error("Failed to sign request: %s", signing_error)
+
+        return super().request(method=method, url=url, *args, data=data, headers=headers, **kwargs)

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

@@ -0,0 +1,45 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+from typing import Dict, Optional
+
+from amazon.opentelemetry.distro.exporter.otlp.aws.common.aws_auth_session import AwsAuthSession
+from opentelemetry.exporter.otlp.proto.http import Compression
+from opentelemetry.exporter.otlp.proto.http._log_exporter import OTLPLogExporter
+
+
+class OTLPAwsLogExporter(OTLPLogExporter):
+    """
+    This exporter extends the functionality of the OTLPLogExporter to allow logs to be exported to the
+    CloudWatch Logs OTLP endpoint https://logs.[AWSRegion].amazonaws.com/v1/logs. Utilizes the botocore
+    library to sign and directly inject SigV4 Authentication to the exported request's headers.
+
+    https://docs.aws.amazon.com/AmazonCloudWatch/latest/monitoring/CloudWatch-OTLPEndpoint.html
+    """
+
+    # pylint: disable=too-many-arguments
+    def __init__(
+        self,
+        endpoint: Optional[str] = None,
+        certificate_file: Optional[str] = None,
+        client_key_file: Optional[str] = None,
+        client_certificate_file: Optional[str] = None,
+        headers: Optional[Dict[str, str]] = None,
+        timeout: Optional[int] = None,
+        compression: Optional[Compression] = None,
+    ):
+        rsession = None
+
+        if endpoint:
+            rsession = AwsAuthSession(endpoint.split(".")[1], "logs")
+
+        super().__init__(
+            endpoint=endpoint,
+            certificate_file=certificate_file,
+            client_key_file=client_key_file,
+            client_certificate_file=client_certificate_file,
+            headers=headers,
+            timeout=timeout,
+            compression=compression,
+            session=rsession,
+        )

aws-opentelemetry-distro/src/amazon/opentelemetry/distro/exporter/otlp/aws/traces/otlp_aws_span_exporter.py

@@ -0,0 +1,45 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+from typing import Dict, Optional
+
+from amazon.opentelemetry.distro.exporter.otlp.aws.common.aws_auth_session import AwsAuthSession
+from opentelemetry.exporter.otlp.proto.http import Compression
+from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
+
+
+class OTLPAwsSpanExporter(OTLPSpanExporter):
+    """
+    This exporter extends the functionality of the OTLPSpanExporter to allow spans to be exported to the
+    XRay Traces OTLP endpoint https://xray.[AWSRegion].amazonaws.com/v1/traces. Utilizes the botocore
+    library to sign and directly inject SigV4 Authentication to the exported request's headers.
+
+    https://docs.aws.amazon.com/AmazonCloudWatch/latest/monitoring/CloudWatch-OTLPEndpoint.html
+    """
+
+    # pylint: disable=too-many-arguments
+    def __init__(
+        self,
+        endpoint: Optional[str] = None,
+        certificate_file: Optional[str] = None,
+        client_key_file: Optional[str] = None,
+        client_certificate_file: Optional[str] = None,
+        headers: Optional[Dict[str, str]] = None,
+        timeout: Optional[int] = None,
+        compression: Optional[Compression] = None,
+    ):
+        rsession = None
+
+        if endpoint:
+            rsession = AwsAuthSession(endpoint.split(".")[1], "xray")
+
+        super().__init__(
+            endpoint=endpoint,
+            certificate_file=certificate_file,
+            client_key_file=client_key_file,
+            client_certificate_file=client_certificate_file,
+            headers=headers,
+            timeout=timeout,
+            compression=compression,
+            session=rsession,
+        )