feat(kinesis): add ability to create alarms for firehose streams approaching traffic limit quota (#382)

## Note
- adding the following alarm for firehose streams
 - IncomingBytesExceedThresholdAlarm
 - IncomingRecordsExceedThresholdAlarm
 - IncomingPutRequestsExceedThresholdAlarm
- updated `metricIncomingRecordCount` name from `"Incoming"` to `"Incoming (Records)"` to be consistent with `metricIncomingPutRequests` and `metricIncomingBytes`

## Local testing executed
Successfully executed the following command
```
yarn test
yarn build
yarn package-all
```

---

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

Author: jiashuChen

API.md

@@ -14,6 +14,13 @@ export interface RecordsThrottledThreshold extends CustomAlarmThreshold {
   readonly maxRecordsThrottledThreshold: number;
 }
 
+export interface FirehoseStreamLimitThreshold extends CustomAlarmThreshold {
+  /**
+   * Threshold value between [0.0, 1.0) for when the alarm should be triggered.
+   */
+  readonly safetyThresholdLimit: number;
+}
+
 export interface RecordsFailedThreshold extends CustomAlarmThreshold {
   readonly maxRecordsFailedThreshold: number;
 }
@@ -90,6 +97,36 @@ export class KinesisAlarmFactory {
     });
   }
 
