Add Application Signals runtime metrics

Author: bjrara

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

@@ -22,6 +22,7 @@
     AWS_SQS_QUEUE_NAME,
     AWS_SQS_QUEUE_URL,
 )
+from amazon.opentelemetry.distro._aws_resource_attribute_configurator import get_service_attribute
 from amazon.opentelemetry.distro._aws_span_processing_util import (
     GEN_AI_REQUEST_MODEL,
     LOCAL_ROOT,
@@ -30,7 +31,6 @@
     UNKNOWN_OPERATION,
     UNKNOWN_REMOTE_OPERATION,
     UNKNOWN_REMOTE_SERVICE,
-    UNKNOWN_SERVICE,
     extract_api_path_value,
     get_egress_operation,
     get_ingress_operation,
@@ -47,12 +47,11 @@
     MetricAttributeGenerator,
 )
 from amazon.opentelemetry.distro.sqs_url_parser import SqsUrlParser
-from opentelemetry.sdk.resources import Resource, ResourceAttributes
+from opentelemetry.sdk.resources import Resource
 from opentelemetry.sdk.trace import BoundedAttributes, ReadableSpan
 from opentelemetry.semconv.trace import SpanAttributes
 
 # Pertinent OTEL attribute keys
-_SERVICE_NAME: str = ResourceAttributes.SERVICE_NAME
 _DB_CONNECTION_STRING: str = SpanAttributes.DB_CONNECTION_STRING
 _DB_NAME: str = SpanAttributes.DB_NAME
 _DB_OPERATION: str = SpanAttributes.DB_OPERATION
@@ -92,10 +91,6 @@
 # Special DEPENDENCY attribute value if GRAPHQL_OPERATION_TYPE attribute key is present.
 _GRAPHQL: str = "graphql"
 
-# As per https://opentelemetry.io/docs/specs/semconv/resource/#service, if service name is not specified, SDK defaults
-# the service name to unknown_service:<process name> or just unknown_service.
-_OTEL_UNKNOWN_SERVICE_PREFIX: str = "unknown_service"
-
 _logger: Logger = getLogger(__name__)
 
 
