# Virus Health Check: a validation tool for HETDEX/VIRUS data
# Copyright (C) 2015, 2016, 2017, 2018 "The HETDEX collaboration"
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
"""Set up the VHC logging system.
Two file handler are created:
* ``log.txt``: logs everything with level above ``LOWER_LEVEL``
* ``v_results.txt``: logs everything with level above ``ERORR_LEVEL``. This
is intended only to notify about failed tests
The two handler are then added to the logger and a
:class:`logging.LoggerAdapter` instance is created passing it an empty
:class:`~libvhc.VCheck` instance, to avoid having to pass it to every logger
call. The :class:`logging.LoggerAdapter` instance is then saved using
similar name conventions as for :class:`logging.Logger`
The correct way to use it is to use the logger is:
* initialise the logger with :func:`set_logger`
* update the ``LoggerAdapter`` for before calling every driver function with
:func:`update_logger`
* where needed retrieve the "adapted" logger with :func:`getLogger` and log the
message and with a normal logger.
* alternatively you can get the standard logger and then log a message adding
the keyword option ``extra``
Examples
--------
>>> import os
>>> from libvhc import loggers
>>> from libvhc import VCheck
>>> # set up the logger
>>> # in the example as in the code, the name of the logger is set
>>> # to the ``path``.
>>> path = "tests/data/doctest"
>>> loggers.set_logger(path, os.path.join(path, loggers.LOG_FILE.format(0)),
... os.path.join(path, loggers.RESULT_FILE.format(0)))
>>> recipe = "flat"
>>> checks = ['flat:check1', 'flat:check2']
>>> # do the actual work
>>> log = loggers.getLogger(path)
>>> def check(vcheck, path, args):
... # do the check
... log.info("all fine")
... log.error("are you sure?")
>>> with loggers.logger_context(path):
... for c in checks:
... vcheck = VCheck(recipe, c)
... loggers.update_logger(vcheck, path)
... check(vcheck, path, [])
"""
from __future__ import (absolute_import, division, print_function,
unicode_literals)
import contextlib
import logging
import multiprocessing
import copy
import pyhetdex.tools.logging_helper as phlog
import pyhetdex.tools.queue as phqueue
import libvhc
LOWER_LEVEL = logging.INFO
"lower logging level: INFO"
ERORR_LEVEL = logging.ERROR
"error logging level: ERROR"
JSON_KEY = '+++VHCJSON+++ '
"Log message key to filter o"
LOG_FILE = 'log_{}.txt'
"Template for the log file name"
RESULT_FILE = 'v_results_{}.txt'
"Template for the result file name"
JSON_FILE = 'v_json_{}.txt'
"Template for the json file name"
manager = multiprocessing.Manager()
# local dictionary storing log adapters
_extras = manager.dict()
# local dictionary storing the queue listener instances
_listeners = {}
# queues used by the loggers
_queues = {}
# save the queue handler to be able to properly remove it
_queue_handler = {}
[docs]class JSONFilter(logging.Filter):
def __init__(self, print_json):
self.json = print_json
[docs] def filter(self, record):
msg = record.getMessage()
rmsg = '' # Remaining message
for line in msg.split('\n'):
if line.startswith(JSON_KEY):
if self.json:
rmsg += line.replace(JSON_KEY, '')+'\n'
else:
if not self.json:
rmsg += line+'\n'
return len(rmsg.strip()) # Return true if there is any message left
[docs]def _name(name):
"""Return the name to use (privately) with the module (private)
dictionaries.
Use the return string of this function instead of the name itself in parts
that have interaction with child processes, to allow for future
modifications.
Parameters
----------
name : string
Returns
-------
string
``name``
"""
return name
[docs]def set_logger(name, logfile, result_file, json_file, vcheck=libvhc.VCheck()):
"""Create and add the logging handlers to the logger ``name``.
Save a :class:`~logging.LoggerAdapter` created with the given
:class:`~libvhc.VCheck` instance.
Parameters
----------
name : string
name of the logger
logfile, result_file : string
name of the log and the result files
json_file : string
name of the optional json output file. If empty, do not create a
json log file
vcheck : instance of :class:`~libvhc.VCheck`, optional
store the recipe name and the check currently executing
"""
# create and save the queue
q = get_queue(name)
init_logger_adapt(name, q, extras=vcheck)
# log.txt handler
log_fmt = "[%(recipe)s, %(check)s] - [%(asctime)s] - %(levelname)s:"
log_fmt += " %(message)s"
logh = _make_file_handler(logfile, LOWER_LEVEL,
JSONFormatter(False, log_fmt),
filter_=JSONFilter(False))
# v_results.txt handler
results_fmt = "[%(recipe)s, %(check)s] %(levelname)s: %(message)s"
resulth = _make_file_handler(result_file, ERORR_LEVEL,
JSONFormatter(False, results_fmt),
filter_=JSONFilter(False))
handlers = [logh, resulth]
if json_file != '':
# v_json.txt handler
jsonh = _make_file_handler(json_file, logging.DEBUG,
JSONFormatter(True),
filter_=JSONFilter(True))
handlers.append(jsonh)
qlistener = phlog.SetupQueueListener(q, handlers=handlers)
_listeners[name] = qlistener
[docs]def update_logger(vcheck, name):
"""Update the :class:`~libvhc.VCheck` entry of the
:class:`~logging.LoggerAdapter` named ``name``
Parameters
----------
vcheck : instance of :class:`~libvhc.VCheck`
store the recipe name and the check currently executing
name : string
name of the logger
Raises
------
KeyError
if the logger called ``name`` has not been created using
:func:`set_logger`
"""
_extras[name] = vcheck.as_dict()
[docs]def getLogger(name, adapter=True):
"""Returns the logger named ``name``.
Parameters
----------
name : string
name of the logger
adapter : bool, optional
get the adapted logger, instead of the original one
Returns
-------
:class:`logging.LoggerAdapter` or :class:`logging.Logger` instance
Raises
------
`KeyError`
if the ``name`` logger is not found
"""
logger = logging.getLogger(name=_name(name))
if adapter:
return logging.LoggerAdapter(logger, _extras[_name(name)])
else:
return logger
[docs]def get_listener(name):
"""Return the queue listener
Parameters
----------
name : string
name of the logger
Returns
-------
:class:`~pyhetdex.tools.logging_helper.SetupQueueListener`
can be used once in a :keyword:`with`
Raises
------
`KeyError`
if the ``name`` listener is not found
"""
return _listeners[name]
[docs]def get_queue(name):
"""Returns the queue saved under the ``name``.
If it does not exist, create it first
Parameters
----------
path : string
path of the two above files
Returns
-------
queue-like instance
"""
try:
return _queues[name]
except KeyError:
q = phqueue.QueueContext()
_queues[name] = q
return q
[docs]@contextlib.contextmanager
def logger_context(name):
"""Yield the listener and the queue associated with ``name`` from within a
:keyword:`with` statement
Parameters
----------
name : string
name of the listener and the queue
Yields
------
queue-like instance
:class:`~pyhetdex.tools.logging_helper.SetupQueueListener`
"""
try:
with get_queue(name) as q, get_listener(name) as l:
yield q, l
finally:
clear_logger(name)
[docs]def init_logger_adapt(name, queue_, level=LOWER_LEVEL, extras=libvhc.VCheck()):
"""Initialise the logger and save the adapter passing ``extras`` to it
Parameters
----------
name : string
name of the logger
queue_ : queue-like instance
level : int, optional
minimum logging level
extras : dict-like instance
extras for the :class:`logging.LoggerAdapter`
"""
# create the logger
_init_logger(name, queue_, level=level)
# then save it
_extras[_name(name)] = extras.as_dict()
[docs]def clear_logger(name):
"""Remove the queue, the listener and reset the logger to pristine
situation.
Parameters
----------
name : string
name of the queue and logger
"""
_queues.pop(name, None)
listener = _listeners.pop(name, None)
qhandler = _queue_handler.pop(name, None)
log = logging.getLogger(_name(name))
if qhandler:
log.removeHandler(qhandler)
if listener:
for h in listener.handlers:
h.close()
[docs]def _init_logger(name, queue_, level=LOWER_LEVEL):
"""Initialise the logger, using the
:class:`~pyhetdex.tools.logging_helper.QueueHandler`
Parameters
----------
name : string
name of the logger
queue_ : queue-like instance
level : int, optional
minimum logging level
Returns
-------
logger : :class:`logging.Logger`
"""
# create the logger
logger = logging.getLogger(name=_name(name))
# set the lower logger lever
logger.setLevel(level)
# set the queue handler
qhandler = phlog.QueueHandler(queue_)
qhandler.setLevel(level)
# add the queue handler
logger.addHandler(qhandler)
_queue_handler[name] = qhandler
return logger
[docs]def _make_file_handler(fname, level, fmt, filter_=None):
"""Create the file handler
Parameters
----------
fname : string
name of the file where the logging happens
level : int
level of the handler
fmt : string or :class:`logging.Formatter`
formatting of the handler
filter_ : :class:`logging.Filter`
filter to apply to the handler
Returns
-------
fhandler : instance of :class:`logging.FileHandler`
configured handler
"""
fhandler = logging.FileHandler(fname)
fhandler.setLevel(level)
if isinstance(fmt, logging.Formatter):
fhandler.setFormatter(fmt)
else:
fhandler.setFormatter(logging.Formatter(fmt=fmt))
if filter_ is not None:
fhandler.addFilter(filter_)
return fhandler