- init commit

.gitignore

@@ -0,0 +1,6 @@
+# package directories
+node_modules
+jspm_packages
+
+# Serverless directories
+.serverless

.vscode/launch.json

@@ -0,0 +1,22 @@
+{
+  // Use IntelliSense to learn about possible Node.js debug attributes.
+  // Hover to view descriptions of existing attributes.
+  // For more information, visit: https://go.microsoft.com/fwlink/?linkid=830387
+  "version": "0.2.0",
+  "configurations": [    
+    {
+      "type": "node",
+      "request": "launch",
+      "name": "ship-logs",
+      "program": "${workspaceRoot}/node_modules/.bin/sls",
+      "args": [
+        "invoke",
+        "local",
+        "-f",
+        "ship-logs",
+        "-p",
+        "events/log-message.json"
+      ]
+    }
+  ]
+}

README.md

@@ -1,2 +1,3 @@
 # lambda-logging-demo
+
 Demo for shipping logs to ELK stack, and to auto-subscribe new log groups

events/create-log-group.json

@@ -0,0 +1,42 @@
+{
+  "version": "0",
+  "id": "52d15c46-1bbc-47da-84e4-cef8c4dfc5b7",
+  "detail-type": "AWS API Call via CloudTrail",
+  "source": "aws.logs",
+  "account": "xxx",
+  "time": "2017-08-20T18:50:54Z",
+  "region": "us-east-1",
+  "resources": [],
+  "detail": {
+      "eventVersion": "1.04",
+      "userIdentity": {
+          "type": "IAMUser",
+          "principalId": "AIDAIP5G4Z37ZNNNKVEUW",
+          "arn": "arn:aws:iam::xxx:user/yancui",
+          "accountId": "xxx",
+          "accessKeyId": "xxx",
+          "userName": "yancui",
+          "sessionContext": {
+              "attributes": {
+                  "mfaAuthenticated": "false",
+                  "creationDate": "2017-08-20T18:50:48Z"
+              }
+          },
+          "invokedBy": "cloudformation.amazonaws.com"
+      },
+      "eventTime": "2017-08-20T18:50:54Z",
+      "eventSource": "logs.amazonaws.com",
+      "eventName": "CreateLogGroup",
+      "awsRegion": "us-east-1",
+      "sourceIPAddress": "cloudformation.amazonaws.com",
+      "userAgent": "cloudformation.amazonaws.com",
+      "requestParameters": {
+          "logGroupName": "/aws/lambda/logging-demo-dev-api"
+      },
+      "responseElements": null,
+      "requestID": "7f14f744-85d8-11e7-b277-a3957840b2c0",
+      "eventID": "ddcd560b-b39e-4991-8e2b-e39fd8d7dabf",
+      "eventType": "AwsApiCall",
+      "apiVersion": "20140328"
+  }
+}

events/log-message.json

@@ -0,0 +1,16 @@
+{
+  "messageType": "DATA_MESSAGE",
+  "owner": "499014364541",
+  "logGroup": "/aws/lambda/log",
+  "logStream": "2017/08/20/[$LATEST]40237edd48f34b24970905cc8e7aa220",
+  "subscriptionFilters": [
+      "ship-logs"
+  ],
+  "logEvents": [
+      {
+          "id": "33523879000509538715657644838039675459397576285294297088",
+          "timestamp": 1503262725172,
+          "message": "START RequestId: 5ae184ed-85ea-11e7-a771-25a3e4ddb615 Version: $LATEST\n"
+      }
+  ]
+}

functions/set-retention/handler.js

@@ -0,0 +1,28 @@
+'use strict';
+
+const co             = require('co');
+const Promise        = require('bluebird');
+const AWS            = require('aws-sdk');
+const cloudWatchLogs = Promise.promisifyAll(new AWS.CloudWatchLogs());
+const retentionDays  = 30;
+
+let setExpiry = function* (logGroupName) {
+  let params = {
+    logGroupName    : logGroupName,
+    retentionInDays : retentionDays
+  };
+
+  yield cloudWatchLogs.putRetentionPolicyAsync(params);
+};
+
+module.exports.handler = co.wrap(function* (event, context, callback) {
+  console.log(JSON.stringify(event));
+
+  let logGroupName = event.detail.requestParameters.logGroupName;
+  console.log(`log group: ${logGroupName}`);
+
+  yield setExpiry(logGroupName);
+  console.log(`updated [${logGroupName}]'s retention policy to ${retentionDays} days`);
+
+  callback(null, 'ok');
+});

