import sys
import logging
import colorlog
from bs4 import BeautifulSoup, MarkupResemblesLocatorWarning
from firexkit.result import ChainInterruptedException
import warnings
# BeautifulSoup thinks we're giving it an URL because there is an URL in msg.
# Not good. Keep stderr clean by ignoring this warning.
warnings.filterwarnings("ignore", category=UserWarning, module='bs4')
warnings.filterwarnings("ignore", category=MarkupResemblesLocatorWarning)
console_stdout = None
console_stderr = None
[docs]
class RequeueingUndeliverableFilter(logging.Filter):
[docs]
def filter(self, record):
return 'Requeuing undeliverable message for queue' not in record.getMessage()
[docs]
class DistlibWarningsFilter(logging.Filter):
[docs]
def filter(self, record):
pathname = record.pathname
return not pathname.endswith('distlib/metadata.py') and not pathname.endswith('distlib/database.py')
[docs]
class RetryFilter(logging.Filter):
[docs]
def filter(self, record):
return 'Retry in' not in record.getMessage()
[docs]
class ChainInterruptedExceptionFilter(logging.Filter):
[docs]
def filter(self, record):
return ChainInterruptedException.__name__ not in record.getMessage()
[docs]
def setup_console_logging(module=None,
stdout_logging_level=logging.INFO,
console_logging_formatter='%(green)s[%(asctime)s]%(reset)s[%(hostname)s] %(log_color)s%(message)s',
console_datefmt="%H:%M:%S",
stderr_logging_level=logging.ERROR,
module_logger_logging_level=None):
formatter = FireXColoredConsoleFormatter(fmt=console_logging_formatter,
datefmt=console_datefmt,
log_colors={'DEBUG': 'cyan',
'INFO': 'bold',
'WARNING': 'yellow',
'ERROR': 'bold_red',
'CRITICAL': 'red,bg_white'})
class LogLevelFilter(logging.Filter):
"""Filters (lets through) all messages with level < LEVEL"""
def __init__(self, level):
self.level = level
super(LogLevelFilter, self).__init__()
def filter(self, record):
# "<" instead of "<=": since logger.setLevel is inclusive, this should
# be exclusive
return record.levelno < self.level
if module == "__main__":
# For program entry point, use root logger
module_logger = logging.getLogger()
if module_logger_logging_level is None:
module_logger_logging_level = logging.DEBUG
module_logger.setLevel(module_logger_logging_level)
global console_stdout
global console_stderr
if not console_stdout:
# This setup hasn't been done before
from firexapp.engine.logging import add_hostname_to_log_records
add_hostname_to_log_records()
console_stdout = logging.StreamHandler(sys.stdout)
console_stdout.setLevel(stdout_logging_level)
console_stdout.setFormatter(formatter)
console_stdout.addFilter(LogLevelFilter(stderr_logging_level))
console_stdout.addFilter(DistlibWarningsFilter())
console_stdout.addFilter(RequeueingUndeliverableFilter())
module_logger.addHandler(console_stdout)
console_stderr = logging.StreamHandler()
console_stderr.setLevel(stderr_logging_level)
console_stderr.setFormatter(formatter)
console_stderr.addFilter(RetryFilter())
console_stderr.addFilter(ChainInterruptedExceptionFilter())
module_logger.addHandler(console_stderr)
else:
# For submodules, create sub-loggers
module_logger = logging.getLogger(module)
if module_logger_logging_level is None:
module_logger_logging_level = logging.NOTSET
module_logger.setLevel(module_logger_logging_level)
return module_logger
[docs]
def set_console_log_level(log_level):
global console_stdout
global console_stderr
console_stdout.setLevel(log_level)
if log_level == logging.CRITICAL:
console_stderr.setLevel(log_level)
[docs]
def add_filter_to_console(log_filter):
assert isinstance(log_filter, logging.Filter)
console_stdout.addFilter(log_filter)
set_console_log_level(logging.DEBUG)