KMS-661: Add Monitoring and Metrics for event publishing (KMS)

bin/deploy-bamboo.sh

@@ -64,6 +64,7 @@ dockerRun() {
         --env "EDL_PASSWORD=$bamboo_EDL_PASSWORD" \
         --env "CMR_BASE_URL=$bamboo_CMR_BASE_URL" \
         --env "BLOCK_PUBLISH_ON_KEYWORD_DIFF_FAILURE=${bamboo_BLOCK_PUBLISH_ON_KEYWORD_DIFF_FAILURE:-false}" \
+        --env "KEYWORD_SYNC_ALARM_EMAILS=${bamboo_KEYWORD_SYNC_ALARM_EMAILS:-}" \
         --env "CORS_ORIGIN=$bamboo_CORS_ORIGIN" \
         --env "RDF4J_INSTANCE_TYPE=$bamboo_RDF4J_INSTANCE_TYPE" \
         --env "RDF4J_CONTAINER_MEMORY_LIMIT=$bamboo_RDF4J_CONTAINER_MEMORY_LIMIT" \

bin/localstack/start.sh

@@ -6,7 +6,9 @@ SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
 # shellcheck source=bin/env/local_env.sh
 source "${SCRIPT_DIR}/../env/local_env.sh"
 
-REQUIRED_SERVICES="sns,sqs,events"
+# CloudWatch is enabled here so keyword sync metrics can be verified locally
+# alongside the existing SNS/SQS/EventBridge workflow.
+REQUIRED_SERVICES="sns,sqs,events,cloudwatch"
 
 if ! docker network inspect "${KMS_DOCKER_NETWORK}" >/dev/null 2>&1; then
   docker network create "${KMS_DOCKER_NETWORK}" >/dev/null
@@ -24,7 +26,8 @@ if [[ -n "${existing_id}" ]]; then
       || true
   )"
 
-  if [[ ",${configured_services}," != *",events,"* ]]; then
+  if [[ ",${configured_services}," != *",events,"* ]] \
+    || [[ ",${configured_services}," != *",cloudwatch,"* ]]; then
     docker rm -f "${LOCALSTACK_CONTAINER_NAME}" >/dev/null
     echo "Recreating LocalStack container '${LOCALSTACK_CONTAINER_NAME}' to enable services: ${REQUIRED_SERVICES}"
     existing_id=""
@@ -54,5 +57,5 @@ docker run -d \
   "${LOCALSTACK_IMAGE}" >/dev/null
 
 echo "Started LocalStack container '${LOCALSTACK_CONTAINER_NAME}' on ${LOCALSTACK_PORT}->4566"
-echo "SNS/SQS/EventBridge endpoint for SAM/Lambda containers: ${AWS_ENDPOINT_URL}"
-echo "SNS/SQS/EventBridge endpoint from host: http://localhost:${LOCALSTACK_PORT}"
+echo "SNS/SQS/EventBridge/CloudWatch endpoint for SAM/Lambda containers: ${AWS_ENDPOINT_URL}"
+echo "SNS/SQS/EventBridge/CloudWatch endpoint from host: http://localhost:${LOCALSTACK_PORT}"

cdk/app/lib/KmsStack.ts

@@ -8,6 +8,7 @@ import { Construct } from 'constructs'
 
 import { ApiResources } from './helper/ApiResources'
 import { IamSetup } from './helper/IamSetup'
+import { KeywordSyncMonitoringSetup } from './helper/KeywordSyncMonitoringSetup'
 import { LambdaFunctions } from './helper/KmsLambdaFunctions'
 import { LogForwardingSetup } from './helper/LogForwardingSetup'
 import { VpcSetup } from './helper/VpcSetup'
