Kubeflow Pipeline Run Audit with Failure and Stuck Run Alerts

id: kubeflow-pipeline-audit
namespace: company.team

description: |
  Kubeflow runs fail quietly, hang in the running state for hours, or finish
  without publishing the artifact downstream jobs expect, and the team notices
  only when the next training job starves. This flow starts from a webhook
  whenever a pipeline run completes, reads the pipeline list with its recent
  run records from the Kubeflow Pipelines API, and audits them in one Python
  task that measures the failure rate over the window, flags runs stuck past
  the threshold, and checks artifacts. A failed or stuck run opens a GitHub
  issue, pings Slack, and logs a note per failed run, while a clean window is
  logged healthy; the verdict and counts are returned as typed outputs.

triggers:
  - id: pipeline_run_completed
    type: io.kestra.plugin.core.trigger.Webhook
    description: Receives the completion event posted by the Kubeflow run notifier
      so the audit starts as soon as a pipeline run finishes.
    key: "{{ secret('KUBEFLOW_WEBHOOK_KEY') }}"
  - id: daily_audit_sweep
    type: io.kestra.plugin.core.trigger.Schedule
    description: Disabled daily net that would re-audit the same window every
      morning until someone switches it on.
    cron: "0 6 * * *"
    disabled: true

inputs:
  - id: kfp_api_base
    type: STRING
    defaults: https://kubeflow.example.com/pipeline
    description: Base url of the Kubeflow Pipelines deployment that the audit calls,
      to which the version path of the pipeline list request is appended.
  - id: stuck_hours
    type: INT
    defaults: 3
    description: Hours a run may stay in the running state before the audit counts
      it as stuck and the escalation gate opens.
  - id: window_hours
    type: INT
    defaults: 24
    description: Lookback window that decides which runs enter the audit and
      therefore the failure rate reported for the window.

