create topic subscriptions on consuming side

To avoid cyclic references [1], we need to create SNS subscriptions on the conuming stack instead of the topic's stack. This change implements a workaround to the issue. aws/aws-cdk#3064
eladb · Jun 25, 2019 · 0a5826c · 0a5826c
1 parent 4329c5a
commit 0a5826c
Show file tree

Hide file tree

Showing 3 changed files with 26 additions and 13 deletions.
diff --git a/grapl-cdk/.gitignore b/grapl-cdk/.gitignore
@@ -0,0 +1 @@
+cdk.out/
diff --git a/grapl-cdk/index.ts b/grapl-cdk/index.ts
@@ -453,7 +453,7 @@ class SysmonSubgraphGenerator extends cdk.Stack {
         const service = new Service(this, 'sysmon-subgraph-generator', environment);
 
         service.readsFrom(reads_from);
-        subscribes_to.addSubscription(new snsSubs.SqsSubscription(service.queues.queue));
+        addSubscription(this, subscribes_to, new snsSubs.SqsSubscription(service.queues.queue));
         service.publishesToBucket(writes_to);
     }
 }
@@ -475,11 +475,25 @@ class GenericSubgraphGenerator extends cdk.Stack {
         const service = new Service(this, 'generic-subgraph-generator', environment);
 
         service.readsFrom(reads_from);
-        subscribes_to.addSubscription(new snsSubs.SqsSubscription(service.queues.queue));
+
+        addSubscription(this, subscribes_to, new snsSubs.SqsSubscription(service.queues.queue));
+
         service.publishesToBucket(writes_to);
     }
 }
 
+function addSubscription(scope, topic, subscription) {
+    const config = subscription.bind(topic);
+
+    new sns.Subscription(scope, 'Subscription', {
+        topic: topic,
+        endpoint: config.endpoint,
+        filterPolicy: config.filterPolicy,
+        protocol: config.protocol,
+        rawMessageDelivery: config.rawMessageDelivery
+    });
+}
+
 
 class NodeIdentityMapper extends cdk.Stack {
 
@@ -500,7 +514,7 @@ class NodeIdentityMapper extends cdk.Stack {
 
         service.readsFrom(reads_from);
 
-        subscribes_to.addSubscription(new snsSubs.SqsSubscription(service.queues.queue));
+        addSubscription(this, subscribes_to, new snsSubs.SqsSubscription(service.queues.queue));
 
         service.event_handler.connections.allowToAnyIPv4(ec2.Port.tcp(443), 'Allow outbound to S3');
         service.event_retry_handler.connections.allowToAnyIPv4(ec2.Port.tcp(443), 'Allow outbound to S3');
@@ -530,7 +544,7 @@ class NodeIdentifier extends cdk.Stack {
 
         history_db.allowReadWrite(service);
         service.publishesToBucket(writes_to);
-        subscribes_to.addSubscription(new snsSubs.SqsSubscription(service.queues.queue));
+        addSubscription(this, subscribes_to,, new snsSubs.SqsSubscription(service.queues.queue));
         service.event_handler.connections.allowToAnyIPv4(ec2.Port.tcp(443), 'Allow outbound to S3');
         service.event_retry_handler.connections.allowToAnyIPv4(ec2.Port.tcp(443), 'Allow outbound to S3');
 
@@ -563,7 +577,7 @@ class GraphMerger extends cdk.Stack {
         service.readsFrom(reads_from);
         service.publishesToTopic(publishes_to);
 
-        subscribes_to.addSubscription(new snsSubs.SqsSubscription(service.queues.queue));
+        addSubscription(this, subscribes_to, new snsSubs.SqsSubscription(service.queues.queue));
         //
         // service.event_handler.connections
         //     .allowToAnyIPv4(new ec2.Port({
@@ -594,12 +608,12 @@ class AnalyzerDispatch extends cdk.Stack {
         };
 
         const service = new Service(this, 'analyzer-dispatcher', environment, vpc);
-;
+
         service.publishesToBucket(writes_to);
         // We need the List capability to find each of the analyzers
         service.readsFrom(reads_from, true);
 
-        subscribes_to.addSubscription(new snsSubs.SqsSubscription(service.queues.queue));
+        addSubscription(this, subscribes_to, new snsSubs.SqsSubscription(service.queues.queue));
 
         service.event_handler.connections.allowToAnyIPv4(ec2.Port.allTcp(), 'Allow outbound to S3');
         service.event_retry_handler.connections.allowToAnyIPv4(ec2.Port.allTcp(), 'Allow outbound to S3');
@@ -658,8 +672,7 @@ class AnalyzerExecutor extends cdk.Stack {
         service.event_handler.addToRolePolicy(policy);
         service.event_retry_handler.addToRolePolicy(policy);
 
-
-        subscribes_to.addSubscription(new snsSubs.SqsSubscription(service.queues.queue));
+        addSubscription(this, subscribes_to, new snsSubs.SqsSubscription(service.queues.queue));
 
         service.event_handler.connections.allowToAnyIPv4(ec2.Port.allTcp(), 'Allow outbound to S3');
         service.event_retry_handler.connections.allowToAnyIPv4(ec2.Port.allTcp(), 'Allow outbound to S3');
@@ -696,7 +709,7 @@ class EngagementCreator extends cdk.Stack {
         service.readsFrom(reads_from);
         service.publishesToTopic(publishes_to);
 
-        subscribes_to.addSubscription(new snsSubs.SqsSubscription(service.queues.queue));
+        addSubscription(this, subscribes_to, new snsSubs.SqsSubscription(service.queues.queue));
 
         service.event_handler.connections.allowToAnyIPv4(ec2.Port.allTcp(), 'Allow outbound to S3');
         service.event_retry_handler.connections.allowToAnyIPv4(ec2.Port.allTcp(), 'Allow outbound to S3');
@@ -997,7 +1010,7 @@ class Grapl extends cdk.App {
     constructor() {
         super();
 
-        env(__dirname + '/.env');
+        // env(__dirname + '/.env');
 
         const mgZeroCount = Number(process.env.MG_ZEROS_COUNT) || 3;
         const mgAlphaCount = Number(process.env.MG_ALPHAS_COUNT) || 5;

diff --git a/grapl-cdk/package.json b/grapl-cdk/package.json
@@ -29,7 +29,6 @@
     "@aws-cdk/core": "^0.36.0",
     "@types/node": "^10.14.6",
     "aws-cdk": "^0.36.0",
-    "node-env-file": "^0.1.8",
-    "@aws-cdk/aws-cloudmap": "latest"
+    "node-env-file": "^0.1.8"
   }
 }