AWS Lambda IteratorAge Stream Processing Lag Sentinel

id: aws-lambda-iterator-age-stream-lag-sentinel
namespace: company.serverless
description: Audits AWS Lambda IteratorAge stream processing metrics via
  CloudWatch, detects stream consumer lag across Kinesis and DynamoDB streams,
  logs a GitHub issue, and alerts Slack.
inputs:
  - id: aws_region
    type: STRING
    displayName: AWS Region
    defaults: us-east-1
  - id: function_name
    type: STRING
    displayName: Lambda Function Name
    defaults: event-stream-processor
  - id: iterator_age_ms_threshold
    type: INT
    displayName: Max IteratorAge (Milliseconds)
    defaults: 60000
  - id: target_repository
    type: STRING
    displayName: Serverless Issue Repository
    defaults: my-org/serverless-operations
triggers:
  - id: periodic_iterator_age_check
    type: io.kestra.plugin.core.trigger.Schedule
    cron: '*/15 * * * *'
    disabled: true
tasks:
  - id: probe_iterator_age_metric
    type: io.kestra.plugin.aws.cli.AwsCLI
    description: Queries CloudWatch for Lambda IteratorAge metric over the last 15 minutes
    accessKeyId: '{{ secret(''AWS_ACCESS_KEY_ID'') }}'
    secretKeyId: '{{ secret(''AWS_SECRET_ACCESS_KEY'') }}'
    region: '{{ inputs.aws_region }}'
    commands:
      - aws cloudwatch get-metric-data --metric-data-queries
        '[{"Id":"m1","MetricStat":{"Metric":{"Namespace":"AWS/Lambda","MetricName":"IteratorAge","Dimensions":[{"Name":"FunctionName","Value":"{{
        inputs.function_name
        }}"}]},"Period":900,"Stat":"Maximum"},"ReturnData":true}]' --start-time
        $(date -u -v-15M +%Y-%m-%dT%H:%M:%SZ) --end-time $(date -u
        +%Y-%m-%dT%H:%M:%SZ) --output json | tr -d '\n' | xargs -0 -I {} echo
        '::{"outputs":{}}::'
  - id: evaluate_stream_lag
    type: io.kestra.plugin.core.debug.Return
    description: Evaluates maximum iterator age against lag threshold
    format:
      function: '{{ inputs.function_name }}'
      max_iterator_age_ms: '{{
        outputs.probe_iterator_age_metric.vars.MetricDataResults[0].Values[0] ??
        0 }}'
      threshold_ms: '{{ inputs.iterator_age_ms_threshold }}'
      is_lag_critical: '{{
        (outputs.probe_iterator_age_metric.vars.MetricDataResults[0].Values[0]
        ?? 0) >= inputs.iterator_age_ms_threshold }}'
  - id: branch_on_stream_lag
    type: io.kestra.plugin.core.flow.If
    description: Branches based on whether iterator age exceeded threshold
    condition: '{{ outputs.evaluate_stream_lag.value.is_lag_critical == true }}'
    then:
      - id: create_lambda_incident_issue
        type: io.kestra.plugin.github.issues.Create
        description: Logs a prioritized GitHub issue documenting stream processing lag
        oauthToken: '{{ secret(''GITHUB_TOKEN'') }}'
        repository: '{{ inputs.target_repository }}'
        title: '[Lambda Alert] High IteratorAge Stream Lag on {{ inputs.function_name
          }}'
        body: "## 🚨 AWS Lambda Stream Processing Latency (IteratorAge)


          The serverless sentinel detected elevated IteratorAge on streaming
          event source:

          - **Function Name:** `{{ inputs.function_name }}`

          - **Peak IteratorAge:** **{{
          outputs.evaluate_stream_lag.value.max_iterator_age_ms }} ms**

          - **Threshold:** `{{ inputs.iterator_age_ms_threshold }} ms`


          Please verify Lambda batch size, concurrency throttling, and execution
          duration.

          *Generated automatically by Kestra Lambda Sentinel.*"
      - id: alert_serverless_slack
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Dispatches high-priority alert to serverless Slack channel
        url: '{{ secret(''SLACK_WEBHOOK'') }}'
        payload: "{{ {'text': '🚨 *AWS Lambda Stream Lag Alert*\\n*Function:* `' ~
          inputs.function_name ~ '`\\n*IteratorAge:* *' ~
          outputs.evaluate_stream_lag.value.max_iterator_age_ms ~ 'ms*
          (Threshold: `' ~ inputs.iterator_age_ms_threshold ~ 'ms`)\\n*Incident
          Ticket:* <' ~ (outputs.create_lambda_incident_issue.issueUrl ?? '#') ~
          '|View GitHub Issue>'} | toJson }}"
    else:
      - id: log_stream_healthy
        type: io.kestra.plugin.core.log.Log
        message: Lambda function {{ inputs.function_name }} stream processing is up to
          date ({{ outputs.evaluate_stream_lag.value.max_iterator_age_ms }}ms <=
          {{ inputs.iterator_age_ms_threshold }}ms).
  - id: log_audit_complete
    type: io.kestra.plugin.core.log.Log
    description: Final audit status log
    message: 'Lambda stream lag audit complete for {{ inputs.function_name }}. Age:
      {{ outputs.evaluate_stream_lag.value.max_iterator_age_ms }}ms.'
errors:
  - id: alert_flow_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Alerts Slack on execution failure
    url: '{{ secret(''SLACK_WEBHOOK'') }}'
    payload: "{{ {'text': '⚠️ *AWS Lambda Sentinel Failed* in execution `' ~
      execution.id ~ '`.'} | toJson }}"