<@ULVA73B9P> I want to display logging messages fr...
# ask-marvin
b
@Marvin I want to display logging messages from a custom Python package which is used in my flows and tasks. I am running Prefect locally in docker and I have set the
PREFECT_LOGGING_EXTRA_LOGGERS: mypackage
environment variable on the Prefect worker service. However, none of the logging messages emitted by
mypackage
are showing up in the UI.
m
thought for 331.5 seconds
Likely cause: with a Docker work pool, your flows run in separate containers. Setting PREFECT_LOGGING_EXTRA_LOGGERS only on the worker service won’t affect the flow-run containers, so logs from your package won’t be captured by the Prefect APILogHandler and won’t reach the UI. In Prefect 3.x, set PREFECT_LOGGING_EXTRA_LOGGERS inside the flow-run container. You can do that in any of these ways: - In your deployment’s job variables (recommended) - Bake it into the image used for the flow - Set it in a profile/env file that’s present at runtime in the flow container Examples 1) prefect.yaml (Docker work pool)
Copy code
deployments:
  - name: my-deployment
    entrypoint: flows/my_flow.py:my_flow
    work_pool:
      name: my-docker-pool
      job_variables:
        env:
          PREFECT_LOGGING_EXTRA_LOGGERS: "mypackage"
          # optionally raise level if your package logs at INFO/DEBUG
          PREFECT_LOGGING_LEVEL: "INFO"
2) In code when deploying
Copy code
from prefect import flow

@flow
def my_flow():
    ...

flow.from_source(".").deploy(
    name="my-deployment",
    work_pool_name="my-docker-pool",
    job_variables={
        "env": {
            "PREFECT_LOGGING_EXTRA_LOGGERS": "mypackage",
            "PREFECT_LOGGING_LEVEL": "INFO",
        }
    },
)
3) Bake into the image used by the Docker work pool
Copy code
ENV PREFECT_LOGGING_EXTRA_LOGGERS="mypackage"
ENV PREFECT_LOGGING_LEVEL="INFO"
Common gotchas to check - Logger location: The value must match the logger’s name. If your package uses loggers like logging.getLogger("mypackage.submodule"), adding "mypackage" is sufficient. If it uses a different root (e.g., "mycompany.mypackage"), add that instead. - Level/propagation: Extra loggers are attached to Prefect’s APILogHandler but their level/propagation still matters. Ensure your package’s logger is not set to propagate=False and is emitting at a level >= the handler level (try INFO). - Timing/context: Only logs emitted during a flow/task run (with a flow_run_id/task_run_id) go to the UI. Import-time logs before the run starts won’t be captured. - Message size: Single messages > ~1MB are dropped. - Verify from inside the flow container: Add a quick check in a task:
Copy code
from prefect.settings import get_current_settings
  import logging

  @task
  def check_logging():
      s = get_current_settings()
      print("extra_loggers:", s.logging.extra_loggers)
      logging.getLogger("mypackage").info("Test log from mypackage")
If you can share: - Are you using a Docker work pool? - How is your deployment defined (prefect.yaml or deploy code)? - The image your flow runs in and whether the env var is present there (docker exec env inside the running flow container) Docs and references - Advanced logging customization: Prefect 3.x logging customization - Settings and profiles: Settings &amp; profiles - Customize job variables: Customize job variables - Docker deployment guide: Docker deployments If you move the PREFECT_LOGGING_EXTRA_LOGGERS setting into the flow-run container (via job_variables or the image), your package logs should start appearing in the UI.
b
@Marvin How can I set the level of the Prefect logger and extra logger within a flow, so that I can set the logging level as a flow parameter
m
thought for 48.0 seconds
Yes—assuming Prefect 3.x, you can set both the Prefect logger level and your extra logger(s) at runtime from a flow parameter. There are two good patterns: Pattern A (recommended): reconfigure Prefect’s logging with temporary settings - This updates Prefect’s logging config so APILogHandler is attached to your extra loggers and the level is applied globally.
Copy code
import logging
from typing import Sequence
from prefect import flow, task
from prefect.logging import get_run_logger
from prefect.settings import temporary_settings, PREFECT_LOGGING_LEVEL, PREFECT_LOGGING_EXTRA_LOGGERS
from prefect.logging.configuration import setup_logging

def normalize_level(level_str: str) -> str:
    # Accepts "debug", "INFO", etc., returns standard name like "DEBUG"
    level_str = (level_str or "INFO").upper()
    return level_str if level_str in logging._nameToLevel else "INFO"

