#!/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 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
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",
+ help="The global default level below which to squelch log messages",
)
cfg.add_argument(
"--logging_format",
)
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.",
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...)"
- ),
-)
cfg.add_argument(
"--logging_clear_preexisting_handlers",
action=argparse_utils.ActionNoYes,
+ "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
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] = {}
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
+ """
+
+ 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!!!
"""
- A formatter for adding milliseconds to log messages which, for
+
+ 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"]
and preexisting_handlers_count > 0
logger.debug(
"Logging code is forbidden to request probabilistic logging; messages always logged. (--no_logging_probabilistically)"
)
- if config.config["lmodule"]:
- logger.debug(
- f'Logging dynamic per-module logging enabled. (--lmodule={config.config["lmodule"]})'
- )
if config.config["logging_captures_prints"]:
logger.debug(
"Logging will capture printed data as logger.info messages. (--logging_captures_prints)"
)
-def initialize_logging(logger=None) -> logging.Logger:
- """Initialize logging for the program. See module level comments
- for information about what functionality this provides and how to
- enable or disable functionality via the commandline.
-
- If you use the
- :meth:`bootstrap.initialize` decorator on your program's entry point,
- it will call this for you. See :meth:`pyutils.bootstrap.initialize`
- for more details.
- """
- global LOGGING_INITIALIZED
- if LOGGING_INITIALIZED:
- return logging.getLogger()
- LOGGING_INITIALIZED = True
-
- if logger is None:
- 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"]:
- 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"])
- return logger
-
- handlers: List[logging.Handler] = []
- handler: Optional[logging.Handler] = None
-
- # Global default logging level (--logging_level); messages below
- # this level will be silenced.
- 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"]}')
+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]: %(message)s"
+ fmt = "%(levelname).1s:%(filename)s[%(process)d]: "
else:
- fmt = "%(levelname).1s:%(asctime)s: %(message)s"
+ 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
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.
- 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(
+ 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=fmt,
+ fmt=logging_format,
datefmt=config.config["logging_date_format"],
)
)
- handlers.append(handler)
+ retval.append(syslog_handler)
# --logging_filename (with friends --logging_filename_count and
# --logging_filename_maxsize) set up logging to a file on the
assert max_bytes and isinstance(max_bytes, int)
backup_count = config.config["logging_filename_count"]
assert backup_count and isinstance(backup_count, int)
- handler = RotatingFileHandler(
+ file_handler = RotatingFileHandler(
config.config["logging_filename"],
maxBytes=max_bytes,
backupCount=backup_count,
)
- handler.setFormatter(
+ file_handler.setFormatter(
MillisecondAwareFormatter(
- fmt=fmt,
+ fmt=logging_format,
datefmt=config.config["logging_date_format"],
)
)
- handlers.append(handler)
+ retval.append(file_handler)
# --logging_console is, ahem, logging to the console.
if config.config["logging_console"]:
- handler = logging.StreamHandler(sys.stderr)
- handler.setFormatter(
+ console_handler = logging.StreamHandler(sys.stderr)
+ console_handler.setFormatter(
MillisecondAwareFormatter(
- fmt=fmt,
+ fmt=logging_format,
datefmt=config.config["logging_date_format"],
)
)
- handlers.append(handler)
-
- if len(handlers) == 0:
- handlers.append(logging.NullHandler())
- for handler in handlers:
- logger.addHandler(handler)
+ 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"]:
- handler = logging.StreamHandler(sys.stdout)
- handler.addFilter(OnlyInfoFilter())
- logger.addHandler(handler)
+ print_handler = logging.StreamHandler(sys.stdout)
+ print_handler.addFilter(OnlyInfoFilter())
+ retval.append(print_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:
+ 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"]:
- for handler in handlers:
+ # --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)
- # --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
+
+def initialize_logging(logger=None) -> logging.Logger:
+ """Initialize logging for the program. See module level comments
+ for information about what functionality this provides and how to
+ enable or disable functionality via the commandline.
+
+ If you use the
+ :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.
+ 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"])
+ return logger
+
+ # Global default logging level (--logging_level); messages below
+ # this level will be silenced.
+ 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)
+
+ 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"]
+ 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.
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)
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