diff --git a/application/pom.xml b/application/pom.xml
index 3693fe6dbe..96c9ab7b39 100644
--- a/application/pom.xml
+++ b/application/pom.xml
@@ -286,7 +286,6 @@
com.jayway.jsonpath
json-path
- test
com.jayway.jsonpath
diff --git a/application/src/main/data/upgrade/3.4.1/schema_update.sql b/application/src/main/data/upgrade/3.4.1/schema_update.sql
new file mode 100644
index 0000000000..79cde8167f
--- /dev/null
+++ b/application/src/main/data/upgrade/3.4.1/schema_update.sql
@@ -0,0 +1,74 @@
+--
+-- Copyright © 2016-2022 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.
+--
+
+DO
+$$
+ DECLARE table_partition RECORD;
+ BEGIN
+ -- in case of running the upgrade script a second time:
+ IF NOT (SELECT exists(SELECT FROM pg_tables WHERE tablename = 'old_audit_log')) THEN
+ ALTER TABLE audit_log RENAME TO old_audit_log;
+ ALTER INDEX IF EXISTS idx_audit_log_tenant_id_and_created_time RENAME TO idx_old_audit_log_tenant_id_and_created_time;
+
+ FOR table_partition IN SELECT tablename AS name, split_part(tablename, '_', 3) AS partition_ts
+ FROM pg_tables WHERE tablename LIKE 'audit_log_%'
+ LOOP
+ EXECUTE format('ALTER TABLE %s RENAME TO old_audit_log_%s', table_partition.name, table_partition.partition_ts);
+ END LOOP;
+ ELSE
+ RAISE NOTICE 'Table old_audit_log already exists, leaving as is';
+ END IF;
+ END;
+$$;
+
+CREATE TABLE IF NOT EXISTS audit_log (
+ id uuid NOT NULL,
+ created_time bigint NOT NULL,
+ tenant_id uuid,
+ customer_id uuid,
+ entity_id uuid,
+ entity_type varchar(255),
+ entity_name varchar(255),
+ user_id uuid,
+ user_name varchar(255),
+ action_type varchar(255),
+ action_data varchar(1000000),
+ action_status varchar(255),
+ action_failure_details varchar(1000000)
+) PARTITION BY RANGE (created_time);
+CREATE INDEX IF NOT EXISTS idx_audit_log_tenant_id_and_created_time ON audit_log(tenant_id, created_time DESC);
+
+CREATE OR REPLACE PROCEDURE migrate_audit_logs(IN start_time_ms BIGINT, IN end_time_ms BIGINT, IN partition_size_ms BIGINT)
+ LANGUAGE plpgsql AS
+$$
+DECLARE
+ p RECORD;
+ partition_end_ts BIGINT;
+BEGIN
+ FOR p IN SELECT DISTINCT (created_time - created_time % partition_size_ms) AS partition_ts FROM old_audit_log
+ WHERE created_time >= start_time_ms AND created_time < end_time_ms
+ LOOP
+ partition_end_ts = p.partition_ts + partition_size_ms;
+ RAISE NOTICE '[audit_log] Partition to create : [%-%]', p.partition_ts, partition_end_ts;
+ EXECUTE format('CREATE TABLE IF NOT EXISTS audit_log_%s PARTITION OF audit_log ' ||
+ 'FOR VALUES FROM ( %s ) TO ( %s )', p.partition_ts, p.partition_ts, partition_end_ts);
+ END LOOP;
+
+ INSERT INTO audit_log
+ SELECT * FROM old_audit_log
+ WHERE created_time >= start_time_ms AND created_time < end_time_ms;
+END;
+$$;
diff --git a/application/src/main/java/org/thingsboard/server/actors/app/AppActor.java b/application/src/main/java/org/thingsboard/server/actors/app/AppActor.java
index 2817b1e359..eeb6d8e82e 100644
--- a/application/src/main/java/org/thingsboard/server/actors/app/AppActor.java
+++ b/application/src/main/java/org/thingsboard/server/actors/app/AppActor.java
@@ -36,7 +36,7 @@ import org.thingsboard.server.common.data.plugin.ComponentLifecycleEvent;
import org.thingsboard.server.common.msg.MsgType;
import org.thingsboard.server.common.msg.TbActorMsg;
import org.thingsboard.server.common.msg.aware.TenantAwareMsg;
-import org.thingsboard.server.common.msg.edge.EdgeEventUpdateMsg;
+import org.thingsboard.server.common.msg.edge.EdgeSessionMsg;
import org.thingsboard.server.common.msg.plugin.ComponentLifecycleMsg;
import org.thingsboard.server.common.msg.queue.QueueToRuleEngineMsg;
import org.thingsboard.server.common.msg.queue.RuleEngineException;
@@ -105,7 +105,9 @@ public class AppActor extends ContextAwareActor {
onToDeviceActorMsg((TenantAwareMsg) msg, true);
break;
case EDGE_EVENT_UPDATE_TO_EDGE_SESSION_MSG:
- onToTenantActorMsg((EdgeEventUpdateMsg) msg);
+ case EDGE_SYNC_REQUEST_TO_EDGE_SESSION_MSG:
+ case EDGE_SYNC_RESPONSE_FROM_EDGE_SESSION_MSG:
+ onToEdgeSessionMsg((EdgeSessionMsg) msg);
break;
case SESSION_TIMEOUT_MSG:
ctx.broadcastToChildrenByType(msg, EntityType.TENANT);
@@ -193,7 +195,7 @@ public class AppActor extends ContextAwareActor {
() -> new TenantActor.ActorCreator(systemContext, tenantId));
}
- private void onToTenantActorMsg(EdgeEventUpdateMsg msg) {
+ private void onToEdgeSessionMsg(EdgeSessionMsg msg) {
TbActorRef target = null;
if (ModelConstants.SYSTEM_TENANT.equals(msg.getTenantId())) {
log.warn("Message has system tenant id: {}", msg);
@@ -203,7 +205,7 @@ public class AppActor extends ContextAwareActor {
if (target != null) {
target.tellWithHighPriority(msg);
} else {
- log.debug("[{}] Invalid edge event update msg: {}", msg.getTenantId(), msg);
+ log.debug("[{}] Invalid edge session msg: {}", msg.getTenantId(), msg);
}
}
diff --git a/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java b/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java
index 239466fe81..3eec2e942b 100644
--- a/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java
+++ b/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java
@@ -17,8 +17,11 @@ package org.thingsboard.server.actors.ruleChain;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ArrayNode;
+import com.fasterxml.jackson.databind.node.ObjectNode;
import io.netty.channel.EventLoopGroup;
import lombok.extern.slf4j.Slf4j;
+import org.thingsboard.common.util.JacksonUtil;
import org.thingsboard.common.util.ListeningExecutor;
import org.thingsboard.rule.engine.api.MailService;
import org.thingsboard.rule.engine.api.RuleEngineAlarmService;
@@ -40,6 +43,7 @@ import org.thingsboard.server.common.data.DataConstants;
import org.thingsboard.server.common.data.Device;
import org.thingsboard.server.common.data.DeviceProfile;
import org.thingsboard.server.common.data.EntityType;
+import org.thingsboard.server.common.data.HasRuleEngineProfile;
import org.thingsboard.server.common.data.StringUtils;
import org.thingsboard.server.common.data.TenantProfile;
import org.thingsboard.server.common.data.alarm.Alarm;
@@ -53,6 +57,7 @@ import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.id.RuleChainId;
import org.thingsboard.server.common.data.id.RuleNodeId;
import org.thingsboard.server.common.data.id.TenantId;
+import org.thingsboard.server.common.data.kv.AttributeKvEntry;
import org.thingsboard.server.common.data.page.PageData;
import org.thingsboard.server.common.data.page.PageLink;
import org.thingsboard.server.common.data.rule.RuleNode;
@@ -89,6 +94,7 @@ import org.thingsboard.server.queue.TbQueueMsgMetadata;
import org.thingsboard.server.service.script.RuleNodeJsScriptEngine;
import java.util.Collections;
+import java.util.List;
import java.util.Set;
import java.util.function.BiConsumer;
import java.util.function.Consumer;
@@ -331,58 +337,62 @@ class DefaultTbContext implements TbContext {
}
public TbMsg deviceCreatedMsg(Device device, RuleNodeId ruleNodeId) {
- RuleChainId ruleChainId = null;
- String queueName = null;
+ DeviceProfile deviceProfile = null;
if (device.getDeviceProfileId() != null) {
- DeviceProfile deviceProfile = mainCtx.getDeviceProfileCache().find(device.getDeviceProfileId());
- if (deviceProfile == null) {
- log.warn("[{}] Device profile is null!", device.getDeviceProfileId());
- } else {
- ruleChainId = deviceProfile.getDefaultRuleChainId();
- queueName = deviceProfile.getDefaultQueueName();
- }
+ deviceProfile = mainCtx.getDeviceProfileCache().find(device.getDeviceProfileId());
}
- return entityActionMsg(device, device.getId(), ruleNodeId, DataConstants.ENTITY_CREATED, queueName, ruleChainId);
+ return entityActionMsg(device, device.getId(), ruleNodeId, DataConstants.ENTITY_CREATED, deviceProfile);
}
public TbMsg assetCreatedMsg(Asset asset, RuleNodeId ruleNodeId) {
- RuleChainId ruleChainId = null;
- String queueName = null;
+ AssetProfile assetProfile = null;
if (asset.getAssetProfileId() != null) {
- AssetProfile assetProfile = mainCtx.getAssetProfileCache().find(asset.getAssetProfileId());
- if (assetProfile == null) {
- log.warn("[{}] Asset profile is null!", asset.getAssetProfileId());
- } else {
- ruleChainId = assetProfile.getDefaultRuleChainId();
- queueName = assetProfile.getDefaultQueueName();
- }
+ assetProfile = mainCtx.getAssetProfileCache().find(asset.getAssetProfileId());
}
- return entityActionMsg(asset, asset.getId(), ruleNodeId, DataConstants.ENTITY_CREATED, queueName, ruleChainId);
+ return entityActionMsg(asset, asset.getId(), ruleNodeId, DataConstants.ENTITY_CREATED, assetProfile);
}
public TbMsg alarmActionMsg(Alarm alarm, RuleNodeId ruleNodeId, String action) {
- RuleChainId ruleChainId = null;
- String queueName = null;
+ HasRuleEngineProfile profile = null;
if (EntityType.DEVICE.equals(alarm.getOriginator().getEntityType())) {
DeviceId deviceId = new DeviceId(alarm.getOriginator().getId());
- DeviceProfile deviceProfile = mainCtx.getDeviceProfileCache().get(getTenantId(), deviceId);
- if (deviceProfile == null) {
- log.warn("[{}] Device profile is null!", deviceId);
- } else {
- ruleChainId = deviceProfile.getDefaultRuleChainId();
- queueName = deviceProfile.getDefaultQueueName();
- }
+ profile = mainCtx.getDeviceProfileCache().get(getTenantId(), deviceId);
} else if (EntityType.ASSET.equals(alarm.getOriginator().getEntityType())) {
AssetId assetId = new AssetId(alarm.getOriginator().getId());
- AssetProfile assetProfile = mainCtx.getAssetProfileCache().get(getTenantId(), assetId);
- if (assetProfile == null) {
- log.warn("[{}] Asset profile is null!", assetId);
- } else {
- ruleChainId = assetProfile.getDefaultRuleChainId();
- queueName = assetProfile.getDefaultQueueName();
- }
+ profile = mainCtx.getAssetProfileCache().get(getTenantId(), assetId);
+ }
+ return entityActionMsg(alarm, alarm.getOriginator(), ruleNodeId, action, profile);
+ }
+
+ public TbMsg attributesUpdatedActionMsg(EntityId originator, RuleNodeId ruleNodeId, String scope, List attributes) {
+ ObjectNode entityNode = JacksonUtil.newObjectNode();
+ if (attributes != null) {
+ attributes.forEach(attributeKvEntry -> JacksonUtil.addKvEntry(entityNode, attributeKvEntry));
}
- return entityActionMsg(alarm, alarm.getId(), ruleNodeId, action, queueName, ruleChainId);
+ return attributesActionMsg(originator, ruleNodeId, scope, DataConstants.ATTRIBUTES_UPDATED, JacksonUtil.toString(entityNode));
+ }
+
+ public TbMsg attributesDeletedActionMsg(EntityId originator, RuleNodeId ruleNodeId, String scope, List keys) {
+ ObjectNode entityNode = JacksonUtil.newObjectNode();
+ ArrayNode attrsArrayNode = entityNode.putArray("attributes");
+ if (keys != null) {
+ keys.forEach(attrsArrayNode::add);
+ }
+ return attributesActionMsg(originator, ruleNodeId, scope, DataConstants.ATTRIBUTES_DELETED, JacksonUtil.toString(entityNode));
+ }
+
+ private TbMsg attributesActionMsg(EntityId originator, RuleNodeId ruleNodeId, String scope, String action, String msgData) {
+ TbMsgMetaData tbMsgMetaData = getActionMetaData(ruleNodeId);
+ tbMsgMetaData.putValue("scope", scope);
+ HasRuleEngineProfile profile = null;
+ if (EntityType.DEVICE.equals(originator.getEntityType())) {
+ DeviceId deviceId = new DeviceId(originator.getId());
+ profile = mainCtx.getDeviceProfileCache().get(getTenantId(), deviceId);
+ } else if (EntityType.ASSET.equals(originator.getEntityType())) {
+ AssetId assetId = new AssetId(originator.getId());
+ profile = mainCtx.getAssetProfileCache().get(getTenantId(), assetId);
+ }
+ return entityActionMsg(originator, tbMsgMetaData, msgData, action, profile);
}
@Override
@@ -391,17 +401,27 @@ class DefaultTbContext implements TbContext {
}
public TbMsg entityActionMsg(E entity, I id, RuleNodeId ruleNodeId, String action) {
- return entityActionMsg(entity, id, ruleNodeId, action, null, null);
+ return entityActionMsg(entity, id, ruleNodeId, action, null);
}
- public TbMsg entityActionMsg(E entity, I id, RuleNodeId ruleNodeId, String action, String queueName, RuleChainId ruleChainId) {
+ public TbMsg entityActionMsg(E entity, I id, RuleNodeId ruleNodeId, String action, K profile) {
try {
- return TbMsg.newMsg(queueName, action, id, getActionMetaData(ruleNodeId), mapper.writeValueAsString(mapper.valueToTree(entity)), ruleChainId, null);
+ return entityActionMsg(id, getActionMetaData(ruleNodeId), mapper.writeValueAsString(mapper.valueToTree(entity)), action, profile);
} catch (JsonProcessingException | IllegalArgumentException e) {
throw new RuntimeException("Failed to process " + id.getEntityType().name().toLowerCase() + " " + action + " msg: " + e);
}
}
+ private TbMsg entityActionMsg(I id, TbMsgMetaData msgMetaData, String msgData, String action, K profile) {
+ String defaultQueueName = null;
+ RuleChainId defaultRuleChainId = null;
+ if (profile != null) {
+ defaultQueueName = profile.getDefaultQueueName();
+ defaultRuleChainId = profile.getDefaultRuleChainId();
+ }
+ return TbMsg.newMsg(defaultQueueName, action, id, msgMetaData, msgData, defaultRuleChainId, null);
+ }
+
@Override
public RuleNodeId getSelfId() {
return nodeCtx.getSelf().getId();
diff --git a/application/src/main/java/org/thingsboard/server/actors/tenant/TenantActor.java b/application/src/main/java/org/thingsboard/server/actors/tenant/TenantActor.java
index c22b44d7cf..5718f581ad 100644
--- a/application/src/main/java/org/thingsboard/server/actors/tenant/TenantActor.java
+++ b/application/src/main/java/org/thingsboard/server/actors/tenant/TenantActor.java
@@ -47,7 +47,7 @@ import org.thingsboard.server.common.msg.TbActorMsg;
import org.thingsboard.server.common.msg.TbMsg;
import org.thingsboard.server.common.msg.aware.DeviceAwareMsg;
import org.thingsboard.server.common.msg.aware.RuleChainAwareMsg;
-import org.thingsboard.server.common.msg.edge.EdgeEventUpdateMsg;
+import org.thingsboard.server.common.msg.edge.EdgeSessionMsg;
import org.thingsboard.server.common.msg.plugin.ComponentLifecycleMsg;
import org.thingsboard.server.common.msg.queue.PartitionChangeMsg;
import org.thingsboard.server.common.msg.queue.QueueToRuleEngineMsg;
@@ -167,7 +167,9 @@ public class TenantActor extends RuleChainManagerActor {
onRuleChainMsg((RuleChainAwareMsg) msg);
break;
case EDGE_EVENT_UPDATE_TO_EDGE_SESSION_MSG:
- onToEdgeSessionMsg((EdgeEventUpdateMsg) msg);
+ case EDGE_SYNC_REQUEST_TO_EDGE_SESSION_MSG:
+ case EDGE_SYNC_RESPONSE_FROM_EDGE_SESSION_MSG:
+ onToEdgeSessionMsg((EdgeSessionMsg) msg);
break;
default:
return false;
@@ -271,9 +273,8 @@ public class TenantActor extends RuleChainManagerActor {
() -> new DeviceActorCreator(systemContext, tenantId, deviceId));
}
- private void onToEdgeSessionMsg(EdgeEventUpdateMsg msg) {
- log.trace("[{}] onToEdgeSessionMsg [{}]", msg.getTenantId(), msg);
- systemContext.getEdgeRpcService().onEdgeEvent(tenantId, msg.getEdgeId());
+ private void onToEdgeSessionMsg(EdgeSessionMsg msg) {
+ systemContext.getEdgeRpcService().onToEdgeSessionMsg(tenantId, msg);
}
private ApiUsageState getApiUsageState() {
diff --git a/application/src/main/java/org/thingsboard/server/controller/BaseController.java b/application/src/main/java/org/thingsboard/server/controller/BaseController.java
index 21edd71ddd..ffc3fcf3cd 100644
--- a/application/src/main/java/org/thingsboard/server/controller/BaseController.java
+++ b/application/src/main/java/org/thingsboard/server/controller/BaseController.java
@@ -279,10 +279,7 @@ public abstract class BaseController {
protected EdgeService edgeService;
@Autowired(required = false)
- protected EdgeNotificationService edgeNotificationService;
-
- @Autowired(required = false)
- protected EdgeRpcService edgeGrpcService;
+ protected EdgeRpcService edgeRpcService;
@Autowired
protected TbNotificationEntityService notificationEntityService;
diff --git a/application/src/main/java/org/thingsboard/server/controller/EdgeController.java b/application/src/main/java/org/thingsboard/server/controller/EdgeController.java
index 59c9eb39c1..6a5dd44d0b 100644
--- a/application/src/main/java/org/thingsboard/server/controller/EdgeController.java
+++ b/application/src/main/java/org/thingsboard/server/controller/EdgeController.java
@@ -22,6 +22,7 @@ import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.http.HttpStatus;
import org.springframework.http.MediaType;
+import org.springframework.http.ResponseEntity;
import org.springframework.security.access.prepost.PreAuthorize;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.PostMapping;
@@ -32,6 +33,7 @@ import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.ResponseBody;
import org.springframework.web.bind.annotation.ResponseStatus;
import org.springframework.web.bind.annotation.RestController;
+import org.springframework.web.context.request.async.DeferredResult;
import org.thingsboard.rule.engine.flow.TbRuleChainInputNode;
import org.thingsboard.server.common.data.Customer;
import org.thingsboard.server.common.data.EntitySubtype;
@@ -47,6 +49,8 @@ import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.page.PageData;
import org.thingsboard.server.common.data.page.PageLink;
import org.thingsboard.server.common.data.rule.RuleChain;
+import org.thingsboard.server.common.msg.edge.FromEdgeSyncResponse;
+import org.thingsboard.server.common.msg.edge.ToEdgeSyncRequest;
import org.thingsboard.server.dao.exception.DataValidationException;
import org.thingsboard.server.dao.exception.IncorrectParameterException;
import org.thingsboard.server.dao.model.ModelConstants;
@@ -61,6 +65,7 @@ import org.thingsboard.server.service.security.permission.Resource;
import java.util.ArrayList;
import java.util.List;
+import java.util.UUID;
import java.util.stream.Collectors;
import static org.thingsboard.server.controller.ControllerConstants.CUSTOMER_ID_PARAM_DESCRIPTION;
@@ -529,24 +534,35 @@ public class EdgeController extends BaseController {
"All entities that are assigned to particular edge are going to be send to remote edge service." + TENANT_AUTHORITY_PARAGRAPH)
@PreAuthorize("hasAuthority('TENANT_ADMIN')")
@RequestMapping(value = "/edge/sync/{edgeId}", method = RequestMethod.POST)
- public void syncEdge(@ApiParam(value = EDGE_ID_PARAM_DESCRIPTION, required = true)
+ public DeferredResult syncEdge(@ApiParam(value = EDGE_ID_PARAM_DESCRIPTION, required = true)
@PathVariable("edgeId") String strEdgeId) throws ThingsboardException {
checkParameter("edgeId", strEdgeId);
try {
+ final DeferredResult response = new DeferredResult<>();
if (isEdgesEnabled()) {
EdgeId edgeId = new EdgeId(toUUID(strEdgeId));
edgeId = checkNotNull(edgeId);
SecurityUser user = getCurrentUser();
TenantId tenantId = user.getTenantId();
- edgeGrpcService.startSyncProcess(tenantId, edgeId);
+ ToEdgeSyncRequest request = new ToEdgeSyncRequest(UUID.randomUUID(), tenantId, edgeId);
+ edgeRpcService.processSyncRequest(request, fromEdgeSyncResponse -> reply(response, fromEdgeSyncResponse));
} else {
throw new ThingsboardException("Edges support disabled", ThingsboardErrorCode.GENERAL);
}
+ return response;
} catch (Exception e) {
throw handleException(e);
}
}
+ private void reply(DeferredResult response, FromEdgeSyncResponse fromEdgeSyncResponse) {
+ if (fromEdgeSyncResponse.isSuccess()) {
+ response.setResult(new ResponseEntity<>(HttpStatus.OK));
+ } else {
+ response.setErrorResult(new ThingsboardException("Edge is not connected", ThingsboardErrorCode.GENERAL));
+ }
+ }
+
@ApiOperation(value = "Find missing rule chains (findMissingToRelatedRuleChains)",
notes = "Returns list of rule chains ids that are not assigned to particular edge, but these rule chains are present in the already assigned rule chains to edge." + TENANT_AUTHORITY_PARAGRAPH)
@PreAuthorize("hasAuthority('TENANT_ADMIN')")
diff --git a/application/src/main/java/org/thingsboard/server/install/ThingsboardInstallService.java b/application/src/main/java/org/thingsboard/server/install/ThingsboardInstallService.java
index c1f352d8eb..33a6d4bdd1 100644
--- a/application/src/main/java/org/thingsboard/server/install/ThingsboardInstallService.java
+++ b/application/src/main/java/org/thingsboard/server/install/ThingsboardInstallService.java
@@ -228,6 +228,7 @@ public class ThingsboardInstallService {
case "3.4.1":
log.info("Upgrading ThingsBoard from version 3.4.1 to 3.4.2 ...");
databaseEntitiesUpgradeService.upgradeDatabase("3.4.1");
+ dataUpdateService.updateData("3.4.1");
log.info("Updating system data...");
systemDataLoaderService.updateSystemWidgets();
break;
diff --git a/application/src/main/java/org/thingsboard/server/service/action/EntityActionService.java b/application/src/main/java/org/thingsboard/server/service/action/EntityActionService.java
index 79968187b6..7057f903be 100644
--- a/application/src/main/java/org/thingsboard/server/service/action/EntityActionService.java
+++ b/application/src/main/java/org/thingsboard/server/service/action/EntityActionService.java
@@ -21,6 +21,7 @@ import com.fasterxml.jackson.databind.node.ObjectNode;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
+import org.thingsboard.common.util.JacksonUtil;
import org.thingsboard.server.cluster.TbClusterService;
import org.thingsboard.server.common.data.DataConstants;
import org.thingsboard.server.common.data.EntityType;
@@ -34,9 +35,8 @@ import org.thingsboard.server.common.data.id.CustomerId;
import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.kv.AttributeKvEntry;
-import org.thingsboard.server.common.data.kv.DataType;
-import org.thingsboard.server.common.data.kv.KvEntry;
import org.thingsboard.server.common.data.kv.TsKvEntry;
+import org.thingsboard.server.common.data.relation.EntityRelation;
import org.thingsboard.server.common.msg.TbMsg;
import org.thingsboard.server.common.msg.TbMsgDataType;
import org.thingsboard.server.common.msg.TbMsgMetaData;
@@ -113,6 +113,15 @@ public class EntityActionService {
case UNASSIGNED_FROM_EDGE:
msgType = DataConstants.ENTITY_UNASSIGNED_FROM_EDGE;
break;
+ case RELATION_ADD_OR_UPDATE:
+ msgType = DataConstants.RELATION_ADD_OR_UPDATE;
+ break;
+ case RELATION_DELETED:
+ msgType = DataConstants.RELATION_DELETED;
+ break;
+ case RELATIONS_DELETED:
+ msgType = DataConstants.RELATIONS_DELETED;
+ break;
}
if (!StringUtils.isEmpty(msgType)) {
try {
@@ -170,7 +179,7 @@ public class EntityActionService {
metaData.putValue(DataConstants.SCOPE, scope);
if (attributes != null) {
for (AttributeKvEntry attr : attributes) {
- addKvEntry(entityNode, attr);
+ JacksonUtil.addKvEntry(entityNode, attr);
}
}
} else if (actionType == ActionType.ATTRIBUTES_DELETED) {
@@ -195,6 +204,8 @@ public class EntityActionService {
}
entityNode.put("startTs", extractParameter(Long.class, 1, additionalInfo));
entityNode.put("endTs", extractParameter(Long.class, 2, additionalInfo));
+ } else if (ActionType.RELATION_ADD_OR_UPDATE.equals(actionType) || ActionType.RELATION_DELETED.equals(actionType)) {
+ entityNode = json.valueToTree(extractParameter(EntityRelation.class, 0, additionalInfo));
}
}
TbMsg tbMsg = TbMsg.newMsg(msgType, entityId, customerId, metaData, TbMsgDataType.JSON, json.writeValueAsString(entityNode));
@@ -246,27 +257,11 @@ public class EntityActionService {
element.put("ts", entry.getKey());
ObjectNode values = element.putObject("values");
for (TsKvEntry tsKvEntry : entry.getValue()) {
- addKvEntry(values, tsKvEntry);
+ JacksonUtil.addKvEntry(values, tsKvEntry);
}
result.add(element);
}
}
}
- private void addKvEntry(ObjectNode entityNode, KvEntry kvEntry) throws Exception {
- if (kvEntry.getDataType() == DataType.BOOLEAN) {
- kvEntry.getBooleanValue().ifPresent(value -> entityNode.put(kvEntry.getKey(), value));
- } else if (kvEntry.getDataType() == DataType.DOUBLE) {
- kvEntry.getDoubleValue().ifPresent(value -> entityNode.put(kvEntry.getKey(), value));
- } else if (kvEntry.getDataType() == DataType.LONG) {
- kvEntry.getLongValue().ifPresent(value -> entityNode.put(kvEntry.getKey(), value));
- } else if (kvEntry.getDataType() == DataType.JSON) {
- if (kvEntry.getJsonValue().isPresent()) {
- entityNode.set(kvEntry.getKey(), json.readTree(kvEntry.getJsonValue().get()));
- }
- } else {
- entityNode.put(kvEntry.getKey(), kvEntry.getValueAsString());
- }
- }
-
}
diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcService.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcService.java
index c56fcd4ed9..4cb6fcdb0a 100644
--- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcService.java
+++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcService.java
@@ -26,6 +26,7 @@ import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Service;
import org.thingsboard.common.util.ThingsBoardThreadFactory;
+import org.thingsboard.server.cluster.TbClusterService;
import org.thingsboard.server.common.data.DataConstants;
import org.thingsboard.server.common.data.ResourceUtils;
import org.thingsboard.server.common.data.edge.Edge;
@@ -34,6 +35,10 @@ import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.kv.BasicTsKvEntry;
import org.thingsboard.server.common.data.kv.BooleanDataEntry;
import org.thingsboard.server.common.data.kv.LongDataEntry;
+import org.thingsboard.server.common.msg.edge.EdgeEventUpdateMsg;
+import org.thingsboard.server.common.msg.edge.EdgeSessionMsg;
+import org.thingsboard.server.common.msg.edge.FromEdgeSyncResponse;
+import org.thingsboard.server.common.msg.edge.ToEdgeSyncRequest;
import org.thingsboard.server.gen.edge.v1.EdgeRpcServiceGrpc;
import org.thingsboard.server.gen.edge.v1.RequestMsg;
import org.thingsboard.server.gen.edge.v1.ResponseMsg;
@@ -59,6 +64,7 @@ import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
+import java.util.function.Consumer;
@Service
@Slf4j
@@ -71,6 +77,8 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i
private final Map sessionNewEvents = new HashMap<>();
private final ConcurrentMap> sessionEdgeEventChecks = new ConcurrentHashMap<>();
+ private final ConcurrentMap> localSyncEdgeRequests = new ConcurrentHashMap<>();
+
@Value("${edges.rpc.port}")
private int rpcPort;
@Value("${edges.rpc.ssl.enabled}")
@@ -98,12 +106,17 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i
@Autowired
private TelemetrySubscriptionService tsSubService;
+ @Autowired
+ private TbClusterService clusterService;
+
private Server server;
private ScheduledExecutorService edgeEventProcessingExecutorService;
private ScheduledExecutorService sendDownlinkExecutorService;
+ private ScheduledExecutorService executorService;
+
@PostConstruct
public void init() {
log.info("Initializing Edge RPC service!");
@@ -129,8 +142,9 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i
log.error("Failed to start Edge RPC server!", e);
throw new RuntimeException("Failed to start Edge RPC server!");
}
- this.edgeEventProcessingExecutorService = Executors.newScheduledThreadPool(schedulerPoolSize, ThingsBoardThreadFactory.forName("edge-scheduler"));
+ this.edgeEventProcessingExecutorService = Executors.newScheduledThreadPool(schedulerPoolSize, ThingsBoardThreadFactory.forName("edge-event-check-scheduler"));
this.sendDownlinkExecutorService = Executors.newScheduledThreadPool(sendSchedulerPoolSize, ThingsBoardThreadFactory.forName("edge-send-scheduler"));
+ this.executorService = Executors.newSingleThreadScheduledExecutor(ThingsBoardThreadFactory.forName("edge-service"));
log.info("Edge RPC service initialized!");
}
@@ -153,6 +167,9 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i
if (sendDownlinkExecutorService != null) {
sendDownlinkExecutorService.shutdownNow();
}
+ if (executorService != null) {
+ executorService.shutdownNow();
+ }
}
@Override
@@ -160,47 +177,76 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i
return new EdgeGrpcSession(ctx, outputStream, this::onEdgeConnect, this::onEdgeDisconnect, sendDownlinkExecutorService).getInputStream();
}
+ @Override
+ public void onToEdgeSessionMsg(TenantId tenantId, EdgeSessionMsg msg) {
+ executorService.execute(() -> {
+ switch (msg.getMsgType()) {
+ case EDGE_EVENT_UPDATE_TO_EDGE_SESSION_MSG:
+ EdgeEventUpdateMsg edgeEventUpdateMsg = (EdgeEventUpdateMsg) msg;
+ log.trace("[{}] onToEdgeSessionMsg [{}]", edgeEventUpdateMsg.getTenantId(), msg);
+ onEdgeEvent(tenantId, edgeEventUpdateMsg.getEdgeId());
+ break;
+ case EDGE_SYNC_REQUEST_TO_EDGE_SESSION_MSG:
+ ToEdgeSyncRequest toEdgeSyncRequest = (ToEdgeSyncRequest) msg;
+ log.trace("[{}] toEdgeSyncRequest [{}]", toEdgeSyncRequest.getTenantId(), msg);
+ startSyncProcess(tenantId, toEdgeSyncRequest.getEdgeId(), toEdgeSyncRequest.getId());
+ break;
+ case EDGE_SYNC_RESPONSE_FROM_EDGE_SESSION_MSG:
+ FromEdgeSyncResponse fromEdgeSyncResponse = (FromEdgeSyncResponse) msg;
+ log.trace("[{}] fromEdgeSyncResponse [{}]", fromEdgeSyncResponse.getTenantId(), msg);
+ processSyncResponse(fromEdgeSyncResponse);
+ break;
+ }
+ });
+ }
+
@Override
public void updateEdge(TenantId tenantId, Edge edge) {
- EdgeGrpcSession session = sessions.get(edge.getId());
- if (session != null && session.isConnected()) {
- log.debug("[{}] Updating configuration for edge [{}] [{}]", tenantId, edge.getName(), edge.getId());
- session.onConfigurationUpdate(edge);
- } else {
- log.debug("[{}] Session doesn't exist for edge [{}] [{}]", tenantId, edge.getName(), edge.getId());
- }
+ executorService.execute(() -> {
+ EdgeGrpcSession session = sessions.get(edge.getId());
+ if (session != null && session.isConnected()) {
+ log.debug("[{}] Updating configuration for edge [{}] [{}]", tenantId, edge.getName(), edge.getId());
+ session.onConfigurationUpdate(edge);
+ } else {
+ log.debug("[{}] Session doesn't exist for edge [{}] [{}]", tenantId, edge.getName(), edge.getId());
+ }
+ });
}
@Override
public void deleteEdge(TenantId tenantId, EdgeId edgeId) {
+ executorService.execute(() -> {
+ EdgeGrpcSession session = sessions.get(edgeId);
+ if (session != null && session.isConnected()) {
+ log.info("[{}] Closing and removing session for edge [{}]", tenantId, edgeId);
+ session.close();
+ sessions.remove(edgeId);
+ final Lock newEventLock = sessionNewEventsLocks.computeIfAbsent(edgeId, id -> new ReentrantLock());
+ newEventLock.lock();
+ try {
+ sessionNewEvents.remove(edgeId);
+ } finally {
+ newEventLock.unlock();
+ }
+ cancelScheduleEdgeEventsCheck(edgeId);
+ }
+ });
+ }
+
+ private void onEdgeEvent(TenantId tenantId, EdgeId edgeId) {
EdgeGrpcSession session = sessions.get(edgeId);
if (session != null && session.isConnected()) {
- log.info("[{}] Closing and removing session for edge [{}]", tenantId, edgeId);
- session.close();
- sessions.remove(edgeId);
+ log.trace("[{}] onEdgeEvent [{}]", tenantId, edgeId.getId());
final Lock newEventLock = sessionNewEventsLocks.computeIfAbsent(edgeId, id -> new ReentrantLock());
newEventLock.lock();
try {
- sessionNewEvents.remove(edgeId);
+ if (Boolean.FALSE.equals(sessionNewEvents.get(edgeId))) {
+ log.trace("[{}] set session new events flag to true [{}]", tenantId, edgeId.getId());
+ sessionNewEvents.put(edgeId, true);
+ }
} finally {
newEventLock.unlock();
}
- cancelScheduleEdgeEventsCheck(edgeId);
- }
- }
-
- @Override
- public void onEdgeEvent(TenantId tenantId, EdgeId edgeId) {
- log.trace("[{}] onEdgeEvent [{}]", tenantId, edgeId.getId());
- final Lock newEventLock = sessionNewEventsLocks.computeIfAbsent(edgeId, id -> new ReentrantLock());
- newEventLock.lock();
- try {
- if (Boolean.FALSE.equals(sessionNewEvents.get(edgeId))) {
- log.trace("[{}] set session new events flag to true [{}]", tenantId, edgeId.getId());
- sessionNewEvents.put(edgeId, true);
- }
- } finally {
- newEventLock.unlock();
}
}
@@ -220,14 +266,47 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i
scheduleEdgeEventsCheck(edgeGrpcSession);
}
- @Override
- public void startSyncProcess(TenantId tenantId, EdgeId edgeId) {
+ private void startSyncProcess(TenantId tenantId, EdgeId edgeId, UUID requestId) {
EdgeGrpcSession session = sessions.get(edgeId);
- if (session != null && session.isConnected()) {
- session.startSyncProcess(tenantId, edgeId);
+ if (session != null) {
+ boolean success = false;
+ if (session.isConnected()) {
+ session.startSyncProcess(tenantId, edgeId);
+ success = true;
+ }
+ clusterService.pushEdgeSyncResponseToCore(new FromEdgeSyncResponse(requestId, tenantId, edgeId, success));
+ }
+ }
+
+ @Override
+ public void processSyncRequest(ToEdgeSyncRequest request, Consumer responseConsumer) {
+ log.trace("[{}][{}] Processing sync edge request [{}]", request.getTenantId(), request.getId(), request.getEdgeId());
+ UUID requestId = request.getId();
+ localSyncEdgeRequests.put(requestId, responseConsumer);
+ clusterService.pushEdgeSyncRequestToCore(request);
+ scheduleSyncRequestTimeout(request, requestId);
+ }
+
+ private void scheduleSyncRequestTimeout(ToEdgeSyncRequest request, UUID requestId) {
+ log.trace("[{}] scheduling sync edge request", requestId);
+ executorService.schedule(() -> {
+ log.trace("[{}] checking if sync edge request is not processed...", requestId);
+ Consumer consumer = localSyncEdgeRequests.remove(requestId);
+ if (consumer != null) {
+ log.trace("[{}] timeout for processing sync edge request.", requestId);
+ consumer.accept(new FromEdgeSyncResponse(requestId, request.getTenantId(), request.getEdgeId(), false));
+ }
+ }, 20, TimeUnit.SECONDS);
+ }
+
+ private void processSyncResponse(FromEdgeSyncResponse response) {
+ log.trace("[{}] Received response from sync service: [{}]", response.getId(), response);
+ UUID requestId = response.getId();
+ Consumer consumer = localSyncEdgeRequests.remove(requestId);
+ if (consumer != null) {
+ consumer.accept(response);
} else {
- log.error("[{}] Edge is not connected [{}]", tenantId, edgeId);
- throw new RuntimeException("Edge is not connected");
+ log.trace("[{}] Unknown or stale sync response received [{}]", requestId, response);
}
}
diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeRpcService.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeRpcService.java
index ad65c15235..6c5515337e 100644
--- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeRpcService.java
+++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeRpcService.java
@@ -18,14 +18,19 @@ package org.thingsboard.server.service.edge.rpc;
import org.thingsboard.server.common.data.edge.Edge;
import org.thingsboard.server.common.data.id.EdgeId;
import org.thingsboard.server.common.data.id.TenantId;
+import org.thingsboard.server.common.msg.edge.EdgeSessionMsg;
+import org.thingsboard.server.common.msg.edge.FromEdgeSyncResponse;
+import org.thingsboard.server.common.msg.edge.ToEdgeSyncRequest;
+
+import java.util.function.Consumer;
public interface EdgeRpcService {
+ void onToEdgeSessionMsg(TenantId tenantId, EdgeSessionMsg msg);
+
void updateEdge(TenantId tenantId, Edge edge);
void deleteEdge(TenantId tenantId, EdgeId edgeId);
- void onEdgeEvent(TenantId tenantId, EdgeId edgeId);
-
- void startSyncProcess(TenantId tenantId, EdgeId edgeId);
+ void processSyncRequest(ToEdgeSyncRequest request, Consumer responseConsumer);
}
diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/BaseEdgeProcessor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/BaseEdgeProcessor.java
index 018acb7668..de71fa32a3 100644
--- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/BaseEdgeProcessor.java
+++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/BaseEdgeProcessor.java
@@ -76,10 +76,12 @@ import org.thingsboard.server.service.edge.rpc.constructor.RuleChainMsgConstruct
import org.thingsboard.server.service.edge.rpc.constructor.UserMsgConstructor;
import org.thingsboard.server.service.edge.rpc.constructor.WidgetTypeMsgConstructor;
import org.thingsboard.server.service.edge.rpc.constructor.WidgetsBundleMsgConstructor;
+import org.thingsboard.server.service.entitiy.TbNotificationEntityService;
import org.thingsboard.server.service.executors.DbCallbackExecutorService;
import org.thingsboard.server.service.profile.TbAssetProfileCache;
import org.thingsboard.server.service.profile.TbDeviceProfileCache;
import org.thingsboard.server.service.state.DeviceStateService;
+import org.thingsboard.server.service.telemetry.TelemetrySubscriptionService;
import java.util.ArrayList;
import java.util.List;
@@ -91,6 +93,12 @@ public abstract class BaseEdgeProcessor {
protected static final int DEFAULT_PAGE_SIZE = 1000;
+ @Autowired
+ protected TelemetrySubscriptionService tsSubService;
+
+ @Autowired
+ protected TbNotificationEntityService notificationEntityService;
+
@Autowired
protected RuleChainService ruleChainService;
diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/TelemetryEdgeProcessor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/TelemetryEdgeProcessor.java
index f3bb328886..171d5518fc 100644
--- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/TelemetryEdgeProcessor.java
+++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/TelemetryEdgeProcessor.java
@@ -35,6 +35,7 @@ import org.thingsboard.server.common.data.EdgeUtils;
import org.thingsboard.server.common.data.EntityType;
import org.thingsboard.server.common.data.EntityView;
import org.thingsboard.server.common.data.asset.Asset;
+import org.thingsboard.server.common.data.audit.ActionType;
import org.thingsboard.server.common.data.asset.AssetProfile;
import org.thingsboard.server.common.data.edge.EdgeEvent;
import org.thingsboard.server.common.data.edge.EdgeEventActionType;
@@ -57,6 +58,7 @@ import org.thingsboard.server.common.msg.queue.TopicPartitionInfo;
import org.thingsboard.server.common.msg.session.SessionMsgType;
import org.thingsboard.server.common.transport.adaptor.JsonConverter;
import org.thingsboard.server.common.transport.util.JsonUtils;
+import org.thingsboard.server.controller.BaseController;
import org.thingsboard.server.gen.edge.v1.AttributeDeleteMsg;
import org.thingsboard.server.gen.edge.v1.DownlinkMsg;
import org.thingsboard.server.gen.edge.v1.EntityDataProto;
@@ -101,7 +103,7 @@ public class TelemetryEdgeProcessor extends BaseEdgeProcessor {
}
if (entityData.hasAttributesUpdatedMsg()) {
metaData.putValue("scope", entityData.getPostAttributeScope());
- result.add(processAttributesUpdate(tenantId, customerId, entityId, entityData.getAttributesUpdatedMsg(), metaData));
+ result.add(processAttributesUpdate(tenantId, entityId, entityData.getAttributesUpdatedMsg(), metaData));
}
if (entityData.hasPostTelemetryMsg()) {
result.add(processPostTelemetry(tenantId, customerId, entityId, entityData.getPostTelemetryMsg(), metaData));
@@ -224,39 +226,36 @@ public class TelemetryEdgeProcessor extends BaseEdgeProcessor {
return futureToSet;
}
- private ListenableFuture processAttributesUpdate(TenantId tenantId, CustomerId customerId, EntityId entityId, TransportProtos.PostAttributeMsg msg, TbMsgMetaData metaData) {
+ private ListenableFuture processAttributesUpdate(TenantId tenantId,
+ EntityId entityId,
+ TransportProtos.PostAttributeMsg msg,
+ TbMsgMetaData metaData) {
SettableFuture futureToSet = SettableFuture.create();
JsonObject json = JsonUtils.getJsonObject(msg.getKvList());
- Set attributes = JsonConverter.convertToAttributes(json);
- ListenableFuture> future = attributesService.save(tenantId, entityId, metaData.getValue("scope"), new ArrayList<>(attributes));
- Futures.addCallback(future, new FutureCallback<>() {
+ List attributes = new ArrayList<>(JsonConverter.convertToAttributes(json));
+ String scope = metaData.getValue("scope");
+ tsSubService.saveAndNotify(tenantId, entityId, scope, attributes, new FutureCallback() {
@Override
- public void onSuccess(@Nullable List keys) {
- var defaultQueueAndRuleChain = getDefaultQueueNameAndRuleChainId(tenantId, entityId);
- TbMsg tbMsg = TbMsg.newMsg(defaultQueueAndRuleChain.getKey(), DataConstants.ATTRIBUTES_UPDATED, entityId, customerId, metaData, gson.toJson(json), defaultQueueAndRuleChain.getValue(), null);
- tbClusterService.pushMsgToRuleEngine(tenantId, tbMsg.getOriginator(), tbMsg, new TbQueueCallback() {
- @Override
- public void onSuccess(TbQueueMsgMetadata metadata) {
- futureToSet.set(null);
- }
-
- @Override
- public void onFailure(Throwable t) {
- log.error("Can't process attributes update [{}]", msg, t);
- futureToSet.setException(t);
- }
- });
+ public void onSuccess(@Nullable Void tmp) {
+ logAttributesUpdated(tenantId, entityId, scope, attributes, null);
+ futureToSet.set(null);
}
@Override
public void onFailure(Throwable t) {
log.error("Can't process attributes update [{}]", msg, t);
+ logAttributesUpdated(tenantId, entityId, scope, attributes, t);
futureToSet.setException(t);
}
- }, dbCallbackExecutorService);
+ });
return futureToSet;
}
+ private void logAttributesUpdated(TenantId tenantId, EntityId entityId, String scope, List attributes, Throwable e) {
+ notificationEntityService.logEntityAction(tenantId, entityId, ActionType.ATTRIBUTES_UPDATED, null,
+ BaseController.toException(e), scope, attributes);
+ }
+
private ListenableFuture processAttributeDeleteMsg(TenantId tenantId, EntityId entityId, AttributeDeleteMsg attributeDeleteMsg, String entityType) {
SettableFuture futureToSet = SettableFuture.create();
String scope = attributeDeleteMsg.getScope();
diff --git a/application/src/main/java/org/thingsboard/server/service/install/SqlDatabaseUpgradeService.java b/application/src/main/java/org/thingsboard/server/service/install/SqlDatabaseUpgradeService.java
index c5c81bfc91..092726d0c8 100644
--- a/application/src/main/java/org/thingsboard/server/service/install/SqlDatabaseUpgradeService.java
+++ b/application/src/main/java/org/thingsboard/server/service/install/SqlDatabaseUpgradeService.java
@@ -15,8 +15,6 @@
*/
package org.thingsboard.server.service.install;
-import com.fasterxml.jackson.databind.JsonNode;
-import com.fasterxml.jackson.databind.node.ObjectNode;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.springframework.beans.factory.annotation.Autowired;
@@ -25,10 +23,7 @@ import org.springframework.context.annotation.Lazy;
import org.springframework.context.annotation.Profile;
import org.springframework.stereotype.Service;
import org.thingsboard.server.common.data.EntitySubtype;
-import org.thingsboard.server.common.data.StringUtils;
import org.thingsboard.server.common.data.Tenant;
-import org.thingsboard.server.common.data.TenantProfile;
-import org.thingsboard.server.common.data.id.QueueId;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.page.PageData;
import org.thingsboard.server.common.data.page.PageLink;
@@ -37,8 +32,6 @@ import org.thingsboard.server.common.data.queue.ProcessingStrategyType;
import org.thingsboard.server.common.data.queue.Queue;
import org.thingsboard.server.common.data.queue.SubmitStrategy;
import org.thingsboard.server.common.data.queue.SubmitStrategyType;
-import org.thingsboard.server.common.data.rule.RuleNode;
-import org.thingsboard.server.common.data.tenant.profile.TenantProfileQueueConfiguration;
import org.thingsboard.server.dao.asset.AssetProfileService;
import org.thingsboard.server.dao.asset.AssetService;
import org.thingsboard.server.dao.dashboard.DashboardService;
@@ -63,11 +56,8 @@ import java.sql.SQLException;
import java.sql.SQLSyntaxErrorException;
import java.sql.SQLWarning;
import java.sql.Statement;
-import java.util.Collections;
import java.util.List;
-import java.util.Map;
import java.util.concurrent.TimeUnit;
-import java.util.stream.Collectors;
import static org.thingsboard.server.service.install.DatabaseHelper.ADDITIONAL_INFO;
import static org.thingsboard.server.service.install.DatabaseHelper.ASSIGNED_CUSTOMERS;
@@ -620,8 +610,8 @@ public class SqlDatabaseUpgradeService implements DatabaseEntitiesUpgradeService
case "3.4.1":
try (Connection conn = DriverManager.getConnection(dbUrl, dbUserName, dbPassword)) {
log.info("Updating schema ...");
+ runSchemaUpdateScript(conn, "3.4.1");
if (isOldSchema(conn, 3004001)) {
-
try {
conn.createStatement().execute("ALTER TABLE asset ADD COLUMN asset_profile_id uuid");
} catch (Exception e) {
@@ -669,6 +659,11 @@ public class SqlDatabaseUpgradeService implements DatabaseEntitiesUpgradeService
}
}
+ private void runSchemaUpdateScript(Connection connection, String version) throws Exception {
+ Path schemaUpdateFile = Paths.get(installScripts.getDataDir(), "upgrade", version, SCHEMA_UPDATE_SQL);
+ loadSql(schemaUpdateFile, connection);
+ }
+
private void loadSql(Path sqlFile, Connection conn) throws Exception {
String sql = new String(Files.readAllBytes(sqlFile), Charset.forName("UTF-8"));
Statement st = conn.createStatement();
diff --git a/application/src/main/java/org/thingsboard/server/service/install/update/DefaultDataUpdateService.java b/application/src/main/java/org/thingsboard/server/service/install/update/DefaultDataUpdateService.java
index d02f50b37b..a8cf5374a2 100644
--- a/application/src/main/java/org/thingsboard/server/service/install/update/DefaultDataUpdateService.java
+++ b/application/src/main/java/org/thingsboard/server/service/install/update/DefaultDataUpdateService.java
@@ -63,6 +63,7 @@ import org.thingsboard.server.common.data.rule.RuleNode;
import org.thingsboard.server.common.data.tenant.profile.TenantProfileQueueConfiguration;
import org.thingsboard.server.dao.DaoUtil;
import org.thingsboard.server.dao.alarm.AlarmDao;
+import org.thingsboard.server.dao.audit.AuditLogDao;
import org.thingsboard.server.dao.entity.EntityService;
import org.thingsboard.server.dao.entityview.EntityViewService;
import org.thingsboard.server.dao.event.EventService;
@@ -138,6 +139,9 @@ public class DefaultDataUpdateService implements DataUpdateService {
@Autowired
private EventService eventService;
+ @Autowired
+ private AuditLogDao auditLogDao;
+
@Override
public void updateData(String fromVersion) throws Exception {
switch (fromVersion) {
@@ -170,12 +174,22 @@ public class DefaultDataUpdateService implements DataUpdateService {
rateLimitsUpdater.updateEntities();
break;
case "3.4.0":
- String skipEventsMigration = System.getenv("TB_SKIP_EVENTS_MIGRATION");
- if (skipEventsMigration == null || skipEventsMigration.equalsIgnoreCase("false")) {
+ boolean skipEventsMigration = getEnv("TB_SKIP_EVENTS_MIGRATION", false);
+ if (!skipEventsMigration) {
log.info("Updating data from version 3.4.0 to 3.4.1 ...");
eventService.migrateEvents();
}
break;
+ case "3.4.1":
+ boolean skipAuditLogsMigration = getEnv("TB_SKIP_AUDIT_LOGS_MIGRATION", false);
+ if (!skipAuditLogsMigration) {
+ log.info("Updating data from version 3.4.1 to 3.4.2 ...");
+ log.info("Starting audit logs migration. Can be skipped with TB_SKIP_AUDIT_LOGS_MIGRATION env variable set to true");
+ auditLogDao.migrateAuditLogs();
+ } else {
+ log.info("Skipping audit logs migration");
+ }
+ break;
default:
throw new RuntimeException("Unable to update data, unsupported fromVersion: " + fromVersion);
}
@@ -645,4 +659,13 @@ public class DefaultDataUpdateService implements DataUpdateService {
return mainQueueConfiguration;
}
+ private boolean getEnv(String name, boolean defaultValue) {
+ String env = System.getenv(name);
+ if (env == null) {
+ return defaultValue;
+ } else {
+ return Boolean.parseBoolean(env);
+ }
+ }
+
}
diff --git a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbClusterService.java b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbClusterService.java
index ca654ec802..7983a0adcd 100644
--- a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbClusterService.java
+++ b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbClusterService.java
@@ -51,6 +51,8 @@ import org.thingsboard.server.common.data.queue.Queue;
import org.thingsboard.server.common.msg.TbMsg;
import org.thingsboard.server.common.msg.ToDeviceActorNotificationMsg;
import org.thingsboard.server.common.msg.edge.EdgeEventUpdateMsg;
+import org.thingsboard.server.common.msg.edge.FromEdgeSyncResponse;
+import org.thingsboard.server.common.msg.edge.ToEdgeSyncRequest;
import org.thingsboard.server.common.msg.plugin.ComponentLifecycleMsg;
import org.thingsboard.server.common.msg.queue.ServiceType;
import org.thingsboard.server.common.msg.queue.TopicPartitionInfo;
@@ -372,12 +374,32 @@ public class DefaultTbClusterService implements TbClusterService {
log.trace("[{}] Processing edge {} event update ", tenantId, edgeId);
EdgeEventUpdateMsg msg = new EdgeEventUpdateMsg(tenantId, edgeId);
byte[] msgBytes = encodingService.encode(msg);
+ ToCoreNotificationMsg toCoreMsg = ToCoreNotificationMsg.newBuilder().setEdgeEventUpdateMsg(ByteString.copyFrom(msgBytes)).build();
+ pushEdgeSyncMsgToCore(edgeId, toCoreMsg);
+ }
+
+ @Override
+ public void pushEdgeSyncRequestToCore(ToEdgeSyncRequest toEdgeSyncRequest) {
+ log.trace("[{}] Processing edge sync request {} ", toEdgeSyncRequest.getTenantId(), toEdgeSyncRequest);
+ byte[] msgBytes = encodingService.encode(toEdgeSyncRequest);
+ ToCoreNotificationMsg toCoreMsg = ToCoreNotificationMsg.newBuilder().setToEdgeSyncRequestMsg(ByteString.copyFrom(msgBytes)).build();
+ pushEdgeSyncMsgToCore(toEdgeSyncRequest.getEdgeId(), toCoreMsg);
+ }
+
+ @Override
+ public void pushEdgeSyncResponseToCore(FromEdgeSyncResponse fromEdgeSyncResponse) {
+ log.trace("[{}] Processing edge sync response {}", fromEdgeSyncResponse.getTenantId(), fromEdgeSyncResponse);
+ byte[] msgBytes = encodingService.encode(fromEdgeSyncResponse);
+ ToCoreNotificationMsg toCoreMsg = ToCoreNotificationMsg.newBuilder().setFromEdgeSyncResponseMsg(ByteString.copyFrom(msgBytes)).build();
+ pushEdgeSyncMsgToCore(fromEdgeSyncResponse.getEdgeId(), toCoreMsg);
+ }
+
+ private void pushEdgeSyncMsgToCore(EdgeId edgeId, ToCoreNotificationMsg toCoreMsg) {
TbQueueProducer> toCoreNfProducer = producerProvider.getTbCoreNotificationsMsgProducer();
Set tbCoreServices = partitionService.getAllServiceIds(ServiceType.TB_CORE);
for (String serviceId : tbCoreServices) {
TopicPartitionInfo tpi = notificationsTopicService.getNotificationsTopic(ServiceType.TB_CORE, serviceId);
- ToCoreNotificationMsg toCoreMsg = ToCoreNotificationMsg.newBuilder().setEdgeEventUpdateMsg(ByteString.copyFrom(msgBytes)).build();
- toCoreNfProducer.send(tpi, new TbProtoQueueMsg<>(msg.getEdgeId().getId(), toCoreMsg), null);
+ toCoreNfProducer.send(tpi, new TbProtoQueueMsg<>(edgeId.getId(), toCoreMsg), null);
toCoreNfs.incrementAndGet();
}
}
diff --git a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java
index 9cdf3e4aa2..ec58ac1aa9 100644
--- a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java
+++ b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java
@@ -20,8 +20,6 @@ import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.context.event.ApplicationReadyEvent;
-import org.springframework.context.event.EventListener;
-import org.springframework.core.annotation.Order;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import org.thingsboard.common.util.JacksonUtil;
@@ -77,7 +75,6 @@ import org.thingsboard.server.service.state.DeviceStateService;
import org.thingsboard.server.service.subscription.SubscriptionManagerService;
import org.thingsboard.server.service.subscription.TbLocalSubscriptionService;
import org.thingsboard.server.service.subscription.TbSubscriptionUtils;
-import org.thingsboard.server.service.sync.vc.EntitiesVersionControlService;
import org.thingsboard.server.service.sync.vc.GitVersionControlQueueService;
import org.thingsboard.server.service.transport.msg.TransportToDeviceActorMsgWrapper;
@@ -319,13 +316,12 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService actorMsg = encodingService.decode(toCoreNotification.getEdgeEventUpdateMsg().toByteArray());
- if (actorMsg.isPresent()) {
- log.trace("[{}] Forwarding message to App Actor {}", id, actorMsg.get());
- actorContext.tellWithHighPriority(actorMsg.get());
- }
- callback.onSuccess();
+ } else if (!toCoreNotification.getEdgeEventUpdateMsg().isEmpty()) {
+ forwardToAppActor(id, encodingService.decode(toCoreNotification.getEdgeEventUpdateMsg().toByteArray()), callback);
+ } else if (!toCoreNotification.getToEdgeSyncRequestMsg().isEmpty()) {
+ forwardToAppActor(id, encodingService.decode(toCoreNotification.getToEdgeSyncRequestMsg().toByteArray()), callback);
+ } else if (!toCoreNotification.getFromEdgeSyncResponseMsg().isEmpty()) {
+ forwardToAppActor(id, encodingService.decode(toCoreNotification.getFromEdgeSyncResponseMsg().toByteArray()), callback);
} else if (toCoreNotification.hasQueueUpdateMsg()) {
TransportProtos.QueueUpdateMsg queue = toCoreNotification.getQueueUpdateMsg();
partitionService.updateQueue(queue);
@@ -554,6 +550,14 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService actorMsg, TbCallback callback) {
+ if (actorMsg.isPresent()) {
+ log.trace("[{}] Forwarding message to App Actor {}", id, actorMsg.get());
+ actorContext.tell(actorMsg.get());
+ }
+ callback.onSuccess();
+ }
+
private void throwNotHandled(Object msg, TbCallback callback) {
log.warn("Message not handled: {}", msg);
callback.onFailure(new RuntimeException("Message not handled!"));
diff --git a/application/src/main/java/org/thingsboard/server/service/script/AbstractJsInvokeService.java b/application/src/main/java/org/thingsboard/server/service/script/AbstractJsInvokeService.java
index eef49ae518..cc46f0a74f 100644
--- a/application/src/main/java/org/thingsboard/server/service/script/AbstractJsInvokeService.java
+++ b/application/src/main/java/org/thingsboard/server/service/script/AbstractJsInvokeService.java
@@ -15,9 +15,14 @@
*/
package org.thingsboard.server.service.script;
+import com.google.common.hash.Hashing;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.MoreExecutors;
+import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.data.util.Pair;
import org.thingsboard.common.util.ThingsBoardThreadFactory;
import org.thingsboard.server.common.data.ApiUsageRecordKey;
import org.thingsboard.server.common.data.id.CustomerId;
@@ -30,20 +35,33 @@ import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;
+import static java.lang.String.format;
+
/**
* Created by ashvayka on 26.09.18.
*/
@Slf4j
+@SuppressWarnings("UnstableApiUsage")
public abstract class AbstractJsInvokeService implements JsInvokeService {
private final TbApiUsageStateService apiUsageStateService;
private final TbApiUsageClient apiUsageClient;
protected ScheduledExecutorService timeoutExecutorService;
- protected Map scriptIdToNameMap = new ConcurrentHashMap<>();
- protected Map disabledFunctions = new ConcurrentHashMap<>();
+
+ protected final Map> scriptIdToNameAndHashMap = new ConcurrentHashMap<>();
+ protected final Map disabledFunctions = new ConcurrentHashMap<>();
+
+ @Getter
+ @Value("${js.max_total_args_size:100000}")
+ private long maxTotalArgsSize;
+ @Getter
+ @Value("${js.max_result_size:300000}")
+ private long maxResultSize;
+ @Getter
+ @Value("${js.max_script_body_size:50000}")
+ private long maxScriptBodySize;
protected AbstractJsInvokeService(TbApiUsageStateService apiUsageStateService, TbApiUsageClient apiUsageClient) {
this.apiUsageStateService = apiUsageStateService;
@@ -65,44 +83,63 @@ public abstract class AbstractJsInvokeService implements JsInvokeService {
@Override
public ListenableFuture eval(TenantId tenantId, JsScriptType scriptType, String scriptBody, String... argNames) {
if (apiUsageStateService.getApiUsageState(tenantId).isJsExecEnabled()) {
+ if (scriptBodySizeExceeded(scriptBody)) {
+ return error(format("Script body exceeds maximum allowed size of %s symbols", getMaxScriptBodySize()));
+ }
UUID scriptId = UUID.randomUUID();
- String functionName = "invokeInternal_" + scriptId.toString().replace('-', '_');
+ String scriptHash = hash(tenantId, scriptBody);
+ String functionName = constructFunctionName(scriptId, scriptHash);
String jsScript = generateJsScript(scriptType, functionName, scriptBody, argNames);
- return doEval(scriptId, functionName, jsScript);
+ return doEval(scriptId, scriptHash, functionName, jsScript);
} else {
- return Futures.immediateFailedFuture(new RuntimeException("JS Execution is disabled due to API limits!"));
+ return error("JS Execution is disabled due to API limits!");
}
}
+ protected String constructFunctionName(UUID scriptId, String scriptHash) {
+ return "invokeInternal_" + scriptId.toString().replace('-', '_');
+ }
+
@Override
- public ListenableFuture