Merge pull request #98 from ykitamura-mdsol/feature/log_stream_filter

Add log stream name prefix filtering, especially for AWS Batch logs

Author: himanshu219

cloudwatchlogs-with-dlq/DLQLambdaCloudFormation.json

@@ -28,6 +28,10 @@
         "Default": "false",
         "AllowedValues" : ["true" ,"false"],
         "Description": "Select true to get loggroup/logstream values in logs"
+      },
+      "LogStreamPrefix": {
+        "Type": "String",
+        "Description": "Enter comma separated list of logStream name prefixes to filter by logStream"
       }
     },
     "Mappings" : {
@@ -221,7 +225,8 @@
                     "Variables": {
                         "SUMO_ENDPOINT": {"Ref": "SumoEndPointURL"},
                         "LOG_FORMAT": {"Ref": "LogFormat"},
-                        "INCLUDE_LOG_INFO": {"Ref": "IncludeLogGroupInfo"}
+                        "INCLUDE_LOG_INFO": {"Ref": "IncludeLogGroupInfo"},
+                        "LOG_STREAM_PREFIX": {"Ref": "LogStreamPrefix"}
 
                     }
                 }
@@ -296,7 +301,8 @@
                         },
                         "NUM_OF_WORKERS": {"Ref": "NumOfWorkers"},
                         "LOG_FORMAT": {"Ref": "LogFormat"},
-                        "INCLUDE_LOG_INFO": {"Ref": "IncludeLogGroupInfo"}
+                        "INCLUDE_LOG_INFO": {"Ref": "IncludeLogGroupInfo"},
+                        "LOG_STREAM_PREFIX": {"Ref": "LogStreamPrefix"}
                     }
                 }
             }

cloudwatchlogs-with-dlq/Readme.md

@@ -21,6 +21,7 @@ The following AWS Lambda environment variables are supported in both the lambda
 * SOURCE_NAME_OVERRIDE (OPTIONAL) - Override _sourceName metadata field within SumoLogic.
 * INCLUDE_LOG_INFO (OPTIONAL) - Set it to true when loggroup/logstream values needs to be included in logs. Default is false
 * LOG_FORMAT - Default is Others. One can choose VPC-JSON for VPC flow logs in json format and VPC-RAW for only RAW message line
+* LOG_STREAM_PREFIX (OPTIONAL) - Comma separated list of logStream name prefixes to filter by logStream, especially for AWS Batch logs
 
 ### Configuring Lambda for VPC Flow Logs
 The following AWS Lambda environment variables are supported in both the lambda functions for VPC flow logs.

cloudwatchlogs-with-dlq/cloudwatchlogs_lambda.js

