feat(*): Adding log processer lambda to firehose

daniel-cottone · Jul 29, 2018 · 205b1a0 · 205b1a0
1 parent c6f706a
commit 205b1a0
Show file tree

Hide file tree

Showing 8 changed files with 438 additions and 18 deletions.
diff --git a/package.json b/package.json
@@ -36,14 +36,16 @@
     ]
   },
   "dependencies": {
-    "lodash": "^4.17.10"
+    "fs-extra": "7.0.0",
+    "lodash": "4.17.10"
   },
   "devDependencies": {
     "@semantic-release/changelog": "3.0.0",
     "@semantic-release/git": "7.0.1",
     "@semantic-release/github": "5.0.1",
+    "@types/fs-extra": "5.0.4",
     "@types/lodash": "4.14.115",
-    "@types/node": "^10.5.4",
+    "@types/node": "10.5.4",
     "rimraf": "2.6.2",
     "semantic-release": "15.8.1",
     "tslint": "5.11.0",

diff --git a/src/formatters/index.ts b/src/formatters/index.ts
@@ -32,6 +32,11 @@ export function formatIamFirehoseRole(opts: IFormatterOpts): any {
     service,
     'es-logs-cw-policy',
   ];
+  properties.Policies[3].PolicyName['Fn::Join'][1] = [
+    stage,
+    service,
+    'es-logs-lambda-policy',
+  ];
   properties.RoleName['Fn::Join'][1] = [
     service,
     stage,

diff --git a/src/index.ts b/src/index.ts
@@ -1,4 +1,6 @@
 import _ from 'lodash';
+import fs from 'fs-extra';
+import path from 'path';
 
 import { formatFirehose, formatIamFirehoseRole, formatLogGroup } from './formatters';
 
@@ -8,6 +10,7 @@ const cloudwatchLogStreamEsTemplate = require('../templates/cloudwatch-log-strea
 const cloudwatchLogStreamS3Template = require('../templates/cloudwatch-log-stream-s3.json');
 const firehoseTemplate = require('../templates/firehose.json');
 const iamCloudwatchTemplate = require('../templates/iam-cloudwatch.json');
+const iamLambdaTemplate = require('../templates/iam-lambda.json');
 const iamFirehoseTemplate = require('../templates/iam-firehose.json');
 const s3Template = require('../templates/s3.json');
 // tslint:enable:no-var-requires
@@ -18,6 +21,8 @@ class ServerlessEsLogsPlugin {
   private serverless: any;
   private options: { [name: string]: any };
   private custom: { [name: string]: any };
+  private logProcesserDir: string = '_es-logs';
+  private logProcesserName: string = 'esLogsProcesser';
 
   constructor(serverless: any, options: { [name: string]: any }) {
     this.serverless = serverless;
@@ -26,24 +31,34 @@ class ServerlessEsLogsPlugin {
     this.custom = serverless.service.custom || {};
     this.hooks = {
       'after:package:initialize': this.afterPackageInitialize.bind(this),
-      'aws:package:finalize:mergeCustomProviderResources': this.mergeResources.bind(this),
+      'after:package:createDeploymentArtifacts': this.afterPackageCreateDeploymentArtifacts.bind(this),
+      'aws:package:finalize:mergeCustomProviderResources': this.mergeCustomProviderResources.bind(this),
     };
   }
 
+  private afterPackageCreateDeploymentArtifacts(): void {
+    this.serverless.cli.log('ServerlessEsLogsPlugin.afterPackageCreateDeploymentArtifacts()');
+    this.cleanupFiles();
+  }
+
   private afterPackageInitialize(): void {
     this.serverless.cli.log('ServerlessEsLogsPlugin.afterPackageInitialize()');
-    this.options.stage  = this.options.stage
+    this.options.stage = this.options.stage
       || this.serverless.service.provider.stage
       || (this.serverless.service.defaults && this.serverless.service.defaults.stage)
       || 'dev';
     this.options.region = this.options.region
       || this.serverless.service.provider.region
       || (this.serverless.service.defaults && this.serverless.service.defaults.region)
       || 'us-east-1';
+
+    // Add log processing lambda
+    // TODO: Find the right lifecycle method for this
+    this.addLogProcesser();
   }
 
-  private mergeResources(): void {
-    this.serverless.cli.log('ServerlessEsLogsPlugin.mergeResources()');
+  private mergeCustomProviderResources(): void {
+    this.serverless.cli.log('ServerlessEsLogsPlugin.mergeCustomProviderResources()');
     const { stage, region } = this.options;
     const template = this.serverless.service.provider.compiledCloudFormationTemplate;
     const formatOpts = {
@@ -55,10 +70,10 @@ class ServerlessEsLogsPlugin {
       stage,
     };
 
-    // Add cloudwatch subscriptions to existing functions
+    // Add cloudwatch subscriptions to firehose for functions' log groups
     this.addCloudwatchSubscriptions();
 
-    // Add resources for firehose -> elasticsearch
+    // Add custom resources for firehose -> elasticsearch
     _.merge(template.Resources, s3Template);
     _.merge(template.Resources, formatLogGroup({
       ...formatOpts,
@@ -67,6 +82,7 @@ class ServerlessEsLogsPlugin {
     _.merge(template.Resources, cloudwatchLogStreamEsTemplate);
     _.merge(template.Resources, cloudwatchLogStreamS3Template);
     _.merge(template.Resources, iamCloudwatchTemplate);
+    _.merge(template.Resources, iamLambdaTemplate);
     _.merge(template.Resources, formatIamFirehoseRole({
       ...formatOpts,
       template: iamFirehoseTemplate,
@@ -76,14 +92,19 @@ class ServerlessEsLogsPlugin {
       template: firehoseTemplate,
     }));
 
-    // console.log(JSON.stringify(template, null, 2));
+    // Patch log processer lambda role
+    this.patchLogProcesserRole();
   }
 
   private addCloudwatchSubscriptions(): void {
     const template = this.serverless.service.provider.compiledCloudFormationTemplate;
     const subscriptionsTemplate: { [name: string]: any } = {};
     const functions = this.serverless.service.getAllFunctions();
     functions.forEach((name: string) => {
+      if (name === this.logProcesserName) {
+        return;
+      }
+
       const normalizedFunctionName = this.provider.naming.getNormalizedFunctionName(name);
       const logicalId = `${normalizedFunctionName}SubscriptionFilter`;
       const logGroupLogicalId = `${normalizedFunctionName}LogGroup`;
@@ -114,6 +135,47 @@ class ServerlessEsLogsPlugin {
     });
     _.merge(template.Resources, subscriptionsTemplate);
   }
+
+  private addLogProcesser(): void {
+    const dirPath = path.join(this.serverless.config.servicePath, this.logProcesserDir);
+    const filePath = path.join(dirPath, 'index.js');
+    const handler = `${this.logProcesserDir}/index.handler`;
+    const name = `${this.serverless.service.service}-${this.options.stage}-es-logs-plugin`;
+    fs.ensureDirSync(dirPath);
+    fs.copySync(path.resolve(__dirname, '../templates/logProcesser.js'), filePath);
+    this.serverless.service.functions[this.logProcesserName] = {
+      description: 'Serverless ES Logs Plugin',
+      handler,
+      events: [],
+      memorySize: 512,
+      name,
+      runtime: 'nodejs8.10',
+      package: {
+        individually: true,
+        exclude: ['**'],
+        include: [this.logProcesserDir + '/**'],
+      },
+      timeout: 60,
+    };
+  }
+
+  private patchLogProcesserRole(): void {
+    const normalizedFunctionName = this.provider.naming.getNormalizedFunctionName(this.logProcesserName);
+    const templateKey = `${normalizedFunctionName}LambdaFunction`;
+    const template = this.serverless.service.provider.compiledCloudFormationTemplate;
+    template.Resources[templateKey].DependsOn.push('ServerlessEsLogsLambdaIAMRole');
+    template.Resources[templateKey].Properties.Role = {
+      'Fn::GetAtt': [
+        'ServerlessEsLogsLambdaIAMRole',
+        'Arn',
+      ],
+    };
+  }
+
+  private cleanupFiles(): void {
+    const dirPath = path.join(this.serverless.config.servicePath, this.logProcesserDir);
+    fs.removeSync(dirPath);
+  }
 }
 
 export = ServerlessEsLogsPlugin;
diff --git a/templates/firehose.json b/templates/firehose.json
@@ -6,7 +6,8 @@
       "ServerlessEsLogsLogGroup",
       "ServerlessEsLogsEsLogStream",
       "ServerlessEsLogsS3LogStream",
-      "ServerlessEsLogsIAMRole"
+      "ServerlessEsLogsIAMRole",
+      "EsLogsProcesserLambdaFunction"
     ],
     "Properties": {
       "DeliveryStreamName": "",
@@ -24,7 +25,7 @@
         "IndexRotationPeriod": "OneDay",
         "BufferingHints": {
           "IntervalInSeconds": 60,
-          "SizeInMBs": 5
+          "SizeInMBs": 3
         },
         "RetryOptions": {
           "DurationInSeconds": 60
@@ -45,7 +46,7 @@
           },
           "BufferingHints": {
             "IntervalInSeconds": 60,
-            "SizeInMBs": 5
+            "SizeInMBs": 3
           },
           "CompressionFormat": "UNCOMPRESSED",
           "EncryptionConfiguration": {
@@ -62,8 +63,44 @@
           }
         },
         "ProcessingConfiguration": {
-          "Enabled": false,
-          "Processors": []
+          "Enabled": true,
+          "Processors": [
+            {
+              "Type": "Lambda",
+              "Parameters": [
+                {
+                  "ParameterName": "LambdaArn",
+                  "ParameterValue": {
+                    "Fn::GetAtt": [
+                      "EsLogsProcesserLambdaFunction",
+                      "Arn"
+                    ]
+                  }
+                },
+                {
+                  "ParameterName": "NumberOfRetries",
+                  "ParameterValue": "3"
+                },
+                {
+                  "ParameterName": "RoleArn",
+                  "ParameterValue": {
+                    "Fn::GetAtt": [
+                      "ServerlessEsLogsIAMRole",
+                      "Arn"
+                    ]
+                  }
+                },
+                {
+                  "ParameterName": "BufferSizeInMBs",
+                  "ParameterValue": "3"
+                },
+                {
+                  "ParameterName": "BufferIntervalInSeconds",
+                  "ParameterValue": "60"
+                }
+              ]
+            }
+          ]
         },
         "CloudWatchLoggingOptions": {
           "Enabled": true,

diff --git a/templates/iam-firehose.json b/templates/iam-firehose.json
@@ -134,7 +134,7 @@
                 "Resource": [
                   {
                     "Fn::GetAtt": [
-                      "ServerlessEsLogsLogGroup",
+                      "EsLogsProcesserLambdaFunction",
                       "Arn"
                     ]
                   }

diff --git a/templates/iam-lambda.json b/templates/iam-lambda.json
@@ -0,0 +1,42 @@
+{
+  "ServerlessEsLogsLambdaIAMRole": {
+    "Type": "AWS::IAM::Role",
+    "Properties": {
+      "AssumeRolePolicyDocument": {
+        "Version": "2012-10-17",
+        "Statement": [
+          {
+            "Effect": "Allow",
+            "Principal": {
+              "Service": [
+                "lambda.amazonaws.com"
+              ]
+            },
+            "Action": [
+              "sts:AssumeRole"
+            ]
+          }
+        ]
+      },
+      "Policies": [
+        {
+          "PolicyName": "lambda-to-firehose-policy",
+          "PolicyDocument": {
+            "Version": "2012-10-17",
+            "Statement": [
+              {
+                "Effect": "Allow",
+                "Action": [
+                  "firehose:*"
+                ],
+                "Resource": {
+                  "Fn::Sub": "arn:aws:firehose:${AWS::Region}:${AWS::AccountId}:deliverystream/*"
+                }
+              }
+            ]
+          }
+        }
+      ]
+    }
+  }
+}