Great question — you’re on the right track. Here’s how I’d approach Loki integration with Prefect 3.x in Kubernetes, plus a concrete pattern to bridge Prefect logging and Loguru so you keep both Prefect run context and your existing structured fields. Assumptions: Prefect 3.x, Kubernetes pods, Promtail scraping stdout, Loki backend. Answers to your numbered questions 1) Recommended pattern in K8s - Emit JSON to stdout from your Prefect pods; let Promtail scrape and parse it, then label key fields for Loki queries. - Keep Prefect’s API logging enabled so logs still show in the Prefect UI; this gives you “dual logging” (UI + Loki) without custom transports. - Summary: JSON to stdout + Promtail → Loki, and Prefect’s APILogHandler → Prefect API. 2) JSON-structured logs vs. “native Loki integration” - There is no native Loki integration in Prefect 3.x. - Use Prefect’s built-in JSON formatter to write structured logs to stdout (for Promtail) and keep the APILogHandler for Prefect UI. 3) Ensuring Prefect run metadata is included -

get_run_logger()

attaches Prefect context as LogRecord extras automatically: - flow_run_id, flow_run_name, flow_name - task_run_id, task_run_name, task_name (in task context) - If you use Prefect’s JSON formatter on the console handler, these fields will be present in the JSON Promtail scrapes. - For code that uses Loguru directly, add a Loguru patcher that enriches each record with Prefect runtime context (

prefect.runtime.flow_run

/

task_run

) so Loguru logs also carry run IDs. 4) Prefect logger configuration for Loki-friendliness - Use Prefect’s

JsonFormatter

for the console handler (stdout) and (optionally) for API logs. Keep log level and batching defaults unless you have specific needs. - Prefer emitting JSON keys that Promtail can extract into labels: flow_run_id, task_run_id, level, name/logger, and a small set of business keys (e.g., error_domain, exception_class) if cardinality allows. 5) Best practice for correlation across flow → task → retries - Label with: - flow_run_id (primary correlation key) - task_run_id (for per-task drilldown) - task_run_attempt/run_count (for retries) - flow_name, task_name (optional, watch cardinality) - Query examples: - All logs for a flow run:

{job="prefect", flow_run_id="UUID"}

- All failed IGXE registrations today: add labels/fields

error_domain="IGXE"

and filter by

level="ERROR"

or match text:

{job="prefect"} |= "IGXE" |= "registration" | level="ERROR"

Concrete setup A) Prefect logging configuration (JSON to stdout + keep API) Create a logging.yml and mount it into your container:

version: 1
disable_existing_loggers: False

formatters:
  json:
    class: prefect.logging.formatters.JsonFormatter
    format: "default"  # or "pretty"

handlers:
  console_json:
    class: logging.StreamHandler
    formatter: json
    stream: <ext://sys.stdout>
    level: 0

  api:
    class: prefect.logging.handlers.APILogHandler
    formatter: json
    level: 0

loggers:
  prefect.flow_runs:
    level: NOTSET
    handlers: [console_json, api]
    propagate: False

  prefect.task_runs:
    level: NOTSET
    handlers: [console_json, api]
    propagate: False

# Optional: capture other Prefect internals in JSON too
  prefect:
    level: INFO
    handlers: [console_json]
    propagate: False

root:
  level: WARNING
  handlers: [console_json]

Set env vars in your pod:

PREFECT_LOGGING_CONFIG_PATH=/etc/prefect/logging.yml
PREFECT_LOGGING_TO_API_ENABLED=true
# Optional tweaks:
# PREFECT_LOGGING_LEVEL=INFO
# PREFECT_LOGGING_LOG_PRINTS=true

Docs for reference: - Add logging - Logging customization

- API refs: Handlers, Formatters, Settings B) Promtail pipeline to parse JSON and promote labels Example Promtail snippet to parse and label key fields:

scrape_configs:
- job_name: prefect
  kubernetes_sd_configs:
  - role: pod
  pipeline_stages:
  - json:
      expressions:
        flow_run_id: flow_run_id
        task_run_id: task_run_id
        flow_name: flow_name
        task_name: task_name
        level: levelname
        logger: name
        task_run_attempt: run_count
        error_domain: error_domain
        exception_class: exception_class
  - labels:
      flow_run_id:
      task_run_id:
      level:
      logger:
      flow_name:
      task_name:
      task_run_attempt:
      error_domain:
      exception_class:
  - timestamp:
      source: asctime
      format: RFC3339Nano

Notes: - Only promote a small set of stable, low-cardinality labels. Consider leaving

exception_class

and

error_domain

as parsed fields (not labels) if they vary widely in your workload. - Keep

message

as the log body, not a label. C) Bridging Prefect logger with Loguru (dual logging) Goal: When your code uses Prefect’s

