Late Arrival Alerts for Window Completeness

id: late-arrival-data-alert
namespace: company.team
description: |
  Streaming and ELT pipelines keep publishing windows that arrived too late
  to matter, and nothing tells anyone the window was incomplete. A keyed
  webhook accepts the per-window arrival stats, a shared workspace
  materializes them, and a Python step grades every window's received share
  against a completeness SLA. Windows below the SLA post a late-arrival
  alert to Slack with the worst gap and the missing row count, an unreadable
  payload escalates through its own branch, and the verdict, the incomplete
  window list, and the worst completeness percentage are published as
  outputs.

triggers:
  - id: post_window_stats
    type: io.kestra.plugin.core.trigger.Webhook
    description: The streaming collector or watermark monitor POSTs the arrival
      stats for the completed windows here.
    key: "{{ secret('LATE_ARRIVAL_WEBHOOK_KEY') }}"
  - id: hourly_arrival_sweep
    type: io.kestra.plugin.core.trigger.Schedule
    description: Fallback net, an hourly completeness sweep of the last closed
      window, shipped disabled until an operator switches it on.
    cron: "30 * * * *"
    disabled: true

inputs:
  - id: arrival_stats
    type: JSON
    defaults:
      pipeline: orders-enrichment
      windows:
        - window_start: 2026-10-15T14:00:00Z
          expected_rows: 120000
          received_rows: 119840
        - window_start: 2026-10-15T15:00:00Z
          expected_rows: 118500
          received_rows: 117960
        - window_start: 2026-10-15T16:00:00Z
          expected_rows: 121000
          received_rows: 120640
    description: Arrival stats the grade uses when no webhook body is posted, shaped
      as a pipeline name and an array of windows carrying expected and received
      row counts.
  - id: min_completeness_pct
    type: FLOAT
    defaults: 98.0
    description: Lowest received share of expected rows a window may reach before it
      counts as late arriving.
  - id: pipeline_name
    type: STRING
    defaults: orders-enrichment
    description: Pipeline the grade reports on when the posted stats do not name
      one, used in the Slack messages and the summary log line.

