feat(metrics) - add support for high resolution metrics

aws_lambda_powertools/metrics/__init__.py

@@ -1,7 +1,12 @@
 """CloudWatch Embedded Metric Format utility
 """
-from .base import MetricUnit
-from .exceptions import MetricUnitError, MetricValueError, SchemaValidationError
+from .base import MetricResolution, MetricUnit
+from .exceptions import (
+    MetricResolutionError,
+    MetricUnitError,
+    MetricValueError,
+    SchemaValidationError,
+)
 from .metric import single_metric
 from .metrics import EphemeralMetrics, Metrics
 
@@ -11,6 +16,8 @@
     "single_metric",
     "MetricUnit",
     "MetricUnitError",
+    "MetricResolution",
+    "MetricResolutionError",
     "SchemaValidationError",
     "MetricValueError",
 ]

aws_lambda_powertools/metrics/base.py

@@ -12,7 +12,13 @@
 
 from ..shared import constants
 from ..shared.functions import resolve_env_var_choice
-from .exceptions import MetricUnitError, MetricValueError, SchemaValidationError
+from .exceptions import (
+    MetricResolutionError,
+    MetricUnitError,
+    MetricValueError,
+    SchemaValidationError,
+)
+from .types import MetricNameUnitResolution
 
 logger = logging.getLogger(__name__)
 
@@ -22,6 +28,11 @@
 is_cold_start = True
 
 
+class MetricResolution(Enum):
+    Standard = 60
+    High = 1
+
+
 class MetricUnit(Enum):
     Seconds = "Seconds"
     Microseconds = "Microseconds"
@@ -72,7 +83,9 @@ class MetricManager:
     Raises
     ------
     MetricUnitError
-        When metric metric isn't supported by CloudWatch
+        When metric unit isn't supported by CloudWatch
+    MetricResolutionError
+        When metric resolution isn't supported by CloudWatch
     MetricValueError
         When metric value isn't a number
     SchemaValidationError
