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

lib/common/monitoring/alarms/KinesisDataAnalyticsAlarmFactory.ts

@@ -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/monitoring/aws-kinesisanalytics/KinesisDataAnalyticsMetricFactory.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/KinesisDataAnalyticsMonitoring.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,
     });
   }