Browse Source

Merge with develop

pull/2687/head
Andrii Shvaika 7 years ago
parent
commit
8abb718349
  1. 1274
      application/src/main/data/json/demo/dashboards/gateways.json
  2. 520
      application/src/main/data/json/demo/dashboards/rule_engine_statistics.json
  3. 1236
      application/src/main/data/json/demo/dashboards/theromstats.json
  4. 146
      application/src/main/data/json/demo/rule_chains/root_rule_chain.json
  5. 137
      application/src/main/data/json/demo/rule_chains/thermostat_alarms.json
  6. 25
      application/src/main/data/upgrade/2.4.3/schema_update_ttl.sql
  7. 3
      application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java
  8. 2
      application/src/main/java/org/thingsboard/server/controller/QueueController.java
  9. 14
      application/src/main/java/org/thingsboard/server/controller/RuleChainController.java
  10. 5
      application/src/main/java/org/thingsboard/server/controller/TelemetryController.java
  11. 2
      application/src/main/java/org/thingsboard/server/install/ThingsboardInstallService.java
  12. 34
      application/src/main/java/org/thingsboard/server/service/install/DefaultSystemDataLoaderService.java
  13. 27
      application/src/main/java/org/thingsboard/server/service/install/InstallScripts.java
  14. 5
      application/src/main/java/org/thingsboard/server/service/install/SqlDatabaseUpgradeService.java
  15. 15
      application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java
  16. 8
      application/src/main/java/org/thingsboard/server/service/queue/DefaultTbRuleEngineConsumerService.java
  17. 2
      application/src/main/java/org/thingsboard/server/service/queue/TbCoreConsumerStats.java
  18. 8
      application/src/main/java/org/thingsboard/server/service/queue/TbMsgPackCallback.java
  19. 4
      application/src/main/java/org/thingsboard/server/service/queue/TbMsgPackProcessingContext.java
  20. 28
      application/src/main/java/org/thingsboard/server/service/queue/TbPackCallback.java
  21. 90
      application/src/main/java/org/thingsboard/server/service/queue/TbPackProcessingContext.java
  22. 11
      application/src/main/java/org/thingsboard/server/service/queue/processing/AbstractConsumerService.java
  23. 6
      application/src/main/java/org/thingsboard/server/service/queue/processing/TbRuleEngineProcessingResult.java
  24. 4
      application/src/main/java/org/thingsboard/server/service/queue/processing/TbRuleEngineSubmitStrategyFactory.java
  25. 2
      application/src/main/java/org/thingsboard/server/service/script/RemoteJsInvokeService.java
  26. 8
      application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetryWebSocketService.java
  27. 40
      application/src/main/java/org/thingsboard/server/service/ttl/AbstractCleanUpService.java
  28. 59
      application/src/main/java/org/thingsboard/server/service/ttl/events/EventsCleanUpService.java
  29. 49
      application/src/main/java/org/thingsboard/server/service/ttl/timeseries/AbstractTimeseriesCleanUpService.java
  30. 2
      application/src/main/java/org/thingsboard/server/service/ttl/timeseries/PsqlTimeseriesCleanUpService.java
  31. 2
      application/src/main/java/org/thingsboard/server/service/ttl/timeseries/TimescaleTimeseriesCleanUpService.java
  32. 84
      application/src/main/resources/thingsboard.yml
  33. 5
      application/src/test/java/org/thingsboard/server/mqtt/telemetry/AbstractMqttTelemetryIntegrationTest.java
  34. 21
      application/src/test/java/org/thingsboard/server/service/cluster/routing/ConsistentHashParitionServiceTest.java
  35. 4
      application/src/test/java/org/thingsboard/server/service/queue/TbMsgPackProcessingContextTest.java
  36. 2
      application/src/test/resources/logback.xml
  37. 12
      common/queue/src/main/java/org/thingsboard/server/queue/discovery/ConsistentHashPartitionService.java
  38. 6
      common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java
  39. 3
      dao/src/main/java/org/thingsboard/server/dao/model/ModelConstants.java
  40. 14
      dao/src/main/java/org/thingsboard/server/dao/model/sql/EventEntity.java
  41. 25
      dao/src/main/java/org/thingsboard/server/dao/rule/BaseRuleChainService.java
  42. 3
      dao/src/main/java/org/thingsboard/server/dao/sql/event/AbstractEventInsertRepository.java
  43. 4
      dao/src/main/java/org/thingsboard/server/dao/sql/event/HsqlEventInsertRepository.java
  44. 2
      dao/src/main/java/org/thingsboard/server/dao/sql/event/PsqlEventInsertRepository.java
  45. 2
      dao/src/main/resources/sql/schema-entities-hsql.sql
  46. 26
      dao/src/main/resources/sql/schema-entities.sql
  47. 2
      dao/src/main/resources/sql/schema-timescale.sql
  48. 2
      dao/src/main/resources/sql/schema-ts-psql.sql
  49. 3
      dao/src/test/resources/sql-test.properties
  50. 2
      docker/README.md
  51. 18
      docker/docker-compose.cassandra.yml
  52. 10
      docker/docker-compose.postgres.volumes.yml
  53. 18
      docker/docker-compose.postgres.yml
  54. 79
      docker/docker-compose.yml
  55. 2
      docker/docker-install-tb.sh
  56. 4
      docker/docker-upgrade-tb.sh
  57. 4
      docker/haproxy/config/haproxy.cfg
  58. 2
      docker/kafka.env
  59. 3
      docker/tb-coap-transport.env
  60. 3
      docker/tb-http-transport.env
  61. 2
      docker/tb-js-executor.env
  62. 3
      docker/tb-mqtt-transport.env
  63. 2
      docker/tb-node.env
  64. 4
      docker/tb-node/conf/logback.xml
  65. 2
      docker/tb-node/conf/thingsboard.conf
  66. 4
      docker/tb-transports/coap/conf/logback.xml
  67. 2
      docker/tb-transports/coap/conf/tb-coap-transport.conf
  68. 4
      docker/tb-transports/http/conf/logback.xml
  69. 2
      docker/tb-transports/http/conf/tb-http-transport.conf
  70. 4
      docker/tb-transports/mqtt/conf/logback.xml
  71. 2
      docker/tb-transports/mqtt/conf/tb-mqtt-transport.conf
  72. 6
      k8s/tb-coap-transport-configmap.yml
  73. 6
      k8s/tb-http-transport-configmap.yml
  74. 6
      k8s/tb-mqtt-transport-configmap.yml
  75. 6
      k8s/tb-node-configmap.yml
  76. 36
      k8s/thingsboard.yml
  77. 2
      msa/black-box-tests/src/test/java/org/thingsboard/server/msa/ThingsBoardDbInstaller.java
  78. 2
      rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/TbGetDeviceAttrNode.java
  79. 2
      rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/transaction/TbSynchronizationBeginNode.java
  80. 2
      rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/transaction/TbSynchronizationEndNode.java

1274
application/src/main/data/json/demo/dashboards/gateways.json

File diff suppressed because it is too large

520
application/src/main/data/json/demo/dashboards/rule_engine_statistics.json

@ -0,0 +1,520 @@
{
"title": "Rule Engine Statistics",
"configuration": {
"widgets": {
"81987f19-3eac-e4ce-b790-d96e9b54d9a0": {
"isSystemType": true,
"bundleAlias": "charts",
"typeAlias": "basic_timeseries",
"type": "timeseries",
"title": "New widget",
"sizeX": 12,
"sizeY": 7,
"config": {
"datasources": [
{
"type": "entity",
"dataKeys": [
{
"name": "successfulMsgs",
"type": "timeseries",
"label": "${entityName} Successful",
"color": "#4caf50",
"settings": {
"excludeFromStacking": false,
"hideDataByDefault": false,
"disableDataHiding": false,
"removeFromLegend": false,
"showLines": true,
"fillLines": false,
"showPoints": false,
"showPointShape": "circle",
"pointShapeFormatter": "var size = radius * Math.sqrt(Math.PI) / 2;\nctx.moveTo(x - size, y - size);\nctx.lineTo(x + size, y + size);\nctx.moveTo(x - size, y + size);\nctx.lineTo(x + size, y - size);",
"showPointsLineWidth": 5,
"showPointsRadius": 3,
"showSeparateAxis": false,
"axisPosition": "left",
"thresholds": [
{
"thresholdValueSource": "predefinedValue"
}
],
"comparisonSettings": {
"showValuesForComparison": true
}
},
"_hash": 0.15490750967648736
},
{
"name": "failedMsgs",
"type": "timeseries",
"label": "${entityName} Permanent Failures",
"color": "#ef5350",
"settings": {
"excludeFromStacking": false,
"hideDataByDefault": false,
"disableDataHiding": false,
"removeFromLegend": false,
"showLines": true,
"fillLines": false,
"showPoints": false,
"showPointShape": "circle",
"pointShapeFormatter": "var size = radius * Math.sqrt(Math.PI) / 2;\nctx.moveTo(x - size, y - size);\nctx.lineTo(x + size, y + size);\nctx.moveTo(x - size, y + size);\nctx.lineTo(x + size, y - size);",
"showPointsLineWidth": 5,
"showPointsRadius": 3,
"showSeparateAxis": false,
"axisPosition": "left",
"thresholds": [
{
"thresholdValueSource": "predefinedValue"
}
],
"comparisonSettings": {
"showValuesForComparison": true
}
},
"_hash": 0.4186621166514697
},
{
"name": "tmpFailed",
"type": "timeseries",
"label": "${entityName} Processing Failures",
"color": "#ffc107",
"settings": {
"excludeFromStacking": false,
"hideDataByDefault": false,
"disableDataHiding": false,
"removeFromLegend": false,
"showLines": true,
"fillLines": false,
"showPoints": false,
"showPointShape": "circle",
"pointShapeFormatter": "var size = radius * Math.sqrt(Math.PI) / 2;\nctx.moveTo(x - size, y - size);\nctx.lineTo(x + size, y + size);\nctx.moveTo(x - size, y + size);\nctx.lineTo(x + size, y - size);",
"showPointsLineWidth": 5,
"showPointsRadius": 3,
"showSeparateAxis": false,
"axisPosition": "left",
"thresholds": [
{
"thresholdValueSource": "predefinedValue"
}
],
"comparisonSettings": {
"showValuesForComparison": true
}
},
"_hash": 0.49891007198715376
}
],
"entityAliasId": "140f23dd-e3a0-ed98-6189-03c49d2d8018"
}
],
"timewindow": {
"realtime": {
"interval": 1000,
"timewindowMs": 300000
},
"aggregation": {
"type": "NONE",
"limit": 8640
},
"hideInterval": false,
"hideAggregation": false,
"hideAggInterval": false
},
"showTitle": true,
"backgroundColor": "#fff",
"color": "rgba(0, 0, 0, 0.87)",
"padding": "8px",
"settings": {
"shadowSize": 4,
"fontColor": "#545454",
"fontSize": 10,
"xaxis": {
"showLabels": true,
"color": "#545454"
},
"yaxis": {
"showLabels": true,
"color": "#545454"
},
"grid": {
"color": "#545454",
"tickColor": "#DDDDDD",
"verticalLines": true,
"horizontalLines": true,
"outlineWidth": 1
},
"stack": false,
"tooltipIndividual": false,
"timeForComparison": "months",
"xaxisSecond": {
"axisPosition": "top",
"showLabels": true
}
},
"title": "Queue Stats",
"dropShadow": true,
"enableFullscreen": true,
"titleStyle": {
"fontSize": "16px",
"fontWeight": 400
},
"mobileHeight": null,
"showTitleIcon": false,
"titleIcon": null,
"iconColor": "rgba(0, 0, 0, 0.87)",
"iconSize": "24px",
"titleTooltip": "",
"widgetStyle": {},
"useDashboardTimewindow": false,
"displayTimewindow": true,
"showLegend": true,
"actions": {},
"legendConfig": {
"direction": "column",
"position": "bottom",
"showMin": true,
"showMax": true,
"showAvg": false,
"showTotal": true
}
},
"id": "81987f19-3eac-e4ce-b790-d96e9b54d9a0"
},
"5eb79712-5c24-3060-7e4f-6af36b8f842d": {
"isSystemType": true,
"bundleAlias": "cards",
"typeAlias": "timeseries_table",
"type": "timeseries",
"title": "New widget",
"sizeX": 24,
"sizeY": 5,
"config": {
"datasources": [
{
"type": "entity",
"dataKeys": [
{
"name": "ruleEngineException",
"type": "timeseries",
"label": "Rule Chain",
"color": "#2196f3",
"settings": {
"useCellStyleFunction": false,
"useCellContentFunction": true,
"cellContentFunction": "return JSON.parse(value).ruleChainName;"
},
"_hash": 0.9954481282345906
},
{
"name": "ruleEngineException",
"type": "timeseries",
"label": "Rule Node",
"color": "#4caf50",
"settings": {
"useCellStyleFunction": false,
"useCellContentFunction": true,
"cellContentFunction": "return JSON.parse(value).ruleNodeName;"
},
"_hash": 0.18580357036589978
},
{
"name": "ruleEngineException",
"type": "timeseries",
"label": "Latest Error",
"color": "#f44336",
"settings": {
"useCellStyleFunction": false,
"useCellContentFunction": true,
"cellContentFunction": "return JSON.parse(value).message;"
},
"_hash": 0.7255162989552142
}
],
"entityAliasId": "140f23dd-e3a0-ed98-6189-03c49d2d8018"
}
],
"timewindow": {
"realtime": {
"interval": 1000,
"timewindowMs": 86400000
},
"aggregation": {
"type": "NONE",
"limit": 200
}
},
"showTitle": true,
"backgroundColor": "rgb(255, 255, 255)",
"color": "rgba(0, 0, 0, 0.87)",
"padding": "8px",
"settings": {
"showTimestamp": true,
"displayPagination": true,
"defaultPageSize": 10
},
"title": "Exceptions",
"dropShadow": true,
"enableFullscreen": true,
"titleStyle": {
"fontSize": "16px",
"fontWeight": 400
},
"useDashboardTimewindow": false,
"showLegend": false,
"widgetStyle": {},
"actions": {},
"showTitleIcon": false,
"titleIcon": null,
"iconColor": "rgba(0, 0, 0, 0.87)",
"iconSize": "24px",
"titleTooltip": "",
"displayTimewindow": true
},
"id": "5eb79712-5c24-3060-7e4f-6af36b8f842d"
},
"ad3f1417-87a8-750e-fc67-49a2de1466d4": {
"isSystemType": true,
"bundleAlias": "charts",
"typeAlias": "basic_timeseries",
"type": "timeseries",
"title": "New widget",
"sizeX": 12,
"sizeY": 7,
"config": {
"datasources": [
{
"type": "entity",
"dataKeys": [
{
"name": "timeoutMsgs",
"type": "timeseries",
"label": "${entityName} Permanent Timeouts",
"color": "#4caf50",
"settings": {
"excludeFromStacking": false,
"hideDataByDefault": false,
"disableDataHiding": false,
"removeFromLegend": false,
"showLines": true,
"fillLines": false,
"showPoints": false,
"showPointShape": "circle",
"pointShapeFormatter": "var size = radius * Math.sqrt(Math.PI) / 2;\nctx.moveTo(x - size, y - size);\nctx.lineTo(x + size, y + size);\nctx.moveTo(x - size, y + size);\nctx.lineTo(x + size, y - size);",
"showPointsLineWidth": 5,
"showPointsRadius": 3,
"showSeparateAxis": false,
"axisPosition": "left",
"thresholds": [
{
"thresholdValueSource": "predefinedValue"
}
],
"comparisonSettings": {
"showValuesForComparison": true
}
},
"_hash": 0.565222981550328
},
{
"name": "tmpTimeout",
"type": "timeseries",
"label": "${entityName} Processing Timeouts",
"color": "#9c27b0",
"settings": {
"excludeFromStacking": false,
"hideDataByDefault": false,
"disableDataHiding": false,
"removeFromLegend": false,
"showLines": true,
"fillLines": false,
"showPoints": false,
"showPointShape": "circle",
"pointShapeFormatter": "var size = radius * Math.sqrt(Math.PI) / 2;\nctx.moveTo(x - size, y - size);\nctx.lineTo(x + size, y + size);\nctx.moveTo(x - size, y + size);\nctx.lineTo(x + size, y - size);",
"showPointsLineWidth": 5,
"showPointsRadius": 3,
"showSeparateAxis": false,
"axisPosition": "left",
"thresholds": [
{
"thresholdValueSource": "predefinedValue"
}
],
"comparisonSettings": {
"showValuesForComparison": true
}
},
"_hash": 0.2679547062508352
}
],
"entityAliasId": "140f23dd-e3a0-ed98-6189-03c49d2d8018"
}
],
"timewindow": {
"realtime": {
"interval": 1000,
"timewindowMs": 300000
},
"aggregation": {
"type": "NONE",
"limit": 8640
},
"hideInterval": false,
"hideAggregation": false,
"hideAggInterval": false
},
"showTitle": true,
"backgroundColor": "#fff",
"color": "rgba(0, 0, 0, 0.87)",
"padding": "8px",
"settings": {
"shadowSize": 4,
"fontColor": "#545454",
"fontSize": 10,
"xaxis": {
"showLabels": true,
"color": "#545454"
},
"yaxis": {
"showLabels": true,
"color": "#545454"
},
"grid": {
"color": "#545454",
"tickColor": "#DDDDDD",
"verticalLines": true,
"horizontalLines": true,
"outlineWidth": 1
},
"stack": false,
"tooltipIndividual": false,
"timeForComparison": "months",
"xaxisSecond": {
"axisPosition": "top",
"showLabels": true
}
},
"title": "Processing Failures and Timeouts",
"dropShadow": true,
"enableFullscreen": true,
"titleStyle": {
"fontSize": "16px",
"fontWeight": 400
},
"mobileHeight": null,
"showTitleIcon": false,
"titleIcon": null,
"iconColor": "rgba(0, 0, 0, 0.87)",
"iconSize": "24px",
"titleTooltip": "",
"widgetStyle": {},
"useDashboardTimewindow": false,
"displayTimewindow": true,
"showLegend": true,
"actions": {},
"legendConfig": {
"direction": "column",
"position": "bottom",
"showMin": true,
"showMax": true,
"showAvg": false,
"showTotal": true
}
},
"id": "ad3f1417-87a8-750e-fc67-49a2de1466d4"
}
},
"states": {
"default": {
"name": "Rule Engine Statistics",
"root": true,
"layouts": {
"main": {
"widgets": {
"81987f19-3eac-e4ce-b790-d96e9b54d9a0": {
"sizeX": 12,
"sizeY": 7,
"mobileHeight": null,
"row": 0,
"col": 0
},
"5eb79712-5c24-3060-7e4f-6af36b8f842d": {
"sizeX": 24,
"sizeY": 5,
"row": 7,
"col": 0
},
"ad3f1417-87a8-750e-fc67-49a2de1466d4": {
"sizeX": 12,
"sizeY": 7,
"mobileHeight": null,
"row": 0,
"col": 12
}
},
"gridSettings": {
"backgroundColor": "#eeeeee",
"color": "rgba(0,0,0,0.870588)",
"columns": 24,
"margins": [
10,
10
],
"backgroundSizeMode": "100%",
"autoFillHeight": true,
"mobileAutoFillHeight": false,
"mobileRowHeight": 70
}
}
}
}
},
"entityAliases": {
"140f23dd-e3a0-ed98-6189-03c49d2d8018": {
"id": "140f23dd-e3a0-ed98-6189-03c49d2d8018",
"alias": "TbServiceQueues",
"filter": {
"type": "assetType",
"resolveMultiple": true,
"assetType": "TbServiceQueue",
"assetNameFilter": ""
}
}
},
"timewindow": {
"displayValue": "",
"selectedTab": 0,
"hideInterval": false,
"hideAggregation": false,
"hideAggInterval": false,
"realtime": {
"interval": 1000,
"timewindowMs": 60000
},
"history": {
"historyType": 0,
"interval": 1000,
"timewindowMs": 60000,
"fixedTimewindow": {
"startTimeMs": 1586176634823,
"endTimeMs": 1586263034823
}
},
"aggregation": {
"type": "AVG",
"limit": 25000
}
},
"settings": {
"stateControllerId": "entity",
"showTitle": false,
"showDashboardsSelect": true,
"showEntitiesSelect": true,
"showDashboardTimewindow": true,
"showDashboardExport": true,
"toolbarAlwaysOpen": true
}
},
"name": "Rule Engine Statistics"
}

