Øyvind Monsen
11/06/2025, 4:03 PMMarvin
11/06/2025, 4:03 PMMarvin
11/06/2025, 4:09 PMlogging.getLogger(...) everywhere and have Prefect “sniff” those logs so they show up on the right flow/task in the UI.
Recommended approach (no code changes):
- Add your application and third‑party logger names to the PREFECT_LOGGING_EXTRA_LOGGERS setting. During a flow/task run, Prefect will attach its API log handler to those loggers, pick up their records, and associate them with the active flow/task run automatically.
- This lets you keep your existing handlers (e.g., Grafana/Loki) and just add Prefect as another sink.
Example
1) Use your existing loggers in code:
import logging
from prefect import flow, task
app_logger = logging.getLogger("myapp") # your existing logger
@task
def do_work():
<http://app_logger.info|app_logger.info>("Task did some work")
return 42
@flow(log_prints=True) # optional: capture print() as logs
def main():
app_logger.warning("Flow starting")
result = do_work()
print("this print will be captured if log_prints=True")
return result
if __name__ == "__main__":
main()
2) Tell Prefect to capture that logger name (and any others you care about):
export PREFECT_LOGGING_EXTRA_LOGGERS=myapp,urllib3,boto3
# set this in your worker/pod/container environment, or Prefect profile
Notes and best practices
- Choose the top‑level logger names for broad coverage. For example, listing myapp will typically capture myapp.api, myapp.db, etc. (children usually propagate to the parent by default).
- Keep your existing Grafana/Loki handler(s); Prefect’s API handler is independent and won’t interfere. If you see duplicates in your own logs, check:
- Whether your child loggers and root logger both have Loki handlers (set propagate=False on child loggers if needed).
- Avoid adding the same handler twice.
- Useful settings:
- PREFECT_LOGGING_LEVEL and PREFECT_LOGGING_ROOT_LEVEL to control verbosity
- PREFECT_LOGGING_EXTRA_LOGGERS to list extra loggers to capture
- PREFECT_LOG_PRINTS=true to capture print() in flows/tasks
- Advanced: point to a custom logging config via PREFECT_LOGGING_SETTINGS_PATH if you want full control (handlers/formatters/filters in YAML).
Alternative (code-based, if you want to be explicit)
Attach Prefect’s API handler to your logger(s) at runtime:
import logging
from prefect.logging.handlers import APILogHandler
def attach_prefect_handler(logger_name="myapp"):
logger = logging.getLogger(logger_name)
if not any(isinstance(h, APILogHandler) for h in logger.handlers):
logger.addHandler(APILogHandler())
Call this inside your flow/task (so a run context exists), or during startup for workers. The APILogHandler will associate records to the active run automatically.
When to use get_run_logger
- get_run_logger() is still the most direct way to emit Prefect‑annotated logs in flows/tasks, but it’s optional if you configure your existing loggers as above.
Docs and references
- How to add logging to workflows: Add logging (Prefect 3)
- Advanced logging customization (handlers/formatters/filters, YAML): Advanced logging customization
- API reference for loggers/handlers: prefect.logging.loggers and prefect.logging.handlers
If you can share:
- The logger names you use (e.g., a top‑level like mycompany),
- Where your flows/tasks run (local, Docker, Kubernetes),
I can suggest the exact PREFECT_LOGGING_EXTRA_LOGGERS value and the best place to set it (profile vs. work pool/worker env).Øyvind Monsen
11/06/2025, 4:15 PMMarvin
11/06/2025, 4:15 PMMarvin
11/06/2025, 4:16 PMlogging.getLogger(...) (or Prefect’s prefect.logging.get_logger(...)). Add those names—ideally the top‑level ones—to PREFECT_LOGGING_EXTRA_LOGGERS so Prefect attaches its API handler and captures their records during a run.
Example:
- Your code: logging.getLogger("myapp.api")
- Set: PREFECT_LOGGING_EXTRA_LOGGERS=myapp
Notes:
- Child loggers propagate to their parent by default, so listing myapp typically captures myapp.api, myapp.db, etc.
- If a library sets propagate=False, add that exact logger name (e.g., some.lib.logger) or attach APILogHandler programmatically.
- This won’t affect your existing Grafana/Loki handlers; it just adds Prefect as another sink.
Refs:
- Add logging (Prefect 3)
- Advanced logging customizationØyvind Monsen
11/06/2025, 4:17 PMMarvin
11/06/2025, 4:17 PMMarvin
11/06/2025, 4:17 PMlogging.getLogger(__name__) in each module:
- The logger name becomes the module’s dotted path (e.g., myapp.ingest.loader).
- You only need to add the top-level package once to PREFECT_LOGGING_EXTRA_LOGGERS (e.g., myapp) and Prefect will capture all child module logs via propagation.
Examples
- Package code:
- In `myapp/ingest/loader.py`: logging.getLogger(__name__) -> myapp.ingest.loader
- Set once: PREFECT_LOGGING_EXTRA_LOGGERS=myapp
- Script entrypoint run directly:
- In main.py run as `python main.py`: logger name is __main__
- Options:
- Add __main__ to PREFECT_LOGGING_EXTRA_LOGGERS (e.g., myapp,__main__), or
- Run as a module so the name is your package path: python -m myapp.main, or
- Use an explicit app-level logger in the script: logging.getLogger("myapp")
Notes
- If a library or your logger sets propagate=False, include that exact logger name in PREFECT_LOGGING_EXTRA_LOGGERS or attach APILogHandler programmatically.
- You can list multiple prefixes: PREFECT_LOGGING_EXTRA_LOGGERS=myapp,otherpkg,urllib3.
Refs
- Add logging (Prefect 3)
- Advanced logging customizationØyvind Monsen
11/06/2025, 4:18 PMMarvin
11/06/2025, 4:18 PMMarvin
11/06/2025, 4:19 PMPREFECT_LOGGING_EXTRA_LOGGERS in the job/container env (e.g., work pool default job variables in the UI, job_variables on your deployment, or bake it into the image).
- Subprocess worker: the flow runs in a subprocess of the worker. Set it on the worker process (it will inherit) or set env for the job.
- Local/manual runs: set it in your shell or Prefect profile before running the flow.
Setting this on the Prefect server has no effect; the server only receives logs that the run process emits.
Example: set via deployment
from prefect import flow
@flow
def my_flow():
...
my_flow.from_source("<https://github.com/org/repo>", entrypoint="flows/my_flow.py:my_flow").deploy(
name="prod",
work_pool_name="k8s-pool",
job_variables={
"env": {
"PREFECT_LOGGING_EXTRA_LOGGERS": "myapp,urllib3,boto3",
"PREFECT_LOGGING_LEVEL": "INFO",
}
},
)
Docs
- Add logging: Add logging (Prefect 3)
- Advanced logging customization: Advanced logging customizationØyvind Monsen
11/06/2025, 4:56 PMMarvin
11/06/2025, 4:56 PMMarvin
11/06/2025, 4:56 PMPREFECT_LOGGING_EXTRA_LOGGERS works, but a few common gotchas can prevent capture.
Quick checklist
1) Emit logs inside an active flow/task run
- Prefect only associates logs during a run context. Import-time or “before/after the flow” logs won’t show up in the UI.
- Sanity check: do get_run_logger().info("hello from run logger") in the same spot — if that appears, the run context is fine.
2) Logger name matches exactly
- If you use logging.getLogger(__name__) in scripts run as python script.py, the logger name is __main__. Add __main__ to PREFECT_LOGGING_EXTRA_LOGGERS or run as a module (e.g., python -m mypkg.script) or use a top-level logger like logging.getLogger("mypkg").
- Avoid spaces in the env var; use a comma-separated list like myapp,urllib3,boto3.
3) Levels aren’t filtering your logs
- Set PREFECT_LOGGING_LEVEL=DEBUG (and/or your app logger level to DEBUG) to test.
4) Your logging config isn’t removing Prefect’s handler
- If you call logging.config.dictConfig, ensure disable_existing_loggers: False. Otherwise it can wipe Prefect’s handlers.
- Don’t replace handlers on your logger mid-run (e.g., logger.handlers = [...]).
- Keep logger.propagate=True unless you explicitly attach APILogHandler to that specific logger.
5) Environment is set where the code runs
- Set the env var on the container/job that executes your flow code (e.g., work pool job variables, deployment job_variables.env, or Docker image env). Setting it on the Prefect server does nothing.
- For subprocess workers, ensure the child process inherits the env.
6) Frameworks that reconfigure logging
- Gunicorn/uvicorn and some libraries reset logging — they can clobber handlers after Prefect attaches them. If you use them, ensure your config preserves existing handlers or explicitly attach Prefect’s handler.
Two quick in-task tests
- Verify the handler got attached to your logger:
import logging
from prefect import task
@task
def debug_handlers():
log = logging.getLogger("myapp") # use the same name you expect to capture
print("logger name:", log.name)
print("effective level:", log.getEffectiveLevel())
print("handlers:", [type(h).__name__ for h in log.handlers])
- Force-attach Prefect’s handler and emit a log:
import logging
from prefect import task
from prefect.logging.handlers import APILogHandler
@task
def force_attach_and_log():
log = logging.getLogger("myapp")
if not any(isinstance(h, APILogHandler) for h in log.handlers):
log.addHandler(APILogHandler())
log.warning("Hello from myapp via APILogHandler")
If this shows up, the env-based auto-attach path is being overridden somewhere (likely your logging config).
If you can share:
- How you run the flow (local, Docker, K8s; which work pool/worker type),
- The exact value of PREFECT_LOGGING_EXTRA_LOGGERS,
- How you obtain your logger (e.g., logging.getLogger(__name__) or a fixed name),
- Whether get_run_logger() logs appear,
- Any logging config (basicConfig/dictConfig) used,
I can pinpoint the issue and suggest the exact fix.
Docs
- Add logging: Add logging (Prefect 3)
- Advanced logging customization: Advanced logging customization