Module `aws_lambda_powertools.metrics`

CloudWatch Embedded Metric Format utility

Expand source code

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

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

Sub-modules

aws_lambda_powertools.metrics.base
aws_lambda_powertools.metrics.exceptions
aws_lambda_powertools.metrics.metric
aws_lambda_powertools.metrics.metrics

Functions

def single_metric(name: str, unit: MetricUnit, value: float, namespace: Optional[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

with single_metric(name="ColdStart", unit=MetricUnit.Count, value=1, namespace="ServerlessAirline") as metric:
    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

with single_metric(name="ColdStart", unit=MetricUnit.Count, value=1) as metric:
    metric.add_dimension(name="function_version", value="47")

Parameters

name : str: Metric name
unit : MetricUnit: aws_lambda_powertools.helper.models.MetricUnit
value : float: Metric value
namespace : str: Namespace for metrics

Yields

SingleMetric: SingleMetric class instance

Raises

MetricUnitError: When metric metric 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, namespace: Optional[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

        with single_metric(name="ColdStart", unit=MetricUnit.Count, value=1, namespace="ServerlessAirline") as metric:
            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

        with single_metric(name="ColdStart", unit=MetricUnit.Count, value=1) as metric:
            metric.add_dimension(name="function_version", value="47")

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

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

    Raises
    ------
    MetricUnitError
        When metric metric 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)
        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
- log_metrics
- serialize_metric_set

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 metric 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 metric 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
- 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