feat(logger): log uncaught exceptions via system's exception hook (aws-powertools#1727)

Author: heitorlessa

aws_lambda_powertools/logging/logger.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import functools
 import inspect
 import io
@@ -96,6 +98,11 @@ class Logger(logging.Logger):  # lgtm [py/missing-call-to-init]
         custom logging formatter that implements PowertoolsFormatter
     logger_handler: logging.Handler, optional
         custom logging handler e.g. logging.FileHandler("file.log")
+    log_uncaught_exceptions: bool, by default False
+        logs uncaught exception using sys.excepthook
+
+        See: https://docs.python.org/3/library/sys.html#sys.excepthook
+
 
     Parameters propagated to LambdaPowertoolsFormatter
     --------------------------------------------------
@@ -203,6 +210,7 @@ def __init__(
         stream: Optional[IO[str]] = None,
         logger_formatter: Optional[PowertoolsFormatter] = None,
         logger_handler: Optional[logging.Handler] = None,
+        log_uncaught_exceptions: bool = False,
         json_serializer: Optional[Callable[[Dict], str]] = None,
         json_deserializer: Optional[Callable[[Union[Dict, str, bool, int, float]], str]] = None,
         json_default: Optional[Callable[[Any], Any]] = None,
@@ -222,6 +230,8 @@ def __init__(
         self.child = child
         self.logger_formatter = logger_formatter
         self.logger_handler = logger_handler or logging.StreamHandler(stream)
+        self.log_uncaught_exceptions = log_uncaught_exceptions
+
         self.log_level = self._get_log_level(level)
         self._is_deduplication_disabled = resolve_truthy_env_var_choice(
             env=os.getenv(constants.LOGGER_LOG_DEDUPLICATION_ENV, "false")
@@ -244,6 +254,10 @@ def __init__(
 
         self._init_logger(formatter_options=formatter_options, **kwargs)
 
+        if self.log_uncaught_exceptions:
+            logger.debug("Replacing exception hook")
+            sys.excepthook = functools.partial(log_uncaught_exception_hook, logger=self)
+
     # Prevent __getattr__ from shielding unknown attribute errors in type checkers
     # https://github.com/awslabs/aws-lambda-powertools-python/issues/1660
     if not TYPE_CHECKING:
@@ -735,3 +749,8 @@ def _is_internal_frame(frame):  # pragma: no cover
     """Signal whether the frame is a CPython or logging module internal."""
     filename = os.path.normcase(frame.f_code.co_filename)
     return filename == logging._srcfile or ("importlib" in filename and "_bootstrap" in filename)
+
+
+def log_uncaught_exception_hook(exc_type, exc_value, exc_traceback, logger: Logger):
+    """Callback function for sys.excepthook to use Logger to log uncaught exceptions"""
+    logger.exception("Uncaught exception", exc_info=(exc_type, exc_value, exc_traceback))  # pragma: no cover

docs/core/logger.md

@@ -291,6 +291,30 @@ Use `logger.exception` method to log contextual information about exceptions. Lo
     --8<-- "examples/logger/src/logging_exceptions_output.json"
     ```
 
+#### Uncaught exceptions
+
+Logger can optionally log uncaught exceptions by setting `log_uncaught_exceptions=True` at initialization.
+
+!!! info "Logger will replace any exception hook previously registered via [sys.excepthook](https://docs.python.org/3/library/sys.html#sys.excepthook){target='_blank'}."
+
+??? question "What are uncaught exceptions?"
+
+    It's any raised exception that wasn't handled by the [`except` statement](https://docs.python.org/3.9/tutorial/errors.html#handling-exceptions){target="_blank"}, leading a Python program to a non-successful exit.
+
+    They are typically raised intentionally to signal a problem (`raise ValueError`), or a propagated exception from elsewhere in your code that you didn't handle it willingly or not (`KeyError`, `jsonDecoderError`, etc.).
+
+=== "logging_uncaught_exceptions.py"
+
+    ```python hl_lines="7"
+    --8<-- "examples/logger/src/logging_uncaught_exceptions.py"
+    ```
+
+=== "logging_uncaught_exceptions_output.json"
+
+    ```json hl_lines="7-8"
+    --8<-- "examples/logger/src/logging_uncaught_exceptions_output.json"
+    ```
+
 ### Date formatting
 
 Logger uses Python's standard logging date format with the addition of timezone: `2021-05-03 11:47:12,494+0200`.

examples/logger/src/logging_uncaught_exceptions.py

@@ -0,0 +1,16 @@
+import requests
+
+from aws_lambda_powertools import Logger
+from aws_lambda_powertools.utilities.typing import LambdaContext
+
+ENDPOINT = "http://httpbin.org/status/500"
+logger = Logger(log_uncaught_exceptions=True)
+
+
+def handler(event: dict, context: LambdaContext) -> str:
+    ret = requests.get(ENDPOINT)
+    # HTTP 4xx/5xx status will lead to requests.HTTPError
+    # Logger will log this exception before this program exits non-successfully
+    ret.raise_for_status()
+
+    return "hello world"

examples/logger/src/logging_uncaught_exceptions_output.json

@@ -0,0 +1,9 @@
+{
+    "level": "ERROR",
+    "location": "log_uncaught_exception_hook:756",
+    "message": "Uncaught exception",
+    "timestamp": "2022-11-16 13:51:29,198+0100",
+    "service": "payment",
+    "exception": "Traceback (most recent call last):\n  File \"<input>\", line 52, in <module>\n    handler({}, {})\n  File \"<input>\", line 17, in handler\n    ret.raise_for_status()\n  File \"<input>/lib/python3.9/site-packages/requests/models.py\", line 1021, in raise_for_status\n    raise HTTPError(http_error_msg, response=self)\nrequests.exceptions.HTTPError: 500 Server Error: INTERNAL SERVER ERROR for url: http://httpbin.org/status/500",
+    "exception_name": "HTTPError"
+}

tests/functional/test_logger.py

@@ -1,10 +1,12 @@
+import functools
 import inspect
 import io
 import json
 import logging
 import random
 import re
 import string
+import sys
 import warnings
 from ast import Dict
 from collections import namedtuple
@@ -892,3 +894,15 @@ def test_powertools_debug_env_var_warning(monkeypatch: pytest.MonkeyPatch):
         set_package_logger_handler()
         assert len(w) == 1
         assert str(w[0].message) == warning_message
+
+
+def test_logger_log_uncaught_exceptions(service_name, stdout):
+    # GIVEN an initialized Logger is set with log_uncaught_exceptions
+    logger = Logger(service=service_name, stream=stdout, log_uncaught_exceptions=True)
+
+    # WHEN Python's exception hook is inspected
+    exception_hook = sys.excepthook
+
+    # THEN it should contain our custom exception hook with a copy of our logger
+    assert isinstance(exception_hook, functools.partial)
+    assert exception_hook.keywords.get("logger") == logger