From eab398c85baa0c9c75e0b969ced9bc6ba1c095c5 Mon Sep 17 00:00:00 2001 From: mploski Date: Wed, 5 Jan 2022 14:10:52 +0100 Subject: [PATCH] feat(logger): clone powertools logger config to any Python logger (#927) Co-authored-by: Michal Ploski Co-authored-by: Heitor Lessa --- aws_lambda_powertools/logging/utils.py | 81 +++++++++++ docs/core/logger.md | 18 +++ tests/functional/test_logger_utils.py | 193 +++++++++++++++++++++++++ 3 files changed, 292 insertions(+) create mode 100644 aws_lambda_powertools/logging/utils.py create mode 100644 tests/functional/test_logger_utils.py diff --git a/aws_lambda_powertools/logging/utils.py b/aws_lambda_powertools/logging/utils.py new file mode 100644 index 00000000000..f0e39ddf8f0 --- /dev/null +++ b/aws_lambda_powertools/logging/utils.py @@ -0,0 +1,81 @@ +import logging +from typing import Callable, List, Optional, Set, Union + +from .logger import Logger + + +def copy_config_to_registered_loggers( + source_logger: Logger, + log_level: Optional[str] = None, + exclude: Optional[Set[str]] = None, + include: Optional[Set[str]] = None, +) -> None: + + """Copies source Logger level and handler to all registered loggers for consistent formatting. + + Parameters + ---------- + source_logger : Logger + Powertools Logger to copy configuration from + log_level : str, optional + Logging level to set to registered loggers, by default uses source_logger logging level + include : Optional[Set[str]], optional + List of logger names to include, by default all registered loggers are included + exclude : Optional[Set[str]], optional + List of logger names to exclude, by default None + """ + + level = log_level or source_logger.level + + # Assumptions: Only take parent loggers not children (dot notation rule) + # Steps: + # 1. Default operation: Include all registered loggers + # 2. Only include set? Only add Loggers in the list and ignore all else + # 3. Include and exclude set? Add Logger if it’s in include and not in exclude + # 4. Only exclude set? Ignore Logger in the excluding list + + # Exclude source logger by default + if exclude: + exclude.add(source_logger.name) + else: + exclude = set(source_logger.name) + + # Prepare loggers set + if include: + loggers = include.difference(exclude) + filter_func = _include_registered_loggers_filter + else: + loggers = exclude + filter_func = _exclude_registered_loggers_filter + + registered_loggers = _find_registered_loggers(source_logger, loggers, filter_func) + for logger in registered_loggers: + _configure_logger(source_logger, logger, level) + + +def _include_registered_loggers_filter(loggers: Set[str]): + return [logging.getLogger(name) for name in logging.root.manager.loggerDict if "." not in name and name in loggers] + + +def _exclude_registered_loggers_filter(loggers: Set[str]) -> List[logging.Logger]: + return [ + logging.getLogger(name) for name in logging.root.manager.loggerDict if "." not in name and name not in loggers + ] + + +def _find_registered_loggers( + source_logger: Logger, loggers: Set[str], filter_func: Callable[[Set[str]], List[logging.Logger]] +) -> List[logging.Logger]: + """Filter root loggers based on provided parameters.""" + root_loggers = filter_func(loggers) + source_logger.debug(f"Filtered root loggers: {root_loggers}") + return root_loggers + + +def _configure_logger(source_logger: Logger, logger: logging.Logger, level: Union[int, str]) -> None: + logger.handlers = [] + logger.setLevel(level) + source_logger.debug(f"Logger {logger} reconfigured to use logging level {level}") + for source_handler in source_logger.handlers: + logger.addHandler(source_handler) + source_logger.debug(f"Logger {logger} reconfigured to use {source_handler}") diff --git a/docs/core/logger.md b/docs/core/logger.md index 6b3370be174..db2d12a2013 100644 --- a/docs/core/logger.md +++ b/docs/core/logger.md @@ -1066,6 +1066,24 @@ def handler(event: Dict, context: LambdaContext) -> List: return response.get("Buckets", []) ``` +**How can I enable powertools logging for imported libraries?** + +You can copy the Logger setup to all or sub-sets of registered external loggers. Use the `copy_config_to_registered_logger` method to do this. By default all registered loggers will be modified. You can change this behaviour by providing `include` and `exclude` attributes. You can also provide optional `log_level` attribute external loggers will be configured with. + + +```python hl_lines="10" title="Cloning Logger config to all other registered standard loggers" +import logging + +from aws_lambda_powertools import Logger +from aws_lambda_powertools.logging import utils + +logger = Logger() + +external_logger = logging.logger() + +utils.copy_config_to_registered_loggers(source_logger=logger) +external_logger.info("test message") + **What's the difference between `append_keys` and `extra`?** Keys added with `append_keys` will persist across multiple log messages while keys added via `extra` will only be available in a given log message operation. diff --git a/tests/functional/test_logger_utils.py b/tests/functional/test_logger_utils.py new file mode 100644 index 00000000000..1317fefc6ab --- /dev/null +++ b/tests/functional/test_logger_utils.py @@ -0,0 +1,193 @@ +import io +import json +import logging +import random +import string +from enum import Enum + +import pytest + +from aws_lambda_powertools import Logger +from aws_lambda_powertools.logging import formatter, utils + + +@pytest.fixture +def stdout(): + return io.StringIO() + + +@pytest.fixture +def log_level(): + class LogLevel(Enum): + NOTSET = 0 + INFO = 20 + WARNING = 30 + CRITICAL = 50 + + return LogLevel + + +@pytest.fixture +def logger(stdout, log_level): + def _logger(): + logging.basicConfig(stream=stdout, level=log_level.NOTSET.value) + logger = logging.getLogger(name=service_name()) + return logger + + return _logger + + +def capture_logging_output(stdout): + return json.loads(stdout.getvalue().strip()) + + +def capture_multiple_logging_statements_output(stdout): + return [json.loads(line.strip()) for line in stdout.getvalue().split("\n") if line] + + +def service_name(): + chars = string.ascii_letters + string.digits + return "".join(random.SystemRandom().choice(chars) for _ in range(15)) + + +def test_copy_config_to_ext_loggers(stdout, logger, log_level): + + msg = "test message" + + # GIVEN a external logger and powertools logger initialized + logger_1 = logger() + logger_2 = logger() + + powertools_logger = Logger(service=service_name(), level=log_level.INFO.value, stream=stdout) + + # WHEN configuration copied from powertools logger to ALL external loggers AND our external logger used + utils.copy_config_to_registered_loggers(source_logger=powertools_logger) + logger_1.info(msg) + logger_2.info(msg) + logs = capture_multiple_logging_statements_output(stdout) + + # THEN + for index, logger in enumerate([logger_1, logger_2]): + assert len(logger.handlers) == 1 + assert type(logger.handlers[0]) is logging.StreamHandler + assert type(logger.handlers[0].formatter) is formatter.LambdaPowertoolsFormatter + assert logger.level == log_level.INFO.value + assert logs[index]["message"] == msg + assert logs[index]["level"] == log_level.INFO.name + + +def test_copy_config_to_ext_loggers_include(stdout, logger, log_level): + + msg = "test message" + + # GIVEN a external logger and powertools logger initialized + logger = logger() + powertools_logger = Logger(service=service_name(), level=log_level.INFO.value, stream=stdout) + + # WHEN configuration copied from powertools logger to ALL external loggers AND our external logger used + utils.copy_config_to_registered_loggers(source_logger=powertools_logger, include={logger.name}) + logger.info(msg) + log = capture_logging_output(stdout) + + # THEN + assert len(logger.handlers) == 1 + assert type(logger.handlers[0]) is logging.StreamHandler + assert type(logger.handlers[0].formatter) is formatter.LambdaPowertoolsFormatter + assert logger.level == log_level.INFO.value + assert log["message"] == msg + assert log["level"] == log_level.INFO.name + + +def test_copy_config_to_ext_loggers_wrong_include(stdout, logger, log_level): + + # GIVEN a external logger and powertools logger initialized + logger = logger() + powertools_logger = Logger(service=service_name(), level=log_level.INFO.value, stream=stdout) + + # WHEN configuration copied from powertools logger to ALL external loggers AND our external logger used + utils.copy_config_to_registered_loggers(source_logger=powertools_logger, include={"non-existing-logger"}) + + # THEN + assert not logger.handlers + + +def test_copy_config_to_ext_loggers_exclude(stdout, logger, log_level): + + # GIVEN a external logger and powertools logger initialized + logger = logger() + powertools_logger = Logger(service=service_name(), level=log_level.INFO.value, stream=stdout) + + # WHEN configuration copied from powertools logger to ALL external loggers AND our external logger used + utils.copy_config_to_registered_loggers(source_logger=powertools_logger, exclude={logger.name}) + + # THEN + assert not logger.handlers + + +def test_copy_config_to_ext_loggers_include_exclude(stdout, logger, log_level): + + msg = "test message" + + # GIVEN a external logger and powertools logger initialized + logger_1 = logger() + logger_2 = logger() + + powertools_logger = Logger(service=service_name(), level=log_level.INFO.value, stream=stdout) + + # WHEN configuration copied from powertools logger to ALL external loggers AND our external logger used + utils.copy_config_to_registered_loggers( + source_logger=powertools_logger, include={logger_1.name, logger_2.name}, exclude={logger_1.name} + ) + logger_2.info(msg) + log = capture_logging_output(stdout) + + # THEN + assert not logger_1.handlers + assert len(logger_2.handlers) == 1 + assert type(logger_2.handlers[0]) is logging.StreamHandler + assert type(logger_2.handlers[0].formatter) is formatter.LambdaPowertoolsFormatter + assert logger_2.level == log_level.INFO.value + assert log["message"] == msg + assert log["level"] == log_level.INFO.name + + +def test_copy_config_to_ext_loggers_clean_old_handlers(stdout, logger, log_level): + + # GIVEN a external logger with handler and powertools logger initialized + logger = logger() + handler = logging.FileHandler("logfile") + logger.addHandler(handler) + powertools_logger = Logger(service=service_name(), level=log_level.INFO.value, stream=stdout) + + # WHEN configuration copied from powertools logger to ALL external loggers AND our external logger used + utils.copy_config_to_registered_loggers(source_logger=powertools_logger) + + # THEN + assert len(logger.handlers) == 1 + assert type(logger.handlers[0]) is logging.StreamHandler + assert type(logger.handlers[0].formatter) is formatter.LambdaPowertoolsFormatter + + +def test_copy_config_to_ext_loggers_custom_log_level(stdout, logger, log_level): + + msg = "test message" + + # GIVEN a external logger and powertools logger initialized + logger = logger() + powertools_logger = Logger(service=service_name(), level=log_level.CRITICAL.value, stream=stdout) + level = log_level.WARNING.name + + # WHEN configuration copied from powertools logger to ALL external loggers + # AND our external logger used with custom log_level + utils.copy_config_to_registered_loggers(source_logger=powertools_logger, include={logger.name}, log_level=level) + logger.warning(msg) + log = capture_logging_output(stdout) + + # THEN + assert len(logger.handlers) == 1 + assert type(logger.handlers[0]) is logging.StreamHandler + assert type(logger.handlers[0].formatter) is formatter.LambdaPowertoolsFormatter + assert powertools_logger.level == log_level.CRITICAL.value + assert logger.level == log_level.WARNING.value + assert log["message"] == msg + assert log["level"] == log_level.WARNING.name