1236
application/src/main/data/json/demo/dashboards/theromstats.json

File diff suppressed because it is too large

146
application/src/main/data/json/demo/rule_chains/root_rule_chain.json

@ -0,0 +1,146 @@
{
"ruleChain": {
"additionalInfo": null,
"name": "Root Rule Chain",
"firstRuleNodeId": null,
"root": true,
"debugMode": false,
"configuration": null
},
"metadata": {
"firstNodeIndex": 2,
"nodes": [
{
"additionalInfo": {
"layoutX": 824,
"layoutY": 156
},
"type": "org.thingsboard.rule.engine.telemetry.TbMsgTimeseriesNode",
"name": "Save Timeseries",
"debugMode": false,
"configuration": {
"defaultTTL": 0
}
},
{
"additionalInfo": {
"layoutX": 825,
"layoutY": 52
},
"type": "org.thingsboard.rule.engine.telemetry.TbMsgAttributesNode",
"name": "Save Client Attributes",
"debugMode": false,
"configuration": {
"scope": "CLIENT_SCOPE"
}
},
{
"additionalInfo": {
"layoutX": 347,
"layoutY": 149
},
"type": "org.thingsboard.rule.engine.filter.TbMsgTypeSwitchNode",
"name": "Message Type Switch",
"debugMode": false,
"configuration": {
"version": 0
}
},
{
"additionalInfo": {
"layoutX": 825,
"layoutY": 266
},
"type": "org.thingsboard.rule.engine.action.TbLogNode",
"name": "Log RPC from Device",
"debugMode": false,
"configuration": {
"jsScript": "return '\\nIncoming message:\\n' + JSON.stringify(msg) + '\\nIncoming metadata:\\n' + JSON.stringify(metadata);"
}
},
{
"additionalInfo": {
"layoutX": 825,
"layoutY": 379
},
"type": "org.thingsboard.rule.engine.action.TbLogNode",
"name": "Log Other",
"debugMode": false,
"configuration": {
"jsScript": "return '\\nIncoming message:\\n' + JSON.stringify(msg) + '\\nIncoming metadata:\\n' + JSON.stringify(metadata);"
}
},
{
"additionalInfo": {
"layoutX": 825,
"layoutY": 468
},
"type": "org.thingsboard.rule.engine.rpc.TbSendRPCRequestNode",
"name": "RPC Call Request",
"debugMode": false,
"configuration": {
"timeoutInSeconds": 60
}
},
{
"additionalInfo": {
"layoutX": 1069,
"layoutY": 90
},
"type": "org.thingsboard.rule.engine.filter.TbJsFilterNode",
"name": "Is Thermostat?",
"debugMode": false,
"configuration": {
"jsScript": "return metadata[\"deviceType\"] === \"thermostat\";"
}
}
],
"connections": [
{
"fromIndex": 0,
"toIndex": 6,
"type": "Success"
},
{
"fromIndex": 2,
"toIndex": 4,
"type": "Other"
},
{
"fromIndex": 2,
"toIndex": 1,
"type": "Post attributes"
},
{
"fromIndex": 2,
"toIndex": 0,
"type": "Post telemetry"
},
{
"fromIndex": 2,
"toIndex": 3,
"type": "RPC Request from Device"
},
{
"fromIndex": 2,
"toIndex": 5,
"type": "RPC Request to Device"
}
],
"ruleChainConnections": [
{
"fromIndex": 6,
"targetRuleChainId": {
"entityType": "RULE_CHAIN",
"id": "83d42540-85fd-11ea-aee2-794850541ced"
},
"additionalInfo": {
"layoutX": 1088,
"layoutY": 203,
"ruleChainNodeId": "rule-chain-node-9"
},
"type": "True"
}
]
}
}

137
application/src/main/data/json/demo/rule_chains/thermostat_alarms.json

@ -0,0 +1,137 @@
{
"ruleChain": {
"additionalInfo": null,
"name": "Thermostat Alarms",
"firstRuleNodeId": null,
"root": false,
"debugMode": false,
"configuration": null
},
"metadata": {
"firstNodeIndex": 5,
"nodes": [
{
"additionalInfo": {
"layoutX": 929,
"layoutY": 67
},
"type": "org.thingsboard.rule.engine.action.TbCreateAlarmNode",
"name": "Create Temp Alarm",
"debugMode": false,
"configuration": {
"alarmType": "High Temperature",
"alarmDetailsBuildJs": "var details = {};\nif (metadata.prevAlarmDetails) {\n details = JSON.parse(metadata.prevAlarmDetails);\n}\ndetails.triggerValue = msg.temperature;\nreturn details;",
"severity": "MAJOR",
"propagate": true,
"useMessageAlarmData": false,
"relationTypes": []
}
},
{
"additionalInfo": {
"layoutX": 930,
"layoutY": 201
},
"type": "org.thingsboard.rule.engine.action.TbClearAlarmNode",
"name": "Clear Temp Alarm",
"debugMode": false,
"configuration": {
"alarmType": "High Temperature",
"alarmDetailsBuildJs": "var details = {};\nif (metadata.prevAlarmDetails) {\n details = JSON.parse(metadata.prevAlarmDetails);\n}\nreturn details;"
}
},
{
"additionalInfo": {
"layoutX": 930,
"layoutY": 131
},
"type": "org.thingsboard.rule.engine.action.TbCreateAlarmNode",
"name": "Create Humidity Alarm",
"debugMode": false,
"configuration": {
"alarmType": "Low Humidity",
"alarmDetailsBuildJs": "var details = {};\nif (metadata.prevAlarmDetails) {\n details = JSON.parse(metadata.prevAlarmDetails);\n}\ndetails.triggerValue = msg.humidity;\nreturn details;",
"severity": "MINOR",
"propagate": true,
"useMessageAlarmData": false,
"relationTypes": []
}
},
{
"additionalInfo": {
"layoutX": 929,
"layoutY": 275
},
"type": "org.thingsboard.rule.engine.action.TbClearAlarmNode",
"name": "Clear Humidity Alarm",
"debugMode": false,
"configuration": {
"alarmType": "Low Humidity",
"alarmDetailsBuildJs": "var details = {};\nif (metadata.prevAlarmDetails) {\n details = JSON.parse(metadata.prevAlarmDetails);\n}\nreturn details;"
}
},
{
"additionalInfo": {
"layoutX": 586,
"layoutY": 148
},
"type": "org.thingsboard.rule.engine.filter.TbJsSwitchNode",
"name": "Check Alarms",
"debugMode": true,
"configuration": {
"jsScript": "var relations = [];\nif(metadata[\"ss_alarmTemperature\"] === \"true\"){\n if(msg.temperature > metadata[\"ss_thresholdTemperature\"]){\n relations.push(\"NewTempAlarm\");\n } else {\n relations.push(\"ClearTempAlarm\");\n }\n}\nif(metadata[\"ss_alarmHumidity\"] === \"true\"){\n if(msg.humidity < metadata[\"ss_thresholdHumidity\"]){\n relations.push(\"NewHumidityAlarm\");\n } else {\n relations.push(\"ClearHumidityAlarm\");\n }\n}\n\nreturn relations;"
}
},
{
"additionalInfo": {
"layoutX": 321,
"layoutY": 149
},
"type": "org.thingsboard.rule.engine.metadata.TbGetAttributesNode",
"name": "Fetch Configuration",
"debugMode": true,
"configuration": {
"clientAttributeNames": [],
"sharedAttributeNames": [],
"serverAttributeNames": [
"alarmTemperature",
"thresholdTemperature",
"alarmHumidity",
"thresholdHumidity"
],
"latestTsKeyNames": [],
"tellFailureIfAbsent": false,
"getLatestValueWithTs": false
}
}
],
"connections": [
{
"fromIndex": 4,
"toIndex": 0,
"type": "NewTempAlarm"
},
{
"fromIndex": 4,
"toIndex": 1,
"type": "ClearTempAlarm"
},
{
"fromIndex": 4,
"toIndex": 2,
"type": "NewHumidityAlarm"
},
{
"fromIndex": 4,
"toIndex": 3,
"type": "ClearHumidityAlarm"
},
{
"fromIndex": 5,
"toIndex": 4,
"type": "Success"
}
],
"ruleChainConnections": null
}
}

25
application/src/main/data/upgrade/2.4.3/schema_update_ttl.sql

@ -123,3 +123,28 @@ BEGIN
END LOOP;
END
$$;
CREATE OR REPLACE PROCEDURE cleanup_events_by_ttl(IN ttl bigint, IN debug_ttl bigint, INOUT deleted bigint)
LANGUAGE plpgsql AS
$$
DECLARE
ttl_ts bigint;
debug_ttl_ts bigint;
ttl_deleted_count bigint DEFAULT 0;
debug_ttl_deleted_count bigint DEFAULT 0;
BEGIN
IF ttl > 0 THEN
ttl_ts := (EXTRACT(EPOCH FROM current_timestamp) * 1000 - ttl::bigint * 1000)::bigint;
EXECUTE format(
'WITH deleted AS (DELETE FROM event WHERE ts < %L::bigint AND (event_type != %L::varchar AND event_type != %L::varchar) RETURNING *) SELECT count(*) FROM deleted', ttl_ts, 'DEBUG_RULE_NODE', 'DEBUG_RULE_CHAIN') into ttl_deleted_count;
END IF;
IF debug_ttl > 0 THEN
debug_ttl_ts := (EXTRACT(EPOCH FROM current_timestamp) * 1000 - debug_ttl::bigint * 1000)::bigint;
EXECUTE format(
'WITH deleted AS (DELETE FROM event WHERE ts < %L::bigint AND (event_type = %L::varchar OR event_type = %L::varchar) RETURNING *) SELECT count(*) FROM deleted', debug_ttl_ts, 'DEBUG_RULE_NODE', 'DEBUG_RULE_CHAIN') into debug_ttl_deleted_count;
END IF;
RAISE NOTICE 'Events removed by ttl: %', ttl_deleted_count;
RAISE NOTICE 'Debug Events removed by ttl: %', debug_ttl_deleted_count;
deleted := ttl_deleted_count + debug_ttl_deleted_count;
END
$$;

3
application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java

