Mux Asset Processing Failure Sentinel for Kestra

id: mux-asset-processing-failure-sentinel
namespace: company.team
description: |
  Every hour, read recent video assets from the Mux Video API (read-only:
  nothing is uploaded, re-encoded or deleted) and classify video processing
  as HEALTHY, WARNING, CRITICAL, LOW_VOLUME or UNREACHABLE. Log quiet runs
  and alert Slack when uploaded videos start failing to process or get stuck
  in preparing, with the most common Mux error.

inputs:
  - id: lookback_hours
    type: INT
    defaults: 24
    displayName: "Lookback Window (hours)"
    description: "Judge assets created in the last this many hours."
  - id: warning_error_rate
    type: FLOAT
    defaults: 5.0
    displayName: "Warning Error Rate (%)"
    description: "At or above this share of errored assets among finished ones, the
      check is WARNING."
  - id: critical_error_rate
    type: FLOAT
    defaults: 20.0
    displayName: "Critical Error Rate (%)"
    description: "At or above this share of errored assets among finished ones, the
      check is CRITICAL."
  - id: min_assets
    type: INT
    defaults: 10
    displayName: "Minimum Assets"
    description: "Below this many finished assets, the rate is too noisy to judge
      and the run is LOW_VOLUME (stuck assets still alert)."
  - id: stuck_minutes
    type: INT
    defaults: 60
    displayName: "Stuck Preparing Threshold (minutes)"
    description: "An asset still preparing after this many minutes counts as stuck
      and makes the check WARNING."

triggers:
  - id: hourly_check
    type: io.kestra.plugin.core.trigger.Schedule
    description: Check video processing every hour. Shipped disabled; enable it once
      the Mux access token secrets are set and a manual run has been verified.
    cron: "35 * * * *"
    disabled: true

