Hi! How can I help with Prefect today?

Short answer: use Python logging handlers. Prefect’s loggers are standard logging loggers; the Prefect UI is powered by Prefect’s APILogHandler. To send the same logs to multiple places, attach multiple handlers (console, API, file, syslog, CloudWatch, etc.) to the Prefect logger. Two good ways to do it in Prefect 3.x: 1) Configure via a logging YAML (recommended) - Create a dictConfig YAML that includes Prefect’s APILogHandler for the UI plus whatever extra handlers you need. - Point Prefect to it with

PREFECT_LOGGING_SETTINGS_PATH

(or set it in your Prefect profile). Example logging.yml

version: 1
disable_existing_loggers: false

formatters:
  standard:
    format: "%(asctime)s | %(levelname)s | %(name)s | %(message)s"
  json:
    "()": "prefect.logging.formatters.JsonFormatter"

handlers:
  console:
    class: "prefect.logging.handlers.PrefectConsoleHandler"
    level: "INFO"
    formatter: "standard"

  api:
    class: "prefect.logging.handlers.APILogHandler"
    level: "INFO"
    formatter: "json"

  file:
    class: "logging.handlers.RotatingFileHandler"
    level: "INFO"
    formatter: "standard"
    filename: "/var/log/prefect/flow.log"
    maxBytes: 10485760
    backupCount: 5

  syslog:
    class: "logging.handlers.SysLogHandler"
    level: "INFO"
    formatter: "standard"
    # For Linux socket: address: "/dev/log"
    address: ["localhost", 514]

loggers:
  prefect:
    level: "INFO"
    handlers: ["console", "api", "file", "syslog"]
    propagate: false

root:
  level: "WARNING"
  handlers: ["console"]

Point Prefect to this file - One-off in your shell:

export PREFECT_LOGGING_SETTINGS_PATH=/path/to/logging.yml

- Persist in your Prefect profile:

prefect config set PREFECT_LOGGING_SETTINGS_PATH=/path/to/logging.yml

2) Add a handler programmatically (quick and simple)

import logging
from logging.handlers import RotatingFileHandler
from prefect import flow
from prefect.logging import get_run_logger

@flow
def my_flow():
    # Add a file handler to the Prefect logger (once per process)
    file_handler = RotatingFileHandler("flow.log", maxBytes=10_000_000, backupCount=3)
    file_handler.setFormatter(logging.Formatter("%(asctime)s | %(levelname)s | %(name)s | %(message)s"))
    logging.getLogger("prefect").addHandler(file_handler)

    logger = get_run_logger()
    <http://logger.info|logger.info>("This goes to the Prefect UI and flow.log (and console if enabled)")

my_flow()

Notes and tips - Keep APILogHandler: If you want logs in the Prefect UI, make sure the

prefect

logger includes

prefect.logging.handlers.APILogHandler

(as above). Logs are only sent to the UI when a flow/task is running with an active Prefect API context (e.g., in Prefect Cloud or Prefect Server). - Be precise with handler placement: Attaching handlers to the

prefect

logger (and setting

propagate: false

) avoids duplicates. If you attach to both

root

and

prefect

, you may double-log. - Third-party destinations: You can add any stdlib-compatible handler. For example: - CloudWatch (via watchtower):

handlers:
      cloudwatch:
        class: "watchtower.CloudWatchLogHandler"
        level: "INFO"
        formatter: "json"
        log_group: "/prefect/flows"
        stream_name: "{logger_name}"
        create_log_group: true
    loggers:
      prefect:
        handlers: ["console", "api", "cloudwatch"]
        level: "INFO"
        propagate: false

Make sure watchtower/boto3 are installed and credentials are available. - Syslog: use

logging.handlers.SysLogHandler

as shown in the YAML above. - Use get_run_logger in flows/tasks: Always log with

get_run_logger()

inside flows/tasks so messages include run context and are eligible for UI delivery. - Persist settings in profiles: Use

prefect config set

to make your logging config the default for your environment. Docs and references - Prefect 3 - Logging customization

- APILogHandler and other logging handlers - Logging settings reference If you share which destinations you need (e.g., Datadog, Splunk, ELK, CloudWatch), I can drop in an exact YAML snippet for that handler.

