Initial implementation for supporting High Resolution Metrics

Author: Meshwa Savalia

aws_embedded_metrics/logger/metric.py

@@ -13,9 +13,10 @@
 
 
 class Metric(object):
-    def __init__(self, value: float, unit: str = None):
+    def __init__(self, value: float, unit: str = None, storageResolution: int = 60):
         self.values = [value]
         self.unit = unit or "None"
+        self.storageResolution = storageResolution
 
     def add_value(self, value: float) -> None:
         self.values.append(value)

aws_embedded_metrics/logger/metrics_context.py

@@ -39,8 +39,9 @@ def __init__(
         self.metrics: Dict[str, Metric] = {}
         self.should_use_default_dimensions = True
         self.meta: Dict[str, Any] = {"Timestamp": utils.now()}
+        self.metricNameAndResolutionMap: Dict[str, int] = {}
 
-    def put_metric(self, key: str, value: float, unit: str = None) -> None:
+    def put_metric(self, key: str, value: float, unit: str = None, storageResolution: int = 60) -> None:
         """
         Adds a metric measurement to the context.
         Multiple calls using the same key will be stored as an
@@ -49,13 +50,13 @@ def put_metric(self, key: str, value: float, unit: str = None) -> None:
         context.put_metric("Latency", 100, "Milliseconds")
         ```
         """
-        validate_metric(key, value, unit)
+        validate_metric(key, value, unit, storageResolution, self.metricNameAndResolutionMap)
         metric = self.metrics.get(key)
         if metric:
             # TODO: we should log a warning if the unit has been changed
             metric.add_value(value)
         else:
-            self.metrics[key] = Metric(value, unit)
+            self.metrics[key] = Metric(value, unit, storageResolution)
 
     def put_dimensions(self, dimension_set: Dict[str, str]) -> None:
         """

aws_embedded_metrics/logger/metrics_logger.py

@@ -78,8 +78,8 @@ def set_namespace(self, namespace: str) -> "MetricsLogger":
         self.context.namespace = namespace
         return self
 
-    def put_metric(self, key: str, value: float, unit: str = "None") -> "MetricsLogger":
-        self.context.put_metric(key, value, unit)
+    def put_metric(self, key: str, value: float, unit: str = "None", storageResolution: int = 60) -> "MetricsLogger":
+        self.context.put_metric(key, value, unit, storageResolution)
         return self
 
     def add_stack_trace(self, key: str, details: Any = None, exc_info: Tuple = None) -> "MetricsLogger":

aws_embedded_metrics/serializers/log_serializer.py

@@ -87,8 +87,11 @@ def create_body() -> Dict[str, Any]:
                     if len(metric.values) > end_index:
                         remaining_data = True
 
+                metricBody = {"Name": metric_name, "Unit": metric.unit}
+                if metric.storageResolution == 1:
+                    metricBody["StorageResolution"] = metric.storageResolution
                 if not config.disable_metric_extraction:
-                    current_body["_aws"]["CloudWatchMetrics"][0]["Metrics"].append({"Name": metric_name, "Unit": metric.unit})
+                    current_body["_aws"]["CloudWatchMetrics"][0]["Metrics"].append(metricBody)
                 num_metrics_in_current_body += 1
 
                 if (num_metrics_in_current_body == MAX_METRICS_PER_EVENT):

aws_embedded_metrics/storageResolution.py

@@ -0,0 +1,16 @@
+from enum import Enum, EnumMeta
+
+
+class StorageResolutionMeta(EnumMeta):
+    def __contains__(self, item: object) -> bool:
+        try:
+            self(item)
+        except (ValueError, TypeError):
+            return False
+        else:
+            return True
+
+
+class StorageResolution(Enum, metaclass=StorageResolutionMeta):
+    STANDARD = 60
+    HIGH = 1

aws_embedded_metrics/validator.py

@@ -15,6 +15,7 @@
 import re
 from typing import Dict, Optional
 from aws_embedded_metrics.unit import Unit
+from aws_embedded_metrics.storageResolution import StorageResolution
 from aws_embedded_metrics.exceptions import DimensionSetExceededError, InvalidDimensionError, InvalidMetricError, InvalidNamespaceError
 import aws_embedded_metrics.constants as constants
 
@@ -57,14 +58,15 @@ def validate_dimension_set(dimension_set: Dict[str, str]) -> None:
             raise InvalidDimensionError("Dimension name cannot start with ':'")
 
 
-def validate_metric(name: str, value: float, unit: Optional[str]) -> None:
+def validate_metric(name: str, value: float, unit: Optional[str], storageResolution: Optional[int], metricNameAndResolutionMap: Optional[dict]) -> None:  # noqa: E501
     """
     Validates a metric
 
         Parameters:
             name (str): The name of the metric
             value (float): The value of the metric
             unit (Optional[str]): The unit of the metric
+            storageResolution (Optional[int]): The storage resolution of metric
 
         Raises:
             InvalidMetricError: If the metric is invalid
@@ -81,6 +83,16 @@ def validate_metric(name: str, value: float, unit: Optional[str]) -> None:
     if unit is not None and unit not in Unit:
         raise InvalidMetricError(f"Metric unit is not valid: {unit}")
 
+    if storageResolution is None or storageResolution not in StorageResolution:
+        raise InvalidMetricError(f"Metric storage Resolution is not valid: {storageResolution}")
+
+    if metricNameAndResolutionMap and name in metricNameAndResolutionMap:
+        if metricNameAndResolutionMap.get(name) is not storageResolution:
+            raise InvalidMetricError(
+                f"Resolution for metrics ${name} is already set. A single log event cannot have a metric with two different resolutions.")
+    else:
+        metricNameAndResolutionMap[name] = storageResolution
+
 
 def validate_namespace(namespace: str) -> None:
     """

tests/canary/agent/canary.py

@@ -26,7 +26,7 @@ async def app(init, last_run_duration, metrics):
     metrics.set_dimensions({"Runtime": 'Python', "Platform": 'ECS', "Agent": 'CloudWatchAgent', "Version": version})
     metrics.put_metric('Invoke', 1, "Count")
     metrics.put_metric('Duration', last_run_duration, 'Seconds')
-    metrics.put_metric('Memory.RSS', process.memory_info().rss, 'Bytes')
+    metrics.put_metric('Memory.RSS', process.memory_info().rss, 'Bytes', 1)
 
 
 async def main():

tests/logger/test_metrics_context.py

@@ -3,6 +3,7 @@
 import random
 from aws_embedded_metrics import constants
 from aws_embedded_metrics.unit import Unit
+from aws_embedded_metrics.storageResolution import StorageResolution
 from aws_embedded_metrics import config
 from aws_embedded_metrics.logger.metrics_context import MetricsContext
 from aws_embedded_metrics.constants import DEFAULT_NAMESPACE
@@ -263,14 +264,16 @@ def test_put_metric_adds_metrics():
     metric_key = fake.word()
     metric_value = fake.random.random()
     metric_unit = random.choice(list(Unit)).value
+    metric_storageResolution = random.choice(list(StorageResolution)).value
 
     # act
-    context.put_metric(metric_key, metric_value, metric_unit)
+    context.put_metric(metric_key, metric_value, metric_unit, metric_storageResolution)
 
     # assert
     metric = context.metrics[metric_key]
     assert metric.unit == metric_unit
     assert metric.values == [metric_value]
+    assert metric.storageResolution == metric_storageResolution
 
 
 def test_put_metric_uses_none_unit_if_not_provided():
@@ -287,26 +290,42 @@ def test_put_metric_uses_none_unit_if_not_provided():
     assert metric.unit == "None"
 
 
+def test_put_metric_uses_60_storage_resolution_if_not_provided():
+    # arrange
+    context = MetricsContext()
+    metric_key = fake.word()
+    metric_value = fake.random.random()
+
+    # act
+    context.put_metric(metric_key, metric_value)
+
+    # assert
+    metric = context.metrics[metric_key]
+    assert metric.storageResolution == 60
+
+
 @pytest.mark.parametrize(
-    "name, value, unit",
+    "name, value, unit, storageResolution",
     [
-        ("", 1, "None"),
-        (" ", 1, "Seconds"),
-        ("a" * (constants.MAX_METRIC_NAME_LENGTH + 1), 1, "None"),
-        ("metric", float("inf"), "Count"),
-        ("metric", float("-inf"), "Count"),
-        ("metric", float("nan"), "Count"),
-        ("metric", math.inf, "Seconds"),
-        ("metric", -math.inf, "Seconds"),
-        ("metric", math.nan, "Seconds"),
-        ("metric", 1, "Kilometers/Fahrenheit")
+        ("", 1, "None", 60),
+        (" ", 1, "Seconds", 60),
+        ("a" * (constants.MAX_METRIC_NAME_LENGTH + 1), 1, "None", 60),
+        ("metric", float("inf"), "Count", 60),
+        ("metric", float("-inf"), "Count", 60),
+        ("metric", float("nan"), "Count", 60),
+        ("metric", math.inf, "Seconds", 60),
+        ("metric", -math.inf, "Seconds", 60),
+        ("metric", math.nan, "Seconds", 60),
+        ("metric", 1, "Kilometers/Fahrenheit", 60),
+        ("metric", 1, "Seconds", 2),
+        ("metric", 1, "Seconds", None)
     ]
 )
-def test_put_invalid_metric_raises_exception(name, value, unit):
+def test_put_invalid_metric_raises_exception(name, value, unit, storageResolution):
     context = MetricsContext()
 
     with pytest.raises(InvalidMetricError):
-        context.put_metric(name, value, unit)
+        context.put_metric(name, value, unit, storageResolution)
 
 
 def test_create_copy_with_context_creates_new_instance():

tests/logger/test_metrics_logger.py

@@ -2,7 +2,7 @@
 from aws_embedded_metrics.logger import metrics_logger
 from aws_embedded_metrics.sinks import Sink
 from aws_embedded_metrics.environment import Environment
-from aws_embedded_metrics.exceptions import InvalidNamespaceError
+from aws_embedded_metrics.exceptions import InvalidNamespaceError, InvalidMetricError
 import aws_embedded_metrics.constants as constants
 import pytest
 from faker import Faker
@@ -53,6 +53,51 @@ async def test_can_put_metric(mocker):
     assert context.metrics[expected_key].unit == "None"
 
 
+@pytest.mark.asyncio
+async def test_can_put_metric_with_different_storage_resolution_different_flush(mocker):
+    # arrange
+    expected_key = fake.word()
+    expected_value = fake.random.randrange(100)
+    metric_storageResolution = 1
+
+    logger, sink, env = get_logger_and_sink(mocker)
+
+    # act
+    logger.put_metric(expected_key, expected_value, None, metric_storageResolution)
+    await logger.flush()
+
+    # assert
+    context = sink.accept.call_args[0][0]
+    assert context.metrics[expected_key].values == [expected_value]
+    assert context.metrics[expected_key].unit == "None"
+    assert context.metrics[expected_key].storageResolution == metric_storageResolution
+
+    expected_key = fake.word()
+    expected_value = fake.random.randrange(100)
+    logger.put_metric(expected_key, expected_value, None)
+    await logger.flush()
+    context = sink.accept.call_args[0][0]
+    assert context.metrics[expected_key].values == [expected_value]
+    assert context.metrics[expected_key].unit == "None"
+    assert context.metrics[expected_key].storageResolution == 60
+
+
+@pytest.mark.asyncio
+async def test_cannot_put_metric_with_different_storage_resolution_same_flush(mocker):
+    # arrange
+    expected_key = fake.word()
+    expected_value = fake.random.randrange(100)
+    metric_storageResolution = 1
+
+    logger, sink, env = get_logger_and_sink(mocker)
+
+    # act
+    with pytest.raises(InvalidMetricError):
+        logger.put_metric(expected_key, expected_value, None, metric_storageResolution)
+        logger.put_metric(expected_key, expected_value, None, 60)
+        await logger.flush()
+
+
 @pytest.mark.asyncio
 async def test_can_add_stack_trace(mocker):
     # arrange

tests/serializer/test_log_serializer.py

@@ -92,6 +92,29 @@ def test_serialize_metrics():
     assert_json_equality(result_json, expected)
 
 
+def test_serialize_metrics_with_storageResolution():
+    # arrange
+    expected_key = fake.word()
+    expected_value = fake.random.randrange(0, 100)
+
+    expected_metric_definition = {"Name": expected_key, "Unit": "None", "StorageResolution": 1}
+
+    expected = {**get_empty_payload()}
+    expected[expected_key] = expected_value
+    expected["_aws"]["CloudWatchMetrics"][0]["Metrics"].append(
+        expected_metric_definition
+    )
+
+    context = get_context()
+    context.put_metric(expected_key, expected_value, "None", 1)
+
+    # act
+    result_json = serializer.serialize(context)[0]
+
+    # assert
+    assert_json_equality(result_json, expected)
+
+
 def test_serialize_more_than_100_metrics():
     # arrange
     expected_value = fake.random.randrange(0, 100)