get_run_logger()

, the log goes to: - Prefect API (for UI) - Stdout as JSON (for Promtail) - Loguru (preserving your structured extras like module, function, line, error_domain, exception_class) 1) Forward stdlib logging (Prefect logger) to Loguru in addition to Prefect’s handlers: ``` import logging import sys from loguru import logger as loguru_logger def setup_loguru_bridge(): # Configure Loguru to emit JSON to stdout (for Promtail) loguru_logger.remove() loguru_logger.add( sys.stdout, serialize=True, # JSON output with Loguru's structured fields backtrace=False, diagnose=False, level="INFO", ) # Enrich every Loguru record with Prefect runtime context (for logs not using get_run_logger) def add_prefect_context(record): try: from prefect.runtime import flow_run, task_run, deployment try: fr_id = flow_run.id() if fr_id: record["extra"].setdefault("flow_run_id", str(fr_id)) record["extra"].setdefault("flow_run_name", flow_run.name()) record["extra"].setdefault("flow_name", flow_run.flow_name()) except Exception: pass try: tr_id = task_run.id() if tr_id: record["extra"].setdefault("task_run_id", str(tr_id)) record["extra"].setdefault("task_run_name", task_run.name()) record["extra"].setdefault("task_name", task_run.task_name()) record["extra"].setdefault("run_count", task_run.run_count()) except Exception: pass try: dep_name = deployment.name() if dep_name: record["extra"].setdefault("deployment_name", dep_name) except Exception: pass except Exception: # runtime context not available pass loguru_logger.configure(patcher=add_prefect_context) # Add a stdlib logging handler that forwards Prefect LogRecords to Loguru class LoguruHandler(logging.Handler): def emit(self, record: logging.LogRecord): try: level = loguru_logger.level(record.levelname).name except Exception: level = record.levelno # Preserve Prefect context extras on the Loguru record extras = {} for key in ( "flow_run_id", "flow_run_name", "flow_name", "task_run_id", "task_run_name", "task_name",

"deployment_name", "run_count", "error_domain", "exception_class", ): val = getattr(record, key, None) if val is not None: extras[key] = str(val) loguru_logger.bind(**extras).opt( exception=record.exc_info, depth=1, record=True, ).log(level, record.getMessage()) # Attach the bridge to Prefect loggers WITHOUT removing Prefect's default handlers for name in ("prefect.flow_runs", "prefect.task_runs"): logging.getLogger(name).addHandler(LoguruHandler()) # Keep propagate=False on these loggers so you don't double-log via root

- This preserves Prefect → API and Prefect → stdout JSON, and additionally forwards to Loguru so your existing Loguru sinks/indexing get the same events with Prefect context.

2) Use Prefect’s logger in your flows/tasks (so Prefect context is present):

from prefect import flow, task from prefect.logging import get_run_logger @task def do_work(): logger = get_run_logger() logger.info("Task running", extra={"error_domain": "IGXE"}) @flow def my_flow(): setup_loguru_bridge() logger = get_run_logger() logger.info("Starting flow") do_work.submit() ``` D) Getting your existing Loguru JSON fields - With

serialize=True

, Loguru emits its rich fields (module, function, line). Your

patcher

adds Prefect context fields to the same JSON. Promtail will parse it. - If your business logic binds extra fields (e.g.,

logger = logger.bind(error_domain="IGXE")

), those will appear under

extra

in the JSON. You can promote them to labels in Promtail if cardinality is acceptable. E) Queries in Loki/Grafana - All logs for a flow run:

{job="prefect", flow_run_id="xxx-uuid"}

- All failed IGXE registration attempts today (if labeled): -

{job="prefect", error_domain="IGXE", level="ERROR"} |= "registration"

- Or if

error_domain

is only in JSON body, use a filter:

{job="prefect"} | json | error_domain="IGXE" | level="ERROR" |= "registration"

Notes and gotchas - Don’t remove Prefect’s

APILogHandler

if you want logs in the Prefect UI. - Avoid high-cardinality labels in Loki. Prefer keeping many fields in JSON and only promote stable keys (flow_run_id, task_run_id, level). -

prefect.runtime

context works inside running flow/task processes. Outside of a run, those functions may raise; guard with try/except as shown. - If you want

print()

statements captured, set

PREFECT_LOGGING_LOG_PRINTS=true

. Relevant docs and references - How to add logging: Add logging - Logging customization (JSON, handlers): Logging customization - API reference: Handlers prefect.logging.handlers, Formatters prefect.logging.formatters, Settings prefect.settings.models.logging - Kubernetes deployments (general): Kubernetes guide If you share your Promtail pipeline and any Loki label constraints you have, I can tailor the label selection and queries to fit your environment and data volumes.