https://prefect.io logo
Join Slack
Powered by
<@ULVA73B9P> - I cant get logging to work with my ...
# ask-marvin
j
@Marvin - I cant get logging to work with my own custom logger. I am using 
PREFECT_LOGGING_EXTRA_LOGGERS=src
plus
Copy code
try:
        # Configure the main Prefect loggers
        prefect_loggers = [
            "prefect",
            "prefect.flow_runs",
            "prefect.task_runs",
        ]

        for logger_name in prefect_loggers:
            prefect_logger = logging.getLogger(logger_name)
            prefect_logger.setLevel(level)

            # Replace handlers with our JSON handler
            for existing_handler in prefect_logger.handlers[:]:
                prefect_logger.removeHandler(existing_handler)

            json_handler = logging.StreamHandler(sys.stdout)
            json_handler.setFormatter(formatter)
            prefect_logger.addHandler(json_handler)

    except Exception:
        # If Prefect logging config fails, continue with standard logging
        pass
I dont see any logs in my prefect runs still but they show up in worker logs
m
✅ thought for 178.9 seconds
Thanks for the details — this behavior makes sense given your snippet. What’s happening - Prefect run logs are sent to the API by a special handler on the Prefect loggers called 
APILogHandler
. - In your code you’re removing all handlers on 
prefect
, 
prefect.flow_runs
, and 
prefect.task_runs
and replacing them with a 
StreamHandler
. That removes 
APILogHandler
, so the run logs never get sent to the API/UI. They still show up in worker logs because you’re writing to stdout. Relevant source for reference: - APILogHandler and APILogWorker - Logging configuration &amp; EXTRA_LOGGERS wiring - Formatters (including JsonFormatter) How to fix (keep run logs + your JSON) Option A — Minimal code change: don’t remove handlers; just add yours
Copy code
import logging, sys

level = <http://logging.INFO|logging.INFO>
formatter = ...  # your JSON formatter

for logger_name in ["prefect.flow_runs", "prefect.task_runs", "prefect"]:
    l = logging.getLogger(logger_name)
    l.setLevel(level)

    # DO NOT remove existing handlers; APILogHandler must remain
    json_handler = logging.StreamHandler(sys.stdout)
    json_handler.setFormatter(formatter)
    l.addHandler(json_handler)
Option B — Recommended: configure via logging.yml and keep APILogHandler 1) Create a logging.yml:
Copy code
formatters:
  json:
    class: prefect.logging.formatters.JsonFormatter

handlers:
  api:
    class: prefect.logging.handlers.APILogHandler
    level: NOTSET
    formatter: json   # use JSON in the Prefect UI as well
  json_stdout:
    class: logging.StreamHandler
    level: NOTSET
    formatter: json

loggers:
  prefect.flow_runs:
    level: NOTSET
    propagate: false
    handlers: [api, json_stdout]
  prefect.task_runs:
    level: NOTSET
    propagate: false
    handlers: [api, json_stdout]
  prefect:
    level: NOTSET
    propagate: false
    handlers: [json_stdout]
2) Point Prefect to it:
Copy code
export PREFECT_LOGGING_SETTINGS_PATH=/path/to/logging.yml
About PREFECT_LOGGING_EXTRA_LOGGERS - 
PREFECT_LOGGING_EXTRA_LOGGERS=src
attaches Prefect’s “extra” handlers (including 
APILogHandler
) to the 
src
logger so third-party/custom logs can also go to the API — but only if logs are emitted while a flow/task run context is active. It does not change levels or propagation. - This does not replace the need to keep 
APILogHandler
on 
prefect.flow_runs
and 
prefect.task_runs
for run logs. Quick debug checks - Verify APILogHandler is still attached:
Copy code
import logging
print([type(h).__name__ for h in logging.getLogger("prefect.flow_runs").handlers])
# Should include 'APILogHandler'
- Ensure API logging is enabled:
Copy code
PREFECT_LOGGING_TO_API_ENABLED=true
- If needed, turn on internal debug to see warnings:
Copy code
PREFECT_LOGGING_INTERNAL_LEVEL=DEBUG
A couple of clarifying questions to make sure we get you to green: - Where is your logging configuration code running — inside the flow run process itself, or only at worker startup? - Do you want the Prefect UI log message to be JSON as well? If yes, set your JSON formatter on the 
APILogHandler
(Option B above). - For 
PREFECT_LOGGING_EXTRA_LOGGERS=src
, is your application actually logging with 
logging.getLogger("src")
(or children like 
src.module
)? If not, we can adjust the logger names.
j
@Marvin - I have created config/logging.yml and set 
PREFECT_LOGGING_SETTINGS_PATH: "/workspace/config/logging.yml"
in my docker. I am still not getting logs in the flow. I see no logs in tasks or the flow in the UI. Here is an example of what I do to do logging within task
Copy code
@task(name="fetch_tickers", retries=2)
async def fetch_tickers() -> bytes:
    
    logger = logging.getLogger(__name__)
    url = "redacted
...
    <http://logger.info|logger.info>("Fetching SEC company tickers", extra={"url": url})
m
✅ thought for 84.0 seconds
Got it — thanks for the extra context. Given that you don’t see any logs in the Prefect UI (not from your tasks nor from Prefect itself), there are a few likely culprits: What typically breaks UI logs - APILogHandler was removed by your custom logging.yml (or by code) from the 
prefect.flow_runs
/ 
prefect.task_runs
loggers. Without this handler, nothing is sent to the API/UI. - Your task logs use 
logging.getLogger(__name__)
and that logger isn’t wired to Prefect’s APILogHandler (via 
PREFECT_LOGGING_EXTRA_LOGGERS
or logging.yml), or propagation got disabled. - The env var 
PREFECT_LOGGING_SETTINGS_PATH
is not present in the flow run container/process (e.g., set only on the worker, not the image that actually runs the flow). - 
PREFECT_LOGGING_TO_API_ENABLED
is false. First, quick checks Run this at the start of a task to see what handlers are attached and if records will propagate. Paste the output here.
Copy code
import logging

def debug_logging_setup():
    names = [
        "prefect.flow_runs",
        "prefect.task_runs",
        "prefect.extra",
        __name__.split('.')[0],  # your package root (e.g., "src")
        __name__,
    ]
    for n in names:
        l = logging.getLogger(n)
        handlers = [type(h).__name__ for h in l.handlers]
        print(f"logger={n} handlers={handlers} propagate={l.propagate}")

debug_logging_setup()
You should see 
APILogHandler
on 
prefect.flow_runs
and 
prefect.task_runs
. If those lists don’t include 
APILogHandler
, the UI won’t get logs. Also verify the basics inside the flow run process: - 
PREFECT_LOGGING_TO_API_ENABLED=true
- 
PREFECT_LOGGING_SETTINGS_PATH
is set in the flow run container/process (not just the worker) How to make your task logger emit to the UI You’re using 
logger = logging.getLogger(__name__)
. For those messages to reach Prefect: - Either add your package root (e.g., 
src
) to extra loggers: 
PREFECT_LOGGING_EXTRA_LOGGERS=src
- Or explicitly attach APILogHandler to that logger (or its ancestor) in logging.yml - Or, simplest, use 
get_run_logger()
inside the task/flow:
Copy code
from prefect import task, get_run_logger

@task(name="fetch_tickers", retries=2)
async def fetch_tickers() -> bytes:
    logger = get_run_logger()
    url = "redacted"
    <http://logger.info|logger.info>("Fetching SEC company tickers", extra={"url": url})
A working logging.yml (keeps UI logs and adds JSON) This keeps the APILogHandler on Prefect run loggers and also wires your app logger (assuming it’s under the 
src
package). Adjust 
src
if your module root is different.
Copy code
formatters:
  json:
    class: prefect.logging.formatters.JsonFormatter

handlers:
  api:
    class: prefect.logging.handlers.APILogHandler
    level: NOTSET
    formatter: json
  json_stdout:
    class: logging.StreamHandler
    level: NOTSET
    formatter: json

loggers:
  prefect.flow_runs:
    level: NOTSET
    handlers: [api, json_stdout]
    propagate: false
  prefect.task_runs:
    level: NOTSET
    handlers: [api, json_stdout]
    propagate: false

  # Your application package root so that logging.getLogger(__name__) under src.* reaches APILogHandler
  src:
    level: INFO
    handlers: [api, json_stdout]
    propagate: true
