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 = 14 * 24 * 60 * 60 * 1000  # 14 days
+MAX_TIMESTAMP_FUTURE_AGE = 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,8 @@
 # limitations under the License.
 
 
-from aws_embedded_metrics import constants, utils
+from datetime import datetime
+from aws_embedded_metrics import constants, utils, 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 +40,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 +177,21 @@ 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 the timestamp field in the metadata.
+
+        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.
+
+        Parameters:
+            timestamp (datetime): The timestamp value to be set.
+
+        Raises:
+            InvalidTimestampError: If the provided 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

@@ -17,7 +17,9 @@
 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
-import aws_embedded_metrics.constants as constants
+from aws_embedded_metrics.exceptions import InvalidTimestampError
+from datetime import datetime
+from aws_embedded_metrics import constants, utils
 
 
 def validate_dimension_set(dimension_set: Dict[str, str]) -> None:
@@ -114,3 +116,39 @@ 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 given timestamp based on CloudWatch Timestamp guidelines.
+
+    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.
+
+    Parameters:
+        timestamp (datetime): Datetime object representing the timestamp to validate.
+
+    Raises:
+        InvalidTimestampError: If the timestamp is either None, too old, or too far in the future.
+    """
+    if not timestamp:
+        raise InvalidTimestampError("Timestamp must be a valid datetime object")
+
+    timestamp_past_age_error_message = f"Timestamp {str(timestamp)} must not be older than {str(constants.MAX_TIMESTAMP_PAST_AGE)} milliseconds"
+    timestamp_future_age_error_message = f"Timestamp {str(timestamp)} must not be newer than {str(constants.MAX_TIMESTAMP_FUTURE_AGE)} milliseconds"
+
+    if timestamp == datetime.min:
+        raise InvalidTimestampError(timestamp_past_age_error_message)
+
+    if timestamp == datetime.max:
+        raise InvalidTimestampError(timestamp_future_age_error_message)
+
+    given_time_in_milliseconds = utils.convert_to_milliseconds(timestamp)
+    current_time_in_milliseconds = utils.now()
+
+    if given_time_in_milliseconds < (current_time_in_milliseconds - constants.MAX_TIMESTAMP_PAST_AGE):
+        raise InvalidTimestampError(timestamp_past_age_error_message)
+
+    if given_time_in_milliseconds > (current_time_in_milliseconds + constants.MAX_TIMESTAMP_FUTURE_AGE):
+        raise InvalidTimestampError(timestamp_future_age_error_message)

tests/logger/test_metrics_context.py

@@ -1,15 +1,17 @@
+from faker import Faker
+from importlib import reload
+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.constants import DEFAULT_NAMESPACE, MAX_TIMESTAMP_FUTURE_AGE, MAX_TIMESTAMP_PAST_AGE
 from aws_embedded_metrics.exceptions import DimensionSetExceededError, InvalidDimensionError, InvalidMetricError
-from importlib import reload
-from faker import Faker
+from aws_embedded_metrics.exceptions import InvalidTimestampError
 
 fake = Faker()
 
@@ -458,6 +460,43 @@ def test_cannot_put_more_than_30_dimensions():
         context.put_dimensions(dimension_set)
 
 
+@pytest.mark.parametrize(
+    "timestamp",
+    [
+        datetime.datetime.now(),
+        datetime.datetime.now() - datetime.timedelta(milliseconds=MAX_TIMESTAMP_PAST_AGE - 5000),
+        datetime.datetime.now() + datetime.timedelta(milliseconds=MAX_TIMESTAMP_FUTURE_AGE - 5000)
+    ]
+)
+def test_set_timestamp_sets_timestamp(timestamp: datetime.datetime):
+    context = MetricsContext()
+
+    context.set_timestamp(timestamp)
+
+    assert context.meta[constants.TIMESTAMP] == utils.convert_to_milliseconds(timestamp)
+
+
+@pytest.mark.parametrize(
+    "timestamp",
+    [
+        None,
+        datetime.datetime.min,
+        datetime.datetime.max,
+        datetime.datetime(1, 1, 1, 0, 0, 0, 0, None),
+        datetime.datetime(1, 1, 1),
+        datetime.datetime(1, 1, 1, 0, 0),
+        datetime.datetime(9999, 12, 31, 23, 59, 59, 999999),
+        datetime.datetime.now() - datetime.timedelta(milliseconds=MAX_TIMESTAMP_PAST_AGE + 5000),
+        datetime.datetime.now() + datetime.timedelta(milliseconds=MAX_TIMESTAMP_FUTURE_AGE + 5000)
+    ]
+)
+def test_set_timestamp_raise_exception(timestamp: datetime.datetime):
+    context = MetricsContext()
+
+    with pytest.raises(InvalidTimestampError):
+        context.set_timestamp(timestamp)
+
+
 # Test utility method