Draft - Metrics EMF

__init__.py

"""CloudWatch Embedded Metric Format utility
"""
from lambda_python_powertools.helper.models import MetricUnit

from .exceptions import (
    MetricUnitError,
    MetricValueError,
    SchemaValidationError,
    UniqueNamespaceError,
)
from .metric import single_metric
from .metrics import Metrics

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

base.py

import datetime
import json
import logging
import numbers
import os
from typing import Dict, List, Union

import jsonschema

from lambda_python_powertools.helper.models import MetricUnit

from .exceptions import MetricUnitError, MetricValueError, SchemaValidationError, UniqueNamespaceError

logger = logging.getLogger(__name__)
logger.setLevel(os.getenv("LOG_LEVEL", "INFO"))

CLOUDWATCH_EMF_SCHEMA = {
    "type": "object",
    "title": "Root Node",
    "required": ["_aws"],
    "properties": {
        "_aws": {
            "$id": "#/properties/_aws",
            "type": "object",
            "title": "Metadata",
            "required": ["Timestamp", "CloudWatchMetrics"],
            "properties": {
                "Timestamp": {
                    "$id": "#/properties/_aws/properties/Timestamp",
                    "type": "integer",
                    "title": "The Timestamp Schema",
                    "examples": [1565375354953],
                },
                "CloudWatchMetrics": {
                    "$id": "#/properties/_aws/properties/CloudWatchMetrics",
                    "type": "array",
                    "title": "MetricDirectives",
                    "items": {
                        "$id": "#/properties/_aws/properties/CloudWatchMetrics/items",
                        "type": "object",
                        "title": "MetricDirective",
                        "required": ["Namespace", "Dimensions", "Metrics"],
                        "properties": {
                            "Namespace": {
                                "$id": "#/properties/_aws/properties/CloudWatchMetrics/items/properties/Namespace",
                                "type": "string",
                                "title": "CloudWatch Metrics Namespace",
                                "examples": ["MyApp"],
                                "pattern": "^(.*)$",
                                "minLength": 1,
                            },
                            "Dimensions": {
                                "$id": "#/properties/_aws/properties/CloudWatchMetrics/items/properties/Dimensions",
                                "type": "array",
                                "title": "The Dimensions Schema",
                                "minItems": 1,
                                "items": {
                                    "$id": "#/properties/_aws/properties/CloudWatchMetrics/items/properties/Dimensions/items",
                                    "type": "array",
                                    "title": "DimensionSet",
                                    "minItems": 1,
                                    "maxItems": 9,
                                    "items": {
                                        "$id": "#/properties/_aws/properties/CloudWatchMetrics/items/properties/Dimensions/items/items",
                                        "type": "string",
                                        "title": "DimensionReference",
                                        "examples": ["Operation"],
                                        "pattern": "^(.*)$",
                                        "minItems": 1,
                                    },
                                },
                            },
                            "Metrics": {
                                "$id": "#/properties/_aws/properties/CloudWatchMetrics/items/properties/Metrics",
                                "type": "array",
                                "title": "MetricDefinitions",
                                "minItems": 1,
                                "items": {
                                    "$id": "#/properties/_aws/properties/CloudWatchMetrics/items/properties/Metrics/items",
                                    "type": "object",
                                    "title": "MetricDefinition",
                                    "required": ["Name"],
                                    "minItems": 1,
                                    "properties": {
                                        "Name": {
                                            "$id": "#/properties/_aws/properties/CloudWatchMetrics/items/properties/Metrics/items/properties/Name",
                                            "type": "string",
                                            "title": "MetricName",
                                            "examples": ["ProcessingLatency"],
                                            "pattern": "^(.*)$",
                                            "minLength": 1,
                                        },
                                        "Unit": {
                                            "$id": "#/properties/_aws/properties/CloudWatchMetrics/items/properties/Metrics/items/properties/Unit",
                                            "type": "string",
                                            "title": "MetricUnit",
                                            "examples": ["Milliseconds"],
                                            "pattern": "^(Seconds|Microseconds|Milliseconds|Bytes|Kilobytes|Megabytes|Gigabytes|Terabytes|Bits|Kilobits|Megabits|Gigabits|Terabits|Percent|Count|Bytes\\/Second|Kilobytes\\/Second|Megabytes\\/Second|Gigabytes\\/Second|Terabytes\\/Second|Bits\\/Second|Kilobits\\/Second|Megabits\\/Second|Gigabits\\/Second|Terabits\\/Second|Count\\/Second|None)$",
                                        },
                                    },
                                },
                            },
                        },
                    },
                },
            },
        }
    },
}


