InfluxDB version 0.9 API support

Signed-off-by: imre Fitos <imre@act.md>

Author: imreACTmd

README.md

@@ -47,6 +47,7 @@ You can configure the following settings in your StatsD config file.
   influxdb: {
     host: '127.0.0.1',   // InfluxDB host. (default 127.0.0.1)
     port: 8086,          // InfluxDB port. (default 8086)
+    version: 0.8,        // InfluxDB port. (default 0.8)
     ssl: false,          // InfluxDB is hosted over SSL. (default false)
     database: 'dbname',  // InfluxDB database instance. (required)
     username: 'user',    // InfluxDB database username. (required)

lib/influxdb.js

@@ -45,13 +45,15 @@ function InfluxdbBackend(startupTime, config, events) {
 
   self.defaultHost = '127.0.0.1';
   self.defaultPort = 8086;
+  self.defaultVersion = 0.8;
   self.defaultFlushEnable = true;
   self.defaultProxyEnable = false;
   self.defaultProxySuffix = 'raw';
   self.defaultProxyFlushInterval = 1000;
 
   self.host = self.defaultHost;
   self.port = self.defaultPort;
+  self.version = self.defaultVersion;
   self.protocol = http;
   self.flushEnable = self.defaultFlushEnable;
   self.proxyEnable = self.defaultProxyEnable;
@@ -66,6 +68,7 @@ function InfluxdbBackend(startupTime, config, events) {
   if (config.influxdb) {
     self.host = config.influxdb.host || self.defaultHost;
     self.port = config.influxdb.port || self.defaultPort;
+    self.version = config.influxdb.version || self.defaultVersion;
     self.user = config.influxdb.username;
     self.pass = config.influxdb.password;
     self.database = config.influxdb.database;
@@ -87,6 +90,14 @@ function InfluxdbBackend(startupTime, config, events) {
     }
   }
 
+  if (self.version >= 0.9) {
+    self.assembleEvent = self.assembleEvent_v09;
+    self.httpPOST = self.httpPOST_v09;
+  } else {
+    self.assembleEvent = self.assembleEvent_v08;
+    self.httpPOST = self.httpPOST_v08;
+  }
+
   if (self.proxyEnable) {
     self.log('Starting the buffer flush interval. (every ' + self.proxyFlushInterval + 'ms)');
     setInterval(function () {
@@ -355,7 +366,7 @@ InfluxdbBackend.prototype.clearRegistry = function () {
   return registry;
 }
 
-InfluxdbBackend.prototype.assembleEvent = function (name, events) {
+InfluxdbBackend.prototype.assembleEvent_v08 = function (name, events) {
   var self = this;
 
   var payload = {
@@ -380,7 +391,19 @@ InfluxdbBackend.prototype.assembleEvent = function (name, events) {
   return payload;
 }
 
-InfluxdbBackend.prototype.httpPOST = function (points) {
+InfluxdbBackend.prototype.assembleEvent_v09 = function (name, events) {
+  var self = this;
+
+  var payload = {
+    name: name,
+    timestamp: events[0]['time'],
+    fields: { value: events[0]['value'] }
+  }
+
+  return payload;
+}
+
+InfluxdbBackend.prototype.httpPOST_v08 = function (points) {
   /* Do not send if there are no points. */
   if (!points.length) { return; }
 
@@ -392,7 +415,7 @@ InfluxdbBackend.prototype.httpPOST = function (points) {
   self.logDebug(function () {
     return 'Sending ' + points.length + ' different points via ' + protocolName;
   });
-  
+
   self.influxdbStats.numStats = points.length;
 
   var options = {
@@ -435,6 +458,66 @@ InfluxdbBackend.prototype.httpPOST = function (points) {
   req.end();
 }
 
+InfluxdbBackend.prototype.httpPOST_v09 = function (points) {
+  /* Do not send if there are no points. */
+  if (!points.length) { return; }
+
+  var self = this,
+      query = {u: self.user, p: self.pass},
+      protocolName = self.protocol == http ? 'HTTP' : 'HTTPS',
+      startTime;
+
+  self.logDebug(function () {
+    return 'Sending ' + points.length + ' different points via ' + protocolName;
+  });
+
+  self.influxdbStats.numStats = points.length;
+
+  var options = {
+    hostname: self.host,
+    port: self.port,
+    path: '/write?' + querystring.stringify(query),
+    method: 'POST',
+    agent: false // Is it okay to use "undefined" here? (keep-alive)
+  };
+
+  var req = self.protocol.request(options);
+
+  req.on('socket', function (res) {
+    startTime = process.hrtime();
+  });
+
+  req.on('response', function (res) {
+    var status = res.statusCode;
+
+    self.influxdbStats.httpResponseTime = millisecondsSince(startTime);
+
+    if (status !== 200) {
+      self.log(protocolName + ' Error: ' + status);
+    }
+  });
+
+  req.on('error', function (e, i) {
+    self.log(e);
+  });
+
+  var payload = JSON.stringify({
+    database: self.database,
+    time_precision: 'ms',
+    points: points
+  });
+
+  self.influxdbStats.payloadSize = Buffer.byteLength(payload);
+
+  self.logDebug(function () {
+    var size = (self.influxdbStats.payloadSize / 1024).toFixed(2);
+    return 'Payload size ' + size + ' KB';
+  });
+
+  req.write(payload);
+  req.end();
+}
+
 InfluxdbBackend.prototype.configCheck = function () {
   var self = this,
       success = true;