@flow
def my_flow(log_level: str = "INFO", extra_loggers: Sequence[str] = ("mypackage",)):
    # Apply logging config at the very start of the flow
    level = normalize_level(log_level)
    extra = ",".join(extra_loggers) if extra_loggers else ""

    with temporary_settings({
        PREFECT_LOGGING_LEVEL: level,
        PREFECT_LOGGING_EXTRA_LOGGERS: extra,
    }):
        # Re-apply logging config so changes take effect in this process
        setup_logging()

        logger = get_run_logger()
        <http://logger.info|logger.info>(f"Runtime logging set to {level}; extra_loggers={extra or '(none)'}")

        # ...rest of your flow...
Notes: - Put this at the top of the flow so logs after this point respect your settings. - This will attach Prefect’s APILogHandler to the given extra logger names for UI delivery. Pattern B: set levels directly on loggers (if handler already attached) - Use this if you’ve already configured
PREFECT_LOGGING_EXTRA_LOGGERS
via environment/image/job variables (so handlers are attached), and you just want to change levels at runtime.
Copy code
import logging
from typing import Sequence
from prefect import flow
from prefect.logging import get_run_logger

def set_runtime_levels(level_name: str, extra_loggers: Sequence[str]):
    level = logging._nameToLevel.get(level_name.upper(), <http://logging.INFO|logging.INFO>)

    # Prefect loggers commonly used during runs
    for name in ("prefect", "prefect.flow_runs", "prefect.task_runs"):
        logging.getLogger(name).setLevel(level)

    # Your extra loggers
    for name in extra_loggers:
        lg = logging.getLogger(name)
        lg.setLevel(level)
        lg.propagate = True  # ensure messages reach Prefect handlers

@flow
def my_flow(log_level: str = "INFO", extra_loggers: Sequence[str] = ("mypackage",)):
    set_runtime_levels(log_level, extra_loggers)
    get_run_logger().info("Logger levels updated at runtime")
    # ...rest of your flow...
If handlers are not already attached to your extra loggers and you don’t want to use Pattern A, you can programmatically attach APILogHandler:
Copy code
from prefect.logging.handlers import APILogHandler

def attach_api_handler(logger_name: str):
    lg = logging.getLogger(logger_name)
    if not any(isinstance(h, APILogHandler) for h in lg.handlers):
        lg.addHandler(APILogHandler())
    lg.propagate = True
Practical tips - Only logs emitted after the reconfiguration will reflect the new level. - Make sure your extra logger names match exactly (e.g.,
mypackage
vs
mycompany.mypackage
). - Keep this configuration at the start of the flow to affect all subsequent tasks. Docs for reference - Logging customization: Prefect 3.x logging customization - Settings and profiles: Settings &amp; profiles
b
@Marvin If I use pattern B, and apply
attach_api_handler
function on each logger passed to the
extra_loggers
argument passed to
set_runtime_levels
, do I still need to set the PREFECT_LOGGING_EXTRA_LOGGERS environment variable?
m
thought for 29.9 seconds
Short answer: No. If you programmatically attach Prefect’s APILogHandler to your extra loggers at runtime, you do not need to set PREFECT_LOGGING_EXTRA_LOGGERS. That env var is just a convenient way for Prefect to attach the handler for you during setup. Recommended pattern when doing it manually: - Do it early in the flow run (before emitting logs) - Avoid double attachment if you also set the env var (you’ll get duplicate logs) - Ensure propagate=True and set an appropriate level Example
Copy code
import logging
from prefect import flow
from prefect.logging import get_run_logger
from prefect.logging.handlers import APILogHandler

def ensure_api_handler(logger_name: str):
    lg = logging.getLogger(logger_name)
    if not any(isinstance(h, APILogHandler) for h in lg.handlers):
        lg.addHandler(APILogHandler())
    lg.propagate = True

def set_runtime_levels(level_name: str, extra_loggers):
    level = logging._nameToLevel.get(level_name.upper(), <http://logging.INFO|logging.INFO>)

    # Prefect loggers
    for name in ("prefect", "prefect.flow_runs", "prefect.task_runs"):
        logging.getLogger(name).setLevel(level)

    # Extra loggers
    for name in extra_loggers:
        ensure_api_handler(name)
        logging.getLogger(name).setLevel(level)

@flow
def my_flow(log_level: str = "INFO", extra_loggers=("mypackage",)):
    set_runtime_levels(log_level, extra_loggers)
    get_run_logger().info("Logger levels and handlers set at runtime")
    # your tasks here...
Notes and gotchas: - Only logs emitted during a flow/task run (when a run id is present) will appear in the UI. - If you’re using Dask/Ray/multiprocessing, attach handlers inside each worker/process (handlers don’t always propagate across process boundaries). - Don’t combine manual attachment with PREFECT_LOGGING_EXTRA_LOGGERS unless you check for existing APILogHandler to avoid duplicates. Docs for reference: - Prefect 3.x logging customization