diff --git a/application/src/main/data/upgrade/2.4.3/schema_update_psql_ts.sql b/application/src/main/data/upgrade/2.4.3/schema_update_psql_ts.sql index 3d17bbef2f..e10f3ee2bd 100644 --- a/application/src/main/data/upgrade/2.4.3/schema_update_psql_ts.sql +++ b/application/src/main/data/upgrade/2.4.3/schema_update_psql_ts.sql @@ -14,24 +14,6 @@ -- limitations under the License. -- --- call check_version(); - -CREATE OR REPLACE PROCEDURE check_version(INOUT valid_version boolean) LANGUAGE plpgsql AS $BODY$ -DECLARE - current_version integer; -BEGIN - RAISE NOTICE 'Check the current installed PostgreSQL version...'; - SELECT current_setting('server_version_num') INTO current_version; - IF current_version > 110000 THEN - RAISE NOTICE 'PostgreSQL version is valid!'; - RAISE NOTICE 'Schema update started...'; - SELECT true INTO valid_version; - ELSE - RAISE NOTICE 'Postgres version should be at least more than 10!'; - END IF; -END; -$BODY$; - -- call create_partition_ts_kv_table(); CREATE OR REPLACE PROCEDURE create_partition_ts_kv_table() LANGUAGE plpgsql AS $$ diff --git a/application/src/main/data/upgrade/2.4.3/schema_update_timescale_ts.sql b/application/src/main/data/upgrade/2.4.3/schema_update_timescale_ts.sql index ebbc6933ae..982ec9d85f 100644 --- a/application/src/main/data/upgrade/2.4.3/schema_update_timescale_ts.sql +++ b/application/src/main/data/upgrade/2.4.3/schema_update_timescale_ts.sql @@ -14,25 +14,6 @@ -- limitations under the License. -- --- call check_version(); - -CREATE OR REPLACE PROCEDURE check_version(INOUT valid_version boolean) LANGUAGE plpgsql AS $BODY$ - -DECLARE - current_version integer; -BEGIN - RAISE NOTICE 'Check the current installed PostgreSQL version...'; - SELECT current_setting('server_version_num') INTO current_version; - IF current_version > 110000 THEN - RAISE NOTICE 'PostgreSQL version is valid!'; - RAISE NOTICE 'Schema update started...'; - SELECT true INTO valid_version; - ELSE - RAISE NOTICE 'Postgres version should be at least more than 10!'; - END IF; -END; -$BODY$; - -- call create_new_ts_kv_table(); CREATE OR REPLACE PROCEDURE create_new_ts_kv_table() LANGUAGE plpgsql AS $$ diff --git a/application/src/main/java/org/thingsboard/server/service/install/AbstractSqlTsDatabaseUpgradeService.java b/application/src/main/java/org/thingsboard/server/service/install/AbstractSqlTsDatabaseUpgradeService.java index 901f773515..70e56f9489 100644 --- a/application/src/main/java/org/thingsboard/server/service/install/AbstractSqlTsDatabaseUpgradeService.java +++ b/application/src/main/java/org/thingsboard/server/service/install/AbstractSqlTsDatabaseUpgradeService.java @@ -32,11 +32,8 @@ import java.sql.Statement; public abstract class AbstractSqlTsDatabaseUpgradeService { protected static final String CALL_REGEX = "call "; - protected static final String CHECK_VERSION = "check_version(false)"; - protected static final String CHECK_VERSION_TO_DELETE = "check_version(INOUT valid_version boolean)"; protected static final String DROP_TABLE = "DROP TABLE "; protected static final String DROP_PROCEDURE_IF_EXISTS = "DROP PROCEDURE IF EXISTS "; - protected static final String DROP_PROCEDURE_CHECK_VERSION = DROP_PROCEDURE_IF_EXISTS + CHECK_VERSION_TO_DELETE; @Value("${spring.datasource.url}") protected String dbUrl; @@ -58,13 +55,14 @@ public abstract class AbstractSqlTsDatabaseUpgradeService { } protected boolean checkVersion(Connection conn) { - log.info("Check the current PostgreSQL version..."); boolean versionValid = false; try { Statement statement = conn.createStatement(); - ResultSet resultSet = statement.executeQuery(CALL_REGEX + CHECK_VERSION); + ResultSet resultSet = statement.executeQuery("SELECT current_setting('server_version_num')"); resultSet.next(); - versionValid = resultSet.getBoolean(1); + if(resultSet.getLong(1) > 110000) { + versionValid = true; + } statement.close(); } catch (Exception e) { log.info("Failed to check current PostgreSQL version due to: {}", e.getMessage()); diff --git a/application/src/main/java/org/thingsboard/server/service/install/PsqlTsDatabaseUpgradeService.java b/application/src/main/java/org/thingsboard/server/service/install/PsqlTsDatabaseUpgradeService.java index 96f2c126a0..a630540981 100644 --- a/application/src/main/java/org/thingsboard/server/service/install/PsqlTsDatabaseUpgradeService.java +++ b/application/src/main/java/org/thingsboard/server/service/install/PsqlTsDatabaseUpgradeService.java @@ -70,16 +70,15 @@ public class PsqlTsDatabaseUpgradeService extends AbstractSqlTsDatabaseUpgradeSe switch (fromVersion) { case "2.4.3": try (Connection conn = DriverManager.getConnection(dbUrl, dbUserName, dbPassword)) { - log.info("Updating timeseries schema ..."); - log.info("Load upgrade functions ..."); - loadSql(conn); + log.info("Check the current PostgreSQL version..."); boolean versionValid = checkVersion(conn); if (!versionValid) { - log.info("PostgreSQL version should be at least more than 11!"); - log.info("Please upgrade your PostgreSQL and restart the script!"); + throw new RuntimeException("PostgreSQL version should be at least more than 11, please upgrade your PostgreSQL and restart the script!"); } else { log.info("PostgreSQL version is valid!"); - log.info("Updating schema ..."); + log.info("Load upgrade functions ..."); + loadSql(conn); + log.info("Updating timeseries schema ..."); executeQuery(conn, CALL_CREATE_PARTITION_TS_KV_TABLE); executeQuery(conn, CALL_CREATE_PARTITIONS); executeQuery(conn, CALL_CREATE_TS_KV_DICTIONARY_TABLE); @@ -91,7 +90,6 @@ public class PsqlTsDatabaseUpgradeService extends AbstractSqlTsDatabaseUpgradeSe executeQuery(conn, DROP_TABLE_TS_KV_OLD); executeQuery(conn, DROP_TABLE_TS_KV_LATEST_OLD); - executeQuery(conn, DROP_PROCEDURE_CHECK_VERSION); executeQuery(conn, DROP_PROCEDURE_CREATE_PARTITION_TS_KV_TABLE); executeQuery(conn, DROP_PROCEDURE_CREATE_PARTITIONS); executeQuery(conn, DROP_PROCEDURE_CREATE_TS_KV_DICTIONARY_TABLE); diff --git a/application/src/main/java/org/thingsboard/server/service/install/TimescaleTsDatabaseUpgradeService.java b/application/src/main/java/org/thingsboard/server/service/install/TimescaleTsDatabaseUpgradeService.java index e438f965c8..99afd59a50 100644 --- a/application/src/main/java/org/thingsboard/server/service/install/TimescaleTsDatabaseUpgradeService.java +++ b/application/src/main/java/org/thingsboard/server/service/install/TimescaleTsDatabaseUpgradeService.java @@ -73,16 +73,15 @@ public class TimescaleTsDatabaseUpgradeService extends AbstractSqlTsDatabaseUpgr switch (fromVersion) { case "2.4.3": try (Connection conn = DriverManager.getConnection(dbUrl, dbUserName, dbPassword)) { - log.info("Updating timescale schema ..."); - log.info("Load upgrade functions ..."); - loadSql(conn); + log.info("Check the current PostgreSQL version..."); boolean versionValid = checkVersion(conn); if (!versionValid) { - log.info("PostgreSQL version should be at least more than 11!"); - log.info("Please upgrade your PostgreSQL and restart the script!"); + throw new RuntimeException("PostgreSQL version should be at least more than 11, please upgrade your PostgreSQL and restart the script!"); } else { log.info("PostgreSQL version is valid!"); - log.info("Updating schema ..."); + log.info("Load upgrade functions ..."); + loadSql(conn); + log.info("Updating timescale schema ..."); executeQuery(conn, CALL_CREATE_TS_KV_LATEST_TABLE); executeQuery(conn, CALL_CREATE_NEW_TENANT_TS_KV_TABLE); @@ -105,7 +104,7 @@ public class TimescaleTsDatabaseUpgradeService extends AbstractSqlTsDatabaseUpgr executeQuery(conn, "ALTER TABLE ts_kv ADD COLUMN json_v json;"); executeQuery(conn, "ALTER TABLE ts_kv_latest ADD COLUMN json_v json;"); - log.info("schema timeseries updated!"); + log.info("schema timescale updated!"); } } break; diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/kafka/TbKafkaNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/kafka/TbKafkaNode.java index 216fea54e3..267e8711aa 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/kafka/TbKafkaNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/kafka/TbKafkaNode.java @@ -16,16 +16,21 @@ package org.thingsboard.rule.engine.kafka; import lombok.extern.slf4j.Slf4j; +import org.apache.commons.lang3.BooleanUtils; import org.apache.kafka.clients.producer.*; +import org.apache.kafka.common.header.Headers; +import org.apache.kafka.common.header.internals.RecordHeader; +import org.apache.kafka.common.header.internals.RecordHeaders; import org.thingsboard.rule.engine.api.util.TbNodeUtils; import org.thingsboard.rule.engine.api.*; import org.thingsboard.server.common.data.plugin.ComponentType; import org.thingsboard.server.common.msg.TbMsg; import org.thingsboard.server.common.msg.TbMsgMetaData; +import java.nio.charset.Charset; +import java.nio.charset.StandardCharsets; import java.util.Properties; import java.util.concurrent.ExecutionException; -import java.util.concurrent.atomic.AtomicInteger; @Slf4j @RuleNode( @@ -46,8 +51,11 @@ public class TbKafkaNode implements TbNode { private static final String PARTITION = "partition"; private static final String TOPIC = "topic"; private static final String ERROR = "error"; + public static final String TB_MSG_MD_PREFIX = "tb_msg_md_"; private TbKafkaNodeConfiguration config; + private boolean addMetadataKeyValuesAsKafkaHeaders; + private Charset toBytesCharset; private Producer producer; @@ -66,8 +74,10 @@ public class TbKafkaNode implements TbNode { properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG, config.getBufferMemory()); if (config.getOtherProperties() != null) { config.getOtherProperties() - .forEach((k,v) -> properties.put(k, v)); + .forEach(properties::put); } + addMetadataKeyValuesAsKafkaHeaders = BooleanUtils.toBooleanDefaultIfNull(config.isAddMetadataKeyValuesAsKafkaHeaders(), false); + toBytesCharset = config.getKafkaHeadersCharset() != null ? Charset.forName(config.getKafkaHeadersCharset()) : StandardCharsets.UTF_8; try { this.producer = new KafkaProducer<>(properties); } catch (Exception e) { @@ -79,16 +89,16 @@ public class TbKafkaNode implements TbNode { public void onMsg(TbContext ctx, TbMsg msg) throws ExecutionException, InterruptedException, TbNodeException { String topic = TbNodeUtils.processPattern(config.getTopicPattern(), msg.getMetaData()); try { - producer.send(new ProducerRecord<>(topic, msg.getData()), - (metadata, e) -> { - if (metadata != null) { - TbMsg next = processResponse(ctx, msg, metadata); - ctx.tellNext(next, TbRelationTypes.SUCCESS); - } else { - TbMsg next = processException(ctx, msg, e); - ctx.tellFailure(next, e); - } - }); + if (!addMetadataKeyValuesAsKafkaHeaders) { + producer.send(new ProducerRecord<>(topic, msg.getData()), + (metadata, e) -> processRecord(ctx, msg, metadata, e)); + } else { + Headers headers = new RecordHeaders(); + msg.getMetaData().values().forEach((key, value) -> headers.add(new RecordHeader(TB_MSG_MD_PREFIX + key, value.getBytes(toBytesCharset)))); + producer.send(new ProducerRecord<>(topic, null, null, null, msg.getData(), headers), + (metadata, e) -> processRecord(ctx, msg, metadata, e)); + } + } catch (Exception e) { ctx.tellFailure(msg, e); } @@ -105,6 +115,16 @@ public class TbKafkaNode implements TbNode { } } + private void processRecord(TbContext ctx, TbMsg msg, RecordMetadata metadata, Exception e) { + if (metadata != null) { + TbMsg next = processResponse(ctx, msg, metadata); + ctx.tellNext(next, TbRelationTypes.SUCCESS); + } else { + TbMsg next = processException(ctx, msg, e); + ctx.tellFailure(next, e); + } + } + private TbMsg processResponse(TbContext ctx, TbMsg origMsg, RecordMetadata recordMetadata) { TbMsgMetaData metaData = origMsg.getMetaData().copy(); metaData.putValue(OFFSET, String.valueOf(recordMetadata.offset())); diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/kafka/TbKafkaNodeConfiguration.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/kafka/TbKafkaNodeConfiguration.java index 1e8fe5b0c7..a1d4eedbb4 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/kafka/TbKafkaNodeConfiguration.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/kafka/TbKafkaNodeConfiguration.java @@ -36,6 +36,9 @@ public class TbKafkaNodeConfiguration implements NodeConfiguration otherProperties; + private boolean addMetadataKeyValuesAsKafkaHeaders; + private String kafkaHeadersCharset; + @Override public TbKafkaNodeConfiguration defaultConfiguration() { TbKafkaNodeConfiguration configuration = new TbKafkaNodeConfiguration(); @@ -49,6 +52,8 @@ public class TbKafkaNodeConfiguration implements NodeConfiguration 0) { this.localSettings.levelColors = this.settingLevelColorsSubscribe(this.localSettings.fixedLevelColors); - this.updateLevelColors(this.localSettings.levelColors); } } + if (this.localSettings.showTicks) { + if (this.localSettings.ticksValue && this.localSettings.ticksValue.length) { + this.localSettings.ticks = this.settingTicksSubscribe(this.localSettings.ticksValue); + } + } + this.updateSetting(); + } + + static generateDatasorce(ctx, datasources, entityAlias, attribute, settings){ + let entityAliasId = ctx.aliasController.getEntityAliasId(entityAlias); + if (!entityAliasId) { + throw new Error('Not valid entity aliase name ' + entityAlias); + } + + let datasource = datasources.filter((datasource) => { + return datasource.entityAliasId === entityAliasId; + })[0]; + + let dataKey = { + type: ctx.$scope.$injector.get('types').dataKeyType.attribute, + name: attribute, + label: attribute, + settings: [settings], + _hash: Math.random() + }; + + if (datasource) { + let findDataKey = datasource.dataKeys.filter((dataKey) => { + return dataKey.name === attribute; + })[0]; + + if (findDataKey) { + findDataKey.settings.push(settings); + } else { + datasource.dataKeys.push(dataKey) + } + } else { + datasource = { + type: ctx.$scope.$injector.get('types').datasourceType.entity, + name: entityAlias, + aliasName: entityAlias, + entityAliasId: entityAliasId, + dataKeys: [dataKey] + }; + datasources.push(datasource); + } + + return datasources; + } + + settingTicksSubscribe(options) { + let ticksDatasource = []; + let predefineTicks = []; + + for (let i = 0; i < options.length; i++) { + let tick = options[i]; + if (tick.valueSource === 'predefinedValue' && isFinite(tick.value)) { + predefineTicks.push(tick.value) + } else if (tick.entityAlias && tick.attribute) { + try { + ticksDatasource = TbCanvasDigitalGauge.generateDatasorce(this.ctx, ticksDatasource, tick.entityAlias, tick.attribute, predefineTicks.length); + } catch (e) { + continue; + } + predefineTicks.push(null); + } + } + + this.subscribeAttributes(ticksDatasource, 'ticks').then((subscription) => { + this.ticksSourcesSubscription = subscription; + }); + + return predefineTicks; } settingLevelColorsSubscribe(options) { @@ -220,50 +302,14 @@ export default class TbCanvasDigitalGauge { color: color }) } else if (levelSetting.entityAlias && levelSetting.attribute) { - let entityAliasId = this.ctx.aliasController.getEntityAliasId(levelSetting.entityAlias); - if (!entityAliasId) { - return; - } - - let datasource = levelColorsDatasource.filter((datasource) => { - return datasource.entityAliasId === entityAliasId; - })[0]; - - let dataKey = { - type: this.ctx.$scope.$injector.get('types').dataKeyType.attribute, - name: levelSetting.attribute, - label: levelSetting.attribute, - settings: [{ + try { + levelColorsDatasource = TbCanvasDigitalGauge.generateDatasorce(this.ctx, levelColorsDatasource, levelSetting.entityAlias, levelSetting.attribute, { color: color, index: predefineLevelColors.length - }], - _hash: Math.random() - }; - - if (datasource) { - let findDataKey = datasource.dataKeys.filter((dataKey) => { - return dataKey.name === levelSetting.attribute; - })[0]; - - if (findDataKey) { - findDataKey.settings.push({ - color: color, - index: predefineLevelColors.length - }); - } else { - datasource.dataKeys.push(dataKey) - } - } else { - datasource = { - type: this.ctx.$scope.$injector.get('types').datasourceType.entity, - name: levelSetting.entityAlias, - aliasName: levelSetting.entityAlias, - entityAliasId: entityAliasId, - dataKeys: [dataKey] - }; - levelColorsDatasource.push(datasource); + }); + } catch (e) { + return; } - predefineLevelColors.push(null); } } @@ -278,49 +324,63 @@ export default class TbCanvasDigitalGauge { } } - this.subscribeLevelColorsAttributes(levelColorsDatasource); + this.subscribeAttributes(levelColorsDatasource, 'levelColors').then((subscription) => { + this.levelColorSourcesSubscription = subscription; + }); return predefineLevelColors; } - updateLevelColors(levelColors) { - this.gauge.options.levelColors = levelColors; - this.gauge.options = CanvasDigitalGauge.configure(this.gauge.options); - this.gauge.update(); - } + subscribeAttributes(datasources, typeAttributes) { + if (!datasources.length) { + return this.ctx.$scope.$injector.get('$q').when(null); + } - subscribeLevelColorsAttributes(datasources) { - let TbCanvasDigitalGauge = this; let levelColorsSourcesSubscriptionOptions = { datasources: datasources, useDashboardTimewindow: false, type: this.ctx.$scope.$injector.get('types').widgetType.latest.value, callbacks: { onDataUpdated: (subscription) => { - for (let i = 0; i < subscription.data.length; i++) { - let keyData = subscription.data[i]; - if (keyData && keyData.data && keyData.data[0]) { - let attrValue = keyData.data[0][1]; - if (isFinite(attrValue)) { - for (let i = 0; i < keyData.dataKey.settings.length; i++) { - let setting = keyData.dataKey.settings[i]; - this.localSettings.levelColors[setting.index] = { - value: attrValue, - color: setting.color - }; - } - } - } - } - this.updateLevelColors(this.localSettings.levelColors); + this.updateAttribute(subscription.data, typeAttributes); } } }; - this.ctx.subscriptionApi.createSubscription(levelColorsSourcesSubscriptionOptions, true).then( - (subscription) => { - TbCanvasDigitalGauge.levelColorSourcesSubscription = subscription; + + return this.ctx.subscriptionApi.createSubscription(levelColorsSourcesSubscriptionOptions, true); + } + + updateAttribute(data, typeAttributes) { + for (let i = 0; i < data.length; i++) { + let keyData = data[i]; + if (keyData && keyData.data && keyData.data[0]) { + let attrValue = keyData.data[0][1]; + if (isFinite(attrValue)) { + for (let i = 0; i < keyData.dataKey.settings.length; i++) { + let setting = keyData.dataKey.settings[i]; + switch (typeAttributes) { + case 'levelColors': + this.localSettings.levelColors[setting.index] = { + value: attrValue, + color: setting.color + }; + break; + case 'ticks': + this.localSettings.ticks[setting] = attrValue; + break; + } + } + } } - ); + } + this.updateSetting(); + } + + updateSetting() { + this.gauge.options.ticks = this.localSettings.ticks; + this.gauge.options.levelColors = this.localSettings.levelColors; + this.gauge.options = CanvasDigitalGauge.configure(this.gauge.options); + this.gauge.update(); } update() { @@ -526,6 +586,48 @@ export default class TbCanvasDigitalGauge { } } }, + "showTicks": { + "title": "Show ticks", + "type": "boolean", + "default": false + }, + "tickWidth": { + "title": "Width ticks", + "type": "number", + "default": 4 + }, + "colorTicks": { + "title": "Color ticks", + "type": "string", + "default": "#666" + }, + "ticksValue": { + "title": "The ticks predefined value", + "type": "array", + "items": { + "title": "tickValue", + "type": "object", + "properties": { + "valueSource": { + "title": "Value source", + "type": "string", + "default": "predefinedValue" + }, + "entityAlias": { + "title": "Source entity alias", + "type": "string" + }, + "attribute": { + "title": "Source entity attribute", + "type": "string" + }, + "value": { + "title": "Value (if predefined value is selected)", + "type": "number" + } + } + } + }, "animation": { "title": "Enable animation", "type": "boolean", @@ -771,6 +873,40 @@ export default class TbCanvasDigitalGauge { } ] }, + "showTicks", + { + "key": "tickWidth", + "condition": "model.showTicks === true" + }, + { + "key": "colorTicks", + "condition": "model.showTicks === true", + "type": "color" + }, + { + "key": "ticksValue", + "condition": "model.showTicks === true", + "items": [ + { + "key": "ticksValue[].valueSource", + "type": "rc-select", + "multiple": false, + "items": [ + { + "value": "predefinedValue", + "label": "Predefined value (Default)" + }, + { + "value": "entityAttribute", + "label": "Value taken from entity attribute" + } + ] + }, + "ticksValue[].value", + "ticksValue[].entityAlias", + "ticksValue[].attribute" + ] + }, "animation", "animationDuration", {