import copy
import functools
import itertools
import json
import logging
import os
import random
import sys
import warnings
from distutils.util import strtobool
from typing import Any, Callable, Dict, Union
from ..helper.models import MetricUnit, build_lambda_context_model, build_metric_unit_from_str
from .exceptions import InvalidLoggerSamplingRateError
logger = logging.getLogger(__name__)
is_cold_start = True
def json_formatter(unserialized_value: Any):
"""JSON custom serializer to cast unserialisable values to strings.
Example
-------
**Serialize unserialisable value to string**
class X: pass
value = {"x": X()}
json.dumps(value, default=json_formatter)
Parameters
----------
unserialized_value: Any
Python object unserializable by JSON
"""
return str(unserialized_value)
class JsonFormatter(logging.Formatter):
"""AWS Lambda Logging formatter.
Formats the log message as a JSON encoded string. If the message is a
dict it will be used directly. If the message can be parsed as JSON, then
the parse d value is used in the output record.
Originally taken from https://gitlab.com/hadrien/aws_lambda_logging/
"""
def __init__(self, **kwargs):
"""Return a JsonFormatter instance.
The `json_default` kwarg is used to specify a formatter for otherwise
unserialisable values. It must not throw. Defaults to a function that
coerces the value to a string.
Other kwargs are used to specify log field format strings.
"""
datefmt = kwargs.pop("datefmt", None)
super(JsonFormatter, self).__init__(datefmt=datefmt)
self.reserved_keys = ["timestamp", "level", "location"]
self.format_dict = {
"timestamp": "%(asctime)s",
"level": "%(levelname)s",
"location": "%(funcName)s:%(lineno)d",
}
self.format_dict.update(kwargs)
self.default_json_formatter = kwargs.pop("json_default", json_formatter)
def format(self, record): # noqa: A003
record_dict = record.__dict__.copy()
record_dict["asctime"] = self.formatTime(record, self.datefmt)
log_dict = {}
for key, value in self.format_dict.items():
if value and key in self.reserved_keys:
# converts default logging expr to its record value
# e.g. '%(asctime)s' to '2020-04-24 09:35:40,698'
log_dict[key] = value % record_dict
else:
log_dict[key] = value
if isinstance(record_dict["msg"], dict):
log_dict["message"] = record_dict["msg"]
else:
log_dict["message"] = record.getMessage()
# Attempt to decode the message as JSON, if so, merge it with the
# overall message for clarity.
try:
log_dict["message"] = json.loads(log_dict["message"])
except (json.decoder.JSONDecodeError, TypeError, ValueError):
pass
if record.exc_info:
# Cache the traceback text to avoid converting it multiple times
# (it's constant anyway)
# from logging.Formatter:format
if not record.exc_text:
record.exc_text = self.formatException(record.exc_info)
if record.exc_text:
log_dict["exception"] = record.exc_text
json_record = json.dumps(log_dict, default=self.default_json_formatter)
if hasattr(json_record, "decode"): # pragma: no cover
json_record = json_record.decode("utf-8")
return json_record
def logger_setup(
service: str = None, level: str = None, sampling_rate: float = 0.0, legacy: bool = False, **kwargs
) -> DeprecationWarning:
"""DEPRECATED
This will be removed when GA - Use `aws_lambda_powertools.logging.logger.Logger` instead
Example
-------
**Logger class - Same UX**
from aws_lambda_powertools import Logger
logger = Logger(service="payment") # same env var still applies
"""
raise DeprecationWarning("Use Logger instead - This method will be removed when GA")
def logger_inject_lambda_context(
lambda_handler: Callable[[Dict, Any], Any] = None, log_event: bool = False
) -> DeprecationWarning:
"""DEPRECATED
This will be removed when GA - Use `aws_lambda_powertools.logging.logger.Logger` instead
Example
-------
**Logger class - Same UX**
from aws_lambda_powertools import Logger
logger = Logger(service="payment") # same env var still applies
@logger.inject_lambda_context
def handler(evt, ctx):
pass
"""
raise DeprecationWarning("Use Logger instead - This method will be removed when GA")
def _is_cold_start() -> bool:
"""Verifies whether is cold start
Returns
-------
bool
cold start bool value
"""
cold_start = False
global is_cold_start
if is_cold_start:
cold_start = is_cold_start
is_cold_start = False
return cold_start
def log_metric(
name: str, namespace: str, unit: MetricUnit, value: float = 0, service: str = "service_undefined", **dimensions,
):
"""Logs a custom metric in a statsD-esque format to stdout.
**This will be removed when GA - Use `aws_lambda_powertools.metrics.metrics.Metrics` instead**
Creating Custom Metrics synchronously impact on performance/execution time.
Instead, log_metric prints a metric to CloudWatch Logs.
That allows us to pick them up asynchronously via another Lambda function and create them as a metric.
NOTE: It takes up to 9 dimensions by default, and Metric units are conveniently available via MetricUnit Enum.
If service is not passed as arg or via env var, "service_undefined" will be used as dimension instead.
**Output in CloudWatch Logs**: `MONITORING|<metric_value>|<metric_unit>|<metric_name>|<namespace>|<dimensions>`
Serverless Application Repository App that creates custom metric from this log output:
https://serverlessrepo.aws.amazon.com/applications/arn:aws:serverlessrepo:us-east-1:374852340823:applications~async-custom-metrics
Environment variables
---------------------
POWERTOOLS_SERVICE_NAME: str
service name
Parameters
----------
name : str
metric name, by default None
namespace : str
metric namespace (e.g. application name), by default None
unit : MetricUnit, by default MetricUnit.Count
metric unit enum value (e.g. MetricUnit.Seconds), by default None\n
API Info: https://docs.aws.amazon.com/AmazonCloudWatch/latest/APIReference/API_MetricDatum.html
value : float, optional
metric value, by default 0
service : str, optional
service name used as dimension, by default "service_undefined"
dimensions: dict, optional
keyword arguments as additional dimensions (e.g. `customer=customerId`)
Example
-------
**Log metric to count number of successful payments; define service via env var**
$ export POWERTOOLS_SERVICE_NAME="payment"
from aws_lambda_powertools.logging import MetricUnit, log_metric
log_metric(
name="SuccessfulPayments",
unit=MetricUnit.Count,
value=1,
namespace="DemoApp"
)
**Log metric to count number of successful payments per campaign & customer**
from aws_lambda_powertools.logging import MetricUnit, log_metric
log_metric(
name="SuccessfulPayments",
service="payment",
unit=MetricUnit.Count,
value=1,
namespace="DemoApp",
campaign=campaign_id,
customer=customer_id
)
"""
warnings.warn(message="This method will be removed in GA; use Metrics instead", category=DeprecationWarning)
logger.debug(f"Building new custom metric. Name: {name}, Unit: {unit}, Value: {value}, Dimensions: {dimensions}")
service = os.getenv("POWERTOOLS_SERVICE_NAME") or service
dimensions = __build_dimensions(**dimensions)
unit = build_metric_unit_from_str(unit)
metric = f"MONITORING|{value}|{unit.name}|{name}|{namespace}|service={service}"
if dimensions:
metric = f"MONITORING|{value}|{unit.name}|{name}|{namespace}|service={service},{dimensions}"
print(metric)
def __build_dimensions(**dimensions) -> str:
"""Builds correct format for custom metric dimensions from kwargs
Parameters
----------
dimensions: dict, optional
additional dimensions
Returns
-------
str
Dimensions in the form of "key=value,key2=value2"
"""
MAX_DIMENSIONS = 10
dimension = ""
# CloudWatch accepts a max of 10 dimensions per metric
# We include service name as a dimension
# so we take up to 9 values as additional dimensions
# before we convert everything to a string of key=value
dimensions_partition = dict(itertools.islice(dimensions.items(), MAX_DIMENSIONS))
dimensions_list = [dimension + "=" + value for dimension, value in dimensions_partition.items() if value]
dimension = ",".join(dimensions_list)
return dimension
class Logger(logging.Logger):
"""Creates and setups a logger to format statements in JSON.
Includes service name and any additional key=value into logs
It also accepts both service name or level explicitly via env vars
Environment variables
---------------------
POWERTOOLS_SERVICE_NAME : str
service name
LOG_LEVEL: str, int
logging level (e.g. INFO, DEBUG)
POWERTOOLS_LOGGER_SAMPLE_RATE: float
samping rate ranging from 0 to 1, 1 being 100% sampling
Parameters
----------
service : str, optional
service name to be appended in logs, by default "service_undefined"
level : str, optional
logging.level, by default "INFO"
sample_rate: float, optional
sample rate for debug calls within execution context defaults to 0.0
stream: sys.stdout, optional
valid output for a logging stream, by default sys.stdout
Example
-------
**Setups structured logging in JSON for Lambda functions with explicit service name**
>>> from aws_lambda_powertools import Logger
>>> logger = Logger(service="payment")
>>>
>>> def handler(event, context):
logger.info("Hello")
**Setups structured logging in JSON for Lambda functions using env vars**
$ export POWERTOOLS_SERVICE_NAME="payment"
$ export POWERTOOLS_LOGGER_SAMPLE_RATE=0.01 # 1% debug sampling
>>> from aws_lambda_powertools import Logger
>>> logger = Logger()
>>>
>>> def handler(event, context):
logger.info("Hello")
**Append payment_id to previously setup structured log logger**
>>> from aws_lambda_powertools import Logger
>>> logger = Logger(service="payment")
>>>
>>> def handler(event, context):
logger.structure_logs(append=True, payment_id=event["payment_id"])
logger.info("Hello")
Parameters
----------
logging : logging.Logger
Inherits Logger
service: str
name of the service to create the logger for, "service_undefined" by default
level: str, int
log level, INFO by default
sampling_rate: float
debug log sampling rate, 0.0 by default
stream: sys.stdout
log stream, stdout by default
Raises
------
InvalidLoggerSamplingRateError
When sampling rate provided is not a float
"""
def __init__(
self,
service: str = None,
level: Union[str, int] = None,
sampling_rate: float = None,
stream: sys.stdout = None,
**kwargs,
):
self.service = service or os.getenv("POWERTOOLS_SERVICE_NAME") or "service_undefined"
self.sampling_rate = sampling_rate or os.getenv("POWERTOOLS_LOGGER_SAMPLE_RATE") or 0.0
self.log_level = level or os.getenv("LOG_LEVEL") or logging.INFO
self.handler = logging.StreamHandler(stream) if stream is not None else logging.StreamHandler(sys.stdout)
self._default_log_keys = {"service": self.service, "sampling_rate": self.sampling_rate}
self.log_keys = copy.copy(self._default_log_keys)
super().__init__(name=self.service, level=self.log_level)
try:
if self.sampling_rate and random.random() <= float(self.sampling_rate):
logger.debug("Setting log level to Debug due to sampling rate")
self.log_level = logging.DEBUG
except ValueError:
raise InvalidLoggerSamplingRateError(
f"Expected a float value ranging 0 to 1, but received {self.sampling_rate} instead. Please review POWERTOOLS_LOGGER_SAMPLE_RATE environment variable." # noqa E501
)
self.setLevel(self.log_level)
self.structure_logs(**kwargs)
self.addHandler(self.handler)
def inject_lambda_context(self, lambda_handler: Callable[[Dict, Any], Any] = None, log_event: bool = False):
"""Decorator to capture Lambda contextual info and inject into struct logging
Parameters
----------
log_event : bool, optional
Instructs logger to log Lambda Event, by default False
Environment variables
---------------------
POWERTOOLS_LOGGER_LOG_EVENT : str
instruct logger to log Lambda Event (e.g. `"true", "True", "TRUE"`)
Example
-------
**Captures Lambda contextual runtime info (e.g memory, arn, req_id)**
from aws_lambda_powertools import Logger
logger = Logger(service="payment")
@logger.inject_lambda_context
def handler(event, context):
logger.info("Hello")
**Captures Lambda contextual runtime info and logs incoming request**
from aws_lambda_powertools import Logger
logger = Logger(service="payment")
@logger.inject_lambda_context(log_event=True)
def handler(event, context):
logger.info("Hello")
Returns
-------
decorate : Callable
Decorated lambda handler
"""
# If handler is None we've been called with parameters
# Return a partial function with args filled
if lambda_handler is None:
logger.debug("Decorator called with parameters")
return functools.partial(self.inject_lambda_context, log_event=log_event)
log_event_env_option = str(os.getenv("POWERTOOLS_LOGGER_LOG_EVENT", "false"))
log_event = strtobool(log_event_env_option) or log_event
@functools.wraps(lambda_handler)
def decorate(event, context):
if log_event:
logger.debug("Event received")
self.info(event)
lambda_context = build_lambda_context_model(context)
cold_start = _is_cold_start()
self.structure_logs(cold_start=cold_start, **lambda_context.__dict__)
return lambda_handler(event, context)
return decorate
def structure_logs(self, append: bool = False, **kwargs):
"""Sets logging formatting to JSON.
Optionally, it can append keyword arguments
to an existing logger so it is available
across future log statements.
Last keyword argument and value wins if duplicated.
Parameters
----------
append : bool, optional
[description], by default False
"""
self.handler.setFormatter(JsonFormatter(**self._default_log_keys, **kwargs))
if append:
new_keys = {**self.log_keys, **kwargs}
self.handler.setFormatter(JsonFormatter(**new_keys))
self.log_keys.update(**kwargs)
def set_package_logger(
level: Union[str, int] = logging.DEBUG, stream: sys.stdout = None, formatter: logging.Formatter = None
):
"""Set an additional stream handler, formatter, and log level for aws_lambda_powertools package logger.
**Package log by default is supressed (NullHandler), this should only used for debugging.
This is separate from application Logger class utility**
Example
-------
**Enables debug logging for AWS Lambda Powertools package**
>>> from aws_lambda_powertools.logging.logger import set_package_logger
>>> set_package_logger()
Parameters
----------
level: str, int
log level, DEBUG by default
stream: sys.stdout
log stream, stdout by default
formatter: logging.Formatter
log formatter, "%(asctime)s %(name)s [%(levelname)s] %(message)s" by default
"""
if formatter is None:
formatter = logging.Formatter("%(asctime)s %(name)s [%(levelname)s] %(message)s")
if stream is None:
stream = sys.stdout
logger = logging.getLogger("aws_lambda_powertools")
logger.setLevel(level)
handler = logging.StreamHandler(stream)
handler.setFormatter(formatter)
logger.addHandler(handler)