S3 File Arrival SLA Monitoring with Kestra

id: s3-file-arrival-sla-monitor
namespace: company.team
description: |
  Watch the files you expect from banks, vendors, partners, and upstream
  teams. Every 15 minutes each expected feed is checked in S3 (or any
  S3-compatible store); Slack is alerted once when a feed is missing past its
  deadline, arrives empty, or is delivered twice, and again when it recovers.
  A daily arrival report summarises every feed.

inputs:
  - id: feeds
    type: JSON
    description: |
      Expected feeds. Fields per feed — name (required), prefix (required, supports {yyyy} {MM} {dd} tokens),
      pattern (regex on the full key, default ".*"), deadline ("HH:mm" local time, default "23:59"),
      min_files (default 1), min_bytes (default 1), days (e.g. ["MON","TUE","WED","THU","FRI"], default every day),
      bucket (optional, overrides the bucket input).
    defaults: |
      [
        {"name": "bank_statements", "prefix": "inbound/bank/{yyyy}/{MM}/{dd}/", "pattern": ".*\\.csv$", "deadline": "07:00", "min_bytes": 1024, "days": ["MON", "TUE", "WED", "THU", "FRI"]},
        {"name": "crm_customer_export", "prefix": "inbound/crm/", "pattern": ".*customers_.*\\.json$", "deadline": "09:30", "min_bytes": 512},
        {"name": "payroll_hours", "prefix": "inbound/payroll/", "pattern": ".*\\.xlsx$", "deadline": "12:00", "min_files": 2, "days": ["MON"]}
      ]
  - id: bucket
    type: STRING
    defaults: my-inbound-bucket
    description: Default bucket that receives the files. A feed can override it with
      its own `bucket` field.
  - id: timezone
    type: STRING
    defaults: UTC
    description: IANA time zone in which deadlines, "today", and weekdays are
      evaluated (e.g. Europe/Berlin, America/New_York, Australia/Sydney).
  - id: mode
    type: SELECT
    values:
      - check
      - report
    defaults: check
    description: "`check` evaluates feeds and alerts on changes; `report` does the
      same and then posts the daily arrival summary."

triggers:
  - id: check_every_15_min
    type: io.kestra.plugin.core.trigger.Schedule
    description: Frequent check so a missed deadline is reported within 15 minutes.
    cron: "*/15 * * * *"

  - id: daily_arrival_report
    type: io.kestra.plugin.core.trigger.Schedule
    description: End-of-day summary of every feed. Set `timezone` to match the
      `timezone` input.
    cron: "0 18 * * *"
    timezone: UTC
    inputs:
      mode: report

  - id: on_upload_event
    type: io.kestra.plugin.core.trigger.Webhook
    description: Optional instant check — call it from an S3 Event Notification (via
      EventBridge API destination or a tiny Lambda) or from the sender's upload
      script so empty or duplicate files are flagged seconds after upload.
      Change the key before enabling.
    key: change-me-file-sla-check

concurrency:
  behavior: QUEUE
  limit: 1

pluginDefaults:
  - type: io.kestra.plugin.aws
    values:
      accessKeyId: "{{ secret('AWS_ACCESS_KEY_ID') }}"
      secretKeyId: "{{ secret('AWS_SECRET_ACCESS_KEY') }}"
      region: "{{ secret('AWS_DEFAULT_REGION') }}"