Short answer: use Prefect’s run logger (get_run_logger) for UI logs, then add standard Python logging handlers for any extra destinations. You can do this per-run in code or globally via a logging config so you don’t duplicate handlers. How logs reach the Prefect UI - Inside a flow/task run, get_run_logger returns a logger that already includes an APILogHandler which ships logs to the Prefect API/UI. You just call logger.info(...) and they appear in the UI. - If you run code outside a Prefect run context (e.g., a plain script), UI shipping won’t occur. Option A: Add an extra destination programmatically (per run)

from prefect import flow, task
from prefect.logging import get_run_logger
import logging
from logging.handlers import RotatingFileHandler

def add_file_handler_once(run_logger, path="/tmp/prefect-flow.log"):
    # get_run_logger may return a LoggingAdapter; grab the base logger
    base = getattr(run_logger, "logger", run_logger)

    # avoid adding the same handler repeatedly
    if not any(isinstance(h, RotatingFileHandler) and getattr(h, "_pf_path", None) == path for h in base.handlers):
        fh = RotatingFileHandler(path, maxBytes=5_000_000, backupCount=3, encoding="utf-8")
        fh.setLevel(<http://logging.INFO|logging.INFO>)
        fh.setFormatter(logging.Formatter("%(asctime)s | %(levelname)s | %(name)s | %(message)s"))
        fh._pf_path = path  # mark to detect duplicates
        base.addHandler(fh)

@task
def work():
    logger = get_run_logger()
    add_file_handler_once(logger)  # UI + console (default) + file
    <http://logger.info|logger.info>("This goes to the Prefect UI and the file.")

@flow
def my_flow():
    work()

if __name__ == "__main__":
    my_flow()

Option B: Add destinations globally with a logging config (recommended for teams) - Let Prefect initialize its defaults (UI + console), then append your handlers using Python’s dictConfig in “incremental” mode. This avoids duplicate handlers. Minimal example (append a file handler to the Prefect logger):

import logging.config
from prefect.logging.configuration import setup_logging

LOGGING_UPDATE = {
    "version": 1,
    "incremental": True,  # only apply changes, keep Prefect’s defaults
    "handlers": {
        "file": {
            "class": "logging.handlers.RotatingFileHandler",
            "level": "INFO",
            "filename": "/tmp/prefect.log",
            "maxBytes": 5_000_000,
            "backupCount": 3,
            "encoding": "utf-8",
        }
    },
    "loggers": {
        "prefect": {
            "handlers": ["file"]  # add file; keep existing API + console
        }
    },
}

# Initialize Prefect logging first, then apply your additions
setup_logging()
logging.config.dictConfig(LOGGING_UPDATE)

Using a config file - Place a JSON/YAML/TOML logging config on disk and point Prefect at it with the environment variable PREFECT_LOGGING_SETTINGS_PATH. Prefect will load it automatically. - Tip: Use

"incremental": true

in your config to add handlers without replacing Prefect’s defaults. Example YAML snippet (append a file handler):

version: 1
incremental: true
handlers:
  file:
    class: logging.handlers.RotatingFileHandler
    level: INFO
    filename: /tmp/prefect.log
    maxBytes: 5000000
    backupCount: 3
    encoding: utf-8
loggers:
  prefect:
    handlers: [file]

Export the path:

export PREFECT_LOGGING_SETTINGS_PATH=/path/to/logging.yml

Avoiding duplicate logs - Don’t attach handlers to both the root logger and the Prefect logger with propagation turned on; prefer attaching to the Prefect logger and set propagate to False if you redefine it. - If adding handlers in code, check existing handlers (as shown) or gate on an env var so you only attach once. - When using dictConfig, prefer

"incremental": true

and only define the handlers you’re adding. Notes - In Prefect 3.x, the APILogHandler and PrefectConsoleHandler are used under the hood; you generally don’t need to add APILogHandler yourself.

- This approach is similar to Prefect 2.x, but handler names and the configuration helpers live under prefect.logging.* in 3.x. Docs - Advanced logging customization: Prefect 3: Logging customization - Adding logging to flows/tasks: Prefect 3: Add logging to workflows If you have a specific destination in mind (e.g., Datadog, CloudWatch, Logstash), tell me which one and I can share a ready-to-use handler config.

Great question. For Promtail, you’ve got three practical patterns from Prefect 3.x: - Most robust: write logs to a file and let Promtail tail the file - Push over Syslog to a Promtail syslog receiver - Send directly to Loki from Python (bypassing Promtail) with python-logging-loki Below are drop-in snippets that keep Prefect’s UI logs intact and add your Promtail/Loki destination. Option 1 — File + Promtail tailing (recommended) - Prefect keeps logging to the UI; you add a file handler. - Promtail tails that file and forwards to Loki. Append a file handler to the Prefect logger (no duplicates) via incremental dictConfig:

# e.g., run this at app startup
import logging.config
from prefect.logging.configuration import setup_logging

LOGGING_UPDATE = {
    "version": 1,
    "incremental": True,  # keep Prefect's defaults (APILogHandler + console)
    "handlers": {
        "prefect_file": {
            "class": "logging.handlers.RotatingFileHandler",
            "level": "INFO",
            "filename": "/var/log/prefect/flows.log",
            "maxBytes": 10_000_000,
            "backupCount": 5,
            "encoding": "utf-8",
        }
    },
    "loggers": {
        "prefect": {"handlers": ["prefect_file"]}
    },
}

setup_logging()
logging.config.dictConfig(LOGGING_UPDATE)

Promtail config to scrape that file:

scrape_configs:
  - job_name: prefect_files
    static_configs:
      - targets: [localhost]
        labels:
          job: prefect
          app: prefect
          __path__: /var/log/prefect/flows.log

Option 2 — Syslog to Promtail - Promtail can receive Syslog and forward to Loki. - Add a SysLogHandler to the Prefect logger. Append a Syslog handler:

import logging, socket
import logging.config
from prefect.logging.configuration import setup_logging

LOGGING_UPDATE = {
    "version": 1,
    "incremental": True,
    "handlers": {
        "prefect_syslog": {
            "class": "logging.handlers.SysLogHandler",
            "level": "INFO",
            "address": ["promtail.your-net", 1514],  # host:port where Promtail listens
            "socktype": socket.SOCK_DGRAM           # or SOCK_STREAM to use TCP
        }
    },
    "loggers": {
        "prefect": {"handlers": ["prefect_syslog"]}
    },
}

setup_logging()
logging.config.dictConfig(LOGGING_UPDATE)

Promtail syslog receiver:

scrape_configs:
  - job_name: prefect_syslog
    syslog:
      listen_address: 0.0.0.0:1514
      labels:
        job: prefect
        app: prefect
    relabel_configs:
      - source_labels: ['__syslog_message_hostname']
        target_label: 'host'

Option 3 — Send directly to Loki from Python - This bypasses Promtail and uses a Python handler to push to Loki. - Uses the third-party library python-logging-loki. Install and add handler:

# pip install python-logging-loki
import logging.config
from prefect.logging.configuration import setup_logging

LOGGING_UPDATE = {
    "version": 1,
    "incremental": True,
    "handlers": {
        "loki": {
            "class": "logging_loki.LokiHandler",
            "level": "INFO",
            "url": "<http://loki:3100/loki/api/v1/push>",
            "tags": {"app": "prefect"},  # static labels
            "version": "1"
        }
    },
    "loggers": {
        "prefect": {"handlers": ["loki"]}
    },
}

setup_logging()
logging.config.dictConfig(LOGGING_UPDATE)

Keep Prefect UI logs - In Prefect 3.x, logs appear in the UI via the built-in APILogHandler. The incremental configs above only add handlers, preserving UI delivery. - Always log inside flows/tasks with

get_run_logger()

so your messages are linked to the run context in the UI. Avoid duplicates - Prefer adding handlers to the

prefect

logger and leave

propagate

alone (or set it to

false

if you redefine the whole logger) to prevent duplicate emissions. - Use “incremental: true” when using dictConfig so you don’t replace Prefect’s defaults.

If you share which path you prefer (file tail, syslog, or direct Loki) and any labeling you need (e.g., flow name, run IDs), I can tailor the exact config. References - Prefect logging customization: Prefect 3: Logging customization - Syslog receiver (Promtail): Promtail syslog receiver - File tailing (Promtail): Promtail static config (file targets) - python-logging-loki: python-logging-loki