feat(fmannotator): add DLQ for fmannotator function

config/constants.ts

@@ -128,6 +128,9 @@ export const dataSchemaRegistryName = 'orcabus.data';
 export const eventBusName = 'OrcaBusMain';
 export const eventSourceQueueName = 'orcabus-event-source-queue';
 
+// DLQs for stateless stack functions
+export const eventDlqNameFMAnnotator = 'orcabus-event-dlq-fmannotator';
+
 /**
  * Configuration for resources created in TokenServiceStack
  */

config/stacks/fmAnnotator.ts

@@ -1,10 +1,11 @@
 import { FMAnnotatorConfigurableProps } from '../../lib/workload/stateless/stacks/fmannotator/deploy/stack';
-import { eventBusName, jwtSecretName, vpcProps } from '../constants';
+import { eventBusName, eventDlqNameFMAnnotator, jwtSecretName, vpcProps } from '../constants';
 
 export const getFmAnnotatorProps = (): FMAnnotatorConfigurableProps => {
   return {
     vpcProps,
     eventBusName,
     jwtSecretName,
+    eventDLQName: eventDlqNameFMAnnotator,
   };
 };

config/stacks/shared.ts

@@ -11,6 +11,7 @@ import {
   dbClusterIdentifier,
   dbClusterResourceIdParameterName,
   eventBusName,
+  eventDlqNameFMAnnotator,
   eventSchemaRegistryName,
   eventSourceQueueName,
   icav2ArchiveAnalysisBucket,
@@ -27,6 +28,7 @@ import {
 } from '../../lib/workload/stateful/stacks/shared/constructs/event-bus';
 import { ComputeProps } from '../../lib/workload/stateful/stacks/shared/constructs/compute';
 import { EventSourceProps } from '../../lib/workload/stateful/stacks/shared/constructs/event-source';
+import { EventDLQProps } from '../../lib/workload/stateful/stacks/shared/constructs/event-dlq';
 
 const getEventSchemaRegistryConstructProps = (): SchemaRegistryProps => {
   return {
@@ -115,6 +117,15 @@ const getEventSourceConstructProps = (stage: AppStage): EventSourceProps => {
   return props;
 };
 
+const getEventDLQConstructProps = (): EventDLQProps[] => {
+  return [
+    {
+      queueName: eventDlqNameFMAnnotator,
+      alarmName: 'Orcabus FMAnnotator DLQ Alarm',
+    },
+  ];
+};
+
 const getDatabaseConstructProps = (stage: AppStage): ConfigurableDatabaseProps => {
   const baseConfig = {
     clusterIdentifier: dbClusterIdentifier,
@@ -170,5 +181,6 @@ export const getSharedStackProps = (stage: AppStage): SharedStackProps => {
     databaseProps: getDatabaseConstructProps(stage),
     computeProps: getComputeConstructProps(),
     eventSourceProps: getEventSourceConstructProps(stage),
+    eventDLQProps: getEventDLQConstructProps(),
   };
 };

lib/workload/stateful/stacks/shared/constructs/event-dlq/index.ts

@@ -0,0 +1,65 @@
+import { Construct } from 'constructs';
+import { IQueue, Queue } from 'aws-cdk-lib/aws-sqs';
+import { Alarm, ComparisonOperator, MathExpression } from 'aws-cdk-lib/aws-cloudwatch';
+
+/**
+ * Properties for the EventDLQConstruct.
+ */
+export type EventDLQProps = {
+  /**
+   * The name of the dead letter queue for the construct.
+   */
+  queueName: string;
+  /**
+   * Specify the name of the alarm.
+   */
+  alarmName: string;
+};
+
+/**
+ * A wrapper around an SQS queue that should act as a dead-letter queue.
+ * Note that this is intentionally not a Construct so that the queue is flattened
+ * within the parent construct.
+ */
+export class EventDLQConstruct {
+  readonly deadLetterQueue: Queue;
+  readonly alarm: Alarm;
+
+  constructor(scope: Construct, queueId: string, alarmId: string, props: EventDLQProps) {
+    this.deadLetterQueue = new Queue(scope, queueId, {
+      queueName: `${props.queueName}`,
+      enforceSSL: true,
+    });
+
+    const rateOfMessages = new MathExpression({
+      expression: 'RATE(visible + notVisible)',
+      usingMetrics: {
+        visible: this.deadLetterQueue.metricApproximateNumberOfMessagesVisible(),
+        notVisible: this.deadLetterQueue.metricApproximateNumberOfMessagesVisible(),
+      },
+    });
+
+    this.alarm = new Alarm(scope, alarmId, {
+      metric: rateOfMessages,
+      comparisonOperator: ComparisonOperator.GREATER_THAN_THRESHOLD,
+      threshold: 0,
+      evaluationPeriods: 1,
+      alarmName: props.alarmName,
+      alarmDescription: 'An event has been received in the dead letter queue.',
+    });
+  }
+
+  /**
+   * Get the SQS queue ARN.
+   */
+  get queueArn(): string {
+    return this.queue.queueArn;
+  }
+
+  /**
+   * Get the dead letter queue.
+   */
+  get queue(): IQueue {
+    return this.deadLetterQueue;
+  }
+}

lib/workload/stateful/stacks/shared/constructs/event-source/index.ts

@@ -2,8 +2,8 @@ import { Construct } from 'constructs';
 import { Rule } from 'aws-cdk-lib/aws-events';
 import { Queue } from 'aws-cdk-lib/aws-sqs';
 import { SqsQueue } from 'aws-cdk-lib/aws-events-targets';
-import { Alarm, ComparisonOperator, MathExpression } from 'aws-cdk-lib/aws-cloudwatch';
 import { ServicePrincipal } from 'aws-cdk-lib/aws-iam';
+import { EventDLQConstruct } from '../event-dlq';
 
 /**
  * Properties for defining an S3 EventBridge rule.
@@ -53,22 +53,21 @@ export type EventSourceProps = {
  */
 export class EventSourceConstruct extends Construct {
   readonly queue: Queue;
-  readonly deadLetterQueue: Queue;
-  readonly alarm: Alarm;
+  readonly deadLetterQueue: EventDLQConstruct;
 
   constructor(scope: Construct, id: string, props: EventSourceProps) {
     super(scope, id);
 
-    this.deadLetterQueue = new Queue(this, 'DeadLetterQueue', {
+    this.deadLetterQueue = new EventDLQConstruct(this, 'DeadLetterQueue', 'Alarm', {
       queueName: `${props.queueName}-dlq`,
-      enforceSSL: true,
+      alarmName: 'Orcabus EventSource Alarm',
     });
     this.queue = new Queue(this, 'Queue', {
       queueName: props.queueName,
       enforceSSL: true,
       deadLetterQueue: {
         maxReceiveCount: props.maxReceiveCount,
-        queue: this.deadLetterQueue,
+        queue: this.deadLetterQueue.queue,
       },
     });
 
@@ -98,23 +97,6 @@ export class EventSourceConstruct extends Construct {
     }
 
     this.queue.grantSendMessages(new ServicePrincipal('events.amazonaws.com'));
-
-    const rateOfMessages = new MathExpression({
-      expression: 'RATE(visible + notVisible)',
-      usingMetrics: {
-        visible: this.deadLetterQueue.metricApproximateNumberOfMessagesVisible(),
-        notVisible: this.deadLetterQueue.metricApproximateNumberOfMessagesVisible(),
-      },
-    });
-
-    this.alarm = new Alarm(this, 'Alarm', {
-      metric: rateOfMessages,
-      comparisonOperator: ComparisonOperator.GREATER_THAN_THRESHOLD,
-      threshold: 0,
-      evaluationPeriods: 1,
-      alarmName: 'Orcabus EventSource Alarm',
-      alarmDescription: 'An event has been received in the dead letter queue.',
-    });
   }
 
   /**

lib/workload/stateful/stacks/shared/stack.ts

@@ -6,6 +6,7 @@ import { ConfigurableDatabaseProps, DatabaseConstruct } from './constructs/datab
 import { ComputeProps, ComputeConstruct } from './constructs/compute';
 import { SchemaRegistryConstruct, SchemaRegistryProps } from './constructs/schema-registry';
 import { EventSourceConstruct, EventSourceProps } from './constructs/event-source';
+import { EventDLQConstruct, EventDLQProps } from './constructs/event-dlq';
 
 export interface SharedStackProps {
   /**
@@ -29,6 +30,10 @@ export interface SharedStackProps {
    * Any configuration related to event source
    */
   eventSourceProps?: EventSourceProps;
+  /**
+   * Any configuration related to event DLQs
+   */
+  eventDLQProps?: EventDLQProps[];
   /**
    * VPC (lookup props) that will be used by resources
    */
@@ -66,5 +71,13 @@ export class SharedStack extends Stack {
     if (props.eventSourceProps) {
       new EventSourceConstruct(this, 'EventSourceConstruct', props.eventSourceProps);
     }
+
+    for (const prop of props.eventDLQProps ?? []) {
+      // Convert kebab-case to PascalCase.
+      const name = prop.queueName
+        .toLowerCase()
+        .replace(/(^.)|(-[a-z])/g, (group) => group.toUpperCase().replace('-', ''));
+      new EventDLQConstruct(this, `${name}`, `${name}Alarm`, prop);
+    }
   }
 }

lib/workload/stateless/stacks/fmannotator/deploy/stack.ts

@@ -1,4 +1,4 @@
-import { aws_events_targets as targets, Duration, Stack, StackProps } from 'aws-cdk-lib';
+import { Arn, aws_events_targets as targets, Duration, Stack, StackProps } from 'aws-cdk-lib';
 import {
   ISecurityGroup,
   IVpc,
@@ -15,13 +15,15 @@ import { EventBus, IEventBus, Rule } from 'aws-cdk-lib/aws-events';
 import { Secret } from 'aws-cdk-lib/aws-secretsmanager';
 import { NamedLambdaRole } from '../../../../components/named-lambda-role';
 import { ManagedPolicy, PolicyStatement, Role } from 'aws-cdk-lib/aws-iam';
+import { IQueue, Queue } from 'aws-cdk-lib/aws-sqs';
 
 /**
  * Config for the FM annotator.
  */
 export type FMAnnotatorConfig = {
   vpcProps: VpcLookupOptions;
   eventBusName: string;
+  eventDLQName: string;
   jwtSecretName: string;
 };
 
@@ -45,6 +47,7 @@ export class FMAnnotator extends Stack {
   private readonly securityGroup: ISecurityGroup;
   private readonly eventBus: IEventBus;
   private readonly role: Role;
+  private readonly dlq: IQueue;
 
   constructor(scope: Construct, id: string, props: FMAnnotatorProps) {
     super(scope, id, props);
@@ -65,6 +68,18 @@ export class FMAnnotator extends Stack {
     // Need access to secrets to fetch FM JWT token.
     tokenSecret.grantRead(this.role);
 
+    this.dlq = Queue.fromQueueArn(
+      this,
+      'FilemanagerQueue',
+      Arn.format(
+        {
+          resource: props.eventDLQName,
+          service: 'sqs',
+        },
+        this
+      )
+    );
+
     const entry = path.join(__dirname, '..', 'cmd', 'portalrunid');
     const fn = new GoFunction(this, 'handler', {
       entry,
@@ -80,6 +95,8 @@ export class FMAnnotator extends Stack {
       vpc: this.vpc,
       vpcSubnets: { subnetType: SubnetType.PRIVATE_WITH_EGRESS },
       securityGroups: [this.securityGroup],
+      deadLetterQueue: this.dlq,
+      deadLetterQueueEnabled: true,
     });
 
     const eventRule = new Rule(this, 'EventRule', {

test/stateful/pipeline/deployment.test.ts

@@ -135,6 +135,19 @@ function applyNagSuppression(stackId: string, stack: Stack) {
           },
         ]
       );
+      NagSuppressions.addResourceSuppressionsByPath(
+        stack,
+        '/SharedStack/OrcabusEventDlqFmannotator/Resource',
+        [
+          {
+            id: 'AwsSolutions-SQS3',
+            reason:
+              'it is expected that the DLQ construct has a Queue without a DLQ, because that ' +
+              'queue itself acts as the DLQ for other constructs.',
+          },
+        ],
+        true
+      );
       break;
 
     case 'PostgresManagerStack':

test/stateful/shared/eventDLQConstruct.test.ts

@@ -0,0 +1,35 @@
+import * as cdk from 'aws-cdk-lib';
+import { Template } from 'aws-cdk-lib/assertions';
+import { EventDLQConstruct } from '../../../lib/workload/stateful/stacks/shared/constructs/event-dlq';
+
+let stack: cdk.Stack;
+
+function assert_common(template: Template) {
+  template.resourceCountIs('AWS::SQS::Queue', 1);
+
+  template.hasResourceProperties('AWS::SQS::Queue', {
+    QueueName: 'queue',
+  });
+
+  template.hasResourceProperties('AWS::CloudWatch::Alarm', {
+    ComparisonOperator: 'GreaterThanThreshold',
+    EvaluationPeriods: 1,
+    Threshold: 0,
+  });
+}
+
+beforeEach(() => {
+  stack = new cdk.Stack();
+});
+
+test('Test EventSourceConstruct created props', () => {
+  new EventDLQConstruct(stack, 'TestEventDLQConstruct', 'TestEventDLQAlarm', {
+    queueName: 'queue',
+    alarmName: 'alarm',
+  });
+  const template = Template.fromStack(stack);
+
+  console.log(JSON.stringify(template, undefined, 2));
+
+  assert_common(template);
+});