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

If you are on server, you can very simply achieve this by setting:

prefect config set PREFECT_LOGGING_EXTRA_LOGGERS="flow-logger"
prefect config set PREFECT_LOGGING_LEVEL="WARNING"

And then, in your flows:

from logging import getLogger

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

Thanks! Logging levels are working fine, I was talking specifically about filtering specific substrings from loggers and handlers

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. 🤷‍♂️

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:

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

Looks great! My suggestion was actually just for getting rid of the logs about tasks stage changes only.

@Tony Zeljkovic how did you apply that to the root logger in a run/deployment?

just put the function in the flow.

How did you specify the root level loggers? I can't seem to hide messages from those

root? Not sure, our issue wasn't with the root level loggers