execution-payloads-benchmarks/src/expb/logging/__init__.py at main · NethermindEth/execution-payloads-benchmarks · 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
import logging
import sys
import structlog


class Logger:
    def __init__(
        self,
        inner_logger: structlog.BoundLogger | None = None,
    ) -> None:
        self.inner_logger = inner_logger

    def info(self, *args, **kwargs):
        if self.inner_logger is None:
            return
        self.inner_logger.info(*args, **kwargs)

    async def ainfo(self, *args, **kwargs):
        if self.inner_logger is None:
            return
        await self.inner_logger.ainfo(*args, **kwargs)

    def error(self, *args, **kwargs):
        if self.inner_logger is None:
            return
        self.inner_logger.error(*args, **kwargs)

    async def aerror(self, *args, **kwargs):
        if self.inner_logger is None:
            return
        await self.inner_logger.aerror(*args, **kwargs)

    def debug(self, *args, **kwargs):
        if self.inner_logger is None:
            return
        self.inner_logger.debug(*args, **kwargs)

    async def adebug(self, *args, **kwargs):
        if self.inner_logger is None:
            return
        await self.inner_logger.adebug(*args, **kwargs)

    def warning(self, *args, **kwargs):
        if self.inner_logger is None:
            return
        self.inner_logger.warning(*args, **kwargs)

    async def awarning(self, *args, **kwargs):
        if self.inner_logger is None:
            return
        await self.inner_logger.awarning(*args, **kwargs)

    def critical(self, *args, **kwargs):
        if self.inner_logger is None:
            return
        self.inner_logger.critical(*args, **kwargs)

    async def acritical(self, *args, **kwargs):
        if self.inner_logger is None:
            return
        await self.inner_logger.acritical(*args, **kwargs)


def setup_logging(log_level: str = "INFO") -> Logger:
    log_level = log_level.upper()
    logging.basicConfig(level=log_level, stream=sys.stdout, format="%(message)s")

    structlog.configure(
        processors=[
            structlog.stdlib.add_log_level,
            # structlog.stdlib.add_logger_name,
            structlog.processors.TimeStamper(fmt="iso"),
            structlog.processors.StackInfoRenderer(),
            structlog.processors.format_exc_info,
            structlog.processors.LogfmtRenderer(
                drop_missing=True,
                key_order=["timestamp", "level", "event"],
            ),
        ],
        context_class=dict,
        logger_factory=structlog.stdlib.LoggerFactory(),
        wrapper_class=structlog.stdlib.BoundLogger,
        cache_logger_on_first_use=True,
    )

    log = structlog.get_logger()
    log.info("logging configured", level=log_level)

    return Logger(log)