OpenAI Batch Job Sentinel for Kestra

id: openai-batch-job-sentinel
namespace: company.team
description: |
  Every hour, list recent jobs from the OpenAI Batch API (read-only: no
  batch is created or canceled) and classify them as HEALTHY, WARNING,
  CRITICAL, LOW_VOLUME or UNREACHABLE. Log quiet runs and alert Slack when a
  batch has failed or expired, when one is close to its 24-hour completion
  window without finishing, or when too many requests inside finished
  batches failed.

inputs:
  - id: lookback_hours
    type: INT
    defaults: 48
    displayName: "Lookback Window (hours)"
    description: "Only batches created in the last this many hours are judged."
  - id: expiry_warning_hours
    type: INT
    defaults: 4
    displayName: "Expiry Warning (hours)"
    description: "A batch still validating, in progress or finalizing with less than
      this many hours left before expires_at is counted as at risk."
  - id: warning_failed_request_pct
    type: FLOAT
    defaults: 2.0
    displayName: "Warning Failed Requests (%)"
    description: "At or above this share of failed requests across completed
      batches, the check is WARNING."
  - id: critical_failed_request_pct
    type: FLOAT
    defaults: 10.0
    displayName: "Critical Failed Requests (%)"
    description: "At or above this share of failed requests across completed
      batches, the check is CRITICAL."

triggers:
  - id: hourly_check
    type: io.kestra.plugin.core.trigger.Schedule
    description: Check batch jobs every hour. Shipped disabled; enable it once the
      OpenAI API key secret is set and a manual run has been verified.
    cron: "10 * * * *"
    disabled: true