tasks:
  - id: list_assets
    type: io.kestra.plugin.core.http.Request
    description: Read the 100 most recent video assets with their status, creation
      time and errors from the Mux Video API. This is a GET; nothing is
      uploaded, re-encoded or deleted.
    uri: "https://api.mux.com/video/v1/assets?limit=100"
    method: GET
    options:
      auth:
        type: BASIC
        username: "{{ secret('MUX_TOKEN_ID') }}"
        password: "{{ secret('MUX_TOKEN_SECRET') }}"
      allowFailed: true
      timeout:
        connectTimeout: PT10S
        readIdleTimeout: PT30S
    retry:
      type: constant
      interval: PT10S
      maxAttempts: 3
    allowFailure: true

  - id: summarize_assets
    type: io.kestra.plugin.core.debug.Return
    description: Keep assets created in the lookback window, compute the share that
      errored among finished ones, find the most common error type and message,
      count assets stuck in preparing past the threshold, and classify in one jq
      query. A missing or non-200 response becomes UNREACHABLE.
    format: >-
      {{ (outputs.list_assets.code is not defined or outputs.list_assets.code !=
      200) ?
      '{"status":"UNREACHABLE","finished":0,"ready":0,"errored":0,"rate":0,"preparing":0,"stuck":0,"top_error":"none","more_pages":false}'
      : (fromJson(outputs.list_assets.body) | jq(' ' ~ (now() |
      dateAdd(-inputs.lookback_hours, 'HOURS') | timestamp) ~ ' as $since | ' ~
      (now() | dateAdd(-inputs.stuck_minutes, 'MINUTES') | timestamp) ~ ' as
      $stuck_before | [(.data // [])[] | select(((.created_at // "0") |
      tonumber) >= $since)] as $a | [$a[] | select(.status == "ready" or .status
      == "errored")] as $f | [$f[] | select(.status == "errored")] as $bad |
      [$a[] | select(.status == "preparing")] as $prep | [$prep[] |
      select(((.created_at // "0") | tonumber) < $stuck_before)] as $stuck | ($f
      | length) as $n | (if $n == 0 then 0 else ((($bad | length) * 1000 / $n) |
      floor) / 10 end) as $r | {finished: $n, ready: ($n - ($bad | length)),
      errored: ($bad | length), rate: $r, preparing: ($prep | length), stuck:
      ($stuck | length), top_error: (if ($bad | length) == 0 then "none" else
      ([$bad[] | (((.errors.type // "unknown") | tostring) + ": " +
      ((.errors.messages // ["no error message"]) | map(tostring) | join(" ")) |
      split("\"") | join("") | split("\n") | join(" ") | .[0:120])] |
      group_by(.) | map({e: .[0], c: length}) | max_by(.c) | (.e + " (x" + (.c |
      tostring) + ")")) end), more_pages: (((.data // []) | length) >= 100 and
      ([(.data // [])[] | select(((.created_at // "0") | tonumber) < $since)] |
      length) == 0), status: (if $n >= ' ~ inputs.min_assets ~ ' and $r >= ' ~
      inputs.critical_error_rate ~ ' then "CRITICAL" elif ($n >= ' ~
      inputs.min_assets ~ ' and $r >= ' ~ inputs.warning_error_rate ~ ') or
      ($stuck | length) > 0 then "WARNING" elif $n < ' ~ inputs.min_assets ~ '
      then "LOW_VOLUME" else "HEALTHY" end)} ') | first | toJson) }}

  - id: route_by_errors
    type: io.kestra.plugin.core.flow.Switch
    description: Route to a quiet log when processing is healthy or volume is too
      low to judge, a Slack message when errors cross the warning line or assets
      are stuck, or a Slack page when errors cross the critical line or Mux
      could not be read.
    value: "{{ fromJson(outputs.summarize_assets.value).status }}"
    cases:
      "HEALTHY":
        - id: log_healthy
          type: io.kestra.plugin.core.log.Log
          description: Log a quiet line when the check is healthy; no Slack message is sent.
          message: "Mux video processing healthy: {{
            fromJson(outputs.summarize_assets.value).errored }} of {{
            fromJson(outputs.summarize_assets.value).finished }} finished assets
            errored ({{ fromJson(outputs.summarize_assets.value).rate }}%) in
            the last {{ inputs.lookback_hours }} hour(s), {{
            fromJson(outputs.summarize_assets.value).preparing }} preparing."
      "LOW_VOLUME":
        - id: log_low_volume
          type: io.kestra.plugin.core.log.Log
          description: Log a quiet line when there are too few finished assets in the
            window to judge; no Slack message is sent.
          message: "Mux video processing: only {{
            fromJson(outputs.summarize_assets.value).finished }} finished
            asset(s) in the last {{ inputs.lookback_hours }} hour(s), below the
            minimum of {{ inputs.min_assets }}; rate not judged this run."
      "WARNING":
        - id: alert_warning_errors
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          description: Post a Slack message when the asset error rate is at or above the
            warning line or assets are stuck in preparing.
          url: "{{ secret('SLACK_WEBHOOK') }}"
          payload: |
            {
              "text": "Mux video processing WARNING: {{ fromJson(outputs.summarize_assets.value).errored }} of {{ fromJson(outputs.summarize_assets.value).finished }} finished assets errored ({{ fromJson(outputs.summarize_assets.value).rate }}%) in the last {{ inputs.lookback_hours }} hour(s), {{ fromJson(outputs.summarize_assets.value).stuck }} asset(s) preparing for more than {{ inputs.stuck_minutes }} minutes. Most common error: {{ fromJson(outputs.summarize_assets.value).top_error }}. Execution {{ execution.id }}."
            }
      "CRITICAL":
        - id: alert_critical_errors
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          description: Page Slack when the asset error rate is at or above the critical
            line, which usually means an upload source, an input format or an
            encoding setting is broken.
          url: "{{ secret('SLACK_WEBHOOK') }}"
          payload: |
            {
              "text": "Mux video processing CRITICAL: {{ fromJson(outputs.summarize_assets.value).errored }} of {{ fromJson(outputs.summarize_assets.value).finished }} finished assets errored ({{ fromJson(outputs.summarize_assets.value).rate }}%) in the last {{ inputs.lookback_hours }} hour(s). Most common error: {{ fromJson(outputs.summarize_assets.value).top_error }}. Uploaded videos are not becoming playable; check the upload source and input settings now. Execution {{ execution.id }}."
            }
      "UNREACHABLE":
        - id: alert_unreachable_api
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          description: Page Slack when the asset list could not be read, indicating an
            outage or an invalid access token.
          url: "{{ secret('SLACK_WEBHOOK') }}"
          payload: |
            {
              "text": "Mux asset sentinel UNREACHABLE: the asset list did not return a usable response (HTTP {{ outputs.list_assets.code ?? 'no response' }}). Check the access token ID and secret and their Mux Video permission. Execution {{ execution.id }}."
            }
    defaults:
      - id: log_unclassified
        type: io.kestra.plugin.core.log.Log
        description: Fallback for any classification the cases above do not cover, so an
          unexpected value is visible in the logs.
        message: "Mux asset sentinel completed with an unexpected classification: {{
          outputs.summarize_assets.value }}."

  - id: log_audit_result
    type: io.kestra.plugin.core.log.Log
    description: Record one audit line for every run, whichever branch fired, so the
      processing history stays queryable even if Slack delivery fails.
    message: "Mux asset sentinel audit: lookback_hours={{ inputs.lookback_hours }}
      http={{ outputs.list_assets.code ?? 'n/a' }} summary={{
      outputs.summarize_assets.value }} execution={{ execution.id }}."

errors:
  - id: alert_flow_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Alert Slack when the sentinel flow itself fails, so a broken check
      is never mistaken for healthy video processing.
    url: "{{ secret('SLACK_WEBHOOK') }}"
    payload: |
      {
        "text": "Mux asset sentinel flow {{ flow.id }} FAILED (execution {{ execution.id }}). Video processing health is now unknown; check the task logs and the Mux dashboard."
      }

outputs:
  - id: status
    type: STRING
    value: "{{ fromJson(outputs.summarize_assets.value).status }}"

A video platform on Mux fails one upload at a time: a creator uploads a file that is not really video, a signed source URL expires before Mux fetches it, or a new client starts sending a format the encoder rejects. Each asset just ends with status: errored and an errors object, the player shows nothing, and support hears about it before engineering does. This blueprint turns a scheduled, read-only call to the Mux Video API into a processing health gate: io.kestra.plugin.core.http.Request reads recent assets, one jq query computes the error rate, the most common error and assets stuck in preparing, and the result routes to a quiet log or a Slack alert.

How it works

list_assets (io.kestra.plugin.core.http.Request) calls GET /video/v1/assets?limit=100 with HTTP basic auth (access token ID and secret), which returns the most recent assets with their status, created_at (Unix seconds) and errors. It is a read: nothing is uploaded, re-encoded or deleted. Retries cover short network blips, and allowFailed plus allowFailure keep an API error from aborting the flow so it can be reported instead.
summarize_assets (io.kestra.plugin.core.debug.Return) first guards for a missing or non-200 response (UNREACHABLE). Otherwise one jq query keeps assets created inside the lookback window, treats ready and errored as finished, computes the share that errored, groups errored assets by errors.type and errors.messages, and flags assets still preparing after stuck_minutes. At or above critical_error_rate is CRITICAL; at or above warning_error_rate, or any stuck asset, is WARNING; fewer than min_assets finished assets is LOW_VOLUME; otherwise HEALTHY.
route_by_errors (io.kestra.plugin.core.flow.Switch) branches on that classification: HEALTHY and LOW_VOLUME log quietly, WARNING posts a Slack note, CRITICAL and UNREACHABLE page Slack. A defaults fallback logs anything unexpected.
log_audit_result (io.kestra.plugin.core.log.Log) writes one audit line per run with the HTTP status and the full summary.
The flow-level errors block pages Slack if the flow itself fails, and the hourly_check trigger runs at 35 minutes past every hour, shipped disabled: true.

What you get

An hourly, read-only check of video processing with no custom scripting and no extra plugin to install.
A status output (HEALTHY / WARNING / CRITICAL / LOW_VOLUME / UNREACHABLE) other flows can gate on, for example to pause a bulk import while uploads are failing.
Slack messages with the error rate, the number of stuck assets and the most common Mux error type and message, which usually points at the cause.
An audit line on every run, separate from Slack.

Who it's for

Product teams running video upload, course or social features on Mux.
Media engineers who ingest from storage buckets or partner URLs and want to hear about a broken source before users do.
Teams that already run their media pipelines and alerting in Kestra.

Why orchestrate this with Kestra

Mux sends a webhook per asset event, which is great for updating one record and poor at telling you that one upload in five failed today. Kestra turns a plain, read-only API call into a scheduled, typed classification with retries, a failure path and an execution history, and the status output lets your ingestion flows react automatically.

Prerequisites

A Mux account and an access token with Mux Video read permission (Settings > Access Tokens), for the environment you want to watch.
A Slack incoming webhook for the WARNING, CRITICAL, UNREACHABLE and flow-failure alerts.

Secrets

MUX_TOKEN_ID: the access token ID, used as the basic-auth username.
MUX_TOKEN_SECRET: the access token secret, used as the basic-auth password.
SLACK_WEBHOOK: Slack incoming webhook used for every alert branch and the flow-failure alert.

In Kestra OSS, secrets are supplied as environment variables prefixed SECRET_ and base64-encoded, for example SECRET_MUX_TOKEN_ID=$(echo -n 'your-token-id' | base64), and read back with {{ secret('MUX_TOKEN_ID') }}. Use the Enterprise secrets backend for encryption at rest and access control.

Quick start

Upload a short video, plus a file that is not a video, to the Mux environment the token belongs to.
Add MUX_TOKEN_ID, MUX_TOKEN_SECRET and SLACK_WEBHOOK as secrets.
Run the flow manually with a low min_assets and compare the counts and the error in the logs with the Assets page in the Mux dashboard.
Enable hourly_check.

How to extend

Follow next_cursor when more than 100 assets are created in the window, or shorten lookback_hours (the summary reports more_pages).
Split the rate by passthrough or meta to see which upload source or customer fails.
Store the IDs of errored assets in a KV store and alert only on new ones.
Add a second request to GET /video/v1/live-streams to watch live streams in the same run.

Inputs

lookback_hours: type INT, default 24.
warning_error_rate: type FLOAT, default 5.0 (percent).
critical_error_rate: type FLOAT, default 20.0 (percent).
min_assets: type INT, default 10.
stuck_minutes: type INT, default 60.

Outputs

status: one of HEALTHY, WARNING, CRITICAL, LOW_VOLUME, UNREACHABLE.
Example path: {{ fromJson(outputs.summarize_assets.value).top_error }} for the most common Mux error.

Pitfalls

Only the first page (100 assets, newest first) is read; high-volume environments need next_cursor paging or a shorter window.
An access token belongs to one Mux environment; watch each environment (for example production and development) with its own token.
Long or high-resolution videos can legitimately stay in preparing for a while; raise stuck_minutes if you ingest large files.
This flow only reads; it never creates, updates or deletes an asset.