diff --git a/.github/release.yml b/.github/release.yml new file mode 100644 index 0000000000..01ba7d863e --- /dev/null +++ b/.github/release.yml @@ -0,0 +1,88 @@ +# +# 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. +# + +changelog: + exclude: + labels: + - Ignore for release + categories: + - title: 'Major Core & Rule Engine' + labels: + - 'Major Core' + - 'Major Rule Engine' + exclude: + labels: + - 'Bug' + - title: 'Major UI' + labels: + - 'Major UI' + exclude: + labels: + - 'Bug' + - title: 'Major Transport' + labels: + - 'Major Transport' + exclude: + labels: + - 'Bug' + - title: 'Major Edge' + labels: + - 'Major Edge' + exclude: + labels: + - 'Bug' + - title: 'Core & Rule Engine' + labels: + - 'Core' + - 'Rule Engine' + exclude: + labels: + - 'Bug' + - title: 'UI' + labels: + - 'UI' + exclude: + labels: + - 'Bug' + - title: 'Transport' + labels: + - 'Transport' + exclude: + labels: + - 'Bug' + - title: 'Edge' + labels: + - 'Edge' + exclude: + labels: + - 'Bug' + - title: 'Bug: Core & Rule Engine' + labels: + - 'Core' + - 'Rule Engine' + - 'Bug' + - title: 'Bug: UI' + labels: + - 'UI' + - 'Bug' + - title: 'Bug: Transport' + labels: + - 'Transport' + - 'Bug' + - title: 'Bug: Edge' + labels: + - 'Edge' + - 'Bug' 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 index a3d7bfb78f..45a75c6bd9 100644 --- a/application/src/main/data/upgrade/3.4.1/schema_update.sql +++ b/application/src/main/data/upgrade/3.4.1/schema_update.sql @@ -22,6 +22,8 @@ $$ -- 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; + CREATE INDEX IF NOT EXISTS idx_old_audit_log_created_time ON old_audit_log(created_time); + 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 @@ -86,6 +88,7 @@ $$ -- in case of running the upgrade script a second time: IF NOT (SELECT exists(SELECT FROM pg_tables WHERE tablename = 'old_edge_event')) THEN ALTER TABLE edge_event RENAME TO old_edge_event; + CREATE INDEX IF NOT EXISTS idx_old_blob_entity_created_time_tmp ON old_blob_entity(created_time); ALTER INDEX IF EXISTS idx_edge_event_tenant_id_and_created_time RENAME TO idx_old_edge_event_tenant_id_and_created_time; FOR table_partition IN SELECT tablename AS name, split_part(tablename, '_', 3) AS partition_ts diff --git a/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java b/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java index e1a41fe401..8b19a6954c 100644 --- a/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java +++ b/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java @@ -50,6 +50,7 @@ import org.thingsboard.server.common.msg.queue.ServiceType; import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; import org.thingsboard.server.common.msg.tools.TbRateLimits; import org.thingsboard.server.common.stats.TbApiUsageReportClient; +import org.thingsboard.server.dao.asset.AssetProfileService; import org.thingsboard.server.dao.asset.AssetService; import org.thingsboard.server.dao.attributes.AttributesService; import org.thingsboard.server.dao.audit.AuditLogService; @@ -58,6 +59,7 @@ import org.thingsboard.server.dao.customer.CustomerService; import org.thingsboard.server.dao.dashboard.DashboardService; import org.thingsboard.server.dao.device.ClaimDevicesService; import org.thingsboard.server.dao.device.DeviceCredentialsService; +import org.thingsboard.server.dao.device.DeviceProfileService; import org.thingsboard.server.dao.device.DeviceService; import org.thingsboard.server.dao.edge.EdgeEventService; import org.thingsboard.server.dao.edge.EdgeService; @@ -176,6 +178,14 @@ public class ActorSystemContext { @Getter private DeviceService deviceService; + @Autowired + @Getter + private DeviceProfileService deviceProfileService; + + @Autowired + @Getter + private AssetProfileService assetProfileService; + @Autowired @Getter private DeviceCredentialsService deviceCredentialsService; 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 ef2de35912..ba1c4e9796 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 @@ -71,12 +71,14 @@ import org.thingsboard.server.common.msg.TbMsgMetaData; import org.thingsboard.server.common.msg.TbMsgProcessingStackItem; import org.thingsboard.server.common.msg.queue.ServiceType; import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; +import org.thingsboard.server.dao.asset.AssetProfileService; import org.thingsboard.server.dao.asset.AssetService; import org.thingsboard.server.dao.attributes.AttributesService; import org.thingsboard.server.dao.cassandra.CassandraCluster; import org.thingsboard.server.dao.customer.CustomerService; import org.thingsboard.server.dao.dashboard.DashboardService; import org.thingsboard.server.dao.device.DeviceCredentialsService; +import org.thingsboard.server.dao.device.DeviceProfileService; import org.thingsboard.server.dao.device.DeviceService; import org.thingsboard.server.dao.edge.EdgeEventService; import org.thingsboard.server.dao.edge.EdgeService; @@ -559,6 +561,16 @@ class DefaultTbContext implements TbContext { return mainCtx.getDeviceService(); } + @Override + public DeviceProfileService getDeviceProfileService() { + return mainCtx.getDeviceProfileService(); + } + + @Override + public AssetProfileService getAssetProfileService() { + return mainCtx.getAssetProfileService(); + } + @Override public DeviceCredentialsService getDeviceCredentialsService() { return mainCtx.getDeviceCredentialsService(); 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 ffc3fcf3cd..43d049926b 100644 --- a/application/src/main/java/org/thingsboard/server/controller/BaseController.java +++ b/application/src/main/java/org/thingsboard/server/controller/BaseController.java @@ -120,6 +120,7 @@ import org.thingsboard.server.dao.queue.QueueService; import org.thingsboard.server.dao.relation.RelationService; import org.thingsboard.server.dao.rpc.RpcService; import org.thingsboard.server.dao.rule.RuleChainService; +import org.thingsboard.server.dao.service.Validator; import org.thingsboard.server.dao.tenant.TbTenantProfileCache; import org.thingsboard.server.dao.tenant.TenantProfileService; import org.thingsboard.server.dao.tenant.TenantService; @@ -131,7 +132,6 @@ import org.thingsboard.server.queue.discovery.PartitionService; import org.thingsboard.server.queue.provider.TbQueueProducerProvider; import org.thingsboard.server.queue.util.TbCoreComponent; import org.thingsboard.server.service.component.ComponentDiscoveryService; -import org.thingsboard.server.service.edge.EdgeNotificationService; import org.thingsboard.server.service.edge.rpc.EdgeRpcService; import org.thingsboard.server.service.entitiy.TbNotificationEntityService; import org.thingsboard.server.service.ota.OtaPackageStateService; @@ -417,9 +417,12 @@ public abstract class BaseController { } PageLink createPageLink(int pageSize, int page, String textSearch, String sortProperty, String sortOrder) throws ThingsboardException { - if (!StringUtils.isEmpty(sortProperty)) { + if (StringUtils.isNotEmpty(sortProperty)) { + if (!Validator.isValidProperty(sortProperty)) { + throw new IllegalArgumentException("Invalid sort property"); + } SortOrder.Direction direction = SortOrder.Direction.ASC; - if (!StringUtils.isEmpty(sortOrder)) { + if (StringUtils.isNotEmpty(sortOrder)) { try { direction = SortOrder.Direction.valueOf(sortOrder.toUpperCase()); } catch (IllegalArgumentException e) { diff --git a/application/src/main/java/org/thingsboard/server/controller/plugin/TbWebSocketHandler.java b/application/src/main/java/org/thingsboard/server/controller/plugin/TbWebSocketHandler.java index 432b35fd80..3a429add9b 100644 --- a/application/src/main/java/org/thingsboard/server/controller/plugin/TbWebSocketHandler.java +++ b/application/src/main/java/org/thingsboard/server/controller/plugin/TbWebSocketHandler.java @@ -28,10 +28,12 @@ import org.springframework.web.socket.WebSocketSession; import org.springframework.web.socket.adapter.NativeWebSocketSession; import org.springframework.web.socket.handler.TextWebSocketHandler; import org.thingsboard.server.common.data.StringUtils; +import org.thingsboard.server.common.data.TenantProfile; import org.thingsboard.server.common.data.exception.ThingsboardErrorCode; import org.thingsboard.server.common.data.id.CustomerId; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.id.UserId; +import org.thingsboard.server.common.data.tenant.profile.DefaultTenantProfileConfiguration; import org.thingsboard.server.common.msg.tools.TbRateLimits; import org.thingsboard.server.config.WebSocketConfiguration; import org.thingsboard.server.dao.tenant.TbTenantProfileCache; @@ -50,6 +52,7 @@ import javax.websocket.Session; import java.io.IOException; import java.net.URI; import java.security.InvalidParameterException; +import java.util.Optional; import java.util.Queue; import java.util.Set; import java.util.UUID; @@ -136,8 +139,9 @@ public class TbWebSocketHandler extends TextWebSocketHandler implements Telemetr if (!checkLimits(session, sessionRef)) { return; } - var tenantProfileConfiguration = tenantProfileCache.get(sessionRef.getSecurityCtx().getTenantId()).getDefaultProfileConfiguration(); - internalSessionMap.put(internalSessionId, new SessionMetaData(session, sessionRef, tenantProfileConfiguration.getWsMsgQueueLimitPerSession() > 0 ? + var tenantProfileConfiguration = getTenantProfileConfiguration(sessionRef); + internalSessionMap.put(internalSessionId, new SessionMetaData(session, sessionRef, + tenantProfileConfiguration != null && tenantProfileConfiguration.getWsMsgQueueLimitPerSession() > 0 ? tenantProfileConfiguration.getWsMsgQueueLimitPerSession() : 500)); externalSessionMap.put(externalSessionId, internalSessionId); @@ -316,22 +320,24 @@ public class TbWebSocketHandler extends TextWebSocketHandler implements Telemetr if (internalId != null) { SessionMetaData sessionMd = internalSessionMap.get(internalId); if (sessionMd != null) { - var tenantProfileConfiguration = tenantProfileCache.get(sessionRef.getSecurityCtx().getTenantId()).getDefaultProfileConfiguration(); - if (StringUtils.isNotEmpty(tenantProfileConfiguration.getWsUpdatesPerSessionRateLimit())) { - TbRateLimits rateLimits = perSessionUpdateLimits.computeIfAbsent(sessionRef.getSessionId(), sid -> new TbRateLimits(tenantProfileConfiguration.getWsUpdatesPerSessionRateLimit())); - if (!rateLimits.tryConsume()) { - if (blacklistedSessions.putIfAbsent(externalId, sessionRef) == null) { - log.info("[{}][{}][{}] Failed to process session update. Max session updates limit reached" - , sessionRef.getSecurityCtx().getTenantId(), sessionRef.getSecurityCtx().getId(), externalId); - sessionMd.sendMsg("{\"subscriptionId\":" + subscriptionId + ", \"errorCode\":" + ThingsboardErrorCode.TOO_MANY_UPDATES.getErrorCode() + ", \"errorMsg\":\"Too many updates!\"}"); + var tenantProfileConfiguration = getTenantProfileConfiguration(sessionRef); + if (tenantProfileConfiguration != null) { + if (StringUtils.isNotEmpty(tenantProfileConfiguration.getWsUpdatesPerSessionRateLimit())) { + TbRateLimits rateLimits = perSessionUpdateLimits.computeIfAbsent(sessionRef.getSessionId(), sid -> new TbRateLimits(tenantProfileConfiguration.getWsUpdatesPerSessionRateLimit())); + if (!rateLimits.tryConsume()) { + if (blacklistedSessions.putIfAbsent(externalId, sessionRef) == null) { + log.info("[{}][{}][{}] Failed to process session update. Max session updates limit reached" + , sessionRef.getSecurityCtx().getTenantId(), sessionRef.getSecurityCtx().getId(), externalId); + sessionMd.sendMsg("{\"subscriptionId\":" + subscriptionId + ", \"errorCode\":" + ThingsboardErrorCode.TOO_MANY_UPDATES.getErrorCode() + ", \"errorMsg\":\"Too many updates!\"}"); + } + return; + } else { + log.debug("[{}][{}][{}] Session is no longer blacklisted.", sessionRef.getSecurityCtx().getTenantId(), sessionRef.getSecurityCtx().getId(), externalId); + blacklistedSessions.remove(externalId); } - return; } else { - log.debug("[{}][{}][{}] Session is no longer blacklisted.", sessionRef.getSecurityCtx().getTenantId(), sessionRef.getSecurityCtx().getId(), externalId); - blacklistedSessions.remove(externalId); + perSessionUpdateLimits.remove(sessionRef.getSessionId()); } - } else { - perSessionUpdateLimits.remove(sessionRef.getSessionId()); } sessionMd.sendMsg(msg); } else { @@ -376,8 +382,7 @@ public class TbWebSocketHandler extends TextWebSocketHandler implements Telemetr } private boolean checkLimits(WebSocketSession session, TelemetryWebSocketSessionRef sessionRef) throws Exception { - var tenantProfileConfiguration = - tenantProfileCache.get(sessionRef.getSecurityCtx().getTenantId()).getDefaultProfileConfiguration(); + var tenantProfileConfiguration = getTenantProfileConfiguration(sessionRef); if (tenantProfileConfiguration == null) { return true; } @@ -444,7 +449,8 @@ public class TbWebSocketHandler extends TextWebSocketHandler implements Telemetr } private void cleanupLimits(WebSocketSession session, TelemetryWebSocketSessionRef sessionRef) { - var tenantProfileConfiguration = tenantProfileCache.get(sessionRef.getSecurityCtx().getTenantId()).getDefaultProfileConfiguration(); + var tenantProfileConfiguration = getTenantProfileConfiguration(sessionRef); + if (tenantProfileConfiguration == null) return; String sessionId = session.getId(); perSessionUpdateLimits.remove(sessionRef.getSessionId()); @@ -477,4 +483,9 @@ public class TbWebSocketHandler extends TextWebSocketHandler implements Telemetr } } + private DefaultTenantProfileConfiguration getTenantProfileConfiguration(TelemetryWebSocketSessionRef sessionRef) { + return Optional.ofNullable(tenantProfileCache.get(sessionRef.getSecurityCtx().getTenantId())) + .map(TenantProfile::getDefaultProfileConfiguration).orElse(null); + } + } \ No newline at end of file diff --git a/application/src/main/java/org/thingsboard/server/service/device/DeviceBulkImportService.java b/application/src/main/java/org/thingsboard/server/service/device/DeviceBulkImportService.java index 7337497194..9f8d58b566 100644 --- a/application/src/main/java/org/thingsboard/server/service/device/DeviceBulkImportService.java +++ b/application/src/main/java/org/thingsboard/server/service/device/DeviceBulkImportService.java @@ -40,6 +40,7 @@ import org.thingsboard.server.common.data.device.profile.DisabledDeviceProfilePr import org.thingsboard.server.common.data.device.profile.Lwm2mDeviceProfileTransportConfiguration; import org.thingsboard.server.common.data.device.profile.lwm2m.OtherConfiguration; import org.thingsboard.server.common.data.device.profile.lwm2m.TelemetryMappingConfiguration; +import org.thingsboard.server.common.data.id.DeviceId; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.security.DeviceCredentials; import org.thingsboard.server.common.data.security.DeviceCredentialsType; @@ -104,7 +105,7 @@ public class DeviceBulkImportService extends AbstractBulkImportService { protected Device saveEntity(SecurityUser user, Device entity, Map fields) { DeviceCredentials deviceCredentials; try { - deviceCredentials = createDeviceCredentials(fields); + deviceCredentials = createDeviceCredentials(entity.getTenantId(), entity.getId(), fields); deviceCredentialsService.formatCredentials(deviceCredentials); } catch (Exception e) { throw new DeviceCredentialsValidationException("Invalid device credentials: " + e.getMessage()); @@ -136,7 +137,7 @@ public class DeviceBulkImportService extends AbstractBulkImportService { } @SneakyThrows - private DeviceCredentials createDeviceCredentials(Map fields) { + private DeviceCredentials createDeviceCredentials(TenantId tenantId, DeviceId deviceId, Map fields) { DeviceCredentials credentials = new DeviceCredentials(); if (fields.containsKey(BulkImportColumnType.LWM2M_CLIENT_ENDPOINT)) { credentials.setCredentialsType(DeviceCredentialsType.LWM2M_CREDENTIALS); @@ -147,7 +148,9 @@ public class DeviceBulkImportService extends AbstractBulkImportService { } else if (CollectionUtils.containsAny(fields.keySet(), EnumSet.of(BulkImportColumnType.MQTT_CLIENT_ID, BulkImportColumnType.MQTT_USER_NAME, BulkImportColumnType.MQTT_PASSWORD))) { credentials.setCredentialsType(DeviceCredentialsType.MQTT_BASIC); setUpBasicMqttCredentials(fields, credentials); - } else { + } else if (deviceId != null && !fields.containsKey(BulkImportColumnType.ACCESS_TOKEN)) { + credentials = deviceCredentialsService.findDeviceCredentialsByDeviceId(tenantId, deviceId); + } else { credentials.setCredentialsType(DeviceCredentialsType.ACCESS_TOKEN); setUpAccessTokenCredentials(fields, credentials); } diff --git a/application/src/main/java/org/thingsboard/server/service/edge/DefaultEdgeNotificationService.java b/application/src/main/java/org/thingsboard/server/service/edge/DefaultEdgeNotificationService.java index 1bb30d6960..6d368fbed8 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/DefaultEdgeNotificationService.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/DefaultEdgeNotificationService.java @@ -16,6 +16,7 @@ package org.thingsboard.server.service.edge; import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.node.ObjectNode; import com.google.common.util.concurrent.FutureCallback; import com.google.common.util.concurrent.Futures; import com.google.common.util.concurrent.ListenableFuture; @@ -23,6 +24,7 @@ import lombok.extern.slf4j.Slf4j; import org.checkerframework.checker.nullness.qual.Nullable; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Service; +import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.common.util.ThingsBoardThreadFactory; import org.thingsboard.server.cluster.TbClusterService; import org.thingsboard.server.common.data.EdgeUtils; @@ -67,6 +69,8 @@ import java.util.concurrent.Executors; @Slf4j public class DefaultEdgeNotificationService implements EdgeNotificationService { + public static final String EDGE_IS_ROOT_BODY_KEY = "isRoot"; + @Autowired private EdgeService edgeService; @@ -142,7 +146,9 @@ public class DefaultEdgeNotificationService implements EdgeNotificationService { public Edge setEdgeRootRuleChain(TenantId tenantId, Edge edge, RuleChainId ruleChainId) throws Exception { edge.setRootRuleChainId(ruleChainId); Edge savedEdge = edgeService.saveEdge(edge); - saveEdgeEvent(tenantId, edge.getId(), EdgeEventType.RULE_CHAIN, EdgeEventActionType.UPDATED, ruleChainId, null).get(); + ObjectNode isRootBody = JacksonUtil.OBJECT_MAPPER.createObjectNode(); + isRootBody.put(EDGE_IS_ROOT_BODY_KEY, Boolean.TRUE); + saveEdgeEvent(tenantId, edge.getId(), EdgeEventType.RULE_CHAIN, EdgeEventActionType.UPDATED, ruleChainId, isRootBody).get(); return savedEdge; } diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java index 9571319226..28cc32efa3 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java @@ -524,7 +524,7 @@ public final class EdgeGrpcSession implements Closeable { case CUSTOMER: return ctx.getCustomerProcessor().convertCustomerEventToDownlink(edgeEvent); case RULE_CHAIN: - return ctx.getRuleChainProcessor().convertRuleChainEventToDownlink(edge, edgeEvent); + return ctx.getRuleChainProcessor().convertRuleChainEventToDownlink(edgeEvent); case RULE_CHAIN_METADATA: return ctx.getRuleChainProcessor().convertRuleChainMetadataEventToDownlink(edgeEvent, this.edgeVersion); case ALARM: diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/RuleChainMsgConstructor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/RuleChainMsgConstructor.java index 35ac760faa..7f5ae72511 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/RuleChainMsgConstructor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/RuleChainMsgConstructor.java @@ -35,13 +35,13 @@ import org.thingsboard.server.service.edge.rpc.constructor.rule.RuleChainMetadat @TbCoreComponent public class RuleChainMsgConstructor { - public RuleChainUpdateMsg constructRuleChainUpdatedMsg(RuleChainId edgeRootRuleChainId, UpdateMsgType msgType, RuleChain ruleChain) { + public RuleChainUpdateMsg constructRuleChainUpdatedMsg(UpdateMsgType msgType, RuleChain ruleChain, boolean isRoot) { RuleChainUpdateMsg.Builder builder = RuleChainUpdateMsg.newBuilder() .setMsgType(msgType) .setIdMSB(ruleChain.getId().getId().getMostSignificantBits()) .setIdLSB(ruleChain.getId().getId().getLeastSignificantBits()) .setName(ruleChain.getName()) - .setRoot(ruleChain.getId().equals(edgeRootRuleChainId)) + .setRoot(isRoot) .setDebugMode(ruleChain.isDebugMode()) .setConfiguration(JacksonUtil.toString(ruleChain.getConfiguration())); if (ruleChain.getFirstRuleNodeId() != null) { diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/RuleChainEdgeProcessor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/RuleChainEdgeProcessor.java index 5704e7d30d..22d8005472 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/RuleChainEdgeProcessor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/RuleChainEdgeProcessor.java @@ -19,7 +19,6 @@ import com.google.common.util.concurrent.ListenableFuture; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Component; import org.thingsboard.server.common.data.EdgeUtils; -import org.thingsboard.server.common.data.edge.Edge; import org.thingsboard.server.common.data.edge.EdgeEvent; import org.thingsboard.server.common.data.id.RuleChainId; import org.thingsboard.server.common.data.id.TenantId; @@ -33,12 +32,14 @@ import org.thingsboard.server.gen.edge.v1.UpdateMsgType; import org.thingsboard.server.gen.transport.TransportProtos; import org.thingsboard.server.queue.util.TbCoreComponent; +import static org.thingsboard.server.service.edge.DefaultEdgeNotificationService.EDGE_IS_ROOT_BODY_KEY; + @Component @Slf4j @TbCoreComponent public class RuleChainEdgeProcessor extends BaseEdgeProcessor { - public DownlinkMsg convertRuleChainEventToDownlink(Edge edge, EdgeEvent edgeEvent) { + public DownlinkMsg convertRuleChainEventToDownlink(EdgeEvent edgeEvent) { RuleChainId ruleChainId = new RuleChainId(edgeEvent.getEntityId()); DownlinkMsg downlinkMsg = null; switch (edgeEvent.getAction()) { @@ -47,9 +48,15 @@ public class RuleChainEdgeProcessor extends BaseEdgeProcessor { case ASSIGNED_TO_EDGE: RuleChain ruleChain = ruleChainService.findRuleChainById(edgeEvent.getTenantId(), ruleChainId); if (ruleChain != null) { + boolean isRoot = false; + if (edgeEvent.getBody() != null && edgeEvent.getBody().get(EDGE_IS_ROOT_BODY_KEY) != null) { + try { + isRoot = Boolean.parseBoolean(edgeEvent.getBody().get(EDGE_IS_ROOT_BODY_KEY).asText()); + } catch (Exception ignored) {} + } UpdateMsgType msgType = getUpdateMsgType(edgeEvent.getAction()); RuleChainUpdateMsg ruleChainUpdateMsg = - ruleChainMsgConstructor.constructRuleChainUpdatedMsg(edge.getRootRuleChainId(), msgType, ruleChain); + ruleChainMsgConstructor.constructRuleChainUpdatedMsg(msgType, ruleChain, isRoot); downlinkMsg = DownlinkMsg.newBuilder() .setDownlinkMsgId(EdgeUtils.nextPositiveInt()) .addRuleChainUpdateMsg(ruleChainUpdateMsg) 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 f4d713b40f..481ee7762e 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 @@ -48,7 +48,7 @@ import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.EntityViewId; import org.thingsboard.server.common.data.id.RuleChainId; import org.thingsboard.server.common.data.id.TenantId; -import org.thingsboard.server.common.data.kv.AttributeKey; +import org.thingsboard.server.common.data.id.UserId; import org.thingsboard.server.common.data.kv.AttributeKvEntry; import org.thingsboard.server.common.msg.TbMsg; import org.thingsboard.server.common.msg.TbMsgMetaData; @@ -71,9 +71,7 @@ import org.thingsboard.server.queue.util.TbCoreComponent; import javax.annotation.Nullable; import javax.annotation.PostConstruct; import java.util.ArrayList; -import java.util.HashSet; import java.util.List; -import java.util.Set; @Component @Slf4j @@ -325,6 +323,9 @@ public class TelemetryEdgeProcessor extends BaseEdgeProcessor { case CUSTOMER: entityId = new CustomerId(edgeEvent.getEntityId()); break; + case USER: + entityId = new UserId(edgeEvent.getEntityId()); + break; case EDGE: entityId = new EdgeId(edgeEvent.getEntityId()); break; diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/sync/DefaultEdgeRequestsService.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/sync/DefaultEdgeRequestsService.java index 9210d3ceec..4be7ddb1c8 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/sync/DefaultEdgeRequestsService.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/sync/DefaultEdgeRequestsService.java @@ -165,7 +165,8 @@ public class DefaultEdgeRequestsService implements EdgeRequestsService { Map entityData = new HashMap<>(); ObjectNode attributes = JacksonUtil.OBJECT_MAPPER.createObjectNode(); for (AttributeKvEntry attr : ssAttributes) { - if (DefaultDeviceStateService.PERSISTENT_ATTRIBUTES.contains(attr.getKey())) { + if (DefaultDeviceStateService.PERSISTENT_ATTRIBUTES.contains(attr.getKey()) + && !DefaultDeviceStateService.INACTIVITY_TIMEOUT.equals(attr.getKey())) { continue; } if (attr.getDataType() == DataType.BOOLEAN && attr.getBooleanValue().isPresent()) { diff --git a/application/src/main/java/org/thingsboard/server/service/entitiy/device/DefaultTbDeviceService.java b/application/src/main/java/org/thingsboard/server/service/entitiy/device/DefaultTbDeviceService.java index f72092bedf..25053676fd 100644 --- a/application/src/main/java/org/thingsboard/server/service/entitiy/device/DefaultTbDeviceService.java +++ b/application/src/main/java/org/thingsboard/server/service/entitiy/device/DefaultTbDeviceService.java @@ -76,17 +76,18 @@ public class DefaultTbDeviceService extends AbstractTbEntityService implements T @Override public Device saveDeviceWithCredentials(Device device, DeviceCredentials credentials, User user) throws ThingsboardException { - ActionType actionType = device.getId() == null ? ActionType.ADDED : ActionType.UPDATED; + boolean isCreate = device.getId() == null; + ActionType actionType = isCreate ? ActionType.ADDED : ActionType.UPDATED; TenantId tenantId = device.getTenantId(); try { + Device oldDevice = isCreate ? null : deviceService.findDeviceById(tenantId, device.getId()); Device savedDevice = checkNotNull(deviceService.saveDeviceWithCredentials(device, credentials)); notificationEntityService.notifyCreateOrUpdateDevice(tenantId, savedDevice.getId(), savedDevice.getCustomerId(), - savedDevice, device, actionType, user); + savedDevice, oldDevice, actionType, user); return savedDevice; } catch (Exception e) { - notificationEntityService.logEntityAction(tenantId, emptyId(EntityType.DEVICE), device, - actionType, user, e); + notificationEntityService.logEntityAction(tenantId, emptyId(EntityType.DEVICE), device, actionType, user, e); throw e; } } diff --git a/application/src/main/java/org/thingsboard/server/service/install/update/DefaultCacheCleanupService.java b/application/src/main/java/org/thingsboard/server/service/install/update/DefaultCacheCleanupService.java index 6e67876dc4..367bb049e5 100644 --- a/application/src/main/java/org/thingsboard/server/service/install/update/DefaultCacheCleanupService.java +++ b/application/src/main/java/org/thingsboard/server/service/install/update/DefaultCacheCleanupService.java @@ -78,6 +78,10 @@ public class DefaultCacheCleanupService implements CacheCleanupService { clearCacheByName("assets"); clearCacheByName("repositorySettings"); break; + case "3.4.2": + log.info("Clearing cache to upgrade from version 3.4.2 to 3.4.3 ..."); + clearCacheByName("repositorySettings"); + break; default: //Do nothing, since cache cleanup is optional. } diff --git a/application/src/main/java/org/thingsboard/server/service/subscription/DefaultTbEntityDataSubscriptionService.java b/application/src/main/java/org/thingsboard/server/service/subscription/DefaultTbEntityDataSubscriptionService.java index a5a45be5b2..3d9077cb92 100644 --- a/application/src/main/java/org/thingsboard/server/service/subscription/DefaultTbEntityDataSubscriptionService.java +++ b/application/src/main/java/org/thingsboard/server/service/subscription/DefaultTbEntityDataSubscriptionService.java @@ -21,6 +21,7 @@ import com.google.common.util.concurrent.ListenableFuture; import com.google.common.util.concurrent.MoreExecutors; import lombok.Getter; import lombok.extern.slf4j.Slf4j; +import org.apache.commons.lang3.ArrayUtils; import org.checkerframework.checker.nullness.qual.Nullable; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; @@ -562,8 +563,13 @@ public class DefaultTbEntityDataSubscriptionService implements TbEntityDataSubsc if (queryResults != null) { for (ReadTsKvQueryResult queryResult : queryResults) { String queryKey = queriesKeys.get(queryResult.getQueryId()); - entityData.getTimeseries().put(queryKey, queryResult.toTsValues()); - lastTsMap.put(queryKey, queryResult.getLastEntryTs()); + if (queryKey != null) { + entityData.getTimeseries().merge(queryKey, queryResult.toTsValues(), ArrayUtils::addAll); + lastTsMap.merge(queryKey, queryResult.getLastEntryTs(), Math::max); + } else { + log.warn("ReadTsKvQueryResult for {} {} has queryId not matching the initial query", + entityData.getEntityId().getEntityType(), entityData.getEntityId()); + } } } // Populate with empty values if no data found. diff --git a/application/src/main/java/org/thingsboard/server/service/subscription/TbAbstractDataSubCtx.java b/application/src/main/java/org/thingsboard/server/service/subscription/TbAbstractDataSubCtx.java index 26e97b6f2c..271de08733 100644 --- a/application/src/main/java/org/thingsboard/server/service/subscription/TbAbstractDataSubCtx.java +++ b/application/src/main/java/org/thingsboard/server/service/subscription/TbAbstractDataSubCtx.java @@ -18,6 +18,7 @@ package org.thingsboard.server.service.subscription; import lombok.Getter; import lombok.extern.slf4j.Slf4j; import org.thingsboard.server.common.data.id.EntityId; +import org.thingsboard.server.common.data.kv.Aggregation; import org.thingsboard.server.common.data.page.PageData; import org.thingsboard.server.common.data.query.AbstractDataQuery; import org.thingsboard.server.common.data.query.EntityData; @@ -197,6 +198,9 @@ public abstract class TbAbstractDataSubCtx { long ts = Arrays.stream(v).map(TsValue::getTs).max(Long::compareTo).orElse(0L); log.trace("[{}][{}] Updating key: {} with ts: {}", serviceId, cmdId, k, ts); + if (!Aggregation.NONE.equals(getCurrentAggregation()) && ts < endTs) { + ts = endTs; + } keyStates.put(k, ts); }); } @@ -247,4 +251,5 @@ public abstract class TbAbstractDataSubCtx { } } + @Override + protected Aggregation getCurrentAggregation() { + return Aggregation.NONE; + } + private void sendWsMsg(String sessionId, AlarmSubscriptionUpdate subscriptionUpdate) { Alarm alarm = subscriptionUpdate.getAlarm(); AlarmId alarmId = alarm.getId(); diff --git a/application/src/main/java/org/thingsboard/server/service/subscription/TbEntityDataSubCtx.java b/application/src/main/java/org/thingsboard/server/service/subscription/TbEntityDataSubCtx.java index 385661f177..016ed198ec 100644 --- a/application/src/main/java/org/thingsboard/server/service/subscription/TbEntityDataSubCtx.java +++ b/application/src/main/java/org/thingsboard/server/service/subscription/TbEntityDataSubCtx.java @@ -19,6 +19,7 @@ import lombok.Getter; import lombok.Setter; import lombok.extern.slf4j.Slf4j; import org.thingsboard.server.common.data.id.EntityId; +import org.thingsboard.server.common.data.kv.Aggregation; import org.thingsboard.server.common.data.page.PageData; import org.thingsboard.server.common.data.query.EntityData; import org.thingsboard.server.common.data.query.EntityDataQuery; @@ -86,6 +87,11 @@ public class TbEntityDataSubCtx extends TbAbstractDataSubCtx { } } + @Override + protected Aggregation getCurrentAggregation() { + return (this.curTsCmd == null || this.curTsCmd.getAgg() == null) ? Aggregation.NONE : this.curTsCmd.getAgg(); + } + private void sendLatestWsMsg(EntityId entityId, String sessionId, TelemetrySubscriptionUpdate subscriptionUpdate, EntityKeyType keyType) { Map latestUpdate = new HashMap<>(); subscriptionUpdate.getData().forEach((k, v) -> { diff --git a/application/src/main/java/org/thingsboard/server/service/sync/ie/importing/csv/AbstractBulkImportService.java b/application/src/main/java/org/thingsboard/server/service/sync/ie/importing/csv/AbstractBulkImportService.java index 37d3cd2e71..d78403a884 100644 --- a/application/src/main/java/org/thingsboard/server/service/sync/ie/importing/csv/AbstractBulkImportService.java +++ b/application/src/main/java/org/thingsboard/server/service/sync/ie/importing/csv/AbstractBulkImportService.java @@ -87,7 +87,7 @@ public abstract class AbstractBulkImportService 0); + Assert.assertEquals(savedTenant.getId(), savedDevice.getTenantId()); + Assert.assertNotNull(savedDevice.getCustomerId()); + Assert.assertEquals(NULL_UUID, savedDevice.getCustomerId().getId()); + Assert.assertEquals(device.getName(), savedDevice.getName()); + + DeviceCredentials foundDeviceCredentials = + doGet("/api/device/" + savedDevice.getId().getId() + "/credentials", DeviceCredentials.class); + + Assert.assertNotNull(foundDeviceCredentials); + Assert.assertNotNull(foundDeviceCredentials.getId()); + Assert.assertEquals(savedDevice.getId(), foundDeviceCredentials.getDeviceId()); + Assert.assertEquals(DeviceCredentialsType.ACCESS_TOKEN, foundDeviceCredentials.getCredentialsType()); + Assert.assertEquals(testToken, foundDeviceCredentials.getCredentialsId()); + + Mockito.reset(tbClusterService, auditLogService, gatewayNotificationsService); + + savedDevice.setName("My new device"); + doPost("/api/device", savedDevice, Device.class); + + testNotifyEntityAllOneTime(savedDevice, savedDevice.getId(), savedDevice.getId(), savedTenant.getId(), + tenantAdmin.getCustomerId(), tenantAdmin.getId(), tenantAdmin.getEmail(), ActionType.UPDATED); + testNotificationUpdateGatewayOneTime(savedDevice, oldDevice); + } + @Test public void saveDeviceWithViolationOfValidation() throws Exception { Device device = new Device(); @@ -1235,6 +1291,66 @@ public abstract class BaseDeviceControllerTest extends AbstractControllerTest { testEntityDaoWithRelationsTransactionalException(deviceDao, savedTenant.getId(), deviceId, "/api/device/" + deviceId); } + @Test + public void testBulkImportDeviceWithoutCredentials() throws Exception { + String deviceName = "some_device"; + String deviceType = "some_type"; + BulkImportRequest request = new BulkImportRequest(); + request.setFile(String.format("NAME,TYPE\n%s,%s", deviceName, deviceType)); + BulkImportRequest.Mapping mapping = new BulkImportRequest.Mapping(); + BulkImportRequest.ColumnMapping name = new BulkImportRequest.ColumnMapping(); + name.setType(BulkImportColumnType.NAME); + BulkImportRequest.ColumnMapping type = new BulkImportRequest.ColumnMapping(); + type.setType(BulkImportColumnType.TYPE); + List columns = new ArrayList<>(); + columns.add(name); + columns.add(type); + + mapping.setColumns(columns); + mapping.setDelimiter(','); + mapping.setUpdate(true); + mapping.setHeader(true); + request.setMapping(mapping); + + BulkImportResult deviceBulkImportResult = doPostWithTypedResponse("/api/device/bulk_import", request, new TypeReference<>() {}); + + Assert.assertEquals(1, deviceBulkImportResult.getCreated().get()); + Assert.assertEquals(0, deviceBulkImportResult.getErrors().get()); + Assert.assertEquals(0, deviceBulkImportResult.getUpdated().get()); + Assert.assertTrue(deviceBulkImportResult.getErrorsList().isEmpty()); + + Device savedDevice = doGet("/api/tenant/devices?deviceName=" + deviceName, Device.class); + + Assert.assertNotNull(savedDevice); + Assert.assertEquals(deviceName, savedDevice.getName()); + Assert.assertEquals(deviceType, savedDevice.getType()); + + DeviceCredentials savedCredentials = + doGet("/api/device/" + savedDevice.getId().getId() + "/credentials", DeviceCredentials.class); + + Assert.assertNotNull(savedCredentials); + Assert.assertNotNull(savedCredentials.getId()); + Assert.assertEquals(savedDevice.getId(), savedCredentials.getDeviceId()); + Assert.assertEquals(DeviceCredentialsType.ACCESS_TOKEN, savedCredentials.getCredentialsType()); + Assert.assertNotNull(savedCredentials.getCredentialsId()); + Assert.assertEquals(20, savedCredentials.getCredentialsId().length()); + + deviceBulkImportResult = doPostWithTypedResponse("/api/device/bulk_import", request, new TypeReference<>() {}); + + Assert.assertEquals(0, deviceBulkImportResult.getCreated().get()); + Assert.assertEquals(0, deviceBulkImportResult.getErrors().get()); + Assert.assertEquals(1, deviceBulkImportResult.getUpdated().get()); + Assert.assertTrue(deviceBulkImportResult.getErrorsList().isEmpty()); + + Device updatedDevice = doGet("/api/device/" + savedDevice.getId().getId(), Device.class); + Assert.assertEquals(savedDevice, updatedDevice); + + DeviceCredentials updatedCredentials = + doGet("/api/device/" + savedDevice.getId().getId() + "/credentials", DeviceCredentials.class); + + Assert.assertEquals(savedCredentials, updatedCredentials); + } + private Device createDevice(String name) { Device device = new Device(); device.setName(name); diff --git a/application/src/test/java/org/thingsboard/server/controller/BaseEntityQueryControllerTest.java b/application/src/test/java/org/thingsboard/server/controller/BaseEntityQueryControllerTest.java index 8c692ef8a2..85d68a479d 100644 --- a/application/src/test/java/org/thingsboard/server/controller/BaseEntityQueryControllerTest.java +++ b/application/src/test/java/org/thingsboard/server/controller/BaseEntityQueryControllerTest.java @@ -22,6 +22,7 @@ import org.junit.After; import org.junit.Assert; import org.junit.Before; import org.junit.Test; +import org.springframework.test.web.servlet.ResultActions; import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.server.common.data.DataConstants; import org.thingsboard.server.common.data.Device; @@ -55,6 +56,7 @@ import java.util.List; import java.util.concurrent.TimeUnit; import java.util.stream.Collectors; +import static org.assertj.core.api.Assertions.assertThat; import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status; public abstract class BaseEntityQueryControllerTest extends AbstractControllerTest { @@ -381,4 +383,24 @@ public abstract class BaseEntityQueryControllerTest extends AbstractControllerTe Assert.assertEquals("1" + i, alarmActiveTime); } } + + @Test + public void givenInvalidEntityDataPageLink_thenReturnError() throws Exception { + DeviceTypeFilter filter = new DeviceTypeFilter(); + filter.setDeviceType("default"); + filter.setDeviceNameFilter(""); + + String invalidSortProperty = "created(Time)"; + EntityDataSortOrder sortOrder = new EntityDataSortOrder( + new EntityKey(EntityKeyType.ENTITY_FIELD, invalidSortProperty), EntityDataSortOrder.Direction.ASC + ); + EntityDataPageLink pageLink = new EntityDataPageLink(10, 0, null, sortOrder); + List entityFields = Collections.singletonList(new EntityKey(EntityKeyType.ENTITY_FIELD, "name")); + List latestValues = Collections.singletonList(new EntityKey(EntityKeyType.ATTRIBUTE, "temperature")); + EntityDataQuery query = new EntityDataQuery(filter, pageLink, entityFields, latestValues, null); + + ResultActions result = doPost("/api/entitiesQuery/find", query).andExpect(status().isBadRequest()); + assertThat(getErrorMessage(result)).contains("Invalid").contains("sort property"); + } + } diff --git a/application/src/test/java/org/thingsboard/server/controller/BaseRuleChainControllerTest.java b/application/src/test/java/org/thingsboard/server/controller/BaseRuleChainControllerTest.java index d32b1baa6b..8daf8356ea 100644 --- a/application/src/test/java/org/thingsboard/server/controller/BaseRuleChainControllerTest.java +++ b/application/src/test/java/org/thingsboard/server/controller/BaseRuleChainControllerTest.java @@ -26,6 +26,8 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Primary; import org.springframework.test.context.ContextConfiguration; +import org.thingsboard.rule.engine.action.TbCreateAlarmNode; +import org.thingsboard.rule.engine.action.TbCreateAlarmNodeConfiguration; import org.thingsboard.server.common.data.StringUtils; import org.thingsboard.server.common.data.Tenant; import org.thingsboard.server.common.data.User; @@ -35,7 +37,9 @@ import org.thingsboard.server.common.data.id.RuleChainId; 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.data.rule.RuleChainMetaData; import org.thingsboard.server.common.data.rule.RuleChainType; +import org.thingsboard.server.common.data.rule.RuleNode; import org.thingsboard.server.common.data.security.Authority; import org.thingsboard.server.dao.exception.DataValidationException; import org.thingsboard.server.dao.rule.RuleChainDao; @@ -44,6 +48,7 @@ import java.util.ArrayList; import java.util.Collections; import java.util.List; +import static org.assertj.core.api.Assertions.assertThat; import static org.hamcrest.Matchers.containsString; import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status; @@ -254,9 +259,35 @@ public abstract class BaseRuleChainControllerTest extends AbstractControllerTest testEntityDaoWithRelationsTransactionalException(ruleChainDao, savedTenant.getId(), ruleChainId, "/api/ruleChain/" + ruleChainId); } + @Test + public void givenRuleNodeWithInvalidConfiguration_thenReturnError() throws Exception { + RuleChain ruleChain = createRuleChain("Rule chain with invalid nodes"); + RuleChainMetaData ruleChainMetaData = new RuleChainMetaData(); + ruleChainMetaData.setRuleChainId(ruleChain.getId()); + + RuleNode createAlarmNode = new RuleNode(); + createAlarmNode.setName("Create alarm"); + createAlarmNode.setType(TbCreateAlarmNode.class.getName()); + TbCreateAlarmNodeConfiguration invalidCreateAlarmNodeConfiguration = new TbCreateAlarmNodeConfiguration(); + invalidCreateAlarmNodeConfiguration.setSeverity("