tremors 0.2.0

Tremors is a library for logging with metrics.

Tremors is a library for logging with metrics. It provides a Logger that can be used like a standard Python Logger. However, the Tremors Logger is also a context manager with collectors. These collectors run whenever messages at target levels are logged, and add information to the LogRecord. These enriched LogRecords can then be processed by the standard logging mechanisms of Filters, Formatters, and Handlers.

Installation

pip install tremors

Usage

A function can be wrapped in a Tremors Logger context with the logged decorator. The function can then be called without a logger argument, and a Logger will automatically be injected into the function by the decorator. The logger parameter is specified in the function signature as shown in my_fn so IDEs and type checkers will work correctly.

import logging
from logging import config

import tremors
from tremors import collector

cfg = {
    "version": 1,
    "formatters": {"tremors": {"format": "Tremors > %(levelname)s:%(name)s:%(message)s"}},
    "handlers": {"console": {"class": "logging.StreamHandler", "formatter": "tremors"}},
    "loggers": {"my_fn": {"level": "INFO", "handlers": ["console"]}},
}
config.dictConfig(cfg)

@tremors.logged
def my_fn(arg: str, *, flag: bool, logger: tremors.Logger = tremors.from_logged) -> None:
    """A function with an injected context."""
    logger.info("arg: %s, flag: %s", arg, flag)

my_fn("foo", flag=True)

A standard Python Logger with the same name of the function, my_fn, is used to log messages according to the logging config. The context also automatically logs entered, and exited messages before, and after each function call.

Tremors > INFO:my_fn:entered: my_fn
Tremors > INFO:my_fn:arg: foo, flag: True
Tremors > INFO:my_fn:exited: my_fn

A typical pattern when naming a logger is to give it the name of the module in which it is instantiated. We can specify the name of the underlying Python Logger, while the name of the context will be the name of the decorated function.

cfg = {
    "version": 1,
    "formatters": {"tremors": {"format": "Tremors > %(levelname)s:%(name)s:%(message)s"}},
    "handlers": {"console": {"class": "logging.StreamHandler", "formatter": "tremors"}},
    "loggers": {__name__: {"level": "INFO", "handlers": ["console"]}},
}
config.dictConfig(cfg)

@tremors.logged(logger_name=__name__)
def my_fn(arg: str, *, flag: bool, logger: tremors.Logger = tremors.from_logged) -> None:
    """A function with an injected context with a named logger."""
    logger.info("arg: %s, flag: %s", arg, flag)

my_fn("foo", flag=True)

The messages now contain the name of the underlying logger that we explicitly specified.

Tremors > INFO:__main__:entered: my_fn
Tremors > INFO:__main__:arg: foo, flag: True
Tremors > INFO:__main__:exited: my_fn

Next we will use a collector included with Tremors to measure, and log the elapsed time since the function started for each logged message. All collector states are added to the LogRecord.tremors dict that maps the collector name to the collector state. The elapsed state will be available at LogRecord.tremors["elapsed"]. We can use a number of approaches to extract, and log this information.

Filters may modify LogRecords before they are formatted, and emitted by Handlers. Let’s use a filter function attached to a Handler to return a modifed LogRecord with an elapsed attribute. We can use a format function provided by Tremors to convert the raw elapsed state data to str with the elapsed seconds as our attribute value. Then we can reference the elapsed attribute, like any other LogRecord attribute, in the format str of the Formatter attached to the Handler.

import copy
import time

def elapsed_filter(record: logging.LogRecord) -> logging.LogRecord:
    """Add elapsed information to a LogRecord."""
    extra = getattr(record, tremors.EXTRA_KEY, None)
    if extra:
        record = copy.copy(record)
        record.elapsed = collector.format_elapsed("{elapsed:.4f}", extra) or ""
    return record

cfg = {
    "version": 1,
    "filters": {"tremors": {"()": lambda: elapsed_filter}},
    "formatters": {
        "tremors": {
            "format": "Tremors elapsed=%(elapsed)s > %(levelname)s:%(name)s:%(message)s"
        }
    },
    "handlers": {
        "console": {
            "class": "logging.StreamHandler",
            "filters": ["tremors"],
            "formatter": "tremors",
        }
    },
    "loggers": {__name__: {"level": "INFO", "handlers": ["console"]}},
}
config.dictConfig(cfg)

