Misthios/grafana-zabbix
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

Refactor DB connector

Browse Source
This commit is contained in:
Alexander Zobnin
2021-08-05 12:15:15 +03:00
parent e12b8cbefb
commit 3831c6e28e
9 changed files with 251 additions and 319 deletions
Download Patch File Download Diff File Expand all files Collapse all files

180
src/datasource-zabbix/zabbix/connectors/dbConnector.js
View File

@@ -1,180 +0,0 @@
import _ from 'lodash';
import { getDataSourceSrv } from '@grafana/runtime';
import responseHandler from "../../responseHandler";
export const DEFAULT_QUERY_LIMIT = 10000;
export const HISTORY_TO_TABLE_MAP = {
'0': 'history',
'1': 'history_str',
'2': 'history_log',
'3': 'history_uint',
'4': 'history_text'
};
export const TREND_TO_TABLE_MAP = {
'0': 'trends',
'3': 'trends_uint'
};
export const consolidateByFunc = {
'avg': 'AVG',
'min': 'MIN',
'max': 'MAX',
'sum': 'SUM',
'count': 'COUNT'
};
export const consolidateByTrendColumns = {
'avg': 'value_avg',
'min': 'value_min',
'max': 'value_max',
'sum': 'num*value_avg' // sum of sums inside the one-hour trend period
};
/**
* Base class for external history database connectors. Subclasses should implement `getHistory()`, `getTrends()` and
* `testDataSource()` methods, which describe how to fetch data from source other than Zabbix API.
*/
export class DBConnector {
constructor(options) {
this.datasourceId = options.datasourceId;
this.datasourceName = options.datasourceName;
this.datasourceTypeId = null;
this.datasourceTypeName = null;
}
static loadDatasource(dsId, dsName) {
if (!dsName && dsId !== undefined) {
let ds = _.find(getDataSourceSrv().getAll(), {'id': dsId});
if (!ds) {
return Promise.reject(`Data Source with ID ${dsId} not found`);
}
dsName = ds.name;
}
if (dsName) {
return getDataSourceSrv().loadDatasource(dsName);
} else {
return Promise.reject(`Data Source name should be specified`);
}
}
loadDBDataSource() {
return DBConnector.loadDatasource(this.datasourceId, this.datasourceName)
.then(ds => {
this.datasourceTypeId = ds.meta.id;
this.datasourceTypeName = ds.meta.name;
if (!this.datasourceName) {
this.datasourceName = ds.name;
}
if (!this.datasourceId) {
this.datasourceId = ds.id;
}
return ds;
});
}
/**
* Send test request to datasource in order to ensure it's working.
*/
testDataSource() {
throw new ZabbixNotImplemented('testDataSource()');
}
/**
* Get history data from external sources.
*/
getHistory() {
throw new ZabbixNotImplemented('getHistory()');
}
/**
* Get trends data from external sources.
*/
getTrends() {
throw new ZabbixNotImplemented('getTrends()');
}
}
// Define Zabbix DB Connector exception type for non-implemented methods
export class ZabbixNotImplemented {
constructor(methodName) {
this.code = null;
this.name = 'ZabbixNotImplemented';
this.message = `Zabbix DB Connector Error: method ${methodName || ''} should be implemented in subclass of DBConnector`;
}
toString() {
return this.message;
}
}
export function handleDBDataSourceResponse(response, items) {
const series = responseHandler.dataResponseToTimeSeries(response, items);
// return convertGrafanaTSResponse(series, items, addHostName);
return series;
}
/**
* Converts time series returned by the data source into format that Grafana expects
* time_series is Array of series:
* ```
* [{
* name: string,
* points: Array<[value: number, timestamp: number]>
* }]
* ```
*/
export function convertGrafanaTSResponse(time_series, items, addHostName) {
if (time_series.length === 0) {
return [];
}
//uniqBy is needed to deduplicate
const hosts = _.uniqBy(_.flatten(_.map(items, 'hosts')), 'hostid');
let grafanaSeries = _.map(_.compact(time_series), series => {
const itemid = series.name;
const item = _.find(items, {'itemid': itemid});
let alias = item.name;
// Add scopedVars for using in alias functions
const scopedVars = {
'__zbx_item': { value: item.name },
'__zbx_item_name': { value: item.name },
'__zbx_item_key': { value: item.key_ },
'__zbx_item_interval': { value: item.delay },
};
if (_.keys(hosts).length > 0) {
const host = _.find(hosts, {'hostid': item.hostid});
scopedVars['__zbx_host'] = { value: host.host };
scopedVars['__zbx_host_name'] = { value: host.name };
// Only add host when multiple hosts selected
if (_.keys(hosts).length > 1 && addHostName) {
alias = host.name + ": " + alias;
}
}
// CachingProxy deduplicates requests and returns one time series for equal queries.
// Clone is needed to prevent changing of series object shared between all targets.
const datapoints = _.cloneDeep(series.points);
return {
target: alias,
datapoints,
scopedVars,
item
};
});
return _.sortBy(grafanaSeries, 'target');
}
const defaults = {
DBConnector,
DEFAULT_QUERY_LIMIT,
HISTORY_TO_TABLE_MAP,
TREND_TO_TABLE_MAP,
consolidateByFunc,
consolidateByTrendColumns
};
export default defaults;

