Monitoring Kinesis: Metrics, Alarms, and Dashboards

Understanding Kinesis Monitoring

Amazon Kinesis Data Streams is a fully managed service for real-time data ingestion and processing. To keep your streaming pipelines healthy and cost-efficient, you need deep visibility into throughput, latency, errors, and resource utilisation. Monitoring Kinesis involves collecting and analysing performance data, setting automated alerts, and building visual dashboards to quickly identify issues before they impact downstream consumers.

Kinesis sends metrics to Amazon CloudWatch automatically, at no extra charge, covering stream-level and shard-level dimensions. These metrics include data flow rates, throttling, user errors, and the critical IteratorAgeMilliseconds indicator of consumer lag. By combining CloudWatch Metrics, Alarms, and Dashboards, you can build a comprehensive observability layer that notifies you of anomalies, helps you right-size capacity, and simplifies troubleshooting.

Core CloudWatch Metrics for Kinesis Data Streams

Kinesis publishes metrics under the AWS/Kinesis namespace. Each metric can be viewed at the stream level (dimension StreamName) or at the shard level (dimensions StreamName and ShardId). Understanding what each metric represents is essential for effective monitoring.

Throughput and Traffic Metrics

• IncomingRecords – The number of records successfully put into the stream (aggregated per second). Used to track write throughput.
• IncomingBytes – The total bytes of data written to the stream. Helps verify payload sizes and bandwidth.
• GetRecords.Records – The number of records retrieved by consumers via GetRecords calls.
• GetRecords.Bytes – Bytes returned to consumers, useful for monitoring egress traffic.
• PutRecord.Bytes / PutRecords.Bytes – Bytes per individual write operation (for single-record and batch APIs).

Error and Throttling Metrics

• WriteProvisionedThroughputExceeded – Number of write requests rejected due to exceeding the shard’s provisioned limit (1 MiB/sec or 1,000 records/sec per shard). High values indicate the need to scale up (increase shards) or batch writes more efficiently.
• ReadProvisionedThroughputExceeded – Number of GetRecords requests throttled because the consumer exceeded 2 MiB/sec per shard (or 5 GetRecords/sec per shard). Often signals a single hot consumer or misconfigured enhanced fan-out.
• UserErrors – Write requests that fail due to invalid data, access issues, or other client-side problems (sum).
• ThrottledReads / ThrottledWrites – Additional throttling indicators (some regions).

Latency and Consumer Health

• IteratorAgeMilliseconds – The age of the last processed record relative to the newest record in the stream. It’s the single most important metric for detecting consumer lag. A high iterator age means your consumer is falling behind, potentially causing data loss if the retention period (default 24h, max 365 days) expires.
• Success – Whether a GetRecords call returned data (1) or not (0). Persistent zeros suggest empty streams or consumer issues.
• GetRecords.Success – Similar to Success, specifically for GetRecords API.
• SubscribeToShard.Success – For enhanced fan-out (EFO) consumers, indicates whether a subscribe call succeeded.
• MillisBehindLatest – For EFO, the lag in milliseconds behind the latest record. Analogous to IteratorAge but per enhanced fan-out consumer.

Operational Metrics

• ActiveShards – The number of shards currently active (useful to monitor scaling operations).
• OpenShardCount – Number of shards in OPEN state.
• DescribeStreams.LimitExceeded – Rate limit errors when calling DescribeStream, indicating overly frequent API calls.

Setting Up CloudWatch Alarms

Alarms translate metric thresholds into actionable notifications. They can trigger Amazon SNS topics (email, SMS, Lambda), autoscaling actions, or even AWS Systems Manager automations.

Choosing What to Alarm On

Start with the most critical indicators:

• WriteProvisionedThroughputExceeded > 0 for any sustained period – indicates write capacity issues.
• IteratorAgeMilliseconds > threshold (e.g., half of retention period) – signals consumer lag.
• ReadProvisionedThroughputExceeded > 0 – throttled reads.
• UserErrors > 0 – invalid write requests.

Example: Creating an Alarm via AWS CLI

Below we create a CloudWatch alarm that triggers when any shard in the stream my-data-stream experiences throttled writes (sum over 1 minute > 0) for two consecutive periods.

aws cloudwatch put-metric-alarm \
  --alarm-name "KinesisWriteThrottled" \
  --alarm-description "Alarm when write throttling exceeds 0" \
  --namespace "AWS/Kinesis" \
  --metric-name "WriteProvisionedThroughputExceeded" \
  --statistic "Sum" \
  --period 60 \
  --evaluation-periods 2 \
  --threshold 0 \
  --comparison-operator "GreaterThanThreshold" \
  --dimensions "Name=StreamName,Value=my-data-stream" \
  --alarm-actions "arn:aws:sns:us-east-1:123456789012:ops-alerts" \
  --treat-missing-data "notBreaching"

