Ship Kinesis events to Logstash

handler.js

const Zlib = require('zlib');
const Logstash = require('logstash-client');

const lambdaVersion = function (logStream) {
    let start = logStream.indexOf('[');
    let end = logStream.indexOf(']');
    return logStream.substring(start+1, end);
}

exports.handler = async (event, context) => {
  const logstashClient = new Logstash({
        type: 'udp',
        host: process.env.LOGSTASH_HOST,
        port: process.env.LOGSTASH_PORT
  });
  event.Records.forEach(record => {
      const payload = new Buffer(record.kinesis.data, 'base64');
      const json = (Zlib.gunzipSync(payload)).toString('utf8');
      const data = JSON.parse(json);

      const functionName = data.logGroup.split('/').reverse()[0];
      const functionVersion = lambdaVersion(data.logStream);
      const awsRegion = record.awsRegion;

      data.logEvents.forEach((event) => {
        if (!(event.message.startsWith('START RequestId') ||
            event.message.startsWith('END RequestId') ||
            event.message.startsWith('REPORT RequestId'))) {
          logstashClient.send({
              message: event.extractedFields.event,
              function: functionName,
              version: functionVersion,
              region: awsRegion,
          });
        }
      })
  })
  return 'done';
}