Source code for lab_utils.custom_logging
""" Implements a custom logging service. The logging setup
is based upon the standard Python :obj:`logging` library
and offers some advantages:
- Standard logging across multiple apps using this module.
- Ease of use and configuration, with only necessary options.
- Extra functionality:
- E-mail notification over TLS.
- Coloured output to a terminal
- Improved file rotation naming.
- Logging over TCP socket compatible with the
:obj:`~lab_utils.socket_comm` module.
Attributes
---------
SUCCESS: int
New log level (25) intended to report successful
events to Slack, between INFO (20) and WARNING (30).
"""
# Imports
import logging
from logging import handlers
from queue import Queue
from sys import stdout
from atexit import register
from typing import Optional, TextIO, Tuple, List
from pkg_resources import resource_filename
import configparser
from os import makedirs
from os.path import abspath, expanduser, sep
from time import strftime
from datetime import datetime
import errno
import smtplib
from email.message import EmailMessage
# Third party
from slacker_log_handler import SlackerLogHandler, NoStacktraceFormatter
from pythonjsonlogger.jsonlogger import JsonFormatter
# New logging level: success, between WARNING and INFO
# It is meant to be used to report non-error events to Slack
SUCCESS = 25
[docs]def string_to_bool(s: str) -> bool:
""" Parses a variety of strings (e.g. 'true' or '1') to a boolean.
Parameters
----------
s: str
The string to parse
Returns
-------
bool:
True if :paramref:`s` is one of:
- 'true'
- '1'
- 't'
- 'y'
- 'yes'
- 'on'
"""
return s.lower() in ['true', '1', 't', 'y', 'yes', 'on']
[docs]class ColoredFormatter(logging.Formatter):
""" Console :class:`formatter<logging.Formatter>`
that prepends the coloured severity level of the
message. Based upon this
`gist <https://gist.github.com/hit9/5635505>`_.
"""
colours = {
'black': 30,
'red': 31,
'green': 32,
'yellow': 33,
'blue': 34,
'magenta': 35,
'cyan': 36,
'white': 37,
'bgred': 41,
'bggrey': 100,
} #: Terminal colour codes.
colour_map = {
'INFO': 'cyan',
'WARNING': 'yellow',
'ERROR': 'red',
'CRITICAL': 'bgred',
'EXCEPTION': 'bgred',
'DEBUG': 'bggrey',
'SUCCESS': 'green'
} #: Colour mapping.
prefix = '\033[' #: Prefix to modify terminal output colour.
suffix = '\033[0m' #: Suffix to modify terminal output colour.
[docs] def format(self, record: logging.LogRecord) -> str:
""" Prepends a fixed-length, coloured trailer with the
log level.
Parameters
----------
record: :class:`~logging.LogRecord`
The record to be logged.
Returns
-------
str:
The formatted message
"""
# Get the plain message from the parent formatter
message = super(ColoredFormatter, self).format(record)
# Get the colour code
colour = self.colour_map.get(record.levelname, 'white')
if colour not in self.colours:
colour = 'white'
colour_code = self.colours[colour]
# Build coloured header
coloured_header = '%s%dm%s%s' % (
self.prefix,
colour_code,
record.levelname,
self.suffix,
)
# Return formatted message, properly padded
return coloured_header + ''.ljust(9-len(record.levelname), ' ') + message
[docs]class CustomTimedRotatingFileHandler(logging.handlers.TimedRotatingFileHandler):
""" Variation of :class:`~logging.handlers.TimedRotatingFileHandler`. The
handler will produce daily log files to a given directory, appending the
date to a given base name."""
# Attributes
path: str = None #: Parent directory to save all logs.
basename: str = None #: Complete file base name where date will be appended, without extension.
extension: str = None #: Log file extension.
# Inherited attributes
stream: TextIO = None #: File stream
rolloverAt = None #: Next rollover time
[docs] def __init__(self, path: str, basename: str, extension: str = '.log'):
""" Calls the parent constructor and creates the logging
directory, if it does not exist.
Parameters
----------
path: str
Parent directory to save all logs.
basename: str
File base name where date will be appended, without extension.
extension: str, optional
Log file extension, default is 'log'.
Raises
------
:class:`OSError`:
The logging directory could not be created. The handler should
not be used if this exception is raised.
"""
# Assign attributes
self.path = abspath(expanduser(path))
self.basename = self.path + sep + basename
self.extension = extension
# Create logging directory if it does not exist
try:
makedirs(self.path)
except OSError as e:
if e.errno != errno.EEXIST:
logging.error("{}: {}".format(type(e).__name__, e))
raise
super(CustomTimedRotatingFileHandler, self).__init__(
filename=self.basename + '_' + strftime("%Y-%m-%d") + self.extension,
when='midnight',
delay=True,
utc=False,
)
[docs] def doRollover(self):
"""
Rotates log files on daily basis. The file name of the current
logfile is :attr:`basename` + :func:`~time.strftime`
+ :attr:`extension`, with time format '%Y-%m-%d'.
"""
# CLose current file
if self.stream:
self.stream.close()
# get the time that this sequence started at and make it a TimeTuple
self.stream = open(self.basename + strftime("%Y-%m-%d") + self.extension, 'a')
# Next rollover
self.rolloverAt = self.computeRollover(datetime.now()) # noqa
[docs]class TlsSMTPHandler(logging.handlers.SMTPHandler):
""" :obj:`~logging.handlers.SMTPHandler`
with TLS support. Based upon this
`gist <http://mynthon.net/howto/-/python/python%20-%20logging.SMTPHandler-how-to-use-gmail-smtp-server.txt>`_.
"""
# Inherited attributes, defined to avoid PyCharm warnings
mailport: int = None #: Mail port (inherited).
mailhost: str = None #: Mail provider (inherited).
fromaddr: str = None #: Sender address (inherited).
toaddrs: List[str] = None #: Recipients addresses (inherited).
username: str = None #: Login username (inherited).
password: str = None #: Login password (inherited).
[docs] def emit(self, record):
"""
Emits a record. Opens a TLS SMTP connection using the
:obj:`smtplib` library and sends an
:class:`~email.message.EmailMessage`.
"""
# Open connection
port = self.mailport
if not port:
port = smtplib.SMTP_PORT
smtp = smtplib.SMTP(self.mailhost, port)
if self.username:
smtp.ehlo() # for tls add this line
smtp.starttls() # for tls add this line
smtp.ehlo() # for tls add this line
smtp.login(self.username, self.password)
# Build email message
msg = EmailMessage()
msg.set_content(self.format(record))
msg['Subject'] = self.getSubject(record)
msg['To'] = ', '.join(self.toaddrs)
msg['From'] = self.fromaddr
smtp.send_message(msg)
smtp.quit()
[docs]class NonPickledSocketHandler(logging.handlers.SocketHandler):
""" Socket handler that sends non-pickled strings. Such strings
are compatible with a :class:`~lab_utils.socket_comm.Server`
listening on the appropriate TCP port."""
[docs] def emit(self, record: logging.LogRecord):
"""
Encodes a :paramref:`~NonPickledSocketHandler.emit.record`
and writes it to the socket. If there is an error with the
socket, silently drops the packet. If there was a problem
with the socket, re-establishes the socket.
Parameters
----------
record: :class:`~logging.LogRecord`
The record to be emitted.
"""
try:
msg = self.format(record)
self.send(msg.encode())
except BaseException:
self.handleError(record)
[docs]class CustomLogger(logging.Logger):
""" Custom logging class based on the default Python
:class:`~logging.Logger`. It introduces the new
logging level :attr:`.SUCCESS` = 25, meant to be used
to notify Slack about important, non-error events."""
[docs] def __init__(self, name, level=logging.NOTSET):
""" Calls the parent constructor and adds
the :attr:`.SUCCESS` = 25 logging level."""
super(CustomLogger, self).__init__(name, level)
[docs] def success(self, message, *args, **kws):
""" Creates a log entry with level :attr:`.SUCCESS`, similar
to the standard :meth:`~logging.Logger.error` and
:meth:`~logging.Logger.info`. """
# Yes, logger takes its '*args' as 'args'.
self._log(SUCCESS, message, args, **kws)
[docs]def getLogger(name: Optional[str] = None) -> CustomLogger: # noqa
""" Overrides the Python standard :func:`logging.getLogger`
to fix type completion hints, referring them to
:class:`CustomLogger` instead of :class:`~logging.Logger`.
Taken from a StackOverflow
`question <https://stackoverflow.com/questions/51400965/in-python-how-to-i-get-an-extended-class-element-to-show-up-in-autocomplete/51403572>`_.
Parameters
----------
name: str, optional
The logger name
Returns
-------
:class:`CustomLogger`
A named instance of the logger.
"""
_logger: CustomLogger = logging.getLogger(name) # noqa
return _logger
[docs]def configure_logging(
config_file: str = resource_filename(__name__, 'conf/logging.ini'),
fallback: bool = False,
):
""" Sets up the custom logger. Loads the configuration
from :paramref:`~configure_logging.config_file` using the
:obj:`configparser` library.
Parameters
----------
config_file : str, optional
Configuration file name, full path.
fallback: bool, optional
If 'True' and the logger setup fails, fall back to the default
:class:`~logging.Logger`.
Raises
------
:class:`configparser.Error`
Error while parsing the file, e.g. no file was found,
a parameter is missing or it has an invalid value.
"""
# Create a list of handlers to populate
handlers_list = []
# Set custom logging class
logging.setLoggerClass(CustomLogger)
# Create new logging level
logging.SUCCESS = SUCCESS # between WARNING and INFO
logging.addLevelName(SUCCESS, 'SUCCESS')
# Load the configuration file
try:
# Initialize config parser and read file
config_parser = configparser.ConfigParser()
config_parser.read(config_file)
# Console handler
if string_to_bool(config_parser.get(section='ConsoleLogger', option='active')):
console_handler = logging.StreamHandler(stream=stdout)
console_handler.setLevel(config_parser.get(section='ConsoleLogger', option='log_level'))
console_handler.setFormatter(ColoredFormatter(
fmt='%(asctime)s - %(name)-35s - %(funcName)-20s - %(message)s',
datefmt='%Y-%m-%d %H:%M:%S',
))
handlers_list.append(console_handler)
# File handler
if string_to_bool(config_parser.get(section='FileLogger', option='active')):
try:
file_handler = CustomTimedRotatingFileHandler(
path=config_parser.get(section='FileLogger', option='log_path'),
basename=config_parser.get(section='FileLogger', option='log_basename'),
extension=config_parser.get(section='FileLogger', option='log_extension'),
)
file_handler.setLevel(config_parser.get(section='FileLogger', option='log_level'))
file_handler.setFormatter(JsonFormatter(
fmt='%(asctime)s - %(levelname)s - %(name)s - %(funcName)s - %(message)s',
datefmt='%Y-%m-%d %H:%M:%S',
))
handlers_list.append(file_handler)
except BaseException as e:
print("{}: {}".format(type(e).__name__, e))
print('File logger disabled')
# Alarm handler
if string_to_bool(config_parser.get(section='AlarmLogger', option='active')):
# Get port
if config_parser.get(section='AlarmLogger', option='port') == 'default':
port = handlers.DEFAULT_TCP_LOGGING_PORT # 9020
else:
port = config_parser.getint(section='AlarmLogger', option='port')
alarm_handler = NonPickledSocketHandler(
host='localhost',
port=port
)
alarm_handler.setLevel(config_parser.get(section='AlarmLogger', option='log_level'))
alarm_handler.setFormatter(logging.Formatter(
fmt='{%(name)s}'
' --date {%(asctime)s}'
' --level {%(levelname)s}'
' --module {%(funcName)s}'
' --message {%(message)s}',
datefmt='%Y-%m-%d %H:%M:%S',
))
handlers_list.append(alarm_handler)
# SMTP handler for email notifications
if string_to_bool(config_parser.get(section='MailHandler', option='active')):
try:
# Read credentials from local file
with open(abspath(expanduser(config_parser.get(section='MailHandler', option='credentials_file'))))\
as f:
credentials: Tuple[str, str] = tuple(filter(None, f.read().rstrip().split('\n', 1))) # noqa
# Parse list of recipients
list_of_rec = config_parser.get(section='MailHandler', option='recipients')
recipients = list(filter(None, (x.strip() for x in list_of_rec.splitlines())))
mail_handler = TlsSMTPHandler(
mailhost=(
config_parser.get(section='MailHandler', option='mail_host'),
config_parser.getint(section='MailHandler', option='port')
),
fromaddr=config_parser.get(section='MailHandler', option='sender'),
toaddrs=recipients,
subject=config_parser.get(section='MailHandler', option='subject'),
credentials=credentials,
)
mail_handler.setLevel(config_parser.get(section='MailHandler', option='log_level'))
mail_handler.setFormatter(logging.Formatter(
fmt='%(asctime)s - %(levelname)-8s - %(name)-35s - %(funcName)-20s - %(message)s',
datefmt='%Y-%m-%d %H:%M:%S',
))
except BaseException as e:
print("{}: {}".format(type(e).__name__, e))
print('The SMTP handler will be disabled')
else:
handlers_list.append(mail_handler)
# Slack handler
if string_to_bool(config_parser.get(section='SlackLogger', option='active')):
try:
# Read API key from local file
with open(abspath(expanduser(config_parser.get(section='SlackLogger', option='api_file')))) as f:
slack_api = f.read().rstrip()
slack_handler = SlackerLogHandler(
api_key=slack_api,
channel=config_parser.get(section='SlackLogger', option='channel'),
username=config_parser.get(section='SlackLogger', option='user'),
fail_silent=False,
stack_trace=True,
)
slack_formatter = NoStacktraceFormatter(
fmt='%(asctime)s - %(message)s',
datefmt='%Y-%m-%d %H:%M:%S',
)
slack_handler.setFormatter(slack_formatter)
slack_handler.setLevel(config_parser.get(section='SlackLogger', option='log_level'))
handlers_list.append(slack_handler)
except BaseException as e:
print("{}: {}".format(type(e).__name__, e))
print('The Slack handler will be disabled')
# Set up logging queue to decouple the loggers and the main app activity
que = Queue(-1) # no limit on size
qh = logging.handlers.QueueHandler(que)
listener = logging.handlers.QueueListener(
que,
*handlers_list,
respect_handler_level=True,
)
# Add queue handler to the default logger
root_logger = logging.getLogger()
root_logger.setLevel(logging.DEBUG)
root_logger.addHandler(qh)
# Start the listener, and register it to stop automatically at exit
listener.start()
register(listener.stop)
except configparser.Error as e:
print("{}: {}".format(type(e).__name__, e))
if not fallback:
raise
except BaseException as e:
# Undefined exception, full traceback to be printed
print("{}: {}".format(type(e).__name__, e))
if not fallback:
raise
else:
logging.info('Logging system set up and running')
logging.debug('Active handlers:')
for handler in handlers_list:
logging.debug(' {}'.format(handler))
return
# Something went wrong
print('Default logger will be used')
logging.setLoggerClass(logging.Logger)
logging.basicConfig()