influx: support retention policy for long-term stored data

src/datasource-zabbix/config.controller.js

@@ -31,6 +31,7 @@ export class ZabbixDSConfigController {
     this.dbDataSources = this.getSupportedDBDataSources();
     this.zabbixVersions = _.cloneDeep(zabbixVersions);
     this.autoDetectZabbixVersion();
+    console.log(this.dbDataSources);
   }
 
   getSupportedDBDataSources() {
@@ -40,6 +41,12 @@ export class ZabbixDSConfigController {
     });
   }
 
+  getCurrentDatasourceType() {
+    const dsId = this.current.jsonData.dbConnectionDatasourceId;
+    const currentDs = _.find(this.dbDataSources, { 'id': dsId });
+    return currentDs ? currentDs.type : null;
+  }
+
   autoDetectZabbixVersion() {
     if (!this.current.id) {
       return;

src/datasource-zabbix/datasource.js

@@ -58,6 +58,7 @@ export class ZabbixDatasource {
     this.enableDirectDBConnection = jsonData.dbConnectionEnable || false;
     this.dbConnectionDatasourceId = jsonData.dbConnectionDatasourceId;
     this.dbConnectionDatasourceName = jsonData.dbConnectionDatasourceName;
+    this.dbConnectionRetentionPolicy = jsonData.dbConnectionRetentionPolicy;
 
     let zabbixOptions = {
       url: this.url,
@@ -69,7 +70,8 @@ export class ZabbixDatasource {
       cacheTTL: this.cacheTTL,
       enableDirectDBConnection: this.enableDirectDBConnection,
       dbConnectionDatasourceId: this.dbConnectionDatasourceId,
-      dbConnectionDatasourceName: this.dbConnectionDatasourceName
+      dbConnectionDatasourceName: this.dbConnectionDatasourceName,
+      dbConnectionRetentionPolicy: this.dbConnectionRetentionPolicy,
     };
 
     this.zabbix = new Zabbix(zabbixOptions, datasourceSrv, backendSrv);

src/datasource-zabbix/partials/config.html

@@ -109,6 +109,22 @@
       </div>
     </div>
   </div>
+  <div ng-if="ctrl.getCurrentDatasourceType() === 'influxdb'">
+    <div class="gf-form max-width-30">
+      <span class="gf-form-label width-12">
+        Retention Policy
+        <info-popover mode="right-normal">
+          Specify retention policy name for fetching long-term stored data (optional).
+          Leave it blank if only default retention policy is using.
+        </info-popover>
+      </span>
+      <input class="gf-form-input max-width-16"
+        type="text"
+        ng-model='ctrl.current.jsonData.dbConnectionRetentionPolicy'
+        placeholder="Retention policy name">
+      </input>
+    </div>
+  </div>
 </div>
 
 <div class="gf-form-group">

src/datasource-zabbix/specs/influxdbConnector.test.js

@@ -5,7 +5,7 @@ describe('InfluxDBConnector', () => {
   let ctx = {};
 
   beforeEach(() => {
-    ctx.options = { datasourceName: 'InfluxDB DS' };
+    ctx.options = { datasourceName: 'InfluxDB DS', retentionPolicy: 'longterm' };
     ctx.datasourceSrvMock = {
       loadDatasource: jest.fn().mockResolvedValue(
         { id: 42, name: 'InfluxDB DS', meta: {} }
@@ -15,15 +15,16 @@ describe('InfluxDBConnector', () => {
     ctx.influxDBConnector.invokeInfluxDBQuery = jest.fn().mockResolvedValue([]);
     ctx.defaultQueryParams = {
       itemids: ['123', '234'],
-      timeFrom: 15000, timeTill: 15100, intervalSec: 5,
+      range: { timeFrom: 15000, timeTill: 15100 },
+      intervalSec: 5,
       table: 'history', aggFunction: 'MAX'
     };
   });
 
   describe('When building InfluxDB query', () => {
     it('should build proper query', () => {
-      const { itemids, timeFrom, timeTill, intervalSec, table, aggFunction } = ctx.defaultQueryParams;
+      const { itemids, range, intervalSec, table, aggFunction } = ctx.defaultQueryParams;
-      const query = ctx.influxDBConnector.buildHistoryQuery(itemids, table, timeFrom, timeTill, intervalSec, aggFunction);
+      const query = ctx.influxDBConnector.buildHistoryQuery(itemids, table, range, intervalSec, aggFunction);
       const expected = compactQuery(`SELECT MAX("value")
         FROM "history" WHERE ("itemid" = '123' OR "itemid" = '234') AND "time" >= 15000s AND "time" <= 15100s
         GROUP BY time(5s), "itemid" fill(none)
@@ -32,9 +33,9 @@ describe('InfluxDBConnector', () => {
     });
 
     it('should use MEAN instead of AVG', () => {
-      const { itemids, timeFrom, timeTill, intervalSec, table } = ctx.defaultQueryParams;
+      const { itemids, range, intervalSec, table } = ctx.defaultQueryParams;
       const aggFunction = 'AVG';
-      const query = ctx.influxDBConnector.buildHistoryQuery(itemids, table, timeFrom, timeTill, intervalSec, aggFunction);
+      const query = ctx.influxDBConnector.buildHistoryQuery(itemids, table, range, intervalSec, aggFunction);
       const expected = compactQuery(`SELECT MEAN("value")
         FROM "history" WHERE ("itemid" = '123' OR "itemid" = '234') AND "time" >= 15000s AND "time" <= 15100s
         GROUP BY time(5s), "itemid" fill(none)
@@ -45,7 +46,7 @@ describe('InfluxDBConnector', () => {
 
   describe('When invoking InfluxDB query', () => {
     it('should query proper table depending on item type', () => {
-      const { timeFrom, timeTill} = ctx.defaultQueryParams;
+      const { timeFrom, timeTill } = ctx.defaultQueryParams.range;
       const options = { intervalMs: 5000 };
       const items = [
         { itemid: '123', value_type: 3 }
@@ -59,7 +60,7 @@ describe('InfluxDBConnector', () => {
     });
 
     it('should split query if different item types are used', () => {
-      const { timeFrom, timeTill} = ctx.defaultQueryParams;
+      const { timeFrom, timeTill } = ctx.defaultQueryParams.range;
       const options = { intervalMs: 5000 };
       const items = [
         { itemid: '123', value_type: 0 },
@@ -78,8 +79,9 @@ describe('InfluxDBConnector', () => {
       expect(ctx.influxDBConnector.invokeInfluxDBQuery).toHaveBeenNthCalledWith(2, expectedQuerySecond);
     });
 
-    it('should use the same table for trends query', () => {
+    it('should use the same table for trends query if no retention policy set', () => {
-      const { timeFrom, timeTill} = ctx.defaultQueryParams;
+      ctx.influxDBConnector.retentionPolicy = '';
+      const { timeFrom, timeTill } = ctx.defaultQueryParams.range;
       const options = { intervalMs: 5000 };
       const items = [
         { itemid: '123', value_type: 3 }
@@ -91,5 +93,19 @@ describe('InfluxDBConnector', () => {
       ctx.influxDBConnector.getTrends(items, timeFrom, timeTill, options);
       expect(ctx.influxDBConnector.invokeInfluxDBQuery).toHaveBeenCalledWith(expectedQuery);
     });
+
+    it('should use retention policy name for trends query if it was set', () => {
+      const { timeFrom, timeTill } = ctx.defaultQueryParams.range;
+      const options = { intervalMs: 5000 };
+      const items = [
+        { itemid: '123', value_type: 3 }
+      ];
+      const expectedQuery = compactQuery(`SELECT MEAN("value")
+        FROM "longterm"."history_uint" WHERE ("itemid" = '123') AND "time" >= 15000s AND "time" <= 15100s
+        GROUP BY time(5s), "itemid" fill(none)
+      `);
+      ctx.influxDBConnector.getTrends(items, timeFrom, timeTill, options);
+      expect(ctx.influxDBConnector.invokeInfluxDBQuery).toHaveBeenCalledWith(expectedQuery);
+    });
   });
 });

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

@@ -5,6 +5,7 @@ import { DBConnector, HISTORY_TO_TABLE_MAP, consolidateByFunc } from '../dbConne
 export class InfluxDBConnector extends DBConnector {
   constructor(options, datasourceSrv) {
     super(options, datasourceSrv);
+    this.retentionPolicy = options.retentionPolicy;
     super.loadDBDataSource().then(ds => {
       this.influxDS = ds;
       return ds;
@@ -19,9 +20,10 @@ export class InfluxDBConnector extends DBConnector {
   }
 
   getHistory(items, timeFrom, timeTill, options) {
-    let {intervalMs, consolidateBy} = options;
+    let { intervalMs, consolidateBy, retentionPolicy } = options;
     const intervalSec = Math.ceil(intervalMs / 1000);
 
+    const range = { timeFrom, timeTill };
     consolidateBy = consolidateBy || 'avg';
     const aggFunction = consolidateByFunc[consolidateBy] || consolidateBy;
 
@@ -30,7 +32,7 @@ export class InfluxDBConnector extends DBConnector {
     const promises = _.map(grouped_items, (items, value_type) => {
       const itemids = _.map(items, 'itemid');
       const table = HISTORY_TO_TABLE_MAP[value_type];
-      const query = this.buildHistoryQuery(itemids, table, timeFrom, timeTill, intervalSec, aggFunction);
+      const query = this.buildHistoryQuery(itemids, table, range, intervalSec, aggFunction, retentionPolicy);
       return this.invokeInfluxDBQuery(query);
     });
 
@@ -42,13 +44,16 @@ export class InfluxDBConnector extends DBConnector {
   }
 
   getTrends(items, timeFrom, timeTill, options) {
+    options.retentionPolicy = this.retentionPolicy;
     return this.getHistory(items, timeFrom, timeTill, options);
   }
 
-  buildHistoryQuery(itemids, table, timeFrom, timeTill, intervalSec, aggFunction) {
+  buildHistoryQuery(itemids, table, range, intervalSec, aggFunction, retentionPolicy) {
+    const { timeFrom, timeTill } = range;
+    const measurement = retentionPolicy ? `"${retentionPolicy}"."${table}"` : `"${table}"`;
     const AGG = aggFunction === 'AVG' ? 'MEAN' : aggFunction;
     const where_clause = this.buildWhereClause(itemids);
-    const query = `SELECT ${AGG}("value") FROM "${table}"
+    const query = `SELECT ${AGG}("value") FROM ${measurement}
       WHERE ${where_clause} AND "time" >= ${timeFrom}s AND "time" <= ${timeTill}s
       GROUP BY time(${intervalSec}s), "itemid" fill(none)`;
     return compactQuery(query);

src/datasource-zabbix/zabbix/zabbix.js

@@ -37,6 +37,7 @@ export class Zabbix {
       enableDirectDBConnection,
       dbConnectionDatasourceId,
       dbConnectionDatasourceName,
+      dbConnectionRetentionPolicy,
     } = options;
 
     this.enableDirectDBConnection = enableDirectDBConnection;
@@ -55,7 +56,8 @@ export class Zabbix {
     this.bindRequests();
 
     if (enableDirectDBConnection) {
-      this.initDBConnector(dbConnectionDatasourceId, dbConnectionDatasourceName, datasourceSrv)
+      const connectorOptions = { dbConnectionRetentionPolicy };
+      this.initDBConnector(dbConnectionDatasourceId, dbConnectionDatasourceName, datasourceSrv, connectorOptions)
       .then(() => {
         this.getHistoryDB = this.cachingProxy.proxyfyWithCache(this.dbConnector.getHistory, 'getHistory', this.dbConnector);
         this.getTrendsDB = this.cachingProxy.proxyfyWithCache(this.dbConnector.getTrends, 'getTrends', this.dbConnector);
@@ -63,14 +65,15 @@ export class Zabbix {
     }
   }
 
-  initDBConnector(datasourceId, datasourceName, datasourceSrv) {
+  initDBConnector(datasourceId, datasourceName, datasourceSrv, options) {
     return DBConnector.loadDatasource(datasourceId, datasourceName, datasourceSrv)
     .then(ds => {
-      const options = { datasourceId, datasourceName };
+      let connectorOptions = { datasourceId, datasourceName };
       if (ds.type === 'influxdb') {
-        this.dbConnector = new InfluxDBConnector(options, datasourceSrv);
+        connectorOptions.retentionPolicy = options.dbConnectionRetentionPolicy;
+        this.dbConnector = new InfluxDBConnector(connectorOptions, datasourceSrv);
       } else {
-        this.dbConnector = new SQLConnector(options, datasourceSrv);
+        this.dbConnector = new SQLConnector(connectorOptions, datasourceSrv);
       }
       return this.dbConnector;
     });