https://prefect.io logo
t

Tony Zeljkovic

08/29/2023, 2:01 PM
Does anyone here have experience with applying filters in logging.yml? The pattern here doesn't work for me and i'm not sure if it's my local setup or some breaking change since then: https://github.com/PrefectHQ/prefect/issues/5952
s

Sanz Al

08/29/2023, 2:59 PM
If you are on server, you can very simply achieve this by setting:
Copy code
prefect config set PREFECT_LOGGING_EXTRA_LOGGERS="flow-logger"
prefect config set PREFECT_LOGGING_LEVEL="WARNING"
And then, in your flows:
Copy code
from logging import getLogger

log = getLogger("flow-logger")
log.setLevel("INFO")
t

Tony Zeljkovic

08/29/2023, 3:11 PM
Thanks! Logging levels are working fine, I was talking specifically about filtering specific substrings from loggers and handlers
s

Sanz Al

08/29/2023, 3:19 PM
With that set up, you leave the default prefect logger only for the warning level up and assign the info level to a custom logger. This way you won’t see the “task_created…” and so on in your flows logs. 🤷‍♂️
t

Tony Zeljkovic

08/29/2023, 3:36 PM
hmm, interesting, I tried going this route but somehow it didn't really work for me. I did write a piece of code which can be used to apply specific substring filters on specific loggers on either the flow or task level:
Copy code
import logging
import functools


class BaseFilter(logging.Filter):
    """
    Base class for filters. Filters are used to filter out log messages of specific loggers.
    Check documentation for more config details: <https://docs.python.org/3/library/logging.html#logging.Filter>
    The default behavior is to filter out all log messages that contain any of the strings in the list except when logging level is set to DEBUG.

    attributes:
        strings_to_filter: list of SUBstrings to filter out. This means that any log message that contains any of the strings in the list will be filtered out.
        loggers_to_filter: list of loggers to filter these messages from.
    """
    def __init__(self, strings_to_filter=None):
        super().__init__()
        if strings_to_filter is None:
            self.strings_to_filter = []
        else:
            self.strings_to_filter = [string.lower() for string in strings_to_filter]

    def filter(self, rec):
        # If logging level is set to DEBUG, do not filter
        if rec.levelno == logging.DEBUG:
            return True

        msg = str(rec.msg)  # Convert to string to be safe
        return not any(string in msg.lower() for string in self.strings_to_filter)

def log_filter(filters=None):
    """ Decorator to apply filters to loggers. To be applied to either a flow or a task.

    Args:
        filters: list of filter classes to apply. Each filter class must inherit from BaseFilter.

    """
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            if filters is not None:
                for filter_class in filters:
                    for logger_name in filter_class().loggers_to_filter:
                        logging.getLogger(logger_name).addFilter(filter_class())
            return func(*args, **kwargs)
        return wrapper
    return decorator
or you know, something like that 😄
s

Sanz Al

08/29/2023, 3:40 PM
Looks great! My suggestion was actually just for getting rid of the logs about tasks stage changes only.
🙂 1
i

Idan

09/20/2023, 1:12 PM
@Tony Zeljkovic how did you apply that to the root logger in a run/deployment?
t

Tony Zeljkovic

09/20/2023, 4:13 PM
just put the function in the flow.
In the end I rewrote it as a regular function
i

Idan

09/20/2023, 4:18 PM
How did you specify the root level loggers? I can't seem to hide messages from those
t

Tony Zeljkovic

09/20/2023, 5:05 PM
root? Not sure, our issue wasn't with the root level loggers