If you prefer using 
PREFECT_LOGGING_EXTRA_LOGGERS=src
instead of defining 
src
in YAML, that also works — just be sure nothing in your YAML disables propagation for those loggers. Common gotchas - A YAML that defines 
prefect.flow_runs
and 
prefect.task_runs
but omits the 
api
handler will stop UI logs. - A YAML that sets 
propagate: false
on your app logger(s) and doesn’t add APILogHandler there will prevent logs from reaching the API. - Setting 
PREFECT_LOGGING_SETTINGS_PATH
only on the worker container won’t affect the flow run container (if using Docker/K8s workers). Ensure it’s in the image/env that runs the flow. Helpful references - APILogHandler and APILogWorker: source
- Logging configuration and EXTRA_LOGGERS wiring: source - Formatters (including JsonFormatter): source To move forward, could you: - Share your 
config/logging.yml
- Confirm your worker type (Process/Docker/Kubernetes) - Confirm 
PREFECT_LOGGING_SETTINGS_PATH
and 
PREFECT_LOGGING_TO_API_ENABLED
are present in the flow run environment - Paste the output of the 
debug_logging_setup()
snippet from inside a task With that, I can pinpoint the exact config change needed.
j
@Marvin Docker compose is my runtime locally for this. 
PREFECT_LOGGING_TO_API_ENABLED
is not explicitly set, settings_path is. Logs
Copy code
logger=prefect.flow_runs handlers=[] propagate=True

logger=prefect.task_runs handlers=[] propagate=True

logger=prefect.extra handlers=[] propagate=True

logger=sync_universe handlers=[] propagate=True

logger=sync_universe handlers=[] propagate=True
logging.yml
Copy code
# Prefect logging configuration
# This file configures Prefect to use JSON formatting while preserving APILogHandler
# for sending logs to the Prefect UI.
#
# Set PREFECT_LOGGING_SETTINGS_PATH to point to this file:
#   export PREFECT_LOGGING_SETTINGS_PATH=/path/to/config/logging.yml

version: 1
disable_existing_loggers: false

formatters:
  json:
    "()": src.core.logging.StructuredJSONFormatter

handlers:
  api:
    class: prefect.logging.handlers.APILogHandler
    level: NOTSET
    formatter: json
  json_stdout:
    class: logging.StreamHandler
    stream: <ext://sys.stdout>
    level: NOTSET
    formatter: json

loggers:
  prefect:
    level: NOTSET
    propagate: false
    handlers: [json_stdout]
  prefect.flow_runs:
    level: NOTSET
    propagate: false
    handlers: [api, json_stdout]
  prefect.task_runs:
    level: NOTSET
    propagate: false
    handlers: [api, json_stdout]
  prefect.extra:
    level: NOTSET
    propagate: false
    handlers: [api, json_stdout]
  src:
    level: NOTSET
    propagate: false
    handlers: [api, json_stdout]

root:
  level: INFO
  handlers: [json_stdout]
m
✅ thought for 268.8 seconds
Thank you — that debug output is the smoking gun. All of these loggers have zero handlers: - prefect.flow_runs - prefect.task_runs - prefect.extra - your app logger(s) By default, Prefect attaches APILogHandler to prefect.flow_runs/task_runs/extra. If those handlers are missing, nothing will ever reach the UI. So something in your environment is removing handlers or preventing Prefect’s logging config from being applied. Most likely causes in your setup - Your earlier “remove handlers and add StreamHandler” code is still running somewhere. If an exception occurs when creating your formatter/handler, your try/except swallows it after you’ve removed handlers — leaving these loggers with no handlers. That exactly matches your output. - Your custom formatter 
src.core.logging.StructuredJSONFormatter
is not importable in the flow run process. If dictConfig tries to resolve it and fails, the config application can error and leave logging in a bad state. - The flow run process is not actually seeing 
PREFECT_LOGGING_SETTINGS_PATH
(env var set only on worker container, not the process that executes the flow). Concrete steps to fix 1) Remove any code that manipulates Prefect loggers If you still have code like this anywhere, delete it. It removes APILogHandler and will break UI logs if anything errors before you add your replacement.
Copy code
for logger_name in ["prefect", "prefect.flow_runs", "prefect.task_runs"]:
    l = logging.getLogger(logger_name)
    for h in l.handlers[:]:
        l.removeHandler(h)
    # then try to add your own handler inside a try/except
2) Verify the flow run process sees your logging file and can import your formatter Run this inside a task and paste results:
Copy code
import os, pathlib, logging

print("PREFECT_LOGGING_SETTINGS_PATH:", os.getenv("PREFECT_LOGGING_SETTINGS_PATH"))
p = pathlib.Path(os.getenv("PREFECT_LOGGING_SETTINGS_PATH") or "")
print("Settings file exists:", p.exists(), "size:", p.stat().st_size if p.exists() else "n/a")

try:
    from src.core.logging import StructuredJSONFormatter
    print("Custom formatter import: OK")
except Exception as e:
    print("Custom formatter import FAILED:", repr(e))
3) Try Prefect defaults to isolate the issue Temporarily unset 
PREFECT_LOGGING_SETTINGS_PATH
and use 
get_run_logger()
in a task:
Copy code
from prefect import get_run_logger

@task
def t():
    get_run_logger().info("hello from task")
- If this shows in the UI, your issue is custom logging config/import. - If not, something is removing handlers later. 4) Start with a minimal working logging.yml using only Prefect’s built-ins Once step 3 works, try this minimal file first (no custom formatter). It should attach APILogHandler and give you UI logs:
Copy code
version: 1
disable_existing_loggers: false

formatters:
  json:
    class: prefect.logging.formatters.JsonFormatter

handlers:
  api:
    class: prefect.logging.handlers.APILogHandler
    level: NOTSET
    formatter: json
  json_stdout:
    class: logging.StreamHandler
    stream: <ext://sys.stdout>
    level: NOTSET
    formatter: json

loggers:
  prefect.flow_runs:
    level: NOTSET
    handlers: [api, json_stdout]
    propagate: false
  prefect.task_runs:
    level: NOTSET
    handlers: [api, json_stdout]
    propagate: false

root:
  level: INFO
  handlers: [json_stdout]
If that works, then switch the formatter to your 
src.core.logging.StructuredJSONFormatter
and confirm the import works in the flow-run container. 5) Re-run your debug check After starting a flow run, inside a task:
Copy code
import logging
for n in ["prefect.flow_runs", "prefect.task_runs", "prefect.extra"]:
    l = logging.getLogger(n)
    print(n, [type(h).__name__ for h in l.handlers], "propagate=", l.propagate)
You should see APILogHandler in both flow/task loggers. 6) Make your app logger reach the UI You can either: - Keep your YAML’s 
src
logger with 
[api, json_stdout]
, or
- Set 
PREFECT_LOGGING_EXTRA_LOGGERS=sync_universe
(or your real package root), which attaches the same handlers as 
prefect.extra
to your logger. 7) Prefer get_run_logger() inside tasks Using 
get_run_logger()
ensures flow/task context fields are added and routed correctly:
Copy code
from prefect import task, get_run_logger

@task
async def fetch_tickers() -> bytes:
    logger = get_run_logger()
    url = "redacted"
    <http://logger.info|logger.info>("Fetching SEC company tickers", extra={"url": url})
Helpful checks - See if API logging is enabled (default is true):
Copy code
from prefect.settings import PREFECT_LOGGING_TO_API_ENABLED
print("to_api_enabled =", PREFECT_LOGGING_TO_API_ENABLED.value())
- Enable internal debug to surface logging setup issues:
Copy code
PREFECT_LOGGING_INTERNAL_LEVEL=DEBUG
Key sources - APILogHandler and APILogWorker - Logging configuration and setup - Built-in formatters If you can confirm: - Whether any code still removes handlers - Whether the custom formatter import works inside the flow container - The output of the “settings file exists” check …I can zero in on the exact step that’s breaking logging in your run environment.
j
Here is my entire logging.py file
Copy code
"""Unified structured logging with Prefect context integration.

This module provides a centralized logging configuration that:
- Outputs structured JSON logs to stdout for non-Prefect code
- Preserves Prefect's APILogHandler so logs appear in the Prefect UI
- Uses YAML-based configuration for Prefect logging (Option B approach)

Usage:
    from src.core.logging import setup_logging
    import logging

    setup_logging()  # Call once at application startup
    logger = logging.getLogger(__name__)
    <http://logger.info|logger.info>("Processing started")

Prefect Integration:
    Logs are sent to the Prefect UI via APILogHandler configured in config/logging.yml.
    Set PREFECT_LOGGING_SETTINGS_PATH to point to the YAML config file.
    
    Inside flows/tasks, prefer using get_run_logger() for best context enrichment:
        from prefect import get_run_logger
        logger = get_run_logger()
"""

import json
import logging
import os
import sys
from datetime import UTC, datetime
from pathlib import Path
from typing import Any

# Application root logger name - Prefect will capture all loggers under this
APP_LOGGER_ROOT = "src"


