feat(kinesis): add more alarms (#105)

- adds alarms of Provisioned Capacity exceeded on Read/Write
- adds alarm of Throttled Records for Kinesis Data Stream
- improves unit tests (adds dashboard code to snapshots)

Closes #53 

---

_By submitting this pull request, I confirm that my contribution is made under the terms of the Apache-2.0 license_

Author: voho

API.md

@@ -82,8 +82,8 @@ You can also browse the documentation at https://constructs.dev/packages/cdk-mon
 | AWS ElastiCache (`.monitorElastiCacheCluster()`) | CPU/memory usage, evictions and connections | | |
 | AWS Glue (`.monitorGlueJob()`) | Traffic, job status, memory/CPU usage | | |
 | AWS Kinesis Data Analytics (`.monitorKinesisDataAnalytics`) | Up/Downtime, CPU/memory usage, KPU usage, checkpoint metrics, and garbage collection metrics | Downtime | |
-| AWS Kinesis Data Stream (`.monitorKinesisDataStream()`) | Put/Get/Incoming Record/s and Throttling | Iterator max age | |
-| AWS Kinesis Firehose (`.monitorKinesisFirehose()`) | Number of records, requests, latency | | |
+| AWS Kinesis Data Stream (`.monitorKinesisDataStream()`) | Put/Get/Incoming Record/s and Throttling | Throttling, iterator max age | |
+| AWS Kinesis Firehose (`.monitorKinesisFirehose()`) | Number of records, requests, latency, throttling | | |
 | AWS Lambda (`.monitorLambdaFunction()`) | Latency, errors, iterator max age | Latency, errors, throttles, iterator max age | Optional Lambda Insights metrics (opt-in) support |
 | AWS Load Balancing (`.monitorNetworkLoadBalancer()`, `.monitorFargateApplicationLoadBalancer()`, `.monitorFargateNetworkLoadBalancer()`, `.monitorEc2ApplicationLoadBalancer()`, `.monitorEc2NetworkLoadBalancer()`) | System resources and task health | Unhealthy task count, running tasks count, (for Fargate/Ec2 apps) CPU/memory usage | Use for FargateService or Ec2Service backed by a NetworkLoadBalancer or ApplicationLoadBalancer |
 | AWS OpenSearch/Elasticsearch (`.monitorOpenSearchCluster()`, `.monitorElasticsearchCluster()`) | Indexing and search latency, disk/memory/CPU usage | Indexing and search latency, disk/memory/CPU usage, cluster status | |

README.md

@@ -82,11 +82,51 @@ export class KinesisAlarmFactory {
         ComparisonOperator.GREATER_THAN_THRESHOLD,
       ...props,
       disambiguator,
-      threshold: threshold,
+      threshold,
       alarmNameSuffix: "PutRecordsFailed",
       alarmDescription: `Number of failed PutRecords exceeded threshold of ${threshold}`,
       // we will dedupe any kind of message count issue to the same ticket
       alarmDedupeStringSuffix: "PutRecordsFailed",
     });
   }
+
+  addProvisionedReadThroughputExceededAlarm(
+    metric: MetricWithAlarmSupport,
+    props: RecordsThrottledThreshold,
+    disambiguator: string
+  ) {
+    const threshold = props.maxRecordsThrottledThreshold;
+    return this.alarmFactory.addAlarm(metric, {
+      treatMissingData:
+        props.treatMissingDataOverride ?? TreatMissingData.NOT_BREACHING,
+      comparisonOperator:
+        props.comparisonOperatorOverride ??
+        ComparisonOperator.GREATER_THAN_THRESHOLD,
+      ...props,
+      disambiguator,
+      threshold,
+      alarmNameSuffix: "ReadThroughputExceeded",
+      alarmDescription: `Number of records resulting in read throughput capacity throttling reached the threshold of ${threshold}.`,
+    });
+  }
+
+  addProvisionedWriteThroughputExceededAlarm(
+    metric: MetricWithAlarmSupport,
+    props: RecordsThrottledThreshold,
+    disambiguator: string
+  ) {
+    const threshold = props.maxRecordsThrottledThreshold;
+    return this.alarmFactory.addAlarm(metric, {
+      treatMissingData:
+        props.treatMissingDataOverride ?? TreatMissingData.NOT_BREACHING,
+      comparisonOperator:
+        props.comparisonOperatorOverride ??
+        ComparisonOperator.GREATER_THAN_THRESHOLD,
+      ...props,
+      disambiguator,
+      threshold,
+      alarmNameSuffix: "WriteThroughputExceeded",
+      alarmDescription: `Number of records resulting in write throughput capacity throttling reached the threshold of ${threshold}.`,
+    });
+  }
 }

lib/common/monitoring/alarms/KinesisAlarmFactory.ts