tasks:
  - id: check_feeds
    type: io.kestra.plugin.core.flow.Loop
    description: Evaluate every expected feed in parallel; each iteration lists,
      classifies, alerts on state change, and records today's state.
    values: "{{ inputs.feeds }}"
    concurrencyLimit: 5
    tasks:
      - id: list_files
        type: io.kestra.plugin.aws.s3.List
        description: List candidate files under the feed prefix, with {yyyy}/{MM}/{dd}
          resolved to today's date in the configured time zone.
        bucket: "{{ fromJson(item.value).bucket ?? inputs.bucket }}"
        prefix: "{{ fromJson(item.value).prefix | replace({'{yyyy}': now() |
          date('yyyy', timeZone=inputs.timezone), '{MM}': now() | date('MM',
          timeZone=inputs.timezone), '{dd}': now() | date('dd',
          timeZone=inputs.timezone)}) }}"
        regexp: "{{ fromJson(item.value).pattern ?? '.*' }}"
        filter: FILES
        maxFiles: 1000

      - id: evaluate
        type: io.kestra.plugin.core.output.OutputValues
        description: |
          Keep only files modified since local midnight, then classify the feed:
          NOT_EXPECTED (not a delivery day), WAITING (before deadline), MISSING (past deadline, fewer than min_files),
          TOO_SMALL (a file under min_bytes), DUPLICATE (same ETag delivered twice), LATE (complete, first file after deadline), ON_TIME.
          The previous state for today is read from the KV store so alerts fire only on transitions.
        values:
          evaluation: |
            {%- set feed = fromJson(item.value) -%}
            {%- set tz = inputs.timezone -%}
            {%- set today = now() | date('yyyy-MM-dd', timeZone=tz) -%}
            {%- set now_hhmm = now() | date('HHmm', timeZone=tz) | number -%}
            {%- set local_midnight = now() | dateAdd(-1 * (now() | date('H', timeZone=tz) | number), 'HOURS') | dateAdd(-1 * (now() | date('m', timeZone=tz) | number), 'MINUTES') -%}
            {%- set day_start_utc = (local_midnight | date("yyyy-MM-dd'T'HH:mm", timeZone='UTC')) ~ ':00Z' -%}
            {%- set files = outputs.list_files.objects | jq('[.[] | select(.lastModified >= "' ~ day_start_utc ~ '")]') | first -%}
            {%- set count = files | length -%}
            {%- set deadline = feed.deadline ?? '23:59' -%}
            {%- set deadline_hhmm = deadline | replace({':': ''}) | number -%}
            {%- set min_files = feed.min_files ?? 1 -%}
            {%- set min_bytes = feed.min_bytes ?? 1 -%}
            {%- set small = files | jq('[.[] | select(.size < ' ~ min_bytes ~ ')] | length') | first -%}
            {%- set duplicates = files | jq('([.[].etag] | length) - ([.[].etag] | unique | length)') | first -%}
            {%- set first_arrival = '' -%}
            {%- set first_hhmm = 0 -%}
            {%- if count > 0 -%}
              {%- set first_ts = files | jq('map(.lastModified) | min') | first -%}
              {%- set first_arrival = first_ts | date('HH:mm', timeZone=tz) -%}
              {%- set first_hhmm = first_ts | date('HHmm', timeZone=tz) | number -%}
            {%- endif -%}
            {%- set weekday = now() | date('EEE', timeZone=tz) | upper -%}
            {%- set days = feed.days ?? [] -%}
            {%- if (days | length) > 0 and not (days contains weekday) -%}
              {%- set status = 'NOT_EXPECTED' -%}
            {%- elseif count < min_files -%}
              {%- set status = now_hhmm >= deadline_hhmm ? 'MISSING' : 'WAITING' -%}
            {%- elseif small > 0 -%}
              {%- set status = 'TOO_SMALL' -%}
            {%- elseif duplicates > 0 -%}
              {%- set status = 'DUPLICATE' -%}
            {%- elseif first_hhmm > deadline_hhmm -%}
              {%- set status = 'LATE' -%}
            {%- else -%}
              {%- set status = 'ON_TIME' -%}
            {%- endif -%}
            {%- set previous = fromJson(kv(key='file_sla_' ~ feed.name ~ '_' ~ today, errorOnMissing=false) ?? '{}').status ?? 'NONE' -%}
            {%- set alerting = ['MISSING', 'TOO_SMALL', 'DUPLICATE'] -%}
            {%- set notify = status != previous and ((alerting contains status) or (alerting contains previous)) -%}
            {%- if status == 'MISSING' -%}
              {%- set message = ':rotating_light: *' ~ feed.name ~ '* is MISSING — expected ' ~ min_files ~ ' file(s) matching `' ~ (feed.pattern ?? '.*') ~ '` under `' ~ feed.prefix ~ '` by ' ~ deadline ~ ' (' ~ tz ~ '), found ' ~ count ~ ' today.' -%}
            {%- elseif status == 'TOO_SMALL' -%}
              {%- set message = ':warning: *' ~ feed.name ~ '* arrived but ' ~ small ~ ' of ' ~ count ~ ' file(s) are under ' ~ min_bytes ~ ' bytes — likely empty or truncated.' -%}
            {%- elseif status == 'DUPLICATE' -%}
              {%- set message = ':warning: *' ~ feed.name ~ '* was delivered more than once today — ' ~ duplicates ~ ' file(s) share an identical ETag with another file.' -%}
            {%- elseif alerting contains previous -%}
              {%- set message = ':white_check_mark: *' ~ feed.name ~ '* recovered from ' ~ previous ~ ' — now ' ~ status ~ ' (' ~ count ~ ' file(s), first at ' ~ first_arrival ~ ', deadline ' ~ deadline ~ ').' -%}
            {%- else -%}
              {%- set message = feed.name ~ ' is ' ~ status -%}
            {%- endif -%}
            {{ {'name': feed.name, 'date': today, 'status': status, 'previous': previous, 'notify': notify, 'files': count, 'first_arrival': first_arrival, 'deadline': deadline, 'too_small': small, 'duplicates': duplicates, 'message': message} | toJson }}

      - id: alert_on_change
        type: io.kestra.plugin.core.flow.If
        description: Alert only when the feed enters a problem state or recovers from
          one — a missing feed is reported once, not every 15 minutes.
        condition: "{{ fromJson(outputs.evaluate.values.evaluation).notify }}"
        then:
          - id: slack_alert
            type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
            url: "{{ secret('SLACK_WEBHOOK_URL') }}"
            payload: |
              {
                "text": {{ fromJson(outputs.evaluate.values.evaluation).message | toJson }}
              }

      - id: record_state
        type: io.kestra.plugin.core.kv.Set
        description: Persist today's state per feed — used for alert de-duplication and
          for the daily report. Entries expire after a week.
        key: "file_sla_{{ fromJson(outputs.evaluate.values.evaluation).name }}_{{
          fromJson(outputs.evaluate.values.evaluation).date }}"
        value: "{{ outputs.evaluate.values.evaluation }}"
        kvType: STRING
        ttl: P7D

  - id: daily_report
    type: io.kestra.plugin.core.flow.If
    condition: "{{ inputs.mode == 'report' }}"
    then:
      - id: slack_report
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: One message listing every feed with its state, file count, first
          arrival time, and deadline.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {%- set today = now() | date('yyyy-MM-dd', timeZone=inputs.timezone) -%}
          {%- set icons = {'ON_TIME': ':large_green_circle:', 'LATE': ':large_yellow_circle:', 'WAITING': ':hourglass:', 'NOT_EXPECTED': ':white_circle:', 'MISSING': ':red_circle:', 'TOO_SMALL': ':red_circle:', 'DUPLICATE': ':red_circle:'} -%}
          {"text": "*File arrival report for {{ today }} ({{ inputs.timezone }})*
          {%- for f in inputs.feeds -%}
            {%- set s = fromJson(kv(key='file_sla_' ~ f.name ~ '_' ~ today, errorOnMissing=false) ?? '{}') -%}
            \n{{ icons[s.status ?? 'UNKNOWN'] ?? ':grey_question:' }} *{{ f.name }}* — {{ s.status ?? 'UNKNOWN' }} · {{ s.files ?? 0 }} file(s){{ (s.first_arrival ?? '') == '' ? '' : ' · first ' ~ s.first_arrival }} · deadline {{ f.deadline ?? '23:59' }}
          {%- endfor -%}
          "}

errors:
  - id: monitor_failed
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: A broken monitor must never look like "all files arrived" — alert
      when the check itself fails (credentials, bucket access, bad feed config).
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "text": ":x: File arrival SLA monitor FAILED in {{ flow.namespace }}.{{ flow.id }} (execution {{ execution.id }}). Feed states were not evaluated — check bucket access and the `feeds` input."
      }

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.

