tag moloch sessions with suricata eve.json alerts

suricata_tagger.js

var fs = require('fs'); 
var byline = require('byline');
var urllib = require('urllib');
var fileName = '/home/vagrant/log/eve.json';
var stream = fs.createReadStream(fileName);
stream = byline.createStream(stream);
stream.on('data', function(linebuf) {
  var suricataEvent = null; 
  try {
    suricataEvent = JSON.parse(linebuf.toString());
  } catch (e) {
    console.error('file is not JSON :: ',e);
    process.exit(1);
  }
  var type = suricataEvent.event_type || null;
  if (type === 'alert') {
    //console.log(suricataEvent);
    stream.pause();
    var time = new Date(suricataEvent.timestamp).getTime();
    if (time === NaN) {
      console.error('not time ::', suricataEvent);
      process.exit(1);
    }
    var src_port = parseInt(suricataEvent.src_port);
    var dest_port = parseInt(suricataEvent.dest_port);
    var src_ip = suricataEvent.src_ip;
    var dest_ip = suricataEvent.dest_ip; 
    var tags = suricataEvent.alert.signature.replace(/ /g,"_");
    urllib.request('http://localhost:8005/addTags?startTime='+((time/1000)-1)+'&stopTime='+((time/1000)+1)+'&expression=ip.src%3D'+src_ip+'%26%26port.src%3D'+src_port+'%26%26ip.dst%3D'+dest_ip+'%26%26port.dst%3D'+dest_port, {
        method: 'POST',
        digestAuth: 'admin:admin',
        contentType: 'json',
        data: {'tags': tags}
      },  function (err, data, res) {
        console.log(data.toString());
        stream.resume();
      });
  }
});