Add contract test for runtime metrics

Author: bjrara

contract-tests/images/mock-collector/mock_collector_client.py

@@ -87,7 +87,7 @@ def wait_condition(exported: List[ExportTraceServiceRequest], current: List[Expo
                         spans.append(ResourceScopeSpan(resource_span, scope_span, span))
         return spans
 
-    def get_metrics(self, present_metrics: Set[str]) -> List[ResourceScopeMetric]:
+    def get_metrics(self, present_metrics: Set[str], exact_match=True) -> List[ResourceScopeMetric]:
         """Get all metrics that are currently stored in the mock collector.
 
         Returns:
@@ -111,7 +111,9 @@ def wait_condition(
                     for scope_metric in resource_metric.scope_metrics:
                         for metric in scope_metric.metrics:
                             received_metrics.add(metric.name.lower())
-            return 0 < len(exported) == len(current) and present_metrics_lower.issubset(received_metrics)
+            if exact_match:
+                return 0 < len(exported) == len(current) and present_metrics_lower.issubset(received_metrics)
+            return present_metrics_lower.issubset(received_metrics)
 
         exported_metrics: List[ExportMetricsServiceRequest] = _wait_for_content(get_export, wait_condition)
         metrics: List[ResourceScopeMetric] = []

contract-tests/tests/test/amazon/base/contract_test_base.py

@@ -90,7 +90,7 @@ def setUp(self) -> None:
             .with_exposed_ports(self.get_application_port())
             .with_env("OTEL_METRIC_EXPORT_INTERVAL", "50")
             .with_env("OTEL_AWS_APPLICATION_SIGNALS_ENABLED", "true")
-            .with_env("OTEL_AWS_APPLICATION_SIGNALS_RUNTIME_ENABLED", "false")
+            .with_env("OTEL_AWS_APPLICATION_SIGNALS_RUNTIME_ENABLED", self.is_runtime_enabled())
             .with_env("OTEL_METRICS_EXPORTER", "none")
             .with_env("OTEL_EXPORTER_OTLP_PROTOCOL", "grpc")
             .with_env("OTEL_BSP_SCHEDULE_DELAY", "1")
@@ -217,6 +217,9 @@ def get_application_otel_service_name(self) -> str:
     def get_application_otel_resource_attributes(self) -> str:
         return "service.name=" + self.get_application_otel_service_name()
 
+    def is_runtime_enabled(self) -> str:
+        return "false"
+
     def _assert_aws_span_attributes(self, resource_scope_spans: List[ResourceScopeSpan], path: str, **kwargs):
         self.fail("Tests must implement this function")
 

contract-tests/tests/test/amazon/runtime/runtime_metrics_test.py

@@ -0,0 +1,104 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+from typing import Dict, List
+
+from mock_collector_client import ResourceScopeMetric
+from requests import Response
+from typing_extensions import override
+
+import amazon.utils.application_signals_constants as constants
+from amazon.base.contract_test_base import ContractTestBase
+from opentelemetry.proto.common.v1.common_pb2 import AnyValue
+from opentelemetry.proto.metrics.v1.metrics_pb2 import Metric, NumberDataPoint
+
+
+class RuntimeMetricsTest(ContractTestBase):
+    @override
+    def is_runtime_enabled(self) -> str:
+        return "true"
+
+    @override
+    @staticmethod
+    def get_application_image_name() -> str:
+        return "aws-application-signals-tests-django-app"
+
+    @override
+    def get_application_wait_pattern(self) -> str:
+        return "Quit the server with CONTROL-C."
+
+    @override
+    def get_application_extra_environment_variables(self):
+        return {"DJANGO_SETTINGS_MODULE": "django_server.settings"}
+
+    def test_runtime_succeeds(self) -> None:
+        self.mock_collector_client.clear_signals()
+        response: Response = self.send_request("GET", "success")
+        self.assertEqual(200, response.status_code)
+
+        metrics: List[ResourceScopeMetric] = self.mock_collector_client.get_metrics(
+            {
+                constants.LATENCY_METRIC,
+                constants.ERROR_METRIC,
+                constants.FAULT_METRIC,
+                constants.PYTHON_PROCESS_CPU_TIME,
+                constants.PYTHON_PROCESS_CPU_UTILIZATION,
+                constants.PYTHON_PROCESS_GC_COUNT,
+                constants.PYTHON_PROCESS_MEMORY_USED,
+                constants.PYTHON_PROCESS_THREAD_COUNT,
+            },
+            False,
+        )
+        self._assert_resource_attributes(metrics)
+        self._assert_counter_attribute_exists(metrics, constants.PYTHON_PROCESS_CPU_TIME, "")
+        self._assert_gauge_attribute_exists(metrics, constants.PYTHON_PROCESS_CPU_UTILIZATION, "")
+        self._assert_gauge_attribute_exists(metrics, constants.PYTHON_PROCESS_GC_COUNT, "count")
+        self._assert_gauge_attribute_exists(metrics, constants.PYTHON_PROCESS_MEMORY_USED, "type")
+        self._assert_gauge_attribute_exists(metrics, constants.PYTHON_PROCESS_THREAD_COUNT, "")
+
+    def _assert_resource_attributes(
+        self,
+        resource_scope_metrics: List[ResourceScopeMetric],
+    ) -> None:
+        for metric in resource_scope_metrics:
+            attribute_dict: Dict[str, AnyValue] = self._get_attributes_dict(metric.resource_metrics.resource.attributes)
+            self._assert_str_attribute(
+                attribute_dict, constants.AWS_LOCAL_SERVICE, self.get_application_otel_service_name()
+            )
+
+    def _assert_gauge_attribute_exists(
+        self,
+        resource_scope_metrics: List[ResourceScopeMetric],
+        metric_name: str,
+        attribute_key: str,
+    ) -> None:
+        target_metrics: List[Metric] = []
+        for resource_scope_metric in resource_scope_metrics:
+            if resource_scope_metric.metric.name.lower() == metric_name.lower():
+                target_metrics.append(resource_scope_metric.metric)
+        self.assertTrue(len(target_metrics) > 0)
+
+        for target_metric in target_metrics:
+            dp_list: List[NumberDataPoint] = target_metric.gauge.data_points
+            self.assertTrue(len(dp_list) > 0)
+            if attribute_key != "":
+                attribute_dict: Dict[str, AnyValue] = self._get_attributes_dict(dp_list[0].attributes)
+                self.assertIsNotNone(attribute_dict.get(attribute_key))
+
+    def _assert_counter_attribute_exists(
+        self,
+        resource_scope_metrics: List[ResourceScopeMetric],
+        metric_name: str,
+        attribute_key: str,
+    ) -> None:
+        target_metrics: List[Metric] = []
+        for resource_scope_metric in resource_scope_metrics:
+            if resource_scope_metric.metric.name.lower() == metric_name.lower():
+                target_metrics.append(resource_scope_metric.metric)
+        self.assertTrue(len(target_metrics) > 0)
+
+        for target_metric in target_metrics:
+            dp_list: List[NumberDataPoint] = target_metric.sum.data_points
+            self.assertTrue(len(dp_list) > 0)
+            if attribute_key != "":
+                attribute_dict: Dict[str, AnyValue] = self._get_attributes_dict(dp_list[0].attributes)
+                self.assertIsNotNone(attribute_dict.get(attribute_key))

contract-tests/tests/test/amazon/utils/application_signals_constants.py

@@ -9,6 +9,12 @@
 ERROR_METRIC: str = "error"
 FAULT_METRIC: str = "fault"
 
+PYTHON_PROCESS_GC_COUNT = "process.runtime.cpython.gc_count"
+PYTHON_PROCESS_MEMORY_USED = "process.runtime.cpython.memory"
+PYTHON_PROCESS_THREAD_COUNT = "process.runtime.cpython.thread_count"
+PYTHON_PROCESS_CPU_TIME = "process.runtime.cpython.cpu_time"
+PYTHON_PROCESS_CPU_UTILIZATION = "process.runtime.cpython.cpu.utilization"
+
 # Attribute names
 AWS_CLOUDFORMATION_PRIMARY_IDENTIFIER: str = "aws.remote.resource.cfn.primary.identifier"
 AWS_LOCAL_SERVICE: str = "aws.local.service"