Find Idle Kinesis Data Streams

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "ListStreamsAndMetrics",
      "Effect": "Allow",
      "Action": ["kinesis:ListStreams", "cloudwatch:GetMetricData"],
      "Resource": "*"
    },
    {
      "Sid": "DescribeStreams",
      "Effect": "Allow",
      "Action": ["kinesis:DescribeStreamSummary", "kinesis:ListStreamConsumers"],
      "Resource": "arn:aws:kinesis:us-east-1:111122223333:stream/*"
    }
  ]
}

// find-idle-kinesis-data-streams.ts
// Lists Kinesis data streams in one Region, sums 14 days of writes (IncomingRecords), GetRecords reads and
// enhanced fan-out reads from CloudWatch, and estimates the fixed monthly charge each stream keeps paying.
// Report only: it never changes or deletes a stream.
// Usage:
//   npx tsx find-idle-kinesis-data-streams.ts [--region us-east-1] [--days 14]
import {
  KinesisClient,
  DescribeStreamSummaryCommand,
  paginateListStreamConsumers,
  paginateListStreams,
} from "@aws-sdk/client-kinesis";
import { CloudWatchClient, GetMetricDataCommand, type MetricDataQuery } from "@aws-sdk/client-cloudwatch";

const args = process.argv.slice(2);
const flag = (name: string): string | undefined => {
  const i = args.indexOf(name);
  return i >= 0 ? args[i + 1] : undefined;
};
const region = flag("--region") ?? process.env.AWS_REGION ?? "us-east-1";
const days = Number(flag("--days") ?? 14);
const HOURS_PER_MONTH = 730;
// us-east-1 prices from the AWS Price List API (AmazonKinesis offer), September 2026.
const PRICE = {
  shardHour: 0.015, // provisioned shard-hour
  extendedRetentionShardHour: 0.02, // provisioned, retention above 24 hours up to 7 days
  efoConsumerShardHour: 0.015, // provisioned, enhanced fan-out consumer-shard hour
  onDemandStreamHour: 0.04, // on-demand (standard) stream-hour
};
const kinesis = new KinesisClient({ region });
const cloudwatch = new CloudWatchClient({ region });

interface Row {
  Stream: string;
  Mode: string;
  Shards: number;
  RetentionH: number;
  EFO: number;
  Writes: number;
  Reads: number;
  Verdict: string;
  FixedUsdPerMonth: number;
}

const errText = (err: unknown): string => (err instanceof Error ? `${err.name}: ${err.message}` : String(err));

// Sums one metric per query over the whole window (a single period covering all the days).
async function sums(queries: { id: string; metric: string; dims: Record<string, string> }[]): Promise<Map<string, number>> {
  const end = new Date();
  const start = new Date(end.getTime() - days * 86_400_000);
  const MetricDataQueries: MetricDataQuery[] = queries.map((q) => ({
    Id: q.id,
    MetricStat: {
      Metric: {
        Namespace: "AWS/Kinesis",
        MetricName: q.metric,
        Dimensions: Object.entries(q.dims).map(([Name, Value]) => ({ Name, Value })),
      },
      Period: days * 86_400,
      Stat: "Sum",
    },
  }));
  const result = new Map<string, number>();
  let NextToken: string | undefined;
  do {
    const out = await cloudwatch.send(new GetMetricDataCommand({ StartTime: start, EndTime: end, MetricDataQueries, NextToken }));
    for (const r of out.MetricDataResults ?? []) {
      const total = (r.Values ?? []).reduce((a, b) => a + b, 0);
      result.set(r.Id ?? "", (result.get(r.Id ?? "") ?? 0) + total);
    }
    NextToken = out.NextToken;
  } while (NextToken);
  return result;
}

