mirror of
https://github.com/thegeeklab/prometheus-pve-sd.git
synced 2024-11-10 03:50:40 +00:00
793ec14b80
Co-authored-by: renovate[bot] <29139614+renovate[bot]@users.noreply.github.com> Co-authored-by: Robert Kaussow <mail@thegeeklab.de>
246 lines
7.5 KiB
Python
246 lines
7.5 KiB
Python
#!/usr/bin/env python3
|
|
"""Global utility methods and classes."""
|
|
|
|
import logging
|
|
import os
|
|
import sys
|
|
|
|
import colorama
|
|
from pythonjsonlogger import jsonlogger
|
|
|
|
from prometheuspvesd.utils import Singleton, to_bool
|
|
|
|
CONSOLE_FORMAT = "{}{}[%(levelname)s]{} %(message)s"
|
|
JSON_FORMAT = "%(asctime)s %(levelname)s %(message)s"
|
|
|
|
|
|
def _should_do_markup():
|
|
py_colors = os.environ.get("PY_COLORS", None)
|
|
if py_colors is not None:
|
|
return to_bool(py_colors)
|
|
|
|
return sys.stdout.isatty() and os.environ.get("TERM") != "dumb"
|
|
|
|
|
|
colorama.init(autoreset=True, strip=not _should_do_markup())
|
|
|
|
|
|
class LogFilter:
|
|
"""A custom log filter which excludes log messages above the logged level."""
|
|
|
|
def __init__(self, level):
|
|
"""
|
|
Initialize a new custom log filter.
|
|
|
|
:param level: Log level limit
|
|
:returns: None
|
|
|
|
"""
|
|
self.__level = level
|
|
|
|
def filter(self, logRecord): # noqa
|
|
# https://docs.python.org/3/library/logging.html#logrecord-attributes
|
|
return logRecord.levelno <= self.__level
|
|
|
|
|
|
class SimpleFormatter(logging.Formatter):
|
|
"""Logging Formatter for simple logs."""
|
|
|
|
def format(self, record):
|
|
return logging.Formatter.format(self, record)
|
|
|
|
|
|
class MultilineFormatter(logging.Formatter):
|
|
"""Logging Formatter to reset color after newline characters."""
|
|
|
|
def format(self, record):
|
|
record.msg = record.msg.replace("\n", f"\n{colorama.Style.RESET_ALL}... ")
|
|
return logging.Formatter.format(self, record)
|
|
|
|
|
|
class MultilineJsonFormatter(jsonlogger.JsonFormatter):
|
|
"""Logging Formatter to remove newline characters."""
|
|
|
|
def format(self, record):
|
|
record.msg = record.msg.replace("\n", " ")
|
|
return jsonlogger.JsonFormatter.format(self, record)
|
|
|
|
|
|
class Log:
|
|
"""Handle logging."""
|
|
|
|
def __init__(self, level=logging.WARNING, name="prometheuspvesd", log_format="console"):
|
|
self.logger = logging.getLogger(name)
|
|
self.logger.setLevel(level)
|
|
self.logger.addHandler(self._get_error_handler(log_format))
|
|
self.logger.addHandler(self._get_warning_handler(log_format))
|
|
self.logger.addHandler(self._get_info_handler(log_format))
|
|
self.logger.addHandler(self._get_critical_handler(log_format))
|
|
self.logger.addHandler(self._get_debug_handler(log_format))
|
|
self.logger.propagate = False
|
|
|
|
def _get_error_handler(self, log_format):
|
|
handler = logging.StreamHandler(sys.stderr)
|
|
handler.setLevel(logging.ERROR)
|
|
handler.addFilter(LogFilter(logging.ERROR))
|
|
|
|
if log_format == "json":
|
|
handler.setFormatter(MultilineJsonFormatter(JSON_FORMAT))
|
|
elif log_format == "simple":
|
|
handler.setFormatter(SimpleFormatter())
|
|
else:
|
|
handler.setFormatter(
|
|
MultilineFormatter(
|
|
self.error(
|
|
CONSOLE_FORMAT.format(
|
|
colorama.Fore.RED, colorama.Style.BRIGHT, colorama.Style.RESET_ALL
|
|
)
|
|
)
|
|
)
|
|
)
|
|
|
|
return handler
|
|
|
|
def _get_warning_handler(self, log_format):
|
|
handler = logging.StreamHandler(sys.stdout)
|
|
handler.setLevel(logging.WARNING)
|
|
handler.addFilter(LogFilter(logging.WARNING))
|
|
|
|
if log_format == "json":
|
|
handler.setFormatter(MultilineJsonFormatter(JSON_FORMAT))
|
|
elif log_format == "simple":
|
|
handler.setFormatter(SimpleFormatter())
|
|
else:
|
|
handler.setFormatter(
|
|
MultilineFormatter(
|
|
self.warning(
|
|
CONSOLE_FORMAT.format(
|
|
colorama.Fore.YELLOW, colorama.Style.BRIGHT, colorama.Style.RESET_ALL
|
|
)
|
|
)
|
|
)
|
|
)
|
|
|
|
return handler
|
|
|
|
def _get_info_handler(self, log_format):
|
|
handler = logging.StreamHandler(sys.stdout)
|
|
handler.setLevel(logging.INFO)
|
|
handler.addFilter(LogFilter(logging.INFO))
|
|
|
|
if log_format == "json":
|
|
handler.setFormatter(MultilineJsonFormatter(JSON_FORMAT))
|
|
elif log_format == "simple":
|
|
handler.setFormatter(SimpleFormatter())
|
|
else:
|
|
handler.setFormatter(
|
|
MultilineFormatter(
|
|
self.info(
|
|
CONSOLE_FORMAT.format(
|
|
colorama.Fore.CYAN, colorama.Style.BRIGHT, colorama.Style.RESET_ALL
|
|
)
|
|
)
|
|
)
|
|
)
|
|
|
|
return handler
|
|
|
|
def _get_critical_handler(self, log_format):
|
|
handler = logging.StreamHandler(sys.stderr)
|
|
handler.setLevel(logging.CRITICAL)
|
|
handler.addFilter(LogFilter(logging.CRITICAL))
|
|
|
|
if log_format == "json":
|
|
handler.setFormatter(MultilineJsonFormatter(JSON_FORMAT))
|
|
elif log_format == "simple":
|
|
handler.setFormatter(SimpleFormatter())
|
|
else:
|
|
handler.setFormatter(
|
|
MultilineFormatter(
|
|
self.critical(
|
|
CONSOLE_FORMAT.format(
|
|
colorama.Fore.RED, colorama.Style.BRIGHT, colorama.Style.RESET_ALL
|
|
)
|
|
)
|
|
)
|
|
)
|
|
|
|
return handler
|
|
|
|
def _get_debug_handler(self, log_format):
|
|
handler = logging.StreamHandler(sys.stderr)
|
|
handler.setLevel(logging.DEBUG)
|
|
handler.addFilter(LogFilter(logging.DEBUG))
|
|
|
|
if log_format == "json":
|
|
handler.setFormatter(MultilineJsonFormatter(JSON_FORMAT))
|
|
elif log_format == "simple":
|
|
handler.setFormatter(SimpleFormatter())
|
|
else:
|
|
handler.setFormatter(
|
|
MultilineFormatter(
|
|
self.critical(
|
|
CONSOLE_FORMAT.format(
|
|
colorama.Fore.BLUE, colorama.Style.BRIGHT, colorama.Style.RESET_ALL
|
|
)
|
|
)
|
|
)
|
|
)
|
|
|
|
return handler
|
|
|
|
def update_logger(self, level=None, log_level=None):
|
|
for handler in self.logger.handlers[:]:
|
|
self.logger.removeHandler(handler)
|
|
|
|
self.logger.setLevel(level)
|
|
self.logger.addHandler(self._get_error_handler(log_level))
|
|
self.logger.addHandler(self._get_warning_handler(log_level))
|
|
self.logger.addHandler(self._get_info_handler(log_level))
|
|
self.logger.addHandler(self._get_critical_handler(log_level))
|
|
self.logger.addHandler(self._get_debug_handler(log_level))
|
|
|
|
def debug(self, msg):
|
|
"""Format info messages and return string."""
|
|
return msg
|
|
|
|
def critical(self, msg):
|
|
"""Format critical messages and return string."""
|
|
return msg
|
|
|
|
def error(self, msg):
|
|
"""Format error messages and return string."""
|
|
return msg
|
|
|
|
def warning(self, msg):
|
|
"""Format warning messages and return string."""
|
|
return msg
|
|
|
|
def info(self, msg):
|
|
"""Format info messages and return string."""
|
|
return msg
|
|
|
|
def _color_text(self, color, msg):
|
|
"""
|
|
Colorize strings.
|
|
|
|
:param color: colorama color settings
|
|
:param msg: string to colorize
|
|
:returns: string
|
|
|
|
"""
|
|
return f"{color}{msg}{colorama.Style.RESET_ALL}"
|
|
|
|
def sysexit(self, code=1):
|
|
sys.exit(code)
|
|
|
|
def sysexit_with_message(self, msg, code=1):
|
|
self.logger.critical(str(msg))
|
|
self.sysexit(code)
|
|
|
|
|
|
class SingleLog(Log, metaclass=Singleton):
|
|
"""Singleton logging class."""
|
|
|
|
pass
|