class StructuredJSONFormatter(logging.Formatter):
    """Format logs as structured JSON with optional Prefect context.

    Output schema:
        {
            "timestamp": "2025-11-28T12:34:56.789Z",
            "level": "INFO",
            "logger": "src.ingestion.flows.sync_universe",
            "message": "Fetching SEC company tickers...",
            "flow_run_id": "abc-123",  # Only when in Prefect context
            "task_run_id": "def-456",  # Only when in Prefect context
            "exception": "..."         # Only when exception info present
        }
    """

    def format(self, record: logging.LogRecord) -> str:
        """Format log record as JSON string with Prefect context."""
        log_data: dict[str, Any] = {
            "timestamp": datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z",
            "level": record.levelname,
            "logger": record.name,
            "message": record.getMessage(),
        }

        # Add Prefect context from record attributes (set by Prefect's logger)
        if hasattr(record, "flow_run_id") and record.flow_run_id:
            log_data["flow_run_id"] = str(record.flow_run_id)
        if hasattr(record, "task_run_id") and record.task_run_id:
            log_data["task_run_id"] = str(record.task_run_id)

        # Fallback: try to get context directly if not in record
        if "flow_run_id" not in log_data:
            prefect_context = self._get_prefect_context()
            if prefect_context.get("flow_run_id"):
                log_data["flow_run_id"] = prefect_context["flow_run_id"]
            if prefect_context.get("task_run_id"):
                log_data["task_run_id"] = prefect_context["task_run_id"]

        # Add exception info if present
        if record.exc_info:
            log_data["exception"] = self.formatException(record.exc_info)

        return json.dumps(log_data)

    def _get_prefect_context(self) -> dict[str, str | None]:
        """Extract Prefect flow_run_id and task_run_id if available.

        Returns:
            Dict with flow_run_id and task_run_id (None if not in Prefect context)
        """
        result: dict[str, str | None] = {"flow_run_id": None, "task_run_id": None}

        try:
            from prefect.context import get_run_context

            ctx = get_run_context()
            if ctx is not None:
                # Handle flow run context
                if hasattr(ctx, "flow_run") and ctx.flow_run is not None:
                    result["flow_run_id"] = str(ctx.flow_run.id)
                # Handle task run context
                if hasattr(ctx, "task_run") and ctx.task_run is not None:
                    result["task_run_id"] = str(ctx.task_run.id)
        except (ImportError, RuntimeError):
            # Prefect not installed or not in a run context - this is fine
            pass

        return result


def _get_logging_config_path() -> Path:
    """Get the path to the logging.yml configuration file.

    Returns:
        Path to the logging.yml file in the config directory.
    """
    # Try to find config relative to this file's location
    module_dir = Path(__file__).parent
    # Go up to src/, then to project root, then into config/
    project_root = module_dir.parent.parent
    return project_root / "config" / "logging.yml"


def setup_logging(level: str | None = None) -> None:
    """Configure logging with structured JSON output.

    This function:
    1. Sets PREFECT_LOGGING_SETTINGS_PATH to use our YAML config (preserves APILogHandler)
    2. Sets PREFECT_LOGGING_EXTRA_LOGGERS to capture application logs
    3. Configures the root logger with JSON formatting for non-Prefect code

    IMPORTANT: This does NOT touch Prefect's loggers directly. Prefect's logging
    is configured via the YAML file to preserve APILogHandler.

    Args:
        level: Log level override. If None, reads from LOG_LEVEL env var (default: INFO)
    """
    if level is None:
        level = os.getenv("LOG_LEVEL", "INFO")

    # Validate level
    numeric_level = getattr(logging, level.upper(), None)
    if not isinstance(numeric_level, int):
        raise ValueError(f"Invalid log level: {level}")

    # Configure Prefect to use our YAML logging config (preserves APILogHandler)
    # This MUST be set before Prefect initializes its logging
    _configure_prefect_logging_path()

    # Configure Prefect to capture our application logs
    _configure_prefect_extra_loggers()

    # Create our JSON formatter for the root logger
    json_formatter = StructuredJSONFormatter()

    # Configure root logger for non-Prefect code paths
    # NOTE: We do NOT touch Prefect loggers here - they are configured via YAML
    handler = logging.StreamHandler(sys.stdout)
    handler.setFormatter(json_formatter)
    logging.basicConfig(level=numeric_level, handlers=[handler], force=True)

    # Configure the src logger explicitly (but don't add handlers - let Prefect do that)
    src_logger = logging.getLogger(APP_LOGGER_ROOT)
    src_logger.setLevel(numeric_level)


def _configure_prefect_logging_path() -> None:
    """Set PREFECT_LOGGING_SETTINGS_PATH to use our YAML config.

    This ensures Prefect uses our logging.yml which:
    - Keeps APILogHandler intact (logs go to Prefect UI)
    - Uses Prefect's JsonFormatter for consistent output
    """
    # Only set if not already configured
    if "PREFECT_LOGGING_SETTINGS_PATH" not in os.environ:
        config_path = _get_logging_config_path()
        if config_path.exists():
            os.environ["PREFECT_LOGGING_SETTINGS_PATH"] = str(config_path)


def _configure_prefect_extra_loggers() -> None:
    """Configure Prefect to capture application logs.

    Sets PREFECT_LOGGING_EXTRA_LOGGERS to include our application root logger.
    This tells Prefect to send logs from our loggers to the Prefect UI.
    """
    extra_loggers = os.getenv("PREFECT_LOGGING_EXTRA_LOGGERS", "")

    # Add our app logger if not already present
    if APP_LOGGER_ROOT not in extra_loggers:
        if extra_loggers:
            extra_loggers = f"{extra_loggers},{APP_LOGGER_ROOT}"
        else:
            extra_loggers = APP_LOGGER_ROOT

        os.environ["PREFECT_LOGGING_EXTRA_LOGGERS"] = extra_loggers
Here are logs from debug ind ocker
Copy code
PREFECT_LOGGING_SETTINGS_PATH: /workspace/config/logging.yml

Settings file exists: True size: 1121

Custom formatter import: OK

prefect.flow_runs [] propagate= True

prefect.task_runs [] propagate= True

prefect.extra [] propagate= True
@Marvin please help
m
Your message was too long, here's your message at the allowed limit: Here is my entire logging py file ``` Unified structured logging with Prefect context integration This module provides a centralized logging configuration that - Outputs structured JSON logs to stdout for non-Prefect code - Preserves Prefect's APILogHandler so logs appear in the Prefect UI - Uses YAML-based configuration for Prefect logging (Option B approach) Usage from src core logging import setup_logging import logging setup_logging() # Call once at application startup logger = logging getLogger(name) <http //logger info|logger info>( Processing started ) Prefect Integration Logs are sent to the Prefect UI via APILogHandler configured in config/logging yml Set PREFECT_LOGGING_SETTINGS_PATH to point to the YAML config file Inside flows/tasks prefer using get_run_logger() for best context enrichment from prefect import get_run_logger logger = get_run_logger() import json import logging import os import sys from datetime import UTC datetime from pathlib import Path from typing import Any # Application root logger name - Prefect will capture all loggers under this APP_LOGGER_ROOT = src class StructuredJSONFormatter(logging Formatter) Format logs as structured JSON with optional Prefect context Output schema { timestamp 2025-11-28T12 34 56 789Z level INFO logger src ingestion flows sync_universe message Fetching SEC company tickers flow_run_id abc-123 # Only when in Prefect context task_run_id def-456 # Only when in Prefect context exception # Only when exception info present } def format(self record logging LogRecord) -> str Format log record as JSON string with Prefect context log_data dict[str Any] = { timestamp datetime now(UTC) strftime( %Y-%m-%dT%H %M %S %f )[ -3] + Z level record levelname logger record name message record getMessage() } # Add Prefect context from record attributes (set by Prefect's logger) if hasattr(record flow_run_id ) and record flow_run_id log_data[ flow_run_id ] = str(record flow_run_id) if hasattr(record task_run_id ) and record task_run_id log_data[ task_run_id ] = str(record task_run_id) # Fallback try to get context directly if not in record if flow_run_id not in log_data prefect_context = self _get_prefect_context() if prefect_context get( flow_run_id ) log_data[ flow_run_id ] = prefect_context[ flow_run_id ] if prefect_context get( task_run_id ) log_data[ task_run_id ] = prefect_context[ task_run_id ] # Add exception info if present if record exc_info log_data[ exception ] = self formatException(record exc_info) return json dumps(log_data) def _get_prefect_context(self) -> dict[str str | None] Extract Prefect flow_run_id and task_run_id if available Returns Dict with flow_run_id and task_run_id (None if not in Prefect context) result dict[str str | None] = { flow_run_id None task_run_id None} try from prefect context import get_run_context ctx = get_run_context() if ctx is not None # Handle flow run context if hasattr(ctx flow_run ) and ctx flow_run is not None result[ flow_run_id ] = str(ctx flow_run id) # Handle task run context if hasattr(ctx task_run ) and ctx task_run is not None result[ task_run_id ] = str(ctx task_run id) except (ImportError RuntimeError) # Prefect not installed or not in a run context - this is fine pass return result def _get_logging_config_path() -> Path Get the path to the logging yml configuration file Returns Path to the logging yml file in the config directory # Try to find
j
Here are parts of my logging file that touch prefect
Copy code
def setup_logging(level: str | None = None) -> None:
    """Configure logging with structured JSON output.

    This function:
    1. Sets PREFECT_LOGGING_SETTINGS_PATH to use our YAML config (preserves APILogHandler)
    2. Sets PREFECT_LOGGING_EXTRA_LOGGERS to capture application logs
    3. Configures the root logger with JSON formatting for non-Prefect code

    IMPORTANT: This does NOT touch Prefect's loggers directly. Prefect's logging
    is configured via the YAML file to preserve APILogHandler.

    Args:
        level: Log level override. If None, reads from LOG_LEVEL env var (default: INFO)
    """
    if level is None:
        level = os.getenv("LOG_LEVEL", "INFO")

    # Validate level
    numeric_level = getattr(logging, level.upper(), None)
    if not isinstance(numeric_level, int):
        raise ValueError(f"Invalid log level: {level}")

    # Configure Prefect to use our YAML logging config (preserves APILogHandler)
    # This MUST be set before Prefect initializes its logging
    _configure_prefect_logging_path()

    # Configure Prefect to capture our application logs
    _configure_prefect_extra_loggers()

    # Create our JSON formatter for the root logger
    json_formatter = StructuredJSONFormatter()

    # Configure root logger for non-Prefect code paths
    # NOTE: We do NOT touch Prefect loggers here - they are configured via YAML
    handler = logging.StreamHandler(sys.stdout)
    handler.setFormatter(json_formatter)
    logging.basicConfig(level=numeric_level, handlers=[handler], force=True)

    # Configure the src logger explicitly (but don't add handlers - let Prefect do that)
    src_logger = logging.getLogger(APP_LOGGER_ROOT)
    src_logger.setLevel(numeric_level)