class MetricManager:
    """Base class for metric functionality (namespace, metric, dimension, serialization)

    MetricManager creates metrics asynchronously thanks to CloudWatch Embedded Metric Format (EMF).
    CloudWatch EMF can create up to 100 metrics per EMF object
    and metrics, dimensions, and namespace created via MetricManager
    will adhere to the specification[1], will be serialized and validated against EMF Schema[1].

    Use Metrics and SingleMetric classes to create EMF metrics.

    [1] https://docs.aws.amazon.com/AmazonCloudWatch/latest/monitoring/CloudWatch_Embedded_Metric_Format_Specification.html

    Environment variables
    ---------------------
    POWERTOOLS_METRICS_NAMESPACE : str
        metric namespace

    Raises
    ------
    MetricUnitError
        Raised when metric added doesn't provide correct metric unit.
    SchemaValidationError
        Raised when metric object fails EMF schema validation
    """

    def __init__(
        self, metric_set: Dict[str, str] = None, dimension_set: Dict = None, namespace: str = None
    ):
        self.metric_set = metric_set or {}
        self.dimension_set = dimension_set or {}
        self.namespace = os.getenv("POWERTOOLS_METRICS_NAMESPACE") or namespace

    def add_namespace(self, name: str):
        """Adds given metric namespace

        Example
        -------
        Add metric namespace

            >>> metric.add_namespace(name="ServerlessAirline")

        Parameters
        ----------
        name : str
            Metric namespace
        """
        if self.namespace is not None:
            raise UniqueNamespaceError(f"Namespace '{self.namespace}' already set - Only one namespace is allowed across metrics")
        logger.debug(f"Adding metrics namespace: {name}")
        self.namespace = name

    def add_metric(self, name: str, unit: MetricUnit, value: Union[float, int]):
        """Adds given metric

        Example
        -------
        Add given metric using MetricUnit enum

            >>> metric.add_metric(name="BookingConfirmation", unit=MetricUnit.Count, value=1)

        Add given metric using plain string but value unit

            >>> metric.add_metric(name="BookingConfirmation", unit="Count", value=1)

        Parameters
        ----------
        name : str
            Metric name
        unit : MetricUnit
            Metric unit (e.g. "Seconds", MetricUnit.Seconds)
        value : float
            Metric value

        Raises
        ------
        MetricUnitError
            Raised when metric unit is not supported by CloudWatch
        """
        if len(self.metric_set) == 100:
            logger.debug("Exceeded maximum of 100 metrics - Publishing existing metric set")
            metrics = self.serialize_metric_set()
            print(json.dumps(metrics))
            self.metric_set = {}

        if not isinstance(value, numbers.Number):
            raise MetricValueError(f"{value} is not a valid number")

        if not isinstance(unit, MetricUnit):
            try:
                unit = MetricUnit[unit]
            except KeyError:
                unit_options = list(MetricUnit.__members__)
                raise MetricUnitError(
                    f"Invalid metric unit '{unit}', expected either option: {unit_options}"
                )

        metric = {"Unit": unit.value, "Value": float(value)}
        logger.debug(f"Adding metric: {name} with {metric}")
        self.metric_set[name] = metric

    def serialize_metric_set(self, metrics: Dict = None, dimensions: Dict = None) -> Dict:
        """Serializes metric and dimensions set

        Parameters
        ----------
        metrics : Dict, optional
            Dictionary of metrics to serialize, by default None
        dimensions : Dict, optional
            Dictionary of dimensions to serialize, by default None

        Example
        -------
        Serialize metrics into EMF format
            >>> metrics = MetricManager()
            >>> ...add metrics, dimensions, namespace
            >>> ret = metrics.serialize_metric_set()

        Returns
        -------
        Dict
            Serialized metrics following EMF specification

        Raises
        ------
        SchemaValidationError
            Raised when serialization fail schema validation
        """
        if metrics is None:
            metrics = self.metric_set

        if dimensions is None:
            dimensions = self.dimension_set

        logger.debug("Serializing...", {"metrics": metrics, "dimensions": dimensions})

        dimension_keys: List[str] = list(dimensions.keys())
        metric_names_unit: List[Dict[str, str]] = []
        metric_set: Dict[str, str] = {}

        for metric_name in metrics:
            metric: str = metrics[metric_name]
            metric_value: int = metric.get("Value", 0)
            metric_unit: str = metric.get("Unit")

            if metric_value > 0 and metric_unit is not None:
                metric_names_unit.append({"Name": metric_name, "Unit": metric["Unit"]})
                metric_set.update({metric_name: metric["Value"]})

        metrics_definition = {
            "CloudWatchMetrics": [
                {
                    "Namespace": self.namespace,
                    "Dimensions": [dimension_keys],
                    "Metrics": metric_names_unit,
                }
            ]
        }
        metrics_timestamp = {"Timestamp": int(datetime.datetime.now().timestamp() * 1000)}
        metric_set["_aws"] = {**metrics_timestamp, **metrics_definition}

        try:
            logger.debug("Validating serialized metrics against CloudWatch EMF schema", metric_set)
            jsonschema.validate(metric_set, schema=CLOUDWATCH_EMF_SCHEMA)
        except jsonschema.exceptions.ValidationError as e:
            message = f"Invalid format. Error: {e.message} ({e.validator}), Invalid item: {e.absolute_schema_path}"  # noqa: B306
            raise SchemaValidationError(message)
        return metric_set

    def add_dimension(self, name: str, value: str):
        """Adds given dimension to all metrics

        Parameters
        ----------
        name : str
            Dimension name
        value : str
            Dimension value
        """
        logger.debug(f"Adding dimension: {name}:{value}")
        self.dimension_set[name] = value

