Kinesis Stream Throughput Quota Gate with Kestra

id: kinesis-stream-throughput-quota-gate
namespace: company.team

description: |
  Read a Kinesis stream's real WriteProvisionedThroughputExceeded and
  GetRecords.IteratorAgeMilliseconds CloudWatch metrics, classify the stream HEALTHY,
  CRITICAL or PROBE_FAILED from the actual values returned, alert Slack with those values,
  and log a final audit summary either way.

inputs:
  - id: stream_name
    type: STRING
    displayName: Kinesis stream name
    description: Name of the Kinesis stream to audit. Used as the StreamName
      dimension on both CloudWatch queries.
    defaults: "orders-stream"

  - id: window
    type: STRING
    displayName: Lookback window
    description: ISO-8601 duration looked back from now for both metrics, for
      example PT15M for 15 minutes. Maps to the CloudWatch Query task's window
      field.
    defaults: "PT15M"

  - id: window_seconds
    type: INT
    displayName: Aggregation period in seconds
    description: Must equal the number of seconds in window, so CloudWatch returns
      exactly one aggregated datapoint for the whole lookback window instead of
      several shorter ones. Maps to the CloudWatch Query task's periodSeconds
      field.
    defaults: 900

  - id: throttled_records_threshold
    type: INT
    displayName: Throttled records threshold
    description: WriteProvisionedThroughputExceeded sum at or above which the stream
      is classified CRITICAL for producer throttling.
    defaults: 1

  - id: iterator_age_critical_ms
    type: INT
    displayName: Iterator age critical threshold in milliseconds
    description: GetRecords.IteratorAgeMilliseconds maximum at or above which the
      stream is classified CRITICAL for consumer lag. Defaults to 12 hours, 50
      percent of the default 24 hour retention period, matching AWS's own
      recommended alarm threshold.
    defaults: 43200000

  - id: alert_on
    type: SELECT
    displayName: Signal to gate on
    description: Which metric drives the CRITICAL classification. throughput only
      watches producer throttling, lag only watches consumer iterator age, both
      watches either.
    values:
      - "throughput"
      - "lag"
      - "both"
    defaults: "both"

tasks:
  - id: probe_throughput_exceeded
    type: io.kestra.plugin.aws.cloudwatch.Query
    description: Read-only. Sum of records rejected due to producer-side throttling
      over the window.
    allowFailure: true
    accessKeyId: "{{ secret('AWS_ACCESS_KEY_ID') }}"
    secretKeyId: "{{ secret('AWS_SECRET_ACCESS_KEY') }}"
    region: "{{ secret('AWS_DEFAULT_REGION') }}"
    namespace: "AWS/Kinesis"
    metricName: "WriteProvisionedThroughputExceeded"
    statistic: "Sum"
    periodSeconds: "{{ inputs.window_seconds }}"
    window: "{{ inputs.window }}"
    dimensions:
      - name: "StreamName"
        value: "{{ inputs.stream_name }}"

  - id: probe_iterator_age
    type: io.kestra.plugin.aws.cloudwatch.Query
    description: Read-only. Maximum age of the oldest unread record across
      GetRecords calls over the window.
    allowFailure: true
    accessKeyId: "{{ secret('AWS_ACCESS_KEY_ID') }}"
    secretKeyId: "{{ secret('AWS_SECRET_ACCESS_KEY') }}"
    region: "{{ secret('AWS_DEFAULT_REGION') }}"
    namespace: "AWS/Kinesis"
    metricName: "GetRecords.IteratorAgeMilliseconds"
    statistic: "Maximum"
    periodSeconds: "{{ inputs.window_seconds }}"
    window: "{{ inputs.window }}"
    dimensions:
      - name: "StreamName"
        value: "{{ inputs.stream_name }}"

  - id: classify_stream_health
    type: io.kestra.plugin.core.debug.Return
    description: PROBE_FAILED when either probe produced no output at all, CRITICAL
      when the signal selected by alert_on is at or above its threshold, HEALTHY
      otherwise.
    format: "{{ (outputs.probe_throughput_exceeded is not defined or
      outputs.probe_iterator_age is not defined) ? 'PROBE_FAILED' :
      ((((inputs.alert_on == 'throughput' or inputs.alert_on == 'both') and
      (outputs.probe_throughput_exceeded.series | length) > 0 and
      (outputs.probe_throughput_exceeded.series | first).sum >=
      inputs.throttled_records_threshold) or ((inputs.alert_on == 'lag' or
      inputs.alert_on == 'both') and (outputs.probe_iterator_age.series |
      length) > 0 and (outputs.probe_iterator_age.series | first).maximum >=
      inputs.iterator_age_critical_ms)) ? 'CRITICAL' : 'HEALTHY') }}"

  - id: route_by_stream_health
    type: io.kestra.plugin.core.flow.Switch
    description: Branch A logs a healthy stream. Branch B alerts Slack when the
      stream is being throttled, a consumer is falling behind, or the probe
      itself failed.
    value: "{{ outputs.classify_stream_health.value }}"
    cases:
      "HEALTHY":
        - id: log_healthy
          type: io.kestra.plugin.core.log.Log
          message: "Kinesis stream {{ inputs.stream_name }} healthy over the last {{
            inputs.window }}. No throttling or excessive iterator age detected."
      "CRITICAL":
        - id: alert_critical
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          url: "{{ secret('SLACK_WEBHOOK') }}"
          payload: |
            {{ {"text": "Kinesis stream " ~ inputs.stream_name ~ " CRITICAL over the last " ~ inputs.window ~ ". WriteProvisionedThroughputExceeded sum=" ~ ((outputs.probe_throughput_exceeded.series | length) > 0 ? (outputs.probe_throughput_exceeded.series | first).sum : "no data") ~ " (threshold " ~ inputs.throttled_records_threshold ~ "). GetRecords.IteratorAgeMilliseconds max=" ~ ((outputs.probe_iterator_age.series | length) > 0 ? (outputs.probe_iterator_age.series | first).maximum : "no data") ~ " ms (threshold " ~ inputs.iterator_age_critical_ms ~ " ms)."} | toJson }}
      "PROBE_FAILED":
        - id: alert_probe_failed
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          url: "{{ secret('SLACK_WEBHOOK') }}"
          payload: |
            {{ {"text": "Kinesis quota probe for stream " ~ inputs.stream_name ~ " did not return a result from one or both CloudWatch queries. Check AWS_ACCESS_KEY_ID, AWS_DEFAULT_REGION, and that the stream exists in that region."} | toJson }}
    defaults:
      - id: log_unhandled_status
        type: io.kestra.plugin.core.log.Log
        message: "Unhandled stream health status: {{
          outputs.classify_stream_health.value }}"

  - id: log_audit_summary
    type: io.kestra.plugin.core.log.Log
    description: Final audit log confirming the gate ran, regardless of the branch
      taken above.
    message: "Kinesis stream throughput quota gate completed. stream={{
      inputs.stream_name }} status={{ outputs.classify_stream_health.value }}
      alert_on={{ inputs.alert_on }} window={{ inputs.window }}"