functions/ship-logs/handler.js

@@ -0,0 +1,9 @@
+'use strict';
+
+const co = require('co');
+const processAll = require('./lib');
+
+module.exports.handler = co.wrap(function* (event, context, callback) {
+  yield processAll(event.logGroup, event.logStream, event.logEvents);
+  callback(null, `Successfully processed ${event.logEvents.length} log events.`);
+});

functions/ship-logs/lib.js

@@ -0,0 +1,46 @@
+'use strict';
+
+const co      = require('co');
+const Promise = require('bluebird');
+const parse   = require('./parse');
+const tls     = require('tls');
+
+let processAll = co.wrap(function* (logGroup, logStream, logEvents) {
+  let lambdaVer = parse.lambdaVersion(logStream);
+
+  let tlsOptions = {
+    host: process.env.logstash_host,
+    port: process.env.logstash_port
+  };
+
+  yield new Promise((resolve, reject) => {
+    let socket = tls.connect(tlsOptions, function() {
+      console.log(`connected to ${process.env.logstash_host}:${process.env.logstash_port}`);
+      socket.setEncoding('utf8');
+
+      for (let logEvent of logEvents) {
+        try {
+          let log = parse.logMessage(logEvent.message);
+          log.level         = log.level || 'debug';
+          log.logStream     = logStream;
+          log.logGroup      = logGroup;
+          log.lambdaVersion = lambdaVer;
+          log.fields        = log.fields || {};
+          log.type          = "cloudwatch";
+          log['@timestamp'] = new Date(logEvent.timestamp);
+
+          console.log("sending : ", log);
+          socket.write(JSON.stringify(log) + '\n');
+        } catch (err) {
+          console.log(err, err.stack);
+        }
+      }
+
+      socket.end();
+
+      resolve();
+    });
+  });
+});
+
+module.exports = processAll;

functions/ship-logs/parse.js

@@ -0,0 +1,88 @@
+'use strict';
+
+// logGroup looks like this:
+//    "logGroup": "/aws/lambda/service-env-funcName"
+let functionName = function (logGroup) {
+  return logGroup.split('/').reverse()[0];
+};
+
+// logStream looks like this:
+//    "logStream": "2016/08/17/[76]afe5c000d5344c33b5d88be7a4c55816"
+let lambdaVersion = function (logStream) {
+  let start = logStream.indexOf('[');
+  let end = logStream.indexOf(']');
+  return logStream.substring(start+1, end);
+};
+
+let requestIdFromSysLog = message => {
+  let idx = message.indexOf('RequestId: ');
+
+  // funny, JS string has multiple substring methods..
+  // substr takes starting index & length
+  // https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Global_Objects/String/substr
+  // NOTE: "RequestId: " is 11 chars long, hence `+ 11` below, and request ID is
+  // always a 36 char guid
+  return message.substr(idx + 11, 36);
+};
+
+let isDate = function (str) {
+  return !isNaN(Date.parse(str));
+}
+
+// a typical API Gateway log message looks like this:
+//    "2017-04-26T10:41:09.023Z	db95c6da-2a6c-11e7-9550-c91b65931beb\tloading index.html...\n"
+// but there are START, END and REPORT messages too:
+//    "START RequestId: 67c005bb-641f-11e6-b35d-6b6c651a2f01 Version: 31\n"
+//    "END RequestId: 5e665f81-641f-11e6-ab0f-b1affae60d28\n"
+//    "REPORT RequestId: 5e665f81-641f-11e6-ab0f-b1affae60d28\tDuration: 1095.52 ms\tBilled Duration: 1100 ms \tMemory Size: 128 MB\tMax Memory Used: 32 MB\t\n"
+let logMessage = function (message) {
+  let type;
+  if (message.startsWith('START RequestId') ||
+      message.startsWith('END RequestId') ||
+      message.startsWith('REPORT RequestId')) {
+    type = 'SYS';
+  } else {
+    type = 'LOG';
+  }
+
+  if (type === 'SYS') {
+    console.log('system message: ', message);
+    let requestId = requestIdFromSysLog(message);
+    return { 
+      level  : 'debug',
+      fields : { requestId }, 
+      message };
+  }
+
+  console.log('non-system message: ', message);
+
+  let parts = message.split('\t', 3);
+
+  // likely API Gateway log message
+  if (parts.length === 3 && isDate(parts[0])) {
+    let timestamp  = parts[0];
+    let requestId  = parts[1];
+    let logMessage = parts[2];
+
+    return { 
+      level   : 'debug',
+      message : logMessage,
+      fields  : {
+        '@timestamp' : timestamp,
+        requestId
+      }
+    };
+  }
+
+  return {
+    level   : 'debug',
+    message : message
+  };
+};
+
+module.exports = {
+  functionName,
+  lambdaVersion,
+  requestIdFromSysLog,
+  logMessage
+};