exceptions.py

class MetricUnitError(Exception):
    pass


class SchemaValidationError(Exception):
    pass


class MetricValueError(Exception):
    pass


class UniqueNamespaceError(Exception):
    pass

metric.py

import json
import logging
import os
from contextlib import contextmanager
from typing import Dict

from lambda_python_powertools.helper.models import MetricUnit
from lambda_python_powertools.metrics.base import MetricManager

logger = logging.getLogger(__name__)
logger.setLevel(os.getenv("LOG_LEVEL", "INFO"))


class SingleMetric(MetricManager):
    """SingleMetric creates an EMF object with a single metric.

    EMF specification doesn't allow metrics with different dimensions.
    SingleMetric overrides MetricManager's add_metric method to do just that.

    Use SingleMetric when you need to create metrics with different dimensions,
    and for simplicity use single_metric() context manager.

    Environment variables
    ---------------------
    POWERTOOLS_METRICS_NAMESPACE : str
        metric namespace

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

        >>> from lambda_python_powertools.metrics import SingleMetric, MetricUnit
        >>> import json
        >>> metric = Single_Metric()
        >>> metric.add_namespace(name="ServerlessAirline")
        >>> metric.add_metric(name="ColdStart", unit=MetricUnit.Count, value=1)
        >>> metric.add_dimension(name="function_version", value=47)
        >>> print(json.dumps(metric.serialize_metric_set(), indent=4))

    Parameters
    ----------
    MetricManager : MetricManager
        Inherits from MetricManager
    """

    def add_metric(self, name: str, unit: MetricUnit, value: float):
        """Method to prevent more than one metric being created

        Parameters
        ----------
        name : str
            Metric name (e.g. BookingConfirmation)
        unit : MetricUnit
            Metric unit (e.g. "Seconds", MetricUnit.Seconds)
        value : float
            Metric value
        """
        if len(self.metric_set) > 0:
            logger.debug(f"Metric {name} already set, skipping...")
            return
        return super().add_metric(name, unit, value)


