<@ULVA73B9P> what is the best way to have prefect ...
# ask-marvin
ø
@Marvin what is the best way to have prefect pick up on logs using logging.get_logger in python? We use that everywhere and also send logs to grafana. Ideally prefect would sniff them up and show them for the appropriate flow / task
m
thought for 319.5 seconds
Great question. In Prefect 3.x, you can keep using the standard Python
logging.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:
Copy 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):
Copy code
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:
Copy code
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).
ø
@Marvin so the name should be what I pass as name to get_logger?
m
thought for 29.7 seconds
Short answer: yes. Use the same logger names you pass to
logging.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
ø
@Marvin what if we just pass _`__name__`_ to the name?
m
thought for 21.5 seconds
Yes—that’s a good pattern. If you do
logging.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
ø
@Marvin so PREFECT_LOGGING_EXTRA_LOGGERS should be set on the container instances running the flow code? or on our prefect_worker or server?
m
thought for 39.8 seconds
Set it in the environment where your flow/task code actually runs. - Docker/Kubernetes runs: set
PREFECT_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
Copy code
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
ø
@Marvin I have set the extra loggers env var, but still, no logs are comming through
m
thought for 43.8 seconds
Got it — let’s troubleshoot. In Prefect 3.x,
PREFECT_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:
Copy code
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:
Copy code
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