Ben Griffiths
10/21/2025, 2:23 PMPREFECT_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.Marvin
10/21/2025, 2:23 PMMarvin
10/21/2025, 2:29 PMdeployments:
- 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
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
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:
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 & 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.Ben Griffiths
10/21/2025, 2:32 PMMarvin
10/21/2025, 2:32 PMMarvin
10/21/2025, 2:33 PMimport 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.
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:
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 & profilesBen Griffiths
10/21/2025, 2:55 PMattach_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?Marvin
10/21/2025, 2:55 PMMarvin
10/21/2025, 2:55 PMimport 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