https://prefect.io logo
Join Slack
Powered by
<@ULVA73B9P> Hi, in a very simple hello world flow...
# ask-marvin
l
@Marvin Hi, in a very simple hello world flow, I’d like to have a custom outcome, I found something like 
return Completed(name="custom_success")
works well. However I’m trying to have another custom failure state , but doing 
return Failed(name="custom_failure")
, even though the actual outcome ends up as I want to, with a red “custom_failure” for the run, the log is inconsistent: • if running the flow locally with 
my_flow.serve()
, it works as expected, there’s a graceful log without traceback on the UI, and there is the complete one on the worker, showing a long traceback (which I can be fine with, even if very verbose)
Copy code
10:26:31.101 | ERROR   | prefect.engine - Execution of flow run '757a6a85-ecdb-47d2-b5ca-62c466830a81' exited with unexpected exception
Traceback (most recent call last):
  File "./.venv/lib/python3.12/site-packages/prefect/engine.py", line 57, in handle_engine_signals
    yield
  File "./.venv/lib/python3.12/site-packages/prefect/flow_engine.py", line 1629, in run_flow_with_env
    maybe_coro = run_flow(*args, **kwargs)
                 ^^^^^^^^^^^^^^^^^^^^^^^^^
  File "./.venv/lib/python3.12/site-packages/prefect/flow_engine.py", line 1554, in run_flow
    ret_val = run_flow_sync(**kwargs)
              ^^^^^^^^^^^^^^^^^^^^^^^
  File "./.venv/lib/python3.12/site-packages/prefect/flow_engine.py", line 1399, in run_flow_sync
    return engine.state if return_type == "state" else engine.result()
                                                       ^^^^^^^^^^^^^^^
  File "./.venv/lib/python3.12/site-packages/prefect/flow_engine.py", line 371, in result
    _result = self.state.result(raise_on_failure=raise_on_failure, _sync=True)  # type: ignore
              ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "./.venv/lib/python3.12/site-packages/prefect/_internal/compatibility/async_dispatch.py", line 94, in wrapper
    return fn(*args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^
  File "./.venv/lib/python3.12/site-packages/prefect/client/schemas/objects.py", line 375, in result
    return run_coro_as_sync(
           ^^^^^^^^^^^^^^^^^
  File "./.venv/lib/python3.12/site-packages/prefect/utilities/asyncutils.py", line 207, in run_coro_as_sync
    return call.result()
           ^^^^^^^^^^^^^
  File "./.venv/lib/python3.12/site-packages/prefect/_internal/concurrency/calls.py", line 365, in result
    return self.future.result(timeout=timeout)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "./.venv/lib/python3.12/site-packages/prefect/_internal/concurrency/calls.py", line 192, in result
    return self.__get_result()
           ^^^^^^^^^^^^^^^^^^^
  File "/usr/local/Cellar/python@3.12/3.12.10/Frameworks/Python.framework/Versions/3.12/lib/python3.12/concurrent/futures/_base.py", line 401, in __get_result
    raise self._exception
  File "./.venv/lib/python3.12/site-packages/prefect/_internal/concurrency/calls.py", line 441, in _run_async
    result = await coro
             ^^^^^^^^^^
  File "./.venv/lib/python3.12/site-packages/prefect/utilities/asyncutils.py", line 188, in coroutine_wrapper
    return await task
           ^^^^^^^^^^
  File "./.venv/lib/python3.12/site-packages/prefect/states.py", line 85, in get_state_result
    return await _get_state_result(
           ^^^^^^^^^^^^^^^^^^^^^^^^
  File "./.venv/lib/python3.12/site-packages/prefect/states.py", line 157, in _get_state_result
    raise await get_state_exception(state)
prefect.exceptions.FailedRun: custom failure
10:26:31.345 | ERROR   | prefect.flow_runs.runner - Process for flow run 'granite-okapi' exited with status code: 1
• however, if running the flow on a local process worker pool , deployed through 
prefect.yaml
, I get the exact same log + traceback on the worker, but it also gets printed in the UI log, which looks really bad for an outcome that is expected: some frequent cases in my flow are indeed expected to result in this custom failed state, I don’t want to see the prefect internal machinery traceback showing in that case. Why are the 2 behaviours different, and how can I have a consistent run no matter how deployed, without this kind of traceback ? I’ve noticed from the prefect API that there might be some support for what I want, with 
raise_on_failure
in the 
FlowRunEngine
but it doesn’t look we can leverage this as a user
m
✅ thought for 914.4 seconds
Great question — and you’re not imagining things. In Prefect 3.x, returning a Failed state is supported, but the logging differs depending on how the flow is executed. Why you see different behavior - flow.serve(): Runs your flow in-process. When you return Failed(...), the engine does not go through the subprocess boundary that logs “unexpected exception” with a traceback, so the UI stays clean. You’ll still see a traceback in the local console when the engine retrieves the result and raises a FailedRun, but it’s not sent to the API/UI. - Deployment on a process worker: Runs your flow in a spawned subprocess. When the engine retrieves the result for a failed state, it raises a FailedRun. At the subprocess boundary, Prefect logs an error with traceback (“Execution of flow run ... exited with unexpected exception”) via logger.exception and sends it to the API — which is why it shows up in the UI. Code paths for reference - Returning a Failed state is supported and used as-is: prefect/states.py → return_value_to_state - When a failed state’s “result” is read, Prefect raises a FailedRun: prefect/states.py → get_state_result/aget_state_exception - The subprocess wrapper logs the traceback to API: prefect/engine.py → handle_engine_signals - The flow engine will call engine.result() by default, which triggers the FailedRun raise path: prefect/flow_engine.py → run_flow_sync Links: - Source: states.py, flow_engine.py, engine.py - Docs: States, Logging Is there a user-facing raise_on_failure toggle? - Not for engine execution. raise_on_failure is only honored when you explicitly call state.result(...), not as a global engine setting. There’s also no env var to change the engine’s return_type (“result” vs “state”). How to avoid tracebacks in UI for “expected” failures Option A — Preferable: Adjust logging so engine exceptions aren’t sent to the API - Create a custom logging.yml that excludes API handler for the engine boundary loggers. This keeps your worker console fully verbose (tracebacks still there), but the UI logs stay clean. Example logging.yml
Copy code
version: 1
formatters:
  simple:
    format: "%(asctime)s | %(levelname)-8s | %(name)s - %(message)s"
handlers:
  console:
    class: logging.StreamHandler
    level: INFO
    formatter: simple
    stream: <ext://sys.stdout>
  api:
    class: prefect.logging.handlers.APILogHandler
    level: INFO
    formatter: simple
loggers:
  prefect:
    level: INFO
    handlers: [console, api]
    propagate: false

  # Downgrade engine boundary logs and exclude them from API
  prefect.engine:
    level: WARNING
    handlers: [console]
    propagate: false
  prefect.flow_runs.runner:
    level: WARNING
    handlers: [console]
    propagate: false
  prefect.flow_engine:
    level: WARNING
    handlers: [console]
    propagate: false

root:
  level: INFO
  handlers: [console]
Enable it:
Copy code
# Either put it at ~/.prefect/logging.yml
# or point to it explicitly
export PREFECT_LOGGING_SETTINGS_PATH=/path/to/logging.yml
If you only want a quick level change (but can’t alter handlers), you can reduce the noise with env vars:
Copy code
export PREFECT_LOGGING_LOGGERS_PREFECT_ENGINE_LEVEL=WARNING
export PREFECT_LOGGING_LOGGERS_PREFECT_FLOW_RUNS_RUNNER_LEVEL=WARNING
export PREFECT_LOGGING_LOGGERS_PREFECT_FLOW_ENGINE_LEVEL=WARNING
Note: This won’t exclude those logs from API; it just suppresses them if they’re <= WARNING. To truly exclude from API, use logging.yml to remove the API handler for those loggers. Option B — When calling flows from Python (not via worker), request the state instead of the result ```
state = my_flow(return_state=True) # returns State; does not raise FailedRun ``` This avoids raising FailedRun locally, but it doesn’t change worker/deployment behavior. What you’re asking about — an engine-level raise_on_failure=False — is not currently user-exposed in deployments. Given your use case (expected Failed terminal states without tracebacks), it’s reasonable to track this as a feature request: treat user-returned Failed states as “graceful” and skip traceback logging at the subprocess boundary. If you’d like, I can help open a GitHub issue or discussion summarizing your scenario. A couple of clarifying questions so I can tailor the exact logging.yml: - Can you confirm your Prefect version (e.g., 3.0.x)? - Do you want to suppress just the two logs you pasted (
prefect.engine
“unexpected exception” and 
prefect.flow_runs.runner
“exited with status code 1”), or all engine-level exceptions from the UI? - Are there other loggers you want to keep sending to the API unchanged? Once you confirm, I’ll provide a minimal logging.yml that only filters those specific lines while preserving everything else in the UI.
l
@Marvin I don’t see how your logging tweaks would work given that the emitted log+traceback are at error level, and I think it would be best not to silence these loggers , for other actually relevant errors. About the 
return_state=True
maybe there’s a way to do this a the task level instead (which would then return my custom failure), would that help ? why not about opening an issue, if relevant and not a duplicate of existing one, • Prefect version 3.4.22 • I’d like to know if there could be a best practice to follow to have “expected” custom failure states, that avoid what we mention here
m
✅ thought for 371.9 seconds
Thanks for the detailed report — I can reproduce/trace this in Prefect 3.4.x. What’s going on - Returning 
Failed(name="...")
from a flow is supported and the state is used as-is. - However, when the engine returns the flow’s “result”, it calls 
state.result(raise_on_failure=True)
under the hood. For a Failed state, that raises a 
prefect.exceptions.FailedRun
. - In deployments/workers (subprocess execution), that exception bubbles to the subprocess boundary and Prefect logs it with a traceback (“Execution of flow run … exited with unexpected exception”), which is sent to the API and shows in the UI. - In 
flow.serve()
runs (in-process), you typically won’t see that boundary log in the UI, though you will see the traceback locally. Relevant sources: - states.py (result retrieval raising FailedRun) - flow_engine.py (engine returns result by default) - engine.py (subprocess boundary traceback log) Re: “silencing the logger” You’re right — lowering 
prefect.engine
to WARNING would also hide other important errors. Instead of changing levels, the only way today to keep the UI clean for “expected” Failed states is to selectively drop just those specific boundary error records from the API handler while keeping them in the worker console. Concrete best practice today - Keep returning explicit 
Failed(name="your_custom_name", message="your message")
for expected terminal outcomes. - Add a logging Filter that drops only the subprocess-boundary error for those specific expected failures from the API handler (but not from the console). This avoids hiding unrelated errors. Example filter that drops only FailedRun records that match your custom marker 1) Create a small module importable by your worker, e.g. `my_prefect_filters.py`:
Copy code
import logging
from prefect.exceptions import FailedRun

class SuppressExpectedFailedRun(logging.Filter):
    """
    Drop only the subprocess-boundary error log for expected Failed states,
    so the UI doesn't show a big traceback for those runs.
    """
    def filter(self, record: logging.LogRecord) -> bool:
        # Only apply to engine boundary loggers
        if record.name not in {"prefect.engine", "prefect.flow_runs.runner"}:
            return True

        # Only consider records that have an exception
        if not record.exc_info:
            return True

        exc = record.exc_info[1]
        if not isinstance(exc, FailedRun):
            return True

        # Match your marker in the FailedRun message; customize this as needed
        msg = str(exc)
        # Example: drop logs when your state message contains "custom failure"
        if "custom failure" in msg or "custom_failure" in (record.getMessage() or ""):
            return False  # drop from API handler

        return True  # keep all others
2) In a custom 
logging.yml
used by your worker: ``` version: 1 formatters: simple: format: "%(asctime)s | %(levelname)-8s | %(name)s - %(message)s" filters: suppress_expected_failed_run: (): my_prefect_filters.SuppressExpectedFailedRun handlers: console: class: logging.StreamHandler level: INFO formatter: simple stream: ext://sys.stdout api: class: prefect.logging.handlers.APILogHandler level: INFO formatter: simple # Only this handler gets the filter; console is untouched filters: [suppress_expected_failed_run] loggers: # Keep engine logs at INFO so real errors still show up; # but the API handler will drop only matching ExpectedFailedRun logs. prefect.engine: level: INFO handlers: [console, api] propagate: false prefect.flow_runs.runner: level: INFO handlers: [console, api] propagate: false prefect: level: INFO handlers: [console, api] propagate: false
root: level: INFO handlers: [console]
Copy code
3) Point your worker to this config:
export PREFECT_LOGGING_SETTINGS_PATH=/path/to/logging.yml # Ensure my_prefect_filters.py is importable (PYTHONPATH/site-packages/container image) ``` Result: - Worker console still shows the full traceback for those runs (useful for debugging). - The UI will not receive that specific engine-boundary traceback for “expected” Failed states. - All other ERROR logs continue to flow to the UI. About using return_state=True at the task level - Using 
return_state=True
on tasks plus 
allow_failure(...)
can be great to keep flows running and make “expected failures” first-class data for downstream logic. See source-backed details above and docs: States. - However, it will not change the flow-level behavior you’re hitting: if your flow ultimately returns a Failed state, the deployment/worker engine still retrieves the flow “result” and raises 
FailedRun
, triggering the subprocess-boundary traceback log. - In other words, task-level `return_state`/`allow_failure` helps avoid unexpected upstream errors during the flow, but it doesn’t suppress the final flow-level traceback when the run ends as Failed. If you’re open to a pattern shift - Some teams model “expected no-op/validation failure” as a Completed flow with a domain result (e.g., a status value) plus an artifact or a label, and use automations/notifications on that status instead of a Failed flow state. That keeps the UI green by design and avoids any failure tracebacks. I realize you specifically want the run to be red with a custom name, but mentioning this for completeness. Feature request (what would be ideal) - A deployment/engine option to treat user-supplied terminal Failed states as “graceful”: - Do not call 
state.result(raise_on_failure=True)
at the end (or call it with 
raise_on_failure=False
) - Avoid the subprocess-boundary logger.exception for this case - Or a flow/deployment option to set engine 
return_type="state"
so the engine returns the final state without result retrieval (and without raising/tracebacks) - Either would give you a clean, consistent UI for expected custom Failed states without needing logging filters. I did not find an existing issue that specifically covers “suppress subprocess-boundary traceback when a user returns a Failed state”. If you’d like, I can open this as a feature request on your behalf with: - Prefect 3.4.22 - Repro steps (return 
Failed(name="custom_failure")
in a deployed flow; observe UI traceback) - Desired behavior and rationale (expected failures should not look like engine crashes) - Proposed options (graceful-user-failure flag, or engine return_type/raise_on_failure control) Let me know: - Should I go ahead and file the issue? If yes, do you want me to include your “custom_failure” message as the matching example, or share a different marker? - Do you want a pared-down logging.yml/filter example that matches your exact custom state name/message? References: - states.py (state.result and FailedRun) - flow_engine.py (engine result retrieval) - engine.py (subprocess boundary traceback log) - Logging guide
l
@Marvin ok to open issue if no duplicates exist then
m
✅ thought for 48.3 seconds
Done — I created a GitHub discussion to capture this as a feature request and reference your scenario and version: - Discussion: Suppress traceback in UI for expected custom Failed flow states returned by users (3.4.22) Summary of what I proposed: - When a flow intentionally returns a Failed state (e.g., 
Failed(name="custom_failure")
) via deployments/workers, the engine currently logs a subprocess-boundary traceback that is sent to the API/UI. - Users want these expected failures to show as red with custom names/messages but without long internal stack traces in the UI. - Suggested solutions: an engine/deployment option to treat user-returned Failed as “graceful” (e.g., return final state without calling result/raising), or a flag to suppress the boundary traceback only for this case. I’ll keep an eye on the thread and update you with any guidance from the team. If you want me to add more details (e.g., exact message patterns you use for matching) or a minimal reproducer snippet, I can append it there.
5 Views
Open in Slack
PreviousNext
Powered by