tasks:
  - id: list_batches
    type: io.kestra.plugin.core.http.Request
    description: List the 100 most recent batches of the organization or project
      from the OpenAI Batch API, with their status, timestamps and request
      counts. This is a GET; no batch is created, canceled or changed.
    uri: "https://api.openai.com/v1/batches?limit=100"
    method: GET
    headers:
      Authorization: "Bearer {{ secret('OPENAI_API_KEY') }}"
    options:
      allowFailed: true
      timeout:
        connectTimeout: PT10S
        readIdleTimeout: PT30S
    retry:
      type: constant
      interval: PT10S
      maxAttempts: 3
    allowFailure: true

  - id: summarize_batches
    type: io.kestra.plugin.core.debug.Return
    description: Keep batches created in the lookback window, list ones that failed
      or expired, list ones still running close to expires_at, add up
      request_counts across completed batches to get the failed-request share,
      and classify in one jq query. A missing or non-200 response becomes
      UNREACHABLE.
    format: >-
      {{ (outputs.list_batches.code is not defined or outputs.list_batches.code
      != 200) ?
      '{"status":"UNREACHABLE","batches":0,"completed":0,"running":0,"requests":0,"failed_requests":0,"failed_pct":0,"dead":"none","at_risk":"none"}'
      : (fromJson(outputs.list_batches.body) | jq(' ' ~ (now() | timestamp) ~ '
      as $now | [(.data // [])[] | select((.created_at // 0) >= ($now - ' ~
      inputs.lookback_hours ~ ' * 3600))] as $b | [$b[] | select(.status ==
      "failed" or .status == "expired") | (.id + " (" + .status + ", " +
      (.endpoint // "?") + ((.errors.data // [])[0].message // "" | if . == ""
      then "" else ": " + (split("\"") | join("") | .[0:80]) end) + ")")] as
      $dead | [$b[] | select((.status == "validating" or .status ==
      "in_progress" or .status == "finalizing") and (.expires_at != null) and
      ((.expires_at - $now) < ' ~ inputs.expiry_warning_hours ~ ' * 3600)) |
      (.id + " (" + .status + ", " + ((.request_counts.completed // 0) |
      tostring) + "/" + ((.request_counts.total // 0) | tostring) + " done, " +
      ((((.expires_at - $now) / 360) | floor) / 10 | tostring) + " h left)")] as
      $risk | [$b[] | select(.status == "completed")] as $done | ([$done[] |
      (.request_counts.total // 0)] | add // 0) as $req | ([$done[] |
      (.request_counts.failed // 0)] | add // 0) as $freq | (if $req == 0 then 0
      else (($freq * 1000 / $req) | floor) / 10 end) as $pct | {batches: ($b |
      length), completed: ($done | length), running: ([$b[] | select(.status ==
      "validating" or .status == "in_progress" or .status == "finalizing")] |
      length), requests: $req, failed_requests: $freq, failed_pct: $pct, dead:
      (if ($dead | length) == 0 then "none" else ($dead | join("; ")) end),
      at_risk: (if ($risk | length) == 0 then "none" else ($risk | join("; "))
      end), status: (if ($dead | length) > 0 or $pct >= ' ~
      inputs.critical_failed_request_pct ~ ' then "CRITICAL" elif ($risk |
      length) > 0 or $pct >= ' ~ inputs.warning_failed_request_pct ~ ' then
      "WARNING" elif ($b | length) == 0 then "LOW_VOLUME" else "HEALTHY" end)}
      ') | first | toJson) }}

  - id: route_by_batch_health
    type: io.kestra.plugin.core.flow.Switch
    description: Route to a quiet log when batches are healthy or none ran in the
      window, a Slack message when a batch is at risk of expiring or failed
      requests cross the warning line, or a Slack page when a batch failed or
      expired, failed requests cross the critical line, or the API could not be
      read.
    value: "{{ fromJson(outputs.summarize_batches.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: "OpenAI batches healthy: {{
            fromJson(outputs.summarize_batches.value).completed }} completed and
            {{ fromJson(outputs.summarize_batches.value).running }} running out
            of {{ fromJson(outputs.summarize_batches.value).batches }} in the
            last {{ inputs.lookback_hours }} hour(s); {{
            fromJson(outputs.summarize_batches.value).failed_requests }} of {{
            fromJson(outputs.summarize_batches.value).requests }} requests
            failed ({{ fromJson(outputs.summarize_batches.value).failed_pct
            }}%)."
      "LOW_VOLUME":
        - id: log_low_volume
          type: io.kestra.plugin.core.log.Log
          description: Log a quiet line when there is too little activity in the window to
            judge; no Slack message is sent.
          message: "OpenAI batches: no batch was created in the last {{
            inputs.lookback_hours }} hour(s); nothing to judge this run."
      "WARNING":
        - id: alert_warning_batches
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          description: Post a Slack message when a running batch is close to its expiry or
            the failed-request share is at or above the warning line.
          url: "{{ secret('SLACK_WEBHOOK') }}"
          payload: |
            {
              "text": "OpenAI batches WARNING. At risk of expiring: {{ fromJson(outputs.summarize_batches.value).at_risk }}. Failed requests in completed batches: {{ fromJson(outputs.summarize_batches.value).failed_requests }} of {{ fromJson(outputs.summarize_batches.value).requests }} ({{ fromJson(outputs.summarize_batches.value).failed_pct }}%). Execution {{ execution.id }}."
            }
      "CRITICAL":
        - id: alert_critical_batches
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          description: Page Slack when a batch failed validation or expired, or the
            failed-request share is at or above the critical line, because
            downstream jobs are missing results.
          url: "{{ secret('SLACK_WEBHOOK') }}"
          payload: |
            {
              "text": "OpenAI batches CRITICAL. Failed or expired: {{ fromJson(outputs.summarize_batches.value).dead }}. Failed requests in completed batches: {{ fromJson(outputs.summarize_batches.value).failed_requests }} of {{ fromJson(outputs.summarize_batches.value).requests }} ({{ fromJson(outputs.summarize_batches.value).failed_pct }}%). Results are missing; check the error files and resubmit. Execution {{ execution.id }}."
            }
      "UNREACHABLE":
        - id: alert_unreachable_api
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          description: Page Slack when the batch list could not be read, indicating an
            outage or an invalid or under-scoped API key.
          url: "{{ secret('SLACK_WEBHOOK') }}"
          payload: |
            {
              "text": "OpenAI batch sentinel UNREACHABLE: the Batch API did not return a usable response (HTTP {{ outputs.list_batches.code ?? 'no response' }}). Check the API key and its permissions. 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: "OpenAI batch sentinel completed with an unexpected classification: {{
          outputs.summarize_batches.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
      batch history stays queryable even if Slack delivery fails.
    message: "OpenAI batch sentinel audit: lookback_hours={{ inputs.lookback_hours
      }} http={{ outputs.list_batches.code ?? 'n/a' }} summary={{
      outputs.summarize_batches.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 batches.
    url: "{{ secret('SLACK_WEBHOOK') }}"
    payload: |
      {
        "text": "OpenAI batch sentinel flow {{ flow.id }} FAILED (execution {{ execution.id }}). Batch health is now unknown; check the task logs and the Batches page of the OpenAI platform."
      }

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

The OpenAI Batch API runs large jobs (embeddings for a whole catalog, nightly classification, evaluation sets) at half price, in exchange for a completion window of 24 hours. That trade has quiet failure modes: a batch fails validation because one line of the input file is malformed, a busy period means it is still in_progress when the window closes and it ends expired with only part of the results, or it completes but a share of its requests failed. The job that was waiting for the output file just finds nothing, often the next morning. This blueprint turns a scheduled, read-only call to the Batch API into a batch health gate: io.kestra.plugin.core.http.Request lists recent batches, one jq query finds failed, expired and at-risk batches and the failed-request share, and the result routes to a quiet log or a Slack alert.

How it works

list_batches (io.kestra.plugin.core.http.Request) calls GET /v1/batches?limit=100 with a bearer API key. It is a read: no batch is created, canceled or changed, and no tokens are spent. Retries cover short network blips, and allowFailed plus allowFailure keep an API error from aborting the flow so it can be reported instead.
summarize_batches (io.kestra.plugin.core.debug.Return) first guards for a missing or non-200 response (UNREACHABLE). Otherwise one jq query keeps batches whose created_at falls in the lookback window and lists those with status failed or expired (with the first validation error message), and those still validating, in_progress or finalizing with less than expiry_warning_hours left before expires_at. It adds up request_counts.total and request_counts.failed across completed batches. Any failed or expired batch, or a failed-request share at or above critical_failed_request_pct, is CRITICAL; an at-risk batch or a share at or above warning_failed_request_pct is WARNING; no batch in the window is LOW_VOLUME; otherwise HEALTHY.
route_by_batch_health (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 10 minutes past every hour, shipped disabled: true.

What you get

An hourly, read-only check of every batch job 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 hold a downstream load until the batch results are complete, or to resubmit the failed lines.
Slack messages that name each failed, expired or at-risk batch with its endpoint, progress and hours left.
An audit line on every run, separate from Slack.

Who it's for

Data and ML teams that run embeddings, classification, extraction or evaluation jobs through the Batch API.
Teams that use Kestra to submit batches and want an independent check that they actually finished.
Anyone who has found an expired batch the morning after.

Why orchestrate this with Kestra

The OpenAI platform shows batches in a list, but nothing tells you a batch is about to miss its window or that 8% of its lines failed. 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 the flows that consume batch results wait, retry or resubmit automatically.

Prerequisites

An OpenAI API key with read access to batches for the organization or project that submits them.
A Slack incoming webhook for the WARNING, CRITICAL, UNREACHABLE and flow-failure alerts.

Secrets

OPENAI_API_KEY: an API key (a project key is enough) for the project whose batches you want to watch.
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_OPENAI_API_KEY=$(echo -n 'sk-proj-xxxxxxxx' | base64), and read back with {{ secret('OPENAI_API_KEY') }}. Use the Enterprise secrets backend for encryption at rest and access control.

Quick start

Submit a small batch (a few lines against /v1/chat/completions or /v1/embeddings), plus one with a malformed input file to produce a failed batch.
Add OPENAI_API_KEY and SLACK_WEBHOOK as secrets.
Run the flow manually and compare the batch names, statuses and request counts in the logs with the Batches page of the OpenAI platform.
Enable hourly_check.

How to extend

Follow has_more and last_id with the after parameter when you create more than 100 batches in the window.
Filter on metadata (for example a job key you set at submission) to give each pipeline its own limits.
Download error_file_id with the Files API and attach the first failing lines to the Slack message.
Store the last status in a KV store and alert only when it changes.

Inputs

lookback_hours: type INT, default 48.
expiry_warning_hours: type INT, default 4.
warning_failed_request_pct: type FLOAT, default 2.0 (percent).
critical_failed_request_pct: type FLOAT, default 10.0 (percent).

Outputs

status: one of HEALTHY, WARNING, CRITICAL, LOW_VOLUME, UNREACHABLE.
Example path: {{ fromJson(outputs.summarize_batches.value).dead }} for the failed or expired batches.

Pitfalls

Only the 100 most recent batches are read; page with after if you create more.
A cancelled batch is counted in the window but is not an alert, because it was canceled on purpose.
request_counts.failed covers requests that returned an error; a request that succeeded with an unhelpful answer is not a failure here.
Project API keys only see that project's batches; use one flow per project or a key with wider access.
This flow only reads; it never creates, cancels or resubmits a batch.