Module aws_lambda_powertools.metrics

CloudWatch Embedded Metric Format utility

"""CloudWatch Embedded Metric Format utility
"""
from .base import MetricResolution, MetricUnit
from .exceptions import (
    MetricResolutionError,
    MetricUnitError,
    MetricValueError,
    SchemaValidationError,
)
from .metric import single_metric
from .metrics import EphemeralMetrics, Metrics

__all__ = [
    "Metrics",
    "EphemeralMetrics",
    "single_metric",
    "MetricUnit",
    "MetricUnitError",
    "MetricResolution",
    "MetricResolutionError",
    "SchemaValidationError",
    "MetricValueError",
]

Sub-modules

aws_lambda_powertools.metrics.base

aws_lambda_powertools.metrics.exceptions

aws_lambda_powertools.metrics.metric

aws_lambda_powertools.metrics.metrics

aws_lambda_powertools.metrics.types

Functions

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) ‑> Generator[SingleMetric, None, None]

Context manager to simplify creation of a single metric

Example

Creates cold start metric with function_version as dimension

from aws_lambda_powertools import single_metric
from aws_lambda_powertools.metrics import MetricUnit
from aws_lambda_powertools.metrics import MetricResolution

with single_metric(name="ColdStart", unit=MetricUnit.Count, value=1, resolution=MetricResolution.Standard, namespace="ServerlessAirline") as metric: # noqa E501
    metric.add_dimension(name="function_version", value="47")

Same as above but set namespace using environment variable

$ export POWERTOOLS_METRICS_NAMESPACE="ServerlessAirline"

from aws_lambda_powertools import single_metric
from aws_lambda_powertools.metrics import MetricUnit
from aws_lambda_powertools.metrics import MetricResolution

with single_metric(name="ColdStart", unit=MetricUnit.Count, value=1, resolution=MetricResolution.Standard) as metric: # noqa E501
    metric.add_dimension(name="function_version", value="47")

Parameters

name : str

Metric name

unit : MetricUnit

aws_lambda_powertools.helper.models.MetricUnit

resolution : MetricResolution

aws_lambda_powertools.helper.models.MetricResolution

value : float

Metric value

namespace : str

Namespace for metrics

Yields

SingleMetric

SingleMetric class instance

Raises

MetricUnitError

When metric metric isn't supported by CloudWatch

MetricResolutionError

When metric resolution isn't supported by CloudWatch

MetricValueError

When metric value isn't a number

SchemaValidationError

When metric object fails EMF schema validation

Expand source code

@contextmanager
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,
) -> Generator[SingleMetric, None, None]:
    """Context manager to simplify creation of a single metric

    Example
    -------
    **Creates cold start metric with function_version as dimension**

        from aws_lambda_powertools import single_metric
        from aws_lambda_powertools.metrics import MetricUnit
        from aws_lambda_powertools.metrics import MetricResolution

        with single_metric(name="ColdStart", unit=MetricUnit.Count, value=1, resolution=MetricResolution.Standard, namespace="ServerlessAirline") as metric: # noqa E501
            metric.add_dimension(name="function_version", value="47")

    **Same as above but set namespace using environment variable**

        $ export POWERTOOLS_METRICS_NAMESPACE="ServerlessAirline"

        from aws_lambda_powertools import single_metric
        from aws_lambda_powertools.metrics import MetricUnit
        from aws_lambda_powertools.metrics import MetricResolution

        with single_metric(name="ColdStart", unit=MetricUnit.Count, value=1, resolution=MetricResolution.Standard) as metric: # noqa E501
            metric.add_dimension(name="function_version", value="47")

    Parameters
    ----------
    name : str
        Metric name
    unit : MetricUnit
        `aws_lambda_powertools.helper.models.MetricUnit`
    resolution : MetricResolution
        `aws_lambda_powertools.helper.models.MetricResolution`
    value : float
        Metric value
    namespace: str
        Namespace for metrics

    Yields
    -------
    SingleMetric
        SingleMetric class instance

    Raises
    ------
    MetricUnitError
        When metric metric isn't supported by CloudWatch
    MetricResolutionError
        When metric resolution isn't supported by CloudWatch
    MetricValueError
        When metric value isn't a number
    SchemaValidationError
        When metric object fails EMF schema validation
    """
    metric_set: Optional[Dict] = None
    try:
        metric: SingleMetric = SingleMetric(namespace=namespace)
        metric.add_metric(name=name, unit=unit, value=value, resolution=resolution)

        if default_dimensions:
            for dim_name, dim_value in default_dimensions.items():
                metric.add_dimension(name=dim_name, value=dim_value)

        yield metric
        metric_set = metric.serialize_metric_set()
    finally:
        print(json.dumps(metric_set, separators=(",", ":")))

Classes

class EphemeralMetrics (service: Optional[str] = None, namespace: Optional[str] = None)