@@ -40,6 +40,15 @@ export interface KinesisDataStreamMonitoringOptions
     RecordsThrottledThreshold
   >;
   readonly addPutRecordsFailedAlarm?: Record<string, RecordsFailedThreshold>;
+  readonly addThrottledRecordsAlarm?: Record<string, RecordsThrottledThreshold>;
+  readonly addReadProvisionedThroughputExceededAlarm?: Record<
+    string,
+    RecordsThrottledThreshold
+  >;
+  readonly addWriteProvisionedThroughputExceededAlarm?: Record<
+    string,
+    RecordsThrottledThreshold
+  >;
 }
 
 export interface KinesisDataStreamMonitoringProps
@@ -52,6 +61,7 @@ export class KinesisDataStreamMonitoring extends Monitoring {
 
   protected readonly kinesisAlarmFactory: KinesisAlarmFactory;
   protected readonly ageAnnotations: HorizontalAnnotation[];
+  protected readonly provisionedCapacityAnnotations: HorizontalAnnotation[];
   protected readonly recordCountAnnotations: HorizontalAnnotation[];
 
   protected readonly metricGetRecordSumBytes: MetricWithAlarmSupport;
@@ -90,6 +100,7 @@ export class KinesisDataStreamMonitoring extends Monitoring {
       namingStrategy.resolveAlarmFriendlyName()
     );
     this.kinesisAlarmFactory = new KinesisAlarmFactory(alarmFactory);
+    this.provisionedCapacityAnnotations = [];
     this.ageAnnotations = [];
     this.recordCountAnnotations = [];
 
@@ -162,6 +173,30 @@ export class KinesisDataStreamMonitoring extends Monitoring {
       this.recordCountAnnotations.push(createdAlarm.annotation);
       this.addAlarm(createdAlarm);
     }
+    for (const disambiguator in props.addReadProvisionedThroughputExceededAlarm) {
+      const alarmProps =
+        props.addReadProvisionedThroughputExceededAlarm[disambiguator];
+      const createdAlarm =
+        this.kinesisAlarmFactory.addProvisionedReadThroughputExceededAlarm(
+          this.readProvisionedThroughputExceededMetric,
+          alarmProps,
+          disambiguator
+        );
+      this.provisionedCapacityAnnotations.push(createdAlarm.annotation);
+      this.addAlarm(createdAlarm);
+    }
+    for (const disambiguator in props.addWriteProvisionedThroughputExceededAlarm) {
+      const alarmProps =
+        props.addWriteProvisionedThroughputExceededAlarm[disambiguator];
+      const createdAlarm =
+        this.kinesisAlarmFactory.addProvisionedWriteThroughputExceededAlarm(
+          this.writeProvisionedThroughputExceededMetric,
+          alarmProps,
+          disambiguator
+        );
+      this.provisionedCapacityAnnotations.push(createdAlarm.annotation);
+      this.addAlarm(createdAlarm);
+    }
 
     props.useCreatedAlarms?.consume(this.createdAlarms());
   }
@@ -173,7 +208,7 @@ export class KinesisDataStreamMonitoring extends Monitoring {
         this.createIncomingDataWidget(QuarterWidth, DefaultSummaryWidgetHeight),
         this.createIteratorAgeWidget(QuarterWidth, DefaultSummaryWidgetHeight),
         this.createLatencyWidget(QuarterWidth, DefaultSummaryWidgetHeight),
-        this.createThrottleWidget(QuarterWidth, DefaultSummaryWidgetHeight)
+        this.createCapacityWidget(QuarterWidth, DefaultSummaryWidgetHeight)
       ),
     ];
   }
@@ -185,7 +220,7 @@ export class KinesisDataStreamMonitoring extends Monitoring {
         this.createIncomingDataWidget(QuarterWidth, DefaultGraphWidgetHeight),
         this.createIteratorAgeWidget(QuarterWidth, DefaultGraphWidgetHeight),
         this.createLatencyWidget(QuarterWidth, DefaultGraphWidgetHeight),
-        this.createThrottleWidget(QuarterWidth, DefaultGraphWidgetHeight)
+        this.createCapacityWidget(QuarterWidth, DefaultGraphWidgetHeight)
       ),
       this.createFirstAdditionalRow(),
       this.createSecondAdditionalRow(),
@@ -248,16 +283,17 @@ export class KinesisDataStreamMonitoring extends Monitoring {
     });
   }
 
