How to reliably detect which Flow has failed?

[amirzp]

Hi

We’re looking for a recommended way to detect whether a specific Flow has failed, in a scalable and future-proof way.

---

What we’ve checked so far

1. Metrics

Example:

worker_execution_failed{name="server",tag="default"} 7910

This is useful for alerting and knowing that failures occurred, but:

• It is aggregated
• It does not indicate which Flow or Job failed

---

2. Jobs API

Example:

GET /api/w/{workspace}/jobs/list?script_path_start=...

This allows us to identify failed jobs and see flow/script paths and error messages.
However:

• It is pull-based, meaning we have to continuously query the API to detect failures

Ideally, we would prefer a push-based mechanism, where Windmill actively notifies us when a Flow or Job fails (e.g. via webhook or event), instead of us polling the Jobs API to find out what errored.

---

3. Error handling inside Flows

Handling errors inside Flows (try/catch, error branches, notifications) is helpful for expected errors, but:

• It does not cover crashes or worker-level failures
• It requires per-flow implementation

---

Questions

What approach would you recommend for detecting failed Flows in a way that:

• Works well with alerting
• Identifies the exact Flow
• Scales well for future Flows with minimal changes?

Thanks in advance 🙏

[egvimo]

Maybe a custom script is an option.

• Pull from Windmill Jobs API (filtered)
• Transform into Prometheus metrics with labels
• Expose via HTTP endpoint
• Let Prometheus handle scraping + alerting

import os
from wmill import Windmill


def format_metric(name, value, mtype="gauge", labels=None):
    label_str = ""
    if labels:
        label_str = "{" + ",".join(f'{k}="{v}"' for k, v in labels.items()) + "}"

    return f"# TYPE {name} {mtype}\n{name}{label_str} {value}\n"


def main():
    client = Windmill()
    workspace = os.environ["WM_WORKSPACE"]

    result = client.get(
        f"/w/{workspace}/jobs/list",
        params={
            "success": False,
            "job_kinds": "flow",
        },
    )

    jobs = result.json()

    metrics = ""
    metrics += format_metric(
        "windmill_jobs_total",
        len(jobs),
        labels={"status": "failed", "kind": "flow"},
    )

    return {"windmill_content_type": "text/plain", "result": metrics}