summaryrefslogtreecommitdiffstats
path: root/tests/topotests/munet/mulog.py
blob: f840eae2d8007e74827388aed29214569cc58ec1 (plain)
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
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
# -*- coding: utf-8 eval: (blacken-mode 1) -*-
# SPDX-License-Identifier: GPL-2.0-or-later
#
# December 4 2022, Christian Hopps <chopps@labn.net>
#
# Copyright (c) 2022, LabN Consulting, L.L.C.
#
"""Utilities for logging in munet."""

import logging

from pathlib import Path


class MultiFileHandler(logging.FileHandler):
    """A logging handler that logs to new files based on the logger name.

    The MultiFileHandler operates as a FileHandler with additional functionality. In
    addition to logging to the specified logging file MultiFileHandler also creates new
    FileHandlers for child loggers based on a root logging name path.

    The ``root_path`` determines when to create a new FileHandler. For each received log
    record, ``root_path`` is removed from the logger name of the record if present, and
    the resulting channel path (if any) determines the directory for a new log file to
    also emit the record to. The new file path is constructed by starting with the
    directory ``filename`` resides in, then joining the path determined above after
    converting "." to "/" and finally by adding back the basename of ``filename``.

      record logger path => mutest.output.testingfoo
      root_path => mutest.output
      base filename => /tmp/mutest/mutest-exec.log
      new logfile => /tmp/mutest/testingfoo/mutest-exec.log

    All messages are also emitted to the common FileLogger for ``filename``.

    If a log record is from a logger that does not start with ``root_path`` no file is
    created and the normal emit occurs.

    Args:
        root_path: the logging path of the root level for this handler.
        new_handler_level: logging level for newly created handlers
        log_dir: the log directory to put log files in.
        filename: the base log file.
    """

    def __init__(self, root_path, filename=None, **kwargs):
        self.__root_path = root_path
        self.__basename = Path(filename).name
        if root_path[-1] != ".":
            self.__root_path += "."
        self.__root_pathlen = len(self.__root_path)
        self.__kwargs = kwargs
        self.__log_dir = Path(filename).absolute().parent
        self.__log_dir.mkdir(parents=True, exist_ok=True)
        self.__filenames = {}
        self.__added = set()

        if "new_handler_level" not in kwargs:
            self.__new_handler_level = logging.NOTSET
        else:
            new_handler_level = kwargs["new_handler_level"]
            del kwargs["new_handler_level"]
            self.__new_handler_level = new_handler_level

        super().__init__(filename=filename, **kwargs)

        if self.__new_handler_level is None:
            self.__new_handler_level = self.level

    def __log_filename(self, name):
        if name in self.__filenames:
            return self.__filenames[name]

        if not name.startswith(self.__root_path):
            newname = None
        else:
            newname = name[self.__root_pathlen :]
            newname = Path(newname.replace(".", "/"))
            newname = self.__log_dir.joinpath(newname)
            newname = newname.joinpath(self.__basename)
            self.__filenames[name] = newname

        self.__filenames[name] = newname
        return newname

    def emit(self, record):
        newname = self.__log_filename(record.name)
        if newname:
            if newname not in self.__added:
                self.__added.add(newname)
                h = logging.FileHandler(filename=newname, **self.__kwargs)
                h.setLevel(self.__new_handler_level)
                h.setFormatter(self.formatter)
                logging.getLogger(record.name).addHandler(h)
                h.emit(record)
        super().emit(record)


class ColorFormatter(logging.Formatter):
    """A formatter that adds color sequences based on level."""

    def __init__(self, fmt=None, datefmt=None, style="%", **kwargs):
        grey = "\x1b[90m"
        yellow = "\x1b[33m"
        red = "\x1b[31m"
        bold_red = "\x1b[31;1m"
        reset = "\x1b[0m"
        # basefmt = " ------| %(message)s "

        self.formatters = {
            logging.DEBUG: logging.Formatter(grey + fmt + reset),
            logging.INFO: logging.Formatter(grey + fmt + reset),
            logging.WARNING: logging.Formatter(yellow + fmt + reset),
            logging.ERROR: logging.Formatter(red + fmt + reset),
            logging.CRITICAL: logging.Formatter(bold_red + fmt + reset),
        }
        # Why are we even bothering?
        super().__init__(fmt, datefmt, style, **kwargs)

    def format(self, record):
        formatter = self.formatters.get(record.levelno)
        return formatter.format(record)