@contextmanager
def single_metric(name: str, unit: MetricUnit, value: float):
    """context manager to simplify creation of a single metric

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

        >>> from lambda_python_powertools.metrics import single_metric, MetricUnit
        >>> with single_metric(name="ColdStart", unit=MetricUnit.Count, value=1) as metric:
                metric.add_namespace(name="ServerlessAirline")
                metric.add_dimension(name="function_version", value=47)

    Same as above but set namespace using environment variable

        $ export POWERTOOLS_METRICS_NAMESPACE="ServerlessAirline"
        >>> from lambda_python_powertools.metrics import single_metric, 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
        Metric unit (e.g. "Seconds", MetricUnit.Seconds)
    value : float
        Metric value

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

    Raises
    ------
    e
        Propagate error received
    """
    metric_set = None
    try:
        metric: SingleMetric = SingleMetric()
        metric.add_metric(name=name, unit=unit, value=value)
        yield metric
        logger.debug("Serializing single metric")
        metric_set: Dict = metric.serialize_metric_set()
    except Exception as e:
        logger.error(e)
        raise e
    finally:
        logger.debug("Publishing single metric", {"metric": metric})
        print(json.dumps(metric_set))

metrics.py

import functools
import json
import logging
import os
from typing import Any, Callable

from lambda_python_powertools.metrics.base import MetricManager