Non-singleton version of Metrics to not persist metrics across instances

NOTE: This is useful when you want to:

• Create metrics for distinct namespaces
• Create the same metrics with different dimensions more than once

Expand source code

class EphemeralMetrics(MetricManager):
    """Non-singleton version of Metrics to not persist metrics across instances

    NOTE: This is useful when you want to:

    - Create metrics for distinct namespaces
    - Create the same metrics with different dimensions more than once
    """

    def __init__(self, service: Optional[str] = None, namespace: Optional[str] = None):
        super().__init__(namespace=namespace, service=service)

Ancestors

• MetricManager

Inherited members

• MetricManager:
  • add_dimension
  • add_metadata
  • add_metric
  • flush_metrics
  • log_metrics
  • serialize_metric_set

class MetricResolution (value, names=None, *, module=None, qualname=None, type=None, start=1)

An enumeration.

Expand source code

class MetricResolution(Enum):
    Standard = 60
    High = 1

Ancestors

• enum.Enum

Class variables

var High

var Standard

class MetricResolutionError (*args, **kwargs)

When metric resolution is not supported by CloudWatch

Expand source code

class MetricResolutionError(Exception):
    """When metric resolution is not supported by CloudWatch"""

    pass

Ancestors

• builtins.Exception
• builtins.BaseException

class MetricUnit (value, names=None, *, module=None, qualname=None, type=None, start=1)

An enumeration.

Expand source code

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"

Ancestors

• enum.Enum

Class variables

var Bits

var BitsPerSecond

var Bytes

var BytesPerSecond

var Count

var CountPerSecond

var Gigabits

var GigabitsPerSecond

var Gigabytes

var GigabytesPerSecond

var Kilobits

var KilobitsPerSecond

var Kilobytes

var KilobytesPerSecond

var Megabits

var MegabitsPerSecond

var Megabytes

var MegabytesPerSecond

var Microseconds

var Milliseconds

var Percent

var Seconds

var Terabits

var TerabitsPerSecond

var Terabytes

var TerabytesPerSecond

class MetricUnitError (*args, **kwargs)

When metric unit is not supported by CloudWatch

Expand source code

class MetricUnitError(Exception):
    """When metric unit is not supported by CloudWatch"""

    pass

Ancestors

• builtins.Exception
• builtins.BaseException

class MetricValueError (*args, **kwargs)

When metric value isn't a valid number

Expand source code

class MetricValueError(Exception):
    """When metric value isn't a valid number"""

    pass

Ancestors

• builtins.Exception
• builtins.BaseException

class Metrics (service: Optional[str] = None, namespace: Optional[str] = None)

Metrics create an EMF object with up to 100 metrics

Use Metrics when you need to create multiple metrics that have dimensions in common (e.g. service_name="payment").

Metrics up to 100 metrics in memory and are shared across all its instances. That means it can be safely instantiated outside of a Lambda function, or anywhere else.

A decorator (log_metrics) is provided so metrics are published at the end of its execution. If more than 100 metrics are added at a given function execution, these metrics are serialized and published before adding a given metric to prevent metric truncation.

Example

Creates a few metrics and publish at the end of a function execution

from aws_lambda_powertools import Metrics

metrics = Metrics(namespace="ServerlessAirline", service="payment")

@metrics.log_metrics(capture_cold_start_metric=True)
def lambda_handler():
    metrics.add_metric(name="BookingConfirmation", unit="Count", value=1)
    metrics.add_dimension(name="function_version", value="$LATEST")

    return True

Environment Variables

POWERTOOLS_METRICS_NAMESPACE : str metric namespace POWERTOOLS_SERVICE_NAME : str service name used for default dimension

Parameters

service : str, optional

service name to be used as metric dimension, by default "service_undefined"

namespace : str, optional

Namespace for metrics

Raises

MetricUnitError

When metric unit isn't supported by CloudWatch

MetricResolutionError

When metric resolution isn't supported by CloudWatch

MetricValueError

When metric value isn't a number

SchemaValidationError

When metric object fails EMF schema validation

Expand source code