Almost every company depends on files it does not control: bank statements, payroll exports, vendor price lists, partner order feeds, CRM dumps from another team. When one of them silently does not arrive, nothing fails — the downstream pipeline just loads nothing, and someone notices days later in a dashboard. This blueprint turns "we expect these files every day by these times" into a declarative SLA monitor: you list the feeds, and Kestra tells you once when a feed is missing, empty, or duplicated, tells you again when it recovers, and sends an end-of-day arrival report.

It works with Amazon S3 and any S3-compatible storage (MinIO, Cloudflare R2, Wasabi, Backblaze B2, Ceph) and needs no database, no Lambda, and no changes on the sender's side.

How it works

Triggers. check_every_15_min (Schedule) runs the checks. daily_arrival_report (Schedule, 18:00) runs them with mode: report. on_upload_event (Webhook) lets an S3 event notification or the sender's upload script request an instant check.

Per feed (check_feeds, io.kestra.plugin.core.flow.Loop, 5 in parallel):

list_files (io.kestra.plugin.aws.s3.List) lists objects under the feed's prefix matching its pattern. {yyyy}, {MM} and {dd} in the prefix are replaced with today's date in your time zone, so date-partitioned landing folders work out of the box.

evaluate (io.kestra.plugin.core.output.OutputValues) keeps files modified since local midnight and classifies the feed:

Status	Meaning	Alert
`NOT_EXPECTED`	Today is not in the feed's `days`	–
`WAITING`	Fewer than `min_files`, deadline not reached	–
`MISSING`	Fewer than `min_files` after the deadline	:rotating_light:
`TOO_SMALL`	A file is smaller than `min_bytes` (empty / truncated)	:warning:
`DUPLICATE`	Two files today share an ETag (same content re-sent)	:warning:
`LATE`	Complete, but the first file arrived after the deadline	recovery only
`ON_TIME`	Complete before the deadline	recovery only

alert_on_change (If) posts to Slack only when the state changes into a problem state or out of one, so a missing feed produces one alert and one recovery message, not one alert every 15 minutes.
record_state (io.kestra.plugin.core.kv.Set) stores today's state per feed in the KV store (7-day TTL) for de-duplication and reporting.

Report. In report mode, slack_report posts one message with every feed's state, file count, first arrival time and deadline.
Self-monitoring. If the check itself fails (credentials, bucket access, invalid feed JSON), the errors branch alerts Slack — a broken monitor is never mistaken for "all files arrived".

What you get

One place that documents which files the business depends on, when they are due, and on which days.
Alerts that point at the problem (missing vs. empty vs. duplicated), sent once, with a recovery message when the file finally lands.
A daily arrival report your operations or finance team can read without opening a console.
Arrival history per feed in the KV store and in the execution log.

Who it's for

Data and analytics engineers whose pipelines start from files dropped by partners or other teams.
Finance, payroll and operations teams waiting on bank, PSP, payroll, or vendor files.
Platform teams replacing ad-hoc "did the file come?" cron scripts and Airflow sensors.

Why orchestrate this with Kestra

An S3 event can tell you that a file arrived; it can never tell you that one did not. Detecting absence needs a schedule, a notion of deadlines and delivery days per feed, state to avoid alert storms, and a report — usually spread across cron, a script, and a database. Kestra provides all of it declaratively: schedules and a webhook, a parallel Loop over feeds, the KV store for state, and Slack alerting, in one versioned YAML file. Because each feed is configuration, adding a new partner feed is a one-line change in the feeds input, not new code.

Prerequisites

A bucket where the files land (S3 or S3-compatible) and credentials with s3:ListBucket on it. No read access to file contents is needed.
A Slack incoming webhook.