@@ -141,15 +136,11 @@ def _generate_dependency_metric_attributes(span: ReadableSpan, resource: Resourc
 
 
 def _set_service(resource: Resource, span: ReadableSpan, attributes: BoundedAttributes) -> None:
-    """Service is always derived from SERVICE_NAME"""
-    service: str = resource.attributes.get(_SERVICE_NAME)
-
-    # In practice the service name is never None, but we can be defensive here.
-    if service is None or service.startswith(_OTEL_UNKNOWN_SERVICE_PREFIX):
+    service_name, is_unknown = get_service_attribute(resource)
+    if is_unknown:
         _log_unknown_attribute(AWS_LOCAL_SERVICE, span)
-        service = UNKNOWN_SERVICE
 
-    attributes[AWS_LOCAL_SERVICE] = service
+    attributes[AWS_LOCAL_SERVICE] = service_name
 
 
 def _set_ingress_operation(span: ReadableSpan, attributes: BoundedAttributes) -> None:

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

@@ -0,0 +1,19 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+from amazon.opentelemetry.distro._aws_span_processing_util import UNKNOWN_SERVICE
+from opentelemetry.sdk.resources import SERVICE_NAME, Resource
+
+# As per https://opentelemetry.io/docs/specs/semconv/resource/#service, if service name is not specified, SDK defaults
+# the service name to unknown_service:<process name> or just unknown_service.
+_OTEL_UNKNOWN_SERVICE_PREFIX: str = "unknown_service"
+
+
+def get_service_attribute(resource: Resource) -> (str, bool):
+    """Service is always derived from SERVICE_NAME"""
+    service: str = resource.attributes.get(SERVICE_NAME)
+
+    # In practice the service name is never None, but we can be defensive here.
+    if service is None or service.startswith(_OTEL_UNKNOWN_SERVICE_PREFIX):
+        return UNKNOWN_SERVICE, True
+
+    return service, False

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

@@ -3,11 +3,13 @@
 # Modifications Copyright The OpenTelemetry Authors. Licensed under the Apache License 2.0 License.
 import os
 from logging import Logger, getLogger
-from typing import ClassVar, Dict, Type
+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._aws_resource_attribute_configurator import get_service_attribute
 from amazon.opentelemetry.distro.always_record_sampler import AlwaysRecordSampler
 from amazon.opentelemetry.distro.attribute_propagating_span_processor_builder import (
     AttributePropagatingSpanProcessorBuilder,
@@ -18,7 +20,9 @@
 from amazon.opentelemetry.distro.aws_span_metrics_processor_builder import AwsSpanMetricsProcessorBuilder
 from amazon.opentelemetry.distro.otlp_udp_exporter import OTLPUdpMetricExporter, OTLPUdpSpanExporter
 from amazon.opentelemetry.distro.sampler.aws_xray_remote_sampler import AwsXRayRemoteSampler
+from amazon.opentelemetry.distro.scope_based_exporter import ScopeBasedPeriodicExportingMetricReader
 from opentelemetry.exporter.otlp.proto.http.metric_exporter import OTLPMetricExporter as OTLPHttpOTLPMetricExporter
+from opentelemetry.metrics import set_meter_provider
 from opentelemetry.sdk._configuration import (
     _get_exporter_names,
     _get_id_generator,
@@ -27,7 +31,6 @@
     _import_id_generator,
     _import_sampler,
     _init_logging,
-    _init_metrics,
     _OTelSDKConfigurator,
 )
 from opentelemetry.sdk.environment_variables import (
@@ -48,7 +51,13 @@
     ObservableUpDownCounter,
     UpDownCounter,
 )
-from opentelemetry.sdk.metrics.export import AggregationTemporality, PeriodicExportingMetricReader
+from opentelemetry.sdk.metrics.export import (
+    AggregationTemporality,
+    MetricExporter,
+    MetricReader,
+    PeriodicExportingMetricReader,
+)
+from opentelemetry.sdk.metrics.view import DefaultAggregation, DropAggregation, View
 from opentelemetry.sdk.resources import Resource, get_aggregated_resources
 from opentelemetry.sdk.trace import TracerProvider
 from opentelemetry.sdk.trace.export import BatchSpanProcessor, SpanExporter
@@ -57,9 +66,10 @@
 from opentelemetry.semconv.resource import ResourceAttributes
 from opentelemetry.trace import set_tracer_provider
 
-APP_SIGNALS_ENABLED_CONFIG = "OTEL_AWS_APP_SIGNALS_ENABLED"
+DEPRECATED_APP_SIGNALS_ENABLED_CONFIG = "OTEL_AWS_APP_SIGNALS_ENABLED"
 APPLICATION_SIGNALS_ENABLED_CONFIG = "OTEL_AWS_APPLICATION_SIGNALS_ENABLED"
-APP_SIGNALS_EXPORTER_ENDPOINT_CONFIG = "OTEL_AWS_APP_SIGNALS_EXPORTER_ENDPOINT"
+APPLICATION_SIGNALS_RUNTIME_ENABLED_CONFIG = "OTEL_AWS_APPLICATION_SIGNALS_RUNTIME_ENABLED"
+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"
 DEFAULT_METRIC_EXPORT_INTERVAL = 60000.0
@@ -111,13 +121,16 @@ def _initialize_components():
 
     auto_resource: Dict[str, any] = {}
     auto_resource = _customize_versions(auto_resource)
-    resource = get_aggregated_resources(
-        [
-            AwsEc2ResourceDetector(),
-            AwsEksResourceDetector(),
-            AwsEcsResourceDetector(),
-        ]
-    ).merge(Resource.create(auto_resource))
+    # auto_resource = _set_aws_attributes(auto_resource)
+    resource = _customize_resource(
+        get_aggregated_resources(
+            [
+                AwsEc2ResourceDetector(),
+                AwsEksResourceDetector(),
+                AwsEcsResourceDetector(),
+            ]
+        ).merge(Resource.create(auto_resource))
+    )
 
     sampler_name = _get_sampler()
     sampler = _custom_import_sampler(sampler_name, resource)
@@ -159,6 +172,27 @@ def _init_tracing(
     set_tracer_provider(trace_provider)
 
 
+def _init_metrics(
+    exporters_or_readers: Dict[str, Union[Type[MetricExporter], Type[MetricReader]]],
+    resource: Resource = None,
+):
+    metric_readers = []
+    views = []
+
+    for _, exporter_or_reader_class in exporters_or_readers.items():
+        exporter_args = {}
+
+        if issubclass(exporter_or_reader_class, MetricReader):
+            metric_readers.append(exporter_or_reader_class(**exporter_args))
+        else:
+            metric_readers.append(PeriodicExportingMetricReader(exporter_or_reader_class(**exporter_args)))
+
+    _customize_metric_exporters(metric_readers, views)
+
+    provider = MeterProvider(resource=resource, metric_readers=metric_readers, views=views)
+    set_meter_provider(provider)
+
+
 # END The OpenTelemetry Authors code
 
 
@@ -264,14 +298,9 @@ def _customize_span_processors(provider: TracerProvider, resource: Resource) ->
     # Construct meterProvider
     _logger.info("AWS Application Signals enabled")
     otel_metric_exporter = ApplicationSignalsExporterProvider().create_exporter()
-    export_interval_millis = float(os.environ.get(METRIC_EXPORT_INTERVAL_CONFIG, DEFAULT_METRIC_EXPORT_INTERVAL))
-    _logger.debug("Span Metrics export interval: %s", export_interval_millis)
-    # Cap export interval to 60 seconds. This is currently required for metrics-trace correlation to work correctly.
-    if export_interval_millis > DEFAULT_METRIC_EXPORT_INTERVAL:
-        export_interval_millis = DEFAULT_METRIC_EXPORT_INTERVAL
-        _logger.info("AWS Application Signals metrics export interval capped to %s", export_interval_millis)
+
     periodic_exporting_metric_reader = PeriodicExportingMetricReader(
-        exporter=otel_metric_exporter, export_interval_millis=export_interval_millis
+        exporter=otel_metric_exporter, export_interval_millis=_get_metric_export_interval()
     )
     meter_provider: MeterProvider = MeterProvider(resource=resource, metric_readers=[periodic_exporting_metric_reader])
     # Construct and set application signals metrics processor
@@ -280,25 +309,68 @@ def _customize_span_processors(provider: TracerProvider, resource: Resource) ->
     return
 
 
+def _customize_metric_exporters(metric_readers: List[MetricReader], views: List[View]) -> None:
+    if _is_application_signals_runtime_enabled():
+        system_metrics_scope_name = "opentelemetry.instrumentation.system_metrics"
+        if 0 == len(metric_readers):
+            _logger.info("Registered scope %s", system_metrics_scope_name)
+            views.append(View(meter_name=system_metrics_scope_name, aggregation=DefaultAggregation()))
+            views.append(View(instrument_name="*", aggregation=DropAggregation()))
+
+        otel_metric_exporter = ApplicationSignalsExporterProvider().create_exporter()
+        scope_based_periodic_exporting_metric_reader = ScopeBasedPeriodicExportingMetricReader(
+            exporter=otel_metric_exporter,
+            export_interval_millis=_get_metric_export_interval(),
+            registered_scope_names={system_metrics_scope_name},
+        )
+        metric_readers.append(scope_based_periodic_exporting_metric_reader)
+
+
 def _customize_versions(auto_resource: Dict[str, any]) -> Dict[str, any]:
     distro_version = version("aws-opentelemetry-distro")
     auto_resource[ResourceAttributes.TELEMETRY_AUTO_VERSION] = distro_version + "-aws"
     _logger.debug("aws-opentelementry-distro - version: %s", auto_resource[ResourceAttributes.TELEMETRY_AUTO_VERSION])
     return auto_resource
 
 
+def _customize_resource(resource: Resource) -> Resource:
+    service_name, is_unknown = get_service_attribute(resource)
+    if is_unknown:
+        _logger.debug("No valid service name found")
+
+    return resource.merge(Resource.create({AWS_LOCAL_SERVICE: service_name}))
+
+
 def _is_application_signals_enabled():
     return (
-        os.environ.get(APPLICATION_SIGNALS_ENABLED_CONFIG, os.environ.get(APP_SIGNALS_ENABLED_CONFIG, "false")).lower()
+        os.environ.get(
+            APPLICATION_SIGNALS_ENABLED_CONFIG, os.environ.get(DEPRECATED_APP_SIGNALS_ENABLED_CONFIG, "false")
+        ).lower()
         == "true"
     )
 
 
+def _is_application_signals_runtime_enabled():
+    return _is_application_signals_enabled() and (
+        os.environ.get(APPLICATION_SIGNALS_RUNTIME_ENABLED_CONFIG, "true").lower() == "true"
+    )
+
+
 def _is_lambda_environment():
     # detect if running in AWS Lambda environment
     return AWS_LAMBDA_FUNCTION_NAME_CONFIG in os.environ
 
 
+def _get_metric_export_interval():
+    export_interval_millis = float(os.environ.get(METRIC_EXPORT_INTERVAL_CONFIG, DEFAULT_METRIC_EXPORT_INTERVAL))
+    _logger.debug("Span Metrics export interval: %s", export_interval_millis)
+    # Cap export interval to 60 seconds. This is currently required for metrics-trace correlation to work correctly.
+    if export_interval_millis > DEFAULT_METRIC_EXPORT_INTERVAL:
+        export_interval_millis = DEFAULT_METRIC_EXPORT_INTERVAL
+        _logger.info("AWS Application Signals metrics export interval capped to %s", export_interval_millis)
+    return export_interval_millis
+
+
 class ApplicationSignalsExporterProvider:
     _instance: ClassVar["ApplicationSignalsExporterProvider"] = None
 
@@ -334,7 +406,7 @@ def create_exporter(self):
         if protocol == "http/protobuf":
             application_signals_endpoint = os.environ.get(
                 APPLICATION_SIGNALS_EXPORTER_ENDPOINT_CONFIG,
-                os.environ.get(APP_SIGNALS_EXPORTER_ENDPOINT_CONFIG, "http://localhost:4316/v1/metrics"),
+                os.environ.get(DEPRECATED_APP_SIGNALS_EXPORTER_ENDPOINT_CONFIG, "http://localhost:4316/v1/metrics"),
             )
             _logger.debug("AWS Application Signals export endpoint: %s", application_signals_endpoint)
             return OTLPHttpOTLPMetricExporter(
@@ -350,7 +422,7 @@ def create_exporter(self):
 
             application_signals_endpoint = os.environ.get(
                 APPLICATION_SIGNALS_EXPORTER_ENDPOINT_CONFIG,
-                os.environ.get(APP_SIGNALS_EXPORTER_ENDPOINT_CONFIG, "localhost:4315"),
+                os.environ.get(DEPRECATED_APP_SIGNALS_EXPORTER_ENDPOINT_CONFIG, "localhost:4315"),
             )
             _logger.debug("AWS Application Signals export endpoint: %s", application_signals_endpoint)
             return OTLPGrpcOTLPMetricExporter(

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

@@ -0,0 +1,54 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+from logging import Logger, getLogger
+from typing import Optional, Set
+
+from opentelemetry.context import _SUPPRESS_INSTRUMENTATION_KEY, attach, detach, set_value
+from opentelemetry.sdk.metrics.export import MetricExporter, MetricsData, PeriodicExportingMetricReader, ResourceMetrics
+
+_logger: Logger = getLogger(__name__)
+
+
+class ScopeBasedPeriodicExportingMetricReader(PeriodicExportingMetricReader):
+
+    def __init__(
+        self,
+        exporter: MetricExporter,
+        export_interval_millis: Optional[float] = None,
+        export_timeout_millis: Optional[float] = None,
+        registered_scope_names: Set[str] = None,
+    ):
+        super().__init__(exporter, export_interval_millis, export_timeout_millis)
+        self._registered_scope_names = registered_scope_names
+
+    def _receive_metrics(
+        self,
+        metrics_data: MetricsData,
+        timeout_millis: float = 10_000,
+        **kwargs,
+    ) -> None:
+
+        token = attach(set_value(_SUPPRESS_INSTRUMENTATION_KEY, True))
+        # pylint: disable=broad-exception-caught,invalid-name
+        try:
+            with self._export_lock:
+                exporting_resource_metrics = []
+                for metric in metrics_data.resource_metrics:
+                    exporting_scope_metrics = []
+                    for scope_metric in metric.scope_metrics:
+                        if scope_metric.scope.name in self._registered_scope_names:
+                            exporting_scope_metrics.append(scope_metric)
+                    if len(exporting_scope_metrics) > 0:
+                        exporting_resource_metrics.append(
+                            ResourceMetrics(
+                                resource=metric.resource,
+                                scope_metrics=exporting_scope_metrics,
+                                schema_url=metric.schema_url,
+                            )
+                        )
+                if len(exporting_resource_metrics) > 0:
+                    new_metrics_data = MetricsData(resource_metrics=exporting_resource_metrics)
+                    self._exporter.export(new_metrics_data, timeout_millis=timeout_millis)
+        except Exception as e:
+            _logger.exception("Exception while exporting metrics %s", str(e))
+        detach(token)