errors:
  - id: alert_flow_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK') }}"
    payload: |
      {{ {"text": "Kinesis stream throughput quota gate flow failed (execution " ~ execution.id ~ "). Check the execution logs."} | toJson }}

triggers:
  - id: fifteen_minutely
    type: io.kestra.plugin.core.trigger.Schedule
    description: Shipped disabled. Match the cron interval to window for
      non-overlapping coverage.
    cron: "*/15 * * * *"
    disabled: true
    inputs:
      stream_name: "orders-stream"
      window: "PT15M"
      window_seconds: 900
      throttled_records_threshold: 1
      iterator_age_critical_ms: 43200000
      alert_on: "both"

A Kinesis stream's shard count is a hard capacity limit: 1,000 records or 1 MB per second per shard to write, 2 MB per second per shard to read. Outgrow it and PutRecords starts silently dropping records under WriteProvisionedThroughputExceeded, or a slow consumer falls behind until its iterator age approaches the stream's retention window, by default 24 hours, after which the unread records are gone for good. AWS's own CloudWatch documentation recommends alarming on exactly these two metrics, but nothing fires that alarm until you wire it up. This blueprint turns both metrics into a Day 2 capacity gate: it reads the real values CloudWatch reports, classifies the stream, and alerts Slack with the exact numbers before a producer starts failing or a retention window runs out.

How it works

probe_throughput_exceeded and probe_iterator_age (io.kestra.plugin.aws.cloudwatch.Query) each read one AWS/Kinesis metric for stream_name over window, with periodSeconds set equal to window_seconds so CloudWatch returns exactly one aggregated datapoint instead of several. Both have allowFailure: true so a credentials or region problem does not stop the flow before it can be classified as PROBE_FAILED below.
- probe_throughput_exceeded reads WriteProvisionedThroughputExceeded with the Sum statistic, any non-zero sum means records were throttled on the way in.
- probe_iterator_age reads GetRecords.IteratorAgeMilliseconds with the Maximum statistic, AWS's own guidance is to alarm on this once it passes roughly half the stream's retention period.
classify_stream_health (io.kestra.plugin.core.debug.Return) reads .series | first from whichever queries alert_on selects (guarded with is not defined for a probe that produced no output at all) and compares .sum or .maximum against throttled_records_threshold and iterator_age_critical_ms. PROBE_FAILED, CRITICAL, or HEALTHY.
route_by_stream_health (io.kestra.plugin.core.flow.Switch) branches on that status:
- HEALTHY: log_healthy logs it.
- CRITICAL, PROBE_FAILED: a SlackIncomingWebhook task alerts with the exact sum and maximum values from the two queries, never an invented number.
log_audit_summary always runs last and logs the final classification.
The errors block posts to Slack if the flow itself fails outside the gate logic above.

What you get

A real answer to whether a Kinesis stream is being throttled or falling behind, built from the exact metrics AWS recommends alarming on, not a derived guess.
A Slack alert carrying the actual WriteProvisionedThroughputExceeded sum and GetRecords.IteratorAgeMilliseconds maximum, never an invented number.
A choice of which signal matters for a given stream (alert_on), a stream with bursty producers and a fast consumer may only care about throughput, one with a slow batch consumer may only care about iterator age.

