Datadog Output and Regex URL Filter

Readme.markdown

@@ -480,6 +480,24 @@ Parameters:
 
 Example: ``metric_key=nginx.response.#{status}``
 
+Datadog
+---
+
+This plugin is used send data to the datadog statsd which has some extended functionality (like tags) compared to statsd.
+
+Example: ``output://datadog://localhost:8125?only_type=nginx&metric_type=increment&metric_key=nginx.request&tags=server:web1``, to send, for each line of nginx log, a counter with value 1, key ``nginx.request``, on a datadog statsd instance located on port 8125.
+
+Parameters:
+
+* ``metric_type``: one of ``increment``, ``decrement``, ``counter``, ``timer``, ``gauge``, ``histogram``, ``set``. Type of value to send to datadog statsd.
+* ``metric_key``: key to send to statsd.
+* ``metric_value``: metric value to send to statsd. Mandatory for ``timer``, ``counter``, ``gauge``, ``histogram`` and ``set`` type.
+* ``tags``: tags you want to send to datadog. Example: &tags=status:200,server:web1
+
+``metric_key``, ``metric_value`` and ``tags`` can reference log line properties (see above).
+
+Example: ``metric_key=nginx.response.#{status}``
+
 Gelf
 ---
 
@@ -678,6 +696,32 @@ Parameters:
 
 Note: fields with empty values will not be set.
 
