MySQL Filter

docs/filters/mysql.md

@@ -0,0 +1,34 @@
+MySQL filter
+---
+
+Status : experimental plugin, unit tested and maintained.
+
+The MySQL filter can be used to resolve fields from a mysql database, caching the results for sequential reads.
+
+Config using logstash format:
+````
+filter {
+  mysql {
+    db => 'db_name'
+    user => 'root'
+    password => 'secret'
+    host => 'localhost'
+    query => 'SELECT * from some_table WHERE id=(?) limit 1;'
+    source_field => 'to_user'
+    target_field => 'my_lookup'
+  }
+}
+````
+
+Parameters:
+
+* ``db``: name of the database used for lookups.
+* ``query``: SELECT query with single parameter matching.
+* ``filter``: override filter for lookups, can be used to extract as #{toto}.
+* ``user``: username for the connection. 
+* ``password``: password for the connection. 
+* ``host``: server host to connect to.
+* ``port``: server port to connect to.
+* ``pair``: return results as key: value pairs on multiple rows.
+* ``source_field``: which field to work on for the lookup.
+* ``target_field``: field to store the result. Default: field used for resolution.

lib/filters/filter_mysql.js

@@ -0,0 +1,92 @@
+/* 
+   PASTASH MySQL Filter
+*/
+
+var base_filter = require('../lib/base_filter'),
+  util = require('util'),
+  logger = require('log4node');
+
+var mysql = require('mysql2');
+var sqdb;
+
+function FilterMySql() {
+  base_filter.BaseFilter.call(this);
+  this.mergeConfig({
+    name: 'MySql',
+    optional_params: ['db', 'table', 'query', 'source_field', 'target_field', 'filter', 'host', 'user', 'password', 'port', 'pair'],
+    default_values: {
+      	'host': 'localhost',
+	'port': 3306,
+	'user': 'root',
+	'password': '',
+      	'target_field': 'mysql'
+    },
+    start_hook: this.start,
+  });
+}
+
+util.inherits(FilterMySql, base_filter.BaseFilter);
+
+FilterMySql.prototype.start = function(callback) {
+
+  if (this.db) { 
+	try {
+	  var cfg = { database: this.db, rowsAsArray: true };
+	  if(this.host) cfg.host = this.host;
+	  if(this.user) cfg.user = this.user;
+	  if(this.password) cfg.password = this.password;
+	  sqdb = mysql.createConnection(cfg);
+	  logger.info('Initializing Filter MySql:',this.db);
+	} catch(e){ logger.error('Failed Initializing Filter MySql',e); }
+  }
+  logger.info('Initialized Filter MySql');
+  callback();
+};
+
+FilterMySql.prototype.process = function(raw) {
+   if (!sqdb||!this.query) return raw;
+   if (!this.source_field && !this.filter) return raw;
+
+   try {
+
+	if ( !this.filter && this.source_field ){
+		this.filter = raw[this.source_field];
+	}
+
+	if (this.db) {
+		sqdb.execute(this.query, this.filter, function(err, results, fields) {
+			if (err||!results) this.emit('output', raw);
+			else {
+				console.log('ROW:',results);
+				if (this.pair){
+					raw[this.target_field] = {};
+					results.forEach(function(block){
+						var even = false;
+						block.forEach(function(item){
+							if (item && even){ 
+								raw[this.target_field][even] = item;
+								even = false;
+							} else { even = item; }
+						}.bind(this));
+					}.bind(this));
+
+				} else {
+					raw[this.target_field] = results[0];
+				}
+				this.emit('output', raw);
+			}
+		}.bind(this));	
+	} else { return raw; }
+
+   } catch(e) { logger.info('failed processing mysql!',e); return raw; }
+
+};
+
+FilterMySql.prototype.close = function(callback) {
+  logger.info('Closing Filter MySql');
+  callback();
+};
+
+exports.create = function() {
+  return new FilterMySql();
+};

package.json

@@ -60,7 +60,7 @@
     "optimist": "0.6.1"
   },
   "optionalDependencies": {
-    "@qxip/bencode": "1.0.0",
+   "@qxip/bencode": "1.0.0",
     "amqplib": "0.5.1",
     "asterisk.io": "git+https://git@github.com/lmangani/asterisk.io.git",
     "aws-sdk": "2.112.x",
@@ -79,6 +79,7 @@
     "metrics-influxdb": "1.0.0",
     "modesl": "*",
     "moment": "2.19.3",
+    "mysql2":"1.4.2",
     "mqtt": "*",
     "msgpack": "1.0.x",
     "mustache": "2.3.0",