initial influxdb connector

src/datasource-zabbix/zabbix/connectors/influxdb/influxdbConnector.js

@@ -1,5 +1,6 @@
 import _ from 'lodash';
 import DBConnector from '../dbConnector';
+// import InfluxSeries from 'grafana/app/plugins/datasource/influxdb/influx_series';
 
 const DEFAULT_QUERY_LIMIT = 10000;
 const HISTORY_TO_TABLE_MAP = {
@@ -65,8 +66,10 @@ export class InfluxDBConnector extends DBConnector {
       return this.invokeInfluxDBQuery(query);
     });
 
-    return Promise.all(promises).then(results => {
-      return _.flatten(results);
+    return Promise.all(promises)
+    .then(_.flatten)
+    .then(results => {
+      return handleInfluxHistoryResponse(results);
     });
   }
 
@@ -89,40 +92,82 @@ export class InfluxDBConnector extends DBConnector {
       return this.invokeInfluxDBQuery(query);
     });
 
-    return Promise.all(promises).then(results => {
-      return _.flatten(results);
+    return Promise.all(promises)
+    .then(_.flatten)
+    .then(results => {
+      return handleInfluxHistoryResponse(results);
     });
   }
 
   buildHistoryQuery(itemids, table, timeFrom, timeTill, intervalSec, aggFunction) {
     const AGG = aggFunction === 'AVG' ? 'MEAN' : aggFunction;
-    const where_clause = itemids.map(itemid => `"itemid" = ${itemid}`).join(' AND ');
-    const query = `SELECT "itemid", "time", ${AGG}("value") FROM "${table}"
-      WHERE ${where_clause} AND "time" >= ${timeFrom} AND "time" <= ${timeTill}
-      GROUP BY time(${intervalSec}s)`;
+    const where_clause = this.buildWhereClause(itemids);
+    const query = `SELECT ${AGG}("value") FROM "${table}"
+      WHERE ${where_clause} AND "time" >= ${timeFrom}s AND "time" <= ${timeTill}s
+      GROUP BY time(${intervalSec}s), "itemid" fill(linear)`;
     return compactQuery(query);
   }
 
   buildTrendsQuery(itemids, table, timeFrom, timeTill, intervalSec, aggFunction, valueColumn) {
     const AGG = aggFunction === 'AVG' ? 'MEAN' : aggFunction;
-    const where_clause = itemids.map(itemid => `"itemid" = ${itemid}`).join(' AND ');
-    const query = `SELECT "itemid", "time", ${AGG}("${valueColumn}") FROM "${table}"
-      WHERE ${where_clause} AND "time" >= ${timeFrom} AND "time" <= ${timeTill}
+    const where_clause = this.buildWhereClause(itemids);
+    const query = `SELECT ${AGG}("${valueColumn}") FROM "${table}"
+      WHERE ${where_clause} AND "time" >= ${timeFrom}s AND "time" <= ${timeTill}s
       GROUP BY time(${intervalSec}s)`;
     return compactQuery(query);
   }
 
+  buildWhereClause(itemids) {
+    const itemidsWhere = itemids.map(itemid => `"itemid" = '${itemid}'`).join(' OR ');
+    return `(${itemidsWhere})`;
+  }
+
   handleGrafanaTSResponse(history, items, addHostName = true) {
     return convertGrafanaTSResponse(history, items, addHostName);
   }
 
   invokeInfluxDBQuery(query) {
-    return this.ds._seriesQuery(query);
+    return this.ds._seriesQuery(query).then(data => {
+      return data && data.results ? data.results : [];
+    });
   }
 }
 
 ///////////////////////////////////////////////////////////////////////////////
 
+function handleInfluxHistoryResponse(results) {
+  if (!results) {
+    return [];
+  }
+
+  const seriesList = [];
+  for (let i = 0; i < results.length; i++) {
+    const result = results[i];
+    if (!result || !result.series) {
+      continue;
+    }
+
+    const influxSeriesList = results[i].series;
+
+    for (let y = 0; y < influxSeriesList.length; y++) {
+      const influxSeries = influxSeriesList[y];
+      const datapoints = [];
+      if (influxSeries.values) {
+        for (i = 0; i < influxSeries.values.length; i++) {
+          datapoints[i] = [influxSeries.values[i][1], influxSeries.values[i][0]];
+        }
+      }
+      const timeSeries = {
+        name: influxSeries.tags.itemid,
+        points: datapoints
+      };
+      seriesList.push(timeSeries);
+    }
+  }
+
+  return seriesList;
+}
+
 function convertGrafanaTSResponse(time_series, items, addHostName) {
   //uniqBy is needed to deduplicate
   var hosts = _.uniqBy(_.flatten(_.map(items, 'hosts')), 'hostid');