implement sigv4 support for logs exporter

Author: liustve

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,13 @@
     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.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 +38,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,6 +87,8 @@
 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"
+OTEL_EXPORTER_OTLP_LOGS_ENDPOINT = "OTEL_EXPORTER_OTLP_LOGS_ENDPOINT"
+
 XRAY_OTLP_ENDPOINT_PATTERN = r"https://xray\.([a-z0-9-]+)\.amazonaws\.com/v1/traces$"
 # UDP package size is not larger than 64KB
 LAMBDA_SPAN_EXPORT_BATCH_SIZE = 10
@@ -160,6 +165,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 = {}
+        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,
@@ -338,6 +361,10 @@ 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:
+    return OTLPAwsLogExporter(endpoint=os.getenv(OTEL_EXPORTER_OTLP_LOGS_ENDPOINT))
+
+
 def _customize_span_processors(provider: TracerProvider, resource: Resource) -> None:
     # Add LambdaSpanProcessor to list of processors regardless of application signals.
     if _is_lambda_environment():

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

@@ -0,0 +1,73 @@
+import logging
+from abc import ABC, abstractmethod
+from typing import Optional
+
+import requests
+
+from amazon.opentelemetry.distro._utils import is_installed
+
+_logger = logging.getLogger(__name__)
+
+
+class OTLPBaseAwsExporter(ABC):
+    def __init__(
+        self,
+        endpoint: Optional[str] = None,
+        rsession: Optional[requests.Session] = None,
+    ):
+
+        self._aws_region = None
+        self._has_required_dependencies = False
+        self._endpoint = endpoint
+        self._session = rsession
+
+        # 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 endpoint 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()
+
+            # Assumes only valid endpoints passed are of XRay OTLP format.
+            # The only usecase for this class would be for ADOT Python Auto Instrumentation and that already validates
+            # the endpoint to be an XRay OTLP endpoint.
+            self._aws_region = endpoint.split(".")[1]
+            self._has_required_dependencies = True
+
+        else:
+            _logger.error(
+                "botocore is required to export to %s. Please install it using `pip install botocore`",
+                endpoint,
+            )
+
+    @abstractmethod
+    def get_service(self):
+        pass
+
+    def sigv4_auth(self, serialized_data):
+        if self._has_required_dependencies:
+            request = self._boto_aws_request.AWSRequest(
+                method="POST",
+                url=self._endpoint,
+                data=serialized_data,
+                headers={"Content-Type": "application/x-protobuf"},
+            )
+
+            credentials = self._boto_session.get_credentials()
+
+            if credentials is not None:
+                signer = self._boto_auth.SigV4Auth(credentials, self.get_service(), self._aws_region)
+
+                try:
+                    signer.add_auth(request)
+                    self._session.headers.update(dict(request.headers))
+
+                except Exception as signing_error:  # pylint: disable=broad-except
+                    _logger.error("Failed to sign request: %s", signing_error)
+        else:
+            _logger.debug("botocore is not installed. Failed to sign request to export traces to: %s", self._endpoint)

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

@@ -0,0 +1,43 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+from typing import Dict, Optional
+
+import requests
+from opentelemetry.exporter.otlp.proto.http import Compression
+
+from amazon.opentelemetry.distro.exporter.otlp.aws.common.otlp_aws_exporter import OTLPBaseAwsExporter
+from opentelemetry.exporter.otlp.proto.http._log_exporter import OTLPLogExporter
+
+
+class OTLPAwsLogExporter(OTLPLogExporter, OTLPBaseAwsExporter):
+    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,
+        session: Optional[requests.Session] = None,
+    ):
+        OTLPBaseAwsExporter.__init__(self, endpoint, session)
+        OTLPLogExporter.__init__(
+            self,
+            endpoint,
+            certificate_file,
+            client_key_file,
+            client_certificate_file,
+            headers,
+            timeout,
+            compression,
+            session,
+        )
+
+    def get_service(self):
+        return "logs"
+
+    def _export(self, serialized_data: bytes):
+        self.sigv4_auth(serialized_data)
+        return OTLPLogExporter._export(self, serialized_data)

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

@@ -0,0 +1,54 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+from typing import Dict, Optional
+
+import requests
+
+from amazon.opentelemetry.distro.exporter.otlp.aws.common.otlp_aws_exporter import OTLPBaseAwsExporter
+from opentelemetry.exporter.otlp.proto.http import Compression
+from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
+
+
+class OTLPAwsSpanExporter(OTLPSpanExporter, OTLPBaseAwsExporter):
+    """
+    This exporter extends the functionality of the OTLPSpanExporter to allow spans to be exported to the
+    XRay 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
+    """
+
+    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: Optional[requests.Session] = None,
+    ):
+
+        OTLPBaseAwsExporter.__init__(self, endpoint, rsession)
+        OTLPSpanExporter.__init__(
+            self,
+            endpoint,
+            certificate_file,
+            client_key_file,
+            client_certificate_file,
+            headers,
+            timeout,
+            compression,
+            rsession,
+        )
+
+    def get_service(self):
+        return "xray"
+
+    # Overrides upstream's private implementation of _export. All behaviors are
+    # the same except if the endpoint is an XRay OTLP endpoint, we will sign the request
+    # with SigV4 in headers before sending it to the endpoint. Otherwise, we will skip signing.
+    def _export(self, serialized_data: bytes):
+        self.sigv4_auth(serialized_data)
+        return OTLPSpanExporter._export(self, serialized_data)