def _configure_prefect_logging_path() -> None:
    """Set PREFECT_LOGGING_SETTINGS_PATH to use our YAML config.

    This ensures Prefect uses our logging.yml which:
    - Keeps APILogHandler intact (logs go to Prefect UI)
    - Uses Prefect's JsonFormatter for consistent output
    """
    # Only set if not already configured
    if "PREFECT_LOGGING_SETTINGS_PATH" not in os.environ:
        config_path = _get_logging_config_path()
        if config_path.exists():
            os.environ["PREFECT_LOGGING_SETTINGS_PATH"] = str(config_path)


def _configure_prefect_extra_loggers() -> None:
    """Configure Prefect to capture application logs.

    Sets PREFECT_LOGGING_EXTRA_LOGGERS to include our application root logger.
    This tells Prefect to send logs from our loggers to the Prefect UI.
    """
    extra_loggers = os.getenv("PREFECT_LOGGING_EXTRA_LOGGERS", "")

    # Add our app logger if not already present
    if APP_LOGGER_ROOT not in extra_loggers:
        if extra_loggers:
            extra_loggers = f"{extra_loggers},{APP_LOGGER_ROOT}"
        else:
            extra_loggers = APP_LOGGER_ROOT

        os.environ["PREFECT_LOGGING_EXTRA_LOGGERS"] = extra_loggers
Logs
Copy code
PREFECT_LOGGING_SETTINGS_PATH: /workspace/config/logging.yml
Settings file exists: True size: 1121
Custom formatter import: OK
prefect.flow_runs [] propagate= True
prefect.task_runs [] propagate= True
prefect.extra [] propagate= True
Using 
logger = get_run_logger()
to try to log things @Marvin help
m
✅ thought for 340.7 seconds
Thanks — this explains it. Root cause - Your 
setup_logging()
calls 
logging.basicConfig(..., force=True)
. The 
force=True
resets logging and removes handlers from existing loggers. That wipes Prefect’s 
APILogHandler
from 
prefect.flow_runs
and 
prefect.task_runs
, which is why your debug shows empty handler lists and why 
get_run_logger()
emits nothing to the UI. There’s also a second mismatch: - Your YAML wires the 
src
logger, but your app logs show 
logger=sync_universe
. So even if extra loggers were working, your app logger wouldn’t be matched by the 
src
stanza. What to change Option A — Recommended (no code-based basicConfig) - Remove your code’s 
logging.basicConfig(..., force=True)
entirely. - Let YAML drive everything; you already define the root handler as JSON and attach 
api
to run loggers. Also update the logger name in YAML to match your package root:
Copy code
loggers:
  prefect.flow_runs:
    level: NOTSET
    handlers: [api, json_stdout]
    propagate: false
  prefect.task_runs:
    level: NOTSET
    handlers: [api, json_stdout]
    propagate: false
  prefect.extra:
    level: NOTSET
    handlers: [api, json_stdout]
    propagate: false

  # Change this from 'src' to your actual package root
  sync_universe:
    level: NOTSET
    handlers: [api, json_stdout]
    propagate: false

root:
  level: INFO
  handlers: [json_stdout]
Option B — If you must keep basicConfig, reapply Prefect logging after it
Copy code
import logging, os
from prefect.logging.configuration import setup_logging as setup_prefect_logging

# Your basic config (preferably without force=True)
logging.basicConfig(level=<http://logging.INFO|logging.INFO>)

# Re-apply Prefect’s logging (loads your YAML and reattaches APILogHandler)
setup_prefect_logging(incremental=False)
If you keep 
force=True
, it will continue to wipe handlers; calling 
setup_prefect_logging(incremental=False)
after that is mandatory. Attach your app logger to the API - Either use the YAML stanza for 
sync_universe
as shown above, or set:
Copy code
export PREFECT_LOGGING_EXTRA_LOGGERS=sync_universe
Re-run quick checks inside a task
Copy code
import logging
for n in ["prefect.flow_runs", "prefect.task_runs", "prefect.extra", "sync_universe"]:
    l = logging.getLogger(n)
    print(n, [type(h).__name__ for h in l.handlers], "propagate=", l.propagate)
Expected: 
prefect.flow_runs
and 
prefect.task_runs
include 
APILogHandler
. 
sync_universe
includes 
APILogHandler
and 
StreamHandler
(json_stdout). Then test with get_run_logger
Copy code
from prefect import task, get_run_logger

@task
async def fetch_tickers():
    logger = get_run_logger()
    <http://logger.info|logger.info>("Fetching SEC company tickers", extra={"url": "redacted"})
If anything still doesn’t show up in the UI, set this to surface internal hints:
Copy code
PREFECT_LOGGING_INTERNAL_LEVEL=DEBUG
Relevant source for how Prefect wires handlers - APILogHandler and APILogWorker - Prefect logging setup and config loading - Default logging.yml (shows api handler on flow/task loggers) Give this a try: 1) Remove 
basicConfig(force=True)
and change 
src
-> 
sync_universe
in YAML. 2) Run a flow; share the debug output of the handler check above if logs still don’t appear.
j
@Marvin - does this allow us to use a custom logger for other things though? The whole reason I am doing this is to have logging not only for prefect, but for other modules too
m
✅ thought for 45.6 seconds
Short answer: yes. You can keep Prefect run logs going to the UI while using your own JSON logger for everything else. How to structure it - Keep APILogHandler on Prefect’s run loggers so flow/task logs reach the UI. - Put your JSON handler on: - the root logger to cover “everything” (third‑party, your code), or - specific module roots (e.g., 
sync_universe
, 
requests
) if you want tighter control. - For any non-Prefect logger you want to also appear in the Prefect UI, attach APILogHandler to those loggers too (either via YAML or 
PREFECT_LOGGING_EXTRA_LOGGERS
). A concrete YAML that does both - Uses your custom 
StructuredJSONFormatter
for stdout (works for “other modules”). - Sends Prefect run logs and your app logs to the Prefect UI via APILogHandler. - Adjust 
sync_universe
to your actual package root; add more module names as needed.
Copy code
version: 1
disable_existing_loggers: false

formatters:
  json:
    "()": src.core.logging.StructuredJSONFormatter

handlers:
  api:
    class: prefect.logging.handlers.APILogHandler
    level: NOTSET
    formatter: json
  json_stdout:
    class: logging.StreamHandler
    stream: <ext://sys.stdout>
    level: NOTSET
    formatter: json

loggers:
  # Prefect run loggers → UI (API) + your JSON stdout
  prefect.flow_runs:
    level: NOTSET
    handlers: [api, json_stdout]
    propagate: false
  prefect.task_runs:
    level: NOTSET
    handlers: [api, json_stdout]
    propagate: false
  prefect.extra:
    level: NOTSET
    handlers: [api, json_stdout]
    propagate: false

  # Your app logger(s) → UI (API) + your JSON stdout
  # Replace with your package root; you showed 'sync_universe' in your logs
  sync_universe:
    level: INFO
    handlers: [api, json_stdout]
    propagate: false

  # Optional: third-party modules you want in the Prefect UI too
  requests:
    level: WARNING
    handlers: [api, json_stdout]
    propagate: false