@tremors.logged(logger_name=__name__, collectors=(collector.elapsed(),))
def my_fn(arg: str, *, flag: bool, logger: tremors.Logger = tremors.from_logged) -> None:
    """A function with an injected context that has an elapsed collector."""
    logger.info("arg: %s, flag: %s", arg, flag)
    time.sleep(1)

my_fn("foo", flag=True)

The messages contain elapsed information added to the records by the Filter, then processed by the Formatter.

Tremors elapsed=0.0000 > INFO:__main__:entered: my_fn
Tremors elapsed=0.0001 > INFO:__main__:arg: foo, flag: True
Tremors elapsed=1.0003 > INFO:__main__:exited: my_fn

We can achieve the same outcome as the previous example using a Formatter instead of a Filter. Using a Filter acts at a lower level, is more straightforward, and relies on the documented approach of modifying LogRecords for downstream processing. However, if formatting the collector state is expensive, we may want to do that in a Formatter which only receives records that are to be logged.

class ElapsedFormatter(logging.Formatter):
    """A formatter to add elapsed information to logged messages."""

    def format(self, record: logging.LogRecord) -> str:
        """Add a formatted elapsed attribute to the record.

        Then delegate to the default formatter.
        """
        extra = getattr(record, tremors.EXTRA_KEY, None)
        if extra:
            record = copy.copy(record)
            record.elapsed = collector.format_elapsed("{elapsed:.4f}", extra) or ""
        return super().format(record)

cfg = {
    "version": 1,
    "formatters": {
        "tremors": {
            "class": "__main__.ElapsedFormatter",
            "format": "Tremors elapsed=%(elapsed)s > %(levelname)s:%(name)s:%(message)s",
        }
    },
    "handlers": {"console": {"class": "logging.StreamHandler", "formatter": "tremors"}},
    "loggers": {__name__: {"level": "INFO", "handlers": ["console"]}},
}
config.dictConfig(cfg)

my_fn("foo", flag=True)

The messages contain elapsed information based on the collector state that the Formatter was able to access from the records.

Tremors elapsed=0.0000 > INFO:__main__:entered: my_fn
Tremors elapsed=0.0003 > INFO:__main__:arg: foo, flag: True
Tremors elapsed=1.0006 > INFO:__main__:exited: my_fn

A Logger can have any number of collectors. Here, in addition to the elapsed collector from the previous examples, we’ll add a counter collector. The counter will only run if the level is ERROR or higher. It will be named errors, so it will available at LogRecord.tremors["errors"]

def tremors_filter(record: logging.LogRecord) -> logging.LogRecord:
    """Add counter, and elapsed information to a LogRecord."""
    extra = getattr(record, tremors.EXTRA_KEY, None)
    if extra:
        record = copy.copy(record)
        record.errors = collector.format_counter("{errors}", extra, name="errors") or "0"
        record.elapsed = collector.format_elapsed("{elapsed:.4f}", extra) or ""
    return record

cfg = {
    "version": 1,
    "filters": {"tremors": {"()": lambda: tremors_filter}},
    "formatters": {
        "tremors": {
            "format": "Tremors errors=%(errors)s elapsed=%(elapsed)s"
            " > %(levelname)s:%(name)s:%(message)s"
        }
    },
    "handlers": {
        "console": {
            "class": "logging.StreamHandler",
            "filters": ["tremors"],
            "formatter": "tremors",
        }
    },
    "loggers": {__name__: {"level": "INFO", "handlers": ["console"]}},
}
config.dictConfig(cfg)

@tremors.logged(
    logger_name=__name__,
    collectors=(collector.counter(name="errors", level=logging.ERROR), collector.elapsed()),
)
def my_fn(arg: str, *, flag: bool, logger: tremors.Logger = tremors.from_logged) -> None:
    """A function with an injected context that has multiple collectors."""
    logger.info("arg: %s, flag: %s", arg, flag)
    time.sleep(1)
    logger.error("uh-ho!")

my_fn("foo", flag=True)

The messages contain information from multiple colletors.

Tremors errors=0 elapsed=0.0000 > INFO:__main__:entered: my_fn
Tremors errors=0 elapsed=0.0002 > INFO:__main__:arg: foo, flag: True
Tremors errors=1 elapsed=1.0006 > ERROR:__main__:uh-ho!
Tremors errors=1 elapsed=1.0009 > INFO:__main__:exited: my_fn

You can define your own collector factories and formatters. See, for example, tremors.collector.counter, and tremors.collector.format_counter.