-  protected createThrottleWidget(width: number, height: number) {
+  protected createCapacityWidget(width: number, height: number) {
     return new GraphWidget({
       width,
       height,
-      title: "Throttles",
+      title: "Provisioned Capacity Exceeded",
       left: [
         this.readProvisionedThroughputExceededMetric,
         this.writeProvisionedThroughputExceededMetric,
       ],
       leftYAxis: TimeAxisMillisFromZero,
+      leftAnnotations: this.provisionedCapacityAnnotations,
     });
   }
 

lib/monitoring/aws-kinesis/KinesisDataStreamMonitoring.ts

@@ -58,6 +58,17 @@ export class KinesisFirehoseMetricFactory {
     );
   }
 
+  metricIncomingPutRequests() {
+    return this.metricFactory.createMetric(
+      "IncomingPutRequests",
+      MetricStatistic.SUM,
+      "Incoming (PutRequest)",
+      this.dimensionsMap,
+      undefined,
+      FirehoseNamespace
+    );
+  }
+
   metricIncomingRecordCount() {
     return this.metricFactory.createMetric(
       "IncomingRecords",
@@ -101,4 +112,70 @@ export class KinesisFirehoseMetricFactory {
       FirehoseNamespace
     );
   }
+
+  metricIncomingBytesToLimitRate() {
+    return this.metricFactory.createMetricMath(
+      "(bytes_in / PERIOD(bytes_in)) / bytes_max",
+      {
+        bytes_in: this.metricIncomingBytes(),
+        bytes_max: this.metricBytesPerSecondLimit(),
+      },
+      "Incoming Bytes / Limit"
+    );
+  }
+
+  metricIncomingRecordsToLimitRate() {
+    return this.metricFactory.createMetricMath(
+      "(records_in / PERIOD(records_in)) / records_max",
+      {
+        records_in: this.metricIncomingRecordCount(),
+        records_max: this.metricRecordsPerSecondLimit(),
+      },
+      "Incoming Records / Limit"
+    );
+  }
+
+  metricIncomingPutRequestsToLimitRate() {
+    return this.metricFactory.createMetricMath(
+      "(requests_in / PERIOD(requests_in)) / requests_max",
+      {
+        requests_in: this.metricIncomingPutRequests(),
+        requests_max: this.metricPutRequestsPerSecondLimit(),
+      },
+      "Incoming PutRequests / Limit"
+    );
+  }
+
+  metricBytesPerSecondLimit() {
+    return this.metricFactory.createMetric(
+      "BytesPerSecondLimit",
+      MetricStatistic.AVERAGE,
+      "Incoming Bytes/s Limit",
+      undefined,
+      undefined,
+      FirehoseNamespace
+    );
+  }
+
+  metricRecordsPerSecondLimit() {
+    return this.metricFactory.createMetric(
+      "RecordsPerSecondLimit",
+      MetricStatistic.AVERAGE,
+      "Records/s Limit",
+      undefined,
+      undefined,
+      FirehoseNamespace
+    );
+  }
+
+  metricPutRequestsPerSecondLimit() {
+    return this.metricFactory.createMetric(
+      "PutRequestsPerSecondLimit",
+      MetricStatistic.AVERAGE,
+      "PutRequests/s Limit",
+      undefined,
+      undefined,
+      FirehoseNamespace
+    );
+  }
 }

lib/monitoring/aws-kinesis/KinesisFirehoseMetricFactory.ts