root:
  # Everything else gets your JSON stdout (but not sent to API unless added above)
  level: INFO
  handlers: [json_stdout]
Environment options - If you prefer env vars instead of YAML stanzas for app/third‑party loggers, set:
Copy code
PREFECT_LOGGING_EXTRA_LOGGERS=sync_universe,requests
This attaches the same handlers as 
prefect.extra
(including APILogHandler) to those loggers at runtime. Usage patterns - In tasks/flows, prefer 
get_run_logger()
when you want logs with run context in the UI:
Copy code
from prefect import task, get_run_logger

@task
async def fetch_tickers():
    log = get_run_logger()
    <http://log.info|log.info>("Fetching SEC company tickers", extra={"url": "redacted"})
- For “other modules,” continue using 
logging.getLogger(name)
. With the YAML above, they’ll emit JSON to stdout. If included via YAML or 
PREFECT_LOGGING_EXTRA_LOGGERS
, they’ll also go to the Prefect UI. Important gotchas to avoid - Don’t call 
logging.basicConfig(..., force=True)
after Prefect sets up logging; it wipes APILogHandler. If you must call it, immediately re-run:
Copy code
from prefect.logging.configuration import setup_logging as setup_prefect_logging
setup_prefect_logging(incremental=False)
- Ensure the logger names in YAML match your actual package roots (you had 
sync_universe
, not 
src
). - Make sure 
PREFECT_LOGGING_SETTINGS_PATH
is set in the flow run container/process. References - APILogHandler and APILogWorker - Prefect logging setup and config loading - Built-in formatters (JsonFormatter) This setup gives you: - Prefect run logs in the UI (with context) - Your own JSON-structured logs for everything else - Optional routing of selected non-Prefect loggers into the Prefect UI too
j
@Marvin - is this correct?
Copy code
# Prefect logging configuration
# This file configures Prefect to use JSON formatting while preserving APILogHandler
# for sending logs to the Prefect UI.
#
# Set PREFECT_LOGGING_SETTINGS_PATH to point to this file:
#   export PREFECT_LOGGING_SETTINGS_PATH=/path/to/config/logging.yml
#
# NOTE: We use Prefect's built-in JsonFormatter here to avoid import issues.
# Do NOT use logging.basicConfig(force=True) in code - it wipes APILogHandler!

version: 1
disable_existing_loggers: false

formatters:
  json:
    "()": src.core.logging.StructuredJSONFormatter

handlers:
  api:
    class: prefect.logging.handlers.APILogHandler
    level: DEBUG
    formatter: json
  json_stdout:
    class: logging.StreamHandler
    stream: <ext://sys.stdout>
    level: DEBUG
    formatter: json

loggers:
  prefect.flow_runs:
    level: INFO
    propagate: false
    handlers: [api, json_stdout]
  prefect.task_runs:
    level: INFO
    propagate: false
    handlers: [api, json_stdout]
  prefect.extra:
    level: INFO
    propagate: false
    handlers: [api, json_stdout]

  # Application loggers - these match the module paths used by get_run_logger()
  src:
    level: INFO
    propagate: false
    handlers: [api, json_stdout]
  src.ingestion:
    level: INFO
    propagate: false
    handlers: [api, json_stdout]
  src.ingestion.flows:
    level: INFO
    propagate: false
    handlers: [api, json_stdout]
  src.ingestion.flows.sync_universe:
    level: INFO
    propagate: false
    handlers: [api, json_stdout]

root:
  level: INFO
  handlers: [json_stdout]
m
✅ thought for 116.3 seconds
You’re very close. A few important fixes and clarifications: What to change - Use the correct ext syntax: use 
<ext://sys.stdout>
(no angle brackets) for the stream. - Logger names must match your actual modules. Your earlier output showed 
logger=sync_universe
, so include a top-level 
sync_universe
stanza (or set 
PREFECT_LOGGING_EXTRA_LOGGERS=sync_universe
). The 
src.*
entries won’t catch 
sync_universe
. - Prefer 
level: NOTSET
on `prefect.flow_runs`/`prefect.task_runs` so you don’t accidentally filter logs before the handler. Control verbosity via the root/app loggers. - Your comment says you’re using Prefect’s built-in JsonFormatter, but your formatter points to 
src.core.logging.StructuredJSONFormatter
. If you want to avoid import issues, switch to 
prefect.logging.formatters.JsonFormatter
. If you keep your custom formatter, ensure it’s importable in the flow-run container. Adjusted YAML
Copy code
version: 1
disable_existing_loggers: false

formatters:
  # Option A: use Prefect's built-in JSON formatter (safer)
  json:
    class: prefect.logging.formatters.JsonFormatter
  # Option B: use your custom formatter (ensure it imports in the run env)
  # json:
  #   "()": src.core.logging.StructuredJSONFormatter

handlers:
  api:
    class: prefect.logging.handlers.APILogHandler
    level: NOTSET
    formatter: json        # If you want JSON in the Prefect UI; remove if you want plain text there
  json_stdout:
    class: logging.StreamHandler
    stream: <ext://sys.stdout>
    level: NOTSET
    formatter: json

loggers:
  prefect.flow_runs:
    level: NOTSET
    propagate: false
    handlers: [api, json_stdout]
  prefect.task_runs:
    level: NOTSET
    propagate: false
    handlers: [api, json_stdout]
  prefect.extra:
    level: NOTSET
    propagate: false
    handlers: [api, json_stdout]

  # Add the actual roots your app uses. Earlier you had 'sync_universe'.
  sync_universe:
    level: INFO
    propagate: false
    handlers: [api, json_stdout]
  # Keep these only if your modules are actually under 'src...'
  # src:
  #   level: INFO
  #   propagate: false
  #   handlers: [api, json_stdout]
  # src.ingestion:
  #   level: INFO
  #   propagate: false
  #   handlers: [api, json_stdout]
  # src.ingestion.flows:
  #   level: INFO
  #   propagate: false
  #   handlers: [api, json_stdout]
  # src.ingestion.flows.sync_universe:
  #   level: INFO
  #   propagate: false
  #   handlers: [api, json_stdout]

root:
  level: INFO
  handlers: [json_stdout]
Alternative to listing app loggers in YAML - Use env var to cover your app and any 3rd-party libs you want in the UI:
Copy code
PREFECT_LOGGING_EXTRA_LOGGERS=sync_universe,requests,sqlalchemy
This attaches the same handlers as 
prefect.extra
(including APILogHandler) to those loggers. Verify inside a task after changes
Copy code
import logging
for n in ["prefect.flow_runs", "prefect.task_runs", "prefect.extra", "sync_universe"]:
    l = logging.getLogger(n)
    print(n, [type(h).__name__ for h in l.handlers], "propagate=", l.propagate)
You should see APILogHandler on flow/task loggers and on your app logger if configured. Reminder - Do not call 
logging.basicConfig(..., force=True)
after Prefect initializes logging; it wipes APILogHandler. If you must use it, immediately re-run:
Copy code
from prefect.logging.configuration import setup_logging as setup_prefect_logging
setup_prefect_logging(incremental=False)
References - APILogHandler - Logging setup and config loading - Formatters (JsonFormatter) If you confirm your actual package roots (e.g., 
sync_universe
vs 
src.*
), I can tailor the final YAML exactly.
j
@Marvin - sync_universe is the name of my flow. How can I make sure all flows use the same logger?
m
✅ thought for 64.4 seconds
Yes. Don’t key anything off the flow name. Instead: - Use get_run_logger() inside every flow/task for Prefect run logs (consistent across all flows). - Use a single application logger root for “everything else” (e.g., your package root like src or company_app) and attach handlers once in logging.yml. - If you want those app logs in the Prefect UI too, include that root in PREFECT_LOGGING_EXTRA_LOGGERS or add it in logging.yml with the api handler. A clean setup that works for all flows 1) YAML (shared by all flows via PREFECT_LOGGING_SETTINGS_PATH):
Copy code
version: 1
disable_existing_loggers: false

formatters:
  json:
    class: prefect.logging.formatters.JsonFormatter

handlers:
  api:
    class: prefect.logging.handlers.APILogHandler
    level: NOTSET
    formatter: json
  json_stdout:
    class: logging.StreamHandler
    stream: <ext://sys.stdout>
    level: NOTSET
    formatter: json