class Metrics(MetricManager):
    """Metrics create an EMF object with up to 100 metrics

    Use Metrics when you need to create multiple metrics that have
    dimensions in common (e.g. service_name="payment").

    Metrics up to 100 metrics in memory and are shared across
    all its instances. That means it can be safely instantiated outside
    of a Lambda function, or anywhere else.

    A decorator (log_metrics) is provided so metrics are published at the end of its execution.
    If more than 100 metrics are added at a given function execution,
    these metrics are serialized and published before adding a given metric
    to prevent metric truncation.

    Example
    -------
    **Creates a few metrics and publish at the end of a function execution**

        from aws_lambda_powertools import Metrics

        metrics = Metrics(namespace="ServerlessAirline", service="payment")

        @metrics.log_metrics(capture_cold_start_metric=True)
        def lambda_handler():
            metrics.add_metric(name="BookingConfirmation", unit="Count", value=1)
            metrics.add_dimension(name="function_version", value="$LATEST")

            return True

    Environment variables
    ---------------------
    POWERTOOLS_METRICS_NAMESPACE : str
        metric namespace
    POWERTOOLS_SERVICE_NAME : str
        service name used for default dimension

    Parameters
    ----------
    service : str, optional
        service name to be used as metric dimension, by default "service_undefined"
    namespace : str, optional
        Namespace for metrics

    Raises
    ------
    MetricUnitError
        When metric unit isn't supported by CloudWatch
    MetricResolutionError
        When metric resolution isn't supported by CloudWatch
    MetricValueError
        When metric value isn't a number
    SchemaValidationError
        When metric object fails EMF schema validation
    """

    # NOTE: We use class attrs to share metrics data across instances
    # this allows customers to initialize Metrics() throughout their code base (and middlewares)
    # and not get caught by accident with metrics data loss, or data deduplication
    # e.g., m1 and m2 add metric ProductCreated, however m1 has 'version' dimension  but m2 doesn't
    # Result: ProductCreated is created twice as we now have 2 different EMF blobs
    _metrics: Dict[str, Any] = {}
    _dimensions: Dict[str, str] = {}
    _metadata: Dict[str, Any] = {}
    _default_dimensions: Dict[str, Any] = {}

    def __init__(self, service: Optional[str] = None, namespace: Optional[str] = None):
        self.metric_set = self._metrics
        self.metadata_set = self._metadata
        self.default_dimensions = self._default_dimensions
        self.dimension_set = self._dimensions

        self.dimension_set.update(**self._default_dimensions)
        return super().__init__(
            namespace=namespace,
            service=service,
            metric_set=self.metric_set,
            dimension_set=self.dimension_set,
            metadata_set=self.metadata_set,
        )

    def set_default_dimensions(self, **dimensions) -> None:
        """Persist dimensions across Lambda invocations

        Parameters
        ----------
        dimensions : Dict[str, Any], optional
            metric dimensions as key=value

        Example
        -------
        **Sets some default dimensions that will always be present across metrics and invocations**

            from aws_lambda_powertools import Metrics

            metrics = Metrics(namespace="ServerlessAirline", service="payment")
            metrics.set_default_dimensions(environment="demo", another="one")

            @metrics.log_metrics()
            def lambda_handler():
                return True
        """
        for name, value in dimensions.items():
            self.add_dimension(name, value)

        self.default_dimensions.update(**dimensions)

    def clear_default_dimensions(self) -> None:
        self.default_dimensions.clear()

    def clear_metrics(self) -> None:
        super().clear_metrics()
        # re-add default dimensions
        self.set_default_dimensions(**self.default_dimensions)

Ancestors

• MetricManager

Methods

def clear_default_dimensions(self) ‑> None

Expand source code

def clear_default_dimensions(self) -> None:
    self.default_dimensions.clear()

def clear_metrics(self) ‑> None

Expand source code

def clear_metrics(self) -> None:
    super().clear_metrics()
    # re-add default dimensions
    self.set_default_dimensions(**self.default_dimensions)

def set_default_dimensions(self, **dimensions) ‑> None

Persist dimensions across Lambda invocations

Parameters

dimensions : Dict[str, Any], optional

metric dimensions as key=value

Example

Sets some default dimensions that will always be present across metrics and invocations

from aws_lambda_powertools import Metrics

metrics = Metrics(namespace="ServerlessAirline", service="payment")
metrics.set_default_dimensions(environment="demo", another="one")

@metrics.log_metrics()
def lambda_handler():
    return True

Expand source code

def set_default_dimensions(self, **dimensions) -> None:
    """Persist dimensions across Lambda invocations

    Parameters
    ----------
    dimensions : Dict[str, Any], optional
        metric dimensions as key=value

    Example
    -------
    **Sets some default dimensions that will always be present across metrics and invocations**

        from aws_lambda_powertools import Metrics

        metrics = Metrics(namespace="ServerlessAirline", service="payment")
        metrics.set_default_dimensions(environment="demo", another="one")

        @metrics.log_metrics()
        def lambda_handler():
            return True
    """
    for name, value in dimensions.items():
        self.add_dimension(name, value)

    self.default_dimensions.update(**dimensions)

Inherited members

• MetricManager:
  • add_dimension
  • add_metadata
  • add_metric
  • flush_metrics
  • log_metrics
  • serialize_metric_set

class SchemaValidationError (*args, **kwargs)

When serialization fail schema validation

Expand source code

class SchemaValidationError(Exception):
    """When serialization fail schema validation"""

    pass

Ancestors

• builtins.Exception
• builtins.BaseException