tasks:
  # --- 1. window arrival grade ---
  - id: arrival_workspace
    type: io.kestra.plugin.core.flow.WorkingDirectory
    description: One shared directory so the shell step that writes the stats and
      the Python step that grades them read the same file.
    tasks:
      - id: write_arrival_stats
        type: io.kestra.plugin.scripts.shell.Commands
        description: Materialize the posted arrival stats, or the input default on
          manual runs, into the shared workspace with a quoted heredoc and size
          the file for the log.
        taskRunner:
          type: io.kestra.plugin.scripts.runner.docker.Docker
          entryPoint: []
        containerImage: python:3.12-slim
        commands:
          - |
            set -u
            cat > window_arrivals.json <<'STATS_EOF'
            {{ (trigger.body ?? inputs.arrival_stats) | toJson }}
            STATS_EOF
            echo "wrote $(wc -c < window_arrivals.json) bytes of window arrival stats to the shared workspace"

      - id: grade_completeness
        type: io.kestra.plugin.scripts.python.Script
        description: Compute the received share of every window, collect the windows
          below the floor, sum the rows that missed the watermark, and emit the
          late-arrival verdict through the outputs protocol.
        taskRunner:
          type: io.kestra.plugin.scripts.runner.docker.Docker
        containerImage: python:3.12-slim
        env:
          MIN_COMPLETENESS_PCT: "{{ inputs.min_completeness_pct }}"
          PIPELINE_NAME: "{{ inputs.pipeline_name }}"
        script: |
          import json
          import os

          STATS_FILE = "window_arrivals.json"

          def env_float(name, default):
              try:
                  return float(os.environ.get(name, default))
              except (TypeError, ValueError):
                  return float(default)

          MIN_COMPLETENESS_PCT = env_float("MIN_COMPLETENESS_PCT", 98.0)
          PIPELINE_NAME = os.environ.get("PIPELINE_NAME", "unknown-pipeline")

          def load_stats(path):
              try:
                  if not os.path.exists(path) or os.path.getsize(path) == 0:
                      return None, "arrival stats file is missing or empty"
                  with open(path, "r", encoding="utf-8") as handle:
                      document = json.load(handle)
                  document = json.loads(document) if isinstance(document, str) else document
                  if not isinstance(document, dict):
                      return None, "arrival stats is not a JSON object"
                  return document, ""
              except ValueError:
                  return None, "arrival stats is not valid JSON"

          stats, problem = load_stats(STATS_FILE)

          pipeline = PIPELINE_NAME
          windows_checked = 0
          incomplete = []
          window_rows = []
          worst_pct = 100.0
          total_missing = 0
          error = ""
          verdict = "within_sla"

          if stats is None:
              verdict = "error"
              error = problem
          else:
              pipeline = str(stats.get("pipeline") or PIPELINE_NAME)
              windows = [w for w in (stats.get("windows") or []) if isinstance(w, dict)]
              if not windows:
                  verdict = "error"
                  error = "arrival stats contain no windows"
              else:
                  for window in windows:
                      label = str(window.get("window_start") or window.get("window") or "unnamed-window")
                      try:
                          expected = int(window.get("expected_rows") or 0)
                      except (TypeError, ValueError):
                          expected = 0
                      try:
                          received = int(window.get("received_rows") or 0)
                      except (TypeError, ValueError):
                          received = 0
                      if expected <= 0:
                          pct = 100.0
                          missing = 0
                      else:
                          pct = round(received * 100.0 / expected, 2)
                          missing = max(expected - received, 0)
                      late = pct < MIN_COMPLETENESS_PCT
                      window_rows.append({
                          "window_start": label,
                          "expected_rows": expected,
                          "received_rows": received,
                          "completeness_pct": pct,
                          "missing_rows": missing,
                          "late": late,
                      })
                      if late:
                          incomplete.append(label)
                      worst_pct = min(worst_pct, pct)
                      total_missing += missing
                  windows_checked = len(window_rows)
                  verdict = "late_arrival" if incomplete else "within_sla"

          payload = {
              "verdict": verdict,
              "pipeline": pipeline,
              "windows_checked": windows_checked,
              "incomplete_count": len(incomplete),
              "incomplete_windows": sorted(incomplete),
              "worst_pct": round(worst_pct, 2),
              "total_missing_rows": total_missing,
              "min_completeness_pct": MIN_COMPLETENESS_PCT,
              "windows": window_rows,
              "error": error,
          }
          print("pipeline=%s windows=%d incomplete=%d worst=%.2f verdict=%s" % (
              pipeline, windows_checked, len(incomplete), payload["worst_pct"], verdict))
          print("::" + json.dumps({"outputs": payload}) + "::")

  # --- 2. sla verdict ---
  - id: check_sla
    type: io.kestra.plugin.core.flow.If
    description: Escalate to Slack when at least one window landed below the
      completeness floor, and verify the error branch instead of trusting a
      parse failure.
    condition: "{{ outputs.grade_completeness.vars.verdict == 'late_arrival' }}"
    then:
      - id: notify_late_arrival
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Tell the data channel which windows arrived late, the worst
          received share, and how many rows missed the watermark.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "text": "LATE ARRIVAL: {{ outputs.grade_completeness.vars.pipeline }} - {{ outputs.grade_completeness.vars.incomplete_count }} window(s) below {{ outputs.grade_completeness.vars.min_completeness_pct }}% completeness ({{ outputs.grade_completeness.vars.incomplete_windows | join(', ') }}), worst {{ outputs.grade_completeness.vars.worst_pct }}%, {{ outputs.grade_completeness.vars.total_missing_rows }} row(s) missed the watermark. Execution {{ execution.id }}."
          }
    else:
      - id: check_grade_error
        type: io.kestra.plugin.core.flow.If
        description: Surface arrival stats that could not be graded as their own branch
          so a broken feed never reads as a complete window.
        condition: "{{ outputs.grade_completeness.vars.verdict == 'error' }}"
        then:
          - id: alert_unreadable_stats
            type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
            description: Tell the data channel the arrival stats could not be graded and
              carry the load problem named by the grader.
            url: "{{ secret('SLACK_WEBHOOK_URL') }}"
            payload: |
              {
                "text": "LATE ARRIVAL grade unreadable for {{ outputs.grade_completeness.vars.pipeline }} - {{ outputs.grade_completeness.vars.error }}. Execution {{ execution.id }}."
              }
        else:
          - id: log_within_sla
            type: io.kestra.plugin.core.log.Log
            description: Record the clean completeness grade so the execution history
              doubles as watermark evidence.
            message: "within SLA - {{ outputs.grade_completeness.vars.pipeline }} graded {{
              outputs.grade_completeness.vars.windows_checked }} window(s),
              worst {{ outputs.grade_completeness.vars.worst_pct }}% against a
              {{ outputs.grade_completeness.vars.min_completeness_pct }}%
              floor."

errors:
  - id: notify_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Escalate 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
    description: within_sla when every window met the floor, late_arrival when one
      slipped below it, error when the stats could not be graded.
    value: "{{ outputs.grade_completeness.vars.verdict }}"
  - id: incomplete_windows
    type: ARRAY
    description: Window starts that landed below the completeness floor, for example
      the 15:00 UTC window.
    value: "{{ outputs.grade_completeness.vars.incomplete_windows | toJson }}"
  - id: worst_pct
    type: FLOAT
    description: Lowest received share across all graded windows, for example 60.0.
    value: "{{ outputs.grade_completeness.vars.worst_pct }}"
  - id: total_missing_rows
    type: INT
    description: Rows that never arrived before the watermark across every graded
      window in the batch.
    value: "{{ outputs.grade_completeness.vars.total_missing_rows }}"
  - id: windows_checked
    type: INT
    description: Number of windows the grade scored during this run.
    value: "{{ outputs.grade_completeness.vars.windows_checked }}"

