adding retry mechanism in case of throttling. Preventing deletion of primer resource.

Author: sourabh

loggroup-lambda-connector/sam/template.yaml

@@ -207,6 +207,8 @@ Resources:
     Type: AWS::CloudFormation::CustomResource
     Version: "1.0"
     Condition: invoke_existing
+    DeletionPolicy : "Retain"
+    UpdateReplacePolicy: "Retain"
     Properties:
       ServiceToken: !GetAtt SumoLogGroupExistingLambdaConnector.Arn
       DESTINATION_ARN: !Ref "DestinationArnValue"

loggroup-lambda-connector/src/loggroup-lambda-connector.js

@@ -1,7 +1,8 @@
 var AWS = require("aws-sdk");
+const util = require("util");
 var cwl = new AWS.CloudWatchLogs({apiVersion: '2014-03-28'});
 
-async function createSubscriptionFilter(lambdaLogGroupName, destinationArn, roleArn, errorHandler) {
+async function createSubscriptionFilter(lambdaLogGroupName, destinationArn, roleArn) {
     if (destinationArn.startsWith("arn:aws:lambda")){
         var params = {
             destinationArn: destinationArn,
@@ -20,7 +21,13 @@ async function createSubscriptionFilter(lambdaLogGroupName, destinationArn, role
     }
 
     // handle case where subscription filter exists/case where loggroup generated by target lambda
-   await cwl.putSubscriptionFilter(params, errorHandler);
+    try {
+        await util.promisify(cwl.putSubscriptionFilter.bind(cwl))(params);
+        console.log("Successfully subscribed logGroup: ", lambdaLogGroupName);
+    } catch (err) {
+        console.log("Error in subscribing", lambdaLogGroupName, err);
+        throw err;
+    }
 }
 
 function filterLogGroups(event, logGroupRegex) {
@@ -46,24 +53,29 @@ function filterLogGroups(event, logGroupRegex) {
     return false;
 }
 
-async function subscribeExistingLogGroups(logGroups) {
-    var logGroupName;
+async function subscribeExistingLogGroups(logGroups, counter) {
     var logGroupRegex = new RegExp(process.env.LOG_GROUP_PATTERN, "i");
     var destinationArn = process.env.DESTINATION_ARN;
     var roleArn = process.env.ROLE_ARN;
-    for (var i = logGroups.length - 1; i >= 0; i--) {
-        logGroupName = logGroups[i].logGroupName;
-        if (logGroupName.match(logGroupRegex)) {
-            await createSubscriptionFilter(logGroupName, destinationArn, roleArn, (function(inner_logGroupName) { return function (err, data) {
-                if (err) {
-                    console.log("Error in subscribing", inner_logGroupName, err);
-                } else {
-                    console.log("Successfully subscribed logGroup: ", inner_logGroupName);
-                }
-            };})(logGroupName));
-        } else {
+    const failedLogGroupNames = [];
+    await logGroups.reduce(async (previousPromise, nextLogGroup) => {
+        await previousPromise;
+        const { logGroupName } = nextLogGroup;
+        if (!logGroupName.match(logGroupRegex)) {
             console.log("Unmatched logGroup: ", logGroupName);
+            return Promise.resolve();
+        } else {
+            return createSubscriptionFilter(logGroupName, destinationArn, roleArn).catch( function (err) {
+                if (err && err.code == "ThrottlingException") {
+                    failedLogGroupNames.push({logGroupName: logGroupName});
+                }
+            });
         }
+    }, Promise.resolve());
+
+    if (counter < 4 && failedLogGroupNames.length > 0) {
+        console.log("Retrying Subscription for Failed Log Groups due to throttling with counter number as " + counter);
+        await subscribeExistingLogGroups(failedLogGroupNames, counter + 1);
     }
 }
 
@@ -83,12 +95,12 @@ function processExistingLogGroups(token, context, errorHandler) {
                 reject(err);
             } else {
                 console.log("fetched logGroups: " + data.logGroups.length + " nextToken: " + data.nextToken);
-                subscribeExistingLogGroups(data.logGroups);
                 resolve(data);
             }
         });
     });
-    var cb = function (data) {
+    var cb = async function (data) {
+        await subscribeExistingLogGroups(data.logGroups, 0);
         if (data.nextToken) {// if next set of log groups exists, invoke next instance of lambda
             console.log("Log Groups remaining...Calling the lambda again with token " + data.nextToken);
             invoke_lambda(context, data.nextToken, errorHandler);
@@ -118,7 +130,9 @@ function processEvents(env, event, errorHandler) {
     var logGroupName = event.detail.requestParameters.logGroupName;
     if (filterLogGroups(event, env.LOG_GROUP_PATTERN)) {
         console.log("Subscribing: ", logGroupName, env.DESTINATION_ARN);
-        createSubscriptionFilter(logGroupName, env.DESTINATION_ARN, env.ROLE_ARN, errorHandler);
+        createSubscriptionFilter(logGroupName, env.DESTINATION_ARN, env.ROLE_ARN).catch (function (err) {
+            errorHandler(err, "Error in Subscribing.");
+        });
     } else {
         console.log("Unmatched: ", logGroupName, env.DESTINATION_ARN);
     }