loki-logger-handler/loki_logger_handler/formatters/logger_formatter.py at beed8d09cb1fafedf14a22cae9a838da72df1007 · xente/loki-logger-handler · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
import traceback

from loki_logger_handler.formatters import LogFormatter


class LoggerFormatter(LogFormatter):
    """
    A custom formatter for log records generated by the standard library logger, formatting the record into a structured dictionary.
    """
    LOG_RECORD_FIELDS = {
        "msg",
        "levelname",
        "msecs",
        "name",
        "pathname",
        "filename",
        "module",
        "lineno",
        "funcName",
        "created",
        "thread",
        "threadName",
        "process",
        "processName",
        "relativeCreated",
        "stack_info",
        "args",
        "exc_info",
        "levelno",
        "exc_text",
    }

    def format(self, record):
        """
        Format a log record into a structured dictionary.

        Args:
            record (logging.LogRecord): The log record to format.

        Returns:
            (tuple): A tuple of dictionary representation of the log record and the extracted loki metadata.
        """

        formatted = {
            "message": record.getMessage(),
            "timestamp": record.created,
            "process": record.process,
            "thread": record.thread,
            "function": record.funcName,
            "module": record.module,
            "name": record.name,
            "level": record.levelname,
        }

        # Capture any custom fields added to the log record
        custom_fields = {
            key: value for key, value in record.__dict__.items()
            if key not in self.LOG_RECORD_FIELDS
        }

        loki_metadata = {}

        for key in custom_fields:
            if "loki_metadata" == key:
                value = getattr(record, key)
                if isinstance(value, dict):
                    loki_metadata = value
            else:
                formatted[key] = getattr(record, key)

        # Check if the log level indicates an error (case-insensitive and can be partial)
        if record.levelname.upper().startswith("ER"):
            formatted["file"] = record.filename
            formatted["path"] = record.pathname
            formatted["line"] = record.lineno
            formatted["stacktrace"] = self._format_stacktrace(record.exc_info)

        return formatted, loki_metadata

    @staticmethod
    def _format_stacktrace(exc_info):
        """
        Format the stacktrace if exc_info is present.

        Args:
            exc_info (tuple or None): Exception info tuple as returned by sys.exc_info().

        Returns:
            str or None: Formatted stacktrace as a string, or None if exc_info is not provided.
        """
        if exc_info:
            return "".join(traceback.format_exception(*exc_info))
        return None