@@ -18,6 +19,7 @@ import { VpcSetup } from './helper/VpcSetup'
  */
 export interface KmsStackProps extends cdk.StackProps {
   existingApiId: string | undefined
+  keywordSyncAlarmEmails?: string[]
   prefix: string
   rootResourceId: string | undefined
   stage: string
@@ -83,6 +85,7 @@ export class KmsStack extends cdk.Stack {
     const {
       environment,
       existingApiId,
+      keywordSyncAlarmEmails,
       logDestinationArn,
       prefix,
       rootResourceId,
@@ -147,6 +150,7 @@ export class KmsStack extends cdk.Stack {
     apiResources.configureCors(this, prefix)
     // Use a concrete ARN string during local synth so SAM can inject it into the Lambda env.
     const keywordEventsTopicArn = useLocalstack ? localTopicArn : this.keywordEventsTopic.topicArn
+    // SNS keyword event publishing needs the topic ARN at runtime.
     const lambdaEnvironment = {
       ...environment,
       KEYWORD_EVENTS_TOPIC_ARN: keywordEventsTopicArn
@@ -165,6 +169,12 @@ export class KmsStack extends cdk.Stack {
       useLocalstack
     })
 
+    new KeywordSyncMonitoringSetup(this, {
+      prefix,
+      stage: this.stage,
+      notificationEmails: keywordSyncAlarmEmails || []
+    })
+
     const lambdas = this.lambdaFunctions.getAllLambdas()
     const methods = this.lambdaFunctions.getAllMethods()
 

cdk/app/lib/helper/IamSetup.ts

@@ -115,6 +115,9 @@ export class IamSetup {
 
   /**
    * Adds base KMS Lambda policy to the Lambda role.
+   *
+   * These permissions cover shared runtime needs across the application,
+   * including CloudWatch custom metric emission for keyword sync monitoring.
    * @private
    */
   private addKMSLambdaBasePolicy() {
@@ -123,6 +126,11 @@ export class IamSetup {
       resources: ['*']
     }))
 
+    this.lambdaRole.addToPolicy(new iam.PolicyStatement({
+      actions: ['cloudwatch:PutMetricData'],
+      resources: ['*']
+    }))
+
     this.lambdaRole.addToPolicy(new iam.PolicyStatement({
       actions: ['lambda:InvokeFunction'],
       resources: ['*']

cdk/app/lib/helper/KeywordSyncMonitoringSetup.ts

@@ -0,0 +1,65 @@
+import * as cdk from 'aws-cdk-lib'
+import * as cloudwatch from 'aws-cdk-lib/aws-cloudwatch'
+import * as cloudwatchActions from 'aws-cdk-lib/aws-cloudwatch-actions'
+import * as sns from 'aws-cdk-lib/aws-sns'
+import * as subscriptions from 'aws-cdk-lib/aws-sns-subscriptions'
+import { Construct } from 'constructs'
+
+const KEYWORD_SYNC_METRIC_NAMESPACE = 'KMS/KeywordSync'
+const KEYWORD_EVENT_PUBLISH_FAILURES_METRIC = 'KeywordEventPublishFailures'
+const PUBLISH_FAILURE_ALARM_PERIOD = cdk.Duration.days(1)
+
+export interface KeywordSyncMonitoringSetupProps {
+  prefix: string
+  stage: string
+  notificationEmails?: string[]
+}
+
+/**
+ * Sets up CloudWatch monitoring resources for keyword sync publishing.
+ */
+export class KeywordSyncMonitoringSetup {
+  public readonly publishFailuresAlarm: cloudwatch.Alarm
+
+  /**
+   * Creates the keyword sync monitoring resources.
+   *
+   * @param {Construct} scope - Construct scope for the monitoring resources.
+   * @param {KeywordSyncMonitoringSetupProps} props - Monitoring configuration.
+   */
+  constructor(scope: Construct, props: KeywordSyncMonitoringSetupProps) {
+    const { notificationEmails = [], prefix, stage } = props
+
+    const keywordEventPublishFailuresMetric = new cloudwatch.Metric({
+      namespace: KEYWORD_SYNC_METRIC_NAMESPACE,
+      metricName: KEYWORD_EVENT_PUBLISH_FAILURES_METRIC,
+      statistic: 'Sum',
+      period: PUBLISH_FAILURE_ALARM_PERIOD
+    })
+
+    this.publishFailuresAlarm = new cloudwatch.Alarm(scope, 'KeywordSyncPublishFailuresAlarm', {
+      alarmName: `${prefix}-${stage}-keyword-event-publish-failures`,
+      alarmDescription: 'Alerts when publisher keyword event publishes fail after retries.',
+      metric: keywordEventPublishFailuresMetric,
+      threshold: 1,
+      evaluationPeriods: 1,
+      datapointsToAlarm: 1,
+      comparisonOperator: cloudwatch.ComparisonOperator.GREATER_THAN_OR_EQUAL_TO_THRESHOLD,
+      treatMissingData: cloudwatch.TreatMissingData.NOT_BREACHING
+    })
+
+    if (notificationEmails.length === 0) {
+      return
+    }
+
+    const alarmTopic = new sns.Topic(scope, 'KeywordSyncAlarmTopic', {
+      topicName: `${prefix}-${stage}-keyword-sync-alerts`
+    })
+
+    notificationEmails.forEach((email) => {
+      alarmTopic.addSubscription(new subscriptions.EmailSubscription(email))
+    })
+
+    this.publishFailuresAlarm.addAlarmAction(new cloudwatchActions.SnsAction(alarmTopic))
+  }
+}

cdk/bin/main.ts

@@ -44,6 +44,10 @@ async function main() {
   const existingApiId = process.env.EXISTING_API_ID
   const rootResourceId = process.env.ROOT_RESOURCE_ID
   let logDestinationArn = process.env.LOG_DESTINATION_ARN
+  const keywordSyncAlarmEmails = (process.env.KEYWORD_SYNC_ALARM_EMAILS || '')
+    .split(',')
+    .map((email) => email.trim())
+    .filter(Boolean)
 
   const app = new cdk.App({
     context: {
@@ -165,6 +169,7 @@ async function main() {
     stage,
     existingApiId,
     rootResourceId,
+    keywordSyncAlarmEmails,
     logDestinationArn: logDestinationArn!,
     environment: {
       RDF4J_SERVICE_URL: useLocalstack