Add GELF Input Plugin

.jshintrc

@@ -81,6 +81,8 @@
     "worker"        : false,    // Web Workers
     "wsh"           : false,    // Windows Scripting Host
     "yui"           : false,    // Yahoo User Interface
+    "esversion"     : 6,        // Disable JSHint from unnecessary warnings when code use ES6 syntax 
+
 
     // Custom Globals
     "globals"       : {}        // additional predefined global variables

bin/logagent.js

@@ -38,6 +38,9 @@ var moduleAlias = {
   'input-influxdb-http': '../lib/plugins/input/influxHttp.js',
   'elasticsearch-query': '../lib/plugins/input/elasticsearchQuery.js',
   'input-elasticsearch-http': '../lib/plugins/input/elasticsearchHttp.js',
+  'input-gelf': '../lib/plugins/input/gelf.js',
+  'input-cloudfoundry': '../lib/plugins/input/cloudfoundry.js',
+  'input-heroku': '../lib/plugins/input/heroku.js',
 
   // input filters
   grep: '../lib/plugins/input-filter/grep.js',
@@ -62,6 +65,7 @@ function LaCli (options) {
   this.la = null
   this.throng = null
   this.argv = options || require('../lib/core/cliArgs.js')
+
   this.globPattern = this.argv.glob || process.env.GLOB_PATTERN
   this.logseneToken = this.argv.index || process.env.LOGSENE_TOKEN
   this.loggers = {}
@@ -107,6 +111,7 @@ LaCli.prototype.initPlugins = function (plugins) {
       if (plugin.config) {
         plugin.config.configFile = plugin.globalConfig
       }
+
       var p = new Plugin(plugin.config || this.argv, eventEmitter)
       this.plugins.push(p)
       p.start()

config/examples/cloudfoundry-elasticsearch.yml

@@ -3,8 +3,9 @@ options:
   includeOriginalLine: true
 
 input:
-  cloudFoundry: 
-   port: 9999
+  cloudFoundry:
+    module: input-cloudfoundry
+    port: 9999
 
 # we use here inline patterns, loaded after patterns.yml 
 parser:

config/examples/gelf-stdout.yml

@@ -0,0 +1,9 @@
+# Global options
+input:
+  gelf: 
+    module: input-gelf
+    port: 12100
+    bindAddress: 0.0.0.0
+output:
+  # print parsed logs in YAML format to stdout   
+  stdout: yaml 

config/examples/heroku-stdout-yml.yml

@@ -3,7 +3,8 @@ options:
   includeOriginalLine: true
 
 input:
-  heroku: 
+  heroku:
+   module: input-heroku
    port: 9999
    worker: 4
    blacklist:

lib/core/configLoader.js

@@ -53,8 +53,6 @@ function convertToCliArgs (cfg, argv) {
   argv.configFile = cfg
   if (cfg.input) {
     setProperty(flatCfg, argv, 'input.syslog.port', 'udp')
-    setProperty(flatCfg, argv, 'input.heroku.port', 'heroku')
-    setProperty(flatCfg, argv, 'input.cloudFoundry.port', 'cfhttp')
     if (cfg.input.files && cfg.input.files.length > 1) {
       argv.glob = '{' + cfg.input.files.join(',') + '}'
     }

lib/plugins/input/cloudfoundry.js

@@ -4,13 +4,15 @@ var http = require('http')
 function InputCloudFoundry (config, eventEmitter) {
   this.config = config
   this.eventEmitter = eventEmitter
-  if (config.configFile.input.cloudFoundry && config.configFile.input.cloudFoundry.blacklist) {
-    this.config.blacklist = config.configFile.input.cloudFoundry.blacklist
+  this.config.cfhttp = config.cfhttp || config.port
+  consoleLogger.log('value '+this.config.cfhttp )
+  if (config && config.blacklist) {
+    this.config.blacklist = config.blacklist
   } else {
     this.config.blacklist = {}
   }
-  if (config.configFile.input.cloudFoundry.workers) {
-    this.config.cloudFoundryWorkers = config.configFile.input.heroku.workers
+  if (config.workers) {
+    this.config.cloudFoundryWorkers = config.workers
   } else {
     this.config.cloudFoundryWorkers = undefined
   }
@@ -64,9 +66,6 @@ InputCloudFoundry.prototype.cloudFoundryHandler = function (req, res) {
     if (path.length > 1) {
       if (path[1] && path[1].length > 31 && /[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}/.test(path[1])) {
         token = path[1]
-      } else {
-        // console.log('Bad Url: ' + path)
-        // console.log(JSON.stringify(req.headers))
       }
     }
     if (!token) {
@@ -75,7 +74,7 @@ InputCloudFoundry.prototype.cloudFoundryHandler = function (req, res) {
       return
     }
     if (self.config && self.config.blacklist && self.config.blacklist[token]) {
-      if (self.config.configFile.input.heroku.debug) {
+      if (self.config.debug) {
         consoleLogger.log('blacklisted request for' + token)
       }
       res.statusCode = 404

lib/plugins/input/command.js

@@ -1,6 +1,5 @@
 'use strict'
 var split = require('split2')
-var momenttz = require('moment-timezone')
 
 function InputCommand (config, eventEmitter) {
   this.config = config // config.configFile.input.command

lib/plugins/input/elasticsearchQuery.js

@@ -4,7 +4,7 @@ var elasticsearch = require('elasticsearch')
 
 /**
  * Constructor called by logagent
- * @config cli arguments and config.configFile entries
+ * @config cli arguments and config entries
  * @eventEmitter logent eventEmitter object
  */
 function InputElasticsearchQuery (config, eventEmitter) {

lib/plugins/input/gelf.js

@@ -0,0 +1,36 @@
+'use strict'
+
+var gelfserver = require('graygelf/server')
+var safeStringify = require('fast-safe-stringify')
+var consoleLogger = require('../../util/logger.js')
+
+
+function InputGELF (config, eventEmitter) {
+  this.config = config
+  this.config.port = config.port || 12100
+  this.config.bindAddress = config.bindAddress || '0.0.0.0'
+  this.eventEmitter = eventEmitter
+}
+module.exports = InputGELF
+
+
+InputGELF.prototype.start = function () {
+  if (!this.started) {
+    this.createServer()
+    this.started = true
+  }
+}
+
+InputGELF.prototype.stop = function (cb) {
+  this.server.close(cb)
+}
+
+InputGELF.prototype.createServer = function () {
+  var self = this
+  this.server = gelfserver();
+  this.server.on('message', function (gelf) {
+    //At the moment gelf server doesn't expose rinfo
+    self.eventEmitter.emit('data.raw', safeStringify(gelf), {sourceName: 'input-gelf ',source: this.config.bindAddress })
+  })  
+  this.server.listen(this.config.port ,this.config.bindAddress)
+}

lib/plugins/input/heroku.js

@@ -3,21 +3,21 @@ var http = require('http')
 
 function InputHeroku (config, eventEmitter) {
   this.config = config
-  console.log(config.configFile.input.heroku)
   this.eventEmitter = eventEmitter
-  if (config.configFile.input.heroku && config.configFile.input.heroku.blacklist) {
-    this.config.blacklist = config.configFile.input.heroku.blacklist
+  this.config.port = this.config.heroku || config.port
+  if (config && config.blacklist) {
+    this.config.blacklist = config.blacklist
   } else {
     this.config.blacklist = {}
   }
-  if (config.configFile.input.heroku.workers) {
-    this.config.herokuWorkers = config.configFile.input.heroku.workers
+  if (config.workers) {
+    this.config.herokuWorkers = config.workers
   } else {
     this.config.herokuWorkers = undefined
   }
 }
 InputHeroku.prototype.start = function () {
-  if (this.config.heroku) {
+  if (this.config.port) {
     this.throng = require('throng')
     this.throng({
       workers: this.config.herokuWorkers || this.WORKERS || 2,
@@ -79,9 +79,6 @@ InputHeroku.prototype.herokuHandler = function (req, res) {
     if (path.length > 1) {
       if (path[1] && path[1].length > 31 && /[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}/.test(path[1])) {
         token = path[1]
-      } else {
-        // console.log('Bad Url: ' + path)
-        // console.log(JSON.stringify(req.headers))
       }
     }
     if (!token) {
@@ -90,7 +87,7 @@ InputHeroku.prototype.herokuHandler = function (req, res) {
       return
     }
     if (self.config && self.config.blacklist && self.config.blacklist[token]) {
-      if (self.config.configFile.input.heroku.debug) {
+      if (self.config.debug) {
         consoleLogger.log('blacklisted request for' + token)
       }
       res.statusCode = 404
@@ -114,11 +111,10 @@ InputHeroku.prototype.herokuHandler = function (req, res) {
     consoleLogger.error('Error in Heroku (http): ' + err)
   }
 }
-
 // heroku start function for WEB_CONCURENCY
 InputHeroku.prototype.startHerokuServer = function (id) {
   consoleLogger.log('start heroku worker: ' + id + ', pid:' + process.pid)
-  this.getHttpServer(Number(this.config.heroku), this.herokuHandler.bind(this))
+  this.getHttpServer(Number(this.config.port), this.herokuHandler.bind(this))
   var exitInProgress = false
   var terminate = function (reason) {
     return function () {

lib/plugins/input/tcp.js

@@ -12,12 +12,12 @@ var safeStringify = require('fast-safe-stringify')
  *    bindAddress: 0.0.0.0
  *    returnResult: false
  *
- * @config cli arguments and config.configFile entries
+ * @config cli arguments and config entries
  * @eventEmitter logent eventEmitter object
  */
 function InputTCP (config, eventEmitter) {
-  this.config = config.configFile.input.tcp
-  this.config.maxInputRate = config.configFile.input.tcp.maxInputRate || config.maxInputRate
+  this.config = config
+  this.config.maxInputRate = config.maxInputRate || config.maxInputRate
   this.eventEmitter = eventEmitter
 }
 module.exports = InputTCP

package.json

@@ -66,6 +66,7 @@
     "flat": "^2.0.1",
     "flatstr": "^1.0.4",
     "glob": "^7.1.1",
+    "graygelf": "^1.0.2",
     "http-aws-es": "^3.1.0",
     "influx-line-protocol-parser": "^0.2.0",
     "js-yaml": "^3.8.1",

test/gelf-producer.js

@@ -0,0 +1,6 @@
+var log = require('graygelf')({
+    host: 'localhost',
+    port: 12100
+  })
+
+log.info.a('short', 'full', { foo: 'bar' })