tasks:
  # --- 1. recent pipeline runs ---
  - id: fetch_recent_runs
    type: io.kestra.plugin.core.http.Request
    description: Reads the pipeline list with its recent run records from the
      Kubeflow Pipelines API so the audit works from live status data.
    method: GET
    uri: "{{ inputs.kfp_api_base }}/apis/v1/pipelines"
    headers:
      Authorization: "Bearer {{ secret('KUBEFLOW_TOKEN') }}"
      Accept: "application/json"
    retry:
      type: exponential
      interval: PT2S
      maxInterval: PT15S
      maxAttempts: 3

  # --- 2. audit verdict ---
  - id: audit_runs
    type: io.kestra.plugin.scripts.python.Script
    description: Unwraps the staged run records, counts failures, stuck runs and
      artifact gaps, and prints the verdict payload the gate branches on.
    containerImage: python:3.11-slim
    env:
      STUCK_HOURS: "{{ inputs.stuck_hours }}"
      WINDOW_HOURS: "{{ inputs.window_hours }}"
    inputFiles:
      runs.json: "{{ outputs.fetch_recent_runs.body | toJson }}"
    script: |
      import json
      import os
      from datetime import datetime, timedelta, timezone

      RUNS_FILE = "runs.json"
      NOW = datetime.now(timezone.utc)
      SUCCEEDED = {"succeeded", "success", "completed", "succeeded_with_warnings"}
      FAILED = {"failed", "failure", "error", "errored"}
      ACTIVE = {"running", "pending", "queued", "in_progress", "inprogress"}
      ENVELOPE_KEYS = ("runs", "items", "pipelines", "results", "entries", "data")

      def stamp(hours_ago):
          return (NOW - timedelta(hours=hours_ago)).strftime("%Y-%m-%dT%H:%M:%SZ")

      def fallback_runs():
          return [
              {"id": "run-4821", "name": "nightly-train", "status": "succeeded", "started_at": stamp(7), "finished_at": stamp(6), "artifacts": ["model.bin"]},
              {"id": "run-4822", "name": "nightly-train", "status": "succeeded", "started_at": stamp(6), "finished_at": stamp(5), "artifacts": ["metrics.json"]},
              {"id": "run-4823", "name": "nightly-eval", "status": "succeeded", "started_at": stamp(5), "finished_at": stamp(4), "artifacts": ["report.html"]},
              {"id": "run-4824", "name": "nightly-eval", "status": "failed", "started_at": stamp(5), "finished_at": stamp(4), "artifacts": [], "message": "image pull backoff"},
              {"id": "run-4825", "name": "feature-build", "status": "failed", "started_at": stamp(4), "finished_at": stamp(4), "artifacts": [], "message": "pod evicted"},
              {"id": "run-4826", "name": "feature-build", "status": "running", "started_at": stamp(4), "artifacts": []},
          ]

      def decode(raw):
          if not raw:
              return None
          try:
              data = json.loads(raw)
          except ValueError:
              return None
          while isinstance(data, str):
              try:
                  data = json.loads(data)
              except ValueError:
                  return None
          return data

      def status_of(run):
          for key in ("status", "state", "phase"):
              value = run.get(key)
              if isinstance(value, str) and value.strip():
                  return value.strip().lower().replace("-", "_").replace(" ", "_")
          return ""

      def flatten(payload):
          if isinstance(payload, dict):
              for key in ENVELOPE_KEYS:
                  value = payload.get(key)
                  if isinstance(value, (list, dict)):
                      return flatten(value)
              return [payload] if status_of(payload) else None
          if not isinstance(payload, list):
              return None
          if not payload:
              return []
          rows = []
          for entry in payload:
              if isinstance(entry, dict) and isinstance(entry.get("runs"), list):
                  rows.extend([row for row in entry["runs"] if isinstance(row, dict)])
              elif isinstance(entry, dict):
                  rows.append(entry)
          return rows or None

      def parse_time(value):
          if not isinstance(value, str) or not value.strip():
              return None
          raw = value.strip()
          if raw.endswith(("Z", "z")):
              raw = raw[:-1] + "+00:00"
          try:
              parsed = datetime.fromisoformat(raw)
          except ValueError:
              return None
          return parsed if parsed.tzinfo else parsed.replace(tzinfo=timezone.utc)

      def first_time(run, keys):
          for key in keys:
              parsed = parse_time(run.get(key))
              if parsed is not None:
                  return parsed
          return None

      def has_artifact(run):
          if run.get("artifact_missing") is True or run.get("has_artifact") is False:
              return False
          value = run.get("artifacts")
          if value is None:
              value = run.get("artifact_count", 1)
          if isinstance(value, list):
              return any(bool(item) for item in value)
          if isinstance(value, dict):
              return bool(value)
          return value > 0 if isinstance(value, (int, float)) else bool(value)

      def env_number(name, default):
          try:
              value = float(os.environ.get(name, ""))
          except ValueError:
              return default
          return default if value <= 0 else value

      def main():
          stuck_hours = env_number("STUCK_HOURS", 3.0)
          window_hours = env_number("WINDOW_HOURS", 24.0)
          try:
              with open(RUNS_FILE, "r", encoding="utf-8") as handle:
                  raw = handle.read().strip()
          except OSError:
              raw = ""
          payload = decode(raw)
          rows = flatten(payload) if payload is not None else None
          if rows is None:
              rows = fallback_runs()

          audited = []
          for row in rows:
              if not isinstance(row, dict):
                  continue
              status = status_of(row)
              if not status:
                  continue
              started = first_time(row, ("started_at", "start_time", "startTime", "started", "created_at"))
              finished = first_time(row, ("finished_at", "end_time", "finishTime", "finished", "completed_at"))
              reference = started if status in ACTIVE else (finished or started)
              if reference is not None and (NOW - reference).total_seconds() / 3600.0 > window_hours:
                  continue
              audited.append((row, status, started))

          failed_count = 0
          stuck_count = 0
          failed_runs = []
          for index, item in enumerate(audited, 1):
              row, status, started = item
              run_id = str(row.get("id") or row.get("run_id") or ("run-" + str(index)))
              run_name = str(row.get("name") or row.get("pipeline") or "unknown-pipeline")
              reason = None
              if status in FAILED:
                  reason = "reported status " + status
              elif status in SUCCEEDED and not has_artifact(row):
                  reason = "finished without publishing an artifact"
              if reason is not None:
                  failed_count += 1
                  failed_runs.append({"id": run_id, "name": run_name, "status": status, "reason": reason})
                  continue
              if status in ACTIVE and started is not None and (NOW - started).total_seconds() / 3600.0 > stuck_hours:
                  stuck_count += 1

          total = len(audited)
          failure_rate = round(float(failed_count) / float(total), 4) if total else 0.0
          if not total:
              verdict = "NO_RUNS"
          elif failed_count and stuck_count:
              verdict = "FAILED_AND_STUCK"
          elif failed_count:
              verdict = "FAILED_RUNS"
          elif stuck_count:
              verdict = "STUCK_RUNS"
          else:
              verdict = "HEALTHY"

          result = {"verdict": verdict, "failed_count": failed_count, "stuck_count": stuck_count,
                    "runs": total, "failure_rate": failure_rate, "failed_runs": failed_runs}
          print("::" + json.dumps({"outputs": result}) + "::")

      main()

  # --- 3. the escalation gate ---
  - id: audit_gate
    type: io.kestra.plugin.core.flow.If
    description: Opens when the window holds at least one failed run or one run
      still running past the stuck threshold, otherwise records a clean audit.
    condition: "{{ outputs.audit_runs.vars.failed_count > 0 or
      outputs.audit_runs.vars.stuck_count > 0 }}"
    then:
      - id: open_failure_issue
        type: io.kestra.plugin.github.issues.Create
        description: Files the audit finding in the ops backlog with the verdict, the
          counts, the failure rate and the failed run ids.
        oauthToken: "{{ secret('GITHUB_TOKEN') }}"
        repository: "acme-inc/ml-platform"
        title: "Kubeflow audit found {{ outputs.audit_runs.vars.failed_count }} failed
          and {{ outputs.audit_runs.vars.stuck_count }} stuck runs"
        body: |
          ## Kubeflow pipeline audit

          - Verdict: {{ outputs.audit_runs.vars.verdict }}
          - Runs audited in the last {{ inputs.window_hours }} hours: {{ outputs.audit_runs.vars.runs }}
          - Failed runs: {{ outputs.audit_runs.vars.failed_count }}
          - Stuck runs past {{ inputs.stuck_hours }} hours: {{ outputs.audit_runs.vars.stuck_count }}
          - Failure rate over the window: {{ outputs.audit_runs.vars.failure_rate }}
          - Failed run detail: {{ outputs.audit_runs.vars.failed_runs | toJson }}

          Rerun the failing pipelines after checking their images, volumes and
          quotas, and cancel the runs that have been running past the stuck
          threshold so the queue drains again.
        labels:
          - kubeflow
          - pipelines
      - id: notify_slack
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Pings the mlops channel so an owner reruns or cancels the offending
          pipelines before the next scheduled run.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {"channel": "#mlops-ops", "text": "Kubeflow audit {{ outputs.audit_runs.vars.verdict }} for flow {{ flow.id }}: {{ outputs.audit_runs.vars.failed_count }} failed and {{ outputs.audit_runs.vars.stuck_count }} stuck of {{ outputs.audit_runs.vars.runs }} runs, failure rate {{ outputs.audit_runs.vars.failure_rate }} over {{ inputs.window_hours }} hours, execution {{ execution.id }}."}
      - id: loop_failed_runs
        type: io.kestra.plugin.core.flow.Loop
        description: Walks every failed run from the audit payload and appends a note
          naming the run, its pipeline and the reason it failed.
        values: "{{ outputs.audit_runs.vars.failed_runs | toJson }}"
        concurrencyLimit: 2
        tasks:
          - id: append_run_note
            type: io.kestra.plugin.core.log.Log
            description: Prints one note per failed run so the execution log keeps the run
              id, pipeline name and failure reason.
            message: "Note for run {{ item.value.id }} of {{ item.value.name }}: {{
              item.value.reason }}."
    else:
      - id: log_healthy
        type: io.kestra.plugin.core.log.Log
        description: Records a clean window where no run failed, nothing published
          without an artifact, and nothing stayed running past the threshold.
        message: "Kubeflow audit healthy for {{ flow.id }}: verdict {{
          outputs.audit_runs.vars.verdict }}, {{ outputs.audit_runs.vars.runs }}
          runs in the last {{ inputs.window_hours }} hours, failure rate {{
          outputs.audit_runs.vars.failure_rate }}, no failed or stuck runs."