loggers:
  # Prefect run logs — same for ALL flows/tasks
  prefect.flow_runs:
    level: NOTSET
    handlers: [api, json_stdout]
    propagate: false
  prefect.task_runs:
    level: NOTSET
    handlers: [api, json_stdout]
    propagate: false
  prefect.extra:
    level: NOTSET
    handlers: [api, json_stdout]
    propagate: false

  # Your application root logger — covers all modules across flows
  # Replace 'src' with your actual package root
  src:
    level: INFO
    handlers: [api, json_stdout]   # include api if you want these in the UI
    propagate: false

root:
  level: INFO
  handlers: [json_stdout]
2) Environment
Copy code
# Shared in the flow-run environment/image
PREFECT_LOGGING_SETTINGS_PATH=/workspace/config/logging.yml

# Optional: add more app/third‑party roots to send to UI without editing YAML
PREFECT_LOGGING_EXTRA_LOGGERS=src,requests,sqlalchemy
3) In your code - For Prefect run logs (always consistent across flows):
Copy code
from prefect import flow, task, get_run_logger

@task
def t():
    log = get_run_logger()
    <http://log.info|log.info>("run-scoped log", extra={"foo": "bar"})
- For your app logs (same root across all flows):
Copy code
import logging
app_log = logging.getLogger("src")  # or "src.module", etc.
<http://app_log.info|app_log.info>("app log outside run")
Notes - The logger name you saw as sync_universe was likely the module name where the flow lives. Don’t rely on per-flow names; rely on: - prefect.flow_runs / prefect.task_runs for run logs - your package root (e.g., src) for app logs across all flows - Avoid logging.basicConfig(force=True). If you must call it, immediately reapply Prefect logging:
Copy code
from prefect.logging.configuration import setup_logging as setup_prefect_logging
setup_prefect_logging(incremental=False)
This gives you one consistent logger setup for all flows and tasks, with your JSON formatter everywhere, and optional routing of app/third‑party logs into the Prefect UI.
j
@Marvin I made a mistake - just give me the simple guide to setup logging without any extras or anything. just consistent logging in prefect's json format across all flows centralized in a yaml
m
✅ thought for 21.3 seconds
Perfect — here’s the simplest, central YAML-only setup for consistent JSON logging across all flows in Prefect 3.x. 1) Create config/logging.yml
Copy code
version: 1
disable_existing_loggers: false

formatters:
  json:
    class: prefect.logging.formatters.JsonFormatter

handlers:
  api:
    class: prefect.logging.handlers.APILogHandler
    level: NOTSET
    formatter: json
  json_stdout:
    class: logging.StreamHandler
    stream: <ext://sys.stdout>
    level: NOTSET
    formatter: json

loggers:
  prefect.flow_runs:
    level: NOTSET
    propagate: false
    handlers: [api, json_stdout]
  prefect.task_runs:
    level: NOTSET
    propagate: false
    handlers: [api, json_stdout]
  prefect.extra:
    level: NOTSET
    propagate: false
    handlers: [api, json_stdout]

root:
  level: INFO
  handlers: [json_stdout]
2) Point Prefect to it (in the environment where flows run) - In your docker-compose service that executes the flow:
Copy code
PREFECT_LOGGING_SETTINGS_PATH=/workspace/config/logging.yml
3) In flows/tasks, use get_run_logger for run-scoped logs
Copy code
from prefect import flow, task, get_run_logger

@task
def t():
    log = get_run_logger()
    <http://log.info|log.info>("hello from task")

@flow
def my_flow():
    t()

if __name__ == "__main__":
    my_flow()
Notes - Do not call 
logging.basicConfig(..., force=True)
after Prefect starts; it will remove APILogHandler. - This config sends logs to: - Prefect UI (via APILogHandler) in JSON format - Stdout (json_stdout) in JSON format References (source code) - APILogHandler - JsonFormatter - Logging setup
j
@Marvin I hve followed your config and its still not working, maybe my docker compose is not right?
Copy code
prefect-server:
    image: prefecthq/prefect:3-latest
    container_name: prefect-server
    command: prefect server start
    environment:
      # Core Prefect settings
      PREFECT_SERVER_API_HOST: "0.0.0.0"
      PREFECT_SERVER_API_PORT: "4200"
      PREFECT_API_URL: "<http://127.0.0.1:4200/api>"
      # Application configuration
      ENVIRONMENT: "local"
      USE_API_MOCKS: "true"
    ports:
      # Prefect UI and API
      - "4200:4200"
    volumes:
      # Persistent storage for Prefect database and configuration
      - prefect-server-data:/root/.prefect
    networks:
      - prefect-network
    healthcheck:
      # Use Prefect's documented health endpoint to determine readiness
      test:
        [
          "CMD-SHELL",
          "python - << 'EOF'\nimport http.client, sys\nconn = http.client.HTTPConnection('localhost', 4200, timeout=5)\ntry:\n    conn.request('GET', '/api/health')\n    resp = conn.getresponse()\n    sys.exit(0 if resp.status == 200 else 1)\nexcept Exception:\n    sys.exit(1)\nfinally:\n    conn.close()\nEOF",
        ]
      interval: 10s
      timeout: 10s
      retries: 6
      start_period: 30s

  # Prefect Worker (runs flows)
  prefect-worker:
    build:
      context: .
      dockerfile: Dockerfile.prefect
    container_name: prefect-worker
    command: prefect worker start --pool "default-work-pool" --type "process"
    env_file:
      - .env
    environment:
      # Prefect API connection (overrides .env if set there)
      PREFECT_API_URL: "<http://prefect-server:4200/api>"
      # Logging configuration - use YAML config to preserve APILogHandler
      PREFECT_LOGGING_SETTINGS_PATH: "/workspace/config/logging.yml"
    depends_on:
      prefect-server:
        condition: service_healthy
    networks:
      - prefect-network
    volumes:
      # Mount the application code for development (overrides COPY in Dockerfile)
      - .:/workspace
      # Shared storage for flow artifacts
      - prefect-worker-data:/root/.prefect
I See logs in the prefect-worker stdout but they are not being sent to the UI still
m
✅ thought for 143.6 seconds
Thanks for sharing the compose. Given you’re seeing JSON on worker stdout but nothing in the UI, the problem is almost always one of: - APILogHandler not actually attached in the flow run process, or - APILogHandler is attached but failing to send logs to the API (connection/HTTP error), or - Something is resetting logging after Prefect config is applied (e.g., basicConfig(force=True)). Let’s quickly pinpoint which it is. 1) Verify handlers are attached inside the flow/task process Run this at the top of a task to confirm APILogHandler is present:
Copy code
import logging
from prefect.settings import PREFECT_LOGGING_TO_API_ENABLED
from prefect.settings import PREFECT_API_URL

def debug_handlers():
    for n in ["prefect.flow_runs", "prefect.task_runs", "prefect.extra"]:
        l = logging.getLogger(n)
        print(n, [type(h).__name__ for h in l.handlers], "propagate=", l.propagate)
    print("to_api_enabled =", PREFECT_LOGGING_TO_API_ENABLED.value())
    print("api_url =", PREFECT_API_URL.value())

debug_handlers()
Expected: both 
prefect.flow_runs
and 
prefect.task_runs
show 
APILogHandler
in the list. If they don’t, something (often 
logging.basicConfig(..., force=True)
) is wiping handlers after Prefect starts. Search your codebase and dependencies for 
basicConfig(
, or explicitly re-apply Prefect logging at flow start:
Copy code
from prefect.logging.configuration import setup_logging as setup_prefect_logging
setup_prefect_logging(incremental=False)
2) Enable internal logging to surface send errors Add this to your worker service env and restart:
Copy code
PREFECT_LOGGING_INTERNAL_LEVEL=DEBUG
PREFECT_LOGGING_TO_API_ENABLED=true
If APILogHandler is failing to POST to the API, you’ll see warnings/errors in worker logs. 3) Confirm worker can reach the API From the worker container: - Health:
Copy code
docker exec -it prefect-worker curl -s -o /dev/null -w "%{http_code}\n" <http://prefect-server:4200/api/health>
Expected: 200 - In your task debug above, ensure 
api_url
prints 
<http://prefect-server:4200/api>
(not 127.0.0.1), which you already set in compose. 4) Use the simplest logging.yml (no custom types) Make sure you’re using exactly this (and that the file exists at /workspace/config/logging.yml inside the worker):
Copy code
version: 1
disable_existing_loggers: false

formatters:
  json:
    class: prefect.logging.formatters.JsonFormatter

handlers:
  api:
    class: prefect.logging.handlers.APILogHandler
    level: NOTSET
    formatter: json
  json_stdout:
    class: logging.StreamHandler
    stream: <ext://sys.stdout>
    level: NOTSET
    formatter: json

loggers:
  prefect.flow_runs:
    level: NOTSET
    propagate: false
    handlers: [api, json_stdout]
  prefect.task_runs:
    level: NOTSET
    propagate: false
    handlers: [api, json_stdout]
  prefect.extra:
    level: NOTSET
    propagate: false
    handlers: [api, json_stdout]

