metrics.py

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

from aws_lambda_powertools.metrics.base import MetricManager

logger = logging.getLogger(__name__)


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.metrics import Metrics

        metrics = Metrics(namespace="ServerlessAirline", service="payment")
        metrics.add_metric(name="ColdStart", unit=MetricUnit.Count, value=1)
        metrics.add_metric(name="BookingConfirmation", unit="Count", value=1)
        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)

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

    Parameters
    ----------
    MetricManager : MetricManager
        Inherits from `aws_lambda_powertools.metrics.base.MetricManager`

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

    _metrics = {}
    _dimensions = {}

    def __init__(self, service: str = None, namespace: str = None):
        self.metric_set = self._metrics
        self.dimension_set = self._dimensions
        self.service = service or os.environ.get("POWERTOOLS_SERVICE_NAME")
        self.namespace = namespace
        if self.service:
            self.dimension_set["service"] = self.service
        super().__init__(metric_set=self.metric_set, dimension_set=self.dimension_set, namespace=self.namespace)

    def clear_metrics(self):
        logger.debug("Clearing out existing metric set from memory")
        self.metric_set.clear()
        self.dimension_set.clear()

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

        Be aware that the log_metrics **does call* the decorated function (e.g. lambda_handler).

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

            metrics = Metrics(service="payment")
            tracer = Tracer(service="payment")

            @tracer.capture_lambda_handler
            @metrics.log_metrics
                def handler(event, context)

        Parameters
        ----------
        lambda_handler : Callable[[Any, Any], Any], optional
            Lambda function handler, by default None

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

        @functools.wraps(lambda_handler)
        def decorate(*args, **kwargs):
            try:
                response = lambda_handler(*args, **kwargs)
            finally:
                metrics = self.serialize_metric_set()
                self.clear_metrics()
                logger.debug("Publishing metrics", {"metrics": metrics})
                print(json.dumps(metrics))

            return response

        return decorate