🔔 Implement Alert change streams + add SNS destination for webhooks

infra/lib/alerting.ts

@@ -17,10 +17,14 @@ interface MatanoAlertingProps {
 }
 
 export class MatanoAlerting extends Construct {
+  ruleMatchesTopic: sns.Topic;
   alertingTopic: sns.Topic;
   constructor(scope: Construct, id: string, props: MatanoAlertingProps) {
     super(scope, id);
 
+    this.ruleMatchesTopic = new sns.Topic(this, "RuleMatchesTopic", {
+      displayName: "MatanoRuleMatchesTopic",
+    });
     this.alertingTopic = new sns.Topic(this, "Topic", {
       displayName: "MatanoAlertingTopic",
     });
@@ -29,14 +33,16 @@ export class MatanoAlerting extends Construct {
       const alertForwarder = new AlertForwarder(this, "Forwarder", {
         integrationsStore: props.integrationsStore,
         alertTrackerTable: props.alertTrackerTable,
-        alertsSnsTopic: this.alertingTopic,
+        ruleMatchesSnsTopic: this.ruleMatchesTopic,
+        alertingSnsTopic: this.alertingTopic,
       });
     }
   }
 }
 
 interface AlertForwarderProps {
-  alertsSnsTopic: sns.Topic;
+  ruleMatchesSnsTopic: sns.Topic;
+  alertingSnsTopic: sns.Topic;
   integrationsStore: IntegrationsStore;
   alertTrackerTable: ddb.Table;
 }
@@ -64,6 +70,7 @@ export class AlertForwarder extends Construct {
       environment: {
         RUST_LOG: "warn,alert_forwarder=info",
         ALERT_TRACKER_TABLE_NAME: props.alertTrackerTable.tableName,
+        ALERTING_SNS_TOPIC_ARN: props.alertingSnsTopic.topicArn,
         DESTINATION_TO_CONFIGURATION_MAP: JSON.stringify(props.integrationsStore?.integrationsInfoMap ?? {}),
         DESTINATION_TO_SECRET_ARN_MAP: JSON.stringify(destinationToSecretArnMap),
       },
@@ -90,11 +97,12 @@ export class AlertForwarder extends Construct {
       visibilityTimeout: cdk.Duration.seconds(Math.max(this.function.timeout!.toSeconds(), 30)),
     });
 
-    props.alertsSnsTopic.addSubscription(
+    props.ruleMatchesSnsTopic.addSubscription(
       new SqsSubscription(this.queue, {
         rawMessageDelivery: true,
       })
     );
+    props.alertingSnsTopic.grantPublish(this.function);
 
     this.function.addEventSource(
       new SqsEventSource(this.queue, {

infra/lib/lake-writer.ts

@@ -10,7 +10,7 @@ interface LakeWriterProps {
   realtimeBucket: s3.IBucket;
   outputBucket: s3.IBucket;
   outputObjectPrefix: string;
-  alertingSnsTopic: sns.Topic;
+  ruleMatchesSnsTopic: sns.Topic;
 }
 
 export class LakeWriter extends Construct {
@@ -43,14 +43,14 @@ export class LakeWriter extends Construct {
         RUST_LOG: "warn,lake_writer=info",
         OUT_BUCKET_NAME: props.outputBucket.bucketName,
         OUT_KEY_PREFIX: props.outputObjectPrefix,
-        ALERTING_SNS_TOPIC_ARN: props.alertingSnsTopic.topicArn,
+        RULE_MATCHES_SNS_TOPIC_ARN: props.ruleMatchesSnsTopic.topicArn,
       },
       timeout: cdk.Duration.seconds(120),
       // prevent concurrency
       reservedConcurrentExecutions: 1,
     });
     props.realtimeBucket.grantRead(this.alertsLakeWriterLambda);
     props.outputBucket.grantReadWrite(this.alertsLakeWriterLambda);
-    props.alertingSnsTopic.grantPublish(this.alertsLakeWriterLambda);
+    props.ruleMatchesSnsTopic.grantPublish(this.alertsLakeWriterLambda);
   }
 }

infra/src/DPMainStack.ts

@@ -82,7 +82,7 @@ export class DPMainStack extends MatanoStack {
 
     const lakeWriter = new LakeWriter(this, "LakeWriter", {
       realtimeBucket: props.realtimeBucket,
-      alertingSnsTopic: matanoAlerting.alertingTopic,
+      ruleMatchesSnsTopic: matanoAlerting.ruleMatchesTopic,
       outputBucket: props.lakeStorageBucket.bucket,
       outputObjectPrefix: "lake",
     });

lib/rust/alert_forwarder/Cargo.toml

@@ -31,6 +31,7 @@ lambda_runtime = "0.7.0"
 aws-config = "0.51.0"
 aws_lambda_events = "0.7.2"
 aws-sdk-sqs = "0.21.0"
+aws-sdk-sns = "0.21.0"
 aws-sdk-dynamodb = "0.21.0"
 serde_dynamo = { version = "4", features = ["aws-sdk-dynamodb+0_21"] }