#!/usr/bin/env python3
# -*- coding: utf-8 -*-
-# © Copyright 2021-2022, Scott Gasch
+# © Copyright 2021-2023, Scott Gasch
"""
This is a module that offers an opinionated take on how whole program
* Set the logging default level (debug, info, warning, error, critical)
of the whole program (see: :code:`--logging_level`)... and to override
- the logging level for individual modules/functions based on their names
- (see :code:`--lmodule`),
+ the logging level with :class:`LoggingContext`.
+ * Prepend or append a message to every log record also with
+ :class:`LoggingContext`.
* define the logging message format (see :code:`--logging_format` and
:code:`--logging_date_format`) including easily adding a PID/TID
marker on all messages to help with multithreaded debugging
* 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 random
+import re
import sys
+from logging import FileHandler
from logging.config import fileConfig
from logging.handlers import RotatingFileHandler, SysLogHandler
-from typing import Any, Callable, Dict, Iterable, List, Optional
+from types import TracebackType
+from typing import Any, Callable, Dict, Iterable, List, Optional, Union
import pytz
from overrides import overrides
# This module is commonly used by others in here and should avoid
# taking any unnecessary dependencies back on them.
-from pyutils import argparse_utils, config
+from pyutils import argparse_utils, config, misc_utils
-cfg = config.add_commandline_args(f'Logging ({__file__})', 'Args related to logging')
+cfg = config.add_commandline_args(f"Logging ({__file__})", "Args related to logging")
cfg.add_argument(
- '--logging_config_file',
+ "--logging_config_file",
type=argparse_utils.valid_filename,
default=None,
- metavar='FILENAME',
- help='Config file containing the logging setup, see: https://docs.python.org/3/howto/logging.html#logging-advanced-tutorial',
+ metavar="FILENAME",
+ help="Config file containing the logging setup, see: https://docs.python.org/3/howto/logging.html#logging-advanced-tutorial",
)
cfg.add_argument(
- '--logging_level',
+ "--logging_level",
type=str,
- default='INFO',
- choices=['NOTSET', 'DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'],
- metavar='LEVEL',
- help='The global default level below which to squelch log messages; see also --lmodule',
+ default="INFO",
+ choices=["NOTSET", "DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"],
+ metavar="LEVEL",
+ help="The global default level below which to squelch log messages",
)
cfg.add_argument(
- '--logging_format',
+ "--logging_format",
type=str,
default=None,
- help='The format for lines logged via the logger module. See: https://docs.python.org/3/library/logging.html#formatter-objects',
+ help="The format for lines logged via the logger module. See: https://docs.python.org/3/library/logging.html#formatter-objects",
)
cfg.add_argument(
- '--logging_date_format',
+ "--logging_date_format",
type=str,
- default='%Y/%m/%dT%H:%M:%S.%f%z',
- metavar='DATEFMT',
- help='The format of any dates in --logging_format.',
+ default="%Y/%m/%dT%H:%M:%S.%f%z",
+ metavar="DATEFMT",
+ help="The format of any dates in --logging_format.",
)
cfg.add_argument(
- '--logging_console',
+ "--logging_console",
action=argparse_utils.ActionNoYes,
default=True,
- help='Should we log to the console (stderr)',
+ help="Should we log to the console (stderr)",
)
cfg.add_argument(
- '--logging_filename',
+ "--logging_filename",
type=str,
default=None,
- metavar='FILENAME',
- help='The filename of the logfile to write.',
+ metavar="FILENAME",
+ help="The filename of the logfile to write.",
)
cfg.add_argument(
- '--logging_filename_maxsize',
- type=int,
+ "--logging_filename_maxsize",
+ type=argparse_utils.valid_byte_count,
default=(1024 * 1024),
- metavar='#BYTES',
- help='The maximum size (in bytes) to write to the logging_filename.',
+ metavar="#BYTES",
+ help="The maximum size (in bytes) to write to the logging_filename.",
)
cfg.add_argument(
- '--logging_filename_count',
+ "--logging_filename_count",
type=int,
default=7,
- metavar='COUNT',
- help='The number of logging_filename copies to keep before deleting.',
+ metavar="COUNT",
+ help="The number of logging_filename copies to keep before deleting.",
)
cfg.add_argument(
- '--logging_syslog',
+ "--logging_syslog",
action=argparse_utils.ActionNoYes,
default=False,
- help='Should we log to localhost\'s syslog.',
+ help="Should we log to localhost's syslog.",
)
cfg.add_argument(
- '--logging_syslog_facility',
+ "--logging_syslog_facility",
type=str,
- default='USER',
+ default="USER",
choices=[
- 'NOTSET',
- 'AUTH',
- 'AUTH_PRIV',
- 'CRON',
- 'DAEMON',
- 'FTP',
- 'KERN',
- 'LPR',
- 'MAIL',
- 'NEWS',
- 'SYSLOG',
- 'USER',
- 'UUCP',
- 'LOCAL0',
- 'LOCAL1',
- 'LOCAL2',
- 'LOCAL3',
- 'LOCAL4',
- 'LOCAL5',
- 'LOCAL6',
- 'LOCAL7',
+ "NOTSET",
+ "AUTH",
+ "AUTH_PRIV",
+ "CRON",
+ "DAEMON",
+ "FTP",
+ "KERN",
+ "LPR",
+ "MAIL",
+ "NEWS",
+ "SYSLOG",
+ "USER",
+ "UUCP",
+ "LOCAL0",
+ "LOCAL1",
+ "LOCAL2",
+ "LOCAL3",
+ "LOCAL4",
+ "LOCAL5",
+ "LOCAL6",
+ "LOCAL7",
],
- metavar='SYSLOG_FACILITY_LIST',
- help='The default syslog message facility identifier',
+ metavar="SYSLOG_FACILITY_LIST",
+ help="The default syslog message facility identifier",
)
cfg.add_argument(
- '--logging_debug_threads',
+ "--logging_debug_threads",
action=argparse_utils.ActionNoYes,
default=False,
- help='Should we prepend pid/tid data to all log messages?',
+ help="Should we prepend pid/tid data to all log messages?",
)
cfg.add_argument(
- '--logging_debug_modules',
+ "--logging_debug_modules",
action=argparse_utils.ActionNoYes,
default=False,
- help='Should we prepend module/function data to all log messages?',
+ help="Should we prepend module/function data to all log messages?",
)
cfg.add_argument(
- '--logging_info_is_print',
+ "--logging_info_is_print",
action=argparse_utils.ActionNoYes,
default=False,
- help='logging.info also prints to stdout.',
+ help="logging.info also prints to stdout.",
)
cfg.add_argument(
- '--logging_squelch_repeats',
+ "--logging_squelch_repeats",
action=argparse_utils.ActionNoYes,
default=True,
- help='Do we allow code to indicate that it wants to squelch repeated logging messages or should we always log?',
+ help="Do we allow code to indicate that it wants to squelch repeated logging messages or should we always log?",
)
cfg.add_argument(
- '--logging_probabilistically',
+ "--logging_probabilistically",
action=argparse_utils.ActionNoYes,
default=True,
- help='Do we allow probabilistic logging (for code that wants it) or should we always log?',
+ help="Do we allow probabilistic logging (for code that wants it) or should we always log?",
)
# See also: OutputMultiplexer
cfg.add_argument(
- '--logging_captures_prints',
+ "--logging_captures_prints",
action=argparse_utils.ActionNoYes,
default=False,
- help='When calling print, also log.info automatically.',
-)
-cfg.add_argument(
- '--lmodule',
- type=str,
- metavar='<SCOPE>=<LEVEL>[,<SCOPE>=<LEVEL>...]',
- help=(
- 'Allows per-scope logging levels which override the global level set with --logging-level.'
- + 'Pass a space separated list of <scope>=<level> where <scope> is one of: module, '
- + 'module:function, or :function and <level> is a logging level (e.g. INFO, DEBUG...)'
- ),
+ help="When calling print, also log.info automatically.",
)
cfg.add_argument(
- '--logging_clear_preexisting_handlers',
+ "--logging_clear_preexisting_handlers",
action=argparse_utils.ActionNoYes,
default=True,
help=(
- 'Should logging code clear preexisting global logging handlers and thus insist that is '
- + 'alone can add handlers. Use this to work around annoying modules that insert global '
- + 'handlers with formats and logging levels you might now want. Caveat emptor, this may '
- + 'cause you to miss logging messages.'
+ "Should logging code clear preexisting global logging handlers and thus insist that is "
+ + "alone can add handlers. Use this to work around annoying modules that insert global "
+ + "handlers with formats and logging levels you might now want. Caveat emptor, this may "
+ + "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
@overrides
def filter(self, record: logging.LogRecord) -> bool:
"""Should we drop this log message?"""
- id1 = f'{record.module}:{record.funcName}'
+ id1 = f"{record.module}:{record.funcName}"
if id1 not in squelched_logging_counts:
return True
threshold = squelched_logging_counts[id1]
- logsite = f'{record.pathname}+{record.lineno}+{record.levelno}+{record.msg}'
+ logsite = f"{record.pathname}+{record.lineno}+{record.levelno}+{record.msg}"
count = self.counters[logsite]
self.counters[logsite] += 1
return count < threshold
-class DynamicPerScopeLoggingLevelFilter(logging.Filter):
- """This filter only allows logging messages from an allow list of
- module names or `module:function` names. Blocks all others. This
- filter is used to implement the :code:`--lmodule` commandline option.
-
- .. note::
-
- You probably don't need to use this directly, just use
- :code:`--lmodule`. For example, to set logging level to INFO
- everywhere except "module:function" where it should be DEBUG::
-
- # myprogram.py --logging_level=INFO --lmodule=module:function=DEBUG
-
- """
-
- @staticmethod
- def level_name_to_level(name: str) -> int:
- """Given a level name, return its numberic value."""
- numeric_level = getattr(logging, name, None)
- if not isinstance(numeric_level, int):
- raise ValueError(f'Invalid level: {name}')
- return numeric_level
-
- def __init__(
- self,
- default_logging_level: int,
- per_scope_logging_levels: Optional[str],
- ) -> None:
- """Construct the Filter.
-
- Args:
- default_logging_level: the logging level of the whole program
- per_scope_logging_levels: optional, comma separated overrides of
- logging level per scope of the format scope=level where
- scope is of the form "module:function" or ":function" and
- level is one of NOTSET, DEBUG, INFO, WARNING, ERROR or
- CRITICAL.
- """
- super().__init__()
- self.valid_levels = set(
- ['NOTSET', 'DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL']
- )
- self.default_logging_level = default_logging_level
- self.level_by_scope = {}
- if per_scope_logging_levels is not None:
- for chunk in per_scope_logging_levels.split(','):
- if '=' not in chunk:
- print(
- f'Malformed lmodule directive: "{chunk}", missing "=". Ignored.',
- file=sys.stderr,
- )
- continue
- try:
- (scope, level) = chunk.split('=')
- except ValueError:
- print(
- f'Malformed lmodule directive: "{chunk}". Ignored.',
- file=sys.stderr,
- )
- continue
- scope = scope.strip()
- level = level.strip().upper()
- if level not in self.valid_levels:
- print(
- f'Malformed lmodule directive: "{chunk}", bad level. Ignored.',
- file=sys.stderr,
- )
- continue
- self.level_by_scope[
- scope
- ] = DynamicPerScopeLoggingLevelFilter.level_name_to_level(level)
-
- @overrides
- def filter(self, record: logging.LogRecord) -> bool:
- """Decides whether or not to log based on an allow list."""
-
- # First try to find a logging level by scope (--lmodule)
- if len(self.level_by_scope) > 0:
- min_level = None
- for scope in (
- record.module,
- f'{record.module}:{record.funcName}',
- f':{record.funcName}',
- ):
- level = self.level_by_scope.get(scope, None)
- if level is not None:
- if min_level is None or level < min_level:
- min_level = level
-
- # If we found one, use it instead of the global default level.
- if min_level is not None:
- return record.levelno >= min_level
-
- # Otherwise, use the global logging level (--logging_level)
- return record.levelno >= self.default_logging_level
-
-
# A map from function_identifier -> probability of logging (0.0%..100.0%)
probabilistic_logging_levels: Dict[str, float] = {}
@overrides
def filter(self, record: logging.LogRecord) -> bool:
"""Should the message be logged?"""
- id1 = f'{record.module}:{record.funcName}'
- if id1 not in probabilistic_logging_levels:
- return True
- threshold = probabilistic_logging_levels[id1]
- return (random.random() * 100.0) <= threshold
+ identifier = f"{record.module}:{record.funcName}"
+ threshold = probabilistic_logging_levels.get(identifier, 100.0)
+ return misc_utils.execute_probabilistically(threshold)
class OnlyInfoFilter(logging.Filter):
return record.levelno == logging.INFO
-class MillisecondAwareFormatter(logging.Formatter):
+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
+ else:
+ self.prefix = ""
+ self.klobber = klobber
+
+ @overrides
+ def filter(self, record: logging.LogRecord):
+ if self.klobber:
+ record.prefix = self.prefix
+ elif "prefix" not in record.__dict__:
+ record.prefix = self.prefix
+ return True
+
+
+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
+ else:
+ self.suffix = ""
+ self.klobber = klobber
+
+ @overrides
+ def filter(self, record: logging.LogRecord):
+ if self.klobber:
+ record.suffix = self.suffix
+ elif "suffix" not in record.__dict__:
+ record.suffix = self.suffix
+ return True
+
+
+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
"""
- A formatter for adding milliseconds to log messages which, for
+
+ def process(self, msg, kwargs):
+ return f'{self.extra.get("prefix", "")}{msg}', kwargs
+
+ @staticmethod
+ def wrap_logger(prefix: str, logger: logging.Logger) -> logging.LoggerAdapter:
+ """Helper method around the creation of a LogAdapter that prepends
+ a given string to every log message produced.
+
+ Args:
+ prefix: the message to prepend to every log message.
+ logger: the logger whose messages to modify.
+
+ Returns:
+ A new logger wrapping the old one with the given behavior.
+ The old logger will continue to behave as usual; simply drop
+ the reference to this wrapper when it's no longer needed.
+ """
+ assert prefix
+ return PrependingLogAdapter(logger, {"prefix": prefix})
+
+
+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
+
+ @staticmethod
+ def wrap_logger(suffix: str, logger: logging.Logger) -> logging.LoggerAdapter:
+ """Helper method around the creation of a LoggerAdapter that appends
+ a given string to every log message produced.
+
+ Args:
+ suffix: the message to prepend to every log message.
+ logger: the logger whose messages to modify.
+
+ Returns:
+ A new logger wrapping the old one with the given behavior.
+ The old logger will continue to behave as usual; simply drop
+ the reference to this wrapper when it's no longer needed.
+ """
+ assert suffix
+ return AppendingLogAdapter(logger, {"suffix": suffix})
+
+
+class LoggingContext(contextlib.ContextDecorator):
+ def __init__(
+ self,
+ logger: logging.Logger,
+ *,
+ level: Optional[int] = None,
+ handlers: Optional[List[logging.Handler]] = None,
+ prefix: Optional[str] = None,
+ suffix: Optional[str] = None,
+ ):
+ """This is a logging context that can be used to temporarily change the
+ way we are logging within its scope. Logging changes may include:
+
+ - Changing the logging level (e.g. from INFO to DEBUG)
+ - Adding a prefix or suffix to every log message produced
+ - Adding temporary Handlers to direct the logging output elsewhere
+
+ Setup for doctest / examples. This will normally be taken care of
+ by code in :meth:`initialize_logging` so you don't have to worry
+ about it.
+
+ >>> logging_format = "%(prefix)s%(message)s%(suffix)s"
+ >>> logger = logging.getLogger(__name__ + ".LoggingContext")
+ >>> logger.setLevel(logging.INFO)
+ >>> handler = logging.StreamHandler(sys.stdout)
+ >>> handler.setFormatter(
+ ... MillisecondAwareFormatter(
+ ... fmt=logging_format,
+ ... datefmt='',
+ ... )
+ ... )
+ >>> logger.addHandler(handler)
+ >>> logger.addFilter(PrefixAddingFilter(None))
+ >>> logger.addFilter(SuffixAddingFilter(None))
+
+ First, this logger should be currently be configured to send
+ INFO+ messages to sys.stdout. Let's see it in action:
+
+ >>> logger.info("Hello world!")
+ Hello world!
+ >>> logger.debug("You should not see this")
+
+ The first example is to simply change the level of the logger.
+ Here we temporarily change it to DEBUG within the body of the
+ :class:`LoggingContext`:
+
+ >>> with LoggingContext(logger, level=logging.DEBUG):
+ ... logger.debug("You should see this now")
+ ... logger.info("Of course you should still see this too")
+ You should see this now
+ Of course you should still see this too
+
+ >>> logger.debug("Outside of the context we are at INFO again")
+ >>> logger.debug("(which is why you don't see these)")
+ >>> logger.info("But you should see this at INFO level")
+ But you should see this at INFO level
+
+ The prefix and suffix argument prepend or append a message to
+ all log output. To do this, you need %(prefix)s and
+ %(suffix)s placeholders in your logger format string
+ indicating where to insert the data. This is useful, for
+ example, to add an active request identifier to the set of log
+ messages produced while processing it.
+
+ >>> logger.info("About to work on a new request")
+ About to work on a new request
+
+ >>> with LoggingContext(logger, prefix='10.0.0.13> '):
+ ... logger.info("Working on it now")
+ 10.0.0.13> Working on it now
+
+ >>> logger.info("Done with that request")
+ Done with that request
+
+ LoggingContext can also be used to add temporary handler(s).
+ This code temporarily uses two stdout handlers to double the
+ output for testing purporses but you could also temporarily,
+ e.g., add a RotatingFileHandler or SysLogHandler etc...
+
+ >>> with LoggingContext(logger, handlers=[logging.StreamHandler(sys.stdout)]):
+ ... logger.info("TEST")
+ TEST
+ TEST
+
+ Once leaving the context, logger's behavior is restored. In
+ this case, the extra handler is removed so output will not
+ longer be doubled.
+
+ >>> logger.info("OUTSIDE")
+ OUTSIDE
+
+ LoggingContext can also be used as a decorator if that is more
+ convenient:
+
+ >>> @LoggingContext(logger, level=logging.DEBUG)
+ ... def log_stuff(logger):
+ ... logger.debug("But inside, the decorator has changed us to DEBUG")
+
+ >>> logger.debug("Outside, we're at INFO level and you don't see this")
+ >>> log_stuff(logger)
+ 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
+ self.old_level: Optional[int] = None
+ self.handlers = handlers
+ if prefix:
+ self.prefix: Optional[logging.Filter] = PrefixAddingFilter(prefix, True)
+ else:
+ self.prefix = None
+ if suffix:
+ self.suffix: Optional[logging.Filter] = SuffixAddingFilter(suffix, True)
+ else:
+ self.suffix = None
+
+ def __enter__(self) -> Union[logging.Logger, logging.LoggerAdapter]:
+ assert self.logger
+ if self.level:
+ self.old_level = self.logger.level
+ self.logger.setLevel(self.level)
+
+ if self.handlers:
+ for handler in self.handlers:
+ self.logger.addHandler(handler)
+
+ if self.prefix:
+ self.logger.addFilter(self.prefix)
+
+ if self.suffix:
+ self.logger.addFilter(self.suffix)
+ return self.logger
+
+ def __exit__(self, et, ev, tb) -> None:
+ if self.suffix:
+ self.logger.removeFilter(self.suffix)
+
+ if self.prefix:
+ self.logger.removeFilter(self.prefix)
+
+ if self.handlers:
+ for handler in self.handlers:
+ self.logger.removeHandler(handler)
+ handler.close()
+
+ if self.old_level:
+ self.logger.setLevel(self.old_level)
+ return None # propagate exceptions out
+
+
+class MillisecondAwareFormatter(logging.Formatter):
+ """A formatter for adding milliseconds to log messages which, for
whatever reason, the default Python logger doesn't do.
.. note::
You probably don't need to use this directly but it is
- wired in under :meth:initialize_logging so that the
+ wired in under :meth:`initialize_logging` so that the
timestamps in log messages have millisecond level
precision.
"""
level_name = logging._levelToName.get(
default_logging_level, str(default_logging_level)
)
- logger.debug('Initialized global logging; default logging level is %s.', level_name)
+ logger.debug("Initialized global logging; logging level is %s.", level_name)
if (
- config.config['logging_clear_preexisting_handlers']
+ config.config["logging_clear_preexisting_handlers"]
and preexisting_handlers_count > 0
):
logger.debug(
- 'Logging cleared %d global handlers (--logging_clear_preexisting_handlers)',
+ "Logging cleared %d global handlers (--logging_clear_preexisting_handlers)",
preexisting_handlers_count,
)
logger.debug('Logging format specification is "%s"', fmt)
- if config.config['logging_debug_threads']:
+ if config.config["logging_debug_threads"]:
logger.debug(
- '...Logging format spec captures tid/pid. (--logging_debug_threads)'
+ "...Logging format spec captures tid/pid. (--logging_debug_threads)"
)
- if config.config['logging_debug_modules']:
+ if config.config["logging_debug_modules"]:
logger.debug(
- '...Logging format spec captures files/functions/lineno. (--logging_debug_modules)'
+ "...Logging format spec captures files/functions/lineno. (--logging_debug_modules)"
)
- if config.config['logging_syslog']:
+ if config.config["logging_syslog"]:
logger.debug(
- 'Logging to syslog as %s with priority mapping based on level. (--logging_syslog)',
+ "Logging to syslog as %s with priority mapping based on level. (--logging_syslog)",
facility_name,
)
- if config.config['logging_filename']:
+ if config.config["logging_filename"]:
logger.debug(
'Logging to file "%s". (--logging_filename)',
config.config["logging_filename"],
)
logger.debug(
- '...with %d bytes max file size. (--logging_filename_maxsize)',
+ "...with %d bytes max file size. (--logging_filename_maxsize)",
config.config["logging_filename_maxsize"],
)
logger.debug(
- '...and %d rotating backup file count. (--logging_filename_count)',
+ "...and %d rotating backup file count. (--logging_filename_count)",
config.config["logging_filename_count"],
)
- if config.config['logging_console']:
- logger.debug('Logging to the console (stderr). (--logging_console)')
- if config.config['logging_info_is_print']:
+ if config.config["logging_console"]:
+ logger.debug("Logging to the console (stderr). (--logging_console)")
+ if config.config["logging_info_is_print"]:
logger.debug(
- 'Logging logger.info messages will be repeated on stdout. (--logging_info_is_print)'
+ "Logging logger.info messages will be repeated on stdout. (--logging_info_is_print)"
)
- if config.config['logging_squelch_repeats']:
+ if config.config["logging_squelch_repeats"]:
logger.debug(
- 'Logging code allowed to request repeated messages be squelched. (--logging_squelch_repeats)'
+ "Logging code allowed to request repeated messages be squelched. (--logging_squelch_repeats)"
)
else:
logger.debug(
- 'Logging code forbidden to request messages be squelched; all messages logged. (--no_logging_squelch_repeats)'
+ "Logging code forbidden to request messages be squelched; all messages logged. (--no_logging_squelch_repeats)"
)
- if config.config['logging_probabilistically']:
+ if config.config["logging_probabilistically"]:
logger.debug(
- 'Logging code is allowed to request probabilistic logging. (--logging_probabilistically)'
+ "Logging code is allowed to request probabilistic logging. (--logging_probabilistically)"
)
else:
logger.debug(
- 'Logging code is forbidden to request probabilistic logging; messages always logged. (--no_logging_probabilistically)'
+ "Logging code is forbidden to request probabilistic logging; messages always logged. (--no_logging_probabilistically)"
)
- if config.config['lmodule']:
+ if config.config["logging_captures_prints"]:
logger.debug(
- f'Logging dynamic per-module logging enabled. (--lmodule={config.config["lmodule"]})'
+ "Logging will capture printed data as logger.info messages. (--logging_captures_prints)"
)
- if config.config['logging_captures_prints']:
- logger.debug(
- 'Logging will capture printed data as logger.info messages. (--logging_captures_prints)'
+
+
+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"]
+ elif config.config["logging_syslog"]:
+ fmt = "%(levelname).1s:%(filename)s[%(process)d]: "
+ else:
+ fmt = "%(levelname).1s:%(asctime)s: "
+
+ fmt += "%(prefix)s%(message)s%(suffix)s"
+
+ # --logging_debug_threads and --logging_debug_modules both affect
+ # the format by prepending information about the pid/tid or
+ # file/function.
+ if config.config["logging_debug_threads"]:
+ fmt = f"%(process)d.%(thread)d|{fmt}"
+ if config.config["logging_debug_modules"]:
+ fmt = f"%(filename)s:%(funcName)s:%(lineno)s|{fmt}"
+ return fmt
+
+
+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)
+ # sets up for logging to use the standard system syslogd as a
+ # sink.
+ if config.config["logging_syslog"]:
+ if sys.platform not in ("win32", "cygwin"):
+ assert facility_name
+ facility = SysLogHandler.__dict__.get(facility_name, SysLogHandler.LOG_USER)
+ assert facility
+ syslog_handler = SysLogHandler(facility=facility, address="/dev/log")
+ syslog_handler.setFormatter(
+ MillisecondAwareFormatter(
+ fmt=logging_format,
+ datefmt=config.config["logging_date_format"],
+ )
+ )
+ retval.append(syslog_handler)
+
+ # --logging_filename (with friends --logging_filename_count and
+ # --logging_filename_maxsize) set up logging to a file on the
+ # filesystem with automatic rotation when it gets too big.
+ if config.config["logging_filename"]:
+ max_bytes = config.config["logging_filename_maxsize"]
+ assert max_bytes and isinstance(max_bytes, int)
+ backup_count = config.config["logging_filename_count"]
+ assert backup_count and isinstance(backup_count, int)
+ file_handler = RotatingFileHandler(
+ config.config["logging_filename"],
+ maxBytes=max_bytes,
+ backupCount=backup_count,
)
+ file_handler.setFormatter(
+ MillisecondAwareFormatter(
+ fmt=logging_format,
+ datefmt=config.config["logging_date_format"],
+ )
+ )
+ retval.append(file_handler)
+
+ # --logging_console is, ahem, logging to the console.
+ if config.config["logging_console"]:
+ console_handler = logging.StreamHandler(sys.stderr)
+ console_handler.setFormatter(
+ MillisecondAwareFormatter(
+ fmt=logging_format,
+ datefmt=config.config["logging_date_format"],
+ )
+ )
+ 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("", False))
+ handler.addFilter(SuffixAddingFilter("", False))
+
+ # --logging_squelch_repeats allows code to request repeat logging
+ # messages (identical log site and message contents) to be
+ # silenced. Logging code must request this explicitly, it isn't
+ # automatic. This option just allows the silencing to happen.
+ if config.config["logging_squelch_repeats"]:
+ handler.addFilter(SquelchRepeatedMessagesFilter())
+
+ # --logging_probabilistically allows code to request
+ # non-deterministic logging where messages have some probability
+ # of being produced. Logging code must request this explicitly.
+ # This option just allows the non-deterministic behavior to
+ # happen. Disabling it will cause every log message to be
+ # produced.
+ if config.config["logging_probabilistically"]:
+ handler.addFilter(ProbabilisticFilter())
+ logger.addHandler(handler)
def initialize_logging(logger=None) -> logging.Logger:
: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.basicConfig(force=clear_preexisting)
logger = logging.getLogger()
# --logging_clear_preexisting_handlers removes logging handlers
# that were registered by global statements during imported module
# setup.
- preexisting_handlers_count = 0
- assert config.has_been_parsed()
- if config.config['logging_clear_preexisting_handlers']:
+ 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.
- if config.config['logging_config_file'] is not None:
- fileConfig(config.config['logging_config_file'])
+ if config.config["logging_config_file"] is not None:
+ fileConfig(config.config["logging_config_file"])
return logger
- handlers: List[logging.Handler] = []
- handler: Optional[logging.Handler] = None
-
# Global default logging level (--logging_level); messages below
# this level will be silenced.
- default_logging_level = getattr(
- logging, config.config['logging_level'].upper(), None
- )
+ logging_level = config.config["logging_level"]
+ assert logging_level
+ logging_level = logging_level.upper()
+ default_logging_level = getattr(logging, logging_level, None)
if not isinstance(default_logging_level, int):
raise ValueError(f'Invalid level: {config.config["logging_level"]}')
+ logger.setLevel(default_logging_level)
- # Custom or default --logging_format?
- if config.config['logging_format']:
- fmt = config.config['logging_format']
- else:
- if config.config['logging_syslog']:
- fmt = '%(levelname).1s:%(filename)s[%(process)d]: %(message)s'
- else:
- fmt = '%(levelname).1s:%(asctime)s: %(message)s'
-
- # --logging_debug_threads and --logging_debug_modules both affect
- # the format by prepending information about the pid/tid or
- # file/function.
- if config.config['logging_debug_threads']:
- fmt = f'%(process)d.%(thread)d|{fmt}'
- if config.config['logging_debug_modules']:
- fmt = f'%(filename)s:%(funcName)s:%(lineno)s|{fmt}'
-
- # --logging_syslog (optionally with --logging_syslog_facility)
- # sets up for logging to use the standard system syslogd as a
- # sink.
+ logging_format = _construct_logging_format()
facility_name = None
- if config.config['logging_syslog']:
- if sys.platform not in ('win32', 'cygwin'):
- if config.config['logging_syslog_facility']:
- facility_name = 'LOG_' + config.config['logging_syslog_facility']
- facility = SysLogHandler.__dict__.get(facility_name, SysLogHandler.LOG_USER) # type: ignore
- assert facility is not None
- handler = SysLogHandler(facility=facility, address='/dev/log')
- handler.setFormatter(
- MillisecondAwareFormatter(
- fmt=fmt,
- datefmt=config.config['logging_date_format'],
- )
- )
- handlers.append(handler)
-
- # --logging_filename (with friends --logging_filename_count and
- # --logging_filename_maxsize) set up logging to a file on the
- # filesystem with automatic rotation when it gets too big.
- if config.config['logging_filename']:
- handler = RotatingFileHandler(
- config.config['logging_filename'],
- maxBytes=config.config['logging_filename_maxsize'],
- backupCount=config.config['logging_filename_count'],
- )
- handler.setFormatter(
- MillisecondAwareFormatter(
- fmt=fmt,
- datefmt=config.config['logging_date_format'],
- )
- )
- handlers.append(handler)
-
- # --logging_console is, ahem, logging to the console.
- if config.config['logging_console']:
- handler = logging.StreamHandler(sys.stderr)
- handler.setFormatter(
- MillisecondAwareFormatter(
- fmt=fmt,
- datefmt=config.config['logging_date_format'],
- )
- )
- handlers.append(handler)
-
- if len(handlers) == 0:
- handlers.append(logging.NullHandler())
- for handler in handlers:
- 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)
-
- # --logging_squelch_repeats allows code to request repeat logging
- # messages (identical log site and message contents) to be
- # silenced. Logging code must request this explicitly, it isn't
- # automatic. This option just allows the silencing to happen.
- if config.config['logging_squelch_repeats']:
- for handler in handlers:
- handler.addFilter(SquelchRepeatedMessagesFilter())
-
- # --logging_probabilistically allows code to request
- # non-deterministic logging where messages have some probability
- # of being produced. Logging code must request this explicitly.
- # This option just allows the non-deterministic behavior to
- # happen. Disabling it will cause every log message to be
- # produced.
- if config.config['logging_probabilistically']:
- for handler in handlers:
- handler.addFilter(ProbabilisticFilter())
-
- # --lmodule is a way to have a special logging level for just on
- # module or one set of modules that is different than the one set
- # globally via --logging_level.
- for handler in handlers:
- handler.addFilter(
- DynamicPerScopeLoggingLevelFilter(
- default_logging_level,
- config.config['lmodule'],
- )
- )
- logger.setLevel(0)
- logger.propagate = False
+ if config.config["logging_syslog"]:
+ if sys.platform not in ("win32", "cygwin"):
+ if config.config["logging_syslog_facility"]:
+ facility_name = "LOG_" + config.config["logging_syslog_facility"]
+ handlers: List[logging.Handler] = _construct_handlers_list(
+ logging_format, facility_name
+ )
+ _add_logging_handlers_and_filters(logger, handlers)
# --logging_captures_prints, if set, will capture and log.info
# anything printed on stdout.
- if config.config['logging_captures_prints']:
+ if config.config["logging_captures_prints"]:
import builtins
def print_and_also_log(*arg, **kwarg):
- f = kwarg.get('file', None)
+ f = kwarg.get("file", None)
if f == sys.stderr:
logger.warning(*arg)
else:
logger,
default_logging_level,
preexisting_handlers_count,
- fmt,
+ logging_format,
facility_name,
)
return logger
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)
self.f: Optional[List[Any]] = None
if filenames is not None:
- self.f = [open(filename, 'wb', buffering=0) for filename in filenames]
+ self.f = [open(filename, "wb", buffering=0) for filename in filenames]
else:
if destination_bitv & OutputMultiplexer.Destination.FILENAMES:
raise ValueError("Filenames argument is required if bitv & FILENAMES")
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)
sep = " "
if end is None:
end = "\n"
- if end == '\n':
- buf += '\n'
+ if end == "\n":
+ buf += "\n"
if self.destination_bitv & self.Destination.FILENAMES and self.f is not None:
for _ in self.f:
- _.write(buf.encode('utf-8'))
+ _.write(buf.encode("utf-8"))
_.flush()
if self.destination_bitv & self.Destination.FILEHANDLES and self.h is not 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
os.system(f"/usr/bin/logger -p local7.info -- '{message}'")
-if __name__ == '__main__':
+if __name__ == "__main__":
import doctest
doctest.testmod()
projects / pyutils.git / blobdiff