Added support to set custom timestamp

README.md

@@ -178,6 +178,20 @@ Examples:
 set_namespace("MyApplication")
 ```
 
+- **set_timestamp**(timestamp: datetime) -> MetricsLogger
+
+Sets the timestamp of the metrics. If not set, current time of the client will be used.
+
+Timestamp must meet CloudWatch requirements, otherwise a InvalidTimestampError will be thrown. See [Timestamps](https://docs.aws.amazon.com/AmazonCloudWatch/latest/monitoring/cloudwatch_concepts.html#about_timestamp) for valid values.
+
+Examples:
+
+```py
+    set_timestamp(datetime.datetime.now())
+```
+
+
+
 - **flush**()
 
 Flushes the current MetricsContext to the configured sink and resets all properties and metric values. The namespace and default dimensions will be preserved across flushes.

aws_embedded_metrics/constants.py

@@ -20,3 +20,6 @@
 MAX_METRIC_NAME_LENGTH = 1024
 MAX_NAMESPACE_LENGTH = 256
 VALID_NAMESPACE_REGEX = '^[a-zA-Z0-9._#:/-]+$'
+TIMESTAMP = "Timestamp"
+MAX_TIMESTAMP_PAST_AGE_SECONDS = 14 * 24 * 60 * 60 * 1000  # 14 days
+MAX_TIMESTAMP_FUTURE_AGE_SECONDS = 2 * 60 * 60 * 1000  # 2 hours

aws_embedded_metrics/exceptions.py

@@ -33,3 +33,9 @@ class InvalidNamespaceError(Exception):
     def __init__(self, message: str) -> None:
         # Call the base class constructor with the parameters it needs
         super().__init__(message)
+
+
+class InvalidTimestampError(Exception):
+    def __init__(self, message: str) -> None:
+        # Call the base class constructor with the parameters it needs
+        super().__init__(message)

aws_embedded_metrics/logger/metrics_context.py

@@ -12,7 +12,9 @@
 # limitations under the License.
 
 
+import datetime
 from aws_embedded_metrics import constants, utils
+from aws_embedded_metrics import validator
 from aws_embedded_metrics.config import get_config
 from aws_embedded_metrics.logger.metric import Metric
 from aws_embedded_metrics.validator import validate_dimension_set, validate_metric
@@ -39,7 +41,7 @@ def __init__(
         self.default_dimensions: Dict[str, str] = default_dimensions or {}
         self.metrics: Dict[str, Metric] = {}
         self.should_use_default_dimensions = True
-        self.meta: Dict[str, Any] = {"Timestamp": utils.now()}
+        self.meta: Dict[str, Any] = {constants.TIMESTAMP: utils.now()}
         self.metric_name_and_resolution_map: Dict[str, StorageResolution] = {}
 
     def put_metric(self, key: str, value: float, unit: str = None, storage_resolution: StorageResolution = StorageResolution.STANDARD) -> None:
@@ -176,3 +178,20 @@ def create_copy_with_context(self, preserve_dimensions: bool = False) -> "Metric
     @staticmethod
     def empty() -> "MetricsContext":
         return MetricsContext()
+
+    def set_timestamp(self, timestamp: datetime) -> None:
+        """
+    Update timestamp field in the metadata
+
+    * @see <a
+    *     href="https://docs.aws.amazon.com/AmazonCloudWatch/latest/monitoring/cloudwatch_concepts.html#about_timestamp">CloudWatch
+    *     Timestamp</a>
+
+        Parameters:
+            timestamp (datetime): timestamp value to be set
+
+        Raises:
+            InvalidTimestampError: if timestamp is invalid
+    """
+        validator.validate_timestamp(timestamp)
+        self.meta[constants.TIMESTAMP] = utils.convert_to_milliseconds(timestamp)

aws_embedded_metrics/utils.py

@@ -12,4 +12,9 @@
 # limitations under the License.
 
 import time
+from datetime import datetime
 def now() -> int: return int(round(time.time() * 1000))
+
+
+def convert_to_milliseconds(datetime: datetime) -> int:
+    return int(round(datetime.timestamp() * 1000))

aws_embedded_metrics/validator.py

@@ -16,8 +16,10 @@
 from typing import Dict, Optional
 from aws_embedded_metrics.unit import Unit
 from aws_embedded_metrics.storage_resolution import StorageResolution
-from aws_embedded_metrics.exceptions import DimensionSetExceededError, InvalidDimensionError, InvalidMetricError, InvalidNamespaceError
+from aws_embedded_metrics.exceptions import DimensionSetExceededError, InvalidDimensionError, InvalidMetricError, InvalidNamespaceError, InvalidTimestampError
 import aws_embedded_metrics.constants as constants
+import datetime
+from aws_embedded_metrics import utils, constants
 
 
 def validate_dimension_set(dimension_set: Dict[str, str]) -> None:
@@ -114,3 +116,32 @@ def validate_namespace(namespace: str) -> None:
 
     if not re.match(constants.VALID_NAMESPACE_REGEX, namespace):
         raise InvalidNamespaceError(f"Namespace contains invalid characters: {namespace}")
+
+
+def validate_timestamp(timestamp: datetime) -> None:
+    """
+    Validates a timestamp
+
+    * @see <a
+    *     href="https://docs.aws.amazon.com/AmazonCloudWatch/latest/monitoring/cloudwatch_concepts.html#about_timestamp">CloudWatch
+    *     Timestamp</a>
+
+        Parameters:
+            timestamp (datetime): datetime to validate
+
+        Raises:
+            InvalidTimestampError: if datetime is invalid
+    """
+    if (timestamp is None):
+        raise InvalidTimestampError(f"Timestamp cannot be none")
+
+    given_time_in_milliseconds = utils.convert_to_milliseconds(timestamp)
+    current_time_in_milliseconds = utils.convert_to_milliseconds(datetime.datetime.now())
+
+    if (given_time_in_milliseconds <= (current_time_in_milliseconds - constants.MAX_TIMESTAMP_PAST_AGE_SECONDS)):
+        raise InvalidTimestampError(
+            f"Timestamp {str(timestamp)} must not be older than {str(constants.MAX_TIMESTAMP_PAST_AGE_SECONDS)} milliseconds")
+
+    if (given_time_in_milliseconds >= (current_time_in_milliseconds + constants.MAX_TIMESTAMP_FUTURE_AGE_SECONDS)):
+        raise InvalidTimestampError(
+            f"Timestamp {str(timestamp)} must not be newer than {str(constants.MAX_TIMESTAMP_FUTURE_AGE_SECONDS)} milliseconds")

tests/logger/test_metrics_context.py

@@ -1,13 +1,14 @@
+import datetime
 import pytest
 import math
 import random
-from aws_embedded_metrics import constants
+from aws_embedded_metrics import constants, utils
 from aws_embedded_metrics.unit import Unit
 from aws_embedded_metrics.storage_resolution 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
-from aws_embedded_metrics.exceptions import DimensionSetExceededError, InvalidDimensionError, InvalidMetricError
+from aws_embedded_metrics.constants import DEFAULT_NAMESPACE, MAX_TIMESTAMP_FUTURE_AGE_SECONDS, MAX_TIMESTAMP_PAST_AGE_SECONDS
+from aws_embedded_metrics.exceptions import DimensionSetExceededError, InvalidDimensionError, InvalidMetricError, InvalidTimestampError
 from importlib import reload
 from faker import Faker
 
@@ -467,3 +468,35 @@ def generate_dimension_set(dimensions_to_add):
         dimension_set[f"{i}"] = fake.word()
 
     return dimension_set
+
+
+def test_set_timestamp_verify_timestamp():
+    context = MetricsContext()
+    context.put_metric("TestMetric", 0)
+
+    now = datetime.datetime.now()
+    context.set_timestamp(now)
+
+    assert context.meta[constants.TIMESTAMP] == utils.convert_to_milliseconds(now)
+
+
+def test_set_timestamp_null_raise_exception():
+    context = MetricsContext()
+    past_date = None
+    with pytest.raises(InvalidTimestampError):
+        context.set_timestamp(past_date)
+
+
+def test_set_timestamp_past_14_days_raise_exception():
+    context = MetricsContext()
+    past_date = datetime.datetime.now() - datetime.timedelta(milliseconds=MAX_TIMESTAMP_PAST_AGE_SECONDS + 1)
+    with pytest.raises(InvalidTimestampError):
+        context.set_timestamp(past_date)
+
+
+def test_set_timestamp_future_2_hours_raise_exception():
+    context = MetricsContext()
+    future_date = datetime.datetime.now() + datetime.timedelta(milliseconds=MAX_TIMESTAMP_FUTURE_AGE_SECONDS + 1)
+
+    with pytest.raises(InvalidTimestampError):
+        context.set_timestamp(future_date)