Logging (no repeat and return all at end)

[wolfiex]

As discussed at the start of meeting 34, the code to carry out the desired function is provided below, and can be integrated into the API as needed.

import logging



def reset_loggers():
    """Remove all handlers from all loggers."""
    # Reset root logger
    root = logging.getLogger()
    for handler in root.handlers[:]:
        root.removeHandler(handler)
    root.handlers.clear()

    # Reset all existing named loggers
    loggers = logging.root.manager.loggerDict
    for name, logger in loggers.items():
        if isinstance(logger, logging.Logger):
            logger.handlers.clear()
            logger.filters.clear()
            logger.propagate = True

# Call this early in your script
reset_loggers()





class UniqueErrorFilter(logging.Filter):
    def __init__(self):
        super().__init__()
        self.seen_messages = set()
        self.history = []

    def filter(self, record):
        # Only allow error messages that haven't been seen before
        # print(self.seen_messages)
        self.history.append(record.msg%(record.args))
        if record.levelno >= logging.ERROR:
            msg_id = (record.msg, record.args)
            if msg_id in self.seen_messages:
                return False
            self.seen_messages.add(msg_id)
        return True

# Setup logger
logger = logging.getLogger("DRLogger")
logger.setLevel(logging.ERROR)
# Stop messages going to the root logger
logger.propagate = False
handler = logging.StreamHandler()
handler.addFilter(UniqueErrorFilter())  # Add the filter here
formatter = logging.Formatter('%(levelname)s - %(message)s')
handler.setFormatter(formatter)
# logger.addHandler(handler)
# Ensure only one handler is added
if not logger.hasHandlers():
    logger.addHandler(handler)


# Example usage
logger.error("Something went wrong: %s", "disk full")
logger.error("Something went wrong: %s", "disk full")  # Will be suppressed
logger.error("Something went wrong: %s", "disk full")  # Will be suppressed
logger.error("Something went wrong: %s", "network down")  # Will be shown


print('\n\n\n','-'*50)


# Get all messages as parsed by the filter
all_uef = list(filter(lambda x: isinstance(x, UniqueErrorFilter),handler.filters))
messages = all_uef[0].seen_messages
history = "\n - ".join(all_uef[0].history)

print(f'A unique collection of all the errors recorded:\n {messages}')
print(f'A complete collection of all the errors recorded:\n - {history}')

Should produce output in the from:

ERROR - Something went wrong: disk full
ERROR - Something went wrong: network down

---------------------------------------------------

A unique collection of all the errors recorded:
  {('Something went wrong: %s', ('network down',)), ('Something went wrong: %s', ('disk full',))}
A complete collection of all the errors recorded:
 - Something went wrong: disk full
 - Something went wrong: disk full
 - Something went wrong: disk full
 - Something went wrong: network down

[wolfiex]

If we are expecting a very large number of non duplicate warnings, it may also be a good idea to log these to file rather than sdout/in-memory for debugging (e.g. if the script crashes).

A simple solution would be to turn on the propagation to root logger on again, and redirect that to a temp file.

[wolfiex]

@JamesAnstey

setup

import logging

logger = logging.getLogger("global")
if not logger.hasHandlers():
    handler = logging.StreamHandler()
    handler.setFormatter(logging.Formatter('%(levelname)s - %(message)s'))
    logger.addHandler(handler)
logger.setLevel(logging.INFO)

# set level
  logger.setLevel(getattr(logging, level.upper(), logging.INFO))

files

from logger_config import logger

logger.debug("Debug message")
logger.info("Info message")

[mpmoine]

from TISG meeting 38:

• DE's configured logger to be used in all pieces of the code.

[JamesAnstey]

Started on this in #153

Feedback would be useful before I keep going with this - is it on the right track? @wolfiex @rigoudyg @sol1105