Who it's for

Data platform and streaming teams who own a Kinesis stream's shard count and need to know about throttling or consumer lag before a producer starts failing or records expire.
Teams evaluating the CloudWatch plugin who want a working example of io.kestra.plugin.aws.cloudwatch.Query against a real AWS namespace with dimensions.

Why orchestrate this with Kestra

CloudWatch can fire a native alarm, but wiring an SNS topic, a Lambda, and a Slack integration just to get a readable message is its own small project. Kestra adds the schedule, the single declarative query against both metrics, the classification that picks which signal matters per stream, and a Slack alert built from the real numbers, all as one YAML file you can version and review like code next to the rest of your data pipelines.

Prerequisites

A Kinesis stream and an AWS credential with cloudwatch:GetMetricStatistics permission. Reads from CloudWatch, nothing in this blueprint writes to the stream or its shards.
A Slack incoming webhook for alerts.
This plugin calls the AWS SDK directly from the Kestra worker process, there is no sub-container taskRunner and no Docker network to join, any machine Kestra runs on that can reach the AWS CloudWatch API works.
No remediation action ships with this blueprint on purpose: the Kestra Kinesis plugin has no task to split a shard or call UpdateShardCount, resharding has to be done with the AWS CLI or console. CRITICAL is deliberately alert-only.

Secrets

AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, AWS_DEFAULT_REGION: credentials used by both io.kestra.plugin.aws.cloudwatch.Query tasks.
SLACK_WEBHOOK: Slack incoming webhook URL used by the alert tasks and the errors block.

In Kestra OSS, set all four as base64-encoded environment variables: SECRET_AWS_ACCESS_KEY_ID, SECRET_AWS_SECRET_ACCESS_KEY, SECRET_AWS_DEFAULT_REGION, SECRET_SLACK_WEBHOOK.

Inputs

Input	Type	Default	Allowed values
`stream_name`	STRING	`orders-stream`	any Kinesis stream name
`window`	STRING	`PT15M`	any ISO-8601 duration
`window_seconds`	INT	`900`	must equal window in seconds
`throttled_records_threshold`	INT	`1`	any non-negative integer
`iterator_age_critical_ms`	INT	`43200000`	any non-negative integer, in milliseconds
`alert_on`	SELECT	`both`	`throughput`, `lag`, `both`

Quick start

Add the AWS and Slack secrets.
Set stream_name to a real stream and run this flow manually.
Read log_audit_summary for the classification, and compare window_seconds against window if you change either, they must match.
Enable the fifteen_minutely trigger, whose cron matches the default window.

How to extend

When CRITICAL fires on throughput, chain a downstream task that calls the AWS CLI's aws kinesis update-shard-count via a script task to double the shard count, behind a manual approval task, since the Kestra plugin itself has no such task.
Add the ShardId dimension and enable enhanced shard-level monitoring on the stream to find which specific shard is hot, instead of only the stream-level aggregate.
Read iterator_age_critical_ms from the stream's actual RetentionPeriodHours instead of a fixed constant, so the threshold scales automatically if retention changes.

Pitfalls

periodSeconds and window must describe the same length of time. If window_seconds is smaller than window, CloudWatch returns more than one datapoint and .series | first reads only the earliest one, not the whole window.
WriteProvisionedThroughputExceeded and GetRecords.IteratorAgeMilliseconds are basic, stream-level metrics billed at no extra charge, this blueprint does not require enabling Kinesis enhanced (shard-level) monitoring, which has its own per-shard charge.
A brand new or genuinely idle stream can return zero datapoints for a window, which this blueprint treats as HEALTHY since there is nothing to throttle or fall behind on, not as a reason to alert.
This reads GetRecords.IteratorAgeMilliseconds, which only has data if something is actually calling GetRecords against the stream. A stream with no active consumer reports zero datapoints for this metric even while records pile up unread.

Kinesis Stream Throughput Quota Gate

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Inputs

Quick start

How to extend

Pitfalls

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

Kinesis Stream Throughput Quota Gate

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Inputs

Quick start

How to extend

Pitfalls

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Data Blueprints

Automate Customer Feedback Triage, Sentiment Analysis, and Escalation with Google Gemini and Slack

Ingest Zendesk data into DuckDB using dlt

Read out an A/B test with a sample ratio check and a guardrail metric

Schema Contract Guard for dlt Pipelines

AI-Powered Inventory Demand Forecasting, Auto-Generate Purchase Orders with OpenAI

AI Data Quality Anomaly Advisor and SQL Remediation Generator

More Infrastructure Blueprints

Kubernetes Chaos Game Day with Parallel Blast-Radius Watch

Watch Certificate Transparency logs for certificates issued under your domains

Ceph Cluster Health Gate

Score change requests on risk and build the CAB agenda, pre-approving the low-risk ones

Cassandra Tombstone Density Auditor

Cassandra Keyspace Replication Audit Gate

New to Kestra?