Secrets

AWS_ACCESS_KEY_ID: access key allowed to list the inbound bucket(s).
AWS_SECRET_ACCESS_KEY: matching secret key.
AWS_DEFAULT_REGION: bucket region (e.g. eu-west-1; use us-east-1 or your provider's region for S3-compatible stores).
SLACK_WEBHOOK_URL: Slack incoming webhook for alerts and the daily report.

Inputs

feeds (JSON): list of expected feeds. Per feed:
- name (required): unique identifier, used in alerts and KV keys.
- prefix (required): key prefix; supports {yyyy}, {MM}, {dd}.
- pattern (default .*): Java regex matched against the full key.
- deadline (default 23:59): local HH:mm by which the feed must be complete.
- min_files (default 1), min_bytes (default 1).
- days (default every day): e.g. ["MON","TUE","WED","THU","FRI"].
- bucket (optional): overrides the bucket input.
bucket (STRING, default my-inbound-bucket): default inbound bucket.
timezone (STRING, default UTC): IANA zone for deadlines, "today" and weekdays.
mode (SELECT check | report, default check).

Quick start

Add the secrets to your namespace.
Replace the example feeds with your own, and set bucket and timezone (also set timezone on the daily_arrival_report trigger).
Run once manually — feeds before their deadline show WAITING, feeds past their deadline with no files alert MISSING.
Upload a test file that matches a feed; the next run posts a recovery message.

S3-compatible storage

Add the endpoint to the plugin defaults, for example for MinIO:

pluginDefaults:
  - type: io.kestra.plugin.aws
    values:
      endpointOverride: https://minio.example.com
      compatibilityMode: true
      forcePathStyle: true

Pitfalls

"Today" is based on each object's LastModified in your time zone. Prefixes that hold years of history are listed in full — prefer date-partitioned prefixes ({yyyy}/{MM}/{dd}) or move processed files away.
list_files returns at most 1,000 objects per feed (maxFiles).
Weekday names in days are English three-letter abbreviations.

Expected outputs

outputs.evaluate.values.evaluation (per iteration): JSON such as {"name": "bank_statements", "status": "LATE", "files": 1, "first_arrival": "07:42", "deadline": "07:00", ...}.
KV entries file_sla_<feed>_<yyyy-MM-dd> holding the same JSON.
Slack alerts on state changes and one daily report.

How to extend

Trigger the ingestion flow for a feed as soon as it becomes ON_TIME or LATE with a Subflow task in alert_on_change.
Route alerts per feed by adding a channel or webhook field to each feed.
Open a ticket (Jira, ServiceNow) or page on-call (PagerDuty, Opsgenie) when a business-critical feed is MISSING.
Load the KV states into a warehouse table for monthly SLA reporting per partner.

File Arrival SLA Monitor for S3 Feeds

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Inputs

Quick start

S3-compatible storage

Pitfalls

Expected outputs

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

File Arrival SLA Monitor for S3 Feeds

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Inputs

Quick start

S3-compatible storage

Pitfalls

Expected outputs

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Data Blueprints

Close IFRS 16 leases each month with remeasurements, terminations and the maturity analysis

KeyDB Active-Replica Multi-Master Peering Sentinel

Changelog CI Validation with Liquibase Update-Testing-Rollback

LakeFS Isolated Lakehouse Branching and Atomic Merge Gate

Kubernetes Secret Exposure Sentinel

Detect lasting KPI shifts and name the release that caused them

More Cloud Blueprints

Kubernetes cert-manager ACME Challenge Timeout Sentinel

Kong Ingress Upstream Target Unhealthy and Circuit Breaker Sentinel

Kubernetes API Server Request Latency Spike Sentinel

Kubernetes ConfigMap and Secret Hash Drift Sentinel

Kubernetes CronJob and Batch Job Execution Failure Sentinel

Kubernetes CoreDNS Cluster DNS Resolution and Pod Health Sentinel

New to Kestra?