async function inspect(name: string): Promise<Row> {
  const s = (await kinesis.send(new DescribeStreamSummaryCommand({ StreamName: name }))).StreamDescriptionSummary;
  const mode = s?.StreamModeDetails?.StreamMode ?? "PROVISIONED";
  const shards = s?.OpenShardCount ?? 0;
  const retention = s?.RetentionPeriodHours ?? 24;
  const consumers: string[] = [];
  if (s?.StreamARN) {
    for await (const page of paginateListStreamConsumers({ client: kinesis }, { StreamARN: s.StreamARN })) {
      for (const c of page.Consumers ?? []) if (c.ConsumerName) consumers.push(c.ConsumerName);
    }
  }
  const queries = [
    { id: "writes", metric: "IncomingRecords", dims: { StreamName: name } },
    { id: "reads", metric: "GetRecords.Records", dims: { StreamName: name } },
    ...consumers.map((c, i) => ({ id: `efo${i}`, metric: "SubscribeToShardEvent.Records", dims: { StreamName: name, ConsumerName: c } })),
  ];
  const m = await sums(queries);
  const writes = m.get("writes") ?? 0;
  let reads = m.get("reads") ?? 0;
  consumers.forEach((_c, i) => (reads += m.get(`efo${i}`) ?? 0));

  let fixed: number;
  if (mode === "ON_DEMAND") {
    fixed = PRICE.onDemandStreamHour * HOURS_PER_MONTH;
  } else {
    fixed = shards * PRICE.shardHour * HOURS_PER_MONTH;
    if (retention > 24) fixed += shards * PRICE.extendedRetentionShardHour * HOURS_PER_MONTH;
    fixed += consumers.length * shards * PRICE.efoConsumerShardHour * HOURS_PER_MONTH;
  }
  const verdict = writes === 0 && reads === 0 ? "IDLE" : writes > 0 && reads === 0 ? "no readers" : writes === 0 ? "no new data" : "active";
  return {
    Stream: name, Mode: mode, Shards: shards, RetentionH: retention, EFO: consumers.length,
    Writes: writes, Reads: reads, Verdict: verdict, FixedUsdPerMonth: Number(fixed.toFixed(2)),
  };
}

async function main(): Promise<void> {
  if (!Number.isFinite(days) || days < 1 || days > 63) throw new Error("--days must be between 1 and 63");
  const rows: Row[] = [];
  for await (const page of paginateListStreams({ client: kinesis }, {})) {
    for (const s of page.StreamSummaries ?? []) {
      if (!s.StreamName) continue;
      try {
        rows.push(await inspect(s.StreamName));
      } catch (err) {
        console.error(`Skipping ${s.StreamName}: ${errText(err)}`);
      }
    }
  }
  rows.sort((a, b) => Number(b.Verdict === "IDLE") - Number(a.Verdict === "IDLE") || b.FixedUsdPerMonth - a.FixedUsdPerMonth);
  console.table(rows);
  const idle = rows.filter((r) => r.Verdict === "IDLE");
  const cost = idle.reduce((sum, r) => sum + r.FixedUsdPerMonth, 0);
  console.log(`${rows.length} streams in ${region}; ${idle.length} idle for ${days} days, about $${cost.toFixed(2)}/month in fixed charges. Nothing was changed.`);
}

main().catch((err) => {
  console.error(errText(err));
  process.exit(1);
});

npm install @aws-sdk/client-kinesis @aws-sdk/client-cloudwatch
npm install --save-dev tsx typescript @types/node

# Last 14 days in us-east-1
AWS_PROFILE=readonly npx tsx find-idle-kinesis-data-streams.ts --region us-east-1

# A longer window for monthly consumers
AWS_PROFILE=readonly npx tsx find-idle-kinesis-data-streams.ts --region us-east-1 --days 45

┌─────────┬─────────────────┬───────────────┬────────┬────────────┬─────┬──────────┬──────────┬──────────────┬──────────────────┐
│ (index) │ Stream          │ Mode          │ Shards │ RetentionH │ EFO │ Writes   │ Reads    │ Verdict      │ FixedUsdPerMonth │
├─────────┼─────────────────┼───────────────┼────────┼────────────┼─────┼──────────┼──────────┼──────────────┼──────────────────┤
│ 0       │ 'legacy-orders' │ 'PROVISIONED' │ 4      │ 168        │ 1   │ 0        │ 0        │ 'IDLE'       │ 146              │
│ 1       │ 'iot-test'      │ 'ON_DEMAND'   │ 4      │ 24         │ 0   │ 0        │ 0        │ 'IDLE'       │ 29.2             │
│ 2       │ 'clickstream'   │ 'ON_DEMAND'   │ 4      │ 24         │ 0   │ 48211930 │ 96420011 │ 'active'     │ 29.2             │
│ 3       │ 'audit-events'  │ 'PROVISIONED' │ 2      │ 24         │ 0   │ 120455   │ 0        │ 'no readers' │ 21.9             │
└─────────┴─────────────────┴───────────────┴────────┴────────────┴─────┴──────────┴──────────┴──────────────┴──────────────────┘
4 streams in us-east-1; 2 idle for 14 days, about $175.20/month in fixed charges. Nothing was changed.