97
src/datasource-zabbix/zabbix/connectors/dbConnector.ts Normal file
View File

@@ -0,0 +1,97 @@
import _ from 'lodash';
import { getDataSourceSrv } from '@grafana/runtime';
export const DEFAULT_QUERY_LIMIT = 10000;
export const HISTORY_TO_TABLE_MAP = {
'0': 'history',
'1': 'history_str',
'2': 'history_log',
'3': 'history_uint',
'4': 'history_text'
};
export const TREND_TO_TABLE_MAP = {
'0': 'trends',
'3': 'trends_uint'
};
export const consolidateByFunc = {
'avg': 'AVG',
'min': 'MIN',
'max': 'MAX',
'sum': 'SUM',
'count': 'COUNT'
};
export const consolidateByTrendColumns = {
'avg': 'value_avg',
'min': 'value_min',
'max': 'value_max',
'sum': 'num*value_avg' // sum of sums inside the one-hour trend period
};
export interface IDBConnector {
getHistory(): any;
getTrends(): any;
testDataSource(): any;
}
/**
* Base class for external history database connectors. Subclasses should implement `getHistory()`, `getTrends()` and
* `testDataSource()` methods, which describe how to fetch data from source other than Zabbix API.
*/
export class DBConnector {
protected datasourceId: any;
private datasourceName: any;
protected datasourceTypeId: any;
private datasourceTypeName: any;
constructor(options) {
this.datasourceId = options.datasourceId;
this.datasourceName = options.datasourceName;
this.datasourceTypeId = null;
this.datasourceTypeName = null;
}
static loadDatasource(dsId, dsName) {
if (!dsName && dsId !== undefined) {
const ds = _.find(getDataSourceSrv().getList(), { 'id': dsId });
if (!ds) {
return Promise.reject(`Data Source with ID ${dsId} not found`);
}
dsName = ds.name;
}
if (dsName) {
return getDataSourceSrv().get(dsName);
} else {
return Promise.reject(`Data Source name should be specified`);
}
}
loadDBDataSource() {
return DBConnector.loadDatasource(this.datasourceId, this.datasourceName)
.then(ds => {
this.datasourceTypeId = ds.meta.id;
this.datasourceTypeName = ds.meta.name;
if (!this.datasourceName) {
this.datasourceName = ds.name;
}
if (!this.datasourceId) {
this.datasourceId = ds.id;
}
return ds;
});
}
}
export default {
DBConnector,
DEFAULT_QUERY_LIMIT,
HISTORY_TO_TABLE_MAP,
TREND_TO_TABLE_MAP,
consolidateByFunc,
consolidateByTrendColumns
};

25
src/datasource-zabbix/zabbix/connectors/influxdb/influxdbConnector.js → src/datasource-zabbix/zabbix/connectors/influxdb/influxdbConnector.ts
View File

