BustleEventLog.js

var elasticsearch = require('elasticsearch');
var client = new elasticsearch.Client();
var eventPattern = /([^ ]*) ([^ ]*) ([^ ]*) ([^ ]*) ([^ ]*) ([^ ]*) ([^ ]*) \"(.*?)\" \"(.*?)\" \"(.*?)\" \"(.*?)\" \"(.*?)\"/;

exports.handler = function(data, context) {
  var events = [];
  var encodedEvent, rawEvent, 
      matchedEvent, event, i,
      commands;  

  console.log('data', JSON.stringify(data, null, '  '));

  try {
    for(i = 0; i < data.Records.length; ++i) {
      encodedEvent  = data.Records[i].kinesis.data;
      rawEvent      = new Buffer(encodedEvent, 'base64').toString();
      matchedEvent  = rawEvent.match(eventPattern);

      events.push({
        time       : new Date(parseInt(matchedEvent[1])),
        key        : matchedEvent[2],
        ip         : matchedEvent[3],
        session    : matchedEvent[4],
        is_mobile  : matchedEvent[5] === 'true',
        is_desktop : matchedEvent[6] === 'true',
        browser    : matchedEvent[8],
        version    : matchedEvent[9],
        os         : matchedEvent[10],
        platform   : matchedEvent[11],
        referer    : matchedEvent[12]
      });
    }

    console.log('events', JSON.stringify(events, null, '  '));

    commands = events.reduce(function(bulk, e) {
      return bulk.concat({ 
        index: { 
          _index: 'event_log', 
          _type: 'event' 
        } 
      }, e);
    }, []);

    console.log('commands', JSON.stringify(commands, null, '  '));

    client.bulk({ 
      body: commands 
    }, function(err, resp) {
      console.log('bulk', err, resp);
      context.done(null, 'success');
    });

  } catch(error) {
    console.log('error', error);
    context.done(null, 'fail');
  };
};