errors:
  - id: notify_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Escalates any failure to the ops channel with the execution url.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {"text": "FAILED: {{ flow.id }} {{ execution.id }}"}

outputs:
  - id: verdict
    type: STRING
    value: "{{ outputs.audit_runs.vars.verdict }}"
  - id: failed_count
    type: INT
    value: "{{ outputs.audit_runs.vars.failed_count }}"
  - id: stuck_count
    type: INT
    value: "{{ outputs.audit_runs.vars.stuck_count }}"
  - id: runs
    type: INT
    value: "{{ outputs.audit_runs.vars.runs }}"

Diagram unavailable

We could not build the topology for this blueprint. The flow itself is valid, use the YAML on the left to run it.

Kubeflow runs fail quietly, hang in the running state for hours, or finish without publishing the artifact downstream jobs expect, and nobody notices until the next training job starves. This flow fires from a webhook whenever a run completes, reads the recent run records from the Kubeflow Pipelines API, and audits them in one Python task that measures the failure rate over the window, flags runs stuck past the threshold, and checks artifacts. When a failed or stuck run is found it opens a GitHub issue, pings Slack, and logs a note per failed run; a clean window is logged healthy, and the verdict and counts are returned as typed outputs.

How it works

pipeline_run_completed (Webhook) receives the completion event posted by the Kubeflow run notifier, so the audit starts the moment a pipeline run finishes. daily_audit_sweep is a disabled Schedule kept as a fallback net until someone switches it on.
fetch_recent_runs (io.kestra.plugin.core.http.Request) calls GET /apis/v1/pipelines on the deployment given by kfp_api_base, authenticates with the KUBEFLOW_TOKEN secret, retries on transient failures, and stages the response body as runs.json.
audit_runs (io.kestra.plugin.scripts.python.Script) unwraps runs.json from any of the usual list envelopes, drops completed runs that finished before window_hours ago while always auditing runs that are still active, counts failed runs and successful runs that published no artifact as failures, flags runs still running past stuck_hours, derives the failure rate for the window, and prints one ::{"outputs": ...}:: line carrying verdict, failed_count, stuck_count, runs, failure_rate, and failed_runs. When runs.json is missing, empty, or unparseable, the task falls back to a six run fixture of three succeeded, two failed, and one running for four hours, so the audit never runs blind.
audit_gate (io.kestra.plugin.core.flow.If) opens when failed_count is greater than 0 or stuck_count is greater than 0. Inside it, open_failure_issue (io.kestra.plugin.github.issues.Create) files the finding, notify_slack posts the summary, and loop_failed_runs (io.kestra.plugin.core.flow.Loop) appends one note per failed run through append_run_note. When the gate stays closed, log_healthy records the clean window.