+  addFirehoseStreamExceedSafetyThresholdAlarm(
+    metric: MetricWithAlarmSupport,
+    metricName: string,
+    quotaName: string,
+    props: FirehoseStreamLimitThreshold,
+    disambiguator?: string
+  ) {
+    const threshold = props.safetyThresholdLimit;
+    if (threshold < 0 || threshold >= 1) {
+      throw new Error(
+        `safetyThresholdLimit must be in range [0.0, 1.0) for ${metricName}ExceedThresholdAlarm.`
+      );
+    }
+
+    return this.alarmFactory.addAlarm(metric, {
+      treatMissingData:
+        props.treatMissingDataOverride ?? TreatMissingData.NOT_BREACHING,
+      comparisonOperator:
+        props.comparisonOperatorOverride ??
+        ComparisonOperator.GREATER_THAN_THRESHOLD,
+      ...props,
+      disambiguator,
+      threshold,
+      alarmNameSuffix: metricName,
+      alarmDescription: `${metricName} exceeded ${quotaName} alarming threshold of ${threshold}`,
+      // we will dedupe any kind of message count issue to the same ticket
+      alarmDedupeStringSuffix: `${metricName}ExceedThresholdLimit`,
+    });
+  }
+
   addProvisionedReadThroughputExceededAlarm(
     metric: MetricWithAlarmSupport,
     props: RecordsThrottledThreshold,

lib/common/monitoring/alarms/KinesisAlarmFactory.ts

@@ -73,7 +73,7 @@ export class KinesisFirehoseMetricFactory {
     return this.metricFactory.createMetric(
       "IncomingRecords",
       MetricStatistic.SUM,
-      "Incoming",
+      "Incoming (Records)",
       this.dimensionsMap,
       undefined,
       FirehoseNamespace

lib/monitoring/aws-kinesis/KinesisFirehoseMetricFactory.ts

@@ -21,6 +21,7 @@ import {
   QuarterWidth,
   RateAxisFromZero,
   RecordsThrottledThreshold,
+  FirehoseStreamLimitThreshold,
   TimeAxisMillisFromZero,
 } from "../../common";
 import {
@@ -30,6 +31,18 @@ import {
 
 export interface KinesisFirehoseMonitoringOptions extends BaseMonitoringProps {
   readonly addRecordsThrottledAlarm?: Record<string, RecordsThrottledThreshold>;
+  readonly addIncomingBytesExceedThresholdAlarm?: Record<
+    string,
+    FirehoseStreamLimitThreshold
+  >;
+  readonly addIncomingRecordsExceedThresholdAlarm?: Record<
+    string,
+    FirehoseStreamLimitThreshold
+  >;
+  readonly addIncomingPutRequestsExceedThresholdAlarm?: Record<
+    string,
+    FirehoseStreamLimitThreshold
+  >;
 }
 
 export interface KinesisFirehoseMonitoringProps
@@ -42,6 +55,7 @@ export class KinesisFirehoseMonitoring extends Monitoring {
 
   readonly kinesisAlarmFactory: KinesisAlarmFactory;
   readonly recordCountAnnotations: HorizontalAnnotation[];
+  readonly incomingLimitAnnotations: HorizontalAnnotation[];
 
   readonly incomingBytesMetric: MetricWithAlarmSupport;
   readonly incomingRecordsMetric: MetricWithAlarmSupport;
@@ -75,6 +89,7 @@ export class KinesisFirehoseMonitoring extends Monitoring {
     );
     this.kinesisAlarmFactory = new KinesisAlarmFactory(alarmFactory);
     this.recordCountAnnotations = [];
+    this.incomingLimitAnnotations = [{ value: 1, label: "100% usage" }];
 
     this.incomingBytesMetric = metricFactory.metricIncomingBytes();
     this.incomingRecordsMetric = metricFactory.metricIncomingRecordCount();
@@ -103,6 +118,51 @@ export class KinesisFirehoseMonitoring extends Monitoring {
       this.addAlarm(createdAlarm);
     }
 
+    for (const disambiguator in props.addIncomingBytesExceedThresholdAlarm) {
+      const alarmProps =
+        props.addIncomingBytesExceedThresholdAlarm[disambiguator];
+      const createdAlarm =
+        this.kinesisAlarmFactory.addFirehoseStreamExceedSafetyThresholdAlarm(
+          this.incomingBytesToLimitRate,
+          "IncomingBytes",
+          "BytesPerSecondLimit",
+          alarmProps,
+          disambiguator
+        );
+      this.incomingLimitAnnotations.push(createdAlarm.annotation);
+      this.addAlarm(createdAlarm);
+    }
+
+    for (const disambiguator in props.addIncomingRecordsExceedThresholdAlarm) {
+      const alarmProps =
+        props.addIncomingRecordsExceedThresholdAlarm[disambiguator];
+      const createdAlarm =
+        this.kinesisAlarmFactory.addFirehoseStreamExceedSafetyThresholdAlarm(
+          this.incomingRecordsToLimitRate,
+          "IncomingRecords",
+          "RecordsPerSecondLimit",
+          alarmProps,
+          disambiguator
+        );
+      this.incomingLimitAnnotations.push(createdAlarm.annotation);
+      this.addAlarm(createdAlarm);
+    }
+
+    for (const disambiguator in props.addIncomingPutRequestsExceedThresholdAlarm) {
+      const alarmProps =
+        props.addIncomingPutRequestsExceedThresholdAlarm[disambiguator];
+      const createdAlarm =
+        this.kinesisAlarmFactory.addFirehoseStreamExceedSafetyThresholdAlarm(
+          this.incomingPutRequestsToLimitRate,
+          "IncomingPutRequests",
+          "PutRequestsPerSecondLimit",
+          alarmProps,
+          disambiguator
+        );
+      this.incomingLimitAnnotations.push(createdAlarm.annotation);
+      this.addAlarm(createdAlarm);
+    }
+
     props.useCreatedAlarms?.consume(this.createdAlarms());
   }
 
@@ -174,7 +234,7 @@ export class KinesisFirehoseMonitoring extends Monitoring {
         this.incomingPutRequestsToLimitRate.with({ label: "PutRequests" }),
       ],
       leftYAxis: RateAxisFromZero,
-      leftAnnotations: [{ value: 1, label: "100% usage" }],
+      leftAnnotations: this.incomingLimitAnnotations,
     });
   }
 }

lib/monitoring/aws-kinesis/KinesisFirehoseMonitoring.ts

@@ -32,6 +32,21 @@ test("snapshot test: all alarms", () => {
         maxRecordsThrottledThreshold: 5,
       },
     },
+    addIncomingBytesExceedThresholdAlarm: {
+      Critical: {
+        safetyThresholdLimit: 0.6,
+      },
+    },
+    addIncomingRecordsExceedThresholdAlarm: {
+      Critical: {
+        safetyThresholdLimit: 0.7,
+      },
+    },
+    addIncomingPutRequestsExceedThresholdAlarm: {
+      Critical: {
+        safetyThresholdLimit: 0.8,
+      },
+    },
     useCreatedAlarms: {
       consume(alarms) {
         numAlarmsCreated = alarms.length;
@@ -40,6 +55,46 @@ test("snapshot test: all alarms", () => {
   });
 
   addMonitoringDashboardsToStack(stack, monitoring);
-  expect(numAlarmsCreated).toStrictEqual(1);
+  expect(numAlarmsCreated).toStrictEqual(4);
   expect(Template.fromStack(stack)).toMatchSnapshot();
 });
+
+test("test: validation error if incoming traffic usage alarm threshold equal to 1", () => {
+  const stack = new Stack();
+
+  const scope = new TestMonitoringScope(stack, "Scope");
+
+  expect(() => {
+    new KinesisFirehoseMonitoring(scope, {
+      deliveryStreamName:
+        "my-firehose-delivery-stream-with-unexpected-threshold",
+      addIncomingBytesExceedThresholdAlarm: {
+        Critical: {
+          safetyThresholdLimit: 1.0,
+        },
+      },
+    });
+  }).toThrow(
+    `safetyThresholdLimit must be in range [0.0, 1.0) for IncomingBytesExceedThresholdAlarm.`
+  );
+});
+
+test("test: validation error if incoming traffic usage alarm threshold less than 0", () => {
+  const stack = new Stack();
+
+  const scope = new TestMonitoringScope(stack, "Scope");
+
+  expect(() => {
+    new KinesisFirehoseMonitoring(scope, {
+      deliveryStreamName:
+        "my-firehose-delivery-stream-with-unexpected-threshold",
+      addIncomingRecordsExceedThresholdAlarm: {
+        Critical: {
+          safetyThresholdLimit: -0.1,
+        },
+      },
+    });
+  }).toThrow(
+    `safetyThresholdLimit must be in range [0.0, 1.0) for IncomingRecordsExceedThresholdAlarm.`
+  );
+});