Suppress multiple messages with same content in Python logging module AKA log compression

Question

By design, my application sometimes produces repeating errors which fill up the log file and make it annoying to read. It looks like that:

WARNING:__main__:CRON10: clock unset or no wind update received in 60 sec -> supressed rrd update
WARNING:__main__:CRON10: clock unset or no wind update received in 60 sec -> supressed rrd update
WARNING:__main__:CRON10: clock unset or no wind update received in 60 sec -> supressed rrd update
WARNING:__main__:CRON10: clock unset or no wind update received in 60 sec -> supressed rrd update

How can I use the Python logging module to suppress repeating messages and output something more rsyslog style (http://www.rsyslog.com/doc/rsconf1_repeatedmsgreduction.html):

WARNING:__main__:CRON10: clock unset or no wind update received in 60 sec -> supressed rrd update
--- The last message repeated 3 times

Is there a way to extend logging or do I have to write a completly own logger?

The code I use for logging is:

logging.basicConfig(format='%(asctime)s %(message)s')
logging.basicConfig(level=logging.info)
logger = logging.getLogger(__name__)
hdlr = logging.FileHandler(LOGFILE)
hdlr.setFormatter(formatter)
logger.addHandler(hdlr)

Any ideas on that?

Possible duplicate of How to avoid duplicate outputs using logging module — Anthon
– Anthon, Commented Jun 22, 2017 at 6:27
@Anthon: The linked dupe is about the results of registering multiple handlers. That's not the problem described here. — SiHa
– SiHa, Commented Jun 22, 2017 at 7:04

zwer · Accepted Answer · 2017-06-22 07:49:58Z

33

You can create a logging.Filter that will keep track of the last logged record and filter out any repeated (similar) records, something like:

import logging

class DuplicateFilter(logging.Filter):

    def filter(self, record):
        # add other fields if you need more granular comparison, depends on your app
        current_log = (record.module, record.levelno, record.msg)
        if current_log != getattr(self, "last_log", None):
            self.last_log = current_log
            return True
        return False

Then just add it to the logger/handler you use (i.e. hdlr.addFilter(DuplicateFilter())) or the root logger to filter all default logs. Here's a simple test:

import logging

logging.warn("my test")
logging.warn("my repeated test")
logging.warn("my repeated test")
logging.warn("my repeated test")
logging.warn("my other test")

logger = logging.getLogger()  # get the root logger
logger.addFilter(DuplicateFilter())  # add the filter to it

logging.warn("my test")
logging.warn("my repeated test")
logging.warn("my repeated test")
logging.warn("my repeated test")
logging.warn("my other test")

This will print out:

WARNING:root:my test
WARNING:root:my repeated test
WARNING:root:my repeated test
WARNING:root:my repeated test
WARNING:root:my other test
WARNING:root:my test
WARNING:root:my repeated test
WARNING:root:my other test

edited Jun 22, 2017 at 7:49

answered Jun 22, 2017 at 6:52

zwer

25.9k3 gold badges54 silver badges70 bronze badges

Sign up to request clarification or add additional context in comments.

2 Comments

mvds Over a year ago

CAVEAT: if logging.*() is used with format string arguments (e.g. logging.warn("some error: %s",errstr)) the DuplicateFilter will treat any such line as duplicate, as the format string is checked - not the printed string.

Jorrit Smit Over a year ago

This can be circumvented by calling record.getMessage()

Gregor · Accepted Answer · 2022-05-28 09:49:50Z

Another option is to overwrite the logging.Logger class in order to modify the _log function:

import logging
import zlib


class FilteredLogger(logging.Logger):

    def __init__(self, name, level=logging.NOTSET):
        super().__init__(name, level)
        self._message_lockup = {}

    def _log(self, level, msg, args, exc_info=None, extra=None, stack_info=False, log_intervall=None):
        if log_intervall is None or log_intervall == 1:
            super(FilteredLogger, self)._log(level, msg, args, exc_info, extra, stack_info)
        else:
            message_Id = zlib.crc32(msg.encode('utf-8'))
            if message_Id not in self._message_lockup:
                self._message_lockup[message_Id] = 0
                super(FilteredLogger, self)._log(level, msg, args, exc_info, extra, stack_info)
            elif self._message_lockup[message_Id] % log_intervall == 0:
                msg += f' -- Suppressed {log_intervall} equal messages'
                super(FilteredLogger, self)._log(level, msg, args, exc_info, extra, stack_info)
            self._message_lockup[message_Id] += 1

if __name__ == '__main__':
    logging.setLoggerClass(FilteredLogger)
    logger = logging.getLogger('test_logger')
    consol_handler = logging.StreamHandler()
    logger.addHandler(consol_handler)
    logger.setLevel(logging.DEBUG)
    for count in range(1000):
        logger.info('Test log', log_intervall=100)

The output looks like this:

Test log
Test log -- Suppressed 100 equal messages
Test log -- Suppressed 100 equal messages
Test log -- Suppressed 100 equal messages
Test log -- Suppressed 100 equal messages
Test log -- Suppressed 100 equal messages
Test log -- Suppressed 100 equal messages
Test log -- Suppressed 100 equal messages
Test log -- Suppressed 100 equal messages
Test log -- Suppressed 100 equal messages

This implementation can easily be customized.

Collectives™ on Stack Overflow

Suppress multiple messages with same content in Python logging module AKA log compression

2 Answers 2

2 Comments

Comments

Your Answer

Linked

Hot Network Questions

Collectives™ on Stack Overflow

2 Answers 2

2 Comments

Comments

Your Answer

Sign up or log in

Post as a guest

Linked

Related