AWS Kinesis Management Skill

Analyze and manage Kinesis streams with safe, read-only operations.

MANDATORY: Two-Phase Execution

You MUST follow this two-phase pattern. Skipping Phase 1 causes hallucinated stream/consumer names.

Phase 1: Discovery (ALWAYS run first)

#!/bin/bash

# 1. List streams
aws kinesis list-streams --output json | jq -r '.StreamNames[]'

# 2. Describe stream
aws kinesis describe-stream-summary --stream-name "$STREAM_NAME" --output json

# 3. List shards
aws kinesis list-shards --stream-name "$STREAM_NAME" --output json | jq '.Shards[] | {ShardId, HashKeyRange, SequenceNumberRange}'

# 4. List stream consumers (enhanced fan-out)
aws kinesis list-stream-consumers --stream-arn "$STREAM_ARN" --output json

# 5. Describe stream consumer
aws kinesis describe-stream-consumer --stream-arn "$STREAM_ARN" --consumer-name "$CONSUMER_NAME" --output json 2>/dev/null

Phase 1 outputs:

• Stream names and ARNs
• Shard count and configuration
• Registered consumers (enhanced fan-out)

Phase 2: Analysis (only after Phase 1)

Only reference stream names, shard IDs, and consumer names confirmed in Phase 1.

Shell Script Patterns

Helper Function

#!/bin/bash

# Core Kinesis helper — always use this
kinesis_cmd() {
    aws kinesis "$@" --output json
}

# CloudWatch metric for Kinesis
kinesis_metric() {
    local stream="$1" metric="$2" stat="${3:-Sum}" period="${4:-300}"
    aws cloudwatch get-metric-statistics \
        --namespace AWS/Kinesis \
        --metric-name "$metric" \
        --dimensions Name=StreamName,Value="$stream" \
        --start-time "$(date -u -v-1H +%Y-%m-%dT%H:%M:%S 2>/dev/null || date -u -d '1 hour ago' +%Y-%m-%dT%H:%M:%S)" \
        --end-time "$(date -u +%Y-%m-%dT%H:%M:%S)" \
        --period "$period" \
        --statistics "$stat" \
        --output json
}

Anti-Hallucination Rules

• NEVER reference a stream without confirming via list-streams
• NEVER reference shard IDs without confirming via list-shards
• NEVER assume consumer names — always list stream consumers
• NEVER guess shard count — always describe the stream
• NEVER assume on-demand vs provisioned — check stream mode

Safety Rules

• READ-ONLY ONLY: Use only describe-, list-, CloudWatch get-metric-statistics
• FORBIDDEN: create-stream, delete-stream, put-record, merge-shards, split-shard without explicit user request
• NEVER get records from production streams without explicit user request
• Use CloudWatch metrics for throughput analysis, not get-records

Common Operations

Stream Overview

#!/bin/bash
echo "=== Kinesis Streams ==="
for STREAM in $(aws kinesis list-streams --output json | jq -r '.StreamNames[]'); do
    SUMMARY=$(kinesis_cmd describe-stream-summary --stream-name "$STREAM" | jq '.StreamDescriptionSummary')
    echo "$STREAM:"
    echo "  Status: $(echo $SUMMARY | jq -r '.StreamStatus')"
    echo "  Shards: $(echo $SUMMARY | jq -r '.OpenShardCount')"
    echo "  Mode: $(echo $SUMMARY | jq -r '.StreamModeDetails.StreamMode')"
    echo "  Consumers: $(echo $SUMMARY | jq -r '.ConsumerCount')"
    echo "  Retention: $(echo $SUMMARY | jq -r '.RetentionPeriodHours')h"
done

Shard Analysis

#!/bin/bash
STREAM="${1:-my-stream}"

echo "=== Shards ==="
kinesis_cmd list-shards --stream-name "$STREAM" | jq '.Shards[] | {ShardId, ParentShardId: .ParentShardId, HashKeyRange: "\(.HashKeyRange.StartingHashKey[0:10])...\(.HashKeyRange.EndingHashKey[0:10])"}'

echo ""
echo "=== Write Throughput per Shard ==="
kinesis_metric "$STREAM" "IncomingRecords" "Sum" 60 | jq -r '.Datapoints | sort_by(.Timestamp) | .[-5:][] | "\(.Timestamp)\t\(.Sum) records"'

