loki-logger-handler/loki_logger_handler/formatters/loguru_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
94
import traceback
import sys

from loki_logger_handler.formatters import LogFormatter


class LoguruFormatter(LogFormatter):
    """
    A custom formatter for log records generated by Loguru, formatting the record into a structured dictionary.
    """
    def format(self, record):
        """
        Format a Loguru log record into a structured dictionary.

        Args:
            record (dict): The Loguru log record to format.

        Returns:
            (tuple): A tuple of dictionary representation of the log record and the extracted loki metadata
        """
        # Convert timestamp to a standard format across Python versions
        timestamp = record.get("time")
        if hasattr(timestamp, "timestamp"):
            # Python 3.x
            timestamp = timestamp.timestamp()
        else:
            # Python 2.7: Convert datetime to a Unix timestamp
            timestamp = (timestamp - timestamp.utcoffset()).total_seconds()

        formatted = {
            "message": record.get("message"),
            "timestamp": timestamp,
            "process": record.get("process").id,
            "thread": record.get("thread").id,
            "function": record.get("function"),
            "module": record.get("module"),
            "name": record.get("name"),
            "level": record.get("level").name.upper(),
        }

        loki_metadata = {}
        # Update with extra fields if available
        extra = record.get("extra", {})
        if isinstance(extra, dict):
            # Handle the nested "extra" key correctly
            if "extra" in extra and isinstance(extra["extra"], dict):
                formatted.update(extra["extra"])
            else:
                formatted.update(extra)

            loki_metadata = formatted.get("loki_metadata")
            if loki_metadata:
                if not isinstance(loki_metadata, dict):
                    loki_metadata = {}
                del formatted["loki_metadata"]

        # Check if the log level indicates an error (case-insensitive and can be partial)
        if formatted["level"].startswith("ER"):
            formatted["file"] = record.get("file").name
            formatted["path"] = record.get("file").path
            formatted["line"] = record.get("line")

            self.add_exception_details(record, formatted)

        return formatted, loki_metadata

    @staticmethod
    def add_exception_details(record, formatted):
        """
        Adds exception details to the formatted log record.

        Args:
            record (dict): The log record containing log information.
            formatted (dict): The dictionary to which the formatted exception details will be added.

        Notes:
            - If the log record contains an exception, this method extracts the exception type,
              value, and traceback, formats the traceback, and adds it to the 'stacktrace' key
              in the formatted dictionary.
            - Handles both Python 2.7 and Python 3.x versions for formatting exceptions.
        """
        if record.get("exception"):
            exc_type, exc_value, exc_traceback = record.get("exception")
            if sys.version_info[0] == 2:
                    # Python 2.7: Use the older method for formatting exceptions
                formatted_traceback = traceback.format_exception(
                        exc_type, exc_value, exc_traceback
                    )
            else:
                    # Python 3.x: This is the same
                formatted_traceback = traceback.format_exception(
                        exc_type, exc_value, exc_traceback
                    )
            formatted["stacktrace"] = "".join(formatted_traceback)