root:
  level: INFO
  handlers: [json_stdout]
And in compose (worker only):
Copy code
environment:
  PREFECT_API_URL: "<http://prefect-server:4200/api>"
  PREFECT_LOGGING_SETTINGS_PATH: "/workspace/config/logging.yml"
  PREFECT_LOGGING_TO_API_ENABLED: "true"
  PREFECT_LOGGING_INTERNAL_LEVEL: "DEBUG"
5) Minimal code test Use get_run_logger to ensure you’re emitting on the run loggers:
Copy code
from prefect import flow, task, get_run_logger

@task
def t():
    log = get_run_logger()
    <http://log.info|log.info>("hello from task")

@flow
def f():
    t()

if __name__ == "__main__":
    f()
6) Check server logs for /logs errors Look at the prefect-server container logs while running a flow. If APILogHandler is sending, you may see POST /logs with 4xx/5xx if there’s a server-side issue. If after this: - Handlers are present but logs still not in UI: paste the worker logs with PREFECT_LOGGING_INTERNAL_LEVEL=DEBUG. APILogWorker will log why it’s dropping/failing (e.g., connectivity, payload error).
- Handlers are missing: something is resetting logging. Search for 
basicConfig(
or any code modifying 
prefect.*
loggers. As a workaround, call 
setup_prefect_logging(incremental=False)
at the start of your flow to restore the config. References (source) - APILogHandler and APILogWorker - Logging setup and YAML loading - JsonFormatter Send me the output of step 1 and any APILogWorker messages from step 2, and I’ll pinpoint the next step.
j
@Marvin logs:
Copy code
prefect.flow_runs [] propagate= True

prefect.task_runs [] propagate= True

prefect.extra [] propagate= True

to_api_enabled = True

api_url = <http://prefect-server:4200/api>⁠
server logs (only see this error)
Copy code
sqlalchemy.exc.OperationalError: (sqlite3.OperationalError) database is locked

[SQL: UPDATE deployment SET last_polled=:last_polled, status=:status, updated=strftime('%Y-%m-%d %H:%M:%f000', 'now') WHERE deployment.id IN (SELECT 1 FROM (SELECT 1) WHERE 1!=1) OR deployment.work_queue_id IN (:work_queue_id_1_1)]

[parameters: {'last_polled': '2025-11-29 05:06:23.209285', 'status': 'READY', 'work_queue_id_1_1': '3c7fd78a-4307-4a5a-9981-55a5e2d218b3'}]
m
✅ thought for 41.2 seconds
Thanks — the debug confirms Prefect’s run loggers still have no handlers attached: - prefect.flow_runs handlers=[] - prefect.task_runs handlers=[] That means the APILogHandler is getting removed or never applied inside the flow process. Fast path to green Add this at the very start of your flow (temporarily) to re-apply Prefect’s logging after anything else might have reset it:
Copy code
from prefect.logging.configuration import setup_logging as setup_prefect_logging

@flow
def my_flow():
    setup_prefect_logging(incremental=False)  # reattach APILogHandler
    log = get_run_logger()
    <http://log.info|log.info>("hello from flow")
If logs show up in the UI after this, something in your app (or a library) is calling 
logging.basicConfig(..., force=True)
or otherwise reconfiguring logging. Keep the YAML and remove that code, or always call 
setup_prefect_logging(incremental=False)
right after it. Double-check your minimal YAML and env in the worker - logging.yml (keep it exactly this while testing):
Copy code
version: 1
disable_existing_loggers: false

formatters:
  json:
    class: prefect.logging.formatters.JsonFormatter

handlers:
  api:
    class: prefect.logging.handlers.APILogHandler
    level: NOTSET
    formatter: json
  json_stdout:
    class: logging.StreamHandler
    stream: <ext://sys.stdout>
    level: NOTSET
    formatter: json

loggers:
  prefect.flow_runs:
    level: NOTSET
    propagate: false
    handlers: [api, json_stdout]
  prefect.task_runs:
    level: NOTSET
    propagate: false
    handlers: [api, json_stdout]
  prefect.extra:
    level: NOTSET
    propagate: false
    handlers: [api, json_stdout]

root:
  level: INFO
  handlers: [json_stdout]
- docker-compose (worker):
Copy code
environment:
  PREFECT_API_URL: "<http://prefect-server:4200/api>"
  PREFECT_LOGGING_SETTINGS_PATH: "/workspace/config/logging.yml"
  PREFECT_LOGGING_TO_API_ENABLED: "true"
  PREFECT_LOGGING_INTERNAL_LEVEL: "DEBUG"
- Ensure the file exists in the worker container at that path:
Copy code
docker exec -it prefect-worker ls -l /workspace/config/logging.yml
Run-time checks inside a task after applying setup_prefect_logging
Copy code
import logging
from prefect.settings import PREFECT_LOGGING_TO_API_ENABLED, PREFECT_API_URL

for n in ["prefect.flow_runs", "prefect.task_runs", "prefect.extra"]:
    l = logging.getLogger(n)
    print(n, [type(h).__name__ for h in l.handlers], "propagate=", l.propagate)
print("to_api_enabled =", PREFECT_LOGGING_TO_API_ENABLED.value())
print("api_url =", PREFECT_API_URL.value())
Expected: APILogHandler shows up on flow/task loggers. About the server “database is locked” error That’s from SQLite and can cause log writes to fail even when APILogHandler is attached. For local/dev it can be transient, but if you see it frequently: - Reduce concurrent operations or restarts - Or switch your Prefect Server to Postgres for reliability Why this is happening - Seeing JSON on worker stdout means your root handler is active, but run loggers have no handlers. Most often, a call to 
logging.basicConfig(force=True)
(or a library calling it) runs after Prefect sets up logging and wipes handlers. Re-applying Prefect’s logging fixes it immediately. If the explicit 
setup_prefect_logging(incremental=False)
in the flow makes logs appear in the UI, we can next locate and remove the code that resets logging. If it still doesn’t, please paste the handler debug output after that call and any APILogWorker messages (with 
PREFECT_LOGGING_INTERNAL_LEVEL=DEBUG
) from the worker logs. References: - APILogHandler and APILogWorker - Logging setup and YAML loading - JsonFormatter
j
@Marvin no idea heres my code im running in my task - handlers still not being added after I setup logging.
Copy code
@task(name="fetch_sec_tickers", retries=2)
async def fetch_sec_tickers() -> bytes:
    """Fetch SEC company tickers JSON.

    Returns:
        Raw JSON bytes from SEC

    Raises:
        IngestionNetworkError: If fetch fails after retries
        ValueError: If response indicates an error (non-2xx status)
    """
    import logging
    from prefect.settings import PREFECT_LOGGING_TO_API_ENABLED
    from prefect.settings import PREFECT_API_URL

    def debug_handlers():
        for n in ["prefect.flow_runs", "prefect.task_runs", "prefect.extra"]:
            l = logging.getLogger(n)
            print(n, [type(h).__name__ for h in l.handlers], "propagate=", l.propagate)
        print("to_api_enabled =", PREFECT_LOGGING_TO_API_ENABLED.value())
        print("api_url =", PREFECT_API_URL.value())

    debug_handlers()

    from prefect.logging.configuration import setup_logging as setup_prefect_logging
    setup_prefect_logging(incremental=False)  # reattach APILogHandler
    log = get_run_logger()
    <http://log.info|log.info>("hello from flow")
    debug_handlers()

    logger = get_run_logger()
    url = "<https://www.sec.gov/files/company_tickers.json>"
    settings = get_settings()
Logs
Copy code
prefect.flow_runs [] propagate= True

prefect.task_runs [] propagate= True

prefect.extra [] propagate= True

to_api_enabled = True

api_url = <http://prefect-server:4200/api>⁠

prefect.flow_runs [] propagate= True

prefect.task_runs [] propagate= True

prefect.extra [] propagate= True

to_api_enabled = True

api_url = <http://prefect-server:4200/api>⁠
m
✅ thought for 48.4 seconds
Thanks for the snippet — that output tells us Prefect’s logging config is still not being applied inside the run process. Even calling 
setup_prefect_logging(incremental=False)
didn’t attach handlers, which means the YAML isn’t being applied (or 
dictConfig
is erroring before it can). Let’s force-verify and unblock you with the smallest possible diagnostic that also acts as a fix. Step 1 — Apply dictConfig directly inside the task (diagnostic) Run this at the start of your task, then re-check handlers. This bypasses Prefect’s loader and proves whether the YAML content itself works in your environment.
Copy code
import logging, logging.config

CONFIG = {
  "version": 1,
  "disable_existing_loggers": False,
  "formatters": {
    "json": {"class": "prefect.logging.formatters.JsonFormatter"}
  },
  "handlers": {
    "api": {
      "class": "prefect.logging.handlers.APILogHandler",
      "level": "NOTSET",
      "formatter": "json"
    },
    "json_stdout": {
      "class": "logging.StreamHandler",
      "stream": "<ext://sys.stdout>",
      "level": "NOTSET",
      "formatter": "json"
    }
  },
  "loggers": {
    "prefect.flow_runs": {"level": "NOTSET", "propagate": False, "handlers": ["api","json_stdout"]},
    "prefect.task_runs": {"level": "NOTSET", "propagate": False, "handlers": ["api","json_stdout"]},
    "prefect.extra":     {"level": "NOTSET", "propagate": False, "handlers": ["api","json_stdout"]},
  },
  "root": {"level": "INFO", "handlers": ["json_stdout"]}
}

def apply_and_debug():
    try:
        logging.config.dictConfig(CONFIG)
        for n in ["prefect.flow_runs", "prefect.task_runs", "prefect.extra"]:
            l = logging.getLogger(n)
            print(n, [type(h).__name__ for h in l.handlers], "propagate=", l.propagate)
    except Exception as e:
        print("dictConfig FAILED:", repr(e))

apply_and_debug()
- Expected: you’ll see 
APILogHandler
on 
prefect.flow_runs
and 
prefect.task_runs
. If yes, emit a test log with 
get_run_logger()
and confirm it hits the UI. - If this fails, the exception text will tell us exactly why (e.g., import error, YAML-equivalent issue, etc.). Step 2 — If dictConfig works, but setup_logging doesn’t, print what Prefect is loading This will show what config Prefect is trying to apply and the path it’s using:
Copy code
from prefect.logging.configuration import setup_logging as setup_prefect_logging
from prefect.settings import PREFECT_LOGGING_SETTINGS_PATH

cfg = setup_prefect_logging(incremental=False)
print("settings_path =", PREFECT_LOGGING_SETTINGS_PATH.value())
print("loaded_keys =", list(cfg.keys()))
print("flow_runs_cfg =", cfg.get("loggers",{}).get("prefect.flow_runs"))
- If 
cfg
is empty or does not contain the expected loggers/handlers, Prefect isn’t loading your YAML (or it’s loading a different file). - If 
cfg
looks correct yet handlers don’t attach, something is reconfiguring logging after Prefect runs (e.g., 
logging.basicConfig(force=True)
from your code or a dependency). In that case, keep calling 
setup_prefect_logging(incremental=False)
right after that code, or remove the offending call. Step 3 — Confirm the file in the container Double-check the file actually present and matches the minimal config:
Copy code
docker exec -it prefect-worker sh -lc 'ls -l /workspace/config/logging.yml && echo "---" && cat /workspace/config/logging.yml'
It should match the minimal config we’ve been using (built-in 
JsonFormatter
, 
APILogHandler
on `prefect.flow_runs`/`prefect.task_runs`, 
<ext://sys.stdout>
). Step 4 — Keep internal logging on for clues In the worker container env: - 
PREFECT_LOGGING_INTERNAL_LEVEL=DEBUG
- 
PREFECT_LOGGING_TO_API_ENABLED=true
You should see errors if 
APILogHandler
is failing to send. Note on the server “database is locked”
- SQLite “database is locked” can cause API writes (including logs) to fail intermittently. Once handlers are attached, if you still don’t see logs and your worker shows APILogHandler send failures, this could be the cause. For local dev, reduce concurrent activity or consider switching the server to Postgres for reliability. Source links for reference - APILogHandler and APILogWorker - Logging setup and YAML loading - JsonFormatter Please run Step 1 and paste the handler output (and any exception). That will tell us immediately whether the config can be applied in your run process or if something else is resetting it.
j
@Marvin you were correct
Copy code
dictConfig FAILED: ValueError("Unable to configure formatter 'json'")
m
✅ thought for 156.6 seconds
Boom — we found it. The error is because the formatter config is wrong for Prefect’s JsonFormatter. Fix your YAML/dictConfig to declare the formatter like this: - Use class: prefect.logging.formatters.JsonFormatter - Include format: default (or pretty) - Don’t use the "()" key for this formatter Minimal working YAML
Copy code
version: 1
disable_existing_loggers: false

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

handlers:
  api:
    class: prefect.logging.handlers.APILogHandler
    level: NOTSET
    formatter: json
  json_stdout:
    class: logging.StreamHandler
    stream: <ext://sys.stdout>
    level: NOTSET
    formatter: json

loggers:
  prefect.flow_runs:
    level: NOTSET
    propagate: false
    handlers: [api, json_stdout]
  prefect.task_runs:
    level: NOTSET
    propagate: false
    handlers: [api, json_stdout]
  prefect.extra:
    level: NOTSET
    propagate: false
    handlers: [api, json_stdout]

root:
  level: INFO
  handlers: [json_stdout]
If you prefer dictConfig in code, the equivalent is:
Copy code
'formatters': {
  'json': {
    'class': 'prefect.logging.formatters.JsonFormatter',
    'format': 'default',
  }
}
After updating: 1) Restart the worker so the env and YAML are loaded cleanly. 2) In a task, print handlers again:
Copy code
import logging
for n in ["prefect.flow_runs", "prefect.task_runs", "prefect.extra"]:
    l = logging.getLogger(n)
    print(n, [type(h).__name__ for h in l.handlers], "propagate=", l.propagate)