This uses stream-level dimension. For shard-level, add Name=ShardId,Value=shardId-00000000 (but you'd need one alarm per shard, which is better handled via metric math or composite alarms).

CloudFormation Template Snippet

Resources:
  KinesisWriteAlarm:
    Type: AWS::CloudWatch::Alarm
    Properties:
      AlarmName: "MyStreamWriteThrottled"
      AlarmDescription: "Triggers if any write throttling occurs"
      Namespace: "AWS/Kinesis"
      MetricName: "WriteProvisionedThroughputExceeded"
      Statistic: Sum
      Period: 60
      EvaluationPeriods: 2
      Threshold: 0
      ComparisonOperator: GreaterThanThreshold
      Dimensions:
        - Name: StreamName
          Value: my-data-stream
      AlarmActions:
        - !Ref SnsTopicArn
      TreatMissingData: notBreaching

Creating Alarms with Boto3 (Python)

import boto3

cloudwatch = boto3.client('cloudwatch')

response = cloudwatch.put_metric_alarm(
    AlarmName='KinesisIteratorAgeHigh',
    AlarmDescription='Iterator age exceeds 120 seconds',
    Namespace='AWS/Kinesis',
    MetricName='IteratorAgeMilliseconds',
    Statistic='Maximum',
    Period=60,
    EvaluationPeriods=3,
    Threshold=120000,  # 120 seconds in milliseconds
    ComparisonOperator='GreaterThanThreshold',
    Dimensions=[
        {'Name': 'StreamName', 'Value': 'my-data-stream'},
        {'Name': 'ShardId', 'Value': 'shardId-00000000'}  # replace with actual shard ID
    ],
    AlarmActions=['arn:aws:sns:us-east-1:123456789012:ops-alerts'],
    TreatMissingData='notBreaching'
)
print(response)

Note: For shard-level alarms you must create one alarm per shard. Consider using Metric Math to aggregate across shards (e.g., MAX(METRICS(...))) or use composite alarms for a stream-wide view.

Composite Alarms for Shard-Level Rollup

Instead of individual shard alarms, you can create a composite alarm that watches multiple shard alarms. First create shard alarms, then a composite alarm that goes into ALARM when any shard alarm is in ALARM. This reduces alarm noise and simplifies management.

Building CloudWatch Dashboards

Dashboards give you a single pane of glass for stream health. They can combine metrics, alarms, and text widgets to provide instant context.

Dashboard Structure

A dashboard is defined by a JSON body containing an array of widget definitions. Each widget has a type (metric, alarm, text, etc.), position, size, and properties. You can create dashboards via the AWS Console, CLI, or CloudFormation.

Example Dashboard Body (JSON)

Below is a complete dashboard JSON that displays:

• IncomingRecords and GetRecords.Records (stacked area)
• IteratorAgeMilliseconds (line)
• WriteProvisionedThroughputExceeded (alarm status)
• UserErrors (line)
• A text widget explaining the dashboard

{
  "widgets": [
    {
      "type": "text",
      "x": 0, "y": 0, "width": 24, "height": 2,
      "properties": {
        "markdown": "# Kinesis Stream Dashboard\nMonitoring throughput, lag, and errors for stream `my-data-stream`"
      }
    },
    {
      "type": "metric",
      "x": 0, "y": 2, "width": 12, "height": 6,
      "properties": {
        "view": "timeSeries",
        "stacked": true,
        "metrics": [
          [ "AWS/Kinesis", "IncomingRecords", { "stat": "Sum", "label": "Incoming Records" } ],
          [ ".", "GetRecords.Records", { "stat": "Sum", "label": "GetRecords Records" } ]
        ],
        "region": "us-east-1",
        "period": 60,
        "title": "Throughput (records/min)",
        "yAxis": { "left": { "showUnits": false } }
      }
    },
    {
      "type": "metric",
      "x": 12, "y": 2, "width": 12, "height": 6,
      "properties": {
        "view": "timeSeries",
        "stacked": false,
        "metrics": [
          [ "AWS/Kinesis", "IteratorAgeMilliseconds", { "stat": "Maximum", "label": "Iterator Age (ms)" } ]
        ],
        "region": "us-east-1",
        "period": 60,
        "title": "Consumer Lag",
        "yAxis": { "left": { "label": "Milliseconds" } }
      }
    },
    {
      "type": "metric",
      "x": 0, "y": 8, "width": 8, "height": 6,
      "properties": {
        "view": "timeSeries",
        "stacked": false,
        "metrics": [
          [ "AWS/Kinesis", "WriteProvisionedThroughputExceeded", { "stat": "Sum", "label": "Write Throttles" } ]
        ],
        "region": "us-east-1",
        "period": 60,
        "title": "Write Throttling"
      }
    },
    {
      "type": "metric",
      "x": 8, "y": 8, "width": 8, "height": 6,
      "properties": {
        "view": "timeSeries",
        "stacked": false,
        "metrics": [
          [ "AWS/Kinesis", "UserErrors", { "stat": "Sum", "label": "User Errors" } ]
        ],
        "region": "us-east-1",
        "period": 60,
        "title": "User Errors"
      }
    },
    {
      "type": "alarm",
      "x": 16, "y": 8, "width": 8, "height": 6,
      "properties": {
        "alarms": [
          "arn:aws:cloudwatch:us-east-1:123456789012:alarm:KinesisWriteThrottled"
        ],
        "title": "Write Alarm Status"
      }
    }
  ]
}

Replace