Initial commit

Author: leandrodamascena

aws_lambda_powertools/metrics/base.py

@@ -76,6 +76,8 @@ def __init__(
         self.namespace = resolve_env_var_choice(choice=namespace, env=os.getenv(constants.METRICS_NAMESPACE_ENV))
         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.timestamp: int | None = None
+
         self._metric_units = [unit.value for unit in MetricUnit]
         self._metric_unit_valid_options = list(MetricUnit.__members__)
         self._metric_resolutions = [resolution.value for resolution in MetricResolution]
@@ -224,7 +226,7 @@ def serialize_metric_set(
 
         return {
             "_aws": {
-                "Timestamp": int(datetime.datetime.now().timestamp() * 1000),  # epoch
+                "Timestamp": self.timestamp or int(datetime.datetime.now().timestamp() * 1000),  # epoch
                 "CloudWatchMetrics": [
                     {
                         "Namespace": self.namespace,  # "test_namespace"
@@ -296,6 +298,16 @@ def add_metadata(self, key: str, value: Any) -> None:
         else:
             self.metadata_set[str(key)] = value
 
+    def set_timestamp(self, timestamp: int):
+        if not isinstance(timestamp, int):
+            warnings.warn(
+                "The timestamp key must be an integer value representing an epoch time. ",
+                "The provided value is not valid. Using the current timestamp instead.",
+                stacklevel=2,
+            )
+        else:
+            self.timestamp = timestamp
+
     def clear_metrics(self) -> None:
         logger.debug("Clearing out existing metric set from memory")
         self.metric_set.clear()

aws_lambda_powertools/metrics/metrics.py

@@ -125,6 +125,9 @@ def serialize_metric_set(
     def add_metadata(self, key: str, value: Any) -> None:
         self.provider.add_metadata(key=key, value=value)
 
+    def set_timestamp(self, timestamp: int):
+        self.provider.set_timestamp(timestamp=timestamp)
+
     def flush_metrics(self, raise_on_empty_metrics: bool = False) -> None:
         self.provider.flush_metrics(raise_on_empty_metrics=raise_on_empty_metrics)
 

aws_lambda_powertools/metrics/provider/cloudwatch_emf/cloudwatch.py

@@ -73,6 +73,7 @@ def __init__(
         self.namespace = resolve_env_var_choice(choice=namespace, env=os.getenv(constants.METRICS_NAMESPACE_ENV))
         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.timestamp: int | None = None
 
         self._metric_units = [unit.value for unit in MetricUnit]
         self._metric_unit_valid_options = list(MetricUnit.__members__)
@@ -231,7 +232,7 @@ def serialize_metric_set(
 
         return {
             "_aws": {
-                "Timestamp": int(datetime.datetime.now().timestamp() * 1000),  # epoch
+                "Timestamp": self.timestamp or int(datetime.datetime.now().timestamp() * 1000),  # epoch
                 "CloudWatchMetrics": [
                     {
                         "Namespace": self.namespace,  # "test_namespace"
@@ -304,6 +305,16 @@ def add_metadata(self, key: str, value: Any) -> None:
         else:
             self.metadata_set[str(key)] = value
 
+    def set_timestamp(self, timestamp: int):
+        if not isinstance(timestamp, int):
+            warnings.warn(
+                "The timestamp key must be an integer value representing an epoch time. "
+                "The provided value is not valid. Using the current timestamp instead.",
+                stacklevel=2,
+            )
+        else:
+            self.timestamp = timestamp
+
     def clear_metrics(self) -> None:
         logger.debug("Clearing out existing metric set from memory")
         self.metric_set.clear()

tests/functional/metrics/test_metrics_cloudwatch_emf.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import datetime
 import json
 import warnings
 from collections import namedtuple
@@ -1213,3 +1214,47 @@ def lambda_handler(evt, ctx):
 
     output = capture_metrics_output_multiple_emf_objects(capsys)
     assert len(output) == 2
+
+
+def test_metric_with_custom_timestamp(namespace, metric, capsys):
+    # GIVEN Metrics instance is initialized
+    my_metrics = Metrics(namespace=namespace)
+
+    # Calculate the metric timestamp as 2 days before the current time
+    metric_timestamp = int((datetime.datetime.now() - datetime.timedelta(days=2)).timestamp() * 1000)
+
+    # WHEN we set custom timestamp before to flush the metric
+    @my_metrics.log_metrics
+    def lambda_handler(evt, ctx):
+        my_metrics.add_metric(**metric)
+        my_metrics.set_timestamp(metric_timestamp)
+
+    lambda_handler({}, {})
+    invocation = capture_metrics_output(capsys)
+
+    # THEN Timestamp must be the custom value
+    assert invocation["_aws"]["Timestamp"] == metric_timestamp
+
+
+def test_metric_with_wrong_custom_timestamp(namespace, metric):
+    # GIVEN Metrics instance is initialized
+    my_metrics = Metrics(namespace=namespace)
+
+    # Setting timestamp as datetime
+    metric_timestamp = datetime.datetime.now()
+
+    # WHEN we set a wrong timestamp before to flush the metric
+    @my_metrics.log_metrics
+    def lambda_handler(evt, ctx):
+        my_metrics.add_metric(**metric)
+        my_metrics.set_timestamp(metric_timestamp)
+
+    # THEN should raise a warning
+    with warnings.catch_warnings(record=True) as w:
+        warnings.simplefilter("default")
+        lambda_handler({}, {})
+        assert len(w) == 1
+        assert str(w[-1].message) == (
+            "The timestamp key must be an integer value representing an epoch time. "
+            "The provided value is not valid. Using the current timestamp instead."
+        )