Refactoring logic + mypy still failling

Author: leandrodamascena

aws_lambda_powertools/metrics/functions.py

@@ -1,10 +1,13 @@
 from __future__ import annotations
 
+from datetime import datetime, timezone
+
 from aws_lambda_powertools.metrics.provider.cloudwatch_emf.exceptions import (
     MetricResolutionError,
     MetricUnitError,
 )
 from aws_lambda_powertools.metrics.provider.cloudwatch_emf.metric_properties import MetricResolution, MetricUnit
+from aws_lambda_powertools.shared import constants
 from aws_lambda_powertools.shared.types import List
 
 
@@ -69,3 +72,57 @@ def extract_cloudwatch_metric_unit_value(metric_units: List, metric_valid_option
         unit = unit.value
 
     return unit
+
+
+def validate_emf_timestamp(timestamp: int | datetime) -> bool:
+    """
+    Validates a given timestamp based on CloudWatch Timestamp guidelines.
+
+    Timestamp must meet CloudWatch requirements, otherwise an InvalidTimestampError will be raised.
+    See [Timestamps](https://docs.aws.amazon.com/AmazonCloudWatch/latest/monitoring/cloudwatch_concepts.html#about_timestamp)
+    for valid values.
+
+    Parameters:
+    ----------
+    timestamp: int | datetime
+        Datetime object or epoch time representing the timestamp to validate.
+
+    Returns
+    -------
+    bool
+        Valid or not timestamp values
+    """
+
+    if not isinstance(timestamp, (int, datetime)):
+        return False
+
+    if isinstance(timestamp, datetime):
+        # Assuming the integer timestamp represents seconds since the epoch
+        timestamp = int(timestamp.timestamp() * 1000)
+
+    current_time = int(datetime.now(timezone.utc).timestamp() * 1000)
+    min_valid_timestamp = current_time - constants.EMF_MAX_TIMESTAMP_PAST_AGE
+    max_valid_timestamp = current_time + constants.EMF_MAX_TIMESTAMP_FUTURE_AGE
+
+    return min_valid_timestamp <= timestamp <= max_valid_timestamp
+
+
+def convert_timestamp_to_emf_format(timestamp: int | datetime) -> int:
+    """
+    Converts a timestamp to EMF compatible format.
+
+    Parameters
+    ----------
+    timestamp: int | datetime
+        The timestamp to convert. If already in milliseconds format, returns it as is.
+        If datetime object, converts it to milliseconds since Unix epoch.
+
+    Returns:
+    --------
+    int
+        The timestamp converted to EMF compatible format (milliseconds since Unix epoch).
+    """
+    if isinstance(timestamp, int):
+        return timestamp
+
+    return int(round(timestamp.timestamp() * 1000))

aws_lambda_powertools/metrics/provider/cloudwatch_emf/cloudwatch.py

@@ -14,6 +14,7 @@
 from aws_lambda_powertools.metrics.functions import (
     extract_cloudwatch_metric_resolution_value,
     extract_cloudwatch_metric_unit_value,
+    validate_emf_timestamp,
 )
 from aws_lambda_powertools.metrics.provider.base import BaseProvider
 from aws_lambda_powertools.metrics.provider.cloudwatch_emf.constants import MAX_DIMENSIONS, MAX_METRICS
@@ -305,15 +306,17 @@ 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):
+    def set_timestamp(self, timestamp: int | datetime.datetime):
+        if not validate_emf_timestamp(timestamp):
             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.",
+                "The timestamp must be a Datetime object or an integer representing an epoch time. "
+                "This should not exceed 14 days in the past or be more than 2 hours in the future. "
+                "Any metrics failing to meet this criteria will be skipped by Amazon CloudWatch. "
+                "Please check the EMFValidationErrors metric in AWS/Logs namespace for more details.",
                 stacklevel=2,
             )
-        else:
-            self.timestamp = timestamp
+
+        self.timestamp = timestamp
 
     def clear_metrics(self) -> None:
         logger.debug("Clearing out existing metric set from memory")

aws_lambda_powertools/shared/constants.py

@@ -39,6 +39,11 @@
 METRICS_NAMESPACE_ENV: str = "POWERTOOLS_METRICS_NAMESPACE"
 DATADOG_FLUSH_TO_LOG: str = "DD_FLUSH_TO_LOG"
 SERVICE_NAME_ENV: str = "POWERTOOLS_SERVICE_NAME"
+# If the timestamp of log event is more than 2 hours in future, the log event is skipped.
+# If the timestamp of log event is more than 14 days in past, the log event is skipped.
+# See https://docs.aws.amazon.com/AmazonCloudWatch/latest/logs/AgentReference.html
+EMF_MAX_TIMESTAMP_PAST_AGE = 14 * 24 * 60 * 60 * 1000  # 14 days
+EMF_MAX_TIMESTAMP_FUTURE_AGE = 2 * 60 * 60 * 1000  # 2 hours
 
 # Parameters constants
 PARAMETERS_SSM_DECRYPT_ENV: str = "POWERTOOLS_PARAMETERS_SSM_DECRYPT"