@ -32,6 +32,7 @@ import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Lazy;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
@ -233,6 +234,7 @@ public class ActorSystemContext {
/**
* The following Service will be null if we operate in tb-core mode
*/
@Lazy
@Autowired(required = false)
@Getter
private TbRuleEngineDeviceRpcService tbRuleEngineDeviceRpcService;
@ -240,6 +242,7 @@ public class ActorSystemContext {
/**
* The following Service will be null if we operate in tb-rule-engine mode
*/
@Lazy
@Autowired(required = false)
@Getter
private TbCoreDeviceRpcService tbCoreDeviceRpcService;

2
application/src/main/java/org/thingsboard/server/controller/QueueController.java

@ -43,7 +43,7 @@ public class QueueController extends BaseController {
ServiceType type = ServiceType.valueOf(serviceType);
switch (type) {
case TB_RULE_ENGINE:
return Arrays.asList("HighPriority", "Main");
return Arrays.asList("Main", "HighPriority", "SequentialByOriginator");
default:
return Collections.emptyList();
}

14
application/src/main/java/org/thingsboard/server/controller/RuleChainController.java

@ -161,15 +161,15 @@ public class RuleChainController extends BaseController {
TenantId tenantId = getCurrentUser().getTenantId();
RuleChain previousRootRuleChain = ruleChainService.getRootTenantRuleChain(tenantId);
if (ruleChainService.setRootRuleChain(getTenantId(), ruleChainId)) {
if (previousRootRuleChain != null) {
previousRootRuleChain = ruleChainService.findRuleChainById(getTenantId(), previousRootRuleChain.getId());
previousRootRuleChain = ruleChainService.findRuleChainById(getTenantId(), previousRootRuleChain.getId());
tbClusterService.onEntityStateChange(previousRootRuleChain.getTenantId(), previousRootRuleChain.getId(),
ComponentLifecycleEvent.UPDATED);
logEntityAction(previousRootRuleChain.getId(), previousRootRuleChain,
null, ActionType.UPDATED, null);
tbClusterService.onEntityStateChange(previousRootRuleChain.getTenantId(), previousRootRuleChain.getId(),
ComponentLifecycleEvent.UPDATED);
logEntityAction(previousRootRuleChain.getId(), previousRootRuleChain,
null, ActionType.UPDATED, null);
}
ruleChain = ruleChainService.findRuleChainById(getTenantId(), ruleChainId);
tbClusterService.onEntityStateChange(ruleChain.getTenantId(), ruleChain.getId(),

5
application/src/main/java/org/thingsboard/server/controller/TelemetryController.java

@ -397,11 +397,6 @@ public class TelemetryController extends BaseController {
@Override
public void onSuccess(@Nullable Void tmp) {
logAttributesUpdated(user, entityId, scope, attributes, null);
if (entityIdSrc.getEntityType().equals(EntityType.DEVICE)) {
DeviceId deviceId = new DeviceId(entityId.getId());
tbClusterService.pushMsgToCore(DeviceAttributesEventNotificationMsg.onUpdate(
user.getTenantId(), deviceId, scope, attributes), null);
}
result.setResult(new ResponseEntity(HttpStatus.OK));
}

2
application/src/main/java/org/thingsboard/server/install/ThingsboardInstallService.java

@ -23,8 +23,8 @@ import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.Profile;
import org.springframework.stereotype.Service;
import org.thingsboard.server.service.component.ComponentDiscoveryService;
import org.thingsboard.server.service.install.DatabaseTsUpgradeService;
import org.thingsboard.server.service.install.DatabaseEntitiesUpgradeService;
import org.thingsboard.server.service.install.DatabaseTsUpgradeService;
import org.thingsboard.server.service.install.EntityDatabaseSchemaService;
import org.thingsboard.server.service.install.SystemDataLoaderService;
import org.thingsboard.server.service.install.TsDatabaseSchemaService;

34
application/src/main/java/org/thingsboard/server/service/install/DefaultSystemDataLoaderService.java

@ -25,24 +25,32 @@ import org.springframework.security.crypto.bcrypt.BCryptPasswordEncoder;
import org.springframework.stereotype.Service;
import org.thingsboard.server.common.data.AdminSettings;
import org.thingsboard.server.common.data.Customer;
import org.thingsboard.server.common.data.DataConstants;
import org.thingsboard.server.common.data.Device;
import org.thingsboard.server.common.data.Tenant;
import org.thingsboard.server.common.data.User;
import org.thingsboard.server.common.data.id.CustomerId;
import org.thingsboard.server.common.data.id.DeviceId;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.kv.BaseAttributeKvEntry;
import org.thingsboard.server.common.data.kv.BooleanDataEntry;
import org.thingsboard.server.common.data.kv.DoubleDataEntry;
import org.thingsboard.server.common.data.kv.LongDataEntry;
import org.thingsboard.server.common.data.security.Authority;
import org.thingsboard.server.common.data.security.DeviceCredentials;
import org.thingsboard.server.common.data.security.UserCredentials;
import org.thingsboard.server.common.data.widget.WidgetsBundle;
import org.thingsboard.server.dao.attributes.AttributesService;
import org.thingsboard.server.dao.customer.CustomerService;
import org.thingsboard.server.dao.device.DeviceCredentialsService;
import org.thingsboard.server.dao.device.DeviceService;
import org.thingsboard.server.dao.model.ModelConstants;
import org.thingsboard.server.dao.settings.AdminSettingsService;
import org.thingsboard.server.dao.tenant.TenantService;
import org.thingsboard.server.dao.user.UserService;
import org.thingsboard.server.dao.widget.WidgetsBundleService;
import java.util.Arrays;
@Service
@Profile("install")
@Slf4j
@ -76,6 +84,9 @@ public class DefaultSystemDataLoaderService implements SystemDataLoaderService {
@Autowired
private DeviceService deviceService;
@Autowired
private AttributesService attributesService;
@Autowired
private DeviceCredentialsService deviceCredentialsService;
@ -120,7 +131,7 @@ public class DefaultSystemDataLoaderService implements SystemDataLoaderService {
demoTenant.setRegion("Global");
demoTenant.setTitle("Tenant");
demoTenant = tenantService.saveTenant(demoTenant);
installScripts.createDefaultRuleChains(demoTenant.getId());
installScripts.loadDemoRuleChains(demoTenant.getId());
createUser(Authority.TENANT_ADMIN, demoTenant.getId(), null, "tenant@thingsboard.org", "tenant");
Customer customerA = new Customer();
@ -152,6 +163,25 @@ public class DefaultSystemDataLoaderService implements SystemDataLoaderService {
createDevice(demoTenant.getId(), null, DEFAULT_DEVICE_TYPE, "Raspberry Pi Demo Device", "RASPBERRY_PI_DEMO_TOKEN", "Demo device that is used in " +
"Raspberry Pi GPIO control sample application");
DeviceId t1Id = createDevice(demoTenant.getId(), null, "thermostat", "Thermostat T1", "T1_TEST_TOKEN", "Demo device for Thermostats dashboard").getId();
DeviceId t2Id = createDevice(demoTenant.getId(), null, "thermostat", "Thermostat T2", "T2_TEST_TOKEN", "Demo device for Thermostats dashboard").getId();
attributesService.save(demoTenant.getId(), t1Id, DataConstants.SERVER_SCOPE,
Arrays.asList(new BaseAttributeKvEntry(System.currentTimeMillis(), new DoubleDataEntry("latitude", 37.3948)),
new BaseAttributeKvEntry(System.currentTimeMillis(), new DoubleDataEntry("longitude", -122.1503)),
new BaseAttributeKvEntry(System.currentTimeMillis(), new BooleanDataEntry("alarmTemperature", true)),
new BaseAttributeKvEntry(System.currentTimeMillis(), new BooleanDataEntry("alarmHumidity", true)),
new BaseAttributeKvEntry(System.currentTimeMillis(), new LongDataEntry("thresholdTemperature", (long) 20)),
new BaseAttributeKvEntry(System.currentTimeMillis(), new LongDataEntry("thresholdHumidity", (long) 50))));
attributesService.save(demoTenant.getId(), t2Id, DataConstants.SERVER_SCOPE,
Arrays.asList(new BaseAttributeKvEntry(System.currentTimeMillis(), new DoubleDataEntry("latitude", 37.493801)),
new BaseAttributeKvEntry(System.currentTimeMillis(), new DoubleDataEntry("longitude", -121.948769)),
new BaseAttributeKvEntry(System.currentTimeMillis(), new BooleanDataEntry("alarmTemperature", true)),
new BaseAttributeKvEntry(System.currentTimeMillis(), new BooleanDataEntry("alarmHumidity", true)),
new BaseAttributeKvEntry(System.currentTimeMillis(), new LongDataEntry("thresholdTemperature", (long) 25)),
new BaseAttributeKvEntry(System.currentTimeMillis(), new LongDataEntry("thresholdHumidity", (long) 30))));
installScripts.loadDashboards(demoTenant.getId(), null);
}

27
application/src/main/java/org/thingsboard/server/service/install/InstallScripts.java

@ -24,6 +24,7 @@ import org.springframework.util.StringUtils;
import org.thingsboard.server.common.data.Dashboard;
import org.thingsboard.server.common.data.id.CustomerId;
import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.id.RuleChainId;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.rule.RuleChain;
import org.thingsboard.server.common.data.rule.RuleChainMetaData;
@ -182,4 +183,30 @@ public class InstallScripts {
}
public void loadDemoRuleChains(TenantId tenantId) throws Exception {
Path ruleChainsDir = Paths.get(getDataDir(), JSON_DIR, DEMO_DIR, RULE_CHAINS_DIR);
try {
JsonNode ruleChainJson = objectMapper.readTree(ruleChainsDir.resolve("thermostat_alarms.json").toFile());
RuleChain ruleChain = objectMapper.treeToValue(ruleChainJson.get("ruleChain"), RuleChain.class);
RuleChainMetaData ruleChainMetaData = objectMapper.treeToValue(ruleChainJson.get("metadata"), RuleChainMetaData.class);
ruleChain.setTenantId(tenantId);
ruleChain = ruleChainService.saveRuleChain(ruleChain);
ruleChainMetaData.setRuleChainId(ruleChain.getId());
ruleChainService.saveRuleChainMetaData(new TenantId(EntityId.NULL_UUID), ruleChainMetaData);
JsonNode rootChainJson = objectMapper.readTree(ruleChainsDir.resolve("root_rule_chain.json").toFile());
RuleChain rootChain = objectMapper.treeToValue(rootChainJson.get("ruleChain"), RuleChain.class);
RuleChainMetaData rootChainMetaData = objectMapper.treeToValue(rootChainJson.get("metadata"), RuleChainMetaData.class);
RuleChainId thermostatsRuleChainId = ruleChain.getId();
rootChainMetaData.getRuleChainConnections().forEach(connection -> connection.setTargetRuleChainId(thermostatsRuleChainId));
rootChain.setTenantId(tenantId);
rootChain = ruleChainService.saveRuleChain(rootChain);
rootChainMetaData.setRuleChainId(rootChain.getId());
ruleChainService.saveRuleChainMetaData(new TenantId(EntityId.NULL_UUID), rootChainMetaData);
} catch (Exception e) {
log.error("Unable to load dashboard from json", e);
throw new RuntimeException("Unable to load dashboard from json", e);
}
}
}

5
application/src/main/java/org/thingsboard/server/service/install/SqlDatabaseUpgradeService.java

@ -225,6 +225,11 @@ public class SqlDatabaseUpgradeService implements DatabaseEntitiesUpgradeService
conn.createStatement().execute("ALTER TABLE tenant ADD COLUMN isolated_tb_core boolean DEFAULT (false), ADD COLUMN isolated_tb_rule_engine boolean DEFAULT (false)");
} catch (Exception e) {
}
try {
long ts = System.currentTimeMillis();
conn.createStatement().execute("ALTER TABLE event ADD COLUMN ts bigint DEFAULT " + ts + ";"); //NOSONAR, ignoring because method used to execute thingsboard database upgrade script
} catch (Exception e) {
}
log.info("Schema updated.");
}
break;

15
application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java

@ -52,6 +52,7 @@ import org.thingsboard.server.service.subscription.TbSubscriptionUtils;
import org.thingsboard.server.service.transport.msg.TransportToDeviceActorMsgWrapper;
import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.List;
import java.util.Optional;
import java.util.UUID;
@ -98,6 +99,11 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService<ToCore
super.init("tb-core-consumer", "tb-core-notifications-consumer");
}
@PreDestroy
public void destroy(){
super.destroy();
}
@Override
public void onApplicationEvent(PartitionChangeEvent partitionChangeEvent) {
if (partitionChangeEvent.getServiceType().equals(getServiceType())) {
@ -117,11 +123,12 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService<ToCore
}
ConcurrentMap<UUID, TbProtoQueueMsg<ToCoreMsg>> pendingMap = msgs.stream().collect(
Collectors.toConcurrentMap(s -> UUID.randomUUID(), Function.identity()));
ConcurrentMap<UUID, TbProtoQueueMsg<ToCoreMsg>> failedMap = new ConcurrentHashMap<>();
CountDownLatch processingTimeoutLatch = new CountDownLatch(1);
TbPackProcessingContext<TbProtoQueueMsg<ToCoreMsg>> ctx = new TbPackProcessingContext<>(
processingTimeoutLatch, pendingMap, new ConcurrentHashMap<>());
pendingMap.forEach((id, msg) -> {
log.trace("[{}] Creating main callback for message: {}", id, msg.getValue());
TbCallback callback = new TbPackCallback<>(id, processingTimeoutLatch, pendingMap, failedMap);
TbCallback callback = new TbPackCallback<>(id, ctx);
try {
ToCoreMsg toCoreMsg = msg.getValue();
if (toCoreMsg.hasToSubscriptionMgrMsg()) {
@ -147,8 +154,8 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService<ToCore
}
});
if (!processingTimeoutLatch.await(packProcessingTimeout, TimeUnit.MILLISECONDS)) {
pendingMap.forEach((id, msg) -> log.warn("[{}] Timeout to process message: {}", id, msg.getValue()));
failedMap.forEach((id, msg) -> log.warn("[{}] Failed to process message: {}", id, msg.getValue()));
ctx.getAckMap().forEach((id, msg) -> log.warn("[{}] Timeout to process message: {}", id, msg.getValue()));
ctx.getFailedMap().forEach((id, msg) -> log.warn("[{}] Failed to process message: {}", id, msg.getValue()));
}
mainConsumer.commit();
} catch (Exception e) {

8
application/src/main/java/org/thingsboard/server/service/queue/DefaultTbRuleEngineConsumerService.java

@ -31,7 +31,6 @@ import org.thingsboard.server.common.msg.queue.ServiceQueue;
import org.thingsboard.server.common.msg.queue.ServiceType;
import org.thingsboard.server.common.msg.queue.TbCallback;
import org.thingsboard.server.common.msg.queue.TbMsgCallback;
import org.thingsboard.server.gen.transport.TransportProtos;
import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg;
import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineNotificationMsg;
import org.thingsboard.server.queue.TbQueueConsumer;
@ -64,9 +63,6 @@ import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.function.BiConsumer;
import java.util.function.Function;
import java.util.stream.Collectors;
@Service
@TbRuleEngineComponent
@ -116,10 +112,10 @@ public class DefaultTbRuleEngineConsumerService extends AbstractConsumerService<
@PreDestroy
public void stop() {
super.destroy();
if (submitExecutor != null) {
submitExecutor.shutdownNow();
}
ruleEngineSettings.getQueues().forEach(config -> consumerConfigurations.put(config.getName(), config));
}
@ -156,7 +152,7 @@ public class DefaultTbRuleEngineConsumerService extends AbstractConsumerService<
submitStrategy.init(msgs);
while (!stopped) {
ProcessingAttemptContext ctx = new ProcessingAttemptContext(submitStrategy);
TbMsgPackProcessingContext ctx = new TbMsgPackProcessingContext(submitStrategy);
submitStrategy.submitAttempt((id, msg) -> submitExecutor.submit(() -> {
log.trace("[{}] Creating callback for message: {}", id, msg.getValue());
ToRuleEngineMsg toRuleEngineMsg = msg.getValue();

2
application/src/main/java/org/thingsboard/server/service/queue/TbCoreConsumerStats.java

@ -79,7 +79,7 @@ public class TbCoreConsumerStats {
public void printStats() {
int total = totalCounter.getAndSet(0);
if (total > 0) {
log.info("Transport total [{}] sessionEvents [{}] getAttr [{}] subToAttr [{}] subToRpc [{}] toDevRpc [{}] subInfo [{}] claimDevice [{}]" +
log.info("Total [{}] sessionEvents [{}] getAttr [{}] subToAttr [{}] subToRpc [{}] toDevRpc [{}] subInfo [{}] claimDevice [{}]" +
" deviceState [{}] subMgr [{}] coreNfs [{}]",
total, sessionEventCounter.getAndSet(0),
getAttributesCounter.getAndSet(0), subscribeToAttributesCounter.getAndSet(0),

8
application/src/main/java/org/thingsboard/server/service/queue/TbMsgPackCallback.java

@ -18,21 +18,17 @@ package org.thingsboard.server.service.queue;
import lombok.extern.slf4j.Slf4j;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.msg.queue.RuleEngineException;
import org.thingsboard.server.common.msg.queue.RuleNodeException;
import org.thingsboard.server.common.msg.queue.TbCallback;
import org.thingsboard.server.common.msg.queue.TbMsgCallback;
import java.util.UUID;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CountDownLatch;
@Slf4j
public class TbMsgPackCallback implements TbMsgCallback {
private final UUID id;
private final TenantId tenantId;
private final ProcessingAttemptContext ctx;
private final TbMsgPackProcessingContext ctx;
public TbMsgPackCallback(UUID id, TenantId tenantId, ProcessingAttemptContext ctx) {
public TbMsgPackCallback(UUID id, TenantId tenantId, TbMsgPackProcessingContext ctx) {
this.id = id;
this.tenantId = tenantId;
this.ctx = ctx;

4
application/src/main/java/org/thingsboard/server/service/queue/ProcessingAttemptContext.java → application/src/main/java/org/thingsboard/server/service/queue/TbMsgPackProcessingContext.java

@ -29,7 +29,7 @@ import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
public class ProcessingAttemptContext {
public class TbMsgPackProcessingContext {
private final TbRuleEngineSubmitStrategy submitStrategy;
@ -44,7 +44,7 @@ public class ProcessingAttemptContext {
@Getter
private final ConcurrentMap<TenantId, RuleEngineException> exceptionsMap = new ConcurrentHashMap<>();
public ProcessingAttemptContext(TbRuleEngineSubmitStrategy submitStrategy) {
public TbMsgPackProcessingContext(TbRuleEngineSubmitStrategy submitStrategy) {
this.submitStrategy = submitStrategy;
this.pendingMap = submitStrategy.getPendingMap();
this.pendingCount = new AtomicInteger(pendingMap.size());

28
application/src/main/java/org/thingsboard/server/service/queue/TbPackCallback.java

@ -19,44 +19,26 @@ import lombok.extern.slf4j.Slf4j;
import org.thingsboard.server.common.msg.queue.TbCallback;
import java.util.UUID;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CountDownLatch;
@Slf4j
public class TbPackCallback<T> implements TbCallback {
private final CountDownLatch processingTimeoutLatch;
private final ConcurrentMap<UUID, T> ackMap;
private final ConcurrentMap<UUID, T> failedMap;
private final TbPackProcessingContext<T> ctx;
private final UUID id;
public TbPackCallback(UUID id,
CountDownLatch processingTimeoutLatch,
ConcurrentMap<UUID, T> ackMap,
ConcurrentMap<UUID, T> failedMap) {
public TbPackCallback(UUID id, TbPackProcessingContext<T> ctx) {
this.id = id;
this.processingTimeoutLatch = processingTimeoutLatch;
this.ackMap = ackMap;
this.failedMap = failedMap;
this.ctx = ctx;
}
@Override
public void onSuccess() {
log.trace("[{}] ON SUCCESS", id);
T msg = ackMap.remove(id);
if (msg != null && ackMap.isEmpty()) {
processingTimeoutLatch.countDown();
}
ctx.onSuccess(id);
}
@Override
public void onFailure(Throwable t) {
log.trace("[{}] ON FAILURE", id, t);
T msg = ackMap.remove(id);
if (msg != null) {
failedMap.put(id, msg);
}
if (ackMap.isEmpty()) {
processingTimeoutLatch.countDown();
}
ctx.onFailure(id, t);
}
}

90
application/src/main/java/org/thingsboard/server/service/queue/TbPackProcessingContext.java

@ -0,0 +1,90 @@
/**
* Copyright © 2016-2020 The Thingsboard Authors
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.thingsboard.server.service.queue;
import lombok.extern.slf4j.Slf4j;
import java.util.UUID;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
@Slf4j
public class TbPackProcessingContext<T> {
private final AtomicInteger pendingCount;
private final CountDownLatch processingTimeoutLatch;
private final ConcurrentMap<UUID, T> ackMap;
private final ConcurrentMap<UUID, T> failedMap;
public TbPackProcessingContext(CountDownLatch processingTimeoutLatch,
ConcurrentMap<UUID, T> ackMap,
ConcurrentMap<UUID, T> failedMap) {
this.processingTimeoutLatch = processingTimeoutLatch;
this.pendingCount = new AtomicInteger(ackMap.size());
this.ackMap = ackMap;
this.failedMap = failedMap;
}
public boolean await(long packProcessingTimeout, TimeUnit milliseconds) throws InterruptedException {
return processingTimeoutLatch.await(packProcessingTimeout, milliseconds);
}
public void onSuccess(UUID id) {
boolean empty = false;
T msg = ackMap.remove(id);
if (msg != null) {
empty = pendingCount.decrementAndGet() == 0;
}
if (empty) {
processingTimeoutLatch.countDown();
} else {
if (log.isTraceEnabled()) {
log.trace("Items left: {}", ackMap.size());
for (T t : ackMap.values()) {
log.trace("left item: {}", t);
}
}
}
}
public void onFailure(UUID id, Throwable t) {
boolean empty = false;
T msg = ackMap.remove(id);
if (msg != null) {
empty = pendingCount.decrementAndGet() == 0;
failedMap.put(id, msg);
if (log.isTraceEnabled()) {
log.trace("Items left: {}", ackMap.size());
for (T v : ackMap.values()) {
log.trace("left item: {}", v);
}
}
}
if (empty) {
processingTimeoutLatch.countDown();
}
}
public ConcurrentMap<UUID, T> getAckMap() {
return ackMap;
}
public ConcurrentMap<UUID, T> getFailedMap() {
return failedMap;
}
}

11
application/src/main/java/org/thingsboard/server/service/queue/processing/AbstractConsumerService.java

@ -23,11 +23,13 @@ import org.thingsboard.common.util.ThingsBoardThreadFactory;
import org.thingsboard.server.actors.ActorSystemContext;
import org.thingsboard.server.common.msg.queue.ServiceType;
import org.thingsboard.server.common.msg.queue.TbCallback;
import org.thingsboard.server.gen.transport.TransportProtos;
import org.thingsboard.server.queue.TbQueueConsumer;
import org.thingsboard.server.queue.common.TbProtoQueueMsg;
import org.thingsboard.server.queue.discovery.PartitionChangeEvent;
import org.thingsboard.server.service.encoding.DataDecodingEncodingService;
import org.thingsboard.server.service.queue.TbPackCallback;
import org.thingsboard.server.service.queue.TbPackProcessingContext;
import javax.annotation.PreDestroy;
import java.util.List;
@ -92,11 +94,12 @@ public abstract class AbstractConsumerService<N extends com.google.protobuf.Gene
}
ConcurrentMap<UUID, TbProtoQueueMsg<N>> pendingMap = msgs.stream().collect(
Collectors.toConcurrentMap(s -> UUID.randomUUID(), Function.identity()));
ConcurrentMap<UUID, TbProtoQueueMsg<N>> failedMap = new ConcurrentHashMap<>();
CountDownLatch processingTimeoutLatch = new CountDownLatch(1);
TbPackProcessingContext<TbProtoQueueMsg<N>> ctx = new TbPackProcessingContext<>(
processingTimeoutLatch, pendingMap, new ConcurrentHashMap<>());
pendingMap.forEach((id, msg) -> {
log.trace("[{}] Creating notification callback for message: {}", id, msg.getValue());
TbCallback callback = new TbPackCallback<>(id, processingTimeoutLatch, pendingMap, failedMap);
TbCallback callback = new TbPackCallback<>(id, ctx);
try {
handleNotification(id, msg, callback);
} catch (Throwable e) {
@ -105,8 +108,8 @@ public abstract class AbstractConsumerService<N extends com.google.protobuf.Gene
}
});
if (!processingTimeoutLatch.await(getNotificationPackProcessingTimeout(), TimeUnit.MILLISECONDS)) {
pendingMap.forEach((id, msg) -> log.warn("[{}] Timeout to process notification: {}", id, msg.getValue()));
failedMap.forEach((id, msg) -> log.warn("[{}] Failed to process notification: {}", id, msg.getValue()));
ctx.getAckMap().forEach((id, msg) -> log.warn("[{}] Timeout to process notification: {}", id, msg.getValue()));
ctx.getFailedMap().forEach((id, msg) -> log.warn("[{}] Failed to process notification: {}", id, msg.getValue()));
}
nfConsumer.commit();
} catch (Exception e) {

6
application/src/main/java/org/thingsboard/server/service/queue/processing/TbRuleEngineProcessingResult.java

@ -20,7 +20,7 @@ import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.msg.queue.RuleEngineException;
import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg;
import org.thingsboard.server.queue.common.TbProtoQueueMsg;
import org.thingsboard.server.service.queue.ProcessingAttemptContext;
import org.thingsboard.server.service.queue.TbMsgPackProcessingContext;
import java.util.UUID;
import java.util.concurrent.ConcurrentMap;
@ -32,9 +32,9 @@ public class TbRuleEngineProcessingResult {
@Getter
private final boolean timeout;
@Getter
private final ProcessingAttemptContext ctx;
private final TbMsgPackProcessingContext ctx;
public TbRuleEngineProcessingResult(boolean timeout, ProcessingAttemptContext ctx) {
public TbRuleEngineProcessingResult(boolean timeout, TbMsgPackProcessingContext ctx) {
this.timeout = timeout;
this.ctx = ctx;
this.success = !timeout && ctx.getPendingMap().isEmpty() && ctx.getFailedMap().isEmpty();

4
application/src/main/java/org/thingsboard/server/service/queue/processing/TbRuleEngineSubmitStrategyFactory.java

@ -29,9 +29,9 @@ public class TbRuleEngineSubmitStrategyFactory {
return new BurstTbRuleEngineSubmitStrategy(name);
case "BATCH":
return new BatchTbRuleEngineSubmitStrategy(name, configuration.getBatchSize());
case "SEQUENTIAL_WITHIN_ORIGINATOR":
case "SEQUENTIAL_BY_ORIGINATOR":
return new SequentialByOriginatorIdTbRuleEngineSubmitStrategy(name);
case "SEQUENTIAL_WITHIN_TENANT":
case "SEQUENTIAL_BY_TENANT":
return new SequentialByTenantIdTbRuleEngineSubmitStrategy(name);
case "SEQUENTIAL":
return new SequentialTbRuleEngineSubmitStrategy(name);

2
application/src/main/java/org/thingsboard/server/service/script/RemoteJsInvokeService.java

@ -46,7 +46,7 @@ import java.util.concurrent.atomic.AtomicInteger;
@Service
public class RemoteJsInvokeService extends AbstractJsInvokeService {
@Value("${js.remote.max_requests_timeout}")
@Value("${queue.js.max_requests_timeout}")
private long maxRequestsTimeout;
@Getter

8
application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetryWebSocketService.java

@ -345,6 +345,8 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi
keys.forEach(key -> subState.put(key, 0L));
attributesData.forEach(v -> subState.put(v.getKey(), v.getTs()));
TbAttributeSubscriptionScope scope = StringUtils.isEmpty(cmd.getScope()) ? TbAttributeSubscriptionScope.SERVER_SCOPE : TbAttributeSubscriptionScope.valueOf(cmd.getScope());
TbAttributeSubscription sub = TbAttributeSubscription.builder()
.serviceId(serviceId)
.sessionId(sessionId)
@ -353,7 +355,7 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi
.entityId(entityId)
.allKeys(false)
.keyStates(subState)
.scope(TbAttributeSubscriptionScope.valueOf(cmd.getScope())).build();
.scope(scope).build();
subService.addSubscription(sub);
}
@ -440,6 +442,8 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi
Map<String, Long> subState = new HashMap<>(attributesData.size());
attributesData.forEach(v -> subState.put(v.getKey(), v.getTs()));
TbAttributeSubscriptionScope scope = StringUtils.isEmpty(cmd.getScope()) ? TbAttributeSubscriptionScope.SERVER_SCOPE : TbAttributeSubscriptionScope.valueOf(cmd.getScope());
TbAttributeSubscription sub = TbAttributeSubscription.builder()
.serviceId(serviceId)
.sessionId(sessionId)
@ -448,7 +452,7 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi
.entityId(entityId)
.allKeys(true)
.keyStates(subState)
.scope(TbAttributeSubscriptionScope.valueOf(cmd.getScope())).build();
.scope(scope).build();
subService.addSubscription(sub);
}

40
application/src/main/java/org/thingsboard/server/service/ttl/AbstractTimeseriesCleanUpService.java → application/src/main/java/org/thingsboard/server/service/ttl/AbstractCleanUpService.java

@ -17,47 +17,27 @@ package org.thingsboard.server.service.ttl;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Scheduled;
import org.thingsboard.server.dao.util.PsqlTsAnyDao;
import org.thingsboard.server.dao.util.PsqlDao;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.SQLWarning;
import java.sql.Statement;
@PsqlTsAnyDao
@Slf4j
public abstract class AbstractTimeseriesCleanUpService {
@Value("${sql.ttl.ts_key_value_ttl}")
protected long systemTtl;
@Value("${sql.ttl.enabled}")
private boolean ttlTaskExecutionEnabled;
@Slf4j
@PsqlDao
public abstract class AbstractCleanUpService {
@Value("${spring.datasource.url}")
private String dbUrl;
protected String dbUrl;
@Value("${spring.datasource.username}")
private String dbUserName;
protected String dbUserName;
@Value("${spring.datasource.password}")
private String dbPassword;
@Scheduled(initialDelayString = "${sql.ttl.execution_interval_ms}", fixedDelayString = "${sql.ttl.execution_interval_ms}")
public void cleanUp() {
if (ttlTaskExecutionEnabled) {
try (Connection conn = DriverManager.getConnection(dbUrl, dbUserName, dbPassword)) {
doCleanUp(conn);
} catch (SQLException e) {
log.error("SQLException occurred during TTL task execution ", e);
}
}
}
protected abstract void doCleanUp(Connection connection);
protected String dbPassword;
protected long executeQuery(Connection conn, String query) {
long removed = 0L;
@ -74,7 +54,7 @@ public abstract class AbstractTimeseriesCleanUpService {
return removed;
}
private void getWarnings(Statement statement) throws SQLException {
protected void getWarnings(Statement statement) throws SQLException {
SQLWarning warnings = statement.getWarnings();
if (warnings != null) {
log.debug("{}", warnings.getMessage());
@ -86,4 +66,6 @@ public abstract class AbstractTimeseriesCleanUpService {
}
}
}
protected abstract void doCleanUp(Connection connection);
}

59
application/src/main/java/org/thingsboard/server/service/ttl/events/EventsCleanUpService.java

@ -0,0 +1,59 @@
/**
* Copyright © 2016-2020 The Thingsboard Authors
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.thingsboard.server.service.ttl.events;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import org.thingsboard.server.dao.util.PsqlDao;
import org.thingsboard.server.service.ttl.AbstractCleanUpService;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.SQLException;
@PsqlDao
@Slf4j
@Service
public class EventsCleanUpService extends AbstractCleanUpService {
@Value("${sql.ttl.events.events_ttl}")
private long ttl;
@Value("${sql.ttl.events.debug_events_ttl}")
private long debugTtl;
@Value("${sql.ttl.events.enabled}")
private boolean ttlTaskExecutionEnabled;
@Scheduled(initialDelayString = "${sql.ttl.events.execution_interval_ms}", fixedDelayString = "${sql.ttl.events.execution_interval_ms}")
public void cleanUp() {
if (ttlTaskExecutionEnabled) {
try (Connection conn = DriverManager.getConnection(dbUrl, dbUserName, dbPassword)) {
doCleanUp(conn);
} catch (SQLException e) {
log.error("SQLException occurred during TTL task execution ", e);
}
}
}
@Override
protected void doCleanUp(Connection connection) {
long totalEventsRemoved = executeQuery(connection, "call cleanup_events_by_ttl(" + ttl + ", " + debugTtl + ", 0);");
log.info("Total events removed by TTL: [{}]", totalEventsRemoved);
}
}

49
application/src/main/java/org/thingsboard/server/service/ttl/timeseries/AbstractTimeseriesCleanUpService.java

@ -0,0 +1,49 @@
/**
* Copyright © 2016-2020 The Thingsboard Authors
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.thingsboard.server.service.ttl.timeseries;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Scheduled;
import org.thingsboard.server.dao.util.PsqlTsAnyDao;
import org.thingsboard.server.service.ttl.AbstractCleanUpService;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.SQLException;
@PsqlTsAnyDao
@Slf4j
public abstract class AbstractTimeseriesCleanUpService extends AbstractCleanUpService {
@Value("${sql.ttl.ts.ts_key_value_ttl}")
protected long systemTtl;
@Value("${sql.ttl.ts.enabled}")
private boolean ttlTaskExecutionEnabled;
@Scheduled(initialDelayString = "${sql.ttl.ts.execution_interval_ms}", fixedDelayString = "${sql.ttl.ts.execution_interval_ms}")
public void cleanUp() {
if (ttlTaskExecutionEnabled) {
try (Connection conn = DriverManager.getConnection(dbUrl, dbUserName, dbPassword)) {
doCleanUp(conn);
} catch (SQLException e) {
log.error("SQLException occurred during TTL task execution ", e);
}
}
}
}

2
application/src/main/java/org/thingsboard/server/service/ttl/PsqlTimeseriesCleanUpService.java → application/src/main/java/org/thingsboard/server/service/ttl/timeseries/PsqlTimeseriesCleanUpService.java

@ -13,7 +13,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.thingsboard.server.service.ttl;
package org.thingsboard.server.service.ttl.timeseries;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;

2
application/src/main/java/org/thingsboard/server/service/ttl/TimescaleTimeseriesCleanUpService.java → application/src/main/java/org/thingsboard/server/service/ttl/timeseries/TimescaleTimeseriesCleanUpService.java

@ -13,7 +13,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.thingsboard.server.service.ttl;
package org.thingsboard.server.service.ttl.timeseries;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;

84
application/src/main/resources/thingsboard.yml

@ -181,31 +181,37 @@ cassandra:
# SQL configuration parameters
sql:
# Specify batch size for persisting attribute updates
attributes:
batch_size: "${SQL_ATTRIBUTES_BATCH_SIZE:10000}"
batch_max_delay: "${SQL_ATTRIBUTES_BATCH_MAX_DELAY_MS:100}"
stats_print_interval_ms: "${SQL_ATTRIBUTES_BATCH_STATS_PRINT_MS:10000}"
ts:
batch_size: "${SQL_TS_BATCH_SIZE:10000}"
batch_max_delay: "${SQL_TS_BATCH_MAX_DELAY_MS:100}"
stats_print_interval_ms: "${SQL_TS_BATCH_STATS_PRINT_MS:10000}"
ts_latest:
batch_size: "${SQL_TS_LATEST_BATCH_SIZE:10000}"
batch_max_delay: "${SQL_TS_LATEST_BATCH_MAX_DELAY_MS:100}"
stats_print_interval_ms: "${SQL_TS_LATEST_BATCH_STATS_PRINT_MS:10000}"
# Specify whether to remove null characters from strValue of attributes and timeseries before insert
remove_null_chars: "${SQL_REMOVE_NULL_CHARS:true}"
postgres:
# Specify partitioning size for timestamp key-value storage. Example: DAYS, MONTHS, YEARS, INDEFINITE.
ts_key_value_partitioning: "${SQL_POSTGRES_TS_KV_PARTITIONING:MONTHS}"
timescale:
# Specify Interval size for new data chunks storage.
chunk_time_interval: "${SQL_TIMESCALE_CHUNK_TIME_INTERVAL:604800000}"
ttl:
enabled: "${SQL_TTL_ENABLED:true}"
execution_interval_ms: "${SQL_TTL_EXECUTION_INTERVAL:86400000}" # Number of miliseconds
ts_key_value_ttl: "${SQL_TTL_TS_KEY_VALUE_TTL:0}" # Number of seconds
# Specify batch size for persisting attribute updates
attributes:
batch_size: "${SQL_ATTRIBUTES_BATCH_SIZE:10000}"
batch_max_delay: "${SQL_ATTRIBUTES_BATCH_MAX_DELAY_MS:100}"
stats_print_interval_ms: "${SQL_ATTRIBUTES_BATCH_STATS_PRINT_MS:10000}"
ts:
batch_size: "${SQL_TS_BATCH_SIZE:10000}"
batch_max_delay: "${SQL_TS_BATCH_MAX_DELAY_MS:100}"
stats_print_interval_ms: "${SQL_TS_BATCH_STATS_PRINT_MS:10000}"
ts_latest:
batch_size: "${SQL_TS_LATEST_BATCH_SIZE:10000}"
batch_max_delay: "${SQL_TS_LATEST_BATCH_MAX_DELAY_MS:100}"
stats_print_interval_ms: "${SQL_TS_LATEST_BATCH_STATS_PRINT_MS:10000}"
# Specify whether to remove null characters from strValue of attributes and timeseries before insert
remove_null_chars: "${SQL_REMOVE_NULL_CHARS:true}"
postgres:
# Specify partitioning size for timestamp key-value storage. Example: DAYS, MONTHS, YEARS, INDEFINITE.
ts_key_value_partitioning: "${SQL_POSTGRES_TS_KV_PARTITIONING:MONTHS}"
timescale:
# Specify Interval size for new data chunks storage.
chunk_time_interval: "${SQL_TIMESCALE_CHUNK_TIME_INTERVAL:604800000}"
ttl:
ts:
enabled: "${SQL_TTL_TS_ENABLED:true}"
execution_interval_ms: "${SQL_TTL_TS_EXECUTION_INTERVAL:86400000}" # Number of miliseconds. The current value corresponds to one day
ts_key_value_ttl: "${SQL_TTL_TS_TS_KEY_VALUE_TTL:0}" # Number of seconds
events:
enabled: "${SQL_TTL_EVENTS_ENABLED:true}"
execution_interval_ms: "${SQL_TTL_EVENTS_EXECUTION_INTERVAL:86400000}" # Number of miliseconds. The current value corresponds to one day
events_ttl: "${SQL_TTL_EVENTS_EVENTS_TTL:0}" # Number of seconds
debug_events_ttl: "${SQL_TTL_EVENTS_DEBUG_EVENTS_TTL:604800}" # Number of seconds. The current value corresponds to one week
# Actor system parameters
actors:
@ -410,8 +416,9 @@ audit-log:
password: "${AUDIT_LOG_SINK_PASSWORD:}"
state:
defaultInactivityTimeoutInSec: "${DEFAULT_INACTIVITY_TIMEOUT:10}"
defaultStateCheckIntervalInSec: "${DEFAULT_STATE_CHECK_INTERVAL:10}"
# Should be greater then transport.sessions.report_timeout
defaultInactivityTimeoutInSec: "${DEFAULT_INACTIVITY_TIMEOUT:600}"
defaultStateCheckIntervalInSec: "${DEFAULT_STATE_CHECK_INTERVAL:60}"
persistToTelemetry: "${PERSIST_STATE_TO_TELEMETRY:false}"
js:
@ -595,7 +602,7 @@ queue:
partitions: "${TB_QUEUE_CORE_PARTITIONS:10}"
pack-processing-timeout: "${TB_QUEUE_CORE_PACK_PROCESSING_TIMEOUT_MS:60000}"
stats:
enabled: "${TB_QUEUE_CORE_STATS_ENABLED:false}"
enabled: "${TB_QUEUE_CORE_STATS_ENABLED:true}"
print-interval-ms: "${TB_QUEUE_CORE_STATS_PRINT_INTERVAL_MS:10000}"
js:
# JS Eval request topic
@ -624,7 +631,7 @@ queue:
partitions: "${TB_QUEUE_RE_MAIN_PARTITIONS:10}"
pack-processing-timeout: "${TB_QUEUE_RE_MAIN_PACK_PROCESSING_TIMEOUT_MS:60000}"
submit-strategy:
type: "${TB_QUEUE_RE_MAIN_SUBMIT_STRATEGY_TYPE:BURST}" # BURST, BATCH, SEQUENTIAL_WITHIN_ORIGINATOR, SEQUENTIAL_WITHIN_TENANT, SEQUENTIAL
type: "${TB_QUEUE_RE_MAIN_SUBMIT_STRATEGY_TYPE:BURST}" # BURST, BATCH, SEQUENTIAL_BY_ORIGINATOR, SEQUENTIAL_BY_TENANT, SEQUENTIAL
# For BATCH only
batch-size: "${TB_QUEUE_RE_MAIN_SUBMIT_STRATEGY_BATCH_SIZE:1000}" # Maximum number of messages in batch
processing-strategy:
@ -636,10 +643,10 @@ queue:
- name: "${TB_QUEUE_RE_HP_QUEUE_NAME:HighPriority}"
topic: "${TB_QUEUE_RE_HP_TOPIC:tb_rule_engine.hp}"
poll-interval: "${TB_QUEUE_RE_HP_POLL_INTERVAL_MS:25}"
partitions: "${TB_QUEUE_RE_HP_PARTITIONS:3}"
partitions: "${TB_QUEUE_RE_HP_PARTITIONS:10}"
pack-processing-timeout: "${TB_QUEUE_RE_HP_PACK_PROCESSING_TIMEOUT_MS:60000}"
submit-strategy:
type: "${TB_QUEUE_RE_HP_SUBMIT_STRATEGY_TYPE:SEQUENTIAL_WITHIN_ORIGINATOR}" # BURST, BATCH, SEQUENTIAL_WITHIN_ORIGINATOR, SEQUENTIAL_WITHIN_TENANT, SEQUENTIAL
type: "${TB_QUEUE_RE_HP_SUBMIT_STRATEGY_TYPE:BURST}" # BURST, BATCH, SEQUENTIAL_BY_ORIGINATOR, SEQUENTIAL_BY_TENANT, SEQUENTIAL
# For BATCH only
batch-size: "${TB_QUEUE_RE_HP_SUBMIT_STRATEGY_BATCH_SIZE:100}" # Maximum number of messages in batch
processing-strategy:
@ -648,6 +655,21 @@ queue:
retries: "${TB_QUEUE_RE_HP_PROCESSING_STRATEGY_RETRIES:0}" # Number of retries, 0 is unlimited
failure-percentage: "${TB_QUEUE_RE_HP_PROCESSING_STRATEGY_FAILURE_PERCENTAGE:0}" # Skip retry if failures or timeouts are less then X percentage of messages;
pause-between-retries: "${TB_QUEUE_RE_HP_PROCESSING_STRATEGY_RETRY_PAUSE:5}"# Time in seconds to wait in consumer thread before retries;
- name: "${TB_QUEUE_RE_SQ_QUEUE_NAME:SequentialByOriginator}"
topic: "${TB_QUEUE_RE_SQ_TOPIC:tb_rule_engine.sq}"
poll-interval: "${TB_QUEUE_RE_SQ_POLL_INTERVAL_MS:25}"
partitions: "${TB_QUEUE_RE_SQ_PARTITIONS:10}"
pack-processing-timeout: "${TB_QUEUE_RE_SQ_PACK_PROCESSING_TIMEOUT_MS:60000}"
submit-strategy:
type: "${TB_QUEUE_RE_SQ_SUBMIT_STRATEGY_TYPE:SEQUENTIAL_BY_ORIGINATOR}" # BURST, BATCH, SEQUENTIAL_BY_ORIGINATOR, SEQUENTIAL_BY_TENANT, SEQUENTIAL
# For BATCH only
batch-size: "${TB_QUEUE_RE_SQ_SUBMIT_STRATEGY_BATCH_SIZE:100}" # Maximum number of messages in batch
processing-strategy:
type: "${TB_QUEUE_RE_SQ_PROCESSING_STRATEGY_TYPE:RETRY_FAILED_AND_TIMED_OUT}" # SKIP_ALL_FAILURES, RETRY_ALL, RETRY_FAILED, RETRY_TIMED_OUT, RETRY_FAILED_AND_TIMED_OUT
# For RETRY_ALL, RETRY_FAILED, RETRY_TIMED_OUT, RETRY_FAILED_AND_TIMED_OUT
retries: "${TB_QUEUE_RE_SQ_PROCESSING_STRATEGY_RETRIES:3}" # Number of retries, 0 is unlimited
failure-percentage: "${TB_QUEUE_RE_SQ_PROCESSING_STRATEGY_FAILURE_PERCENTAGE:0}" # Skip retry if failures or timeouts are less then X percentage of messages;
pause-between-retries: "${TB_QUEUE_RE_SQ_PROCESSING_STRATEGY_RETRY_PAUSE:5}"# Time in seconds to wait in consumer thread before retries;
transport:
# For high priority notifications that require minimum latency and processing time
notifications_topic: "${TB_QUEUE_TRANSPORT_NOTIFICATIONS_TOPIC:tb_transport.notifications}"

5
application/src/test/java/org/thingsboard/server/mqtt/telemetry/AbstractMqttTelemetryIntegrationTest.java

@ -97,7 +97,8 @@ public abstract class AbstractMqttTelemetryIntegrationTest extends AbstractContr
assertEquals("4", values.get("key4").get(0).get("value"));
}
@Test
// @Test - Unstable
public void testMqttQoSLevel() throws Exception {
String clientId = MqttAsyncClient.generateClientId();
MqttAsyncClient client = new MqttAsyncClient(MQTT_URL, clientId);
@ -109,7 +110,7 @@ public abstract class AbstractMqttTelemetryIntegrationTest extends AbstractContr
client.setCallback(callback);
client.connect(options).waitForCompletion(5000);
client.subscribe("v1/devices/me/attributes", MqttQoS.AT_MOST_ONCE.value());
String payload = "{\"key\":\"value\"}";
String payload = "{\"key\":\"uniqueValue\"}";
// TODO 3.1: we need to acknowledge subscription only after it is processed by device actor and not when the message is pushed to queue.
// MqttClient -> SUB REQUEST -> Transport -> Kafka -> Device Actor (subscribed)
// MqttClient <- SUB_ACK <- Transport

21
application/src/test/java/org/thingsboard/server/service/cluster/routing/ConsistentHashParitionServiceTest.java

@ -17,6 +17,7 @@ package org.thingsboard.server.service.cluster.routing;
import com.datastax.driver.core.utils.UUIDs;
import lombok.extern.slf4j.Slf4j;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
@ -31,6 +32,8 @@ import org.thingsboard.server.queue.discovery.TbServiceInfoProvider;
import org.thingsboard.server.common.msg.queue.TopicPartitionInfo;
import org.thingsboard.server.gen.transport.TransportProtos;
import org.thingsboard.server.queue.discovery.TenantRoutingInfoService;
import org.thingsboard.server.queue.settings.TbQueueRuleEngineSettings;
import org.thingsboard.server.queue.settings.TbRuleEngineQueueConfiguration;
import java.util.ArrayList;
import java.util.Collections;
@ -41,6 +44,7 @@ import java.util.Map;
import java.util.stream.Collectors;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
@Slf4j
@RunWith(MockitoJUnitRunner.class)
@ -52,6 +56,7 @@ public class ConsistentHashParitionServiceTest {
private TbServiceInfoProvider discoveryService;
private TenantRoutingInfoService routingInfoService;
private ApplicationEventPublisher applicationEventPublisher;
private TbQueueRuleEngineSettings ruleEngineSettings;
private String hashFunctionName = "murmur3_128";
private Integer virtualNodesSize = 16;
@ -62,12 +67,15 @@ public class ConsistentHashParitionServiceTest {
discoveryService = mock(TbServiceInfoProvider.class);
applicationEventPublisher = mock(ApplicationEventPublisher.class);
routingInfoService = mock(TenantRoutingInfoService.class);
clusterRoutingService = new ConsistentHashPartitionService(discoveryService, routingInfoService, applicationEventPublisher);
ruleEngineSettings = mock(TbQueueRuleEngineSettings.class);
clusterRoutingService = new ConsistentHashPartitionService(discoveryService,
routingInfoService,
applicationEventPublisher,
ruleEngineSettings
);
when(ruleEngineSettings.getQueues()).thenReturn(Collections.emptyList());
ReflectionTestUtils.setField(clusterRoutingService, "coreTopic", "tb.core");
ReflectionTestUtils.setField(clusterRoutingService, "corePartitions", 3);
ReflectionTestUtils.setField(clusterRoutingService, "ruleEngineTopic", "tb.rule-engine");
ReflectionTestUtils.setField(clusterRoutingService, "ruleEnginePartitions", 100);
ReflectionTestUtils.setField(clusterRoutingService, "hashFunctionName", hashFunctionName);
ReflectionTestUtils.setField(clusterRoutingService, "virtualNodesSize", virtualNodesSize);
TransportProtos.ServiceInfo currentServer = TransportProtos.ServiceInfo.newBuilder()
@ -107,8 +115,9 @@ public class ConsistentHashParitionServiceTest {
List<Map.Entry<Integer, Integer>> data = map.entrySet().stream().sorted(Comparator.comparingInt(Map.Entry::getValue)).collect(Collectors.toList());
long end = System.currentTimeMillis();
double diff = (data.get(data.size() - 1).getValue() - data.get(0).getValue());
System.out.println("Size: " + virtualNodesSize + " Time: " + (end - start) + " Diff: " + diff + "(" + String.format("%f", (diff / ITERATIONS) * 100.0) + "%)");
double diffPercent = (diff / ITERATIONS) * 100.0;
System.out.println("Size: " + virtualNodesSize + " Time: " + (end - start) + " Diff: " + diff + "(" + String.format("%f", diffPercent) + "%)");
Assert.assertTrue(diffPercent < 0.5);
for (Map.Entry<Integer, Integer> entry : data) {
System.out.println(entry.getKey() + ": " + entry.getValue());
}

4
application/src/test/java/org/thingsboard/server/service/queue/ProcessingAttemptContextTest.java → application/src/test/java/org/thingsboard/server/service/queue/TbMsgPackProcessingContextTest.java

@ -37,7 +37,7 @@ import static org.mockito.Mockito.when;
@Slf4j
@RunWith(MockitoJUnitRunner.class)
public class ProcessingAttemptContextTest {
public class TbMsgPackProcessingContextTest {
@Test
public void testHighConcurrencyCase() throws InterruptedException {
@ -51,7 +51,7 @@ public class ProcessingAttemptContextTest {
messages.put(UUID.randomUUID(), new TbProtoQueueMsg<>(UUID.randomUUID(), null));
}
when(strategyMock.getPendingMap()).thenReturn(messages);
ProcessingAttemptContext context = new ProcessingAttemptContext(strategyMock);
TbMsgPackProcessingContext context = new TbMsgPackProcessingContext(strategyMock);
for (UUID uuid : messages.keySet()) {
for (int i = 0; i < parallelCount; i++) {
executorService.submit(() -> context.onSuccess(uuid));

2
application/src/test/resources/logback.xml

@ -7,7 +7,7 @@
</encoder>
</appender>
<logger name="org.thingsboard.server" level="TRACE"/>
<logger name="org.thingsboard.server" level="WARN"/>
<logger name="org.springframework" level="WARN"/>
<logger name="org.springframework.boot.test" level="WARN"/>
<logger name="org.apache.cassandra" level="WARN"/>

12
common/queue/src/main/java/org/thingsboard/server/queue/discovery/ConsistentHashPartitionService.java

@ -30,6 +30,7 @@ import org.thingsboard.server.common.msg.queue.ServiceType;
import org.thingsboard.server.common.msg.queue.TopicPartitionInfo;
import org.thingsboard.server.gen.transport.TransportProtos;
import org.thingsboard.server.gen.transport.TransportProtos.ServiceInfo;
import org.thingsboard.server.queue.settings.TbQueueRuleEngineSettings;
import javax.annotation.PostConstruct;
import java.nio.charset.StandardCharsets;
@ -61,6 +62,7 @@ public class ConsistentHashPartitionService implements PartitionService {
private final ApplicationEventPublisher applicationEventPublisher;
private final TbServiceInfoProvider serviceInfoProvider;
private final TenantRoutingInfoService tenantRoutingInfoService;
private final TbQueueRuleEngineSettings tbQueueRuleEngineSettings;
private final ConcurrentMap<ServiceQueue, String> partitionTopics = new ConcurrentHashMap<>();
private final ConcurrentMap<ServiceQueue, Integer> partitionSizes = new ConcurrentHashMap<>();
private final ConcurrentMap<TenantId, TenantRoutingInfo> tenantRoutingInfoMap = new ConcurrentHashMap<>();
@ -74,10 +76,14 @@ public class ConsistentHashPartitionService implements PartitionService {
private HashFunction hashFunction;
public ConsistentHashPartitionService(TbServiceInfoProvider serviceInfoProvider, TenantRoutingInfoService tenantRoutingInfoService, ApplicationEventPublisher applicationEventPublisher) {
public ConsistentHashPartitionService(TbServiceInfoProvider serviceInfoProvider,
TenantRoutingInfoService tenantRoutingInfoService,
ApplicationEventPublisher applicationEventPublisher,
TbQueueRuleEngineSettings tbQueueRuleEngineSettings) {
this.serviceInfoProvider = serviceInfoProvider;
this.tenantRoutingInfoService = tenantRoutingInfoService;
this.applicationEventPublisher = applicationEventPublisher;
this.tbQueueRuleEngineSettings = tbQueueRuleEngineSettings;
}
@PostConstruct
@ -85,6 +91,10 @@ public class ConsistentHashPartitionService implements PartitionService {
this.hashFunction = forName(hashFunctionName);
partitionSizes.put(new ServiceQueue(ServiceType.TB_CORE), corePartitions);
partitionTopics.put(new ServiceQueue(ServiceType.TB_CORE), coreTopic);
tbQueueRuleEngineSettings.getQueues().forEach(queueConfiguration -> {
partitionTopics.put(new ServiceQueue(ServiceType.TB_RULE_ENGINE, queueConfiguration.getName()), queueConfiguration.getTopic());
partitionSizes.put(new ServiceQueue(ServiceType.TB_RULE_ENGINE, queueConfiguration.getName()), queueConfiguration.getPartitions());
});
}
@Override

6
common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java

@ -544,6 +544,9 @@ public class DefaultTransportService implements TransportService {
protected void sendToDeviceActor(TransportProtos.SessionInfoProto sessionInfo, TransportToDeviceActorMsg toDeviceActorMsg, TransportServiceCallback<Void> callback) {
TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_CORE, getTenantId(sessionInfo), getDeviceId(sessionInfo));
if (log.isTraceEnabled()) {
log.trace("[{}][{}] Pushing to topic {} message {}", getTenantId(sessionInfo), getDeviceId(sessionInfo), tpi.getFullTopicName(), toDeviceActorMsg);
}
tbCoreMsgProducer.send(tpi,
new TbProtoQueueMsg<>(getRoutingKey(sessionInfo),
ToCoreMsg.newBuilder().setToDeviceActorMsg(toDeviceActorMsg).build()), callback != null ?
@ -552,6 +555,9 @@ public class DefaultTransportService implements TransportService {
protected void sendToRuleEngine(TenantId tenantId, TbMsg tbMsg, TbQueueCallback callback) {
TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_RULE_ENGINE, tenantId, tbMsg.getOriginator());
if (log.isTraceEnabled()) {
log.trace("[{}][{}] Pushing to topic {} message {}", tenantId, tbMsg.getOriginator(), tpi.getFullTopicName(), tbMsg);
}
ToRuleEngineMsg msg = ToRuleEngineMsg.newBuilder().setTbMsg(TbMsg.toByteString(tbMsg))
.setTenantIdMSB(tenantId.getId().getMostSignificantBits())
.setTenantIdLSB(tenantId.getId().getLeastSignificantBits()).build();

3
dao/src/main/java/org/thingsboard/server/dao/model/ModelConstants.java

@ -32,6 +32,9 @@ public class ModelConstants {
public static final String NULL_UUID_STR = UUIDConverter.fromTimeUUID(NULL_UUID);
public static final TenantId SYSTEM_TENANT = new TenantId(ModelConstants.NULL_UUID);
// this is the difference between midnight October 15, 1582 UTC and midnight January 1, 1970 UTC as 100 nanosecond units
public static final long EPOCH_DIFF = 122192928000000000L;
/**
* Generic constants.
*/

14
dao/src/main/java/org/thingsboard/server/dao/model/sql/EventEntity.java

@ -37,6 +37,9 @@ import javax.persistence.EnumType;
import javax.persistence.Enumerated;
import javax.persistence.Table;
import java.util.UUID;
import static org.thingsboard.server.dao.model.ModelConstants.EPOCH_DIFF;
import static org.thingsboard.server.dao.model.ModelConstants.EVENT_BODY_PROPERTY;
import static org.thingsboard.server.dao.model.ModelConstants.EVENT_COLUMN_FAMILY_NAME;
import static org.thingsboard.server.dao.model.ModelConstants.EVENT_ENTITY_ID_PROPERTY;
@ -44,6 +47,7 @@ import static org.thingsboard.server.dao.model.ModelConstants.EVENT_ENTITY_TYPE_
import static org.thingsboard.server.dao.model.ModelConstants.EVENT_TENANT_ID_PROPERTY;
import static org.thingsboard.server.dao.model.ModelConstants.EVENT_TYPE_PROPERTY;
import static org.thingsboard.server.dao.model.ModelConstants.EVENT_UID_PROPERTY;
import static org.thingsboard.server.dao.model.ModelConstants.TS_COLUMN;
@Data
@EqualsAndHashCode(callSuper = true)
@ -73,9 +77,15 @@ public class EventEntity extends BaseSqlEntity<Event> implements BaseEntity<Eve
@Column(name = EVENT_BODY_PROPERTY)
private JsonNode body;
@Column(name = TS_COLUMN)
private long ts;
public EventEntity(Event event) {
if (event.getId() != null) {
this.setUuid(event.getId().getId());
this.ts = getTs(event.getId().getId());
} else {
this.ts = System.currentTimeMillis();
}
if (event.getTenantId() != null) {
this.tenantId = toString(event.getTenantId().getId());
@ -101,4 +111,8 @@ public class EventEntity extends BaseSqlEntity<Event> implements BaseEntity<Eve
event.setUid(eventUid);
return event;
}
private long getTs(UUID uuid) {
return (uuid.timestamp() - EPOCH_DIFF) / 10000;
}
}

25
dao/src/main/java/org/thingsboard/server/dao/rule/BaseRuleChainService.java

@ -88,26 +88,33 @@ public class BaseRuleChainService extends AbstractEntityService implements RuleC
RuleChain ruleChain = ruleChainDao.findById(tenantId, ruleChainId.getId());
if (!ruleChain.isRoot()) {
RuleChain previousRootRuleChain = getRootTenantRuleChain(ruleChain.getTenantId());
if (!previousRootRuleChain.getId().equals(ruleChain.getId())) {
try {
try {
if (previousRootRuleChain == null) {
setRootAndSave(tenantId, ruleChain);
return true;
} else if (!previousRootRuleChain.getId().equals(ruleChain.getId())) {
deleteRelation(tenantId, new EntityRelation(previousRootRuleChain.getTenantId(), previousRootRuleChain.getId(),
EntityRelation.CONTAINS_TYPE, RelationTypeGroup.RULE_CHAIN));
previousRootRuleChain.setRoot(false);
ruleChainDao.save(tenantId, previousRootRuleChain);
createRelation(tenantId, new EntityRelation(ruleChain.getTenantId(), ruleChain.getId(),
EntityRelation.CONTAINS_TYPE, RelationTypeGroup.RULE_CHAIN));
ruleChain.setRoot(true);
ruleChainDao.save(tenantId, ruleChain);
setRootAndSave(tenantId, ruleChain);
return true;
} catch (ExecutionException | InterruptedException e) {
log.warn("[{}] Failed to set root rule chain, ruleChainId: [{}]", ruleChainId);
throw new RuntimeException(e);
}
} catch (ExecutionException | InterruptedException e) {
log.warn("[{}] Failed to set root rule chain, ruleChainId: [{}]", ruleChainId);
throw new RuntimeException(e);
}
}
return false;
}
private void setRootAndSave(TenantId tenantId, RuleChain ruleChain) throws ExecutionException, InterruptedException {
createRelation(tenantId, new EntityRelation(ruleChain.getTenantId(), ruleChain.getId(),
EntityRelation.CONTAINS_TYPE, RelationTypeGroup.RULE_CHAIN));
ruleChain.setRoot(true);
ruleChainDao.save(tenantId, ruleChain);
}
@Override
public RuleChainMetaData saveRuleChainMetaData(TenantId tenantId, RuleChainMetaData ruleChainMetaData) {
Validator.validateId(ruleChainMetaData.getRuleChainId(), "Incorrect rule chain id.");

3
dao/src/main/java/org/thingsboard/server/dao/sql/event/AbstractEventInsertRepository.java

@ -75,7 +75,8 @@ public abstract class AbstractEventInsertRepository implements EventInsertReposi
.setParameter("entity_type", entity.getEntityType().name())
.setParameter("event_type", entity.getEventType())
.setParameter("event_uid", entity.getEventUid())
.setParameter("tenant_id", entity.getTenantId());
.setParameter("tenant_id", entity.getTenantId())
.setParameter("ts", entity.getTs());
}
private EventEntity processSaveOrUpdate(EventEntity entity, String query) {

4
dao/src/main/java/org/thingsboard/server/dao/sql/event/HsqlEventInsertRepository.java

@ -44,7 +44,7 @@ public class HsqlEventInsertRepository extends AbstractEventInsertRepository {
}
private static String getInsertString(String conflictStatement) {
return "MERGE INTO event USING (VALUES :id, :body, :entity_id, :entity_type, :event_type, :event_uid, :tenant_id) I (id, body, entity_id, entity_type, event_type, event_uid, tenant_id) ON " + conflictStatement + " WHEN MATCHED THEN UPDATE SET event.id = I.id, event.body = I.body, event.entity_id = I.entity_id, event.entity_type = I.entity_type, event.event_type = I.event_type, event.event_uid = I.event_uid, event.tenant_id = I.tenant_id" +
" WHEN NOT MATCHED THEN INSERT (id, body, entity_id, entity_type, event_type, event_uid, tenant_id) VALUES (I.id, I.body, I.entity_id, I.entity_type, I.event_type, I.event_uid, I.tenant_id)";
return "MERGE INTO event USING (VALUES :id, :body, :entity_id, :entity_type, :event_type, :event_uid, :tenant_id, :ts) I (id, body, entity_id, entity_type, event_type, event_uid, tenant_id, ts) ON " + conflictStatement + " WHEN MATCHED THEN UPDATE SET event.id = I.id, event.body = I.body, event.entity_id = I.entity_id, event.entity_type = I.entity_type, event.event_type = I.event_type, event.event_uid = I.event_uid, event.tenant_id = I.tenant_id, event.ts = I.ts" +
" WHEN NOT MATCHED THEN INSERT (id, body, entity_id, entity_type, event_type, event_uid, tenant_id, ts) VALUES (I.id, I.body, I.entity_id, I.entity_type, I.event_type, I.event_uid, I.tenant_id, I.ts)";
}
}

2
dao/src/main/java/org/thingsboard/server/dao/sql/event/PsqlEventInsertRepository.java

@ -48,6 +48,6 @@ public class PsqlEventInsertRepository extends AbstractEventInsertRepository {
}
private static String getInsertOrUpdateString(String eventKeyStatement, String updateKeyStatement) {
return "INSERT INTO event (id, body, entity_id, entity_type, event_type, event_uid, tenant_id) VALUES (:id, :body, :entity_id, :entity_type, :event_type, :event_uid, :tenant_id) ON CONFLICT " + eventKeyStatement + " DO UPDATE SET body = :body, " + updateKeyStatement + " returning *";
return "INSERT INTO event (id, body, entity_id, entity_type, event_type, event_uid, tenant_id, ts) VALUES (:id, :body, :entity_id, :entity_type, :event_type, :event_uid, :tenant_id, :ts) ON CONFLICT " + eventKeyStatement + " DO UPDATE SET body = :body, ts = :ts," + updateKeyStatement + " returning *";
}
}

2
dao/src/main/resources/sql/schema-entities-hsql.sql

@ -144,6 +144,7 @@ CREATE TABLE IF NOT EXISTS event (
event_type varchar(255),
event_uid varchar(255),
tenant_id varchar(31),
ts bigint NOT NULL,
CONSTRAINT event_unq_key UNIQUE (tenant_id, entity_type, entity_id, event_type, event_uid)
);
@ -251,3 +252,4 @@ CREATE TABLE IF NOT EXISTS entity_view (
search_text varchar(255),
additional_info varchar
);

26
dao/src/main/resources/sql/schema-entities.sql

@ -144,6 +144,7 @@ CREATE TABLE IF NOT EXISTS event (
event_type varchar(255),
event_uid varchar(255),
tenant_id varchar(31),
ts bigint NOT NULL,
CONSTRAINT event_unq_key UNIQUE (tenant_id, entity_type, entity_id, event_type, event_uid)
);
@ -251,3 +252,28 @@ CREATE TABLE IF NOT EXISTS entity_view (
search_text varchar(255),
additional_info varchar
);
CREATE OR REPLACE PROCEDURE cleanup_events_by_ttl(IN ttl bigint, IN debug_ttl bigint, INOUT deleted bigint)
LANGUAGE plpgsql AS
$$
DECLARE
ttl_ts bigint;
debug_ttl_ts bigint;
ttl_deleted_count bigint DEFAULT 0;
debug_ttl_deleted_count bigint DEFAULT 0;
BEGIN
IF ttl > 0 THEN
ttl_ts := (EXTRACT(EPOCH FROM current_timestamp) * 1000 - ttl::bigint * 1000)::bigint;
EXECUTE format(
'WITH deleted AS (DELETE FROM event WHERE ts < %L::bigint AND (event_type != %L::varchar AND event_type != %L::varchar) RETURNING *) SELECT count(*) FROM deleted', ttl_ts, 'DEBUG_RULE_NODE', 'DEBUG_RULE_CHAIN') into ttl_deleted_count;
END IF;
IF debug_ttl > 0 THEN
debug_ttl_ts := (EXTRACT(EPOCH FROM current_timestamp) * 1000 - debug_ttl::bigint * 1000)::bigint;
EXECUTE format(
'WITH deleted AS (DELETE FROM event WHERE ts < %L::bigint AND (event_type = %L::varchar OR event_type = %L::varchar) RETURNING *) SELECT count(*) FROM deleted', debug_ttl_ts, 'DEBUG_RULE_NODE', 'DEBUG_RULE_CHAIN') into debug_ttl_deleted_count;
END IF;
RAISE NOTICE 'Events removed by ttl: %', ttl_deleted_count;
RAISE NOTICE 'Debug Events removed by ttl: %', debug_ttl_deleted_count;
deleted := ttl_deleted_count + debug_ttl_deleted_count;
END
$$;

2
dao/src/main/resources/sql/schema-timescale.sql

@ -52,7 +52,7 @@ CREATE TABLE IF NOT EXISTS tb_schema_settings
CONSTRAINT tb_schema_settings_pkey PRIMARY KEY (schema_version)
);
INSERT INTO tb_schema_settings (schema_version) VALUES (2005000);
INSERT INTO tb_schema_settings (schema_version) VALUES (2005000) ON CONFLICT (schema_version) DO UPDATE SET schema_version = 2005000;
CREATE OR REPLACE FUNCTION to_uuid(IN entity_id varchar, OUT uuid_id uuid) AS
$$

2
dao/src/main/resources/sql/schema-ts-psql.sql

@ -52,7 +52,7 @@ CREATE TABLE IF NOT EXISTS tb_schema_settings
CONSTRAINT tb_schema_settings_pkey PRIMARY KEY (schema_version)
);
INSERT INTO tb_schema_settings (schema_version) VALUES (2005000);
INSERT INTO tb_schema_settings (schema_version) VALUES (2005000) ON CONFLICT (schema_version) DO UPDATE SET schema_version = 2005000;
CREATE OR REPLACE PROCEDURE drop_partitions_by_max_ttl(IN partition_type varchar, IN system_ttl bigint, INOUT deleted bigint)
LANGUAGE plpgsql AS

3
dao/src/test/resources/sql-test.properties

@ -37,6 +37,9 @@ service.type=monolith
#spring.datasource.driverClassName=org.postgresql.Driver
#spring.datasource.hikari.maximumPoolSize = 50
queue.core.pack-processing-timeout=3000
queue.rule-engine.pack-processing-timeout=3000
queue.rule-engine.queues[0].name=Main
queue.rule-engine.queues[0].topic=tb_rule_engine.main
queue.rule-engine.queues[0].poll-interval=25

2
docker/README.md

@ -58,7 +58,7 @@ In case of any issues you can examine service logs for errors.
For example to see ThingsBoard node logs execute the following command:
`
$ docker-compose logs -f tb1
$ docker-compose logs -f tb-core1 tb-rule-engine1
`
Or use `docker-compose ps` to see the state of all the containers.

18
docker/docker-compose.cassandra.yml

@ -24,14 +24,28 @@ services:
- "9042"
volumes:
- ./tb-node/cassandra:/var/lib/cassandra
tb1:
tb-core1:
env_file:
- tb-node.cassandra.env
depends_on:
- kafka
- redis
- cassandra
tb2:
tb-core2:
env_file:
- tb-node.cassandra.env
depends_on:
- kafka
- redis
- cassandra
tb-rule-engine1:
env_file:
- tb-node.cassandra.env
depends_on:
- kafka
- redis
- cassandra
tb-rule-engine2:
env_file:
- tb-node.cassandra.env
depends_on:

10
docker/docker-compose.postgres.volumes.yml

@ -20,10 +20,16 @@ services:
postgres:
volumes:
- postgres-db-volume:/var/lib/postgresql/data
tb1:
tb-core1:
volumes:
- tb-log-volume:/var/log/thingsboard
tb2:
tb-core2:
volumes:
- tb-log-volume:/var/log/thingsboard
tb-rule-engine1:
volumes:
- tb-log-volume:/var/log/thingsboard
tb-rule-engine2:
volumes:
- tb-log-volume:/var/log/thingsboard
tb-coap-transport:

18
docker/docker-compose.postgres.yml

@ -27,14 +27,28 @@ services:
POSTGRES_PASSWORD: postgres
volumes:
- ./tb-node/postgres:/var/lib/postgresql/data
tb1:
tb-core1:
env_file:
- tb-node.postgres.env
depends_on:
- kafka
- redis
- postgres
tb2:
tb-core2:
env_file:
- tb-node.postgres.env
depends_on:
- kafka
- redis
- postgres
tb-rule-engine1:
env_file:
- tb-node.postgres.env
depends_on:
- kafka
- redis
- postgres
tb-rule-engine2:
env_file:
- tb-node.postgres.env
depends_on:

79
docker/docker-compose.yml

@ -48,7 +48,7 @@ services:
- tb-js-executor.env
depends_on:
- kafka
tb1:
tb-core1:
restart: always
image: "${DOCKER_REPO}/${TB_NODE_DOCKER_NAME}:${TB_VERSION}"
ports:
@ -59,8 +59,8 @@ services:
max-size: "200m"
max-file: "30"
environment:
TB_HOST: tb1
CLUSTER_NODE_ID: tb1
TB_SERVICE_ID: tb-core1
TB_SERVICE_TYPE: tb-core
env_file:
- tb-node.env
volumes:
@ -70,7 +70,9 @@ services:
- kafka
- redis
- tb-js-executor
tb2:
- tb-rule-engine1
- tb-rule-engine2
tb-core2:
restart: always
image: "${DOCKER_REPO}/${TB_NODE_DOCKER_NAME}:${TB_VERSION}"
ports:
@ -81,8 +83,54 @@ services:
max-size: "200m"
max-file: "30"
environment:
TB_HOST: tb2
CLUSTER_NODE_ID: tb2
TB_SERVICE_ID: tb-core2
TB_SERVICE_TYPE: tb-core
env_file:
- tb-node.env
volumes:
- ./tb-node/conf:/config
- ./tb-node/log:/var/log/thingsboard
depends_on:
- kafka
- redis
- tb-js-executor
- tb-rule-engine1
- tb-rule-engine2
tb-rule-engine1:
restart: always
image: "${DOCKER_REPO}/${TB_NODE_DOCKER_NAME}:${TB_VERSION}"
ports:
- "8080"
logging:
driver: "json-file"
options:
max-size: "200m"
max-file: "30"
environment:
TB_SERVICE_ID: tb-rule-engine1
TB_SERVICE_TYPE: tb-rule-engine
env_file:
- tb-node.env
volumes:
- ./tb-node/conf:/config
- ./tb-node/log:/var/log/thingsboard
depends_on:
- kafka
- redis
- tb-js-executor
tb-rule-engine2:
restart: always
image: "${DOCKER_REPO}/${TB_NODE_DOCKER_NAME}:${TB_VERSION}"
ports:
- "8080"
logging:
driver: "json-file"
options:
max-size: "200m"
max-file: "30"
environment:
TB_SERVICE_ID: tb-rule-engine2
TB_SERVICE_TYPE: tb-rule-engine
env_file:
- tb-node.env
volumes:
@ -98,8 +146,7 @@ services:
ports:
- "1883"
environment:
TB_HOST: tb-mqtt-transport1
CLUSTER_NODE_ID: tb-mqtt-transport1
TB_SERVICE_ID: tb-mqtt-transport1
env_file:
- tb-mqtt-transport.env
volumes:
@ -113,8 +160,7 @@ services:
ports:
- "1883"
environment:
TB_HOST: tb-mqtt-transport2
CLUSTER_NODE_ID: tb-mqtt-transport2
TB_SERVICE_ID: tb-mqtt-transport2
env_file:
- tb-mqtt-transport.env
volumes:
@ -128,8 +174,7 @@ services:
ports:
- "8081"
environment:
TB_HOST: tb-http-transport1
CLUSTER_NODE_ID: tb-http-transport1
TB_SERVICE_ID: tb-http-transport1
env_file:
- tb-http-transport.env
volumes:
@ -143,8 +188,7 @@ services:
ports:
- "8081"
environment:
TB_HOST: tb-http-transport2
CLUSTER_NODE_ID: tb-http-transport2
TB_SERVICE_ID: tb-http-transport2
env_file:
- tb-http-transport.env
volumes:
@ -158,8 +202,7 @@ services:
ports:
- "5683:5683/udp"
environment:
TB_HOST: tb-coap-transport
CLUSTER_NODE_ID: tb-coap-transport
TB_SERVICE_ID: tb-coap-transport
env_file:
- tb-coap-transport.env
volumes:
@ -202,8 +245,8 @@ services:
MQTT_PORT: 1883
FORCE_HTTPS_REDIRECT: "false"
links:
- tb1
- tb2
- tb-core1
- tb-core2
- tb-web-ui1
- tb-web-ui2
- tb-mqtt-transport1

2
docker/docker-install-tb.sh

@ -49,6 +49,6 @@ if [ ! -z "${ADDITIONAL_STARTUP_SERVICES// }" ]; then
docker-compose -f docker-compose.yml $ADDITIONAL_COMPOSE_ARGS up -d redis $ADDITIONAL_STARTUP_SERVICES
fi
docker-compose -f docker-compose.yml $ADDITIONAL_COMPOSE_ARGS run --no-deps --rm -e INSTALL_TB=true -e LOAD_DEMO=${loadDemo} tb1
docker-compose -f docker-compose.yml $ADDITIONAL_COMPOSE_ARGS run --no-deps --rm -e INSTALL_TB=true -e LOAD_DEMO=${loadDemo} tb-core1

4
docker/docker-upgrade-tb.sh

@ -44,8 +44,8 @@ ADDITIONAL_COMPOSE_ARGS=$(additionalComposeArgs) || exit $?
ADDITIONAL_STARTUP_SERVICES=$(additionalStartupServices) || exit $?
docker-compose -f docker-compose.yml $ADDITIONAL_COMPOSE_ARGS pull tb1
docker-compose -f docker-compose.yml $ADDITIONAL_COMPOSE_ARGS pull tb-core1
docker-compose -f docker-compose.yml $ADDITIONAL_COMPOSE_ARGS up -d redis $ADDITIONAL_STARTUP_SERVICES
docker-compose -f docker-compose.yml $ADDITIONAL_COMPOSE_ARGS run --no-deps --rm -e UPGRADE_TB=true -e FROM_VERSION=${fromVersion} tb1
docker-compose -f docker-compose.yml $ADDITIONAL_COMPOSE_ARGS run --no-deps --rm -e UPGRADE_TB=true -e FROM_VERSION=${fromVersion} tb-core1

4
docker/haproxy/config/haproxy.cfg

@ -111,6 +111,6 @@ backend tb-api-backend
balance leastconn
option tcp-check
option log-health-checks
server tbApi1 tb1:8080 check inter 5s resolvers docker_resolver resolve-prefer ipv4
server tbApi2 tb2:8080 check inter 5s resolvers docker_resolver resolve-prefer ipv4
server tbApi1 tb-core1:8080 check inter 5s resolvers docker_resolver resolve-prefer ipv4
server tbApi2 tb-core2:8080 check inter 5s resolvers docker_resolver resolve-prefer ipv4
http-request set-header X-Forwarded-Port %[dst_port]

2
docker/kafka.env

@ -4,7 +4,7 @@ KAFKA_LISTENERS=INSIDE://:9093,OUTSIDE://:9092
KAFKA_ADVERTISED_LISTENERS=INSIDE://:9093,OUTSIDE://kafka:9092
KAFKA_LISTENER_SECURITY_PROTOCOL_MAP=INSIDE:PLAINTEXT,OUTSIDE:PLAINTEXT
KAFKA_INTER_BROKER_LISTENER_NAME=INSIDE
KAFKA_CREATE_TOPICS=js.eval.requests:100:1:delete --config=retention.ms=60000 --config=segment.bytes=26214400 --config=retention.bytes=104857600,tb.transport.api.requests:30:1:delete --config=retention.ms=60000 --config=segment.bytes=26214400 --config=retention.bytes=104857600,tb.rule-engine:30:1:delete --config=retention.ms=60000 --config=segment.bytes=26214400 --config=retention.bytes=104857600
KAFKA_CREATE_TOPICS=js_eval.requests:3:1:delete --config=retention.ms=60000 --config=segment.bytes=26214400 --config=retention.bytes=104857600,tb_transport.api.requests:3:1:delete --config=retention.ms=60000 --config=segment.bytes=26214400 --config=retention.bytes=104857600
KAFKA_AUTO_CREATE_TOPICS_ENABLE=false
KAFKA_LOG_RETENTION_BYTES=1073741824
KAFKA_LOG_SEGMENT_BYTES=268435456

3
docker/tb-coap-transport.env

@ -1,6 +1,9 @@
ZOOKEEPER_ENABLED=true
ZOOKEEPER_URL=zookeeper:2181
COAP_BIND_ADDRESS=0.0.0.0
COAP_BIND_PORT=5683
COAP_TIMEOUT=10000
TB_QUEUE_TYPE=kafka
TB_KAFKA_SERVERS=kafka:9092

3
docker/tb-http-transport.env

@ -1,6 +1,9 @@
ZOOKEEPER_ENABLED=true
ZOOKEEPER_URL=zookeeper:2181
HTTP_BIND_ADDRESS=0.0.0.0
HTTP_BIND_PORT=8081
HTTP_REQUEST_TIMEOUT=60000
TB_QUEUE_TYPE=kafka
TB_KAFKA_SERVERS=kafka:9092

2
docker/tb-js-executor.env

@ -1,5 +1,5 @@
REMOTE_JS_EVAL_REQUEST_TOPIC=js.eval.requests
REMOTE_JS_EVAL_REQUEST_TOPIC=js_eval.requests
TB_KAFKA_SERVERS=kafka:9092
LOGGER_LEVEL=info
LOG_FOLDER=logs

3
docker/tb-mqtt-transport.env

@ -1,6 +1,9 @@
ZOOKEEPER_ENABLED=true
ZOOKEEPER_URL=zookeeper:2181
MQTT_BIND_ADDRESS=0.0.0.0
MQTT_BIND_PORT=1883
MQTT_TIMEOUT=10000
TB_QUEUE_TYPE=kafka
TB_KAFKA_SERVERS=kafka:9092

2
docker/tb-node.env

@ -2,7 +2,7 @@
ZOOKEEPER_ENABLED=true
ZOOKEEPER_URL=zookeeper:2181
RPC_HOST=${TB_HOST}
TB_QUEUE_TYPE=kafka
TB_KAFKA_SERVERS=kafka:9092
JS_EVALUATOR=remote
TRANSPORT_TYPE=remote

4
docker/tb-node/conf/logback.xml

@ -21,10 +21,10 @@
<appender name="fileLogAppender"
class="ch.qos.logback.core.rolling.RollingFileAppender">
<file>/var/log/thingsboard/${TB_HOST}/thingsboard.log</file>
<file>/var/log/thingsboard/${TB_SERVICE_ID}/thingsboard.log</file>
<rollingPolicy
class="ch.qos.logback.core.rolling.SizeAndTimeBasedRollingPolicy">
<fileNamePattern>/var/log/thingsboard/${TB_HOST}/thingsboard.%d{yyyy-MM-dd}.%i.log</fileNamePattern>
<fileNamePattern>/var/log/thingsboard/${TB_SERVICE_ID}/thingsboard.%d{yyyy-MM-dd}.%i.log</fileNamePattern>
<maxFileSize>100MB</maxFileSize>
<maxHistory>30</maxHistory>
<totalSizeCap>3GB</totalSizeCap>

2
docker/tb-node/conf/thingsboard.conf

@ -15,7 +15,7 @@
#
export JAVA_OPTS="$JAVA_OPTS -Dplatform=deb -Dinstall.data_dir=/usr/share/thingsboard/data"
export JAVA_OPTS="$JAVA_OPTS -Xloggc:/var/log/thingsboard/${TB_HOST}/gc.log -XX:+IgnoreUnrecognizedVMOptions -XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=/var/log/thingsboard/${TB_HOST}/heapdump.bin -XX:+PrintGCDetails -XX:+PrintGCDateStamps"
export JAVA_OPTS="$JAVA_OPTS -Xloggc:/var/log/thingsboard/${TB_SERVICE_ID}/gc.log -XX:+IgnoreUnrecognizedVMOptions -XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=/var/log/thingsboard/${TB_SERVICE_ID}/heapdump.bin -XX:+PrintGCDetails -XX:+PrintGCDateStamps"
export JAVA_OPTS="$JAVA_OPTS -XX:+PrintHeapAtGC -XX:+PrintTenuringDistribution -XX:+PrintGCApplicationStoppedTime -XX:+UseGCLogFileRotation -XX:NumberOfGCLogFiles=10"
export JAVA_OPTS="$JAVA_OPTS -XX:GCLogFileSize=10M -XX:-UseBiasedLocking -XX:+UseTLAB -XX:+ResizeTLAB -XX:+PerfDisableSharedMem -XX:+UseCondCardMark"
export JAVA_OPTS="$JAVA_OPTS -XX:CMSWaitDuration=10000 -XX:+UseParNewGC -XX:+UseConcMarkSweepGC -XX:+CMSParallelRemarkEnabled -XX:+CMSParallelInitialMarkEnabled"

4
docker/tb-transports/coap/conf/logback.xml

@ -21,10 +21,10 @@
<appender name="fileLogAppender"
class="ch.qos.logback.core.rolling.RollingFileAppender">
<file>/var/log/tb-coap-transport/${TB_HOST}/tb-coap-transport.log</file>
<file>/var/log/tb-coap-transport/${TB_SERVICE_ID}/tb-coap-transport.log</file>
<rollingPolicy
class="ch.qos.logback.core.rolling.SizeAndTimeBasedRollingPolicy">
<fileNamePattern>/var/log/tb-coap-transport/${TB_HOST}/tb-coap-transport.%d{yyyy-MM-dd}.%i.log</fileNamePattern>
<fileNamePattern>/var/log/tb-coap-transport/${TB_SERVICE_ID}/tb-coap-transport.%d{yyyy-MM-dd}.%i.log</fileNamePattern>
<maxFileSize>100MB</maxFileSize>
<maxHistory>30</maxHistory>
<totalSizeCap>3GB</totalSizeCap>

2
docker/tb-transports/coap/conf/tb-coap-transport.conf

@ -14,7 +14,7 @@
# limitations under the License.
#
export JAVA_OPTS="$JAVA_OPTS -Xloggc:/var/log/tb-coap-transport/${TB_HOST}/gc.log -XX:+IgnoreUnrecognizedVMOptions -XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=/var/log/tb-coap-transport/${TB_HOST}/heapdump.bin -XX:+PrintGCDetails -XX:+PrintGCDateStamps"
export JAVA_OPTS="$JAVA_OPTS -Xloggc:/var/log/tb-coap-transport/${TB_SERVICE_ID}/gc.log -XX:+IgnoreUnrecognizedVMOptions -XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=/var/log/tb-coap-transport/${TB_SERVICE_ID}/heapdump.bin -XX:+PrintGCDetails -XX:+PrintGCDateStamps"
export JAVA_OPTS="$JAVA_OPTS -XX:+PrintHeapAtGC -XX:+PrintTenuringDistribution -XX:+PrintGCApplicationStoppedTime -XX:+UseGCLogFileRotation -XX:NumberOfGCLogFiles=10"
export JAVA_OPTS="$JAVA_OPTS -XX:GCLogFileSize=10M -XX:-UseBiasedLocking -XX:+UseTLAB -XX:+ResizeTLAB -XX:+PerfDisableSharedMem -XX:+UseCondCardMark"
export JAVA_OPTS="$JAVA_OPTS -XX:CMSWaitDuration=10000 -XX:+UseParNewGC -XX:+UseConcMarkSweepGC -XX:+CMSParallelRemarkEnabled -XX:+CMSParallelInitialMarkEnabled"

4
docker/tb-transports/http/conf/logback.xml

@ -21,10 +21,10 @@
<appender name="fileLogAppender"
class="ch.qos.logback.core.rolling.RollingFileAppender">
<file>/var/log/tb-http-transport/${TB_HOST}/tb-http-transport.log</file>
<file>/var/log/tb-http-transport/${TB_SERVICE_ID}/tb-http-transport.log</file>
<rollingPolicy
class="ch.qos.logback.core.rolling.SizeAndTimeBasedRollingPolicy">
<fileNamePattern>/var/log/tb-http-transport/${TB_HOST}/tb-http-transport.%d{yyyy-MM-dd}.%i.log</fileNamePattern>
<fileNamePattern>/var/log/tb-http-transport/${TB_SERVICE_ID}/tb-http-transport.%d{yyyy-MM-dd}.%i.log</fileNamePattern>
<maxFileSize>100MB</maxFileSize>
<maxHistory>30</maxHistory>
<totalSizeCap>3GB</totalSizeCap>

2
docker/tb-transports/http/conf/tb-http-transport.conf

@ -14,7 +14,7 @@
# limitations under the License.
#
export JAVA_OPTS="$JAVA_OPTS -Xloggc:/var/log/tb-http-transport/${TB_HOST}/gc.log -XX:+IgnoreUnrecognizedVMOptions -XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=/var/log/tb-http-transport/${TB_HOST}/heapdump.bin -XX:+PrintGCDetails -XX:+PrintGCDateStamps"
export JAVA_OPTS="$JAVA_OPTS -Xloggc:/var/log/tb-http-transport/${TB_SERVICE_ID}/gc.log -XX:+IgnoreUnrecognizedVMOptions -XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=/var/log/tb-http-transport/${TB_SERVICE_ID}/heapdump.bin -XX:+PrintGCDetails -XX:+PrintGCDateStamps"
export JAVA_OPTS="$JAVA_OPTS -XX:+PrintHeapAtGC -XX:+PrintTenuringDistribution -XX:+PrintGCApplicationStoppedTime -XX:+UseGCLogFileRotation -XX:NumberOfGCLogFiles=10"
export JAVA_OPTS="$JAVA_OPTS -XX:GCLogFileSize=10M -XX:-UseBiasedLocking -XX:+UseTLAB -XX:+ResizeTLAB -XX:+PerfDisableSharedMem -XX:+UseCondCardMark"
export JAVA_OPTS="$JAVA_OPTS -XX:CMSWaitDuration=10000 -XX:+UseParNewGC -XX:+UseConcMarkSweepGC -XX:+CMSParallelRemarkEnabled -XX:+CMSParallelInitialMarkEnabled"

4
docker/tb-transports/mqtt/conf/logback.xml

@ -21,10 +21,10 @@
<appender name="fileLogAppender"
class="ch.qos.logback.core.rolling.RollingFileAppender">
<file>/var/log/tb-mqtt-transport/${TB_HOST}/tb-mqtt-transport.log</file>
<file>/var/log/tb-mqtt-transport/${TB_SERVICE_ID}/tb-mqtt-transport.log</file>
<rollingPolicy
class="ch.qos.logback.core.rolling.SizeAndTimeBasedRollingPolicy">
<fileNamePattern>/var/log/tb-mqtt-transport/${TB_HOST}/tb-mqtt-transport.%d{yyyy-MM-dd}.%i.log</fileNamePattern>
<fileNamePattern>/var/log/tb-mqtt-transport/${TB_SERVICE_ID}/tb-mqtt-transport.%d{yyyy-MM-dd}.%i.log</fileNamePattern>
<maxFileSize>100MB</maxFileSize>
<maxHistory>30</maxHistory>
<totalSizeCap>3GB</totalSizeCap>

2
docker/tb-transports/mqtt/conf/tb-mqtt-transport.conf

@ -14,7 +14,7 @@
# limitations under the License.
#
export JAVA_OPTS="$JAVA_OPTS -Xloggc:/var/log/tb-mqtt-transport/${TB_HOST}/gc.log -XX:+IgnoreUnrecognizedVMOptions -XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=/var/log/tb-mqtt-transport/${TB_HOST}/heapdump.bin -XX:+PrintGCDetails -XX:+PrintGCDateStamps"
export JAVA_OPTS="$JAVA_OPTS -Xloggc:/var/log/tb-mqtt-transport/${TB_SERVICE_ID}/gc.log -XX:+IgnoreUnrecognizedVMOptions -XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=/var/log/tb-mqtt-transport/${TB_SERVICE_ID}/heapdump.bin -XX:+PrintGCDetails -XX:+PrintGCDateStamps"
export JAVA_OPTS="$JAVA_OPTS -XX:+PrintHeapAtGC -XX:+PrintTenuringDistribution -XX:+PrintGCApplicationStoppedTime -XX:+UseGCLogFileRotation -XX:NumberOfGCLogFiles=10"
export JAVA_OPTS="$JAVA_OPTS -XX:GCLogFileSize=10M -XX:-UseBiasedLocking -XX:+UseTLAB -XX:+ResizeTLAB -XX:+PerfDisableSharedMem -XX:+UseCondCardMark"
export JAVA_OPTS="$JAVA_OPTS -XX:CMSWaitDuration=10000 -XX:+UseParNewGC -XX:+UseConcMarkSweepGC -XX:+CMSParallelRemarkEnabled -XX:+CMSParallelInitialMarkEnabled"

6
k8s/tb-coap-transport-configmap.yml

@ -23,7 +23,7 @@ metadata:
name: tb-coap-transport-config
data:
conf: |
export JAVA_OPTS="$JAVA_OPTS -Xloggc:/var/log/tb-coap-transport/${TB_HOST}/gc.log -XX:+IgnoreUnrecognizedVMOptions -XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=/var/log/tb-coap-transport/${TB_HOST}/heapdump.bin -XX:+PrintGCDetails -XX:+PrintGCDateStamps"
export JAVA_OPTS="$JAVA_OPTS -Xloggc:/var/log/tb-coap-transport/${TB_SERVICE_ID}/gc.log -XX:+IgnoreUnrecognizedVMOptions -XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=/var/log/tb-coap-transport/${TB_SERVICE_ID}/heapdump.bin -XX:+PrintGCDetails -XX:+PrintGCDateStamps"
export JAVA_OPTS="$JAVA_OPTS -XX:+PrintHeapAtGC -XX:+PrintTenuringDistribution -XX:+PrintGCApplicationStoppedTime -XX:+UseGCLogFileRotation -XX:NumberOfGCLogFiles=10"
export JAVA_OPTS="$JAVA_OPTS -XX:GCLogFileSize=10M -XX:-UseBiasedLocking -XX:+UseTLAB -XX:+ResizeTLAB -XX:+PerfDisableSharedMem -XX:+UseCondCardMark"
export JAVA_OPTS="$JAVA_OPTS -XX:CMSWaitDuration=10000 -XX:+UseParNewGC -XX:+UseConcMarkSweepGC -XX:+CMSParallelRemarkEnabled -XX:+CMSParallelInitialMarkEnabled"
@ -36,10 +36,10 @@ data:
<appender name="fileLogAppender"
class="ch.qos.logback.core.rolling.RollingFileAppender">
<file>/var/log/tb-coap-transport/${TB_HOST}/tb-coap-transport.log</file>
<file>/var/log/tb-coap-transport/${TB_SERVICE_ID}/tb-coap-transport.log</file>
<rollingPolicy
class="ch.qos.logback.core.rolling.SizeAndTimeBasedRollingPolicy">
<fileNamePattern>/var/log/tb-coap-transport/${TB_HOST}/tb-coap-transport.%d{yyyy-MM-dd}.%i.log</fileNamePattern>
<fileNamePattern>/var/log/tb-coap-transport/${TB_SERVICE_ID}/tb-coap-transport.%d{yyyy-MM-dd}.%i.log</fileNamePattern>
<maxFileSize>100MB</maxFileSize>
<maxHistory>30</maxHistory>
<totalSizeCap>3GB</totalSizeCap>

6
k8s/tb-http-transport-configmap.yml

@ -23,7 +23,7 @@ metadata:
name: tb-http-transport-config
data:
conf: |
export JAVA_OPTS="$JAVA_OPTS -Xloggc:/var/log/tb-http-transport/${TB_HOST}/gc.log -XX:+IgnoreUnrecognizedVMOptions -XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=/var/log/tb-http-transport/${TB_HOST}/heapdump.bin -XX:+PrintGCDetails -XX:+PrintGCDateStamps"
export JAVA_OPTS="$JAVA_OPTS -Xloggc:/var/log/tb-http-transport/${TB_SERVICE_ID}/gc.log -XX:+IgnoreUnrecognizedVMOptions -XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=/var/log/tb-http-transport/${TB_SERVICE_ID}/heapdump.bin -XX:+PrintGCDetails -XX:+PrintGCDateStamps"
export JAVA_OPTS="$JAVA_OPTS -XX:+PrintHeapAtGC -XX:+PrintTenuringDistribution -XX:+PrintGCApplicationStoppedTime -XX:+UseGCLogFileRotation -XX:NumberOfGCLogFiles=10"
export JAVA_OPTS="$JAVA_OPTS -XX:GCLogFileSize=10M -XX:-UseBiasedLocking -XX:+UseTLAB -XX:+ResizeTLAB -XX:+PerfDisableSharedMem -XX:+UseCondCardMark"
export JAVA_OPTS="$JAVA_OPTS -XX:CMSWaitDuration=10000 -XX:+UseParNewGC -XX:+UseConcMarkSweepGC -XX:+CMSParallelRemarkEnabled -XX:+CMSParallelInitialMarkEnabled"
@ -36,10 +36,10 @@ data:
<appender name="fileLogAppender"
class="ch.qos.logback.core.rolling.RollingFileAppender">
<file>/var/log/tb-http-transport/${TB_HOST}/tb-http-transport.log</file>
<file>/var/log/tb-http-transport/${TB_SERVICE_ID}/tb-http-transport.log</file>
<rollingPolicy
class="ch.qos.logback.core.rolling.SizeAndTimeBasedRollingPolicy">
<fileNamePattern>/var/log/tb-http-transport/${TB_HOST}/tb-http-transport.%d{yyyy-MM-dd}.%i.log</fileNamePattern>
<fileNamePattern>/var/log/tb-http-transport/${TB_SERVICE_ID}/tb-http-transport.%d{yyyy-MM-dd}.%i.log</fileNamePattern>
<maxFileSize>100MB</maxFileSize>
<maxHistory>30</maxHistory>
<totalSizeCap>3GB</totalSizeCap>

6
k8s/tb-mqtt-transport-configmap.yml

@ -23,7 +23,7 @@ metadata:
name: tb-mqtt-transport-config
data:
conf: |
export JAVA_OPTS="$JAVA_OPTS -Xloggc:/var/log/tb-mqtt-transport/${TB_HOST}/gc.log -XX:+IgnoreUnrecognizedVMOptions -XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=/var/log/tb-mqtt-transport/${TB_HOST}/heapdump.bin -XX:+PrintGCDetails -XX:+PrintGCDateStamps"
export JAVA_OPTS="$JAVA_OPTS -Xloggc:/var/log/tb-mqtt-transport/${TB_SERVICE_ID}/gc.log -XX:+IgnoreUnrecognizedVMOptions -XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=/var/log/tb-mqtt-transport/${TB_SERVICE_ID}/heapdump.bin -XX:+PrintGCDetails -XX:+PrintGCDateStamps"
export JAVA_OPTS="$JAVA_OPTS -XX:+PrintHeapAtGC -XX:+PrintTenuringDistribution -XX:+PrintGCApplicationStoppedTime -XX:+UseGCLogFileRotation -XX:NumberOfGCLogFiles=10"
export JAVA_OPTS="$JAVA_OPTS -XX:GCLogFileSize=10M -XX:-UseBiasedLocking -XX:+UseTLAB -XX:+ResizeTLAB -XX:+PerfDisableSharedMem -XX:+UseCondCardMark"
export JAVA_OPTS="$JAVA_OPTS -XX:CMSWaitDuration=10000 -XX:+UseParNewGC -XX:+UseConcMarkSweepGC -XX:+CMSParallelRemarkEnabled -XX:+CMSParallelInitialMarkEnabled"
@ -36,10 +36,10 @@ data:
<appender name="fileLogAppender"
class="ch.qos.logback.core.rolling.RollingFileAppender">
<file>/var/log/tb-mqtt-transport/${TB_HOST}/tb-mqtt-transport.log</file>
<file>/var/log/tb-mqtt-transport/${TB_SERVICE_ID}/tb-mqtt-transport.log</file>
<rollingPolicy
class="ch.qos.logback.core.rolling.SizeAndTimeBasedRollingPolicy">
<fileNamePattern>/var/log/tb-mqtt-transport/${TB_HOST}/tb-mqtt-transport.%d{yyyy-MM-dd}.%i.log</fileNamePattern>
<fileNamePattern>/var/log/tb-mqtt-transport/${TB_SERVICE_ID}/tb-mqtt-transport.%d{yyyy-MM-dd}.%i.log</fileNamePattern>
<maxFileSize>100MB</maxFileSize>
<maxHistory>30</maxHistory>
<totalSizeCap>3GB</totalSizeCap>

6
k8s/tb-node-configmap.yml

@ -24,7 +24,7 @@ metadata:
data:
conf: |
export JAVA_OPTS="$JAVA_OPTS -Dplatform=deb -Dinstall.data_dir=/usr/share/thingsboard/data"
export JAVA_OPTS="$JAVA_OPTS -Xloggc:/var/log/thingsboard/${TB_HOST}/gc.log -XX:+IgnoreUnrecognizedVMOptions -XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=/var/log/thingsboard/${TB_HOST}/heapdump.bin -XX:+PrintGCDetails -XX:+PrintGCDateStamps"
export JAVA_OPTS="$JAVA_OPTS -Xloggc:/var/log/thingsboard/${TB_SERVICE_ID}/gc.log -XX:+IgnoreUnrecognizedVMOptions -XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=/var/log/thingsboard/${TB_SERVICE_ID}/heapdump.bin -XX:+PrintGCDetails -XX:+PrintGCDateStamps"
export JAVA_OPTS="$JAVA_OPTS -XX:+PrintHeapAtGC -XX:+PrintTenuringDistribution -XX:+PrintGCApplicationStoppedTime -XX:+UseGCLogFileRotation -XX:NumberOfGCLogFiles=10"
export JAVA_OPTS="$JAVA_OPTS -XX:GCLogFileSize=10M -XX:-UseBiasedLocking -XX:+UseTLAB -XX:+ResizeTLAB -XX:+PerfDisableSharedMem -XX:+UseCondCardMark"
export JAVA_OPTS="$JAVA_OPTS -XX:CMSWaitDuration=10000 -XX:+UseParNewGC -XX:+UseConcMarkSweepGC -XX:+CMSParallelRemarkEnabled -XX:+CMSParallelInitialMarkEnabled"
@ -37,10 +37,10 @@ data:
<appender name="fileLogAppender"
class="ch.qos.logback.core.rolling.RollingFileAppender">
<file>/var/log/thingsboard/${TB_HOST}/thingsboard.log</file>
<file>/var/log/thingsboard/${TB_SERVICE_ID}/thingsboard.log</file>
<rollingPolicy
class="ch.qos.logback.core.rolling.SizeAndTimeBasedRollingPolicy">
<fileNamePattern>/var/log/thingsboard/${TB_HOST}/thingsboard.%d{yyyy-MM-dd}.%i.log</fileNamePattern>
<fileNamePattern>/var/log/thingsboard/${TB_SERVICE_ID}/thingsboard.%d{yyyy-MM-dd}.%i.log</fileNamePattern>
<maxFileSize>100MB</maxFileSize>
<maxHistory>30</maxHistory>
<totalSizeCap>3GB</totalSizeCap>

36
k8s/thingsboard.yml

@ -247,18 +247,12 @@ spec:
- containerPort: 9001
name: rpc
env:
- name: RPC_HOST
valueFrom:
fieldRef:
fieldPath: status.podIP
- name: CLUSTER_NODE_ID
valueFrom:
fieldRef:
fieldPath: metadata.name
- name: TB_HOST
- name: TB_SERVICE_ID
valueFrom:
fieldRef:
fieldPath: metadata.name
- name: TB_SERVICE_TYPE
value: "monolith"
- name: ZOOKEEPER_ENABLED
value: "true"
- name: ZOOKEEPER_URL
@ -334,14 +328,12 @@ spec:
- containerPort: 1883
name: mqtt
env:
- name: CLUSTER_NODE_ID
valueFrom:
fieldRef:
fieldPath: metadata.name
- name: TB_HOST
- name: TB_SERVICE_ID
valueFrom:
fieldRef:
fieldPath: metadata.name
- name: TB_SERVICE_TYPE
value: "monolith"
- name: MQTT_BIND_ADDRESS
value: "0.0.0.0"
- name: MQTT_BIND_PORT
@ -409,14 +401,12 @@ spec:
- containerPort: 8080
name: http
env:
- name: CLUSTER_NODE_ID
valueFrom:
fieldRef:
fieldPath: metadata.name
- name: TB_HOST
- name: TB_SERVICE_ID
valueFrom:
fieldRef:
fieldPath: metadata.name
- name: TB_SERVICE_TYPE
value: "monolith"
- name: HTTP_BIND_ADDRESS
value: "0.0.0.0"
- name: HTTP_BIND_PORT
@ -484,14 +474,12 @@ spec:
name: coap
protocol: UDP
env:
- name: CLUSTER_NODE_ID
valueFrom:
fieldRef:
fieldPath: metadata.name
- name: TB_HOST
- name: TB_SERVICE_ID
valueFrom:
fieldRef:
fieldPath: metadata.name
- name: TB_SERVICE_TYPE
value: "monolith"
- name: COAP_BIND_ADDRESS
value: "0.0.0.0"
- name: COAP_BIND_PORT

2
msa/black-box-tests/src/test/java/org/thingsboard/server/msa/ThingsBoardDbInstaller.java

@ -92,7 +92,7 @@ public class ThingsBoardDbInstaller extends ExternalResource {
dockerCompose.withCommand("up -d redis postgres");
dockerCompose.invokeCompose();
dockerCompose.withCommand("run --no-deps --rm -e INSTALL_TB=true -e LOAD_DEMO=true tb1");
dockerCompose.withCommand("run --no-deps --rm -e INSTALL_TB=true -e LOAD_DEMO=true tb-core1");
dockerCompose.invokeCompose();
} finally {

2
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/TbGetDeviceAttrNode.java

@ -30,7 +30,7 @@ import org.thingsboard.server.common.msg.TbMsg;
@Slf4j
@RuleNode(type = ComponentType.ENRICHMENT,
name = "device attributes",
name = "related device attributes",
configClazz = TbGetDeviceAttrNodeConfiguration.class,
nodeDescription = "Add Originators Related Device Attributes and Latest Telemetry value into Message Metadata",
nodeDetails = "If Attributes enrichment configured, <b>CLIENT/SHARED/SERVER</b> attributes are added into Message metadata " +

2
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/transaction/TbSynchronizationBeginNode.java

@ -33,7 +33,7 @@ import static org.thingsboard.rule.engine.api.TbRelationTypes.SUCCESS;
type = ComponentType.ACTION,
name = "synchronization start",
configClazz = EmptyNodeConfiguration.class,
nodeDescription = "Starts synchronization of message processing based on message originator",
nodeDescription = "This Node is now deprecated. Use \"Checkpoint\" instead.",
nodeDetails = "This node should be used together with \"synchronization end\" node. \n This node will put messages into queue based on message originator id. \n" +
"Subsequent messages will not be processed until the previous message processing is completed or timeout event occurs.\n" +
"Size of the queue per originator and timeout values are configurable on a system level",

2
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/transaction/TbSynchronizationEndNode.java

@ -35,7 +35,7 @@ import static org.thingsboard.rule.engine.api.TbRelationTypes.SUCCESS;
type = ComponentType.ACTION,
name = "synchronization end",
configClazz = EmptyNodeConfiguration.class,
nodeDescription = "Stops synchronization of message processing based on message originator",
nodeDescription = "This Node is now deprecated. Use \"Checkpoint\" instead.",
nodeDetails = "",
uiResources = {"static/rulenode/rulenode-core-config.js"},
configDirective = ("tbNodeEmptyConfig")

Loading…
Cancel
Save