feat(kinesis): shard-level metrics for stream

packages/@aws-cdk-testing/framework-integ/test/aws-kinesis/test/integ.stream-shard-level-monitoring.ts

@@ -0,0 +1,48 @@
+import { App, Stack } from 'aws-cdk-lib';
+import * as kinesis from 'aws-cdk-lib/aws-kinesis';
+import { ExpectedResult, IntegTest } from '@aws-cdk/integ-tests-alpha';
+import { RemovalPolicies } from '../../../../../aws-cdk-lib/core/lib/removal-policies';
+
+const app = new App();
+const stack = new Stack(app, 'kinesis-stream-shard-level-monitoring-stack');
+
+const explicitStream = new kinesis.Stream(stack, 'ExplicitStream', {
+  shardLevelMetrics: [
+    kinesis.ShardLevelMetrics.INCOMING_BYTES,
+    kinesis.ShardLevelMetrics.INCOMING_RECORDS,
+    kinesis.ShardLevelMetrics.ITERATOR_AGE_MILLISECONDS,
+    kinesis.ShardLevelMetrics.OUTGOING_BYTES,
+    kinesis.ShardLevelMetrics.OUTGOING_RECORDS,
+    kinesis.ShardLevelMetrics.WRITE_PROVISIONED_THROUGHPUT_EXCEEDED,
+    kinesis.ShardLevelMetrics.READ_PROVISIONED_THROUGHPUT_EXCEEDED,
+  ],
+});
+
+const allStream = new kinesis.Stream(stack, 'AllStream', {
+  shardLevelMetrics: [kinesis.ShardLevelMetrics.ALL],
+});
+
+RemovalPolicies.of(stack).destroy();
+
+const integ = new IntegTest(app, 'integ-kinesis-stream-consumer', {
+  testCases: [stack],
+});
+
+const streams = [explicitStream, allStream];
+streams.forEach((stream) => {
+  integ.assertions.awsApiCall('Kinesis', 'describeStream', {
+    StreamName: stream.streamName,
+  }).expect(ExpectedResult.objectLike({
+    StreamDescription: {
+      ShardLevelMetrics: [
+        'IncomingBytes',
+        'IncomingRecords',
+        'IteratorAgeMilliseconds',
+        'OutgoingBytes',
+        'OutgoingRecords',
+        'WriteProvisionedThroughputExceeded',
+        'ReadProvisionedThroughputExceeded',
+      ],
+    },
+  }));
+});

packages/aws-cdk-lib/aws-kinesis/lib/stream.ts

@@ -31,6 +31,53 @@ const WRITE_OPERATIONS = [
   'kinesis:PutRecords',
 ];
 
+/**
+ * Enhanced shard-level metrics
+ *
+ * @see https://docs.aws.amazon.com/streams/latest/dev/monitoring-with-cloudwatch.html#kinesis-metrics-shard
+ */
+export enum ShardLevelMetrics {
+  /**
+   * The number of bytes successfully put to the shard over the specified time period.
+   */
+  INCOMING_BYTES = 'IncomingBytes',
+
+  /**
+   * The number of records successfully put to the shard over the specified time period.
+   */
+  INCOMING_RECORDS = 'IncomingRecords',
+
+  /**
+   * The age of the last record in all GetRecords calls made against a shard, measured over the specified time period.
+   */
+  ITERATOR_AGE_MILLISECONDS = 'IteratorAgeMilliseconds',
+
+  /**
+   * The number of bytes retrieved from the shard, measured over the specified time period.
+   */
+  OUTGOING_BYTES = 'OutgoingBytes',
+
+  /**
+   * The number of records retrieved from the shard, measured over the specified time period.
+   */
+  OUTGOING_RECORDS = 'OutgoingRecords',
+
+  /**
+   * The number of GetRecords calls throttled for the shard over the specified time period.
+   */
+  READ_PROVISIONED_THROUGHPUT_EXCEEDED = 'ReadProvisionedThroughputExceeded',
+
+  /**
+   * The number of records rejected due to throttling for the shard over the specified time period.
+   */
+  WRITE_PROVISIONED_THROUGHPUT_EXCEEDED = 'WriteProvisionedThroughputExceeded',
+
+  /**
+   * All metrics
+   */
+  ALL = 'ALL',
+}
+
 /**
  * A Kinesis Stream
  */
@@ -784,6 +831,15 @@ export interface StreamProps {
    * @default RemovalPolicy.RETAIN
    */
   readonly removalPolicy?: RemovalPolicy;
+
+  /**
+   * A list of shard-level metrics in properties to enable enhanced monitoring mode.
+   *
+   * @see https://docs.aws.amazon.com/streams/latest/dev/monitoring-with-cloudwatch.html#kinesis-metrics-shard
+   *
+   * @default undefined - AWS Kinesis default is disabled
+   */
+  readonly shardLevelMetrics?: ShardLevelMetrics[];
 }
 
 /**
@@ -861,6 +917,7 @@ export class Stream extends StreamBase {
       retentionPeriodHours,
       shardCount,
       streamEncryption,
+      desiredShardLevelMetrics: props.shardLevelMetrics,
       ...(props.streamMode !== undefined
         ? {
           streamModeDetails: { streamMode: props.streamMode },