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,
@@ -52,7 +52,6 @@
 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,16 @@
+from amazon.opentelemetry.distro._aws_span_processing_util import UNKNOWN_SERVICE
+from opentelemetry.sdk.resources import Resource, SERVICE_NAME
+
+# 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,18 +3,22 @@
 # 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, Type, Union
 
+from amazon.opentelemetry.distro._aws_resource_attribute_configurator import get_service_attribute
 from importlib_metadata import version
+from opentelemetry.metrics import set_meter_provider
+from opentelemetry.sdk.metrics.view import View, DefaultAggregation, DropAggregation
 from typing_extensions import override
-
+from amazon.opentelemetry.distro._aws_attribute_keys import AWS_LOCAL_SERVICE
 from amazon.opentelemetry.distro.always_record_sampler import AlwaysRecordSampler
 from amazon.opentelemetry.distro.attribute_propagating_span_processor_builder import (
     AttributePropagatingSpanProcessorBuilder,
 )
 from amazon.opentelemetry.distro.aws_metric_attributes_span_exporter_builder import (
     AwsMetricAttributesSpanExporterBuilder,
 )
+from amazon.opentelemetry.distro.scope_based_exporter import ScopeBasedPeriodicExportingMetricReader
 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
@@ -27,7 +31,6 @@
     _import_id_generator,
     _import_sampler,
     _init_logging,
-    _init_metrics,
     _OTelSDKConfigurator,
 )
 from opentelemetry.sdk.environment_variables import (
@@ -48,18 +51,19 @@
     ObservableUpDownCounter,
     UpDownCounter,
 )
-from opentelemetry.sdk.metrics.export import AggregationTemporality, PeriodicExportingMetricReader
-from opentelemetry.sdk.resources import Resource, get_aggregated_resources
+from opentelemetry.sdk.metrics.export import MetricExporter, MetricReader, AggregationTemporality, PeriodicExportingMetricReader
+from opentelemetry.sdk.resources import Resource, get_aggregated_resources, SERVICE_NAME
 from opentelemetry.sdk.trace import TracerProvider
 from opentelemetry.sdk.trace.export import BatchSpanProcessor, SpanExporter
 from opentelemetry.sdk.trace.id_generator import IdGenerator
 from opentelemetry.sdk.trace.sampling import Sampler
 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
@@ -105,13 +109,15 @@ 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)
@@ -153,6 +159,49 @@ 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)
+                )
+            )
+
+    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)
+
+    provider = MeterProvider(resource=resource, metric_readers=metric_readers, views=views)
+    set_meter_provider(provider)
+
+
 # END The OpenTelemetry Authors code
 
 
@@ -237,14 +286,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
@@ -260,9 +304,25 @@ def _customize_versions(auto_resource: Dict[str, any]) -> Dict[str, any]:
     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"
     )
 
@@ -272,6 +332,16 @@ def _is_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
 
@@ -307,7 +377,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(
@@ -323,7 +393,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,53 @@
+from logging import Logger, getLogger
+from typing import Optional
+
+from opentelemetry.context import attach, detach, set_value, _SUPPRESS_INSTRUMENTATION_KEY
+from opentelemetry.sdk.metrics.export import ResourceMetrics, MetricsData, MetricExporter, PeriodicExportingMetricReader
+
+_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)

aws-opentelemetry-distro/tests/amazon/opentelemetry/distro/test_aws_opentelementry_configurator.py

@@ -15,7 +15,7 @@
     _customize_exporter,
     _customize_sampler,
     _customize_span_processors,
-    _is_application_signals_enabled,
+    _is_application_signals_enabled, APPLICATION_SIGNALS_RUNTIME_ENABLED_CONFIG,
 )
 from amazon.opentelemetry.distro.aws_opentelemetry_distro import AwsOpenTelemetryDistro
 from amazon.opentelemetry.distro.aws_span_metrics_processor import AwsSpanMetricsProcessor
@@ -53,6 +53,7 @@ def setUpClass(cls):
         os.environ[OTEL_TRACES_EXPORTER] = "none"
         os.environ[OTEL_METRICS_EXPORTER] = "none"
         os.environ[OTEL_LOGS_EXPORTER] = "none"
+        os.environ[APPLICATION_SIGNALS_RUNTIME_ENABLED_CONFIG] = "false"
         os.environ[OTEL_TRACES_SAMPLER] = "traceidratio"
         os.environ[OTEL_TRACES_SAMPLER_ARG] = "0.01"