echo ""
echo "=== Read Throughput ==="
kinesis_metric "$STREAM" "GetRecords.Records" "Sum" 60 | jq -r '.Datapoints | sort_by(.Timestamp) | .[-5:][] | "\(.Timestamp)\t\(.Sum) records"'

Consumer Lag Analysis

#!/bin/bash
STREAM="${1:-my-stream}"

echo "=== Iterator Age (lag) ==="
kinesis_metric "$STREAM" "GetRecords.IteratorAgeMilliseconds" "Maximum" 60 | jq -r '.Datapoints | sort_by(.Timestamp) | .[-5:][] | "\(.Timestamp)\t\(.Maximum)ms"'

echo ""
echo "=== Read Provisioned Throughput Exceeded ==="
kinesis_metric "$STREAM" "ReadProvisionedThroughputExceeded" "Sum" 300 | jq -r '.Datapoints | sort_by(.Timestamp) | .[] | select(.Sum > 0) | "\(.Timestamp)\t\(.Sum)"'

echo ""
echo "=== Write Provisioned Throughput Exceeded ==="
kinesis_metric "$STREAM" "WriteProvisionedThroughputExceeded" "Sum" 300 | jq -r '.Datapoints | sort_by(.Timestamp) | .[] | select(.Sum > 0) | "\(.Timestamp)\t\(.Sum)"'

echo ""
echo "=== Enhanced Fan-out Consumers ==="
STREAM_ARN=$(kinesis_cmd describe-stream-summary --stream-name "$STREAM" | jq -r '.StreamDescriptionSummary.StreamARN')
kinesis_cmd list-stream-consumers --stream-arn "$STREAM_ARN" | jq '.Consumers[] | {ConsumerName, ConsumerStatus, ConsumerCreationTimestamp}'

Throughput Analysis

#!/bin/bash
STREAM="${1:-my-stream}"

echo "=== Incoming Bytes ==="
kinesis_metric "$STREAM" "IncomingBytes" "Sum" 60 | jq -r '.Datapoints | sort_by(.Timestamp) | .[-5:][] | "\(.Timestamp)\t\(.Sum / 1024 | round)KB"'

echo ""
echo "=== Outgoing Bytes ==="
kinesis_metric "$STREAM" "GetRecords.Bytes" "Sum" 60 | jq -r '.Datapoints | sort_by(.Timestamp) | .[-5:][] | "\(.Timestamp)\t\(.Sum / 1024 | round)KB"'

echo ""
echo "=== Put Latency ==="
kinesis_metric "$STREAM" "PutRecord.Latency" "Average" 60 | jq -r '.Datapoints | sort_by(.Timestamp) | .[-5:][] | "\(.Timestamp)\t\(.Average)ms"'

Output Format

Present results as a structured report:

Managing Aws Kinesis Report
═══════════════════════════
Resources discovered: [count]

Resource       Status    Key Metric    Issues
──────────────────────────────────────────────
[name]         [ok/warn] [value]       [findings]

Summary: [total] resources | [ok] healthy | [warn] warnings | [crit] critical
Action Items: [list of prioritized findings]

Target ≤50 lines of output. Use tables for multi-resource comparisons.

Counter-Rationalizations

Shortcut	Counter	Why
"I'll skip discovery and check known resources"	Always run Phase 1 discovery first	Resource names change, new resources appear — assumed names cause errors
"The user only asked for a quick check"	Follow the full discovery → analysis flow	Quick checks miss critical issues; structured analysis catches silent failures
"Default configuration is probably fine"	Audit configuration explicitly	Defaults often leave logging, security, and optimization features disabled
"Metrics aren't needed for this"	Always check relevant metrics when available	API/CLI responses show current state; metrics reveal trends and intermittent issues
"I don't have access to that"	Try the command and report the actual error	Assumed permission failures prevent useful investigation; actual errors are informative

Common Pitfalls

• Shard limits: Each shard supports 1MB/s write and 2MB/s read — exceeding causes throttling
• Hot shards: Skewed partition keys cause hot shards — check per-shard metrics
• Iterator age: High iterator age means consumers are falling behind — investigate consumer health
• Enhanced fan-out cost: Each registered consumer adds cost — use shared throughput when possible
• Resharding lag: Shard splitting/merging creates parent-child relationships — consumers must handle this
• Retention period: Default 24h, max 8760h (365 days) — data beyond retention is lost permanently