feat(kinesisdataanalytics): Fix Checkpoint Failure Rate calculation and add Full Restart Rate monitoring (#668)

### Changes

**Fixed Checkpoint Failure Rate**
- Replaced manual rate calculation with CloudWatch's native RATE()
function
- Changed from (3600 * checkpoints) / PERIOD(checkpoints) to
RATE(numberOfFailedCheckpoints)

**Added Full Restart Rate Monitoring**
- New metricFullRestartRate() method using RATE(fullRestarts)
- New addFullRestartRateAlarm() with custom implementation to avoid
naming conflicts
- Enhanced Full Restarts widget to show both count (left axis) and rate
(right axis)

**Technical**
- Resolved alarm naming conflicts between full restart and checkpoint
failure rate alarms
- Added test coverage for new functionality
- All 5 alarms now supported (was 4)

Provides more accurate rate monitoring and better insights into Kinesis
Data Analytics application performance.

Author: costmatt-amzn

API.md

@@ -70,6 +70,26 @@ export class KinesisDataAnalyticsAlarmFactory {
     });
   }
 
+  addFullRestartRateAlarm(
+    metric: MetricWithAlarmSupport,
+    props: ErrorRateThreshold,
+    disambiguator?: string,
+  ) {
+    return this.alarmFactory.addAlarm(metric, {
+      treatMissingData:
+        props.treatMissingDataOverride ?? TreatMissingData.BREACHING,
+      comparisonOperator:
+        props.comparisonOperatorOverride ??
+        ComparisonOperator.GREATER_THAN_THRESHOLD,
+      ...props,
+      disambiguator,
+      threshold: props.maxErrorRate,
+      alarmNameSuffix: "FullRestartRate",
+      alarmDescription: "Full restart rate is too high",
+      alarmDedupeStringSuffix: "KDAFullRestartRateAlarm",
+    });
+  }
+
   addCheckpointFailureCountAlarm(
     metric: MetricWithAlarmSupport,
     props: ErrorCountThreshold,

lib/common/monitoring/alarms/KinesisDataAnalyticsAlarmFactory.ts

@@ -5,7 +5,6 @@ import {
   BaseMetricFactoryProps,
   MetricFactory,
   MetricStatistic,
-  RateComputationMethod,
 } from "../../common";
 
 export interface KinesisDataAnalyticsMetricFactoryProps
@@ -120,11 +119,34 @@ export class KinesisDataAnalyticsMetricFactory extends BaseMetricFactory<Kinesis
   }
 
   metricCheckpointFailureRate() {
-    return this.metricFactory.toRate(
-      this.metricNumberOfFailedCheckpointsCount(),
-      RateComputationMethod.PER_HOUR,
-      false,
-      "checkpoints",
+    // Flink reports this metric as the latest sum for the lifecycle of a job.
+    // Therefore, we truly care about rate of change
+    return this.metricFactory.createMetricMath(
+      "RATE(numberOfFailedCheckpoints)",
+      {
+        numberOfFailedCheckpoints: this.metricNumberOfFailedCheckpointsCount(),
+      },
+      "Checkpoint Failure Rate",
+      undefined,
+      undefined,
+      this.region,
+      this.account,
+    );
+  }
+
+  metricFullRestartRate() {
+    // Flink reports this metric as the latest sum for the lifecycle of a job.
+    // Therefore, we truly care about rate of change
+    return this.metricFactory.createMetricMath(
+      "RATE(fullRestarts)",
+      {
+        fullRestarts: this.metricFullRestartsCount(),
+      },
+      "Full Restart Rate",
+      undefined,
+      undefined,
+      this.region,
+      this.account,
     );
   }
 

lib/monitoring/aws-kinesisanalytics/KinesisDataAnalyticsMetricFactory.ts

@@ -38,6 +38,8 @@ export interface KinesisDataAnalyticsMonitoringOptions
 
   readonly addFullRestartCountAlarm?: Record<string, FullRestartCountThreshold>;
 
+  readonly addFullRestartRateAlarm?: Record<string, ErrorRateThreshold>;
+
   readonly addCheckpointFailureCountAlarm?: Record<string, ErrorCountThreshold>;
 
   readonly addCheckpointFailureRateAlarm?: Record<string, ErrorRateThreshold>;
@@ -54,6 +56,7 @@ export class KinesisDataAnalyticsMonitoring extends Monitoring {
   readonly kdaAlarmFactory: KinesisDataAnalyticsAlarmFactory;
   readonly downtimeAnnotations: HorizontalAnnotation[];
   readonly fullRestartAnnotations: HorizontalAnnotation[];
+  readonly fullRestartRateAnnotations: HorizontalAnnotation[];
   readonly checkpointFailureCountAnnotations: HorizontalAnnotation[];
   readonly checkpointFailureRateAnnotations: HorizontalAnnotation[];
 
@@ -68,6 +71,7 @@ export class KinesisDataAnalyticsMonitoring extends Monitoring {
   readonly oldGenerationGCCountMetric: MetricWithAlarmSupport;
   readonly oldGenerationGCTimeMsMetric: MetricWithAlarmSupport;
   readonly checkpointFailureRateMetric: MetricWithAlarmSupport;
+  readonly fullRestartRateMetric: MetricWithAlarmSupport;
 
   constructor(
     scope: MonitoringScope,
@@ -90,6 +94,7 @@ export class KinesisDataAnalyticsMonitoring extends Monitoring {
     this.kdaAlarmFactory = new KinesisDataAnalyticsAlarmFactory(alarmFactory);
     this.downtimeAnnotations = [];
     this.fullRestartAnnotations = [];
+    this.fullRestartRateAnnotations = [];
     this.checkpointFailureCountAnnotations = [];
     this.checkpointFailureRateAnnotations = [];
 
@@ -117,6 +122,7 @@ export class KinesisDataAnalyticsMonitoring extends Monitoring {
       metricFactory.metricOldGenerationGCTimeMs();
     this.checkpointFailureRateMetric =
       metricFactory.metricCheckpointFailureRate();
+    this.fullRestartRateMetric = metricFactory.metricFullRestartRate();
 
     for (const disambiguator in props.addDowntimeAlarm) {
       const alarmProps = props.addDowntimeAlarm[disambiguator];
@@ -140,6 +146,17 @@ export class KinesisDataAnalyticsMonitoring extends Monitoring {
       this.addAlarm(createdAlarm);
     }
 
+    for (const disambiguator in props.addFullRestartRateAlarm) {
+      const alarmProps = props.addFullRestartRateAlarm[disambiguator];
+      const createdAlarm = this.kdaAlarmFactory.addFullRestartRateAlarm(
+        this.fullRestartRateMetric,
+        alarmProps,
+        disambiguator,
+      );
+      this.fullRestartRateAnnotations.push(createdAlarm.annotation);
+      this.addAlarm(createdAlarm);
+    }
+
     for (const disambiguator in props.addCheckpointFailureCountAlarm) {
       const alarmProps = props.addCheckpointFailureCountAlarm[disambiguator];
       const createdAlarm = this.kdaAlarmFactory.addCheckpointFailureCountAlarm(
@@ -230,6 +247,9 @@ export class KinesisDataAnalyticsMonitoring extends Monitoring {
       left: [this.fullRestartsCountMetric],
       leftYAxis: CountAxisFromZero,
       leftAnnotations: this.fullRestartAnnotations,
+      right: [this.fullRestartRateMetric],
+      rightYAxis: RateAxisFromZero,
+      rightAnnotations: this.fullRestartRateAnnotations,
     });
   }
 

lib/monitoring/aws-kinesisanalytics/KinesisDataAnalyticsMonitoring.ts

@@ -37,6 +37,11 @@ test("snapshot test: all alarms", () => {
         maxFullRestartCount: 1,
       },
     },
+    addFullRestartRateAlarm: {
+      Warning: {
+        maxErrorRate: 0.1,
+      },
+    },
     addCheckpointFailureCountAlarm: {
       Warning: {
         maxErrorCount: 5,
@@ -55,6 +60,6 @@ test("snapshot test: all alarms", () => {
   });
 
   addMonitoringDashboardsToStack(stack, monitoring);
-  expect(numAlarmsCreated).toStrictEqual(4);
+  expect(numAlarmsCreated).toStrictEqual(5);
   expect(Template.fromStack(stack)).toMatchSnapshot();
 });