* optionally squelch repeated messages (:code:`--logging_squelch_repeats`),
* optionally log probalistically (:code:`--logging_probabilistically`),
* capture printed messages into the info log (:code:`--logging_captures_prints`),
- * and optionally clear unwanted logging handlers added by other imports
+ * optionally clear unwanted logging handlers added by other imports
before this one (:code:`--logging_clear_preexisting_handlers`).
+ * optionally append to system-wide records of non-zero exits
+ (:code:`--logging_non_zero_exits_record_path`) and unhandled exceptions
+ (:code:`--logging_unhandled_top_level_exceptions_record_path`) in
+ cooperation with :mod:`pyutils.bootstrap`.
+ * There are also :class:`LoggerAdapter` classes to implement prefix/suffix
+ functionality without using :class:`LoggingContext` by wrapping the
+ logger included.
To use this functionality, call :meth:`initialize_logging` early
in your program entry point. If you use the
import io
import logging
import os
+import re
import sys
+from logging import FileHandler
from logging.config import fileConfig
from logging.handlers import RotatingFileHandler, SysLogHandler
+from types import TracebackType
from typing import Any, Callable, Dict, Iterable, List, Optional, Union
import pytz
)
cfg.add_argument(
"--logging_filename_maxsize",
- type=int,
+ type=argparse_utils.valid_byte_count,
default=(1024 * 1024),
metavar="#BYTES",
help="The maximum size (in bytes) to write to the logging_filename.",
+ "cause you to miss logging messages."
),
)
+cfg.add_argument(
+ "--logging_non_zero_exits_record_path",
+ type=str,
+ default="/var/log/abnormal_python_exits.log",
+ metavar="FILENAME",
+ help="The filename in which to record non-zero exits.",
+)
+cfg.add_argument(
+ "--logging_unhandled_top_level_exceptions_record_path",
+ type=str,
+ default="/var/log/abnormal_python_exits.log",
+ metavar="FILENAME",
+ help="The filename in which to record unhandled top level exceptions.",
+)
BUILT_IN_PRINT = print
LOGGING_INITIALIZED = False
from pyutils import function_utils
identifier = function_utils.function_identifier(f)
+
+ # Get rid of module paths, e.g. pyutils.ansi:bg -> ansi:bg which
+ # is what we're going to need below.
+ identifier = re.sub(r"[^\.]+\.", "", identifier)
squelched_logging_counts[identifier] = squelch_after_n_repeats
return f
class PrefixAddingFilter(logging.Filter):
+ """A filter that adds a string prefix to the log record for the
+ formatter to later fill in. Requires a %(prefix)s in the format
+ string.
+ """
+
def __init__(self, prefix: str, klobber: bool = False):
+ """
+ Args:
+ prefix: the prefix string to add
+ klobber: should we overwrite other prefixes?
+ """
super().__init__()
if prefix:
self.prefix = prefix
self.prefix = ""
self.klobber = klobber
+ @overrides
def filter(self, record: logging.LogRecord):
if self.klobber:
record.prefix = self.prefix
class SuffixAddingFilter(logging.Filter):
+ """A filter that adds a string suffix to the log record for the
+ formatter to later fill in. Requires a %(suffix)s in the format
+ string.
+ """
+
def __init__(self, suffix: str, klobber: bool = False):
+ """
+ Args:
+ suffix: the suffix string to add
+ klobber: should we overwrite other suffixes?
+ """
super().__init__()
if suffix:
self.suffix = suffix
self.suffix = ""
self.klobber = klobber
+ @overrides
def filter(self, record: logging.LogRecord):
if self.klobber:
record.suffix = self.suffix
class PrependingLogAdapter(logging.LoggerAdapter):
+ """:class:`LoggingContext` adds prefixes and suffixes using a
+ logging.Filter that must insert "prefix" or "suffix" members into
+ each log record by using :class:`PrefixAddingFilter` and
+ :class:`SuffixAddingFilter`. This relies on the logging format
+ string containing a %(prefix)s and a %(suffix)s to work correctly.
+
+ This is an alternate approach that basically just wraps the logger
+ in a class that has the same interface and thunks most calls down
+ to the wrapped logger. It might be useful if you don't want to use
+ :class:`LoggingContext` or its friends.
+
+ >>> logger = logging.getLogger(__name__ + ".PrependingLogAdapter")
+ >>> logger.setLevel(logging.INFO)
+ >>> logger.addHandler(logging.StreamHandler(sys.stdout))
+
+ At this point logger doesn't have any format string and so it is
+ missing %(prefix)s and %(suffix)s. It also doesn't have a
+ :class:`PrefixAddingFilter` or :class:`SuffixAddingFilter` added.
+ So using it in a :class:`LoggingContext` will not work.
+
+ But we can still add a prefix or suffix by just wrapping it:
+
+ >>> logger.info("TEST")
+ TEST
+
+ >>> log = PrependingLogAdapter.wrap_logger('prefix> ', logger)
+ >>> log.info("TEST")
+ prefix> TEST
+ """
+
def process(self, msg, kwargs):
return f'{self.extra.get("prefix", "")}{msg}', kwargs
class AppendingLogAdapter(logging.LoggerAdapter):
+ """:class:`LoggingContext` adds prefixes and suffixes using a
+ logging.Filter that must insert "prefix" or "suffix" members into
+ each log record by using :class:`PrefixAddingFilter` and
+ :class:`SuffixAddingFilter`. This relies on the logging format
+ string containing a %(prefix)s and a %(suffix)s to work correctly.
+
+ This is an alternate approach that basically just wraps the logger
+ in a class that has the same interface and thunks most calls down
+ to the wrapped logger. It might be useful if you don't want to use
+ :class:`LoggingContext` or its friends.
+
+ >>> logger = logging.getLogger(__name__ + ".AppendingLogAdapter")
+ >>> logger.setLevel(logging.INFO)
+ >>> logger.addHandler(logging.StreamHandler(sys.stdout))
+
+ At this point logger doesn't have any format string and so it is
+ missing %(prefix)s and %(suffix)s. It also doesn't have a
+ :class:`PrefixAddingFilter` or :class:`SuffixAddingFilter` added.
+ So using it in a :class:`LoggingContext` will not work.
+
+ But we can still add a prefix or suffix by just wrapping it:
+
+ >>> logger.info("TEST")
+ TEST
+
+ >>> log = AppendingLogAdapter.wrap_logger('!!!', logger)
+ >>> log.info("TEST")
+ TEST!!!
+ """
+
def process(self, msg, kwargs):
return f'{msg}{self.extra.get("suffix", "")}', kwargs
about it.
>>> logging_format = "%(prefix)s%(message)s%(suffix)s"
- >>> logger = logging.getLogger(__name__)
+ >>> logger = logging.getLogger(__name__ + ".LoggingContext")
>>> logger.setLevel(logging.INFO)
>>> handler = logging.StreamHandler(sys.stdout)
>>> handler.setFormatter(
... logger.info("Working on it now")
10.0.0.13> Working on it now
- logger.info("Done with that request")
+ >>> logger.info("Done with that request")
Done with that request
LoggingContext can also be used to add temporary handler(s).
But inside, the decorator has changed us to DEBUG
>>> logger.debug("And, of course, out here we're still at INFO afterwards")
+ Args:
+ logger: the logger on which to operate
+ level: the new level to set for the duration of the context
+ handlers: additional handlers to add for the duration of the context
+ prefix: the prefix string to set for the duration of the context
+ suffix: the suffix string to set for the duration of the context
+
+ Returns:
+ The modified logger.
+
"""
self.logger = logger
self.level = level
wired in under :meth:`initialize_logging` so that the
timestamps in log messages have millisecond level
precision.
-
"""
converter = datetime.datetime.fromtimestamp # type: ignore
def _construct_logging_format() -> str:
+ """Figure out the root logging format string based on commandling args."""
+
# Custom or default --logging_format?
if config.config["logging_format"]:
fmt = config.config["logging_format"]
def _construct_handlers_list(
logging_format: str, facility_name: Optional[str]
) -> List[logging.Handler]:
+ """Create the list of handlers to be added to the root logger
+ based on commandline flags.
+ """
+
retval: List[logging.Handler] = []
# --logging_syslog (optionally with --logging_syslog_facility)
)
retval.append(console_handler)
+ # --logging_info_is_print echoes any message to logger.info(x) as
+ # a print statement on stdout.
+ if config.config["logging_info_is_print"]:
+ print_handler = logging.StreamHandler(sys.stdout)
+ print_handler.addFilter(OnlyInfoFilter())
+ retval.append(print_handler)
+
if len(retval) == 0:
retval.append(logging.NullHandler())
return retval
def _add_logging_handlers_and_filters(
logger: logging.Logger, handlers: List[logging.Handler]
) -> None:
+ """Adds a list of handlers to the root logger."""
for handler in handlers:
# These two are always present and set %(prefix) and %(suffix)
# if appropriate (see :class:`LoggingContext`) or make them
# empty string.
- handler.addFilter(PrefixAddingFilter(""))
- handler.addFilter(SuffixAddingFilter(""))
+ handler.addFilter(PrefixAddingFilter("", False))
+ handler.addFilter(SuffixAddingFilter("", False))
# --logging_squelch_repeats allows code to request repeat logging
# messages (identical log site and message contents) to be
handler.addFilter(ProbabilisticFilter())
logger.addHandler(handler)
- # --logging_info_is_print echoes any message to logger.info(x) as
- # a print statement on stdout.
- if config.config["logging_info_is_print"]:
- handler = logging.StreamHandler(sys.stdout)
- handler.addFilter(OnlyInfoFilter())
- logger.addHandler(handler)
-
def initialize_logging(logger=None) -> logging.Logger:
"""Initialize logging for the program. See module level comments
:meth:`bootstrap.initialize` decorator on your program's entry point,
it will call this for you. See :meth:`pyutils.bootstrap.initialize`
for more details.
+
+ Raises:
+ ValueError: if logging level is invalid
+
"""
+
global LOGGING_INITIALIZED
if LOGGING_INITIALIZED:
return logging.getLogger()
LOGGING_INITIALIZED = True
+ clear_preexisting = config.config["logging_clear_preexisting_handlers"]
preexisting_handlers_count = 0
if logger is None:
- # --logging_clear_preexisting_handlers removes logging handlers
- # that were registered by global statements during imported module
- # setup.
- if config.config["logging_clear_preexisting_handlers"]:
- logging.basicConfig(force=True)
- logger = logging.getLogger()
- while logger.hasHandlers():
- logger.removeHandler(logger.handlers[0])
- preexisting_handlers_count += 1
- else:
- logging.basicConfig()
- logger = logging.getLogger()
+ logging.basicConfig(force=clear_preexisting)
+ logger = logging.getLogger()
+
+ # --logging_clear_preexisting_handlers removes logging handlers
+ # that were registered by global statements during imported module
+ # setup.
+ if clear_preexisting:
+ while logger.hasHandlers():
+ logger.removeHandler(logger.handlers[0])
+ preexisting_handlers_count += 1
# --logging_config_file pulls logging settings from a config file
# skipping the rest of this setup.
return initialize_logging(logger)
-def tprint(*args, **kwargs) -> None:
- """Legacy function for printing a message augmented with thread id
- still needed by some code. Please use --logging_debug_threads in
- new code.
- """
- if config.config["logging_debug_threads"]:
- from pyutils.parallelize.thread_utils import current_thread_id
-
- print(f"{current_thread_id()}", end="")
- print(*args, **kwargs)
- else:
- pass
-
-
class OutputMultiplexer(object):
"""A class that broadcasts printed messages to several sinks
(including various logging levels, different files, different file
handles: if FILEHANDLES bit is set, this should be a list of
already opened filehandles you'd like to output into. The
handles will remain open after the scope of the multiplexer.
+
+ Raises:
+ ValueError: invalid combination of arguments (e.g. the filenames
+ argument is present but the filenames bit isn't set, the handle
+ argument is present but the handles bit isn't set, etc...)
"""
if logger is None:
logger = logging.getLogger(None)
return self.destination_bitv
def set_destination_bitv(self, destination_bitv: int):
- """Change the output destination_bitv to the one provided."""
+ """Change the output destination_bitv to the one provided.
+
+ Args:
+ destination_bitv: the new destination bitvector to set.
+
+ Raises:
+ ValueError: invalid combination of arguments (e.g. the filenames
+ argument is present but the filenames bit isn't set, the handle
+ argument is present but the handles bit isn't set, etc...)
+ """
if destination_bitv & self.Destination.FILENAMES and self.f is None:
raise ValueError("Filename argument is required if bitv & FILENAMES")
if destination_bitv & self.Destination.FILEHANDLES and self.h is None:
self.destination_bitv = destination_bitv
def print(self, *args, **kwargs):
- """Produce some output to all sinks."""
+ """Produce some output to all sinks. Use the same arguments as the
+ print-builtin.
+
+ Raises:
+ TypeError: Illegal argument types encountered
+ """
from pyutils.string_utils import _sprintf, strip_escape_sequences
end = kwargs.pop("end", None)
def __enter__(self):
return self
- def __exit__(self, etype, value, traceback) -> bool:
+ def __exit__(self, etype, value, tb) -> bool:
super().close()
if etype is not None:
return False
return True
+def _get_systemwide_abnormal_exit_handler(filename: str) -> FileHandler:
+ handler = FileHandler(filename)
+ handler.setFormatter(
+ MillisecondAwareFormatter(
+ fmt=_construct_logging_format(),
+ datefmt=config.config["logging_date_format"],
+ )
+ )
+ return handler
+
+
+def non_zero_return_value(ret: Any) -> bool:
+ """
+ Special method hooked from bootstrap.py to optionally keep a system-wide
+ record of non-zero python program exits.
+
+ Args:
+ ret: the return value
+ """
+ try:
+ record = config.config['logging_non_zero_exits_record_path']
+ if record:
+ logger = logging.getLogger()
+ handler = _get_systemwide_abnormal_exit_handler(record)
+ program = config.PROGRAM_NAME
+ args = config.ORIG_ARGV
+ with LoggingContext(logger, handlers=[handler], level=logging.ERROR) as log:
+ log.error('%s (%s) Exit %s', program, args, ret)
+ return True
+ except Exception:
+ pass
+ return False
+
+
+def unhandled_top_level_exception(
+ exc_type: type, exc_value: type, exc_tb: TracebackType
+) -> bool:
+ """
+ Special method hooked from bootstrap.py to optionally keep a system-wide
+ record of unhandled top level exceptions.
+
+ Args:
+ exc_type: the type of the unhandled exception
+ exc_value: the value passed to the exception's c'tor
+ exc_tb: the stack from where the exception was raised
+ """
+ try:
+ record = config.config['logging_unhandled_top_level_exceptions_record_path']
+ if record:
+ import traceback
+
+ logger = logging.getLogger()
+ handler = _get_systemwide_abnormal_exit_handler(record)
+ program = config.PROGRAM_NAME
+ args = config.ORIG_ARGV
+ tb = traceback.extract_tb(exc_tb)
+ original_frame = tb[-1]
+ raise_site_file = original_frame.filename
+ raise_site_lineno = original_frame.lineno
+ with LoggingContext(logger, handlers=[handler], level=logging.ERROR) as log:
+ log.error(
+ '%s (%s) unhandled top-level exception (type=%s(%s)) at %s:%s',
+ program,
+ args,
+ exc_type.__name__,
+ str(exc_value),
+ raise_site_file,
+ raise_site_lineno,
+ )
+ return True
+ except Exception:
+ pass
+ return False
+
+
def hlog(message: str) -> None:
"""Write a message to the house log (syslog facility local7 priority
info) by calling `/usr/bin/logger`. This is pretty hacky but used
projects / pyutils.git / blobdiff