feat(metrics): support to bring your own metrics provider

.gitleaks.toml

@@ -0,0 +1,15 @@
+# Title for the gitleaks configuration file.
+title = "Gitleaks"
+
+[extend]
+# useDefault will extend the base configuration with the default gitleaks config:
+# https://github.com/zricethezav/gitleaks/blob/master/config/gitleaks.toml
+useDefault = true
+
+[allowlist]
+description = "Allow list false positive"
+
+# Allow list paths to ignore due to false positives.
+paths = [
+  '''tests/unit/parser/test_kinesis_firehose\.py''',
+]

aws_lambda_powertools/metrics/__init__.py

@@ -1,23 +1,22 @@
 """CloudWatch Embedded Metric Format utility
 """
-from .base import MetricResolution, MetricUnit
-from .exceptions import (
+from aws_lambda_powertools.metrics.base import MetricResolution, MetricUnit, single_metric
+from aws_lambda_powertools.metrics.exceptions import (
     MetricResolutionError,
     MetricUnitError,
     MetricValueError,
     SchemaValidationError,
 )
-from .metric import single_metric
-from .metrics import EphemeralMetrics, Metrics
+from aws_lambda_powertools.metrics.metrics import EphemeralMetrics, Metrics
 
 __all__ = [
-    "Metrics",
-    "EphemeralMetrics",
     "single_metric",
-    "MetricUnit",
     "MetricUnitError",
-    "MetricResolution",
     "MetricResolutionError",
     "SchemaValidationError",
     "MetricValueError",
+    "Metrics",
+    "EphemeralMetrics",
+    "MetricResolution",
+    "MetricUnit",
 ]

aws_lambda_powertools/metrics/base.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import datetime
 import functools
 import json
@@ -7,59 +9,28 @@
 import warnings
 from collections import defaultdict
 from contextlib import contextmanager
-from enum import Enum
 from typing import Any, Callable, Dict, Generator, List, Optional, Union
 
-from ..shared import constants
-from ..shared.functions import resolve_env_var_choice
-from .exceptions import (
+from aws_lambda_powertools.metrics.exceptions import (
     MetricResolutionError,
     MetricUnitError,
     MetricValueError,
     SchemaValidationError,
 )
-from .types import MetricNameUnitResolution
+from aws_lambda_powertools.metrics.provider.cloudwatch_emf import cold_start
+from aws_lambda_powertools.metrics.provider.cloudwatch_emf.cold_start import (
+    reset_cold_start_flag,  # noqa: F401  # backwards compatibility
+)
+from aws_lambda_powertools.metrics.provider.cloudwatch_emf.constants import MAX_DIMENSIONS, MAX_METRICS
+from aws_lambda_powertools.metrics.provider.cloudwatch_emf.metric_properties import MetricResolution, MetricUnit
+from aws_lambda_powertools.metrics.types import MetricNameUnitResolution
+from aws_lambda_powertools.shared import constants
+from aws_lambda_powertools.shared.functions import resolve_env_var_choice
 
 logger = logging.getLogger(__name__)
 
-MAX_METRICS = 100
-MAX_DIMENSIONS = 29
-
-is_cold_start = True
-
-
-class MetricResolution(Enum):
-    Standard = 60
-    High = 1
-
-
-class MetricUnit(Enum):
-    Seconds = "Seconds"
-    Microseconds = "Microseconds"
-    Milliseconds = "Milliseconds"
-    Bytes = "Bytes"
-    Kilobytes = "Kilobytes"
-    Megabytes = "Megabytes"
-    Gigabytes = "Gigabytes"
-    Terabytes = "Terabytes"
-    Bits = "Bits"
-    Kilobits = "Kilobits"
-    Megabits = "Megabits"
-    Gigabits = "Gigabits"
-    Terabits = "Terabits"
-    Percent = "Percent"
-    Count = "Count"
-    BytesPerSecond = "Bytes/Second"
-    KilobytesPerSecond = "Kilobytes/Second"
-    MegabytesPerSecond = "Megabytes/Second"
-    GigabytesPerSecond = "Gigabytes/Second"
-    TerabytesPerSecond = "Terabytes/Second"
-    BitsPerSecond = "Bits/Second"
-    KilobitsPerSecond = "Kilobits/Second"
-    MegabitsPerSecond = "Megabits/Second"
-    GigabitsPerSecond = "Gigabits/Second"
-    TerabitsPerSecond = "Terabits/Second"
-    CountPerSecond = "Count/Second"
+# Maintenance: alias due to Hyrum's law
+is_cold_start = cold_start.is_cold_start
 
 
 class MetricManager:
@@ -94,11 +65,11 @@ class MetricManager:
 
     def __init__(
         self,
-        metric_set: Optional[Dict[str, Any]] = None,
-        dimension_set: Optional[Dict] = None,
-        namespace: Optional[str] = None,
-        metadata_set: Optional[Dict[str, Any]] = None,
-        service: Optional[str] = None,
+        metric_set: Dict[str, Any] | None = None,
+        dimension_set: Dict | None = None,
+        namespace: str | None = None,
+        metadata_set: Dict[str, Any] | None = None,
+        service: str | None = None,
     ):
         self.metric_set = metric_set if metric_set is not None else {}
         self.dimension_set = dimension_set if dimension_set is not None else {}
@@ -112,9 +83,9 @@ def __init__(
     def add_metric(
         self,
         name: str,
-        unit: Union[MetricUnit, str],
+        unit: MetricUnit | str,
         value: float,
-        resolution: Union[MetricResolution, int] = 60,
+        resolution: MetricResolution | int = 60,
     ) -> None:
         """Adds given metric
 
@@ -173,9 +144,9 @@ def add_metric(
 
     def serialize_metric_set(
         self,
-        metrics: Optional[Dict] = None,
-        dimensions: Optional[Dict] = None,
-        metadata: Optional[Dict] = None,
+        metrics: Dict | None = None,
+        dimensions: Dict | None = None,
+        metadata: Dict | None = None,
     ) -> Dict:
         """Serializes metric and dimensions set
 
@@ -355,10 +326,10 @@ def flush_metrics(self, raise_on_empty_metrics: bool = False) -> None:
 
     def log_metrics(
         self,
-        lambda_handler: Union[Callable[[Dict, Any], Any], Optional[Callable[[Dict, Any, Optional[Dict]], Any]]] = None,
+        lambda_handler: Callable[[Dict, Any], Any] | Optional[Callable[[Dict, Any, Optional[Dict]], Any]] = None,
         capture_cold_start_metric: bool = False,
         raise_on_empty_metrics: bool = False,
-        default_dimensions: Optional[Dict[str, str]] = None,
+        default_dimensions: Dict[str, str] | None = None,
     ):
         """Decorator to serialize and publish metrics at the end of a function execution.
 
@@ -537,9 +508,9 @@ class SingleMetric(MetricManager):
     def add_metric(
         self,
         name: str,
-        unit: Union[MetricUnit, str],
+        unit: MetricUnit | str,
         value: float,
-        resolution: Union[MetricResolution, int] = 60,
+        resolution: MetricResolution | int = 60,
     ) -> None:
         """Method to prevent more than one metric being created
 
@@ -565,9 +536,9 @@ 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,
+    resolution: MetricResolution | int = 60,
+    namespace: str | None = None,
+    default_dimensions: Dict[str, str] | None = None,
 ) -> Generator[SingleMetric, None, None]:
     """Context manager to simplify creation of a single metric
 
@@ -622,7 +593,7 @@ def single_metric(
     SchemaValidationError
         When metric object fails EMF schema validation
     """  # noqa: E501
-    metric_set: Optional[Dict] = None
+    metric_set: Dict | None = None
     try:
         metric: SingleMetric = SingleMetric(namespace=namespace)
         metric.add_metric(name=name, unit=unit, value=value, resolution=resolution)
@@ -635,9 +606,3 @@ def single_metric(
         metric_set = metric.serialize_metric_set()
     finally:
         print(json.dumps(metric_set, separators=(",", ":")))
-
-
-def reset_cold_start_flag():
-    global is_cold_start
-    if not is_cold_start:
-        is_cold_start = True

aws_lambda_powertools/metrics/exceptions.py

@@ -1,13 +1,4 @@
-class MetricUnitError(Exception):
-    """When metric unit is not supported by CloudWatch"""
-
-    pass
-
-
-class MetricResolutionError(Exception):
-    """When metric resolution is not supported by CloudWatch"""
-
-    pass
+from aws_lambda_powertools.metrics.provider.cloudwatch_emf.exceptions import MetricResolutionError, MetricUnitError
 
 
 class SchemaValidationError(Exception):
@@ -20,3 +11,6 @@ class MetricValueError(Exception):
     """When metric value isn't a valid number"""
 
     pass
+
+
+__all__ = ["MetricUnitError", "MetricResolutionError", "SchemaValidationError", "MetricValueError"]

aws_lambda_powertools/metrics/metric.py

@@ -1,4 +1,4 @@
 # NOTE: prevents circular inheritance import
-from .base import SingleMetric, single_metric
+from aws_lambda_powertools.metrics.base import SingleMetric, single_metric
 
 __all__ = ["SingleMetric", "single_metric"]