logger = logging.getLogger(__name__)
logger.setLevel(os.getenv("LOG_LEVEL", "INFO"))


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 lambda_python_powertools.metrics import Metrics
        >>> metrics = Metrics()
        >>> metrics.add_namespace(name="ServerlessAirline")
        >>> metrics.add_metric(name="ColdStart", unit=MetricUnit.Count, value=1)
        >>> metrics.add_metric(name="BookingConfirmation", unit="Count", value=1)
        >>> metrics.add_dimension(name="service", value="booking")
        >>> metrics.add_dimension(name="function_version", value="$LATEST")
        >>> ...
        >>> @tracer.capture_lambda_handler
        >>> @metrics.log_metrics()
        >>> def lambda_handler():
                do_something()
                return True

        >>> def do_something():
                metrics.add_metric(name="Something", unit="Count", value=1)

    Calls lambda function and creates a few metrics and publish.
    Useful log_metrics is the only decorator used, or when no other decorator calls the handler

        >>> from lambda_python_powertools.metrics import Metrics
        >>> metrics = Metrics()
        >>> metrics.add_namespace(name="ServerlessAirline")
        >>> metrics.add_dimension(name="service", value="booking")
        >>> metrics.add_dimension(name="function_version", value="$LATEST")
        >>> ...
        >>> @metrics.log_metrics(call_function=True)
        >>> def lambda_handler():
                if cold_start:
                    metrics.add_metric(name="ColdStart", unit=MetricUnit.Count, value=1)
                    metrics.add_metric(name="BookingConfirmation", unit="Count", value=1)
                do_something()
                return True

        >>> def do_something():
                metrics.add_metric


    Environment variables
    ---------------------
    POWERTOOLS_METRICS_NAMESPACE : str
        metric namespace

    Parameters
    ----------
    MetricManager : MetricManager
        Inherits from MetricManager

    Raises
    ------
    e
        Propagate error received
    """

    _metrics = {}
    _dimensions = {}

    def __init__(self, metric_set=None, dimension_set=None, namespace=None):
        super().__init__(
            metric_set=self._metrics, dimension_set=self._dimensions, namespace=namespace
        )

    def log_metrics(
        self, lambda_handler: Callable[[Any, Any], Any] = None, call_function: bool = False
    ):
        """Decorator to serialize and publish metrics at the end of a function execution.

        By default, it doesn't run the lambda function handler as other decorators
        like Tracer and Logger could be used too.
        However, if you are only using Metrics feature, use `log_metrics(call_function=True)`.

        Example
        -------
        Lambda function using tracer and metrics decorators

            >>> metrics = Metrics()
            >>> tracer = Tracer(service="payment")
            >>> @tracer.capture_lambda_handler
            >>> @metrics.log_metrics
                def handler(event, context)

        Lambda function using metrics decorator only

            >>> metrics = Metrics()
            >>> @metrics.log_metrics(call_function=True)
                def handler(event, context)

        Parameters
        ----------
        lambda_handler : Callable[[Any, Any], Any], optional
            Lambda function handler, by default None
        call_function : bool, optional
            Call function it annotates, by default False

        Raises
        ------
        e
            Propagate error received
        """
        if lambda_handler is None:
            return functools.partial(self.log_metrics, call_function=call_function)

        @functools.wraps(lambda_handler)
        def decorate(*args, **kwargs):
            try:
                if call_function:
                    logger.debug("Calling Lambda handler")
                    lambda_handler(*args, **kwargs)
                metrics = self.serialize_metric_set()
                logger.debug("Publishing metrics", {"metrics": metrics})
                print(json.dumps(metrics))
            except Exception as e:
                logger.error(e)
                raise e

        return decorate

test_metrics.py

import json
from typing import Dict, List

import pytest

from lambda_python_powertools.metrics import (
    Metrics,
    MetricUnit,
    MetricUnitError,
    MetricValueError,
    SchemaValidationError,
    UniqueNamespaceError,
    single_metric,
)
from lambda_python_powertools.metrics.base import MetricManager


@pytest.fixture
def metric() -> Dict[str, str]:
    return {"name": "single_metric", "unit": MetricUnit.Count, "value": 1}


@pytest.fixture
def metrics() -> List[Dict[str, str]]:
    return [
        {"name": "metric_one", "unit": MetricUnit.Count, "value": 1},
        {"name": "metric_two", "unit": MetricUnit.Count, "value": 1},
    ]


@pytest.fixture
def dimension() -> Dict[str, str]:
    return {"name": "test_dimension", "value": "test"}


@pytest.fixture
def dimensions() -> List[Dict[str, str]]:
    return [
        {"name": "test_dimension", "value": "test"},
        {"name": "test_dimension_2", "value": "test"},
    ]


@pytest.fixture
def namespace() -> Dict[str, str]:
    return {"name": "test_namespace"}


@pytest.fixture
def a_hundred_metrics() -> List[Dict[str, str]]:
    metrics = []
    for i in range(100):
        metrics.append({"name": f"metric_{i}", "unit": "Count", "value": 1})

    return metrics


def serialize_metrics(metrics: List[Dict], dimensions: List[Dict], namespace: Dict) -> Dict:
    """ Helper function to build EMF object from a list of metrics, dimensions """
    my_metrics = MetricManager()
    for metric in metrics:
        my_metrics.add_metric(**metric)

    for dimension in dimensions:
        my_metrics.add_dimension(**dimension)

    my_metrics.add_namespace(**namespace)
    return my_metrics.serialize_metric_set()


def serialize_single_metric(metric: Dict, dimension: Dict, namespace: Dict) -> Dict:
    """ Helper function to build EMF object from a given metric, dimension and namespace """
    my_metrics = MetricManager()
    my_metrics.add_metric(**metric)
    my_metrics.add_dimension(**dimension)
    my_metrics.add_namespace(**namespace)
    return my_metrics.serialize_metric_set()


def remove_timestamp(metrics: List):
    for metric in metrics:
        del metric["_aws"]["Timestamp"]


def test_single_metric(capsys, metric, dimension, namespace):
    with single_metric(**metric) as my_metrics:
        my_metrics.add_dimension(**dimension)
        my_metrics.add_namespace(**namespace)

    output = json.loads(capsys.readouterr().out.strip())
    expected = serialize_single_metric(metric=metric, dimension=dimension, namespace=namespace)

    remove_timestamp(metrics=[output, expected])  # Timestamp will always be different
    assert expected["_aws"] == output["_aws"]


def test_single_metric_one_metric_only(capsys, metric, dimension, namespace):
    with single_metric(**metric) as my_metrics:
        my_metrics.add_metric(name="second_metric", unit="Count", value=1)
        my_metrics.add_metric(name="third_metric", unit="Seconds", value=1)
        my_metrics.add_dimension(**dimension)
        my_metrics.add_namespace(**namespace)

    output = json.loads(capsys.readouterr().out.strip())
    expected = serialize_single_metric(metric=metric, dimension=dimension, namespace=namespace)

    remove_timestamp(metrics=[output, expected])  # Timestamp will always be different
    assert expected["_aws"] == output["_aws"]


def test_multiple_metrics(metrics, dimensions, namespace):
    my_metrics = Metrics()
    for metric in metrics:
        my_metrics.add_metric(**metric)

    for dimension in dimensions:
        my_metrics.add_dimension(**dimension)

    my_metrics.add_namespace(**namespace)
    output = my_metrics.serialize_metric_set()
    expected = serialize_metrics(metrics=metrics, dimensions=dimensions, namespace=namespace)

    remove_timestamp(metrics=[output, expected])  # Timestamp will always be different
    assert expected["_aws"] == output["_aws"]


def test_multiple_namespaces(metric, dimension, namespace):
    namespace_a = {"name": "OtherNamespace"}
    namespace_b = {"name": "AnotherNamespace"}

    with pytest.raises(UniqueNamespaceError):
        with single_metric(**metric) as m:
            m.add_dimension(**dimension)
            m.add_namespace(**namespace)
            m.add_namespace(**namespace_a)
            m.add_namespace(**namespace_b)


def test_log_metrics_no_function_call(capsys, metrics, dimensions, namespace):
    my_metrics = Metrics()
    my_metrics.add_namespace(**namespace)
    for metric in metrics:
        my_metrics.add_metric(**metric)
    for dimension in dimensions:
        my_metrics.add_dimension(**dimension)

    @my_metrics.log_metrics
    def lambda_handler(evt, handler):
        return True

    lambda_handler({}, {})
    output = json.loads(capsys.readouterr().out.strip())
    expected = serialize_metrics(metrics=metrics, dimensions=dimensions, namespace=namespace)

    remove_timestamp(metrics=[output, expected])  # Timestamp will always be different
    assert expected["_aws"] == output["_aws"]


def test_log_metrics_call_function(capsys, metrics, dimensions, namespace):
    my_metrics = Metrics()

    @my_metrics.log_metrics(call_function=True)
    def lambda_handler(evt, handler):
        my_metrics.add_namespace(**namespace)
        for metric in metrics:
            my_metrics.add_metric(**metric)
        for dimension in dimensions:
            my_metrics.add_dimension(**dimension)
        return True

    lambda_handler({}, {})

    output = json.loads(capsys.readouterr().out.strip())
    expected = serialize_metrics(metrics=metrics, dimensions=dimensions, namespace=namespace)

    remove_timestamp(metrics=[output, expected])  # Timestamp will always be different
    assert expected["_aws"] == output["_aws"]


def test_namespace_env_var(monkeypatch, capsys, metric, dimension, namespace):
    monkeypatch.setenv("POWERTOOLS_METRICS_NAMESPACE", namespace["name"])

    with single_metric(**metric) as my_metrics:
        my_metrics.add_dimension(**dimension)
        monkeypatch.delenv("POWERTOOLS_METRICS_NAMESPACE")

    output = json.loads(capsys.readouterr().out.strip())
    expected = serialize_single_metric(metric=metric, dimension=dimension, namespace=namespace)

    remove_timestamp(metrics=[output, expected])  # Timestamp will always be different
    assert expected["_aws"] == output["_aws"]


def test_metrics_spillover(capsys, metric, dimension, namespace, a_hundred_metrics):
    my_metrics = Metrics()
    my_metrics.add_namespace(**namespace)
    my_metrics.add_dimension(**dimension)

    for _metric in a_hundred_metrics:
        my_metrics.add_metric(**_metric)

    @my_metrics.log_metrics(call_function=True)
    def lambda_handler(evt, handler):
        my_metrics.add_metric(**metric)
        return True

    lambda_handler({}, {})

    output = capsys.readouterr().out.strip()
    spillover_metrics, single_metric = output.split("\n")
    spillover_metrics = json.loads(spillover_metrics)
    single_metric = json.loads(single_metric)

    expected_single_metric = serialize_single_metric(
        metric=metric, dimension=dimension, namespace=namespace
    )
    expected_spillover_metrics = serialize_metrics(
        metrics=a_hundred_metrics, dimensions=[dimension], namespace=namespace
    )

    remove_timestamp(
        metrics=[
            spillover_metrics,
            expected_spillover_metrics,
            single_metric,
            expected_single_metric,
        ]
    )

    assert single_metric["_aws"] == expected_single_metric["_aws"]
    assert spillover_metrics["_aws"] == expected_spillover_metrics["_aws"]


def test_log_metrics_schema_error(metrics, dimensions, namespace):
    # It should error out because by default log_metrics doesn't invoke a function
    # so when decorator runs it'll raise an error while trying to serialize metrics
    my_metrics = Metrics()

    @my_metrics.log_metrics
    def lambda_handler(evt, handler):
        my_metrics.add_namespace(namespace)
        for metric in metrics:
            my_metrics.add_metric(**metric)
        for dimension in dimensions:
            my_metrics.add_dimension(**dimension)
            return True

    with pytest.raises(SchemaValidationError):
        lambda_handler({}, {})


def test_incorrect_metric_unit(metric, dimension, namespace):
    metric["unit"] = "incorrect_unit"

    with pytest.raises(MetricUnitError):
        with single_metric(**metric) as m:
            m.add_dimension(**dimension)
            m.add_namespace(**namespace)


def test_schema_no_namespace(metric, dimension):
    with pytest.raises(SchemaValidationError):
        with single_metric(**metric) as m:
            m.add_dimension(**dimension)


def test_schema_incorrect_value(metric, dimension, namespace):
    metric["value"] = "some_value"
    with pytest.raises(MetricValueError):
        with single_metric(**metric) as m:
            m.add_dimension(**dimension)
            m.add_namespace(**namespace)


def test_schema_no_metrics(dimensions, namespace):
    my_metrics = Metrics()
    my_metrics.add_namespace(**namespace)
    for dimension in dimensions:
        my_metrics.add_dimension(**dimension)
    with pytest.raises(SchemaValidationError):
        my_metrics.serialize_metric_set()


def test_exceed_number_of_dimensions(metric, namespace):
    dimensions = []
    for i in range(11):
        dimensions.append({"name": f"test_{i}", "value": "test"})

    with pytest.raises(SchemaValidationError):
        with single_metric(**metric) as m:
            m.add_namespace(**namespace)
            for dimension in dimensions:
                m.add_dimension(**dimension)

Features

• Create up to 100 metrics using a single EMF object (large JSON blob)
• Create an one off metric with a different dimension (e.g. Cold Start metric using function_version while all other metrics use service_name as a dimension)
• Validate against common mistakes that CloudWatch would otherwise fail silently
• No stack, custom resource, data collection needed — Metrics are created async by CloudWatch EMF

Multiple metrics

Used alongside other middlewares that call the decorated function

from lambda_python_powertools.metrics import Metrics, MetricUnit

metrics = Metrics()
metrics.add_namespace(name="ServerlessAirline")
metrics.add_metric(name="ColdStart", unit="Count", value=1)
metrics.add_dimension(name="service", value="booking")

@tracer.capture_lambda_handler
@metrics.log_metrics
def testing():
    metrics.add_metric(name="BookingConfirmation", unit="Count", value=1)
    some_code()
    return True

def some_code():
    metrics.add_metric(name="some_other_metric", unit=MetricUnit.Seconds, value=1)
    ...

UX for multiple metrics when no other middleware is being used

...

@metric.log_metrics(call_function=True)
def testing():
    some_code()
    return True

Single metric feature

Context manager with parameters using similar DX as multiple metrics

with single_metric(name="ColdStart", unit=MetricUnit.Count, value=1) as metric:
    metric.add_dimension(name="function_context", value="$LATEST")