forked from bn222/cluster-deployment-automation
-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathlogger.py
58 lines (43 loc) · 1.73 KB
/
logger.py
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
import logging
import os
import sys
import typing
from typing import Optional
from typing import TextIO
class ExtendedLogger(logging.Logger):
def __init__(self, logger: logging.Logger):
self._wrapped_logger = logger
def __getattribute__(self: 'ExtendedLogger', name: str) -> typing.Any:
# ExtendedLogger is-a logging.Logger, but it delegates most calls to
# the wrapped-logger (which is also a logging.Logger).
if name == 'error_and_exit':
return object.__getattribute__(self, name)
logger = object.__getattribute__(self, '_wrapped_logger')
return logger.__getattribute__(name)
def error_and_exit(self: 'ExtendedLogger', msg: str, *, exit_code: int = -1) -> typing.NoReturn:
self.error(msg)
sys.exit(exit_code)
def configure_logger(lvl: Optional[int] = None) -> ExtendedLogger:
logger = logging.getLogger("CDA")
if lvl is None:
lvl = logging.INFO
s = os.environ.get("CDA_LOG_LEVEL")
if s:
s = s.strip().upper()
if s in ("DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"):
lvl = typing.cast(int, getattr(logging, s))
logger.setLevel(lvl)
fmt = "%(asctime)s %(levelname)s [th:%(thread)s] (%(filename)s:%(lineno)d): %(message)s"
datefmt = "%Y-%m-%d %H:%M:%S"
formatter = logging.Formatter(fmt, datefmt)
handler = logging.StreamHandler()
handler.setLevel(lvl)
handler.setFormatter(formatter)
global prev_handler
if prev_handler is not None:
logger.removeHandler(prev_handler)
prev_handler = handler
logger.addHandler(handler)
return ExtendedLogger(logger)
prev_handler: Optional['logging.StreamHandler[TextIO]'] = None
logger = configure_logger()