@@ -1,15 +1,22 @@
-import { GraphWidget, IWidget } from "aws-cdk-lib/aws-cloudwatch";
+import {
+  GraphWidget,
+  HorizontalAnnotation,
+  IWidget,
+} from "aws-cdk-lib/aws-cloudwatch";
 
 import {
   BaseMonitoringProps,
   CountAxisFromZero,
   DefaultGraphWidgetHeight,
   DefaultSummaryWidgetHeight,
   HalfWidth,
+  KinesisAlarmFactory,
   MetricWithAlarmSupport,
   Monitoring,
   MonitoringScope,
-  ThirdWidth,
+  QuarterWidth,
+  RateAxisFromZero,
+  RecordsThrottledThreshold,
   TimeAxisMillisFromZero,
 } from "../../common";
 import {
@@ -21,7 +28,9 @@ import {
   KinesisFirehoseMetricFactoryProps,
 } from "./KinesisFirehoseMetricFactory";
 
-export interface KinesisFirehoseMonitoringOptions extends BaseMonitoringProps {}
+export interface KinesisFirehoseMonitoringOptions extends BaseMonitoringProps {
+  readonly addRecordsThrottledAlarm?: Record<string, RecordsThrottledThreshold>;
+}
 
 export interface KinesisFirehoseMonitoringProps
   extends KinesisFirehoseMetricFactoryProps,
@@ -31,13 +40,19 @@ export class KinesisFirehoseMonitoring extends Monitoring {
   protected readonly title: string;
   protected readonly streamUrl?: string;
 
+  protected readonly kinesisAlarmFactory: KinesisAlarmFactory;
+  protected readonly recordCountAnnotations: HorizontalAnnotation[];
+
   protected readonly incomingBytesMetric: MetricWithAlarmSupport;
   protected readonly incomingRecordsMetric: MetricWithAlarmSupport;
   protected readonly throttledRecordsMetric: MetricWithAlarmSupport;
   protected readonly successfulConversionMetric: MetricWithAlarmSupport;
   protected readonly failedConversionMetric: MetricWithAlarmSupport;
   protected readonly putRecordLatency: MetricWithAlarmSupport;
   protected readonly putRecordBatchLatency: MetricWithAlarmSupport;
+  protected readonly incomingBytesToLimitRate: MetricWithAlarmSupport;
+  protected readonly incomingRecordsToLimitRate: MetricWithAlarmSupport;
+  protected readonly incomingPutRequestsToLimitRate: MetricWithAlarmSupport;
 
   constructor(scope: MonitoringScope, props: KinesisFirehoseMonitoringProps) {
     super(scope);
@@ -55,6 +70,12 @@ export class KinesisFirehoseMonitoring extends Monitoring {
       scope.createMetricFactory(),
       props
     );
+    const alarmFactory = this.createAlarmFactory(
+      namingStrategy.resolveAlarmFriendlyName()
+    );
+    this.kinesisAlarmFactory = new KinesisAlarmFactory(alarmFactory);
+    this.recordCountAnnotations = [];
+
     this.incomingBytesMetric = metricFactory.metricIncomingBytes();
     this.incomingRecordsMetric = metricFactory.metricIncomingRecordCount();
     this.throttledRecordsMetric = metricFactory.metricThrottledRecordCount();
@@ -64,6 +85,25 @@ export class KinesisFirehoseMonitoring extends Monitoring {
     this.putRecordLatency = metricFactory.metricPutRecordLatencyP90InMillis();
     this.putRecordBatchLatency =
       metricFactory.metricPutRecordBatchLatencyP90InMillis();
+    this.incomingBytesToLimitRate =
+      metricFactory.metricIncomingBytesToLimitRate();
+    this.incomingRecordsToLimitRate =
+      metricFactory.metricIncomingRecordsToLimitRate();
+    this.incomingPutRequestsToLimitRate =
+      metricFactory.metricIncomingPutRequestsToLimitRate();
+
+    for (const disambiguator in props.addRecordsThrottledAlarm) {
+      const alarmProps = props.addRecordsThrottledAlarm[disambiguator];
+      const createdAlarm = this.kinesisAlarmFactory.addPutRecordsThrottledAlarm(
+        this.throttledRecordsMetric,
+        alarmProps,
+        disambiguator
+      );
+      this.recordCountAnnotations.push(createdAlarm.annotation);
+      this.addAlarm(createdAlarm);
+    }
+
+    props.useCreatedAlarms?.consume(this.createdAlarms());
   }
 
   summaryWidgets(): IWidget[] {
@@ -77,9 +117,10 @@ export class KinesisFirehoseMonitoring extends Monitoring {
   widgets(): IWidget[] {
     return [
       this.createTitleWidget(),
-      this.createIncomingRecordWidget(ThirdWidth, DefaultGraphWidgetHeight),
-      this.createLatencyWidget(ThirdWidth, DefaultGraphWidgetHeight),
-      this.createConversionWidget(ThirdWidth, DefaultGraphWidgetHeight),
+      this.createIncomingRecordWidget(QuarterWidth, DefaultGraphWidgetHeight),
+      this.createLatencyWidget(QuarterWidth, DefaultGraphWidgetHeight),
+      this.createConversionWidget(QuarterWidth, DefaultGraphWidgetHeight),
+      this.createLimitWidget(QuarterWidth, DefaultGraphWidgetHeight),
     ];
   }
 
@@ -98,6 +139,7 @@ export class KinesisFirehoseMonitoring extends Monitoring {
       title: "Records",
       left: [this.incomingRecordsMetric, this.throttledRecordsMetric],
       leftYAxis: CountAxisFromZero,
+      leftAnnotations: this.recordCountAnnotations,
     });
   }
 
@@ -120,4 +162,19 @@ export class KinesisFirehoseMonitoring extends Monitoring {
       leftYAxis: CountAxisFromZero,
     });
   }
+
+  protected createLimitWidget(width: number, height: number) {
+    return new GraphWidget({
+      width,
+      height,
+      title: "Limits (rate)",
+      left: [
+        this.incomingBytesToLimitRate.with({ label: "Bytes" }),
+        this.incomingRecordsToLimitRate.with({ label: "Records" }),
+        this.incomingPutRequestsToLimitRate.with({ label: "PutRequests" }),
+      ],
+      leftYAxis: RateAxisFromZero,
+      leftAnnotations: [{ value: 1, label: "100% usage" }],
+    });
+  }
 }