@@ -1,6 +1,6 @@
import _ from 'lodash';
import { compactQuery } from '../../../utils';
import { DBConnector, HISTORY_TO_TABLE_MAP, consolidateByTrendColumns } from '../dbConnector';
import { consolidateByTrendColumns, DBConnector, HISTORY_TO_TABLE_MAP } from '../dbConnector';
const consolidateByFunc = {
'avg': 'MEAN',
@@ -11,6 +11,9 @@ const consolidateByFunc = {
};
export class InfluxDBConnector extends DBConnector {
private retentionPolicy: any;
private influxDS: any;
constructor(options) {
super(options);
this.retentionPolicy = options.retentionPolicy;
@@ -26,16 +29,19 @@ export class InfluxDBConnector extends DBConnector {
testDataSource() {
return this.influxDS.testDatasource().then(result => {
if (result.status && result.status === 'error') {
return Promise.reject({ data: {
message: `InfluxDB connection error: ${result.message}`
}});
return Promise.reject({
data: {
message: `InfluxDB connection error: ${result.message}`
}
});
}
return result;
});
}
getHistory(items, timeFrom, timeTill, options) {
let { intervalMs, consolidateBy, retentionPolicy } = options;
const { intervalMs, retentionPolicy } = options;
let { consolidateBy } = options;
const intervalSec = Math.ceil(intervalMs / 1000);
const range = { timeFrom, timeTill };
@@ -71,9 +77,12 @@ export class InfluxDBConnector extends DBConnector {
}
const aggregation = consolidateByFunc[aggFunction] || aggFunction;
const where_clause = this.buildWhereClause(itemids);
const query = `SELECT ${aggregation}("${value}") FROM ${measurement}
WHERE ${where_clause} AND "time" >= ${timeFrom}s AND "time" <= ${timeTill}s
GROUP BY time(${intervalSec}s), "itemid" fill(none)`;
const query = `SELECT ${aggregation}("${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);
}

41
src/datasource-zabbix/zabbix/connectors/sql/mysql.js
View File

@@ -1,41 +0,0 @@
/**
* MySQL queries
*/
function historyQuery(itemids, table, timeFrom, timeTill, intervalSec, aggFunction) {
let query = `
SELECT CAST(itemid AS CHAR) AS metric, MIN(clock) AS time_sec, ${aggFunction}(value) AS value
FROM ${table}
WHERE itemid IN (${itemids})
AND clock > ${timeFrom} AND clock < ${timeTill}
GROUP BY (clock-${timeFrom}) DIV ${intervalSec}, metric
ORDER BY time_sec ASC
`;
return query;
}
function trendsQuery(itemids, table, timeFrom, timeTill, intervalSec, aggFunction, valueColumn) {
let query = `
SELECT CAST(itemid AS CHAR) AS metric, MIN(clock) AS time_sec, ${aggFunction}(${valueColumn}) AS value
FROM ${table}
WHERE itemid IN (${itemids})
AND clock > ${timeFrom} AND clock < ${timeTill}
GROUP BY (clock-${timeFrom}) DIV ${intervalSec}, metric
ORDER BY time_sec ASC
`;
return query;
}
const TEST_QUERY = `SELECT CAST(itemid AS CHAR) AS metric, clock AS time_sec, value_avg AS value FROM trends_uint LIMIT 1`;
function testQuery() {
return TEST_QUERY;
}
const mysql = {
historyQuery,
trendsQuery,
testQuery
};
export default mysql;

44
src/datasource-zabbix/zabbix/connectors/sql/mysql.ts Normal file
View File

@@ -0,0 +1,44 @@
/**
* MySQL queries
*/
function historyQuery(itemids, table, timeFrom, timeTill, intervalSec, aggFunction) {
return `
SELECT CAST(itemid AS CHAR) AS metric, MIN(clock) AS time_sec, ${aggFunction}(value) AS value
FROM ${table}
WHERE itemid IN (${itemids})
AND clock
> ${timeFrom}
AND clock
< ${timeTill}
GROUP BY (clock-${timeFrom}) DIV ${intervalSec}, metric
ORDER BY time_sec ASC
`;
}
function trendsQuery(itemids, table, timeFrom, timeTill, intervalSec, aggFunction, valueColumn) {
return `
SELECT CAST(itemid AS CHAR) AS metric, MIN(clock) AS time_sec, ${aggFunction}(${valueColumn}) AS value
FROM ${table}
WHERE itemid IN (${itemids})
AND clock
> ${timeFrom}
AND clock
< ${timeTill}
GROUP BY (clock-${timeFrom}) DIV ${intervalSec}, metric
ORDER BY time_sec ASC
`;
}
function testQuery() {
return `SELECT CAST(itemid AS CHAR) AS metric, clock AS time_sec, value_avg AS value
FROM trends_uint LIMIT 1`;
}
const mysql = {
historyQuery,
trendsQuery,
testQuery
};
export default mysql;

48
src/datasource-zabbix/zabbix/connectors/sql/postgres.js
View File

@@ -1,48 +0,0 @@
/**
* Postgres queries
*/
const ITEMID_FORMAT = 'FM99999999999999999999';
function historyQuery(itemids, table, timeFrom, timeTill, intervalSec, aggFunction) {
let time_expression = `clock / ${intervalSec} * ${intervalSec}`;
let query = `
SELECT to_char(itemid, '${ITEMID_FORMAT}') AS metric, ${time_expression} AS time, ${aggFunction}(value) AS value
FROM ${table}
WHERE itemid IN (${itemids})
AND clock > ${timeFrom} AND clock < ${timeTill}
GROUP BY 1, 2
ORDER BY time ASC
`;
return query;
}
function trendsQuery(itemids, table, timeFrom, timeTill, intervalSec, aggFunction, valueColumn) {
let time_expression = `clock / ${intervalSec} * ${intervalSec}`;
let query = `
SELECT to_char(itemid, '${ITEMID_FORMAT}') AS metric, ${time_expression} AS time, ${aggFunction}(${valueColumn}) AS value
FROM ${table}
WHERE itemid IN (${itemids})
AND clock > ${timeFrom} AND clock < ${timeTill}
GROUP BY 1, 2
ORDER BY time ASC
`;
return query;
}
const TEST_QUERY = `
SELECT to_char(itemid, '${ITEMID_FORMAT}') AS metric, clock AS time, value_avg AS value
FROM trends_uint LIMIT 1
`;
function testQuery() {
return TEST_QUERY;
}
const postgres = {
historyQuery,
trendsQuery,
testQuery
};
export default postgres;

52
src/datasource-zabbix/zabbix/connectors/sql/postgres.ts Normal file
View File

@@ -0,0 +1,52 @@
/**
* Postgres queries
*/
const ITEMID_FORMAT = 'FM99999999999999999999';
function historyQuery(itemids, table, timeFrom, timeTill, intervalSec, aggFunction) {
const time_expression = `clock / ${intervalSec} * ${intervalSec}`;
return `
SELECT to_char(itemid, '${ITEMID_FORMAT}') AS metric, ${time_expression} AS time, ${aggFunction}(value) AS value
FROM ${table}
WHERE itemid IN (${itemids})
AND clock
> ${timeFrom}
AND clock
< ${timeTill}
GROUP BY 1, 2
ORDER BY time ASC
`;
}
function trendsQuery(itemids, table, timeFrom, timeTill, intervalSec, aggFunction, valueColumn) {
const time_expression = `clock / ${intervalSec} * ${intervalSec}`;
return `
SELECT to_char(itemid, '${ITEMID_FORMAT}') AS metric, ${time_expression} AS time, ${aggFunction}(${valueColumn}) AS value
FROM ${table}
WHERE itemid IN (${itemids})
AND clock
> ${timeFrom}
AND clock
< ${timeTill}
GROUP BY 1, 2
ORDER BY time ASC
`;
}
const TEST_QUERY = `
SELECT to_char(itemid, '${ITEMID_FORMAT}') AS metric, clock AS time, value_avg AS value
FROM trends_uint LIMIT 1
`;
function testQuery() {
return TEST_QUERY;
}
const postgres = {
historyQuery,
trendsQuery,
testQuery
};
export default postgres;

66
src/datasource-zabbix/zabbix/connectors/sql/sqlConnector.js → src/datasource-zabbix/zabbix/connectors/sql/sqlConnector.ts
View File

@@ -11,6 +11,9 @@ const supportedDatabases = {
};
export class SQLConnector extends DBConnector {
private limit: number;
private sqlDialect: any;
constructor(options) {
super(options);
@@ -35,28 +38,18 @@ export class SQLConnector extends DBConnector {
* Try to invoke test query for one of Zabbix database tables.
*/
testDataSource() {
let testQuery = this.sqlDialect.testQuery();
const testQuery = this.sqlDialect.testQuery();
return this.invokeSQLQuery(testQuery);
}
getHistory(items, timeFrom, timeTill, options) {
let {intervalMs, consolidateBy} = options;
let intervalSec = Math.ceil(intervalMs / 1000);
// The interval must match the time range exactly n times, otherwise
// the resulting first and last data points will yield invalid values in the
// calculated average value in downsampleSeries - when using consolidateBy(avg)
let numOfIntervals = Math.ceil((timeTill - timeFrom) / intervalSec);
intervalSec = (timeTill - timeFrom) / numOfIntervals;
consolidateBy = consolidateBy || 'avg';
let aggFunction = dbConnector.consolidateByFunc[consolidateBy];
const { aggFunction, intervalSec } = getAggFunc(timeFrom, timeTill, options);
// Group items by value type and perform request for each value type
let grouped_items = _.groupBy(items, 'value_type');
let promises = _.map(grouped_items, (items, value_type) => {
let itemids = _.map(items, 'itemid').join(', ');
let table = HISTORY_TO_TABLE_MAP[value_type];
const grouped_items = _.groupBy(items, 'value_type');
const promises = _.map(grouped_items, (items, value_type) => {
const itemids = _.map(items, 'itemid').join(', ');
const table = HISTORY_TO_TABLE_MAP[value_type];
let query = this.sqlDialect.historyQuery(itemids, table, timeFrom, timeTill, intervalSec, aggFunction);
query = compactQuery(query);
@@ -69,23 +62,14 @@ export class SQLConnector extends DBConnector {
}
getTrends(items, timeFrom, timeTill, options) {
let { intervalMs, consolidateBy } = options;
let intervalSec = Math.ceil(intervalMs / 1000);
// The interval must match the time range exactly n times, otherwise
// the resulting first and last data points will yield invalid values in the
// calculated average value in downsampleSeries - when using consolidateBy(avg)
let numOfIntervals = Math.ceil((timeTill - timeFrom) / intervalSec);
intervalSec = (timeTill - timeFrom) / numOfIntervals;
consolidateBy = consolidateBy || 'avg';
let aggFunction = dbConnector.consolidateByFunc[consolidateBy];
const { consolidateBy } = options;
const { aggFunction, intervalSec } = getAggFunc(timeFrom, timeTill, options);
// Group items by value type and perform request for each value type
let grouped_items = _.groupBy(items, 'value_type');
let promises = _.map(grouped_items, (items, value_type) => {
let itemids = _.map(items, 'itemid').join(', ');
let table = TREND_TO_TABLE_MAP[value_type];
const grouped_items = _.groupBy(items, 'value_type');
const promises = _.map(grouped_items, (items, value_type) => {
const itemids = _.map(items, 'itemid').join(', ');
const table = TREND_TO_TABLE_MAP[value_type];
let valueColumn = _.includes(['avg', 'min', 'max', 'sum'], consolidateBy) ? consolidateBy : 'avg';
valueColumn = dbConnector.consolidateByTrendColumns[valueColumn];
let query = this.sqlDialect.trendsQuery(itemids, table, timeFrom, timeTill, intervalSec, aggFunction, valueColumn);
@@ -100,7 +84,7 @@ export class SQLConnector extends DBConnector {
}
invokeSQLQuery(query) {
let queryDef = {
const queryDef = {
refId: 'A',
format: 'time_series',
datasourceId: this.datasourceId,
@@ -116,7 +100,7 @@ export class SQLConnector extends DBConnector {
}
})
.then(response => {
let results = response.data.results;
const results = response.data.results;
if (results['A']) {
return results['A'].frames;
} else {
@@ -125,3 +109,19 @@ export class SQLConnector extends DBConnector {
});
}
}
function getAggFunc(timeFrom, timeTill, options) {
const { intervalMs } = options;
let { consolidateBy } = options;
let intervalSec = Math.ceil(intervalMs / 1000);
// The interval must match the time range exactly n times, otherwise
// the resulting first and last data points will yield invalid values in the
// calculated average value in downsampleSeries - when using consolidateBy(avg)
const numOfIntervals = Math.ceil((timeTill - timeFrom) / intervalSec);
intervalSec = (timeTill - timeFrom) / numOfIntervals;
consolidateBy = consolidateBy || 'avg';
const aggFunction = dbConnector.consolidateByFunc[consolidateBy];
return { aggFunction, intervalSec };
}

17
src/datasource-zabbix/zabbix/zabbix.ts
View File

@@ -4,13 +4,12 @@ import semver from 'semver';
import * as utils from '../utils';
import responseHandler from '../responseHandler';
import { CachingProxy } from './proxy/cachingProxy';
// import { ZabbixNotImplemented } from './connectors/dbConnector';
import { DBConnector, handleDBDataSourceResponse } from './connectors/dbConnector';
import { DBConnector } from './connectors/dbConnector';
import { ZabbixAPIConnector } from './connectors/zabbix_api/zabbixAPIConnector';
import { SQLConnector } from './connectors/sql/sqlConnector';
import { InfluxDBConnector } from './connectors/influxdb/influxdbConnector';
import { ZabbixConnector } from './types';
import { joinTriggersWithProblems, joinTriggersWithEvents } from '../problemsHandler';
import { joinTriggersWithEvents, joinTriggersWithProblems } from '../problemsHandler';
import { ProblemDTO } from '../types';
interface AppsResponse extends Array<any> {
@@ -274,7 +273,7 @@ export class Zabbix implements ZabbixConnector {
})
.then(items => {
if (!options.showDisabledItems) {
items = _.filter(items, {'status': '0'});
items = _.filter(items, { 'status': '0' });
}
return items;
@@ -432,7 +431,7 @@ export class Zabbix implements ZabbixConnector {
const [timeFrom, timeTo] = timeRange;
if (this.enableDirectDBConnection) {
return this.getHistoryDB(items, timeFrom, timeTo, options)
.then(history => handleDBDataSourceResponse(history, items));
.then(history => responseHandler.dataResponseToTimeSeries(history, items));
} else {
return this.zabbixAPI.getHistory(items, timeFrom, timeTo)
.then(history => responseHandler.handleHistory(history, items));
@@ -443,7 +442,7 @@ export class Zabbix implements ZabbixConnector {
const [timeFrom, timeTo] = timeRange;
if (this.enableDirectDBConnection) {
return this.getTrendsDB(items, timeFrom, timeTo, options)
.then(history => handleDBDataSourceResponse(history, items));
.then(history => responseHandler.dataResponseToTimeSeries(history, items));
} else {
const valueType = options.consolidateBy || options.valueType;
return this.zabbixAPI.getTrend(items, timeFrom, timeTo)
@@ -473,7 +472,7 @@ export class Zabbix implements ZabbixConnector {
return this.zabbixAPI.getSLA(itServiceIds, timeRange, options)
.then(slaResponse => {
return _.map(itServiceIds, serviceid => {
const itservice = _.find(itservices, {'serviceid': serviceid});
const itservice = _.find(itservices, { 'serviceid': serviceid });
return responseHandler.handleSLAResponse(itservice, target.slaProperty, slaResponse);
});
});
@@ -489,7 +488,7 @@ export class Zabbix implements ZabbixConnector {
* @return array with finded element or empty array
*/
function findByName(list, name) {
const finded = _.find(list, {'name': name});
const finded = _.find(list, { 'name': name });
if (finded) {
return [finded];
} else {
@@ -506,7 +505,7 @@ function findByName(list, name) {
* @return {[type]} array with finded element or empty array
*/
function filterByName(list, name) {
const finded = _.filter(list, {'name': name});
const finded = _.filter(list, { 'name': name });
if (finded) {
return finded;
} else {