This repository has been archived on 2023-12-07. You can view files and clone it, but cannot push or open issues or pull requests.
drone-cleanup-agents/cleanupagents/Logging.py

199 lines
6.2 KiB
Python

#!/usr/bin/env python3
"""Global utility methods and classes."""
import logging
import logging.handlers
import os
import sys
import colorama
from pythonjsonlogger import jsonlogger
import cleanupagents.Utils
from cleanupagents.Utils import Singleton
from cleanupagents.Utils import to_bool
CONSOLE_FORMAT = "{}[%(levelname)s]{} %(message)s"
FILE_FORMAT = "%(asctime)s %(levelname)s %(message)s"
JSON_FORMAT = "(asctime) (levelname) (message)"
TIME_FORMAT = "%Y-%m-%d %H:%M:%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(object):
"""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 MultilineFormatter(logging.Formatter):
"""Logging Formatter to reset color after newline characters."""
def format(self, record): # noqa
record.msg = record.msg.replace("\n", "\n{}... ".format(colorama.Style.RESET_ALL))
return logging.Formatter.format(self, record)
class MultilineJsonFormatter(jsonlogger.JsonFormatter):
"""Logging Formatter to remove newline characters."""
def format(self, record): # noqa
record.msg = record.msg.replace("\n", " ")
return jsonlogger.JsonFormatter.format(self, record)
class Log:
def __init__(self, level=logging.WARN, name="cleanupagents", logfile="/var/log/drone-agents.log", json=False):
self.logger = logging.getLogger(name)
self.logger.setLevel(level)
self.logger.addHandler(self._get_error_handler(json=json))
self.logger.addHandler(self._get_warn_handler(json=json))
self.logger.addHandler(self._get_info_handler(json=json))
self.logger.addHandler(self._get_critical_handler(json=json))
self.logger.addHandler(self._get_debug_handler(json=json))
try:
self.logger.addHandler(self._get_file_handler(logfile=logfile, json=json))
except PermissionError as e:
if e.errno == 13:
self.logger.warning("Can't write to logfile\n{}: '{}'".format(e.strerror, e.filename))
pass
self.logger.propagate = False
def _get_file_handler(self, logfile, json=False):
handler = logging.handlers.RotatingFileHandler(logfile, maxBytes=512000, backupCount=5)
handler.setLevel(logging.DEBUG)
handler.setFormatter(logging.Formatter(FILE_FORMAT, TIME_FORMAT))
if json:
handler.setFormatter(MultilineJsonFormatter(JSON_FORMAT))
return handler
def _get_error_handler(self, json=False):
handler = logging.StreamHandler(sys.stderr)
handler.setLevel(logging.ERROR)
handler.addFilter(LogFilter(logging.ERROR))
handler.setFormatter(MultilineFormatter(
self.error(CONSOLE_FORMAT.format(colorama.Fore.RED, colorama.Style.RESET_ALL))))
if json:
handler.setFormatter(MultilineJsonFormatter(JSON_FORMAT))
return handler
def _get_warn_handler(self, json=False):
handler = logging.StreamHandler(sys.stdout)
handler.setLevel(logging.WARN)
handler.addFilter(LogFilter(logging.WARN))
handler.setFormatter(MultilineFormatter(
self.warn(CONSOLE_FORMAT.format(colorama.Fore.YELLOW, colorama.Style.RESET_ALL))))
if json:
handler.setFormatter(MultilineJsonFormatter(JSON_FORMAT))
return handler
def _get_info_handler(self, json=False):
handler = logging.StreamHandler(sys.stdout)
handler.setLevel(logging.INFO)
handler.addFilter(LogFilter(logging.INFO))
handler.setFormatter(MultilineFormatter(
self.info(CONSOLE_FORMAT.format(colorama.Fore.CYAN, colorama.Style.RESET_ALL))))
if json:
handler.setFormatter(MultilineJsonFormatter(JSON_FORMAT))
return handler
def _get_critical_handler(self, json=False):
handler = logging.StreamHandler(sys.stderr)
handler.setLevel(logging.CRITICAL)
handler.addFilter(LogFilter(logging.CRITICAL))
handler.setFormatter(MultilineFormatter(
self.critical(CONSOLE_FORMAT.format(colorama.Fore.RED, colorama.Style.RESET_ALL))))
if json:
handler.setFormatter(MultilineJsonFormatter(JSON_FORMAT))
return handler
def _get_debug_handler(self, json=False):
handler = logging.StreamHandler(sys.stderr)
handler.setLevel(logging.DEBUG)
handler.addFilter(LogFilter(logging.DEBUG))
handler.setFormatter(MultilineFormatter(
self.critical(CONSOLE_FORMAT.format(colorama.Fore.BLUE, colorama.Style.RESET_ALL))))
if json:
handler.setFormatter(MultilineJsonFormatter(JSON_FORMAT))
return handler
def set_level(self, s):
self.logger.setLevel(s)
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 warn(self, msg):
"""Format warn 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 "{}{}{}".format(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):
pass