+Regex URL
+---
+
+The url regex filter is the same as the regex filter but has additional parameters to parse the GET params from urls (url.parse() is used).
+
+Example 1: ``filter://regex_url://?regex=^(\S)+ &fields=toto``, to extract the first word of a line of logs, and place it into the ``toto`` field.
+
+Example 2: ``filter://regex_url://http_combined?only_type=nginx``, to extract fields following configuration into the http_combined pattern. node-logstash is bundled with [some configurations](https://github.com/bpaquet/node-logstash/tree/master/lib/patterns). You can add your custom patterns directories, see options ``--patterns_directories``.
+
+Example 3: ``filter://regex_url://?regex=(\d+|-)&fields=a&numerical_fields=a``, to force number extraction. If the matched string is not a number but ``-``, the field ``a`` will not be set.
+
+Example 4: ``filter://regex_url://http_combined?only_type=nginx&request_fields=url&request_fields_list=param1,param2&request_prefix=req_``, to force number extraction. If the matched string is not a number but ``-``, the field ``a`` will not be set.
+
+Parameters:
+
+* ``regex``: regex to apply.
+* ``regex_flags: regex flags (eg : g, i, m).
+* ``fields``: name of fields which will receive the pattern extracted (see below for the special field @timestamp).
+* ``numerical_fields``: name of fields which have to contain a numerical value. If value is not numerical, field will not be set.
+* ``request_fields``: name of fields which should parsed with url.parse() - so every GET param in these fields are own fields afterwards (with request_prefix).
+* ``request_fields_list``: whitelist of request_fields - so if this is given only params in this will result in own fields if they are parsed within request_fields.
+* ``request_prefix``: prefix of the request param fields (?param1=foo&param2=bar results in req_param1:foo,req_param2:bar with the request_prefix set to req_)
+* ``date_format``: if ``date_format` is specified and a ``@timestamp`` field is extracted, the filter will process the data extracted with the date\_format, using [moment](http://momentjs.com/docs/#/parsing/string-format/). The result will replace the original timestamp of the log line.
+
+Note: fields with empty values will not be set.
+
 Mutate replace
 ---
 

lib/filters/filter_regex_url.js

@@ -0,0 +1,132 @@
+var base_filter = require('../lib/base_filter'),
+  url = require('url'),
+  util = require('util'),
+  logger = require('log4node'),
+  patterns_loader = require('../lib/patterns_loader'),
+  moment = require('moment');
+
+function FilterRegexUrl() {
+  base_filter.BaseFilter.call(this);
+  this.mergeConfig({
+    name: 'Regex',
+    host_field: 'pattern_field',
+    allow_empty_host: true,
+    required_params: ['regex'],
+    optional_params: ['fields', 'numerical_fields', 'request_fields', 'request_fields_list', 'date_format', 'regex_flags', 'request_prefix'],
+    default_values: {
+      'fields': '',
+      'numerical_fields': '',
+	  'request_fields': '',
+	  'request_fields_list': '',
+	  'request_prefix': '',
+    },
+    config_hook: this.loadPattern,
+    start_hook: this.start,
+  });
+}
+
+util.inherits(FilterRegexUrl, base_filter.BaseFilter);
+
+FilterRegexUrl.prototype.loadPattern = function(callback) {
+  if (this.pattern_field) {
+    logger.info('Try to load config from pattern file ' + this.pattern_field);
+    patterns_loader.load(this.pattern_field, function(err, config) {
+      if (err) {
+        return callback(new Error('Unable to load pattern : ' + this.pattern_field + ' : ' + err));
+      }
+      for (var i in config) {
+        this[i] = config[i];
+      }
+      callback();
+    }.bind(this));
+  }
+  else {
+    callback();
+  }
+};
+
+FilterRegexUrl.prototype.start = function(callback) {
+  this.regex = new RegExp(this.regex, this.regex_flags);
+  this.date_format = this.date_format;
+
+  this.fields = this.fields.split(',');
+  this.numerical_fields = this.numerical_fields.split(',');
+  this.request_fields = this.request_fields.split(',');
+  if (this.request_fields_list){
+	this.request_fields_list = this.request_fields_list.split(',');
+  }
+
+  logger.info('Initializing regex_url filter, regex : ' + this.regex + ', fields ' + this.fields + (this.date_format ? ', date format ' + this.date_format : '') + ', flags: ' + (this.regex_flags || '') + ', request_fields: ' + this.request_fields + ', request_fields_list: ' + this.request_fields_list);
+
+  callback();
+};
+
+FilterRegexUrl.prototype.checkValue = function(v, key){
+	if (v.match(/^[0-9]+$/)) {
+		v = parseInt(v);
+	}
+	else if (v.match(/^[0-9]+[\.,][0-9]+$/)) {
+		v = parseFloat(v.replace(',', '.'));
+	}
+	else {
+		if (key && this.numerical_fields.indexOf(key) !== -1) {
+			v = undefined;
+		}
+	}
+	return v;
+};
+
+
+FilterRegexUrl.prototype.process = function(data) {
+  logger.debug('Trying to match on regex', this.regex, ', input', data.message);
+  var result = data.message.match(this.regex);
+  logger.debug('Match result:', result);
+  if (result) {
+    for (var i = 0; i < this.fields.length; i++) {
+      var v = result[i + 1];
+      if (v) {
+        if (this.date_format && (this.fields[i] === 'timestamp' || this.fields[i] === '@timestamp')) {
+          var m = moment(v, this.date_format);
+          if (m.year() + m.month() + m.date() + m.hours() + m.minutes() + m.seconds() > 1) {
+            if (m.year() === 0) {
+              m.year(moment().year());
+            }
+            data['@timestamp'] = m.format('YYYY-MM-DDTHH:mm:ss.SSSZZ');
+            logger.debug('Event timestamp modified to', data['@timestamp']);
+          }
+        }
+        else if (this.fields[i] === 'host') {
+          data.host = v;
+        }
+        else if (this.fields[i] === 'message') {
+          data.message = v;
+        }
+        else {
+	      if (this.request_fields.indexOf(this.fields[i]) !== -1){
+		    try {
+		      var url_parts = url.parse(v, true);
+			  for (var attrname in url_parts.query) {
+			    if (!this.request_fields_list || this.request_fields_list.indexOf(attrname) !== -1){
+				  v = this.checkValue(v);
+				  data[this.request_prefix+attrname] = url_parts.query[attrname];
+				}
+			  }
+		    }
+		    catch (err){
+			    // we just do nothing here as we couldn't parse the url
+		    }
+	      }
+	      v = this.checkValue(v, this.fields[i]);
+        }
+        if (v !== undefined) {
+          data[this.fields[i]] = v;
+        }
+      }
+    }
+  }
+  return data;
+};
+
+exports.create = function() {
+  return new FilterRegexUrl();
+};

lib/outputs/output_datadog.js

@@ -0,0 +1,83 @@
+var abstract_udp = require('./abstract_udp'),
+  util = require('util'),
+  logger = require('log4node');
+
+function OutputDatadog() {
+  abstract_udp.AbstractUdp.call(this);
+  this.mergeConfig({
+    name: 'Datadog',
+    required_params: ['metric_type', 'metric_key'],
+    optional_params: ['metric_value','tags'],
+    start_hook: this.start
+  });
+}
+
+util.inherits(OutputDatadog, abstract_udp.AbstractUdp);
+
+OutputDatadog.prototype.start = function(callback) {
+  if (this.metric_type === 'counter') {
+    if (!this.metric_value) {
+      return callback(new Error('You have to specify metric_value with metric_type counter'));
+    }
+    this.raw = this.metric_key + ':' + this.metric_value + '|c';
+  }
+  else if (this.metric_type === 'increment') {
+    this.raw = this.metric_key + ':1|c';
+  }
+  else if (this.metric_type === 'decrement') {
+    this.raw = this.metric_key + ':-1|c';
+  }
+  else if (this.metric_type === 'timer') {
+    if (!this.metric_value) {
+      return callback(new Error('You have to specify metric_value with metric_type timer'));
+    }
+    this.raw = this.metric_key + ':' + this.metric_value + '|ms';
+  }
+  else if (this.metric_type === 'gauge') {
+	  if (!this.metric_value) {
+		  return callback(new Error('You have to specify metric_value with metric_type gauge'));
+	  }
+	  this.raw = this.metric_key + ':' + this.metric_value + '|g';
+  }
+  else if (this.metric_type === 'histogram') {
+	  if (!this.metric_value) {
+		  return callback(new Error('You have to specify metric_value with metric_type histogram'));
+	  }
+	  this.raw = this.metric_key + ':' + this.metric_value + '|h';
+  }
+  else if (this.metric_type === 'set') {
+	  if (!this.metric_value) {
+		  return callback(new Error('You have to specify metric_value with metric_type histogram'));
+	  }
+	  this.raw = this.metric_key + ':' + this.metric_value + '|s';
+  }
+  else {
+    return callback(new Error('Wrong metric_type: ' + this.metric_type));
+  }
+
+  if (this.raw && this.tags){
+    this.raw = this.raw + '|#' + this.tags;
+  }
+
+  callback();
+};
+
+OutputDatadog.prototype.formatPayload = function(data, callback) {
+  var raw = this.replaceByFields(data, this.raw);
+  if (raw) {
+    logger.debug('Send to datadog packet', raw);
+    var message = new Buffer(raw);
+    callback(message);
+  }
+  else {
+    logger.debug('Unable to replace fields on', this.raw, 'input', data);
+  }
+};
+
+OutputDatadog.prototype.to = function() {
+  return ' datadog ' + this.host + ':' + this.port + ', metric_type ' + this.metric_type + ', metric_key ' + this.metric_key + ', tags: ' + this.tags;
+};
+
+exports.create = function() {
+  return new OutputDatadog();
+};