@@ -93,9 +106,16 @@ def __init__(
         self.service = resolve_env_var_choice(choice=service, env=os.getenv(constants.SERVICE_NAME_ENV))
         self.metadata_set = metadata_set if metadata_set is not None else {}
         self._metric_units = [unit.value for unit in MetricUnit]
-        self._metric_unit_options = list(MetricUnit.__members__)
+        self._metric_unit_valid_options = list(MetricUnit.__members__)
+        self._metric_resolutions = [resolution.value for resolution in MetricResolution]
 
-    def add_metric(self, name: str, unit: Union[MetricUnit, str], value: float) -> None:
+    def add_metric(
+        self,
+        name: str,
+        unit: Union[MetricUnit, str],
+        value: float,
+        resolution: Union[MetricResolution, int] = 60,
+    ) -> None:
         """Adds given metric
 
         Example
@@ -108,6 +128,10 @@ def add_metric(self, name: str, unit: Union[MetricUnit, str], value: float) -> N
 
             metric.add_metric(name="BookingConfirmation", unit="Count", value=1)
 
+        **Add given metric with MetricResolution non default value**
+
+            metric.add_metric(name="BookingConfirmation", unit="Count", value=1, resolution=MetricResolution.High)
+
         Parameters
         ----------
         name : str
@@ -116,18 +140,24 @@ def add_metric(self, name: str, unit: Union[MetricUnit, str], value: float) -> N
             `aws_lambda_powertools.helper.models.MetricUnit`
         value : float
             Metric value
+        resolution : Union[MetricResolution, int]
+            `aws_lambda_powertools.helper.models.MetricResolution`
 
         Raises
         ------
         MetricUnitError
             When metric unit is not supported by CloudWatch
+        MetricResolutionError
+            When metric resolution is not supported by CloudWatch
         """
         if not isinstance(value, numbers.Number):
             raise MetricValueError(f"{value} is not a valid number")
 
         unit = self._extract_metric_unit_value(unit=unit)
+        resolution = self._extract_metric_resolution_value(resolution=resolution)
         metric: Dict = self.metric_set.get(name, defaultdict(list))
         metric["Unit"] = unit
+        metric["StorageResolution"] = resolution
         metric["Value"].append(float(value))
         logger.debug(f"Adding metric: {name} with {metric}")
         self.metric_set[name] = metric
@@ -194,15 +224,28 @@ def serialize_metric_set(
 
         logger.debug({"details": "Serializing metrics", "metrics": metrics, "dimensions": dimensions})
 
-        metric_names_and_units: List[Dict[str, str]] = []  # [ { "Name": "metric_name", "Unit": "Count" } ]
+        # For standard resolution metrics, don't add StorageResolution field to avoid unnecessary ingestion of data into cloudwatch # noqa E501
+        # Example: [ { "Name": "metric_name", "Unit": "Count"} ] # noqa E800
+        #
+        # In case using high-resolution metrics, add StorageResolution field
+        # Example: [ { "Name": "metric_name", "Unit": "Count", "StorageResolution": 1 } ] # noqa E800
+        metric_definition: List[MetricNameUnitResolution] = []
         metric_names_and_values: Dict[str, float] = {}  # { "metric_name": 1.0 }
 
         for metric_name in metrics:
             metric: dict = metrics[metric_name]
             metric_value: int = metric.get("Value", 0)
             metric_unit: str = metric.get("Unit", "")
+            metric_resolution: int = metric.get("StorageResolution", 60)
+
+            metric_definition_data: MetricNameUnitResolution = {"Name": metric_name, "Unit": metric_unit}
+
+            # high-resolution metrics
+            if metric_resolution == 1:
+                metric_definition_data["StorageResolution"] = metric_resolution
+
+            metric_definition.append(metric_definition_data)
 
-            metric_names_and_units.append({"Name": metric_name, "Unit": metric_unit})
             metric_names_and_values.update({metric_name: metric_value})
 
         return {
@@ -212,7 +255,7 @@ def serialize_metric_set(
                     {
                         "Namespace": self.namespace,  # "test_namespace"
                         "Dimensions": [list(dimensions.keys())],  # [ "service" ]
-                        "Metrics": metric_names_and_units,
+                        "Metrics": metric_definition,
                     }
                 ],
             },
@@ -358,6 +401,34 @@ def decorate(event, context):
 
         return decorate
 
+    def _extract_metric_resolution_value(self, resolution: Union[int, MetricResolution]) -> int:
+        """Return metric value from metric unit whether that's str or MetricResolution enum
+
+        Parameters
+        ----------
+        unit : Union[int, MetricResolution]
+            Metric resolution
+
+        Returns
+        -------
+        int
+            Metric resolution value must be 1 or 60
+
+        Raises
+        ------
+        MetricResolutionError
+            When metric resolution is not supported by CloudWatch
+        """
+        if isinstance(resolution, MetricResolution):
+            return resolution.value
+
+        if isinstance(resolution, int) and resolution in self._metric_resolutions:
+            return resolution
+
+        raise MetricResolutionError(
+            f"Invalid metric resolution '{resolution}', expected either option: {self._metric_resolutions}"  # noqa: E501
+        )
+
     def _extract_metric_unit_value(self, unit: Union[str, MetricUnit]) -> str:
         """Return metric value from metric unit whether that's str or MetricUnit enum
 
@@ -378,12 +449,12 @@ def _extract_metric_unit_value(self, unit: Union[str, MetricUnit]) -> str:
         """
 
         if isinstance(unit, str):
-            if unit in self._metric_unit_options:
+            if unit in self._metric_unit_valid_options:
                 unit = MetricUnit[unit].value
 
             if unit not in self._metric_units:
                 raise MetricUnitError(
-                    f"Invalid metric unit '{unit}', expected either option: {self._metric_unit_options}"
+                    f"Invalid metric unit '{unit}', expected either option: {self._metric_unit_valid_options}"
                 )
 
         if isinstance(unit, MetricUnit):
@@ -429,10 +500,10 @@ class SingleMetric(MetricManager):
     **Creates cold start metric with function_version as dimension**
 
         import json
-        from aws_lambda_powertools.metrics import single_metric, MetricUnit
+        from aws_lambda_powertools.metrics import single_metric, MetricUnit, MetricResolution
         metric = single_metric(namespace="ServerlessAirline")
 
-        metric.add_metric(name="ColdStart", unit=MetricUnit.Count, value=1)
+        metric.add_metric(name="ColdStart", unit=MetricUnit.Count, value=1, resolution=MetricResolution.Standard)
         metric.add_dimension(name="function_version", value=47)
 
         print(json.dumps(metric.serialize_metric_set(), indent=4))
@@ -443,7 +514,13 @@ class SingleMetric(MetricManager):
         Inherits from `aws_lambda_powertools.metrics.base.MetricManager`
     """
 
-    def add_metric(self, name: str, unit: Union[MetricUnit, str], value: float) -> None:
+    def add_metric(
+        self,
+        name: str,
+        unit: Union[MetricUnit, str],
+        value: float,
+        resolution: Union[MetricResolution, int] = 60,
+    ) -> None:
         """Method to prevent more than one metric being created
 
         Parameters
@@ -454,18 +531,21 @@ def add_metric(self, name: str, unit: Union[MetricUnit, str], value: float) -> N
             Metric unit (e.g. "Seconds", MetricUnit.Seconds)
         value : float
             Metric value
+        resolution : MetricResolution
+            Metric resolution (e.g. 60, MetricResolution.Standard)
         """
         if len(self.metric_set) > 0:
             logger.debug(f"Metric {name} already set, skipping...")
             return
-        return super().add_metric(name, unit, value)
+        return super().add_metric(name, unit, value, resolution)
 
 
 @contextmanager
 def single_metric(
     name: str,
     unit: MetricUnit,
     value: float,
+    resolution: Union[MetricResolution, int] = 60,
     namespace: Optional[str] = None,
     default_dimensions: Optional[Dict[str, str]] = None,
 ) -> Generator[SingleMetric, None, None]:
@@ -477,8 +557,9 @@ def single_metric(
 
         from aws_lambda_powertools import single_metric
         from aws_lambda_powertools.metrics import MetricUnit
+        from aws_lambda_powertools.metrics import MetricResolution
 
-        with single_metric(name="ColdStart", unit=MetricUnit.Count, value=1, namespace="ServerlessAirline") as metric:
+        with single_metric(name="ColdStart", unit=MetricUnit.Count, value=1, resolution=MetricResolution.Standard, namespace="ServerlessAirline") as metric: # noqa E501
             metric.add_dimension(name="function_version", value="47")
 
     **Same as above but set namespace using environment variable**
@@ -487,8 +568,9 @@ def single_metric(
 
         from aws_lambda_powertools import single_metric
         from aws_lambda_powertools.metrics import MetricUnit
+        from aws_lambda_powertools.metrics import MetricResolution
 
-        with single_metric(name="ColdStart", unit=MetricUnit.Count, value=1) as metric:
+        with single_metric(name="ColdStart", unit=MetricUnit.Count, value=1, resolution=MetricResolution.Standard) as metric: # noqa E501
             metric.add_dimension(name="function_version", value="47")
 
     Parameters
@@ -497,6 +579,8 @@ def single_metric(
         Metric name
     unit : MetricUnit
         `aws_lambda_powertools.helper.models.MetricUnit`
+    resolution : MetricResolution
+        `aws_lambda_powertools.helper.models.MetricResolution`
     value : float
         Metric value
     namespace: str
@@ -511,6 +595,8 @@ def single_metric(
     ------
     MetricUnitError
         When metric metric isn't supported by CloudWatch
+    MetricResolutionError
+        When metric resolution isn't supported by CloudWatch
     MetricValueError
         When metric value isn't a number
     SchemaValidationError
@@ -519,7 +605,7 @@ def single_metric(
     metric_set: Optional[Dict] = None
     try:
         metric: SingleMetric = SingleMetric(namespace=namespace)
-        metric.add_metric(name=name, unit=unit, value=value)
+        metric.add_metric(name=name, unit=unit, value=value, resolution=resolution)
 
         if default_dimensions:
             for dim_name, dim_value in default_dimensions.items():

aws_lambda_powertools/metrics/exceptions.py

@@ -4,6 +4,12 @@ class MetricUnitError(Exception):
     pass
 
 
+class MetricResolutionError(Exception):
+    """When metric resolution is not supported by CloudWatch"""
+
+    pass
+
+
 class SchemaValidationError(Exception):
     """When serialization fail schema validation"""
 

aws_lambda_powertools/metrics/metrics.py

@@ -50,7 +50,9 @@ def lambda_handler():
     Raises
     ------
     MetricUnitError
-        When metric metric isn't supported by CloudWatch
+        When metric unit isn't supported by CloudWatch
+    MetricResolutionError
+        When metric resolution isn't supported by CloudWatch
     MetricValueError
         When metric value isn't a number
     SchemaValidationError

aws_lambda_powertools/metrics/types.py

@@ -0,0 +1,7 @@
+from typing_extensions import NotRequired, TypedDict
+
+
+class MetricNameUnitResolution(TypedDict):
+    Name: str
+    Unit: str
+    StorageResolution: NotRequired[int]

docs/core/metrics.md

@@ -20,6 +20,9 @@ If you're new to Amazon CloudWatch, there are two terminologies you must be awar
 
 * **Namespace**. It's the highest level container that will group multiple metrics from multiple services for a given application, for example `ServerlessEcommerce`.
 * **Dimensions**. Metrics metadata in key-value format. They help you slice and dice metrics visualization, for example `ColdStart` metric by Payment `service`.
+* **Metric**. It's the name of the metric, for example: `SuccessfulBooking` or `UpdatedBooking`.
+* **Unit**. It's a value representing the unit of measure for the corresponding metric, for example: `Count` or `Seconds`.
+* **Resolution**. It's a value representing the storage resolution for the corresponding metric. Metrics can be either Standard or High resolution. Read more [here](https://docs.aws.amazon.com/AmazonCloudWatch/latest/monitoring/publishingMetrics.html#high-resolution-metrics).
 
 <figure>
   <img src="../../media/metrics_terminology.png" />
@@ -78,6 +81,22 @@ You can create metrics using `add_metric`, and you can create dimensions for all
 ???+ warning "Warning: Do not create metrics or dimensions outside the handler"
     Metrics or dimensions added in the global scope will only be added during cold start. Disregard if you that's the intended behavior.
 
+### Adding high-resolution metrics
+
+You can create [high-resolution metrics](https://aws.amazon.com/pt/about-aws/whats-new/2023/02/amazon-cloudwatch-high-resolution-metric-extraction-structured-logs/) passing `resolution` parameter to `add_metric`.
+
+???+ tip "High-resolution metrics - when is it useful?"
+    High-resolution metrics are data with a granularity of one second and are very useful in several situations such as telemetry, time series, real-time incident management, and others.
+
+=== "add_high_resolution_metrics.py"
+
+    ```python hl_lines="10"
+    --8<-- "examples/metrics/src/add_high_resolution_metric.py"
+    ```
+
+???+ tip "Tip: Autocomplete Metric Resolutions"
+    `MetricResolution` enum facilitates finding a supported metric resolution by CloudWatch. Alternatively, you can pass the values 1 or 60 (must be one of them) as an integer _e.g. `resolution=1`_.
+
 ### Adding multi-value metrics
 
 You can call `add_metric()` with the same metric name multiple times. The values will be grouped together in a list.

examples/metrics/src/add_high_resolution_metric.py

@@ -0,0 +1,10 @@
+from aws_lambda_powertools import Metrics
+from aws_lambda_powertools.metrics import MetricResolution, MetricUnit
+from aws_lambda_powertools.utilities.typing import LambdaContext
+
+metrics = Metrics()
+
+
+@metrics.log_metrics  # ensures metrics are flushed upon request completion/failure
+def lambda_handler(event: dict, context: LambdaContext):
+    metrics.add_metric(name="SuccessfulBooking", unit=MetricUnit.Count, value=1, resolution=MetricResolution.High)