You should now see APILogHandler attached. Then 
get_run_logger().info("hello")
should appear in the UI. Reference (for why): JsonFormatter signature shows it expects a fmt argument mapped from 
format
when using the 
class
key.
j
@Marvin - it is working now - however, the UI json looks unreadable...ex
Copy code
{
  "name": "prefect.flow_runs.worker",
  "msg": "Completed submission of flow run '4eebfb6a-2e61-42c1-b9e2-757d5070808f'",
  "args": [],
  "levelname": "INFO",
  "levelno": 20,
  "pathname": "/opt/venv/lib/python3.14/site-packages/prefect/workers/base.py",
  "filename": "base.py",
  "module": "base",
  "exc_info": null,
  "exc_text": null,
  "stack_info": null,
  "lineno": 1267,
  "funcName": "_submit_run",
  "created": 1764394262.812697,
  "msecs": 812,
  "relativeCreated": 13032.892131,
  "thread": 281472865804320,
  "threadName": "MainThread",
  "processName": "MainProcess",
  "process": 6,
  "taskName": "prefect.workers.base.BaseWorker._submit_run",
  "flow_run_name": "magenta-starling",
  "flow_run_id": "4eebfb6a-2e61-42c1-b9e2-757d5070808f",
  "flow_name": "<unknown>",
  "worker_name": "ProcessWorker f71f1502-a085-4922-93d4-5b01f27871c2",
  "work_pool_name": "default-work-pool",
  "work_pool_id": "97c3cee9-9b40-4e86-9a5c-66023b631bde",
  "severity": "INFO"
}
how can I make it better
m
✅ thought for 25.6 seconds
You’re seeing that because you attached JsonFormatter to the API handler. APILogHandler uses the handler’s formatter to generate the message string that’s sent to the UI—so you’re sending the entire LogRecord as JSON as the “message”. Fix: use a simple text formatter (or no formatter) on the API handler, and keep JsonFormatter only on stdout. Drop-in YAML update
Copy code
version: 1
disable_existing_loggers: false

formatters:
  # JSON for stdout (aggregators, etc.)
  json:
    class: prefect.logging.formatters.JsonFormatter
    format: default   # or 'pretty' for indented stdout

  # Message-only for the Prefect UI
  message:
    class: logging.Formatter
    format: "%(message)s"

handlers:
  # Send clean messages to the Prefect UI
  api:
    class: prefect.logging.handlers.APILogHandler
    level: NOTSET
    formatter: message   # or omit 'formatter' to use message-only default

  # Structured JSON to stdout
  json_stdout:
    class: logging.StreamHandler
    stream: <ext://sys.stdout>
    level: NOTSET
    formatter: json

loggers:
  prefect.flow_runs:
    level: NOTSET
    propagate: false
    handlers: [api, json_stdout]
  prefect.task_runs:
    level: NOTSET
    propagate: false
    handlers: [api, json_stdout]
  prefect.extra:
    level: NOTSET
    propagate: false
    handlers: [api, json_stdout]

root:
  level: INFO
  handlers: [json_stdout]
This keeps: - UI logs readable (just your message text) - Worker/container logs structured JSON on stdout Tip: If you really want JSON in the UI, you can set the API handler formatter to JsonFormatter with 
format: pretty
, but the UI doesn’t pretty-print; it will still be a big blob of text. The message-only formatter is the cleanest UX. Restart the worker after changing the YAML.
18 Views
Open in Slack
PreviousNext
Powered by