@@ -82,7 +82,11 @@ function getConfig(env) {
         "compressData": env.COMPRESS_DATA || true,
         "vpcCIDRPrefix": env.VPC_CIDR_PREFIX || '',
         "includeLogInfo": ("INCLUDE_LOG_INFO" in env) ? env.INCLUDE_LOG_INFO === "true" : false,
-        "includeSecurityGroupInfo": ("INCLUDE_SECURITY_GROUP_INFO" in env) ? env.INCLUDE_SECURITY_GROUP_INFO === "true" : false
+        "includeSecurityGroupInfo": ("INCLUDE_SECURITY_GROUP_INFO" in env) ? env.INCLUDE_SECURITY_GROUP_INFO === "true" : false,
+        // Regex to filter by logStream name prefixes
+        "logStreamPrefixRegex": ("LOG_STREAM_PREFIX" in env)
+                                ? new RegExp('^(' + escapeRegExp(env.LOG_STREAM_PREFIX).replace(/,/g, '|')  + ')', 'i')
+                                : ''
     };
     if (!config.SumoURL) {
         return new Error('Undefined SUMO_ENDPOINT environment variable');
@@ -95,6 +99,10 @@ function getConfig(env) {
     return config;
 }
 
+function escapeRegExp(string) {
+  return string.replace(/[|\\{}()[\]^$+*?.-]/g, '\\$&');
+}
+
 function transformRecords(config, records) {
     return new Promise(function (resolve, reject) {
         if (config.LogFormat === "VPC-JSON" && config.includeSecurityGroupInfo) {
@@ -134,6 +142,8 @@ exports.processLogs = function (env, eventAwslogsData, callback) {
         var records = [];
         if (awslogsData.messageType === 'CONTROL_MESSAGE') {
             console.log('Skipping Control Message');
+        } else if(config.logStreamPrefixRegex && !awslogsData.logStream.match(config.logStreamPrefixRegex)){
+            console.log('Skipping Non-Applicable Log Stream');
         } else {
             records = createRecords(config, awslogsData.logEvents, awslogsData);
             console.log(records.length + " Records Found");

cloudwatchlogs/README.md

@@ -1,4 +1,4 @@
-# Sumo Logic Functions for AWS CloudWatch Logs 
+# Sumo Logic Functions for AWS CloudWatch Logs
 
 AWS Lambda function to collector logs from CloudWatch Logs and post them to [SumoLogic](http://www.sumologic.com) via a [HTTP collector endpoint](http://help.sumologic.com/Send_Data/Sources/02Sources_for_Hosted_Collectors/HTTP_Source)
 
@@ -20,7 +20,7 @@ First create an [HTTP collector endpoint](http://help.sumologic.com/Send_Data/So
    * Copy code from cloudwatchlogs_lambda.js into the Lambda function code.
    * Add Environment variables (See below)
 5. Scroll down to the `Lambda function handle and role` section, make sure you set the right values that match the function. For role, you can just use the basic execution role. Click next.
-6. Finally click on "Create function" to create the function. 
+6. Finally click on "Create function" to create the function.
 7. (Optional) Test this new function with sample AWS CloudWatch Logs template provided by AWS
 
 ## Create Stream from CloudWatch Logs
@@ -41,6 +41,7 @@ The following AWS Lambda environment variables are supported
 * `SOURCE_CATEGORY_OVERRIDE` (OPTIONAL) - Override _sourceCategory metadata field within SumoLogic. If `none` will not be overridden
 * `SOURCE_HOST_OVERRIDE` (OPTIONAL) - Override _sourceHost metadata field within SumoLogic. If `none` will not be overridden
 * `SOURCE_NAME_OVERRIDE` (OPTIONAL) - Override _sourceName metadata field within SumoLogic. If `none` will not be overridden
+* `LOG_STREAM_PREFIX` (OPTIONAL) - Comma separated list of logStream name prefixes to filter by logStream, especially for AWS Batch logs
 
 # Dynamic Metadata Fields
 
@@ -52,7 +53,7 @@ For example:
 
 ```
 exports.handler = (event, context, callback) => {
-    
+
     var serverIp = '123.123.123.123'
 
     console.log(JSON.stringify({
@@ -62,7 +63,7 @@ exports.handler = (event, context, callback) => {
             'source': 'other_source',
             'host': serverIp
         }
-    
+
     }));
     console.log('some other log message with default sourceCategory');
 };

cloudwatchlogs/cloudwatchlogs_lambda.js

@@ -21,6 +21,11 @@ var encoding = process.env.ENCODING || 'utf-8';  // default is utf-8
 // Include logStream and logGroup as json fields within the message. Required for SumoLogic AWS Lambda App
 var includeLogInfo = false;  // default is false
 
+// Regex to filter by logStream name prefixes
+var logStreamPrefixRegex = process.env.LOG_STREAM_PREFIX
+                            ? new RegExp('^(' + escapeRegExp(process.env.LOG_STREAM_PREFIX).replace(/,/g, '|')  + ')', 'i')
+                            : '';
+
 // Regex used to detect logs coming from lambda functions.
 // The regex will parse out the requestID and strip the timestamp
 // Example: 2016-11-10T23:11:54.523Z	108af3bb-a79b-11e6-8bd7-91c363cc05d9    some message
@@ -33,6 +38,9 @@ var https = require('https');
 var zlib = require('zlib');
 var url = require('url');
 
+function escapeRegExp(string) {
+  return string.replace(/[|\\{}()[\]^$+*?.-]/g, '\\$&');
+}
 
 function sumoMetaKey(awslogsData, message) {
     var sourceCategory = '';
@@ -158,6 +166,9 @@ exports.handler = function (event, context, callback) {
         if (awslogsData.messageType === 'CONTROL_MESSAGE') {
             console.log('Control message');
             callback(null, 'Success');
+        } else if(logStreamPrefixRegex && !awslogsData.logStream.match(logStreamPrefixRegex)){
+            console.log('Skipping Non-Applicable Log Stream');
+            return callback(null, 'Success');
         }
 
         var lastRequestID = null;