Merge pull request #208 from SumoLogic/sumo_246483

Added support to subscribe existing log groups by User defined Tags

Author: akhil-sumologic

loggroup-lambda-connector/Readme.md

@@ -1,7 +1,7 @@
 # SumoLogic LogGroup Connector
 This is used to automatically subscribe newly created and existing Cloudwatch LogGroups to a Lambda function.
 
-**Note:**
+> **Note:**
 For existing CloudWatch LogGroups, a Lambda function can subscribe to up to 65,000 LogGroups.
 If the number of LogGroups exceeds 65,000, you can request to disable Lambda recursive loop detection by [contact AWS Support](https://repost.aws/knowledge-center/aws-phone-support).
 
@@ -25,13 +25,30 @@ Made with ❤️ by Sumo Logic. Available on the [AWS Serverless Application Rep
 
 
 ### Configuring Lambda
-It has two environment variables
-
-**LOG_GROUP_PATTERN**: This is a javascript regex to filter out loggroups. Only loggroups which match this pattern will be subscribed to the lambda function.Do not use '/' while writing the pattern and it is case insensitive.
-
-```
-    Test - will match testlogroup, logtestgroup and LogGroupTest
+#### Environment variables
+
+**LOG_GROUP_PATTERN**: This JavaScript regex is used to filter log groups. Only log groups that match this pattern will be subscribed to the Lambda function. The default value is `Test`, which will match log groups like `testlogroup`, `logtestgroup`, and `LogGroupTest`.
+
+##### Use Cases and it's Regex Pattern Example
+
+| Case Description                                                     | Regex Pattern  Example              |
+|----------------------------------------------------------------------|-------------------------------------|
+| To subscribe all loggroup                                            | `/*` or (leave empty)               |
+| To subscribe all loggroup paths only                                 | `/`                                 |
+| To subscribe all loggroup of aws services                            | `/aws/*`                            |
+| To subscribe to loggroups for only one service, such as Lambda       | `/aws/lambda/*`                     |
+| To subscribe loggroup multiple services like lambda, rds, apigateway | `/aws/(lambda\|rds\|apigateway)`    |
+| To subscribe loggroup by key word like `Test` or `Prod`              | `Test` or `Prod` [Case insensitive] |
+| Don't subscribe if `LOG_GROUP_PATTERN`                               | `^$`                                |
+
+**LOG_GROUP_TAGS**: This is used to filter log groups based on tags. Only log groups that match any of the specified key-value pairs will be subscribed to the Lambda function. It is case-sensitive.
+#### For example
+```bash
+LOG_GROUP_TAGS="Environment=Production,Application=MyApp"
 ```
+> 💡 **Tip**: To filter log groups based on tags only, set `LOG_GROUP_PATTERN=^$`.
+
+> **Note**: `LOG_GROUP_PATTERN` and `LOG_GROUP_TAGS` can be used together to subscribe to log groups or can be used separately.
 
 **DESTINATION_ARN**: This specifies ARN of the Destination to Subscribe the log group. 
 
@@ -54,8 +71,6 @@ Lambda Destination ARN :- This specifies ARN of the Lambda function. Also you ha
 
 Kinesis Destination ARN :- This specifies the ARN of the kinesis Stream.
 
-**LOG_GROUP_TAGS**: This is used for filtering out loggroups based on tags.Only loggroups which match any one of the key value pairs will be subscribed to the lambda function. This works only for new loggroups not existing loggroups.
-
 **ROLE_ARN** : This is used when subscription destination ARN is kinesis firehose stream.
 
 ### For Developers

loggroup-lambda-connector/sam/packaged.yaml

@@ -21,10 +21,10 @@ Metadata:
     - serverless
     - loggroups
     - cloudwatch
-    LicenseUrl: s3://appdevstore/LoggroupConnector/v1.0.12/6092dd6c323e33634657102f570628e0
+    LicenseUrl: s3://appdevstore/LoggroupConnector/v1.0.14/6092dd6c323e33634657102f570628e0
     Name: sumologic-loggroup-connector
-    ReadmeUrl: s3://appdevstore/LoggroupConnector/v1.0.12/999bf8292d709fcf681946996eb9071d
-    SemanticVersion: 1.0.12
+    ReadmeUrl: s3://appdevstore/LoggroupConnector/v1.0.14/60b531a8a7a836857dd096ea058dc2c6
+    SemanticVersion: 1.0.14
     SourceCodeUrl: https://github.com/SumoLogic/sumologic-aws-lambda/tree/main/loggroup-lambda-connector
     SpdxLicenseId: Apache-2.0
 Parameters:
@@ -107,7 +107,7 @@ Resources:
   SumoLogGroupLambdaConnector:
     Type: AWS::Serverless::Function
     Properties:
-      CodeUri: s3://appdevstore/LoggroupConnector/v1.0.12/6b0658f101eead0c32f0fd36c4cbedd1
+      CodeUri: s3://appdevstore/LoggroupConnector/v1.0.14/5a44aebff6ae18483b1b5d082d112e85
       Handler: loggroup-lambda-connector.handler
       Runtime: nodejs20.x
       Environment:
@@ -130,6 +130,7 @@ Resources:
           - logs:DescribeLogGroups
           - logs:DescribeLogStreams
           - logs:PutSubscriptionFilter
+          - logs:ListTagsLogGroup
           Resource:
           - Fn::Sub: arn:aws:logs:${AWS::Region}:${AWS::AccountId}:log-group:*
         - Sid: InvokePolicy

loggroup-lambda-connector/sam/sam_package.sh

@@ -10,7 +10,7 @@ else
     AWS_REGION="us-east-2"
 fi
 
-version="1.0.11"
+version="1.0.14"
 
 sam package --template-file template.yaml --s3-bucket $SAM_S3_BUCKET  --output-template-file packaged.yaml --s3-prefix "LoggroupConnector/v$version" --region $AWS_REGION
 

loggroup-lambda-connector/sam/template.yaml

@@ -24,7 +24,7 @@ Metadata:
     LicenseUrl: ../LICENSE
     Name: sumologic-loggroup-connector
     ReadmeUrl: ../Readme.md
-    SemanticVersion: 1.0.12
+    SemanticVersion: 1.0.14
     SourceCodeUrl: https://github.com/SumoLogic/sumologic-aws-lambda/tree/main/loggroup-lambda-connector
     SpdxLicenseId: Apache-2.0
 
@@ -114,6 +114,7 @@ Resources:
                 - logs:DescribeLogGroups
                 - logs:DescribeLogStreams
                 - logs:PutSubscriptionFilter
+                - logs:ListTagsLogGroup
               Resource:
                 - !Sub 'arn:aws:logs:${AWS::Region}:${AWS::AccountId}:log-group:*'
             - Sid: InvokePolicy

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

@@ -1,11 +1,82 @@
-const { CloudWatchLogsClient, PutSubscriptionFilterCommand, DescribeLogGroupsCommand } = require("@aws-sdk/client-cloudwatch-logs");
+const { CloudWatchLogsClient, PutSubscriptionFilterCommand, DescribeLogGroupsCommand, ListTagsLogGroupCommand } = require("@aws-sdk/client-cloudwatch-logs");
 const { LambdaClient, InvokeCommand } = require("@aws-sdk/client-lambda");
 
 const cwl = new CloudWatchLogsClient();
 const lambda = new LambdaClient({ apiVersion: '2015-03-31' }); // Update to the appropriate Lambda API version you require
 const maxRetryCounter = 3;
 const timeoutThreshold = 12000;
 
+
+function validateRegex(pattern) {
+    try {
+        // Attempt to create a RegExp object with the provided pattern
+        return new RegExp(pattern, "i");
+    } catch (e) {
+        // Throw an error with a descriptive message if the pattern is invalid
+        throw new Error(`Invalid regular expression pattern: ${pattern}. Error: ${e.message}`);
+    }
+}
+
+async function getTagsByLogGroupName(logGroupName, retryCounter=0) {
+    var tags = {};
+    const input = {
+      logGroupName: logGroupName, // required
+    };
+    try {
+        // ListTagsLogGroupRequest
+        let response = await cwl.send(new ListTagsLogGroupCommand(input));
+        tags = response.tags
+    } catch (err) {
+          if (err && err.message === "Rate exceeded" && retryCounter <= maxRetryCounter) {
+            retryCounter += 1
+            const delayTime = Math.pow(2, retryCounter) * 2000; // Exponential backoff
+            console.log(`ThrottlingException encountered for ${logGroupName}. Retrying in ${delayTime}ms...Attempt ${retryCounter}/${maxRetryCounter}`);
+            await delay(delayTime);
+            await getTagsByLogGroupName(logGroupName, retryCounter);
+          } else {
+            console.error(`Failed to get tags for ${logGroupName} due to ${err}`)
+          }
+    }
+    return tags
+}
+
+function IsTagMatchToLogGroup(tagMatcherForLogGroup, logGroupTags) {
+    if (tagMatcherForLogGroup && logGroupTags) {
+        let tagMatcherList = tagMatcherForLogGroup.split(",");
+        console.log("logGroupTags: ", logGroupTags);
+        let tag, key, value;
+        for (let i = 0; i < tagMatcherList.length; i++) {
+            tag = tagMatcherList[i].split("=");
+            key = tag[0].trim();
+            value = tag[1].trim();
+            if (logGroupTags[key] && logGroupTags[key] == value) {
+                return true;
+            }
+        }
+    }
+    return false;
+}
+
+async function filterExistingLogGroups(logGroupName, logGroupRegex) {
+    if (logGroupName.match(logGroupRegex)) {
+        return true;
+    }
+    var logGroupTags = await getTagsByLogGroupName(logGroupName)
+    var tagMatcherForLogGroup = process.env.LOG_GROUP_TAGS
+    console.log("Filtering log group:", logGroupName, "with tags:", logGroupTags);
+    return IsTagMatchToLogGroup(tagMatcherForLogGroup, logGroupTags)
+}
+
+function filterNewLogGroups(event, logGroupRegex) {
+    var logGroupName = event.detail.requestParameters.logGroupName;
+    if (logGroupName.match(logGroupRegex) && event.detail.eventName === "CreateLogGroup") {
+        return true;
+    }
+    var logGroupTags = event.detail.requestParameters.tags;
+    var tagMatcherForLogGroup = process.env.LOG_GROUP_TAGS
+    return IsTagMatchToLogGroup(tagMatcherForLogGroup, logGroupTags)
+}
+
 async function createSubscriptionFilter(lambdaLogGroupName, destinationArn, roleArn, additionalArgs) {
     var params={};
     if (destinationArn.startsWith("arn:aws:lambda")) {
@@ -37,46 +108,25 @@ async function createSubscriptionFilter(lambdaLogGroupName, destinationArn, role
     }
 }
 
-function filterLogGroups(event, logGroupRegex) {
-    logGroupRegex = new RegExp(logGroupRegex, "i");
-    let logGroupName = event.detail.requestParameters.logGroupName;
-    if (logGroupName.match(logGroupRegex) && event.detail.eventName === "CreateLogGroup") {
-        return true;
-    }
-    let lg_tags = event.detail.requestParameters.tags;
-    if (process.env.LOG_GROUP_TAGS && lg_tags) {
-        console.log("tags in loggroup: ", lg_tags);
-        var tags_array = process.env.LOG_GROUP_TAGS.split(",");
-        let tag, key, value;
-        for (let i = 0; i < tags_array.length; i++) {
-            tag = tags_array[i].split("=");
-            key = tag[0].trim();
-            value = tag[1].trim();
-            if (lg_tags[key] && lg_tags[key] == value) {
-                return true;
-            }
-        }
-    }
-    return false;
-}
-
 async function subscribeExistingLogGroups(logGroups, retryCounter, additionalArgs) {
-    var logGroupRegex = new RegExp(process.env.LOG_GROUP_PATTERN, "i");
+    var logGroupRegex = validateRegex(process.env.LOG_GROUP_PATTERN);
+    console.log("logGroupRegexPattern: ", logGroupRegex);
     var destinationArn = process.env.DESTINATION_ARN;
     var roleArn = process.env.ROLE_ARN;
     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 {
+        let filterStatus = await filterExistingLogGroups(logGroupName, logGroupRegex);
+        if (filterStatus) {
             return createSubscriptionFilter(logGroupName, destinationArn, roleArn, additionalArgs).catch(function (err) {
                 if (err && err.message === "Rate exceeded") {
                     failedLogGroupNames.push({ logGroupName: logGroupName });
                 }
             });
+        } else {
+            console.log("Unmatched logGroup: ", logGroupName);
+            return Promise.resolve();
         }
     }, Promise.resolve());
 
@@ -128,7 +178,7 @@ async function processExistingLogGroups(context, token, additionalArgs, errorHan
   }
 
 async function invoke_lambda(context, token, additionalArgs, errorHandler) {
-  var payload = { "existingLogs": "true", "token": token, "additionalArgs": additionalArgs};
+  var payload = {"existingLogs": "true", "token": token, "additionalArgs": additionalArgs};
   try {
     await lambda.send(new InvokeCommand({
       InvocationType: 'Event',
@@ -146,7 +196,9 @@ async function delay(ms) {
 
 async function processEvents(env, event, additionalArgs, errorHandler, retryCounter=0) {
   var logGroupName = event.detail.requestParameters.logGroupName;
-  if (filterLogGroups(event, env.LOG_GROUP_PATTERN)) {
+  var logGroupRegex = validateRegex(env.LOG_GROUP_PATTERN);
+  console.log("logGroupRegex: ", logGroupRegex);
+  if (filterNewLogGroups(event, logGroupRegex)) {
     console.log("Subscribing: ", logGroupName, env.DESTINATION_ARN);
     try {
         await createSubscriptionFilter(logGroupName, env.DESTINATION_ARN, env.ROLE_ARN, additionalArgs);
@@ -183,6 +235,9 @@ exports.handler = async function (event, context, callback) {
         callback(null, "Success");
       }
     }
+    if (!process.env.LOG_GROUP_PATTERN || process.env.LOG_GROUP_PATTERN.trim().length === 0) {
+        console.warn("LOG_GROUP_PATTERN is empty, it will subscribe to all loggroups");
+    }
     if (event.existingLogs == "true") {
       await processExistingLogGroups(context, event.token, additionalArgs, errorHandler);
     } else {

loggroup-lambda-connector/test/test-template.yaml

@@ -23,6 +23,13 @@ Parameters:
     AllowedValues: [ "true", "false" ]
     Description: "Select true for subscribing existing logs"
 
+  LogGroupTags:
+    Type: String
+    Default: ""
+    Description: Enter comma separated keyvalue pairs for filtering logGroups using
+      tags. Ex KeyName1=string,KeyName2=string. This is optional leave it blank if
+      tag based filtering is not needed.
+
   BucketName:
     Type: String
     Default: ""
@@ -72,7 +79,7 @@ Resources:
           print("success")
       Handler: index.lambda_handler
       MemorySize: 128
-      Runtime: python3.7
+      Runtime: python3.12
       Timeout: 60
       Role: !GetAtt LambdaRole.Arn
 
@@ -214,6 +221,7 @@ Resources:
         DestinationArnValue: !If [ create_invoke_permission, !GetAtt DummyLambda.Arn, !GetAtt KinesisLogsDeliveryStream.Arn ]
         LogGroupPattern: !Ref LogGroupPattern
         UseExistingLogs: !Ref UseExistingLogs
+        LogGroupTags: !Ref LogGroupTags
         RoleArn: !If [ create_invoke_permission, "", !GetAtt KinesisLogsRole.Arn ]
 
 Outputs:

loggroup-lambda-connector/test/test_loggroup_lambda_connector.py

@@ -72,15 +72,38 @@ def test_4_existing_kinesis(self):
         #self.invoke_lambda()
         self.assert_subscription_filter("SumoLGLBDFilter")
 
-    def create_stack_parameters(self, destination, existing, pattern='test'):
+    def test_5_matching_existing_loggroup_with_pattern_and_tag(self):
+        self.create_log_group_with_tag()
+        self.create_stack(self.stack_name, self.template_data,
+                          self.create_stack_parameters("Kinesis","true", loggroup_tag='env=prod'))
+        print("Testing Stack Creation")
+        self.assertTrue(self.stack_exists(self.stack_name))
+        #self.invoke_lambda()
+        self.assert_subscription_filter("SumoLGLBDFilter")
+
+    def test_6_matching_existing_loggroup_by_tag_only(self):
+        self.create_log_group_with_tag()
+        self.create_stack(self.stack_name, self.template_data,
+                          self.create_stack_parameters("Kinesis","true", loggroup_pattern='^$',
+                                                       loggroup_tag='username=akhil'))
+        print("Testing Stack Creation")
+        self.assertTrue(self.stack_exists(self.stack_name))
+        #self.invoke_lambda()
+        self.assert_subscription_filter("SumoLGLBDFilter")
+
+    def create_stack_parameters(self, destination, existing, loggroup_pattern='test', loggroup_tag=''):
         return [
             {
                 'ParameterKey': 'DestinationType',
                 'ParameterValue': destination
             },
             {
                 'ParameterKey': 'LogGroupPattern',
-                'ParameterValue': pattern
+                'ParameterValue': loggroup_pattern
+            },
+            {
+                'ParameterKey': 'LogGroupTags',
+                'ParameterValue': loggroup_tag
             },
             {
                 'ParameterKey': 'UseExistingLogs',
@@ -137,6 +160,16 @@ def create_log_group(self):
         response = self.log_group_client.create_log_group(logGroupName=self.log_group_name)
         print("creating log group", response)
 
+    def create_log_group_with_tag(self):
+        tags = {
+            'team': 'apps',
+            'env': 'prod'
+        }
+        self.log_group_name = 'mytag-%s' % (datetime.datetime.now().strftime("%d-%m-%y-%H-%M-%S"))
+        print("Loggroup Name", self.log_group_name)
+        response = self.log_group_client.create_log_group(logGroupName=self.log_group_name, tags=tags)
+        print("creating log group", response)
+
     def assert_subscription_filter(self, filter_name):
         sleep(60)
         response = self.log_group_client.describe_subscription_filters(
@@ -205,7 +238,8 @@ def create_sam_package_and_upload():
 
 def _run(command, input=None, check=False, **kwargs):
     if sys.version_info >= (3, 5):
-        return subprocess.run(command, capture_output=True)
+        result = subprocess.run(command, capture_output=True)
+        return result.returncode, result.stdout, result.stderr
     if input is not None:
         if 'stdin' in kwargs:
             raise ValueError('stdin and input arguments may not both be used.')
@@ -226,11 +260,11 @@ def _run(command, input=None, check=False, **kwargs):
 
 
 def run_command(cmdargs):
-    resp = _run(cmdargs)
-    if len(resp.stderr.decode()) > 0:
+    retcode, stdout, stderr = _run(cmdargs)
+    if retcode != 0:
         # traceback.print_exc()
-        raise Exception("Error in run command %s cmd: %s" % (resp, cmdargs))
-    return resp.stdout
+        raise Exception("Error in run command %s cmd: %s" % (stderr, cmdargs))
+    return retcode, stdout, stderr
 
 
 if __name__ == '__main__':