How it works

post_window_stats (io.kestra.plugin.core.trigger.Webhook, keyed) accepts the arrival stats POST from the streaming collector or the watermark monitor. hourly_arrival_sweep (io.kestra.plugin.core.trigger.Schedule) ships disabled: true as a fallback net once an operator switches it on.
arrival_workspace (io.kestra.plugin.core.flow.WorkingDirectory) opens one shared directory so the writer and the grader read the same file, because default tasks are stateless and the temp directory is purged when a task completes.
write_arrival_stats (io.kestra.plugin.scripts.shell.Commands on the io.kestra.plugin.scripts.runner.docker.Docker runner, image python:3.12-slim) writes the posted body, or the arrival_stats input default on manual runs, into window_arrivals.json with a quoted heredoc.
grade_completeness (io.kestra.plugin.scripts.python.Script on python:3.12-slim) computes the received share of every window against min_completeness_pct, collects the windows below the floor, sums the rows that missed the watermark, and emits the verdict through the outputs protocol.
check_sla (io.kestra.plugin.core.flow.If) branches on the verdict. A late window runs notify_late_arrival, the Slack incoming webhook that names the incomplete windows, the worst share, and the missing rows. An unreadable payload runs check_grade_error, a second If whose alert_unreadable_stats branch pages Slack so a broken feed never reads as a clean window, and only a fully complete window reaches log_within_sla.
The errors block alerts Slack whenever the flow itself breaks, because a dead grader must never look like a perfect window.

What you get

A completeness grade for every window in the posted batch, with the exact rows that never arrived before the watermark.
A Slack alert that names the late windows, the worst received share, and the total missing rows the moment the batch is graded.
A separate unreadable payload branch, so a malformed POST escalates instead of quietly passing as zero incomplete windows.
Gateable outputs, verdict, incomplete_windows, worst_pct, total_missing_rows, and windows_checked for any parent pipeline that wants to block its publish step.
A dated execution history that doubles as watermark evidence for the platform team.

Who it's for

Streaming platform teams that publish windows downstream and want proof each window was complete before consumers read it.
Data engineers who currently discover missing rows only after a dashboard or a retrain absorbs them.
On-call engineers who want late data paged to the channel with the window names already attached.

Why orchestrate this with Kestra

A completeness check is one division, but an alert needs the state around it, a keyed webhook for the collector, a hermetic Python runtime pinned in the flow file, a threshold that separates escalation from the all-clear, and an execution record proving the grade ran for that exact batch. Kestra supplies all of that declaratively, with Slack routing, an error handler, and a branch structure that any scheduler already understands. A cron job that greps logs gives you none of that, it only helps the people who remember to look.

Prerequisites

A worker with Docker access, both the writer and the grader run in python:3.12-slim.
A Slack incoming webhook for the late-arrival alert, unreadable payload pages, and failure alerts.
A collector that POSTs arrival stats shaped as a pipeline name and a windows array carrying window_start, expected_rows, and received_rows per window.

Secrets

LATE_ARRIVAL_WEBHOOK_KEY: key segment guarding the webhook URL, anyone holding it can trigger a grade.
SLACK_WEBHOOK_URL: Slack incoming webhook URL used for the late-arrival alert, the unreadable payload page, and failures.

Quick start

Add the secrets above to your Kestra namespace.
Run the flow once with the default inputs and confirm window_arrivals.json lands in the workspace and the outputs carry verdict, worst_pct, and windows_checked.
Lower one window's received_rows in arrival_stats below the floor, re-run, and watch verdict flip to late_arrival with that window named in incomplete_windows.
Point the collector at /api/v1/{tenant}/executions/webhook/{namespace}/late-arrival-data-alert/{LATE_ARRIVAL_WEBHOOK_KEY} once its POST body matches the stats shape.

How to extend

Fan the grade out over many pipelines with a Subflow per pipeline and aggregate the verdicts in a parent flow.
Block the publish step by having the parent flow read verdict and stop on late_arrival instead of only alerting.
Replace the writer with a io.kestra.plugin.git.Clone of a stats repository when the collector publishes files instead of POSTing JSON.
Archive each graded batch to object storage with io.kestra.plugin.aws.s3.Upload so every window keeps its arrival evidence.
Walk the per-window rows with io.kestra.plugin.core.flow.Loop when you want one log line or one ticket per late window.

Alert When Streaming Windows Ship With Late or Missing Rows

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?

Alert When Streaming Windows Ship With Late or Missing Rows

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 Data Blueprints

MongoDB GridFS Orphaned Chunk and Dangling File Reclamation Guard

MongoDB Document Validation Drift Repair Guard

MongoDB Slow Query and Collection Scan Profiler

MongoDB Dead-Letter Mutation Reconciliation and Bulk Repair Gate

MongoDB Duplicate Document Dedup Guard

MongoDB Unbounded Array Growth Trim Guard

New to Kestra?