What you get

verdict (STRING): FAILED_AND_STUCK when both problems appear, FAILED_RUNS for failures only, STUCK_RUNS for a stuck run only, HEALTHY for a clean window, NO_RUNS when the window holds no auditable run.
failed_count (INT): runs that reported a failure status, plus runs that succeeded without publishing an artifact.
stuck_count (INT): runs still in the running state for longer than stuck_hours.
runs (INT): runs audited inside window_hours, which is the denominator of the failure rate.
failure_rate (FLOAT) and failed_runs (ARRAY): carried in the payload for the issue body, the Slack summary, and the per-run note loop.

Who it's for

MLOps and platform teams running Kubeflow Pipelines who discover failed or hung runs only when the next job starves, and who want an issue, a channel ping, and an audit trail instead of a silent stall.

Why orchestrate this with Kestra

The API read, the audit, the gate, and the escalations live in one versioned file instead of a cron script nobody owns, and the disabled Schedule is the fallback sweep ready to be switched on.
The condition that opens the gate is visible in the graph, so the failed and stuck thresholds are reviewable rather than buried in a monitoring tool.
The six run fallback fixture keeps the audit runnable when the API is unreachable, while the errors handler still escalates hard failures to Slack with the execution url.

Prerequisites

A Kestra instance with the GitHub and Slack plugins installed and a Python task runner able to pull python:3.11-slim.
A Kubeflow Pipelines deployment reachable over https, plus a notifier that posts run completion events to the webhook trigger.

Secrets

KUBEFLOW_TOKEN: bearer token used to read the pipeline list and its recent runs from the Kubeflow Pipelines API.
KUBEFLOW_WEBHOOK_KEY: key the webhook trigger expects from the run completion notifier.
GITHUB_TOKEN: GitHub token that can create the audit issue in the ops backlog repository.
SLACK_WEBHOOK_URL: Slack incoming webhook used for the audit ping and for failure alerts.

Quick start

Add the four secrets and import the flow into your namespace.
Point the Kubeflow run notifier at the webhook trigger url, or execute the flow manually with your own kfp_api_base, stuck_hours, and window_hours, then read verdict, failed_count, stuck_count, and runs in the outputs.

How to extend

Lower stuck_hours to catch hung runs earlier, or raise window_hours when you want a weekly instead of a daily failure rate.
Add a third input for a failure rate tolerance and compare failure_rate in the gate condition once you only want to escalate after the window degrades.
Point repository and the Slack channel at one backlog and one channel per Kubeflow namespace by templating them from kfp_api_base.

Audit Kubeflow pipeline runs and escalate failures and stuck runs

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

Audit Kubeflow pipeline runs and escalate failures and stuck runs

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Infrastructure Blueprints

Gate a Kestra upgrade against every live flow

Backport Kestra UI hotfixes to a Git review branch

Gate Kestra Subflow Changes Against Every Caller Before Deploying

Retire Unused Kestra Flows with Quarantine, Approval and Restore

Kinesis Stream Throughput Quota Gate

Knative Service Revision Failure and Scale Sentinel

New to Kestra?