functions/subscribe/handler.js

@@ -0,0 +1,30 @@
+'use strict';
+
+const co             = require('co');
+const Promise        = require('bluebird');
+const AWS            = require('aws-sdk');
+const cloudWatchLogs = Promise.promisifyAll(new AWS.CloudWatchLogs());
+const destinationArn = process.env.DEST_FUNC;
+
+let subscribe = function* (logGroupName) {
+  let options = {
+    destinationArn : destinationArn,
+    logGroupName   : logGroupName,
+    filterName     : 'ship-logs',
+    filterPattern  : ''
+  };
+
+  yield cloudWatchLogs.putSubscriptionFilterAsync(options);
+};
+
+module.exports.handler = co.wrap(function* (event, context, callback) {
+  console.log(JSON.stringify(event));
+
+  let logGroupName = event.detail.requestParameters.logGroupName;
+  console.log(`log group: ${logGroupName}`);
+
+  yield subscribe(logGroupName);
+  console.log(`subscribed [${logGroupName}] to [${destinationArn}]`);
+
+  callback(null, 'ok');
+});

package.json

@@ -0,0 +1,20 @@
+{
+  "name": "lambda-logging-demo",
+  "version": "1.0.0",
+  "description": "",
+  "main": "index.js",
+  "scripts": {
+    "test": "echo \"Error: no test specified\" && exit 1"
+  },
+  "author": "",
+  "license": "ISC",
+  "dependencies": {
+    "bluebird": "^3.5.0",
+    "co": "^4.6.0"
+  },
+  "devDependencies": {
+    "aws-sdk": "^2.100.0",
+    "serverless": "^1.20.2",
+    "serverless-pseudo-parameters": "^1.1.5"
+  }
+}

serverless.yml

@@ -0,0 +1,67 @@
+service: logging-demo
+
+plugins:
+ - serverless-pseudo-parameters
+
+provider:
+  name: aws
+  runtime: nodejs6.10
+  stage: dev
+  region: us-east-1
+
+  iamRoleStatements:
+    - Effect: "Allow"
+      Action:
+        - "logs:PutRetentionPolicy"
+        - "logs:PutSubscriptionFilter"
+      Resource: "*"
+
+functions:
+  ship-logs:
+    handler: functions/ship-logs/handler.handler
+    description: Sends CloudWatch logs to LogStash
+
+  set-retention:
+    handler: functions/set-retention/handler.handler
+    description: Sets the log retention policy to 30 days
+    events:
+      - cloudwatchEvent:
+          event:
+            source:
+              - aws.logs
+            detail-type:
+              - AWS API Call via CloudTrail
+            detail:
+              eventSource:
+                - logs.amazonaws.com
+              eventName:
+                - CreateLogGroup
+
+  subscribe:
+    handler: functions/subscribe/handler.handler
+    description: Subscribe logs to the ship-log function
+    environment:
+      DEST_FUNC: "arn:aws:lambda:#{AWS::Region}:#{AWS::AccountId}:function:${self:service}-${self:provider.stage}-ship-logs"
+    events:
+      - cloudwatchEvent:
+          event:
+            source:
+              - aws.logs
+            detail-type:
+              - AWS API Call via CloudTrail
+            detail:
+              eventSource:
+                - logs.amazonaws.com
+              eventName:
+                - CreateLogGroup
+resources:
+  Resources:
+    LambdaInvokePermission:
+      Type: AWS::Lambda::Permission
+      Properties:
+        Action: lambda:InvokeFunction
+        FunctionName:
+          Fn::Join:
+            - ""
+            - - "Ref" : "ShipDashlogsLambdaFunction"
+        Principal: logs.#{AWS::Region}.amazonaws.com