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 e8339e3202..44d81902a8 100644 --- a/application/src/main/java/org/thingsboard/server/install/ThingsboardInstallService.java +++ b/application/src/main/java/org/thingsboard/server/install/ThingsboardInstallService.java @@ -126,6 +126,7 @@ public class ThingsboardInstallService { case "3.6.2": log.info("Upgrading ThingsBoard from version 3.6.2 to 3.6.3 ..."); databaseEntitiesUpgradeService.upgradeDatabase("3.6.2"); + systemDataLoaderService.updateDefaultNotificationConfigs(); //TODO DON'T FORGET to update switch statement in the CacheCleanupService if you need to clear the cache break; default: diff --git a/application/src/main/java/org/thingsboard/server/service/edge/EdgeContextComponent.java b/application/src/main/java/org/thingsboard/server/service/edge/EdgeContextComponent.java index a10a48244b..2f157af70e 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/EdgeContextComponent.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/EdgeContextComponent.java @@ -20,6 +20,7 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.Lazy; import org.springframework.stereotype.Component; import org.thingsboard.server.cluster.TbClusterService; +import org.thingsboard.server.common.msg.notification.NotificationRuleProcessor; import org.thingsboard.server.dao.asset.AssetProfileService; import org.thingsboard.server.dao.asset.AssetService; import org.thingsboard.server.dao.attributes.AttributesService; @@ -149,6 +150,9 @@ public class EdgeContextComponent { @Autowired private ResourceService resourceService; + @Autowired + private NotificationRuleProcessor notificationRuleProcessor; + @Autowired private AlarmEdgeProcessor alarmProcessor; 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 d3262039d5..c23fd98d8e 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 @@ -38,6 +38,7 @@ 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.data.msg.TbMsgType; +import org.thingsboard.server.common.data.notification.rule.trigger.EdgeConnectionTrigger; import org.thingsboard.server.common.msg.TbMsg; import org.thingsboard.server.common.msg.TbMsgDataType; import org.thingsboard.server.common.msg.TbMsgMetaData; @@ -263,7 +264,8 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i } private void onEdgeConnect(EdgeId edgeId, EdgeGrpcSession edgeGrpcSession) { - TenantId tenantId = edgeGrpcSession.getEdge().getTenantId(); + Edge edge = edgeGrpcSession.getEdge(); + TenantId tenantId = edge.getTenantId(); log.info("[{}][{}] edge [{}] connected successfully.", tenantId, edgeGrpcSession.getSessionId(), edgeId); sessions.put(edgeId, edgeGrpcSession); final Lock newEventLock = sessionNewEventsLocks.computeIfAbsent(edgeId, id -> new ReentrantLock()); @@ -276,7 +278,7 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i save(tenantId, edgeId, DefaultDeviceStateService.ACTIVITY_STATE, true); long lastConnectTs = System.currentTimeMillis(); save(tenantId, edgeId, DefaultDeviceStateService.LAST_CONNECT_TIME, lastConnectTs); - pushRuleEngineMessage(tenantId, edgeId, lastConnectTs, TbMsgType.CONNECT_EVENT); + pushRuleEngineMessage(tenantId, edge, lastConnectTs, TbMsgType.CONNECT_EVENT); cancelScheduleEdgeEventsCheck(edgeId); scheduleEdgeEventsCheck(edgeGrpcSession); } @@ -381,7 +383,8 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i } } - private void onEdgeDisconnect(EdgeId edgeId, UUID sessionId) { + private void onEdgeDisconnect(Edge edge, UUID sessionId) { + EdgeId edgeId = edge.getId(); log.info("[{}][{}] edge disconnected!", edgeId, sessionId); EdgeGrpcSession toRemove = sessions.get(edgeId); if (toRemove.getSessionId().equals(sessionId)) { @@ -397,7 +400,7 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i save(tenantId, edgeId, DefaultDeviceStateService.ACTIVITY_STATE, false); long lastDisconnectTs = System.currentTimeMillis(); save(tenantId, edgeId, DefaultDeviceStateService.LAST_DISCONNECT_TIME, lastDisconnectTs); - pushRuleEngineMessage(toRemove.getEdge().getTenantId(), edgeId, lastDisconnectTs, TbMsgType.DISCONNECT_EVENT); + pushRuleEngineMessage(toRemove.getEdge().getTenantId(), edge, lastDisconnectTs, TbMsgType.DISCONNECT_EVENT); cancelScheduleEdgeEventsCheck(edgeId); } else { log.debug("[{}] edge session [{}] is not available anymore, nothing to remove. most probably this session is already outdated!", edgeId, sessionId); @@ -452,16 +455,24 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i } } - private void pushRuleEngineMessage(TenantId tenantId, EdgeId edgeId, long ts, TbMsgType msgType) { + private void pushRuleEngineMessage(TenantId tenantId, Edge edge, long ts, TbMsgType msgType) { try { + EdgeId edgeId = edge.getId(); ObjectNode edgeState = JacksonUtil.newObjectNode(); - if (msgType.equals(TbMsgType.CONNECT_EVENT)) { + boolean isConnected = TbMsgType.CONNECT_EVENT.equals(msgType); + if (isConnected) { edgeState.put(DefaultDeviceStateService.ACTIVITY_STATE, true); edgeState.put(DefaultDeviceStateService.LAST_CONNECT_TIME, ts); } else { edgeState.put(DefaultDeviceStateService.ACTIVITY_STATE, false); edgeState.put(DefaultDeviceStateService.LAST_DISCONNECT_TIME, ts); } + ctx.getNotificationRuleProcessor().process(EdgeConnectionTrigger.builder() + .tenantId(tenantId) + .customerId(edge.getCustomerId()) + .edgeId(edgeId) + .edgeName(edge.getName()) + .connected(isConnected).build()); String data = JacksonUtil.toString(edgeState); TbMsgMetaData md = new TbMsgMetaData(); if (!persistToTelemetry) { @@ -470,7 +481,8 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i TbMsg tbMsg = TbMsg.newMsg(msgType, edgeId, md, TbMsgDataType.JSON, data); clusterService.pushMsgToRuleEngine(tenantId, edgeId, tbMsg, null); } catch (Exception e) { - log.warn("[{}][{}] Failed to push {}", tenantId, edgeId, msgType, e); + log.warn("[{}][{}] Failed to push {}", tenantId, edge.getId(), msgType, e); } } + } 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 5e6f7a263e..98346674ae 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 @@ -35,6 +35,7 @@ import org.thingsboard.server.common.data.kv.AttributeKvEntry; import org.thingsboard.server.common.data.kv.BaseAttributeKvEntry; import org.thingsboard.server.common.data.kv.LongDataEntry; import org.thingsboard.server.common.data.kv.StringDataEntry; +import org.thingsboard.server.common.data.notification.rule.trigger.EdgeCommunicationFailureTrigger; import org.thingsboard.server.common.data.page.PageData; import org.thingsboard.server.common.data.page.PageLink; import org.thingsboard.server.common.data.page.SortOrder; @@ -111,7 +112,7 @@ public final class EdgeGrpcSession implements Closeable { private final UUID sessionId; private final BiConsumer sessionOpenListener; - private final BiConsumer sessionCloseListener; + private final BiConsumer sessionCloseListener; private final EdgeSessionState sessionState = new EdgeSessionState(); @@ -137,7 +138,7 @@ public final class EdgeGrpcSession implements Closeable { private ScheduledExecutorService sendDownlinkExecutorService; EdgeGrpcSession(EdgeContextComponent ctx, StreamObserver outputStream, BiConsumer sessionOpenListener, - BiConsumer sessionCloseListener, ScheduledExecutorService sendDownlinkExecutorService, int maxInboundMessageSize) { + BiConsumer sessionCloseListener, ScheduledExecutorService sendDownlinkExecutorService, int maxInboundMessageSize) { this.sessionId = UUID.randomUUID(); this.ctx = ctx; this.outputStream = outputStream; @@ -206,7 +207,7 @@ public final class EdgeGrpcSession implements Closeable { connected = false; if (edge != null) { try { - sessionCloseListener.accept(edge.getId(), sessionId); + sessionCloseListener.accept(edge, sessionId); } catch (Exception ignored) { } } @@ -314,7 +315,7 @@ public final class EdgeGrpcSession implements Closeable { } catch (Exception e) { log.error("[{}][{}] Failed to send downlink message [{}]", this.tenantId, this.sessionId, downlinkMsg, e); connected = false; - sessionCloseListener.accept(edge.getId(), sessionId); + sessionCloseListener.accept(edge, sessionId); } finally { downlinkMsgLock.unlock(); } @@ -466,15 +467,26 @@ public final class EdgeGrpcSession implements Closeable { if (isConnected() && sessionState.getPendingMsgsMap().values().size() > 0) { List copy = new ArrayList<>(sessionState.getPendingMsgsMap().values()); if (attempt > 1) { - log.warn("[{}][{}] Failed to deliver the batch: {}, attempt: {}", this.tenantId, this.sessionId, copy, attempt); + String error = "Failed to deliver the batch"; + String failureMsg = String.format("{%s}: {%s}", error, copy); + if (attempt == 2) { + // Send a failure notification only on the second attempt. + // This ensures that failure alerts are sent just once to avoid redundant notifications. + ctx.getNotificationRuleProcessor().process(EdgeCommunicationFailureTrigger.builder().tenantId(tenantId) + .edgeId(edge.getId()).customerId(edge.getCustomerId()).edgeName(edge.getName()).failureMsg(failureMsg).error(error).build()); + } + log.warn("[{}][{}] {}, attempt: {}", this.tenantId, this.sessionId, failureMsg, attempt); } log.trace("[{}][{}][{}] downlink msg(s) are going to be send.", this.tenantId, this.sessionId, copy.size()); for (DownlinkMsg downlinkMsg : copy) { if (this.clientMaxInboundMessageSize != 0 && downlinkMsg.getSerializedSize() > this.clientMaxInboundMessageSize) { - log.error("[{}][{}][{}] Downlink msg size [{}] exceeds client max inbound message size [{}]. Skipping this message. " + - "Please increase value of CLOUD_RPC_MAX_INBOUND_MESSAGE_SIZE env variable on the edge and restart it." + - "Message {}", this.tenantId, edge.getId(), this.sessionId, downlinkMsg.getSerializedSize(), - this.clientMaxInboundMessageSize, downlinkMsg); + String error = String.format("Client max inbound message size [{%s}] is exceeded. Please increase value of CLOUD_RPC_MAX_INBOUND_MESSAGE_SIZE " + + "env variable on the edge and restart it.", this.clientMaxInboundMessageSize); + String message = String.format("Downlink msg size [{%s}] exceeds client max inbound message size [{%s}]. " + + "Please increase value of CLOUD_RPC_MAX_INBOUND_MESSAGE_SIZE env variable on the edge and restart it.", downlinkMsg.getSerializedSize(), this.clientMaxInboundMessageSize); + log.error("[{}][{}][{}] {} Message {}", this.tenantId, edge.getId(), this.sessionId, message, downlinkMsg); + ctx.getNotificationRuleProcessor().process(EdgeCommunicationFailureTrigger.builder().tenantId(tenantId) + .edgeId(edge.getId()).customerId(edge.getCustomerId()).edgeName(edge.getName()).failureMsg(message).error(error).build()); sessionState.getPendingMsgsMap().remove(downlinkMsg.getDownlinkMsgId()); } else { sendDownlinkMsg(ResponseMsg.newBuilder() @@ -485,8 +497,12 @@ public final class EdgeGrpcSession implements Closeable { if (attempt < MAX_DOWNLINK_ATTEMPTS) { scheduleDownlinkMsgsPackSend(attempt + 1); } else { + String failureMsg = String.format("Failed to deliver messages: %s", copy); log.warn("[{}][{}] Failed to deliver the batch after {} attempts. Next messages are going to be discarded {}", this.tenantId, this.sessionId, MAX_DOWNLINK_ATTEMPTS, copy); + ctx.getNotificationRuleProcessor().process(EdgeCommunicationFailureTrigger.builder().tenantId(tenantId).edgeId(edge.getId()) + .customerId(edge.getCustomerId()).edgeName(edge.getName()).failureMsg(failureMsg) + .error("Failed to deliver messages after " + MAX_DOWNLINK_ATTEMPTS + " attempts").build()); stopCurrentSendDownlinkMsgsTask(false); } } else { @@ -791,7 +807,10 @@ public final class EdgeGrpcSession implements Closeable { } } } catch (Exception e) { + String failureMsg = String.format("Can't process uplink msg [%s] from edge", uplinkMsg); log.error("[{}][{}] Can't process uplink msg [{}]", this.tenantId, this.sessionId, uplinkMsg, e); + ctx.getNotificationRuleProcessor().process(EdgeCommunicationFailureTrigger.builder().tenantId(tenantId).edgeId(edge.getId()) + .customerId(edge.getCustomerId()).edgeName(edge.getName()).failureMsg(failureMsg).error(e.getMessage()).build()); return Futures.immediateFailedFuture(e); } return Futures.allAsList(result); @@ -815,15 +834,22 @@ public final class EdgeGrpcSession implements Closeable { .setMaxInboundMessageSize(maxInboundMessageSize) .build(); } + String error = "Failed to validate the edge!"; + String failureMsg = String.format("{%s} Provided request secret: %s", error, request.getEdgeSecret()); + ctx.getNotificationRuleProcessor().process(EdgeCommunicationFailureTrigger.builder().tenantId(tenantId).edgeId(edge.getId()) + .customerId(edge.getCustomerId()).edgeName(edge.getName()).failureMsg(failureMsg).error(error).build()); return ConnectResponseMsg.newBuilder() .setResponseCode(ConnectResponseCode.BAD_CREDENTIALS) - .setErrorMsg("Failed to validate the edge!") + .setErrorMsg(failureMsg) .setConfiguration(EdgeConfiguration.getDefaultInstance()).build(); } catch (Exception e) { - log.error("[{}] Failed to process edge connection!", request.getEdgeRoutingKey(), e); + String failureMsg = "Failed to process edge connection!"; + ctx.getNotificationRuleProcessor().process(EdgeCommunicationFailureTrigger.builder().tenantId(tenantId).edgeId(edge.getId()) + .customerId(edge.getCustomerId()).edgeName(edge.getName()).failureMsg(failureMsg).error(e.getMessage()).build()); + log.error(failureMsg, e); return ConnectResponseMsg.newBuilder() .setResponseCode(ConnectResponseCode.SERVER_UNAVAILABLE) - .setErrorMsg("Failed to process edge connection!") + .setErrorMsg(failureMsg) .setConfiguration(EdgeConfiguration.getDefaultInstance()).build(); } } diff --git a/application/src/main/java/org/thingsboard/server/service/install/DefaultSystemDataLoaderService.java b/application/src/main/java/org/thingsboard/server/service/install/DefaultSystemDataLoaderService.java index b9f4dc482f..b20a548d90 100644 --- a/application/src/main/java/org/thingsboard/server/service/install/DefaultSystemDataLoaderService.java +++ b/application/src/main/java/org/thingsboard/server/service/install/DefaultSystemDataLoaderService.java @@ -691,7 +691,23 @@ public class DefaultSystemDataLoaderService implements SystemDataLoaderService { } @Override + @SneakyThrows public void updateDefaultNotificationConfigs() { + PageDataIterable tenants = new PageDataIterable<>(tenantService::findTenantsIds, 500); + ExecutorService executor = Executors.newFixedThreadPool(Math.max(Runtime.getRuntime().availableProcessors(), 4)); + log.info("Updating default edge failure notification configs for all tenants"); + AtomicInteger count = new AtomicInteger(); + for (TenantId tenantId : tenants) { + executor.submit(() -> { + notificationSettingsService.updateDefaultNotificationConfigs(tenantId); + int n = count.incrementAndGet(); + if (n % 500 == 0) { + log.info("{} tenants processed", n); + } + }); + } + executor.shutdown(); + executor.awaitTermination(Integer.MAX_VALUE, TimeUnit.SECONDS); notificationSettingsService.updateDefaultNotificationConfigs(TenantId.SYS_TENANT_ID); } diff --git a/application/src/main/java/org/thingsboard/server/service/notification/rule/trigger/EdgeCommunicationFailureTriggerProcessor.java b/application/src/main/java/org/thingsboard/server/service/notification/rule/trigger/EdgeCommunicationFailureTriggerProcessor.java new file mode 100644 index 0000000000..38d8ae9805 --- /dev/null +++ b/application/src/main/java/org/thingsboard/server/service/notification/rule/trigger/EdgeCommunicationFailureTriggerProcessor.java @@ -0,0 +1,62 @@ +/** + * Copyright © 2016-2024 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.service.notification.rule.trigger; + +import lombok.RequiredArgsConstructor; +import org.apache.commons.collections.CollectionUtils; +import org.springframework.stereotype.Service; +import org.thingsboard.server.common.data.notification.info.EdgeCommunicationFailureNotificationInfo; +import org.thingsboard.server.common.data.notification.info.RuleOriginatedNotificationInfo; +import org.thingsboard.server.common.data.notification.rule.trigger.EdgeCommunicationFailureTrigger; +import org.thingsboard.server.common.data.notification.rule.trigger.config.EdgeCommunicationFailureNotificationRuleTriggerConfig; +import org.thingsboard.server.common.data.notification.rule.trigger.config.NotificationRuleTriggerType; + +@Service +@RequiredArgsConstructor +public class EdgeCommunicationFailureTriggerProcessor implements NotificationRuleTriggerProcessor { + + @Override + public boolean matchesFilter(EdgeCommunicationFailureTrigger trigger, EdgeCommunicationFailureNotificationRuleTriggerConfig triggerConfig) { + if (CollectionUtils.isNotEmpty(triggerConfig.getEdges())) { + return !triggerConfig.getEdges().contains(trigger.getEdgeId().getId()); + } + return true; + } + + @Override + public RuleOriginatedNotificationInfo constructNotificationInfo(EdgeCommunicationFailureTrigger trigger) { + return EdgeCommunicationFailureNotificationInfo.builder() + .tenantId(trigger.getTenantId()) + .edgeId(trigger.getEdgeId()) + .customerId(trigger.getCustomerId()) + .edgeName(trigger.getEdgeName()) + .failureMsg(truncateFailureMsg(trigger.getFailureMsg())) + .build(); + } + + @Override + public NotificationRuleTriggerType getTriggerType() { + return NotificationRuleTriggerType.EDGE_COMMUNICATION_FAILURE; + } + + private String truncateFailureMsg(String input) { + int maxLength = 500; + if (input != null && input.length() > maxLength) { + return input.substring(0, maxLength); + } + return input; + } +} diff --git a/application/src/main/java/org/thingsboard/server/service/notification/rule/trigger/EdgeConnectionTriggerProcessor.java b/application/src/main/java/org/thingsboard/server/service/notification/rule/trigger/EdgeConnectionTriggerProcessor.java new file mode 100644 index 0000000000..ed1ee74622 --- /dev/null +++ b/application/src/main/java/org/thingsboard/server/service/notification/rule/trigger/EdgeConnectionTriggerProcessor.java @@ -0,0 +1,60 @@ +/** + * Copyright © 2016-2024 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.service.notification.rule.trigger; + +import lombok.RequiredArgsConstructor; +import org.apache.commons.collections.CollectionUtils; +import org.springframework.stereotype.Service; +import org.thingsboard.server.common.data.notification.info.EdgeConnectionNotificationInfo; +import org.thingsboard.server.common.data.notification.info.RuleOriginatedNotificationInfo; +import org.thingsboard.server.common.data.notification.rule.trigger.EdgeConnectionTrigger; +import org.thingsboard.server.common.data.notification.rule.trigger.config.EdgeConnectionNotificationRuleTriggerConfig; +import org.thingsboard.server.common.data.notification.rule.trigger.config.EdgeConnectionNotificationRuleTriggerConfig.EdgeConnectivityEvent; +import org.thingsboard.server.common.data.notification.rule.trigger.config.NotificationRuleTriggerType; + +@Service +@RequiredArgsConstructor +public class EdgeConnectionTriggerProcessor implements NotificationRuleTriggerProcessor { + + @Override + public boolean matchesFilter(EdgeConnectionTrigger trigger, EdgeConnectionNotificationRuleTriggerConfig triggerConfig) { + EdgeConnectivityEvent event = trigger.isConnected() ? EdgeConnectivityEvent.CONNECTED : EdgeConnectivityEvent.DISCONNECTED; + if (CollectionUtils.isEmpty(triggerConfig.getNotifyOn()) || !triggerConfig.getNotifyOn().contains(event)) { + return false; + } + if (CollectionUtils.isNotEmpty(triggerConfig.getEdges())) { + return triggerConfig.getEdges().contains(trigger.getEdgeId().getId()); + } + return true; + } + + @Override + public RuleOriginatedNotificationInfo constructNotificationInfo(EdgeConnectionTrigger trigger) { + return EdgeConnectionNotificationInfo.builder() + .eventType(trigger.isConnected() ? "connected" : "disconnected") + .tenantId(trigger.getTenantId()) + .customerId(trigger.getCustomerId()) + .edgeId(trigger.getEdgeId()) + .edgeName(trigger.getEdgeName()) + .build(); + } + + @Override + public NotificationRuleTriggerType getTriggerType() { + return NotificationRuleTriggerType.EDGE_CONNECTION; + } + +} diff --git a/application/src/main/java/org/thingsboard/server/service/sync/ie/exporting/impl/NotificationRuleExportService.java b/application/src/main/java/org/thingsboard/server/service/sync/ie/exporting/impl/NotificationRuleExportService.java index 7a7d176f81..35f2489f83 100644 --- a/application/src/main/java/org/thingsboard/server/service/sync/ie/exporting/impl/NotificationRuleExportService.java +++ b/application/src/main/java/org/thingsboard/server/service/sync/ie/exporting/impl/NotificationRuleExportService.java @@ -29,6 +29,8 @@ import org.thingsboard.server.common.data.notification.rule.EscalatedNotificatio import org.thingsboard.server.common.data.notification.rule.NotificationRule; import org.thingsboard.server.common.data.notification.rule.NotificationRuleRecipientsConfig; import org.thingsboard.server.common.data.notification.rule.trigger.config.DeviceActivityNotificationRuleTriggerConfig; +import org.thingsboard.server.common.data.notification.rule.trigger.config.EdgeCommunicationFailureNotificationRuleTriggerConfig; +import org.thingsboard.server.common.data.notification.rule.trigger.config.EdgeConnectionNotificationRuleTriggerConfig; import org.thingsboard.server.common.data.notification.rule.trigger.config.NotificationRuleTriggerConfig; import org.thingsboard.server.common.data.notification.rule.trigger.config.RuleEngineComponentLifecycleEventNotificationRuleTriggerConfig; import org.thingsboard.server.common.data.sync.ie.EntityExportData; @@ -65,13 +67,24 @@ public class NotificationRuleExportService ruleChains = triggerConfig.getRuleChains(); if (ruleChains != null) { triggerConfig.setRuleChains(toExternalIds(ruleChains, RuleChainId::new, ctx).collect(Collectors.toSet())); } break; + } + case EDGE_CONNECTION: { + EdgeConnectionNotificationRuleTriggerConfig triggerConfig = (EdgeConnectionNotificationRuleTriggerConfig) ruleTriggerConfig; + triggerConfig.setEdges(null); + break; + } + case EDGE_COMMUNICATION_FAILURE: { + EdgeCommunicationFailureNotificationRuleTriggerConfig triggerConfig = (EdgeCommunicationFailureNotificationRuleTriggerConfig) ruleTriggerConfig; + triggerConfig.setEdges(null); + break; + } } NotificationRuleRecipientsConfig ruleRecipientsConfig = notificationRule.getRecipientsConfig(); diff --git a/application/src/main/java/org/thingsboard/server/service/sync/ie/importing/impl/NotificationRuleImportService.java b/application/src/main/java/org/thingsboard/server/service/sync/ie/importing/impl/NotificationRuleImportService.java index 72f0fa219e..dd53e5b5bc 100644 --- a/application/src/main/java/org/thingsboard/server/service/sync/ie/importing/impl/NotificationRuleImportService.java +++ b/application/src/main/java/org/thingsboard/server/service/sync/ie/importing/impl/NotificationRuleImportService.java @@ -33,6 +33,8 @@ import org.thingsboard.server.common.data.notification.rule.EscalatedNotificatio import org.thingsboard.server.common.data.notification.rule.NotificationRule; import org.thingsboard.server.common.data.notification.rule.NotificationRuleRecipientsConfig; import org.thingsboard.server.common.data.notification.rule.trigger.config.DeviceActivityNotificationRuleTriggerConfig; +import org.thingsboard.server.common.data.notification.rule.trigger.config.EdgeConnectionNotificationRuleTriggerConfig; +import org.thingsboard.server.common.data.notification.rule.trigger.config.EdgeCommunicationFailureNotificationRuleTriggerConfig; import org.thingsboard.server.common.data.notification.rule.trigger.config.NotificationRuleTriggerConfig; import org.thingsboard.server.common.data.notification.rule.trigger.config.NotificationRuleTriggerType; import org.thingsboard.server.common.data.notification.rule.trigger.config.RuleEngineComponentLifecycleEventNotificationRuleTriggerConfig; @@ -86,7 +88,7 @@ public class NotificationRuleImportService extends BaseEntityImportService ruleChains = triggerConfig.getRuleChains(); if (ruleChains != null) { @@ -95,6 +97,17 @@ public class NotificationRuleImportService extends BaseEntityImportService matcherEntityClassEquals = argument -> argument.getClass().equals(entity.getClass()); - ArgumentMatcher matcherOriginatorId = argument -> argument.getClass().equals(originatorId.getClass()); - ArgumentMatcher matcherCustomerId = customerId == null ? - argument -> argument.getClass().equals(CustomerId.class) : argument -> argument.equals(customerId); - ArgumentMatcher matcherUserId = userId == null ? - argument -> argument.getClass().equals(UserId.class) : argument -> argument.equals(userId); - testLogEntityActionAdditionalInfo(matcherEntityClassEquals, matcherOriginatorId, tenantId, matcherCustomerId, matcherUserId, userName, actionType, cntTime, - extractMatcherAdditionalInfo(additionalInfo)); - testPushMsgToRuleEngineTime(matcherOriginatorId, tenantId, entity, cntTime); - Mockito.reset(tbClusterService, auditLogService); - } - protected void testNotifyManyEntityManyTimeMsgToEdgeServiceEntityEqAny(HasName entity, HasName originator, TenantId tenantId, CustomerId customerId, UserId userId, String userName, ActionType actionType, @@ -624,7 +606,7 @@ public abstract class AbstractNotifyEntityTest extends AbstractWebTest { private String entityClassToString(HasName entity) { String className = entity.getClass().toString() .substring(entity.getClass().toString().lastIndexOf(".") + 1); - List str = className.chars() + List str = className.chars() .mapToObj(x -> (Character.isUpperCase(x)) ? "_" + Character.toString(x) : Character.toString(x)) .collect(Collectors.toList()); return String.join("", str).toUpperCase(Locale.ENGLISH).substring(1); diff --git a/application/src/test/java/org/thingsboard/server/service/limits/RateLimitServiceTest.java b/application/src/test/java/org/thingsboard/server/service/limits/RateLimitServiceTest.java index afdb38a2ec..2570b80f9f 100644 --- a/application/src/test/java/org/thingsboard/server/service/limits/RateLimitServiceTest.java +++ b/application/src/test/java/org/thingsboard/server/service/limits/RateLimitServiceTest.java @@ -69,6 +69,8 @@ public class RateLimitServiceTest { profileConfiguration.setCustomerServerRestLimitsConfiguration(rateLimit); profileConfiguration.setWsUpdatesPerSessionRateLimit(rateLimit); profileConfiguration.setCassandraQueryTenantRateLimitsConfiguration(rateLimit); + profileConfiguration.setEdgeEventRateLimits(rateLimit); + profileConfiguration.setEdgeEventRateLimitsPerEdge(rateLimit); updateTenantProfileConfiguration(profileConfiguration); for (LimitedApi limitedApi : List.of( @@ -76,7 +78,9 @@ public class RateLimitServiceTest { LimitedApi.ENTITY_IMPORT, LimitedApi.NOTIFICATION_REQUESTS, LimitedApi.REST_REQUESTS_PER_CUSTOMER, - LimitedApi.CASSANDRA_QUERIES + LimitedApi.CASSANDRA_QUERIES, + LimitedApi.EDGE_EVENTS, + LimitedApi.EDGE_EVENTS_PER_EDGE )) { testRateLimits(limitedApi, max, tenantId); } diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/limit/LimitedApi.java b/common/data/src/main/java/org/thingsboard/server/common/data/limit/LimitedApi.java index 5f743f626a..980eb880a0 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/limit/LimitedApi.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/limit/LimitedApi.java @@ -31,6 +31,8 @@ public enum LimitedApi { REST_REQUESTS_PER_CUSTOMER(DefaultTenantProfileConfiguration::getCustomerServerRestLimitsConfiguration, "REST API requests per customer", false), WS_UPDATES_PER_SESSION(DefaultTenantProfileConfiguration::getWsUpdatesPerSessionRateLimit, "WS updates per session", true), CASSANDRA_QUERIES(DefaultTenantProfileConfiguration::getCassandraQueryTenantRateLimitsConfiguration, "Cassandra queries", true), + EDGE_EVENTS(DefaultTenantProfileConfiguration::getEdgeEventRateLimits, "Edge events", true), + EDGE_EVENTS_PER_EDGE(DefaultTenantProfileConfiguration::getEdgeEventRateLimitsPerEdge, "Edge events per edge", false), PASSWORD_RESET(false, true), TWO_FA_VERIFICATION_CODE_SEND(false, true), TWO_FA_VERIFICATION_CODE_CHECK(false, true), diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/notification/NotificationType.java b/common/data/src/main/java/org/thingsboard/server/common/data/notification/NotificationType.java index 8eb4451e3b..cfc91d8fca 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/notification/NotificationType.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/notification/NotificationType.java @@ -28,6 +28,7 @@ public enum NotificationType { ENTITIES_LIMIT, API_USAGE_LIMIT, RULE_NODE, - RATE_LIMITS - + RATE_LIMITS, + EDGE_CONNECTION, + EDGE_COMMUNICATION_FAILURE } diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/notification/info/EdgeCommunicationFailureNotificationInfo.java b/common/data/src/main/java/org/thingsboard/server/common/data/notification/info/EdgeCommunicationFailureNotificationInfo.java new file mode 100644 index 0000000000..6d0f2ae5f5 --- /dev/null +++ b/common/data/src/main/java/org/thingsboard/server/common/data/notification/info/EdgeCommunicationFailureNotificationInfo.java @@ -0,0 +1,66 @@ +/** + * Copyright © 2016-2024 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.common.data.notification.info; + +import lombok.AllArgsConstructor; +import lombok.Builder; +import lombok.Data; +import lombok.NoArgsConstructor; +import org.thingsboard.server.common.data.id.CustomerId; +import org.thingsboard.server.common.data.id.EdgeId; +import org.thingsboard.server.common.data.id.EntityId; +import org.thingsboard.server.common.data.id.TenantId; + +import java.util.Map; + +import static org.thingsboard.server.common.data.util.CollectionsUtil.mapOf; + +@Data +@NoArgsConstructor +@AllArgsConstructor +@Builder +public class EdgeCommunicationFailureNotificationInfo implements RuleOriginatedNotificationInfo { + + private TenantId tenantId; + private CustomerId customerId; + private EdgeId edgeId; + private String edgeName; + private String failureMsg; + + @Override + public Map getTemplateData() { + return mapOf( + "edgeId", edgeId.toString(), + "edgeName", edgeName, + "failureMsg", failureMsg + ); + } + + @Override + public TenantId getAffectedTenantId() { + return tenantId; + } + + @Override + public CustomerId getAffectedCustomerId() { + return customerId; + } + + @Override + public EntityId getStateEntityId() { + return edgeId; + } +} diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/notification/info/EdgeConnectionNotificationInfo.java b/common/data/src/main/java/org/thingsboard/server/common/data/notification/info/EdgeConnectionNotificationInfo.java new file mode 100644 index 0000000000..62b1370566 --- /dev/null +++ b/common/data/src/main/java/org/thingsboard/server/common/data/notification/info/EdgeConnectionNotificationInfo.java @@ -0,0 +1,66 @@ +/** + * Copyright © 2016-2024 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.common.data.notification.info; + +import lombok.AllArgsConstructor; +import lombok.Builder; +import lombok.Data; +import lombok.NoArgsConstructor; +import org.thingsboard.server.common.data.id.CustomerId; +import org.thingsboard.server.common.data.id.EdgeId; +import org.thingsboard.server.common.data.id.EntityId; +import org.thingsboard.server.common.data.id.TenantId; + +import java.util.Map; + +import static org.thingsboard.server.common.data.util.CollectionsUtil.mapOf; + +@Data +@NoArgsConstructor +@AllArgsConstructor +@Builder +public class EdgeConnectionNotificationInfo implements RuleOriginatedNotificationInfo { + + private String eventType; + private TenantId tenantId; + private CustomerId customerId; + private EdgeId edgeId; + private String edgeName; + + @Override + public Map getTemplateData() { + return mapOf( + "eventType", eventType, + "edgeId", edgeId.toString(), + "edgeName", edgeName + ); + } + + @Override + public TenantId getAffectedTenantId() { + return tenantId; + } + + @Override + public CustomerId getAffectedCustomerId() { + return customerId; + } + + @Override + public EntityId getStateEntityId() { + return edgeId; + } +} diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/notification/rule/trigger/EdgeCommunicationFailureTrigger.java b/common/data/src/main/java/org/thingsboard/server/common/data/notification/rule/trigger/EdgeCommunicationFailureTrigger.java new file mode 100644 index 0000000000..6212c22a0d --- /dev/null +++ b/common/data/src/main/java/org/thingsboard/server/common/data/notification/rule/trigger/EdgeCommunicationFailureTrigger.java @@ -0,0 +1,63 @@ +/** + * Copyright © 2016-2024 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.common.data.notification.rule.trigger; + +import lombok.Builder; +import lombok.Data; +import org.thingsboard.server.common.data.id.CustomerId; +import org.thingsboard.server.common.data.id.EdgeId; +import org.thingsboard.server.common.data.id.EntityId; +import org.thingsboard.server.common.data.id.TenantId; +import org.thingsboard.server.common.data.notification.rule.trigger.config.NotificationRuleTriggerType; + +import java.util.concurrent.TimeUnit; + +@Data +@Builder +public class EdgeCommunicationFailureTrigger implements NotificationRuleTrigger { + + private final TenantId tenantId; + private final CustomerId customerId; + private final EdgeId edgeId; + private final String edgeName; + private final String failureMsg; + private final String error; + + @Override + public boolean deduplicate() { + return true; + } + + @Override + public String getDeduplicationKey() { + return String.join(":", NotificationRuleTrigger.super.getDeduplicationKey(), error); + } + + @Override + public long getDefaultDeduplicationDuration() { + return TimeUnit.MINUTES.toMillis(30); + } + + @Override + public NotificationRuleTriggerType getType() { + return NotificationRuleTriggerType.EDGE_COMMUNICATION_FAILURE; + } + + @Override + public EntityId getOriginatorEntityId() { + return edgeId; + } +} diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/notification/rule/trigger/EdgeConnectionTrigger.java b/common/data/src/main/java/org/thingsboard/server/common/data/notification/rule/trigger/EdgeConnectionTrigger.java new file mode 100644 index 0000000000..766338dba6 --- /dev/null +++ b/common/data/src/main/java/org/thingsboard/server/common/data/notification/rule/trigger/EdgeConnectionTrigger.java @@ -0,0 +1,62 @@ +/** + * Copyright © 2016-2024 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.common.data.notification.rule.trigger; + +import lombok.Builder; +import lombok.Data; +import org.thingsboard.server.common.data.id.CustomerId; +import org.thingsboard.server.common.data.id.EdgeId; +import org.thingsboard.server.common.data.id.EntityId; +import org.thingsboard.server.common.data.id.TenantId; +import org.thingsboard.server.common.data.notification.rule.trigger.config.NotificationRuleTriggerType; + +import java.util.concurrent.TimeUnit; + +@Data +@Builder +public class EdgeConnectionTrigger implements NotificationRuleTrigger { + + private final TenantId tenantId; + private final CustomerId customerId; + private final EdgeId edgeId; + private final boolean connected; + private final String edgeName; + + @Override + public boolean deduplicate() { + return true; + } + + @Override + public String getDeduplicationKey() { + return String.join(":", NotificationRuleTrigger.super.getDeduplicationKey(), String.valueOf(connected)); + } + + @Override + public long getDefaultDeduplicationDuration() { + return TimeUnit.MINUTES.toMillis(1); + } + + @Override + public NotificationRuleTriggerType getType() { + return NotificationRuleTriggerType.EDGE_CONNECTION; + } + + @Override + public EntityId getOriginatorEntityId() { + return edgeId; + } +} diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/notification/rule/trigger/config/EdgeCommunicationFailureNotificationRuleTriggerConfig.java b/common/data/src/main/java/org/thingsboard/server/common/data/notification/rule/trigger/config/EdgeCommunicationFailureNotificationRuleTriggerConfig.java new file mode 100644 index 0000000000..595b0fa7d5 --- /dev/null +++ b/common/data/src/main/java/org/thingsboard/server/common/data/notification/rule/trigger/config/EdgeCommunicationFailureNotificationRuleTriggerConfig.java @@ -0,0 +1,39 @@ +/** + * Copyright © 2016-2024 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.common.data.notification.rule.trigger.config; + +import lombok.AllArgsConstructor; +import lombok.Builder; +import lombok.Data; +import lombok.NoArgsConstructor; + +import java.util.Set; +import java.util.UUID; + +@Data +@NoArgsConstructor +@AllArgsConstructor +@Builder +public class EdgeCommunicationFailureNotificationRuleTriggerConfig implements NotificationRuleTriggerConfig { + + private Set edges; // if empty - all edges + + @Override + public NotificationRuleTriggerType getTriggerType() { + return NotificationRuleTriggerType.EDGE_COMMUNICATION_FAILURE; + } + +} diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/notification/rule/trigger/config/EdgeConnectionNotificationRuleTriggerConfig.java b/common/data/src/main/java/org/thingsboard/server/common/data/notification/rule/trigger/config/EdgeConnectionNotificationRuleTriggerConfig.java new file mode 100644 index 0000000000..8c0905cc59 --- /dev/null +++ b/common/data/src/main/java/org/thingsboard/server/common/data/notification/rule/trigger/config/EdgeConnectionNotificationRuleTriggerConfig.java @@ -0,0 +1,44 @@ +/** + * Copyright © 2016-2024 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.common.data.notification.rule.trigger.config; + +import lombok.AllArgsConstructor; +import lombok.Builder; +import lombok.Data; +import lombok.NoArgsConstructor; + +import java.util.Set; +import java.util.UUID; + +@Data +@NoArgsConstructor +@AllArgsConstructor +@Builder +public class EdgeConnectionNotificationRuleTriggerConfig implements NotificationRuleTriggerConfig { + + private Set edges; // if empty - all edges + private Set notifyOn; + + @Override + public NotificationRuleTriggerType getTriggerType() { + return NotificationRuleTriggerType.EDGE_CONNECTION; + } + + public enum EdgeConnectivityEvent { + CONNECTED, DISCONNECTED + } + +} diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/notification/rule/trigger/config/NotificationRuleTriggerConfig.java b/common/data/src/main/java/org/thingsboard/server/common/data/notification/rule/trigger/config/NotificationRuleTriggerConfig.java index 5ffb8a5fe1..15a5e59255 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/notification/rule/trigger/config/NotificationRuleTriggerConfig.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/notification/rule/trigger/config/NotificationRuleTriggerConfig.java @@ -36,6 +36,8 @@ import java.io.Serializable; @Type(value = EntitiesLimitNotificationRuleTriggerConfig.class, name = "ENTITIES_LIMIT"), @Type(value = ApiUsageLimitNotificationRuleTriggerConfig.class, name = "API_USAGE_LIMIT"), @Type(value = RateLimitsNotificationRuleTriggerConfig.class, name = "RATE_LIMITS"), + @Type(value = EdgeConnectionNotificationRuleTriggerConfig.class, name = "EDGE_CONNECTION"), + @Type(value = EdgeCommunicationFailureNotificationRuleTriggerConfig.class, name = "EDGE_COMMUNICATION_FAILURE"), }) public interface NotificationRuleTriggerConfig extends Serializable { diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/notification/rule/trigger/config/NotificationRuleTriggerType.java b/common/data/src/main/java/org/thingsboard/server/common/data/notification/rule/trigger/config/NotificationRuleTriggerType.java index 6eff7a8114..8469fac752 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/notification/rule/trigger/config/NotificationRuleTriggerType.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/notification/rule/trigger/config/NotificationRuleTriggerType.java @@ -26,6 +26,8 @@ public enum NotificationRuleTriggerType { ALARM_ASSIGNMENT, DEVICE_ACTIVITY, RULE_ENGINE_COMPONENT_LIFECYCLE_EVENT, + EDGE_CONNECTION, + EDGE_COMMUNICATION_FAILURE, NEW_PLATFORM_VERSION(false), ENTITIES_LIMIT(false), API_USAGE_LIMIT(false), diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/tenant/profile/DefaultTenantProfileConfiguration.java b/common/data/src/main/java/org/thingsboard/server/common/data/tenant/profile/DefaultTenantProfileConfiguration.java index 1a4df035f5..3fe110de0b 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/tenant/profile/DefaultTenantProfileConfiguration.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/tenant/profile/DefaultTenantProfileConfiguration.java @@ -81,6 +81,9 @@ public class DefaultTenantProfileConfiguration implements TenantProfileConfigura private String cassandraQueryTenantRateLimitsConfiguration; + private String edgeEventRateLimits; + private String edgeEventRateLimitsPerEdge; + private int defaultStorageTtlDays; private int alarmsTtlDays; private int rpcTtlDays; diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/util/TemplateUtils.java b/common/data/src/main/java/org/thingsboard/server/common/data/util/TemplateUtils.java index 3bc2aac1f8..a6b1c11d42 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/util/TemplateUtils.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/util/TemplateUtils.java @@ -19,6 +19,7 @@ import org.apache.commons.lang3.StringUtils; import java.util.Map; import java.util.function.UnaryOperator; +import java.util.regex.Matcher; import java.util.regex.Pattern; import static com.google.common.base.Strings.nullToEmpty; @@ -49,7 +50,7 @@ public class TemplateUtils { value = FUNCTIONS.get(function).apply(value); } } - return value; + return Matcher.quoteReplacement(value); }); } diff --git a/common/transport/snmp/src/main/java/org/thingsboard/server/transport/snmp/SnmpTransportContext.java b/common/transport/snmp/src/main/java/org/thingsboard/server/transport/snmp/SnmpTransportContext.java index a59013bbfe..6e30b06a2c 100644 --- a/common/transport/snmp/src/main/java/org/thingsboard/server/transport/snmp/SnmpTransportContext.java +++ b/common/transport/snmp/src/main/java/org/thingsboard/server/transport/snmp/SnmpTransportContext.java @@ -152,12 +152,14 @@ public class SnmpTransportContext extends TransportContext { try { if (!newProfileTransportConfiguration.equals(sessionContext.getProfileTransportConfiguration())) { sessionContext.setProfileTransportConfiguration(newProfileTransportConfiguration); + sessionContext.setDevice(device); sessionContext.initializeTarget(newProfileTransportConfiguration, newDeviceTransportConfiguration); snmpTransportService.cancelQueryingTasks(sessionContext); snmpTransportService.createQueryingTasks(sessionContext); transportService.lifecycleEvent(sessionContext.getTenantId(), sessionContext.getDeviceId(), ComponentLifecycleEvent.UPDATED, true, null); } else if (!newDeviceTransportConfiguration.equals(sessionContext.getDeviceTransportConfiguration())) { sessionContext.setDeviceTransportConfiguration(newDeviceTransportConfiguration); + sessionContext.setDevice(device); sessionContext.initializeTarget(newProfileTransportConfiguration, newDeviceTransportConfiguration); transportService.lifecycleEvent(sessionContext.getTenantId(), sessionContext.getDeviceId(), ComponentLifecycleEvent.UPDATED, true, null); } else { diff --git a/common/transport/snmp/src/main/java/org/thingsboard/server/transport/snmp/service/SnmpTransportService.java b/common/transport/snmp/src/main/java/org/thingsboard/server/transport/snmp/service/SnmpTransportService.java index 3a8811df97..3dc293b70b 100644 --- a/common/transport/snmp/src/main/java/org/thingsboard/server/transport/snmp/service/SnmpTransportService.java +++ b/common/transport/snmp/src/main/java/org/thingsboard/server/transport/snmp/service/SnmpTransportService.java @@ -15,6 +15,11 @@ */ package org.thingsboard.server.transport.snmp.service; +import com.google.common.util.concurrent.Futures; +import com.google.common.util.concurrent.ListenableFuture; +import com.google.common.util.concurrent.ListenableScheduledFuture; +import com.google.common.util.concurrent.ListeningScheduledExecutorService; +import com.google.common.util.concurrent.MoreExecutors; import com.google.gson.JsonElement; import com.google.gson.JsonObject; import lombok.Builder; @@ -32,7 +37,7 @@ import org.snmp4j.mp.MPv3; import org.snmp4j.security.SecurityModels; import org.snmp4j.security.SecurityProtocols; import org.snmp4j.security.USM; -import org.snmp4j.smi.Address; +import org.snmp4j.smi.IpAddress; import org.snmp4j.smi.OctetString; import org.snmp4j.smi.TcpAddress; import org.snmp4j.smi.UdpAddress; @@ -44,6 +49,7 @@ import org.springframework.context.annotation.Lazy; import org.springframework.stereotype.Service; import org.thingsboard.common.util.ThingsBoardExecutors; import org.thingsboard.common.util.ThingsBoardThreadFactory; +import org.thingsboard.server.common.adaptor.JsonConverter; import org.thingsboard.server.common.data.DataConstants; import org.thingsboard.server.common.data.TbTransportService; import org.thingsboard.server.common.data.kv.DataType; @@ -53,11 +59,11 @@ import org.thingsboard.server.common.data.transport.snmp.SnmpMethod; import org.thingsboard.server.common.data.transport.snmp.config.RepeatingQueryingSnmpCommunicationConfig; import org.thingsboard.server.common.data.transport.snmp.config.SnmpCommunicationConfig; import org.thingsboard.server.common.transport.TransportService; -import org.thingsboard.server.common.adaptor.JsonConverter; import org.thingsboard.server.gen.transport.TransportProtos; import org.thingsboard.server.queue.util.TbSnmpTransportComponent; import org.thingsboard.server.transport.snmp.SnmpTransportContext; import org.thingsboard.server.transport.snmp.session.DeviceSessionContext; +import org.thingsboard.server.transport.snmp.session.ScheduledTask; import javax.annotation.PostConstruct; import javax.annotation.PreDestroy; @@ -71,8 +77,6 @@ import java.util.Map; import java.util.Optional; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; -import java.util.concurrent.ScheduledExecutorService; -import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; import java.util.stream.Collectors; @@ -80,6 +84,7 @@ import java.util.stream.Collectors; @Service @Slf4j @RequiredArgsConstructor +@SuppressWarnings("UnstableApiUsage") public class SnmpTransportService implements TbTransportService, CommandResponder { private final TransportService transportService; private final PduService pduService; @@ -88,23 +93,27 @@ public class SnmpTransportService implements TbTransportService, CommandResponde @Getter private Snmp snmp; - private ScheduledExecutorService queryingExecutor; - private ExecutorService responseProcessingExecutor; + private ListeningScheduledExecutorService scheduler; + private ExecutorService executor; private final Map responseDataMappers = new EnumMap<>(SnmpCommunicationSpec.class); private final Map responseProcessors = new EnumMap<>(SnmpCommunicationSpec.class); @Value("${transport.snmp.bind_port:1620}") private Integer snmpBindPort; - @Value("${transport.snmp.response_processing.parallelism_level}") - private Integer responseProcessingParallelismLevel; + @Value("${transport.snmp.response_processing.parallelism_level:4}") + private int responseProcessingThreadPoolSize; + @Value("${transport.snmp.scheduler_thread_pool_size:4}") + private int schedulerThreadPoolSize; @Value("${transport.snmp.underlying_protocol}") private String snmpUnderlyingProtocol; + @Value("${transport.snmp.request_chunk_delay_ms:100}") + private int requestChunkDelayMs; @PostConstruct private void init() throws IOException { - queryingExecutor = Executors.newScheduledThreadPool(Runtime.getRuntime().availableProcessors(), ThingsBoardThreadFactory.forName("snmp-querying")); - responseProcessingExecutor = ThingsBoardExecutors.newWorkStealingPool(responseProcessingParallelismLevel, "snmp-response-processing"); + scheduler = MoreExecutors.listeningDecorator(Executors.newScheduledThreadPool(schedulerThreadPoolSize, ThingsBoardThreadFactory.forName("snmp-querying"))); + executor = ThingsBoardExecutors.newWorkStealingPool(responseProcessingThreadPoolSize, "snmp-response-processing"); initializeSnmp(); configureResponseDataMappers(); @@ -115,11 +124,11 @@ public class SnmpTransportService implements TbTransportService, CommandResponde @PreDestroy public void stop() { - if (queryingExecutor != null) { - queryingExecutor.shutdownNow(); + if (scheduler != null) { + scheduler.shutdownNow(); } - if (responseProcessingExecutor != null) { - responseProcessingExecutor.shutdownNow(); + if (executor != null) { + executor.shutdownNow(); } } @@ -144,38 +153,39 @@ public class SnmpTransportService implements TbTransportService, CommandResponde } public void createQueryingTasks(DeviceSessionContext sessionContext) { - List> queryingTasks = sessionContext.getProfileTransportConfiguration().getCommunicationConfigs().stream() + sessionContext.getProfileTransportConfiguration().getCommunicationConfigs().stream() .filter(communicationConfig -> communicationConfig instanceof RepeatingQueryingSnmpCommunicationConfig) - .map(config -> { + .forEach(config -> { RepeatingQueryingSnmpCommunicationConfig repeatingCommunicationConfig = (RepeatingQueryingSnmpCommunicationConfig) config; Long queryingFrequency = repeatingCommunicationConfig.getQueryingFrequencyMs(); - return queryingExecutor.scheduleWithFixedDelay(() -> { + ScheduledTask scheduledTask = new ScheduledTask(); + scheduledTask.init(() -> { try { if (sessionContext.isActive()) { - sendRequest(sessionContext, repeatingCommunicationConfig); + return sendRequest(sessionContext, repeatingCommunicationConfig); } } catch (Exception e) { log.error("Failed to send SNMP request for device {}: {}", sessionContext.getDeviceId(), e.toString()); transportService.errorEvent(sessionContext.getTenantId(), sessionContext.getDeviceId(), config.getSpec().getLabel(), e); } - }, queryingFrequency, queryingFrequency, TimeUnit.MILLISECONDS); - }) - .collect(Collectors.toList()); - sessionContext.getQueryingTasks().addAll(queryingTasks); + return Futures.immediateVoidFuture(); + }, queryingFrequency, scheduler); + sessionContext.getQueryingTasks().add(scheduledTask); + }); } public void cancelQueryingTasks(DeviceSessionContext sessionContext) { - sessionContext.getQueryingTasks().forEach(task -> task.cancel(true)); + sessionContext.getQueryingTasks().forEach(ScheduledTask::cancel); sessionContext.getQueryingTasks().clear(); } - private void sendRequest(DeviceSessionContext sessionContext, SnmpCommunicationConfig communicationConfig) { - sendRequest(sessionContext, communicationConfig, Collections.emptyMap()); + private ListenableFuture sendRequest(DeviceSessionContext sessionContext, SnmpCommunicationConfig communicationConfig) { + return sendRequest(sessionContext, communicationConfig, Collections.emptyMap()); } - private void sendRequest(DeviceSessionContext sessionContext, SnmpCommunicationConfig communicationConfig, Map values) { + private ListenableFuture sendRequest(DeviceSessionContext sessionContext, SnmpCommunicationConfig communicationConfig, Map values) { List request = pduService.createPdus(sessionContext, communicationConfig, values); RequestContext requestContext = RequestContext.builder() .communicationSpec(communicationConfig.getSpec()) @@ -183,19 +193,40 @@ public class SnmpTransportService implements TbTransportService, CommandResponde .responseMappings(communicationConfig.getAllMappings()) .requestSize(request.size()) .build(); - sendRequest(sessionContext, request, requestContext); + return sendRequest(sessionContext, request, requestContext); } - private void sendRequest(DeviceSessionContext sessionContext, List request, RequestContext requestContext) { - for (PDU pdu : request) { - log.debug("Executing SNMP request for device {} with {} variable bindings", sessionContext.getDeviceId(), pdu.size()); - try { - snmp.send(pdu, sessionContext.getTarget(), requestContext, sessionContext); - } catch (IOException e) { - log.error("Failed to send SNMP request to device {}: {}", sessionContext.getDeviceId(), e.toString()); - transportService.errorEvent(sessionContext.getTenantId(), sessionContext.getDeviceId(), requestContext.getCommunicationSpec().getLabel(), e); + private ListenableFuture sendRequest(DeviceSessionContext sessionContext, List request, RequestContext requestContext) { + if (request.size() <= 1 || requestChunkDelayMs == 0) { + for (PDU pdu : request) { + sendPdu(pdu, requestContext, sessionContext); + } + return Futures.immediateVoidFuture(); + } + + List> futures = new ArrayList<>(); + for (int i = 0, delay = 0; i < request.size(); i++, delay += requestChunkDelayMs) { + PDU pdu = request.get(i); + if (delay == 0) { + sendPdu(pdu, requestContext, sessionContext); + } else { + ListenableScheduledFuture future = scheduler.schedule(() -> { + sendPdu(pdu, requestContext, sessionContext); + }, delay, TimeUnit.MILLISECONDS); + futures.add(future); } } + return Futures.whenAllComplete(futures).call(() -> null, MoreExecutors.directExecutor()); + } + + private void sendPdu(PDU pdu, RequestContext requestContext, DeviceSessionContext sessionContext) { + log.debug("[{}] Sending SNMP request with {} variable bindings to {}", sessionContext.getDeviceId(), pdu.size(), sessionContext.getTarget().getAddress()); + try { + snmp.send(pdu, sessionContext.getTarget(), requestContext, sessionContext); + } catch (Exception e) { + log.error("[{}] Failed to send SNMP request", sessionContext.getDeviceId(), e); + transportService.errorEvent(sessionContext.getTenantId(), sessionContext.getDeviceId(), requestContext.getCommunicationSpec().getLabel(), e); + } } public void onAttributeUpdate(DeviceSessionContext sessionContext, TransportProtos.AttributeUpdateNotificationMsg attributeUpdateNotification) { @@ -251,21 +282,19 @@ public class SnmpTransportService implements TbTransportService, CommandResponde ((Snmp) event.getSource()).cancel(event.getRequest(), sessionContext); RequestContext requestContext = (RequestContext) event.getUserObject(); if (event.getError() != null) { - log.warn("SNMP response error: {}", event.getError().toString()); + log.warn("[{}] SNMP response error: {}", sessionContext.getDeviceId(), event.getError().toString()); transportService.errorEvent(sessionContext.getTenantId(), sessionContext.getDeviceId(), requestContext.getCommunicationSpec().getLabel(), new RuntimeException(event.getError())); return; } PDU responsePdu = event.getResponse(); - if (log.isTraceEnabled()) { - log.trace("Received PDU for device {}: {}", sessionContext.getDeviceId(), responsePdu); - } + log.trace("[{}] Received PDU: {}", sessionContext.getDeviceId(), responsePdu); List response; if (requestContext.getRequestSize() == 1) { if (responsePdu == null) { - log.debug("No response from SNMP device {}, requestId: {}", sessionContext.getDeviceId(), event.getRequest().getRequestID()); if (requestContext.getMethod() == SnmpMethod.GET) { + log.debug("[{}][{}] Empty response from device", sessionContext.getDeviceId(), event.getRequest().getRequestID()); transportService.errorEvent(sessionContext.getTenantId(), sessionContext.getDeviceId(), requestContext.getCommunicationSpec().getLabel(), new RuntimeException("No response from device")); } return; @@ -281,14 +310,14 @@ public class SnmpTransportService implements TbTransportService, CommandResponde response.add(responsePart); } } - log.debug("All response parts are collected for request to device {}", sessionContext.getDeviceId()); + log.debug("[{}] All {} response parts are collected for request", sessionContext.getDeviceId(), responseParts.size()); } else { - log.trace("Awaiting other response parts for request to device {}", sessionContext.getDeviceId()); + log.trace("[{}] Awaiting other response parts for request", sessionContext.getDeviceId()); return; } } - responseProcessingExecutor.execute(() -> { + executor.execute(() -> { try { processResponse(sessionContext, response, requestContext); } catch (Exception e) { @@ -298,24 +327,31 @@ public class SnmpTransportService implements TbTransportService, CommandResponde } /* - * SNMP notifications handler - * - * TODO: add check for host uniqueness when saving device (for backward compatibility - only for the ones using from-device RPC requests) - * - * NOTE: SNMP TRAPs support won't work properly when there is more than one SNMP transport, - * due to load-balancing of requests from devices: session might not be on this instance - * */ + * SNMP notifications handler + * + * TODO: add check for host uniqueness when saving device (for backward compatibility - only for the ones using from-device RPC requests) + * + * NOTE: SNMP TRAPs support won't work properly when there is more than one SNMP transport, + * due to load-balancing of requests from devices: session might not be on this instance + * */ @Override public void processPdu(CommandResponderEvent event) { - Address sourceAddress = event.getPeerAddress(); - DeviceSessionContext sessionContext = transportContext.getSessions().stream() - .filter(session -> session.getTarget().getAddress().equals(sourceAddress)) - .findFirst().orElse(null); - if (sessionContext == null) { - log.warn("SNMP TRAP processing failed: couldn't find device session for address {}", sourceAddress); + IpAddress sourceAddress = (IpAddress) event.getPeerAddress(); + List sessions = transportContext.getSessions().stream() + .filter(session -> ((IpAddress) session.getTarget().getAddress()).getInetAddress().equals(sourceAddress.getInetAddress())) + .collect(Collectors.toList()); + if (sessions.isEmpty()) { + log.warn("Couldn't find device session for SNMP TRAP for address {}", sourceAddress); + return; + } else if (sessions.size() > 1) { + for (DeviceSessionContext sessionContext : sessions) { + transportService.errorEvent(sessionContext.getTenantId(), sessionContext.getDeviceId(), SnmpCommunicationSpec.TO_SERVER_RPC_REQUEST.getLabel(), + new IllegalStateException("Found multiple devices for host " + sourceAddress.getInetAddress().getHostAddress())); + } return; } + DeviceSessionContext sessionContext = sessions.get(0); try { processIncomingTrap(sessionContext, event); } catch (Throwable e) { @@ -327,11 +363,11 @@ public class SnmpTransportService implements TbTransportService, CommandResponde private void processIncomingTrap(DeviceSessionContext sessionContext, CommandResponderEvent event) { PDU pdu = event.getPDU(); if (pdu == null) { - log.warn("Got empty trap from device {}", sessionContext.getDeviceId()); + log.warn("[{}] Received empty SNMP trap", sessionContext.getDeviceId()); throw new IllegalArgumentException("Received TRAP with no data"); } - log.debug("Processing SNMP trap from device {} (PDU: {}}", sessionContext.getDeviceId(), pdu); + log.debug("[{}] Processing SNMP trap: {}", sessionContext.getDeviceId(), pdu); SnmpCommunicationConfig communicationConfig = sessionContext.getProfileTransportConfiguration().getCommunicationConfigs().stream() .filter(config -> config.getSpec() == SnmpCommunicationSpec.TO_SERVER_RPC_REQUEST).findFirst() .orElseThrow(() -> new IllegalArgumentException("No config found for to-server RPC requests")); @@ -341,7 +377,7 @@ public class SnmpTransportService implements TbTransportService, CommandResponde .method(SnmpMethod.TRAP) .build(); - responseProcessingExecutor.execute(() -> { + executor.execute(() -> { processResponse(sessionContext, List.of(pdu), requestContext); }); } @@ -352,7 +388,7 @@ public class SnmpTransportService implements TbTransportService, CommandResponde JsonObject responseData = responseDataMappers.get(requestContext.getCommunicationSpec()).map(response, requestContext); if (responseData.size() == 0) { - log.warn("No values in the SNMP response for device {}", sessionContext.getDeviceId()); + log.warn("[{}] No values in the response", sessionContext.getDeviceId()); throw new IllegalArgumentException("No values in the response"); } @@ -428,11 +464,11 @@ public class SnmpTransportService implements TbTransportService, CommandResponde @PreDestroy public void shutdown() { log.info("Stopping SNMP transport!"); - if (queryingExecutor != null) { - queryingExecutor.shutdownNow(); + if (scheduler != null) { + scheduler.shutdownNow(); } - if (responseProcessingExecutor != null) { - responseProcessingExecutor.shutdownNow(); + if (executor != null) { + executor.shutdownNow(); } if (snmp != null) { try { diff --git a/common/transport/snmp/src/main/java/org/thingsboard/server/transport/snmp/session/DeviceSessionContext.java b/common/transport/snmp/src/main/java/org/thingsboard/server/transport/snmp/session/DeviceSessionContext.java index 5819c76f74..8ec0c6841d 100644 --- a/common/transport/snmp/src/main/java/org/thingsboard/server/transport/snmp/session/DeviceSessionContext.java +++ b/common/transport/snmp/src/main/java/org/thingsboard/server/transport/snmp/session/DeviceSessionContext.java @@ -45,7 +45,6 @@ import org.thingsboard.server.transport.snmp.SnmpTransportContext; import java.util.LinkedList; import java.util.List; import java.util.UUID; -import java.util.concurrent.ScheduledFuture; import java.util.concurrent.atomic.AtomicInteger; @Slf4j @@ -60,7 +59,8 @@ public class DeviceSessionContext extends DeviceAwareSessionContext implements S @Setter private SnmpDeviceTransportConfiguration deviceTransportConfiguration; @Getter - private final Device device; + @Setter + private Device device; @Getter private final TenantId tenantId; @@ -73,7 +73,7 @@ public class DeviceSessionContext extends DeviceAwareSessionContext implements S private Runnable sessionTimeoutHandler; @Getter - private final List> queryingTasks = new LinkedList<>(); + private final List queryingTasks = new LinkedList<>(); @Builder public DeviceSessionContext(TenantId tenantId, Device device, DeviceProfile deviceProfile, String token, diff --git a/common/transport/snmp/src/main/java/org/thingsboard/server/transport/snmp/session/ScheduledTask.java b/common/transport/snmp/src/main/java/org/thingsboard/server/transport/snmp/session/ScheduledTask.java new file mode 100644 index 0000000000..43d83e5e2c --- /dev/null +++ b/common/transport/snmp/src/main/java/org/thingsboard/server/transport/snmp/session/ScheduledTask.java @@ -0,0 +1,62 @@ +/** + * Copyright © 2016-2024 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.transport.snmp.session; + +import com.google.common.util.concurrent.AsyncCallable; +import com.google.common.util.concurrent.Futures; +import com.google.common.util.concurrent.ListenableFuture; +import com.google.common.util.concurrent.MoreExecutors; +import lombok.Data; +import lombok.extern.slf4j.Slf4j; + +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; + +@Data +@Slf4j +public class ScheduledTask { + private ListenableFuture scheduledFuture; + private boolean stopped = false; + + public void init(AsyncCallable task, long delayMs, ScheduledExecutorService scheduler) { + schedule(task, delayMs, scheduler); + } + + private void schedule(AsyncCallable task, long delayMs, ScheduledExecutorService scheduler) { + scheduledFuture = Futures.scheduleAsync(() -> { + if (stopped) { + return Futures.immediateCancelledFuture(); + } + try { + return task.call(); + } catch (Throwable t) { + log.error("Unhandled error in scheduled task", t); + return Futures.immediateFailedFuture(t); + } + }, delayMs, TimeUnit.MILLISECONDS, scheduler); + if (!stopped) { + scheduledFuture.addListener(() -> schedule(task, delayMs, scheduler), MoreExecutors.directExecutor()); + } + } + + public void cancel() { + stopped = true; + if (scheduledFuture != null) { + scheduledFuture.cancel(true); + } + } + +} diff --git a/common/transport/snmp/src/test/java/org/thingsboard/server/transport/snmp/SnmpDeviceSimulatorV2.java b/common/transport/snmp/src/test/java/org/thingsboard/server/transport/snmp/SnmpDeviceSimulatorV2.java index 261e67bac8..ced8f21288 100644 --- a/common/transport/snmp/src/test/java/org/thingsboard/server/transport/snmp/SnmpDeviceSimulatorV2.java +++ b/common/transport/snmp/src/test/java/org/thingsboard/server/transport/snmp/SnmpDeviceSimulatorV2.java @@ -58,11 +58,10 @@ public class SnmpDeviceSimulatorV2 extends BaseAgent { private final Target target; private final Address address; + private final Map mappings; private Snmp snmp; - private final String password; - - public SnmpDeviceSimulatorV2(int port, String password) throws IOException { + public SnmpDeviceSimulatorV2(int port, String password, Map mappings) throws IOException { super(new File("conf.agent"), new File("bootCounter.agent"), new CommandProcessor(new OctetString("12312"))); CommunityTarget target = new CommunityTarget(); target.setCommunity(new OctetString(password)); @@ -72,7 +71,7 @@ public class SnmpDeviceSimulatorV2 extends BaseAgent { target.setTimeout(1500); target.setVersion(SnmpConstants.version2c); this.target = target; - this.password = password; + this.mappings = mappings; } public void start() throws IOException { @@ -85,13 +84,6 @@ public class SnmpDeviceSimulatorV2 extends BaseAgent { snmp = new Snmp(transportMappings[0]); } - public void setUpMappings(Map oidToResponseMappings) { - unregisterManagedObject(getSnmpv2MIB()); - oidToResponseMappings.forEach((oid, response) -> { - registerManagedObject(new MOScalar<>(new OID(oid), MOAccessImpl.ACCESS_READ_WRITE, new OctetString(response))); - }); - } - public void sendTrap(String host, int port, Map values) throws IOException { PDU pdu = new PDU(); pdu.addAll(values.entrySet().stream() @@ -107,6 +99,10 @@ public class SnmpDeviceSimulatorV2 extends BaseAgent { @Override protected void registerManagedObjects() { + unregisterManagedObject(getSnmpv2MIB()); + mappings.forEach((oid, response) -> { + registerManagedObject(new MOScalar<>(new OID(oid), MOAccessImpl.ACCESS_READ_WRITE, new OctetString(response))); + }); } protected void registerManagedObject(ManagedObject mo) { @@ -152,6 +148,7 @@ public class SnmpDeviceSimulatorV2 extends BaseAgent { } protected void unregisterManagedObjects() { + unregisterManagedObject(getSnmpv2MIB()); } protected void addCommunities(SnmpCommunityMIB communityMIB) { diff --git a/common/transport/snmp/src/test/java/org/thingsboard/server/transport/snmp/SnmpTestV2.java b/common/transport/snmp/src/test/java/org/thingsboard/server/transport/snmp/SnmpTestV2.java index cb99c85655..c87507f962 100644 --- a/common/transport/snmp/src/test/java/org/thingsboard/server/transport/snmp/SnmpTestV2.java +++ b/common/transport/snmp/src/test/java/org/thingsboard/server/transport/snmp/SnmpTestV2.java @@ -15,28 +15,34 @@ */ package org.thingsboard.server.transport.snmp; +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.node.ArrayNode; +import org.thingsboard.common.util.JacksonUtil; + +import java.io.File; import java.io.IOException; -import java.util.HashMap; +import java.util.LinkedHashMap; import java.util.Map; import java.util.Scanner; +import java.util.stream.Collectors; public class SnmpTestV2 { private static final Scanner scanner = new Scanner(System.in); public static void main(String[] args) throws IOException { - SnmpDeviceSimulatorV2 client = new SnmpDeviceSimulatorV2(1610, "public"); + Map mappings = new LinkedHashMap<>(); + for (int i = 1; i <= 50; i++) { + String oid = String.format("1.3.6.1.2.1.%s.1.52", i); + mappings.put(oid, "value_" + i); + } - client.start(); - Map mappings = new HashMap<>(); -// for (int i = 1; i <= 500; i++) { -// String oid = String.format(".1.3.6.1.2.1.%s.1.52", i); -// mappings.put(oid, "value_" + i); -// } - mappings.put("1.3.6.1.2.1.266.1.52", "****"); + SnmpDeviceSimulatorV2 device = new SnmpDeviceSimulatorV2(1610, "public", mappings); + device.start(); - client.setUpMappings(mappings); - inputTraps(client); + System.out.println("Hosting the following values:\n" + mappings.entrySet().stream() + .map(entry -> entry.getKey() + " - " + entry.getValue()) + .collect(Collectors.joining("\n"))); scanner.nextLine(); } @@ -53,4 +59,18 @@ public class SnmpTestV2 { } } + private static void updateDeviceProfile(String file) throws Exception { + File profileFile = new File(file); + JsonNode deviceProfile = JacksonUtil.OBJECT_MAPPER.readTree(profileFile); + ArrayNode mappingsJson = (ArrayNode) deviceProfile.at("/profileData/transportConfiguration/communicationConfigs/0/mappings"); + for (int i = 1; i <= 50; i++) { + String oid = String.format(".1.3.6.1.2.1.%s.1.52", i); + mappingsJson.add(JacksonUtil.newObjectNode() + .put("oid", oid) + .put("key", "key_" + i) + .put("dataType", "STRING")); + } + JacksonUtil.OBJECT_MAPPER.writeValue(profileFile, deviceProfile); + } + } diff --git a/common/transport/snmp/src/test/resources/snmp-device-profile-transport-config.json b/common/transport/snmp/src/test/resources/snmp-device-profile-transport-config.json deleted file mode 100644 index f74ebca0bf..0000000000 --- a/common/transport/snmp/src/test/resources/snmp-device-profile-transport-config.json +++ /dev/null @@ -1,43 +0,0 @@ -{ - "timeoutMs": 500, - "retries": 0, - "communicationConfigs": [ - { - "spec": "TELEMETRY_QUERYING", - "queryingFrequencyMs": 3000, - "mappings": [ - { - "oid": ".1.3.6.1.2.1.1.1.50", - "key": "temperature", - "dataType": "LONG" - }, - { - "oid": ".1.3.6.1.2.1.2.1.52", - "key": "humidity", - "dataType": "DOUBLE" - } - ] - }, - { - "spec": "CLIENT_ATTRIBUTES_QUERYING", - "queryingFrequencyMs": 5000, - "mappings": [ - { - "oid": ".1.3.6.1.2.1.3.1.54", - "key": "isCool", - "dataType": "STRING" - } - ] - }, - { - "spec": "SHARED_ATTRIBUTES_SETTING", - "mappings": [ - { - "oid": ".1.3.6.1.2.1.7.1.58", - "key": "shared", - "dataType": "STRING" - } - ] - } - ] -} diff --git a/common/transport/snmp/src/test/resources/snmp-device-transport-config-v3.json b/common/transport/snmp/src/test/resources/snmp-device-transport-config-v3.json deleted file mode 100644 index 039e03fa53..0000000000 --- a/common/transport/snmp/src/test/resources/snmp-device-transport-config-v3.json +++ /dev/null @@ -1,13 +0,0 @@ -{ - "address": "192.168.3.23", - "port": 1610, - "protocolVersion": "V3", - - "username": "tb-user", - "engineId": "qwertyuioa", - "securityName": "tb-user", - "authenticationProtocol": "SHA_512", - "authenticationPassphrase": "sdfghjkloifgh", - "privacyProtocol": "DES", - "privacyPassphrase": "rtytguijokod" -} \ No newline at end of file diff --git a/common/transport/snmp/src/test/resources/snmp-device-transport-config.json b/common/transport/snmp/src/test/resources/snmp-device-transport-config.json deleted file mode 100644 index c73d817bfb..0000000000 --- a/common/transport/snmp/src/test/resources/snmp-device-transport-config.json +++ /dev/null @@ -1,6 +0,0 @@ -{ - "address": "127.0.0.1", - "port": 1610, - "community": "public", - "protocolVersion": "V2C" -} \ No newline at end of file diff --git a/common/transport/snmp/src/test/resources/snmp_device_profile.json b/common/transport/snmp/src/test/resources/snmp_device_profile.json new file mode 100644 index 0000000000..0e72c46ce8 --- /dev/null +++ b/common/transport/snmp/src/test/resources/snmp_device_profile.json @@ -0,0 +1,289 @@ +{ + "name": "SNMP Device Profile", + "description": "", + "image": null, + "type": "DEFAULT", + "transportType": "SNMP", + "provisionType": "DISABLED", + "defaultRuleChainId": null, + "defaultDashboardId": null, + "defaultQueueName": null, + "profileData": { + "configuration": { + "type": "DEFAULT" + }, + "transportConfiguration": { + "type": "SNMP", + "timeoutMs": 500, + "retries": 0, + "communicationConfigs": [ + { + "spec": "TELEMETRY_QUERYING", + "mappings": [ + { + "oid": ".1.3.6.1.2.1.1.1.52", + "key": "key_1", + "dataType": "STRING" + }, + { + "oid": ".1.3.6.1.2.1.2.1.52", + "key": "key_2", + "dataType": "STRING" + }, + { + "oid": ".1.3.6.1.2.1.3.1.52", + "key": "key_3", + "dataType": "STRING" + }, + { + "oid": ".1.3.6.1.2.1.4.1.52", + "key": "key_4", + "dataType": "STRING" + }, + { + "oid": ".1.3.6.1.2.1.5.1.52", + "key": "key_5", + "dataType": "STRING" + }, + { + "oid": ".1.3.6.1.2.1.6.1.52", + "key": "key_6", + "dataType": "STRING" + }, + { + "oid": ".1.3.6.1.2.1.7.1.52", + "key": "key_7", + "dataType": "STRING" + }, + { + "oid": ".1.3.6.1.2.1.8.1.52", + "key": "key_8", + "dataType": "STRING" + }, + { + "oid": ".1.3.6.1.2.1.9.1.52", + "key": "key_9", + "dataType": "STRING" + }, + { + "oid": ".1.3.6.1.2.1.10.1.52", + "key": "key_10", + "dataType": "STRING" + }, + { + "oid": ".1.3.6.1.2.1.11.1.52", + "key": "key_11", + "dataType": "STRING" + }, + { + "oid": ".1.3.6.1.2.1.12.1.52", + "key": "key_12", + "dataType": "STRING" + }, + { + "oid": ".1.3.6.1.2.1.13.1.52", + "key": "key_13", + "dataType": "STRING" + }, + { + "oid": ".1.3.6.1.2.1.14.1.52", + "key": "key_14", + "dataType": "STRING" + }, + { + "oid": ".1.3.6.1.2.1.15.1.52", + "key": "key_15", + "dataType": "STRING" + }, + { + "oid": ".1.3.6.1.2.1.16.1.52", + "key": "key_16", + "dataType": "STRING" + }, + { + "oid": ".1.3.6.1.2.1.17.1.52", + "key": "key_17", + "dataType": "STRING" + }, + { + "oid": ".1.3.6.1.2.1.18.1.52", + "key": "key_18", + "dataType": "STRING" + }, + { + "oid": ".1.3.6.1.2.1.19.1.52", + "key": "key_19", + "dataType": "STRING" + }, + { + "oid": ".1.3.6.1.2.1.20.1.52", + "key": "key_20", + "dataType": "STRING" + }, + { + "oid": ".1.3.6.1.2.1.21.1.52", + "key": "key_21", + "dataType": "STRING" + }, + { + "oid": ".1.3.6.1.2.1.22.1.52", + "key": "key_22", + "dataType": "STRING" + }, + { + "oid": ".1.3.6.1.2.1.23.1.52", + "key": "key_23", + "dataType": "STRING" + }, + { + "oid": ".1.3.6.1.2.1.24.1.52", + "key": "key_24", + "dataType": "STRING" + }, + { + "oid": ".1.3.6.1.2.1.25.1.52", + "key": "key_25", + "dataType": "STRING" + }, + { + "oid": ".1.3.6.1.2.1.26.1.52", + "key": "key_26", + "dataType": "STRING" + }, + { + "oid": ".1.3.6.1.2.1.27.1.52", + "key": "key_27", + "dataType": "STRING" + }, + { + "oid": ".1.3.6.1.2.1.28.1.52", + "key": "key_28", + "dataType": "STRING" + }, + { + "oid": ".1.3.6.1.2.1.29.1.52", + "key": "key_29", + "dataType": "STRING" + }, + { + "oid": ".1.3.6.1.2.1.30.1.52", + "key": "key_30", + "dataType": "STRING" + }, + { + "oid": ".1.3.6.1.2.1.31.1.52", + "key": "key_31", + "dataType": "STRING" + }, + { + "oid": ".1.3.6.1.2.1.32.1.52", + "key": "key_32", + "dataType": "STRING" + }, + { + "oid": ".1.3.6.1.2.1.33.1.52", + "key": "key_33", + "dataType": "STRING" + }, + { + "oid": ".1.3.6.1.2.1.34.1.52", + "key": "key_34", + "dataType": "STRING" + }, + { + "oid": ".1.3.6.1.2.1.35.1.52", + "key": "key_35", + "dataType": "STRING" + }, + { + "oid": ".1.3.6.1.2.1.36.1.52", + "key": "key_36", + "dataType": "STRING" + }, + { + "oid": ".1.3.6.1.2.1.37.1.52", + "key": "key_37", + "dataType": "STRING" + }, + { + "oid": ".1.3.6.1.2.1.38.1.52", + "key": "key_38", + "dataType": "STRING" + }, + { + "oid": ".1.3.6.1.2.1.39.1.52", + "key": "key_39", + "dataType": "STRING" + }, + { + "oid": ".1.3.6.1.2.1.40.1.52", + "key": "key_40", + "dataType": "STRING" + }, + { + "oid": ".1.3.6.1.2.1.41.1.52", + "key": "key_41", + "dataType": "STRING" + }, + { + "oid": ".1.3.6.1.2.1.42.1.52", + "key": "key_42", + "dataType": "STRING" + }, + { + "oid": ".1.3.6.1.2.1.43.1.52", + "key": "key_43", + "dataType": "STRING" + }, + { + "oid": ".1.3.6.1.2.1.44.1.52", + "key": "key_44", + "dataType": "STRING" + }, + { + "oid": ".1.3.6.1.2.1.45.1.52", + "key": "key_45", + "dataType": "STRING" + }, + { + "oid": ".1.3.6.1.2.1.46.1.52", + "key": "key_46", + "dataType": "STRING" + }, + { + "oid": ".1.3.6.1.2.1.47.1.52", + "key": "key_47", + "dataType": "STRING" + }, + { + "oid": ".1.3.6.1.2.1.48.1.52", + "key": "key_48", + "dataType": "STRING" + }, + { + "oid": ".1.3.6.1.2.1.49.1.52", + "key": "key_49", + "dataType": "STRING" + }, + { + "oid": ".1.3.6.1.2.1.50.1.52", + "key": "key_50", + "dataType": "STRING" + } + ], + "queryingFrequencyMs": 5000 + } + ] + }, + "provisionConfiguration": { + "type": "DISABLED", + "provisionDeviceSecret": null + }, + "alarms": null + }, + "provisionDeviceKey": null, + "firmwareId": null, + "softwareId": null, + "defaultEdgeRuleChainId": null, + "default": false +} \ No newline at end of file diff --git a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java index 0d7fbfc332..14f7bfffff 100644 --- a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java +++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java @@ -788,7 +788,11 @@ public class DefaultTransportService extends TransportActivityManager implements .setSuccess(success) .setError(error != null ? ExceptionUtils.getStackTrace(error) : "")) .build(); - sendToCore(tenantId, deviceId, msg, deviceId.getId(), TransportServiceCallback.EMPTY); + try { + sendToCore(tenantId, deviceId, msg, deviceId.getId(), TransportServiceCallback.EMPTY); + } catch (Exception e) { + log.error("[{}][{}] Failed to send lifecycle event to core", tenantId, deviceId, e); + } } @Override @@ -801,9 +805,13 @@ public class DefaultTransportService extends TransportActivityManager implements .setEntityIdLSB(deviceId.getId().getLeastSignificantBits()) .setServiceId(serviceInfoProvider.getServiceId()) .setMethod(method) - .setError(ExceptionUtils.getStackTrace(error))) + .setError(ExceptionUtils.getRootCauseMessage(error))) .build(); - sendToCore(tenantId, deviceId, msg, deviceId.getId(), TransportServiceCallback.EMPTY); + try { + sendToCore(tenantId, deviceId, msg, deviceId.getId(), TransportServiceCallback.EMPTY); + } catch (Exception e) { + log.error("[{}][{}] Failed to send error event to core", tenantId, deviceId, e); + } } @Override diff --git a/dao/src/main/java/org/thingsboard/server/dao/device/DeviceServiceImpl.java b/dao/src/main/java/org/thingsboard/server/dao/device/DeviceServiceImpl.java index f190923167..2f7fe0680c 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/device/DeviceServiceImpl.java +++ b/dao/src/main/java/org/thingsboard/server/dao/device/DeviceServiceImpl.java @@ -157,16 +157,19 @@ public class DeviceServiceImpl extends AbstractCachedEntityService deviceDao.findDeviceByTenantIdAndName(tenantId.getId(), name).orElse(null), true); } + @Transactional @Override public Device saveDeviceWithAccessToken(Device device, String accessToken) { return doSaveDevice(device, accessToken, true); } + @Transactional @Override public Device saveDevice(Device device, boolean doValidate) { return doSaveDevice(device, null, doValidate); } + @Transactional @Override public Device saveDevice(Device device) { return doSaveDevice(device, null, true); diff --git a/dao/src/main/java/org/thingsboard/server/dao/edge/BaseEdgeEventService.java b/dao/src/main/java/org/thingsboard/server/dao/edge/BaseEdgeEventService.java index 9ea70f6c5a..4d1e126c49 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/edge/BaseEdgeEventService.java +++ b/dao/src/main/java/org/thingsboard/server/dao/edge/BaseEdgeEventService.java @@ -19,11 +19,15 @@ import com.google.common.util.concurrent.ListenableFuture; import lombok.AllArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Service; +import org.thingsboard.server.cache.limits.RateLimitService; +import org.thingsboard.server.common.data.EntityType; import org.thingsboard.server.common.data.edge.EdgeEvent; import org.thingsboard.server.common.data.id.EdgeId; import org.thingsboard.server.common.data.id.TenantId; +import org.thingsboard.server.common.data.limit.LimitedApi; import org.thingsboard.server.common.data.page.PageData; import org.thingsboard.server.common.data.page.TimePageLink; +import org.thingsboard.server.common.msg.tools.TbRateLimitsException; import org.thingsboard.server.dao.service.DataValidator; @Service @@ -32,11 +36,17 @@ import org.thingsboard.server.dao.service.DataValidator; public class BaseEdgeEventService implements EdgeEventService { private final EdgeEventDao edgeEventDao; - + private final RateLimitService rateLimitService; private final DataValidator edgeEventValidator; @Override public ListenableFuture saveAsync(EdgeEvent edgeEvent) { + if (!rateLimitService.checkRateLimit(LimitedApi.EDGE_EVENTS, edgeEvent.getTenantId())) { + throw new TbRateLimitsException(EntityType.TENANT); + } + if (!rateLimitService.checkRateLimit(LimitedApi.EDGE_EVENTS_PER_EDGE, edgeEvent.getTenantId(), edgeEvent.getEdgeId())) { + throw new TbRateLimitsException(EntityType.EDGE); + } edgeEventValidator.validate(edgeEvent, EdgeEvent::getTenantId); return edgeEventDao.saveAsync(edgeEvent); } diff --git a/dao/src/main/java/org/thingsboard/server/dao/notification/DefaultNotificationSettingsService.java b/dao/src/main/java/org/thingsboard/server/dao/notification/DefaultNotificationSettingsService.java index 59787a01a7..c858eecf62 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/notification/DefaultNotificationSettingsService.java +++ b/dao/src/main/java/org/thingsboard/server/dao/notification/DefaultNotificationSettingsService.java @@ -41,6 +41,7 @@ import org.thingsboard.server.common.data.notification.targets.platform.SystemAd import org.thingsboard.server.common.data.notification.targets.platform.TenantAdministratorsFilter; import org.thingsboard.server.common.data.notification.targets.platform.UsersFilter; import org.thingsboard.server.common.data.notification.targets.platform.UsersFilterType; +import org.thingsboard.server.common.data.notification.template.NotificationTemplate; import org.thingsboard.server.common.data.page.PageLink; import org.thingsboard.server.common.data.settings.UserSettings; import org.thingsboard.server.common.data.settings.UserSettingsType; @@ -53,6 +54,7 @@ import java.util.LinkedHashMap; import java.util.List; import java.util.Map; import java.util.Optional; +import java.util.stream.Collectors; @Service @RequiredArgsConstructor @@ -187,6 +189,8 @@ public class DefaultNotificationSettingsService implements NotificationSettingsS defaultNotifications.create(tenantId, DefaultNotifications.alarmComment, tenantAdmins.getId()); defaultNotifications.create(tenantId, DefaultNotifications.alarmAssignment, affectedUser.getId()); defaultNotifications.create(tenantId, DefaultNotifications.ruleEngineComponentLifecycleFailure, tenantAdmins.getId()); + defaultNotifications.create(tenantId, DefaultNotifications.edgeConnection, tenantAdmins.getId()); + defaultNotifications.create(tenantId, DefaultNotifications.edgeCommunicationFailures, tenantAdmins.getId()); } @Override @@ -198,17 +202,43 @@ public class DefaultNotificationSettingsService implements NotificationSettingsS } NotificationTarget sysAdmins = notificationTargetService.findNotificationTargetsByTenantIdAndUsersFilterType(tenantId, UsersFilterType.SYSTEM_ADMINISTRATORS).stream() - .findFirst().orElseGet(() -> { - return createTarget(tenantId, "System administrators", new SystemAdministratorsFilter(), "All system administrators"); - }); + .findFirst().orElseGet(() -> createTarget(tenantId, "System administrators", new SystemAdministratorsFilter(), "All system administrators")); NotificationTarget affectedTenantAdmins = notificationTargetService.findNotificationTargetsByTenantIdAndUsersFilterType(tenantId, UsersFilterType.AFFECTED_TENANT_ADMINISTRATORS).stream() - .findFirst().orElseGet(() -> { - return createTarget(tenantId, "Affected tenant's administrators", new AffectedTenantAdministratorsFilter(), ""); - }); + .findFirst().orElseGet(() -> createTarget(tenantId, "Affected tenant's administrators", new AffectedTenantAdministratorsFilter(), "")); defaultNotifications.create(tenantId, DefaultNotifications.exceededRateLimits, affectedTenantAdmins.getId()); defaultNotifications.create(tenantId, DefaultNotifications.exceededPerEntityRateLimits, affectedTenantAdmins.getId()); defaultNotifications.create(tenantId, DefaultNotifications.exceededRateLimitsForSysadmin, sysAdmins.getId()); + } else { + var requiredNotificationTypes = List.of(NotificationType.EDGE_CONNECTION, NotificationType.EDGE_COMMUNICATION_FAILURE); + var existingNotificationTypes = notificationTemplateService.findNotificationTemplatesByTenantIdAndNotificationTypes( + tenantId, requiredNotificationTypes, new PageLink(1)) + .getData() + .stream() + .map(NotificationTemplate::getNotificationType) + .collect(Collectors.toSet()); + + if (existingNotificationTypes.containsAll(requiredNotificationTypes)) { + return; + } + + NotificationTarget tenantAdmins = notificationTargetService.findNotificationTargetsByTenantIdAndUsersFilterType(tenantId, UsersFilterType.TENANT_ADMINISTRATORS) + .stream() + .findFirst() + .orElseGet(() -> createTarget(tenantId, "Tenant administrators", new TenantAdministratorsFilter(), "Tenant administrators")); + + for (NotificationType type : requiredNotificationTypes) { + if (!existingNotificationTypes.contains(type)) { + switch (type) { + case EDGE_CONNECTION: + defaultNotifications.create(tenantId, DefaultNotifications.edgeConnection, tenantAdmins.getId()); + break; + case EDGE_COMMUNICATION_FAILURE: + defaultNotifications.create(tenantId, DefaultNotifications.edgeCommunicationFailures, tenantAdmins.getId()); + break; + } + } + } } } diff --git a/dao/src/main/java/org/thingsboard/server/dao/notification/DefaultNotifications.java b/dao/src/main/java/org/thingsboard/server/dao/notification/DefaultNotifications.java index 5a78fa9864..1ef0006e87 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/notification/DefaultNotifications.java +++ b/dao/src/main/java/org/thingsboard/server/dao/notification/DefaultNotifications.java @@ -40,6 +40,9 @@ import org.thingsboard.server.common.data.notification.rule.trigger.config.Alarm import org.thingsboard.server.common.data.notification.rule.trigger.config.ApiUsageLimitNotificationRuleTriggerConfig; import org.thingsboard.server.common.data.notification.rule.trigger.config.DeviceActivityNotificationRuleTriggerConfig; import org.thingsboard.server.common.data.notification.rule.trigger.config.DeviceActivityNotificationRuleTriggerConfig.DeviceEvent; +import org.thingsboard.server.common.data.notification.rule.trigger.config.EdgeConnectionNotificationRuleTriggerConfig; +import org.thingsboard.server.common.data.notification.rule.trigger.config.EdgeConnectionNotificationRuleTriggerConfig.EdgeConnectivityEvent; +import org.thingsboard.server.common.data.notification.rule.trigger.config.EdgeCommunicationFailureNotificationRuleTriggerConfig; import org.thingsboard.server.common.data.notification.rule.trigger.config.EntitiesLimitNotificationRuleTriggerConfig; import org.thingsboard.server.common.data.notification.rule.trigger.config.EntityActionNotificationRuleTriggerConfig; import org.thingsboard.server.common.data.notification.rule.trigger.config.NewPlatformVersionNotificationRuleTriggerConfig; @@ -325,6 +328,35 @@ public class DefaultNotifications { .description("Send notification to tenant admins when any Rule chain or Rule node failed to start, update or stop") .build()) .build(); + public static final DefaultNotification edgeConnection = DefaultNotification.builder() + .name("Edge connection notification") + .type(NotificationType.EDGE_CONNECTION) + .subject("Edge connection status change") + .text("Edge '${edgeName}' is now ${eventType}") + .icon("info").color(null) + .button("Go to Edge").link("/edgeManagement/instances/${edgeId}") + .rule(DefaultRule.builder() + .name("Edge connection status change") + .triggerConfig(EdgeConnectionNotificationRuleTriggerConfig.builder() + .edges(null) + .notifyOn(Set.of(EdgeConnectivityEvent.CONNECTED, EdgeConnectivityEvent.DISCONNECTED)) + .build()) + .description("Send notification to tenant admins when the connection status between TB and Edge changes") + .build()) + .build(); + public static final DefaultNotification edgeCommunicationFailures = DefaultNotification.builder() + .name("Edge communication failure notification") + .type(NotificationType.EDGE_COMMUNICATION_FAILURE) + .subject("Edge '${edgeName}' communication failure occured") + .text("Failure message: '${failureMsg}'") + .icon("error").color(RED_COLOR) + .button("Go to Edge").link("/edgeManagement/instances/${edgeId}") + .rule(DefaultRule.builder() + .name("Edge communication failure") + .triggerConfig(EdgeCommunicationFailureNotificationRuleTriggerConfig.builder().edges(null).build()) + .description("Send notification to tenant admins when communication failures occur") + .build()) + .build(); public static final DefaultNotification jwtSigningKeyIssue = DefaultNotification.builder() .name("JWT Signing Key issue notification") @@ -346,7 +378,7 @@ public class DefaultNotifications { if (defaultNotification.getRule() != null && targets.length > 0) { NotificationRule rule = defaultNotification.toRule(template.getId(), targets); rule.setTenantId(tenantId); - rule = ruleService.saveNotificationRule(tenantId, rule); + ruleService.saveNotificationRule(tenantId, rule); } } diff --git a/dao/src/main/java/org/thingsboard/server/dao/timeseries/CassandraBaseTimeseriesDao.java b/dao/src/main/java/org/thingsboard/server/dao/timeseries/CassandraBaseTimeseriesDao.java index 31e29434e8..6b2ab7e07b 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/timeseries/CassandraBaseTimeseriesDao.java +++ b/dao/src/main/java/org/thingsboard/server/dao/timeseries/CassandraBaseTimeseriesDao.java @@ -65,6 +65,7 @@ import java.util.ArrayList; import java.util.Collections; import java.util.List; import java.util.Optional; +import java.util.UUID; import java.util.concurrent.TimeUnit; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; @@ -85,6 +86,18 @@ public class CassandraBaseTimeseriesDao extends AbstractCassandraBaseTimeseriesD public static final String ASC_ORDER = "ASC"; public static final long SECONDS_IN_DAY = TimeUnit.DAYS.toSeconds(1); protected static final List FIXED_PARTITION = List.of(0L); + protected static final String INSERT_WITH_NULL = INSERT_INTO + ModelConstants.TS_KV_CF + + "(" + ModelConstants.ENTITY_TYPE_COLUMN + + "," + ModelConstants.ENTITY_ID_COLUMN + + "," + ModelConstants.KEY_COLUMN + + "," + ModelConstants.PARTITION_COLUMN + + "," + ModelConstants.TS_COLUMN + + "," + ModelConstants.BOOLEAN_VALUE_COLUMN + + "," + ModelConstants.STRING_VALUE_COLUMN + + "," + ModelConstants.LONG_VALUE_COLUMN + + "," + ModelConstants.DOUBLE_VALUE_COLUMN + + "," + ModelConstants.JSON_VALUE_COLUMN + ")" + + " VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?)"; private CassandraTsPartitionsCache cassandraTsPartitionsCache; @@ -117,6 +130,8 @@ public class CassandraBaseTimeseriesDao extends AbstractCassandraBaseTimeseriesD private PreparedStatement[] fetchStmtsAsc; private PreparedStatement[] fetchStmtsDesc; private PreparedStatement deleteStmt; + private PreparedStatement saveWithNullStmt; + private PreparedStatement saveWithNullWithTtlStmt; private final Lock stmtCreationLock = new ReentrantLock(); private boolean isInstall() { @@ -159,19 +174,36 @@ public class CassandraBaseTimeseriesDao extends AbstractCassandraBaseTimeseriesD ttl = computeTtl(ttl); int dataPointDays = tsKvEntry.getDataPoints() * Math.max(1, (int) (ttl / SECONDS_IN_DAY)); long partition = toPartitionTs(tsKvEntry.getTs()); + String entityType = entityId.getEntityType().name(); + UUID entityIdId = entityId.getId(); + String entryKey = tsKvEntry.getKey(); + long ts = tsKvEntry.getTs(); DataType type = tsKvEntry.getDataType(); + BoundStatementBuilder stmtBuilder; if (setNullValuesEnabled) { - processSetNullValues(tenantId, entityId, tsKvEntry, ttl, futures, partition, type); - } - BoundStatementBuilder stmtBuilder = new BoundStatementBuilder((ttl == 0 ? getSaveStmt(type) : getSaveTtlStmt(type)).bind()); - stmtBuilder.setString(0, entityId.getEntityType().name()) - .setUuid(1, entityId.getId()) - .setString(2, tsKvEntry.getKey()) - .setLong(3, partition) - .setLong(4, tsKvEntry.getTs()); - addValue(tsKvEntry, stmtBuilder, 5); - if (ttl > 0) { - stmtBuilder.setInt(6, (int) ttl); + Boolean booleanValue = tsKvEntry.getBooleanValue().orElse(null); + String strValue = tsKvEntry.getStrValue().orElse(null); + Long longValue = tsKvEntry.getLongValue().orElse(null); + Double doubleValue = tsKvEntry.getDoubleValue().orElse(null); + String jsonValue = tsKvEntry.getJsonValue().orElse(null); + if (ttl == 0) { + stmtBuilder = new BoundStatementBuilder(getSaveWithNullStmt() + .bind(entityType, entityIdId, entryKey, partition, ts, booleanValue, strValue, longValue, doubleValue, jsonValue)); + } else { + stmtBuilder = new BoundStatementBuilder(getSaveWithNullWithTtlStmt() + .bind(entityType, entityIdId, entryKey, partition, ts, booleanValue, strValue, longValue, doubleValue, jsonValue, (int) ttl)); + } + } else { + stmtBuilder = new BoundStatementBuilder((ttl == 0 ? getSaveStmt(type) : getSaveTtlStmt(type)).bind()); + stmtBuilder.setString(0, entityType) + .setUuid(1, entityIdId) + .setString(2, entryKey) + .setLong(3, partition) + .setLong(4, ts); + addValue(tsKvEntry, stmtBuilder, 5); + if (ttl > 0) { + stmtBuilder.setInt(6, (int) ttl); + } } BoundStatement stmt = stmtBuilder.build(); futures.add(getFuture(executeAsyncWrite(tenantId, stmt), rs -> null)); @@ -449,56 +481,6 @@ public class CassandraBaseTimeseriesDao extends AbstractCassandraBaseTimeseriesD return tsFormat.getTruncateUnit().equals(ChronoUnit.FOREVER); } - private void processSetNullValues(TenantId tenantId, EntityId entityId, TsKvEntry tsKvEntry, long ttl, List> futures, long partition, DataType type) { - switch (type) { - case LONG: - futures.add(saveNull(tenantId, entityId, tsKvEntry, ttl, partition, DataType.BOOLEAN)); - futures.add(saveNull(tenantId, entityId, tsKvEntry, ttl, partition, DataType.DOUBLE)); - futures.add(saveNull(tenantId, entityId, tsKvEntry, ttl, partition, DataType.STRING)); - futures.add(saveNull(tenantId, entityId, tsKvEntry, ttl, partition, DataType.JSON)); - break; - case BOOLEAN: - futures.add(saveNull(tenantId, entityId, tsKvEntry, ttl, partition, DataType.DOUBLE)); - futures.add(saveNull(tenantId, entityId, tsKvEntry, ttl, partition, DataType.LONG)); - futures.add(saveNull(tenantId, entityId, tsKvEntry, ttl, partition, DataType.STRING)); - futures.add(saveNull(tenantId, entityId, tsKvEntry, ttl, partition, DataType.JSON)); - break; - case DOUBLE: - futures.add(saveNull(tenantId, entityId, tsKvEntry, ttl, partition, DataType.BOOLEAN)); - futures.add(saveNull(tenantId, entityId, tsKvEntry, ttl, partition, DataType.LONG)); - futures.add(saveNull(tenantId, entityId, tsKvEntry, ttl, partition, DataType.STRING)); - futures.add(saveNull(tenantId, entityId, tsKvEntry, ttl, partition, DataType.JSON)); - break; - case STRING: - futures.add(saveNull(tenantId, entityId, tsKvEntry, ttl, partition, DataType.BOOLEAN)); - futures.add(saveNull(tenantId, entityId, tsKvEntry, ttl, partition, DataType.DOUBLE)); - futures.add(saveNull(tenantId, entityId, tsKvEntry, ttl, partition, DataType.LONG)); - futures.add(saveNull(tenantId, entityId, tsKvEntry, ttl, partition, DataType.JSON)); - break; - case JSON: - futures.add(saveNull(tenantId, entityId, tsKvEntry, ttl, partition, DataType.BOOLEAN)); - futures.add(saveNull(tenantId, entityId, tsKvEntry, ttl, partition, DataType.DOUBLE)); - futures.add(saveNull(tenantId, entityId, tsKvEntry, ttl, partition, DataType.LONG)); - futures.add(saveNull(tenantId, entityId, tsKvEntry, ttl, partition, DataType.STRING)); - break; - } - } - - private ListenableFuture saveNull(TenantId tenantId, EntityId entityId, TsKvEntry tsKvEntry, long ttl, long partition, DataType type) { - BoundStatementBuilder stmtBuilder = new BoundStatementBuilder((ttl == 0 ? getSaveStmt(type) : getSaveTtlStmt(type)).bind()); - stmtBuilder.setString(0, entityId.getEntityType().name()) - .setUuid(1, entityId.getId()) - .setString(2, tsKvEntry.getKey()) - .setLong(3, partition) - .setLong(4, tsKvEntry.getTs()); - stmtBuilder.setToNull(getColumnName(type)); - if (ttl > 0) { - stmtBuilder.setInt(6, (int) ttl); - } - BoundStatement stmt = stmtBuilder.build(); - return getFuture(executeAsyncWrite(tenantId, stmt), rs -> null); - } - private ListenableFuture doSavePartition(TenantId tenantId, EntityId entityId, String key, long ttl, long partition) { log.debug("Saving partition {} for the entity [{}-{}] and key {}", partition, entityId.getEntityType(), entityId.getId(), key); PreparedStatement preparedStatement = ttl == 0 ? getPartitionInsertStmt() : getPartitionInsertTtlStmt(); @@ -591,6 +573,34 @@ public class CassandraBaseTimeseriesDao extends AbstractCassandraBaseTimeseriesD return deleteStmt; } + private PreparedStatement getSaveWithNullStmt() { + if (saveWithNullStmt == null) { + stmtCreationLock.lock(); + try { + if (saveWithNullStmt == null) { + saveWithNullStmt = prepare(INSERT_WITH_NULL); + } + } finally { + stmtCreationLock.unlock(); + } + } + return saveWithNullStmt; + } + + private PreparedStatement getSaveWithNullWithTtlStmt() { + if (saveWithNullWithTtlStmt == null) { + stmtCreationLock.lock(); + try { + if (saveWithNullWithTtlStmt == null) { + saveWithNullWithTtlStmt = prepare(INSERT_WITH_NULL + " USING TTL ?"); + } + } finally { + stmtCreationLock.unlock(); + } + } + return saveWithNullWithTtlStmt; + } + private PreparedStatement getSaveStmt(DataType dataType) { if (saveStmts == null) { stmtCreationLock.lock(); diff --git a/dao/src/test/java/org/thingsboard/server/dao/service/DeviceServiceTest.java b/dao/src/test/java/org/thingsboard/server/dao/service/DeviceServiceTest.java index 61d1c845fc..12bcac72f1 100644 --- a/dao/src/test/java/org/thingsboard/server/dao/service/DeviceServiceTest.java +++ b/dao/src/test/java/org/thingsboard/server/dao/service/DeviceServiceTest.java @@ -16,12 +16,15 @@ package org.thingsboard.server.dao.service; import com.datastax.oss.driver.api.core.uuid.Uuids; +import org.hibernate.exception.ConstraintViolationException; import org.junit.After; import org.junit.Assert; import org.junit.Before; import org.junit.Test; import org.junit.jupiter.api.Assertions; +import org.mockito.Mockito; import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.test.mock.mockito.SpyBean; import org.springframework.transaction.PlatformTransactionManager; import org.springframework.transaction.TransactionStatus; import org.springframework.transaction.support.DefaultTransactionDefinition; @@ -35,8 +38,6 @@ import org.thingsboard.server.common.data.OtaPackage; 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.asset.Asset; -import org.thingsboard.server.common.data.asset.AssetProfile; import org.thingsboard.server.common.data.id.CustomerId; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.ota.ChecksumAlgorithm; @@ -50,15 +51,19 @@ 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.exception.DataValidationException; +import org.thingsboard.server.dao.exception.DeviceCredentialsValidationException; import org.thingsboard.server.dao.ota.OtaPackageService; +import org.thingsboard.server.dao.service.validator.DeviceCredentialsDataValidator; import org.thingsboard.server.dao.tenant.TenantProfileService; import java.nio.ByteBuffer; +import java.sql.SQLException; import java.util.ArrayList; import java.util.Collections; import java.util.List; import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.ArgumentMatchers.any; import static org.thingsboard.server.common.data.ota.OtaPackageType.FIRMWARE; import static org.thingsboard.server.dao.model.ModelConstants.NULL_UUID; @@ -79,6 +84,8 @@ public class DeviceServiceTest extends AbstractServiceTest { TenantProfileService tenantProfileService; @Autowired private PlatformTransactionManager platformTransactionManager; + @SpyBean + private DeviceCredentialsDataValidator validator; private IdComparator idComparator = new IdComparator<>(); private TenantId anotherTenantId; @@ -129,6 +136,67 @@ public class DeviceServiceTest extends AbstractServiceTest { }); } + @Test + public void testSaveDevicesWithTheSameAccessToken() { + Device device = new Device(); + device.setTenantId(tenantId); + device.setName(StringUtils.randomAlphabetic(10)); + device.setType("default"); + String accessToken = StringUtils.generateSafeToken(10); + Device savedDevice = deviceService.saveDeviceWithAccessToken(device, accessToken); + + DeviceCredentials deviceCredentials = deviceCredentialsService.findDeviceCredentialsByDeviceId(tenantId, savedDevice.getId()); + Assert.assertEquals(accessToken, deviceCredentials.getCredentialsId()); + + Device duplicatedDevice = new Device(); + duplicatedDevice.setTenantId(tenantId); + duplicatedDevice.setName(StringUtils.randomAlphabetic(10)); + duplicatedDevice.setType("default"); + assertThatThrownBy(() -> deviceService.saveDeviceWithAccessToken(duplicatedDevice, accessToken)) + .isInstanceOf(DeviceCredentialsValidationException.class) + .hasMessageContaining("Device credentials are already assigned to another device!"); + + Device deviceByName = deviceService.findDeviceByTenantIdAndName(tenantId, duplicatedDevice.getName()); + Assertions.assertNull(deviceByName); + } + + @Test + public void testShouldRollbackNotValidatedDeviceIfDeviceCredentialsValidationFailed() { + Mockito.reset(validator); + Mockito.doThrow(new DataValidationException("mock message")) + .when(validator).validate(any(), any()); + + Device device = new Device(); + device.setTenantId(tenantId); + device.setName(StringUtils.randomAlphabetic(10)); + device.setType("default"); + assertThatThrownBy(() -> deviceService.saveDevice(device, false)) + .isInstanceOf(DataValidationException.class) + .hasMessageContaining("mock message"); + + Device deviceByName = deviceService.findDeviceByTenantIdAndName(tenantId, device.getName()); + Assertions.assertNull(deviceByName); + } + + @Test + public void testShouldRollbackValidatedDeviceIfDeviceCredentialsValidationFailed() { + Mockito.reset(validator); + Mockito.doThrow(new DataValidationException("mock message")) + .when(validator).validate(any(), any()); + + Device device = new Device(); + device.setTenantId(tenantId); + device.setName(StringUtils.randomAlphabetic(10)); + device.setType("default"); + + assertThatThrownBy(() -> deviceService.saveDevice(device)) + .isInstanceOf(DataValidationException.class) + .hasMessageContaining("mock message"); + + Device deviceByName = deviceService.findDeviceByTenantIdAndName(tenantId, device.getName()); + Assertions.assertNull(deviceByName); + } + @Test public void testCountByTenantId() { Assert.assertEquals(0, deviceService.countByTenantId(tenantId)); diff --git a/dao/src/test/java/org/thingsboard/server/dao/service/EdgeEventServiceTest.java b/dao/src/test/java/org/thingsboard/server/dao/service/EdgeEventServiceTest.java index 2a03a1277c..2cc179fae2 100644 --- a/dao/src/test/java/org/thingsboard/server/dao/service/EdgeEventServiceTest.java +++ b/dao/src/test/java/org/thingsboard/server/dao/service/EdgeEventServiceTest.java @@ -40,7 +40,7 @@ import java.text.ParseException; import java.util.ArrayList; import java.util.List; -import static org.apache.commons.lang3.time.DateFormatUtils.ISO_DATETIME_TIME_ZONE_FORMAT; +import static org.apache.commons.lang3.time.DateFormatUtils.ISO_8601_EXTENDED_DATETIME_FORMAT; @DaoSqlTest public class EdgeEventServiceTest extends AbstractServiceTest { @@ -56,19 +56,18 @@ public class EdgeEventServiceTest extends AbstractServiceTest { @Before public void before() throws ParseException { - timeBeforeStartTime = ISO_DATETIME_TIME_ZONE_FORMAT.parse("2016-11-01T11:30:00Z").getTime(); - startTime = ISO_DATETIME_TIME_ZONE_FORMAT.parse("2016-11-01T12:00:00Z").getTime(); - eventTime = ISO_DATETIME_TIME_ZONE_FORMAT.parse("2016-11-01T12:30:00Z").getTime(); - endTime = ISO_DATETIME_TIME_ZONE_FORMAT.parse("2016-11-01T13:00:00Z").getTime(); - timeAfterEndTime = ISO_DATETIME_TIME_ZONE_FORMAT.parse("2016-11-01T13:30:30Z").getTime(); + timeBeforeStartTime = ISO_8601_EXTENDED_DATETIME_FORMAT.parse("2016-11-01T11:30:00").getTime(); + startTime = ISO_8601_EXTENDED_DATETIME_FORMAT.parse("2016-11-01T12:00:00").getTime(); + eventTime = ISO_8601_EXTENDED_DATETIME_FORMAT.parse("2016-11-01T12:30:00").getTime(); + endTime = ISO_8601_EXTENDED_DATETIME_FORMAT.parse("2016-11-01T13:00:00").getTime(); + timeAfterEndTime = ISO_8601_EXTENDED_DATETIME_FORMAT.parse("2016-11-01T13:30:30").getTime(); } @Test public void saveEdgeEvent() throws Exception { EdgeId edgeId = new EdgeId(Uuids.timeBased()); DeviceId deviceId = new DeviceId(Uuids.timeBased()); - TenantId tenantId = new TenantId(Uuids.timeBased()); - EdgeEvent edgeEvent = generateEdgeEvent(tenantId, edgeId, deviceId, EdgeEventActionType.ADDED); + EdgeEvent edgeEvent = generateEdgeEvent(tenantId, edgeId, deviceId); edgeEventService.saveAsync(edgeEvent).get(); PageData edgeEvents = edgeEventService.findEdgeEvents(tenantId, edgeId, 0L, null, new TimePageLink(1)); @@ -81,9 +80,11 @@ public class EdgeEventServiceTest extends AbstractServiceTest { Assert.assertEquals(saved.getType(), edgeEvent.getType()); Assert.assertEquals(saved.getAction(), edgeEvent.getAction()); Assert.assertEquals(saved.getBody(), edgeEvent.getBody()); + + edgeEventService.cleanupEvents(1); } - protected EdgeEvent generateEdgeEvent(TenantId tenantId, EdgeId edgeId, EntityId entityId, EdgeEventActionType edgeEventAction) throws IOException { + protected EdgeEvent generateEdgeEvent(TenantId tenantId, EdgeId edgeId, EntityId entityId) throws IOException { if (tenantId == null) { tenantId = TenantId.fromUUID(Uuids.timeBased()); } @@ -92,7 +93,7 @@ public class EdgeEventServiceTest extends AbstractServiceTest { edgeEvent.setEdgeId(edgeId); edgeEvent.setEntityId(entityId.getId()); edgeEvent.setType(EdgeEventType.DEVICE); - edgeEvent.setAction(edgeEventAction); + edgeEvent.setAction(EdgeEventActionType.ADDED); edgeEvent.setBody(readFromResource("TestJsonData.json")); return edgeEvent; } @@ -101,7 +102,6 @@ public class EdgeEventServiceTest extends AbstractServiceTest { public void findEdgeEventsByTimeDescOrder() throws Exception { EdgeId edgeId = new EdgeId(Uuids.timeBased()); DeviceId deviceId = new DeviceId(Uuids.timeBased()); - TenantId tenantId = TenantId.fromUUID(Uuids.timeBased()); List> futures = new ArrayList<>(); futures.add(saveEdgeEventWithProvidedTime(timeBeforeStartTime, edgeId, deviceId, tenantId)); @@ -133,7 +133,7 @@ public class EdgeEventServiceTest extends AbstractServiceTest { } private ListenableFuture saveEdgeEventWithProvidedTime(long time, EdgeId edgeId, EntityId entityId, TenantId tenantId) throws Exception { - EdgeEvent edgeEvent = generateEdgeEvent(tenantId, edgeId, entityId, EdgeEventActionType.ADDED); + EdgeEvent edgeEvent = generateEdgeEvent(tenantId, edgeId, entityId); edgeEvent.setId(new EdgeEventId(Uuids.startOf(time))); return edgeEventService.saveAsync(edgeEvent); } diff --git a/dao/src/test/java/org/thingsboard/server/dao/service/timeseries/BaseTimeseriesServiceTest.java b/dao/src/test/java/org/thingsboard/server/dao/service/timeseries/BaseTimeseriesServiceTest.java index ad351bb9f2..a624a2b6e5 100644 --- a/dao/src/test/java/org/thingsboard/server/dao/service/timeseries/BaseTimeseriesServiceTest.java +++ b/dao/src/test/java/org/thingsboard/server/dao/service/timeseries/BaseTimeseriesServiceTest.java @@ -33,6 +33,7 @@ import org.thingsboard.server.common.data.kv.BaseReadTsKvQuery; import org.thingsboard.server.common.data.kv.BasicTsKvEntry; import org.thingsboard.server.common.data.kv.BooleanDataEntry; import org.thingsboard.server.common.data.kv.DoubleDataEntry; +import org.thingsboard.server.common.data.kv.JsonDataEntry; import org.thingsboard.server.common.data.kv.KvEntry; import org.thingsboard.server.common.data.kv.LongDataEntry; import org.thingsboard.server.common.data.kv.ReadTsKvQuery; @@ -54,7 +55,9 @@ import java.util.concurrent.TimeoutException; import static org.assertj.core.api.Assertions.assertThat; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertTrue; /** * @author Andrew Shvayka @@ -64,12 +67,12 @@ import static org.junit.Assert.assertNotNull; public abstract class BaseTimeseriesServiceTest extends AbstractServiceTest { @Autowired - TimeseriesService tsService; + protected TimeseriesService tsService; @Autowired EntityViewService entityViewService; - static final int MAX_TIMEOUT = 30; + protected static final int MAX_TIMEOUT = 30; private static final String STRING_KEY = "stringKey"; private static final String LONG_KEY = "longKey"; @@ -84,7 +87,7 @@ public abstract class BaseTimeseriesServiceTest extends AbstractServiceTest { KvEntry doubleKvEntry = new DoubleDataEntry(DOUBLE_KEY, Double.MAX_VALUE); KvEntry booleanKvEntry = new BooleanDataEntry(BOOLEAN_KEY, Boolean.TRUE); - private TenantId tenantId; + protected TenantId tenantId; @Before public void before() { @@ -673,6 +676,32 @@ public abstract class BaseTimeseriesServiceTest extends AbstractServiceTest { assertEquals(3, list.size()); } + @Test + public void shouldSaveEntryOfEachType() throws Exception { + BasicTsKvEntry booleanEntry = new BasicTsKvEntry(TimeUnit.MINUTES.toMillis(1), new BooleanDataEntry("test", true)); + BasicTsKvEntry stringEntry = new BasicTsKvEntry(TimeUnit.MINUTES.toMillis(2), new StringDataEntry("test", "text")); + BasicTsKvEntry longEntry = new BasicTsKvEntry(TimeUnit.MINUTES.toMillis(3), new LongDataEntry("test", 15L)); + BasicTsKvEntry doubleEntry = new BasicTsKvEntry(TimeUnit.MINUTES.toMillis(4), new DoubleDataEntry("test", 10.5)); + BasicTsKvEntry jsonEntry = new BasicTsKvEntry(TimeUnit.MINUTES.toMillis(5), new JsonDataEntry("test", "{\"test\":\"testValue\"}")); + List timeseries = List.of(booleanEntry, stringEntry, longEntry, doubleEntry, jsonEntry); + + DeviceId deviceId = new DeviceId(Uuids.timeBased()); + for (TsKvEntry tsKvEntry : timeseries) { + save(tenantId, deviceId, tsKvEntry); + } + + List listUntil3Minutes = tsService.findAll(tenantId, deviceId, Collections.singletonList(new BaseReadTsKvQuery("test", 0L, + TimeUnit.MINUTES.toMillis(3), 1000, 10, Aggregation.NONE))).get(MAX_TIMEOUT, TimeUnit.SECONDS); + assertEquals(2, listUntil3Minutes.size()); + assertThat(listUntil3Minutes).containsOnlyOnceElementsOf(List.of( + booleanEntry, stringEntry)); + + List fullList = tsService.findAll(tenantId, deviceId, Collections.singletonList(new BaseReadTsKvQuery("test", 0L, + TimeUnit.MINUTES.toMillis(6), 1000, 10, Aggregation.NONE))).get(MAX_TIMEOUT, TimeUnit.SECONDS); + assertEquals(5, fullList.size()); + assertThat(fullList).containsOnlyOnceElementsOf(timeseries); + } + private TsKvEntry save(DeviceId deviceId, long ts, long value) throws Exception { TsKvEntry entry = new BasicTsKvEntry(ts, new LongDataEntry(LONG_KEY, value)); tsService.save(tenantId, deviceId, entry).get(MAX_TIMEOUT, TimeUnit.SECONDS); @@ -691,6 +720,9 @@ public abstract class BaseTimeseriesServiceTest extends AbstractServiceTest { return entry; } + private void save(TenantId tenantId, DeviceId deviceId, TsKvEntry tsKvEntry) throws Exception { + tsService.save(tenantId, deviceId, tsKvEntry).get(MAX_TIMEOUT, TimeUnit.SECONDS); + } private void saveEntries(DeviceId deviceId, long ts) throws ExecutionException, InterruptedException, TimeoutException { tsService.save(tenantId, deviceId, toTsEntry(ts, stringKvEntry)).get(MAX_TIMEOUT, TimeUnit.SECONDS); diff --git a/dao/src/test/java/org/thingsboard/server/dao/service/timeseries/nosql/TimeseriesServiceNoSqlSetNullEnabledTest.java b/dao/src/test/java/org/thingsboard/server/dao/service/timeseries/nosql/TimeseriesServiceNoSqlSetNullEnabledTest.java new file mode 100644 index 0000000000..1bbb37659c --- /dev/null +++ b/dao/src/test/java/org/thingsboard/server/dao/service/timeseries/nosql/TimeseriesServiceNoSqlSetNullEnabledTest.java @@ -0,0 +1,72 @@ +/** + * Copyright © 2016-2024 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.dao.service.timeseries.nosql; + +import com.datastax.oss.driver.api.core.uuid.Uuids; +import org.junit.Test; +import org.springframework.test.context.TestPropertySource; +import org.thingsboard.server.common.data.id.DeviceId; +import org.thingsboard.server.common.data.kv.Aggregation; +import org.thingsboard.server.common.data.kv.BaseReadTsKvQuery; +import org.thingsboard.server.common.data.kv.BasicTsKvEntry; +import org.thingsboard.server.common.data.kv.DoubleDataEntry; +import org.thingsboard.server.common.data.kv.LongDataEntry; +import org.thingsboard.server.common.data.kv.TsKvEntry; +import org.thingsboard.server.dao.service.DaoNoSqlTest; + +import java.util.Collections; +import java.util.List; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +@DaoNoSqlTest +@TestPropertySource(properties = { + "cassandra.query.set_null_values_enabled=true", +}) +public class TimeseriesServiceNoSqlSetNullEnabledTest extends TimeseriesServiceNoSqlTest { + + @Override + @Test + public void testNullValuesOfNoneTargetColumn() throws ExecutionException, InterruptedException, TimeoutException { + long ts = TimeUnit.MINUTES.toMillis(1); + TsKvEntry longEntry = new BasicTsKvEntry(ts, new LongDataEntry("temp", 0L)); + double doubleValue = 20.6; + TsKvEntry doubleEntry = new BasicTsKvEntry(ts, new DoubleDataEntry("temp", doubleValue)); + DeviceId deviceId = new DeviceId(Uuids.timeBased()); + tsService.save(tenantId, deviceId, longEntry).get(MAX_TIMEOUT, TimeUnit.SECONDS); + tsService.save(tenantId, deviceId, doubleEntry).get(MAX_TIMEOUT, TimeUnit.SECONDS); + + List listWithoutAgg = tsService.findAll(tenantId, deviceId, Collections.singletonList(new BaseReadTsKvQuery("temp", 0L, + ts + 1 , 1000, 3, Aggregation.NONE))).get(MAX_TIMEOUT, TimeUnit.SECONDS); + assertEquals(1, listWithoutAgg.size()); + assertFalse(listWithoutAgg.get(0).getLongValue().isPresent()); + assertTrue(listWithoutAgg.get(0).getDoubleValue().isPresent()); + assertThat(listWithoutAgg.get(0).getDoubleValue().get()).isEqualTo(doubleValue); + + // long value should be set to null after second insert, so avg = doubleValue + List listWithAgg = tsService.findAll(tenantId, deviceId, Collections.singletonList(new BaseReadTsKvQuery("temp", 0L, + ts + 1 , 1000, 3, Aggregation.AVG))).get(MAX_TIMEOUT, TimeUnit.SECONDS); + assertEquals(1, listWithAgg.size()); + assertTrue(listWithAgg.get(0).getDoubleValue().isPresent()); + assertThat(listWithAgg.get(0).getDoubleValue().get()).isEqualTo(doubleValue); + } +} diff --git a/dao/src/test/java/org/thingsboard/server/dao/service/timeseries/nosql/TimeseriesServiceNoSqlTest.java b/dao/src/test/java/org/thingsboard/server/dao/service/timeseries/nosql/TimeseriesServiceNoSqlTest.java index 229d5f5842..b8970feca3 100644 --- a/dao/src/test/java/org/thingsboard/server/dao/service/timeseries/nosql/TimeseriesServiceNoSqlTest.java +++ b/dao/src/test/java/org/thingsboard/server/dao/service/timeseries/nosql/TimeseriesServiceNoSqlTest.java @@ -15,9 +15,83 @@ */ package org.thingsboard.server.dao.service.timeseries.nosql; +import com.datastax.oss.driver.api.core.uuid.Uuids; +import org.junit.Test; +import org.thingsboard.server.common.data.id.DeviceId; +import org.thingsboard.server.common.data.kv.Aggregation; +import org.thingsboard.server.common.data.kv.BaseReadTsKvQuery; +import org.thingsboard.server.common.data.kv.BasicTsKvEntry; +import org.thingsboard.server.common.data.kv.BooleanDataEntry; +import org.thingsboard.server.common.data.kv.DoubleDataEntry; +import org.thingsboard.server.common.data.kv.JsonDataEntry; +import org.thingsboard.server.common.data.kv.LongDataEntry; +import org.thingsboard.server.common.data.kv.StringDataEntry; +import org.thingsboard.server.common.data.kv.TsKvEntry; import org.thingsboard.server.dao.service.DaoNoSqlTest; import org.thingsboard.server.dao.service.timeseries.BaseTimeseriesServiceTest; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + @DaoNoSqlTest public class TimeseriesServiceNoSqlTest extends BaseTimeseriesServiceTest { + + @Test + public void shouldSaveEntryOfEachTypeWithTtl() throws ExecutionException, InterruptedException, TimeoutException { + long ttlInSec = TimeUnit.SECONDS.toSeconds(3); + List timeseries = List.of( + new BasicTsKvEntry(TimeUnit.MINUTES.toMillis(1), new BooleanDataEntry("test", true)), + new BasicTsKvEntry(TimeUnit.MINUTES.toMillis(2), new StringDataEntry("test", "text")), + new BasicTsKvEntry(TimeUnit.MINUTES.toMillis(3), new LongDataEntry("test", 15L)), + new BasicTsKvEntry(TimeUnit.MINUTES.toMillis(4), new DoubleDataEntry("test", 10.5)), + new BasicTsKvEntry(TimeUnit.MINUTES.toMillis(5), new JsonDataEntry("test", "{\"test\":\"testValue\"}"))); + + DeviceId deviceId = new DeviceId(Uuids.timeBased()); + tsService.save(tenantId, deviceId, timeseries, ttlInSec); + + List fullList = tsService.findAll(tenantId, deviceId, Collections.singletonList(new BaseReadTsKvQuery("test", 0L, + TimeUnit.MINUTES.toMillis(6), 1000, 10, Aggregation.NONE))).get(MAX_TIMEOUT, TimeUnit.SECONDS); + assertEquals(5, fullList.size()); + + // check entries after ttl + Thread.sleep(TimeUnit.SECONDS.toMillis(ttlInSec + 1)); + List listAfterTtl = tsService.findAll(tenantId, deviceId, Collections.singletonList(new BaseReadTsKvQuery("test", 0L, + TimeUnit.MINUTES.toMillis(6), 1000, 10, Aggregation.NONE))).get(MAX_TIMEOUT, TimeUnit.SECONDS); + assertEquals(0, listAfterTtl.size()); + } + + @Test + public void testNullValuesOfNoneTargetColumn() throws ExecutionException, InterruptedException, TimeoutException { + long ts = TimeUnit.MINUTES.toMillis(1); + long longValue = 10L; + TsKvEntry longEntry = new BasicTsKvEntry(ts, new LongDataEntry("temp", longValue)); + double doubleValue = 20.6; + TsKvEntry doubleEntry = new BasicTsKvEntry(ts, new DoubleDataEntry("temp", doubleValue)); + DeviceId deviceId = new DeviceId(Uuids.timeBased()); + tsService.save(tenantId, deviceId, longEntry).get(MAX_TIMEOUT, TimeUnit.SECONDS); + tsService.save(tenantId, deviceId, doubleEntry).get(MAX_TIMEOUT, TimeUnit.SECONDS); + + List listWithoutAgg = tsService.findAll(tenantId, deviceId, Collections.singletonList(new BaseReadTsKvQuery("temp", 0L, + ts + 1 , 1000, 3, Aggregation.NONE))).get(MAX_TIMEOUT, TimeUnit.SECONDS); + assertEquals(1, listWithoutAgg.size()); + assertTrue(listWithoutAgg.get(0).getLongValue().isPresent()); + assertFalse(listWithoutAgg.get(0).getDoubleValue().isPresent()); + assertThat(listWithoutAgg.get(0).getLongValue().get()).isEqualTo(longValue); + + // long value should not be reset to null, so avg = (doubleValue + longValue)/ 2 + List listWithAgg = tsService.findAll(tenantId, deviceId, Collections.singletonList(new BaseReadTsKvQuery("temp", 0L, + ts + 1, 200000, 3, Aggregation.AVG))).get(MAX_TIMEOUT, TimeUnit.SECONDS); + assertEquals(1, listWithAgg.size()); + assertTrue(listWithAgg.get(0).getDoubleValue().isPresent()); + double expectedValue = (doubleValue + longValue)/ 2; + assertThat(listWithAgg.get(0).getDoubleValue().get()).isEqualTo(expectedValue); + } } diff --git a/transport/snmp/src/main/resources/tb-snmp-transport.yml b/transport/snmp/src/main/resources/tb-snmp-transport.yml index 805ab5a8dd..a40374e1f6 100644 --- a/transport/snmp/src/main/resources/tb-snmp-transport.yml +++ b/transport/snmp/src/main/resources/tb-snmp-transport.yml @@ -128,14 +128,18 @@ transport: bind_port: "${SNMP_BIND_PORT:1620}" response_processing: # parallelism level for executor (workStealingPool) that is responsible for handling responses from SNMP devices - parallelism_level: "${SNMP_RESPONSE_PROCESSING_PARALLELISM_LEVEL:20}" + parallelism_level: "${SNMP_RESPONSE_PROCESSING_PARALLELISM_LEVEL:4}" # to configure SNMP to work over UDP or TCP underlying_protocol: "${SNMP_UNDERLYING_PROTOCOL:udp}" - # Batch size to request OID mappings from the device (useful when the device profile has multiple hundreds of OID mappings) + # Maximum size of a PDU (amount of OID mappings in a single SNMP request). The request will be split into multiple PDUs if mappings amount exceeds this number max_request_oids: "${SNMP_MAX_REQUEST_OIDS:100}" + # Delay after sending each request chunk (in case the request was split into multiple PDUs due to max_request_oids) + request_chunk_delay_ms: "${SNMP_REQUEST_CHUNK_DELAY_MS:100}" response: # To ignore SNMP response values that do not match the data type of the configured OID mapping (by default false - will throw an error if any value of the response not match configured data types) ignore_type_cast_errors: "${SNMP_RESPONSE_IGNORE_TYPE_CAST_ERRORS:false}" + # Thread pool size for scheduler that executes device querying tasks + scheduler_thread_pool_size: "${SNMP_SCHEDULER_THREAD_POOL_SIZE:4}" sessions: # Session inactivity timeout is a global configuration parameter that defines how long the device transport session will be opened after the last message arrives from the device. # The parameter value is in milliseconds. diff --git a/ui-ngx/src/app/modules/home/components/profile/tenant/default-tenant-profile-configuration.component.html b/ui-ngx/src/app/modules/home/components/profile/tenant/default-tenant-profile-configuration.component.html index 4c309f5e80..867cd74f3b 100644 --- a/ui-ngx/src/app/modules/home/components/profile/tenant/default-tenant-profile-configuration.component.html +++ b/ui-ngx/src/app/modules/home/components/profile/tenant/default-tenant-profile-configuration.component.html @@ -523,7 +523,7 @@ -
+
@@ -531,7 +531,7 @@ [type]="rateLimitsType.DEVICE_TELEMETRY_DATA_POINTS">
-
+
@@ -539,7 +539,7 @@ [type]="rateLimitsType.CUSTOMER_SERVER_REST_LIMITS_CONFIGURATION">
-
+
@@ -547,7 +547,7 @@ [type]="rateLimitsType.TENANT_ENTITY_IMPORT_RATE_LIMIT">
-
+
@@ -555,7 +555,7 @@ [type]="rateLimitsType.CASSANDRA_QUERY_TENANT_RATE_LIMITS_CONFIGURATION">
-
+
@@ -563,6 +563,14 @@ [type]="rateLimitsType.TENANT_NOTIFICATION_REQUESTS_PER_RULE_RATE_LIMIT">
+
+ + + + +
diff --git a/ui-ngx/src/app/modules/home/components/profile/tenant/default-tenant-profile-configuration.component.ts b/ui-ngx/src/app/modules/home/components/profile/tenant/default-tenant-profile-configuration.component.ts index a0b8770509..0c95324d85 100644 --- a/ui-ngx/src/app/modules/home/components/profile/tenant/default-tenant-profile-configuration.component.ts +++ b/ui-ngx/src/app/modules/home/components/profile/tenant/default-tenant-profile-configuration.component.ts @@ -106,7 +106,9 @@ export class DefaultTenantProfileConfigurationComponent implements ControlValueA maxWsSubscriptionsPerRegularUser: [null, [Validators.min(0)]], maxWsSubscriptionsPerPublicUser: [null, [Validators.min(0)]], wsUpdatesPerSessionRateLimit: [null, []], - cassandraQueryTenantRateLimitsConfiguration: [null, []] + cassandraQueryTenantRateLimitsConfiguration: [null, []], + edgeEventRateLimits: [null, []], + edgeEventRateLimitsPerEdge: [null, []] }); this.defaultTenantProfileConfigurationFormGroup.get('smsEnabled').valueChanges.pipe( diff --git a/ui-ngx/src/app/modules/home/components/profile/tenant/rate-limits/rate-limits.models.ts b/ui-ngx/src/app/modules/home/components/profile/tenant/rate-limits/rate-limits.models.ts index 257752f5f4..8924682aa1 100644 --- a/ui-ngx/src/app/modules/home/components/profile/tenant/rate-limits/rate-limits.models.ts +++ b/ui-ngx/src/app/modules/home/components/profile/tenant/rate-limits/rate-limits.models.ts @@ -35,7 +35,9 @@ export enum RateLimitsType { TENANT_ENTITY_EXPORT_RATE_LIMIT = 'TENANT_ENTITY_EXPORT_RATE_LIMIT', TENANT_ENTITY_IMPORT_RATE_LIMIT = 'TENANT_ENTITY_IMPORT_RATE_LIMIT', TENANT_NOTIFICATION_REQUEST_RATE_LIMIT = 'TENANT_NOTIFICATION_REQUEST_RATE_LIMIT', - TENANT_NOTIFICATION_REQUESTS_PER_RULE_RATE_LIMIT = 'TENANT_NOTIFICATION_REQUESTS_PER_RULE_RATE_LIMIT' + TENANT_NOTIFICATION_REQUESTS_PER_RULE_RATE_LIMIT = 'TENANT_NOTIFICATION_REQUESTS_PER_RULE_RATE_LIMIT', + EDGE_EVENTS_RATE_LIMIT = 'EDGE_EVENTS_RATE_LIMIT', + EDGE_EVENTS_PER_EDGE_RATE_LIMIT = 'EDGE_EVENTS_PER_EDGE_RATE_LIMIT' } export const rateLimitsLabelTranslationMap = new Map( @@ -54,6 +56,8 @@ export const rateLimitsLabelTranslationMap = new Map( [RateLimitsType.TENANT_ENTITY_IMPORT_RATE_LIMIT, 'tenant-profile.tenant-entity-import-rate-limit'], [RateLimitsType.TENANT_NOTIFICATION_REQUEST_RATE_LIMIT, 'tenant-profile.tenant-notification-request-rate-limit'], [RateLimitsType.TENANT_NOTIFICATION_REQUESTS_PER_RULE_RATE_LIMIT, 'tenant-profile.tenant-notification-requests-per-rule-rate-limit'], + [RateLimitsType.EDGE_EVENTS_RATE_LIMIT, 'tenant-profile.rate-limits.edge-events-rate-limit'], + [RateLimitsType.EDGE_EVENTS_PER_EDGE_RATE_LIMIT, 'tenant-profile.rate-limits.edge-events-per-edge-rate-limit'], ] ); @@ -73,6 +77,8 @@ export const rateLimitsDialogTitleTranslationMap = new Map + + {{ 'notification.edge-trigger-settings' | translate }} +
+
+
+ notification.filter + + + + notification.notify-on + + + {{ edgeConnectionEventTranslationMap.get(edgeEvent) | translate }} + + + +
+
+
+
+
+ + notification.description + + +
+
+ + + + {{ 'notification.edge-trigger-settings' | translate }} +
+
+
+ notification.filter + + +
+
+
+
+
+ + notification.description + + +
+
+ + {{ 'notification.entities-limit-trigger-settings' | translate }} diff --git a/ui-ngx/src/app/modules/home/pages/notification/rule/rule-notification-dialog.component.ts b/ui-ngx/src/app/modules/home/pages/notification/rule/rule-notification-dialog.component.ts index 9604670757..34ff605027 100644 --- a/ui-ngx/src/app/modules/home/pages/notification/rule/rule-notification-dialog.component.ts +++ b/ui-ngx/src/app/modules/home/pages/notification/rule/rule-notification-dialog.component.ts @@ -68,6 +68,7 @@ import { } from '@shared/models/api-usage.models'; import { LimitedApi, LimitedApiTranslationMap } from '@shared/models/limited-api.models'; import { StringItemsOption } from '@shared/components/string-items-list.component'; +import { EdgeConnectionEvent, EdgeConnectionEventTranslationMap } from '@shared/models/edge.models'; export interface RuleNotificationDialogData { rule?: NotificationRule; @@ -98,6 +99,8 @@ export class RuleNotificationDialogComponent extends apiUsageLimitTemplateForm: FormGroup; newPlatformVersionTemplateForm: FormGroup; rateLimitsTemplateForm: FormGroup; + edgeCommunicationFailureTemplateForm: FormGroup; + edgeConnectionTemplateForm: FormGroup; triggerType = TriggerType; triggerTypes: TriggerType[]; @@ -132,6 +135,9 @@ export class RuleNotificationDialogComponent extends apiFeatures: ApiFeature[] = Object.values(ApiFeature); apiFeatureTranslationMap = ApiFeatureTranslationMap; + edgeConnectionEvents: EdgeConnectionEvent[] = Object.values(EdgeConnectionEvent); + edgeConnectionEventTranslationMap = EdgeConnectionEventTranslationMap; + limitedApis: StringItemsOption[]; entityType = EntityType; @@ -221,6 +227,19 @@ export class RuleNotificationDialogComponent extends } }); + this.edgeConnectionTemplateForm = this.fb.group({ + triggerConfig: this.fb.group({ + edges: [null], + notifyOn: [null] + }) + }); + + this.edgeCommunicationFailureTemplateForm = this.fb.group({ + triggerConfig: this.fb.group({ + edges: [null] + }) + }); + this.alarmTemplateForm = this.fb.group({ triggerConfig: this.fb.group({ alarmTypes: [null], @@ -328,7 +347,9 @@ export class RuleNotificationDialogComponent extends [TriggerType.ENTITIES_LIMIT, this.entitiesLimitTemplateForm], [TriggerType.API_USAGE_LIMIT, this.apiUsageLimitTemplateForm], [TriggerType.NEW_PLATFORM_VERSION, this.newPlatformVersionTemplateForm], - [TriggerType.RATE_LIMITS, this.rateLimitsTemplateForm] + [TriggerType.RATE_LIMITS, this.rateLimitsTemplateForm], + [TriggerType.EDGE_COMMUNICATION_FAILURE, this.edgeCommunicationFailureTemplateForm], + [TriggerType.EDGE_CONNECTION, this.edgeConnectionTemplateForm] ]); if (data.isAdd || data.isCopy) { diff --git a/ui-ngx/src/app/shared/models/edge.models.ts b/ui-ngx/src/app/shared/models/edge.models.ts index 7187dc8eaa..569b2bec8a 100644 --- a/ui-ngx/src/app/shared/models/edge.models.ts +++ b/ui-ngx/src/app/shared/models/edge.models.ts @@ -190,3 +190,15 @@ export enum EdgeInstructionsMethod { } export const edgeVersionAttributeKey = 'edgeVersion'; + +export enum EdgeConnectionEvent { + CONNECTED= 'CONNECTED', + DISCONNECTED = 'DISCONNECTED' +} + +export const EdgeConnectionEventTranslationMap = new Map( + [ + [EdgeConnectionEvent.CONNECTED, 'edge.connected'], + [EdgeConnectionEvent.DISCONNECTED, 'edge.disconnected'] + ] +); diff --git a/ui-ngx/src/app/shared/models/limited-api.models.ts b/ui-ngx/src/app/shared/models/limited-api.models.ts index 714441f0b2..d0e2e0cb76 100644 --- a/ui-ngx/src/app/shared/models/limited-api.models.ts +++ b/ui-ngx/src/app/shared/models/limited-api.models.ts @@ -24,7 +24,9 @@ export enum LimitedApi { WS_UPDATES_PER_SESSION = 'WS_UPDATES_PER_SESSION', CASSANDRA_QUERIES = 'CASSANDRA_QUERIES', TRANSPORT_MESSAGES_PER_TENANT = 'TRANSPORT_MESSAGES_PER_TENANT', - TRANSPORT_MESSAGES_PER_DEVICE = 'TRANSPORT_MESSAGES_PER_DEVICE' + TRANSPORT_MESSAGES_PER_DEVICE = 'TRANSPORT_MESSAGES_PER_DEVICE', + EDGE_EVENTS = 'EDGE_EVENTS', + EDGE_EVENTS_PER_EDGE = 'EDGE_EVENTS_PER_EDGE' } export const LimitedApiTranslationMap = new Map( @@ -38,6 +40,8 @@ export const LimitedApiTranslationMap = new Map( [LimitedApi.WS_UPDATES_PER_SESSION, 'api-limit.ws-updates-per-session'], [LimitedApi.CASSANDRA_QUERIES, 'api-limit.cassandra-queries'], [LimitedApi.TRANSPORT_MESSAGES_PER_TENANT, 'api-limit.transport-messages'], - [LimitedApi.TRANSPORT_MESSAGES_PER_DEVICE, 'api-limit.transport-messages-per-device'] + [LimitedApi.TRANSPORT_MESSAGES_PER_DEVICE, 'api-limit.transport-messages-per-device'], + [LimitedApi.EDGE_EVENTS, 'api-limit.edge-events'], + [LimitedApi.EDGE_EVENTS_PER_EDGE, 'api-limit.edge-events-per-edge'], ] ); diff --git a/ui-ngx/src/app/shared/models/notification.models.ts b/ui-ngx/src/app/shared/models/notification.models.ts index 9943bebf3d..ae945b243c 100644 --- a/ui-ngx/src/app/shared/models/notification.models.ts +++ b/ui-ngx/src/app/shared/models/notification.models.ts @@ -474,7 +474,9 @@ export enum NotificationType { API_USAGE_LIMIT = 'API_USAGE_LIMIT', NEW_PLATFORM_VERSION = 'NEW_PLATFORM_VERSION', RULE_NODE = 'RULE_NODE', - RATE_LIMITS = 'RATE_LIMITS' + RATE_LIMITS = 'RATE_LIMITS', + EDGE_CONNECTION = 'EDGE_CONNECTION', + EDGE_COMMUNICATION_FAILURE = 'EDGE_COMMUNICATION_FAILURE' } export const NotificationTypeIcons = new Map([ @@ -585,6 +587,18 @@ export const NotificationTemplateTypeTranslateMap = new Map([ @@ -612,6 +628,8 @@ export const TriggerTypeTranslationMap = new Map([ [TriggerType.API_USAGE_LIMIT, 'notification.trigger.api-usage-limit'], [TriggerType.NEW_PLATFORM_VERSION, 'notification.trigger.new-platform-version'], [TriggerType.RATE_LIMITS, 'notification.trigger.rate-limits'], + [TriggerType.EDGE_CONNECTION, 'notification.trigger.edge-connection'], + [TriggerType.EDGE_COMMUNICATION_FAILURE, 'notification.trigger.edge-communication-failure'] ]); export interface NotificationUserSettings { diff --git a/ui-ngx/src/assets/help/en_US/notification/edge_communication_failure.md b/ui-ngx/src/assets/help/en_US/notification/edge_communication_failure.md new file mode 100644 index 0000000000..712f2b45e7 --- /dev/null +++ b/ui-ngx/src/assets/help/en_US/notification/edge_communication_failure.md @@ -0,0 +1,57 @@ +#### Edge communication failure notification templatization + +
+
+ +Notification subject and message fields support templatization. +The list of available templatization parameters depends on the template type. +See the available types and parameters below: + +Available template parameters: + +* `edgeId` - the edge id as uuid string; +* `edgeName` - the name of the edge; +* `failureMsg` - the string representation of the failure, occurred on the Edge; + +Parameter names must be wrapped using `${...}`. For example: `${edgeName}`. +You may also modify the value of the parameter with one of the suffixes: + +* `upperCase`, for example - `${edgeName:upperCase}` +* `lowerCase`, for example - `${edgeName:lowerCase}` +* `capitalize`, for example - `${edgeName:capitalize}` + +
+ +##### Examples + +Let's assume the notification about the failing of processing connection to Edge. +The following template: + +```text +Edge '${edgeName}' communication failure occurred +{:copy-code} +``` + +will be transformed to: + +```text +Edge 'DatacenterEdge' communication failure occurred +``` + +
+ +The following template: + +```text +Failure message: '${failureMsg}' +{:copy-code} +``` + +will be transformed to: + +```text +Failure message: 'Failed to process edge connection!' +``` + +
+
diff --git a/ui-ngx/src/assets/help/en_US/notification/edge_connection.md b/ui-ngx/src/assets/help/en_US/notification/edge_connection.md new file mode 100644 index 0000000000..37f0ec7573 --- /dev/null +++ b/ui-ngx/src/assets/help/en_US/notification/edge_connection.md @@ -0,0 +1,44 @@ +#### Edge connection notification templatization + +
+
+ +Notification subject and message fields support templatization. +The list of available templatization parameters depends on the template type. +See the available types and parameters below: + +Available template parameters: + +* `edgeId` - the edge id as uuid string; +* `edgeName` - the name of the edge; +* `eventType` - the string representation of the connectivity status: connected or disconnected; + +Parameter names must be wrapped using `${...}`. For example: `${edgeName}`. +You may also modify the value of the parameter with one of the suffixes: + +* `upperCase`, for example - `${edgeName:upperCase}` +* `lowerCase`, for example - `${edgeName:lowerCase}` +* `capitalize`, for example - `${edgeName:capitalize}` + +
+ +##### Examples + +Let's assume the notification about the connecting Edge into the ThingsBoard. +The following template: + +```text +Edge '${edgeName}' is now ${eventType} +{:copy-code} +``` + +will be transformed to: + +```text +Edge 'DatacenterEdge' is now connected +``` + +
+ +
+
diff --git a/ui-ngx/src/assets/help/en_US/notification/rate_limits.md b/ui-ngx/src/assets/help/en_US/notification/rate_limits.md index 8f6319ad80..d314e681c2 100644 --- a/ui-ngx/src/assets/help/en_US/notification/rate_limits.md +++ b/ui-ngx/src/assets/help/en_US/notification/rate_limits.md @@ -11,7 +11,7 @@ Available template parameters: * `api` - rate-limited API label; one of: 'REST API requests', 'REST API requests per customer', 'transport messages', 'transport messages per device', 'Cassandra queries', 'WS updates per session', 'notification requests', 'notification requests per rule', - 'entity version creation', 'entity version load'; + 'entity version creation', 'entity version load', 'Edge events', 'Edge events per edge'; * `limitLevelEntityType` - entity type of the limit level entity, e.g. 'Tenant', 'Device', 'Notification rule', 'Customer', etc.; * `limitLevelEntityId` - id of the limit level entity; * `limitLevelEntityName` - name of the limit level entity; diff --git a/ui-ngx/src/assets/locale/locale.constant-en_US.json b/ui-ngx/src/assets/locale/locale.constant-en_US.json index 92e4ab9895..b3d2821e30 100644 --- a/ui-ngx/src/assets/locale/locale.constant-en_US.json +++ b/ui-ngx/src/assets/locale/locale.constant-en_US.json @@ -856,7 +856,9 @@ "rest-api-requests-per-customer": "REST API requests per customer", "transport-messages": "Transport messages", "transport-messages-per-device": "Transport messages per device", - "ws-updates-per-session": "WS updates per session" + "ws-updates-per-session": "WS updates per session", + "edge-events": "Edge events", + "edge-events-per-edge": "Edge events per edge" }, "audit-log": { "audit": "Audit", @@ -2037,7 +2039,9 @@ "missing-related-rule-chains-title": "Edge has missing related rule chain(s)", "missing-related-rule-chains-text": "Assigned to edge rule chain(s) use rule nodes that forward message(s) to rule chain(s) that are not assigned to this edge.

List of missing rule chain(s):
{{missingRuleChains}}", "upgrade-instructions": "Upgrade Instructions", - "widget-datasource-error": "This widget supports only EDGE entity datasource" + "widget-datasource-error": "This widget supports only EDGE entity datasource", + "connected": "Connected", + "disconnected": "Disconnected" }, "edge-event": { "type-dashboard": "Dashboard", @@ -3288,6 +3292,8 @@ "device-list-rule-hint": "If the field is empty, the trigger will be applied to all devices", "device-profiles-list-rule-hint": "If the field is empty, the trigger will be applied to all device profiles", "disabled": "Disabled", + "edge-trigger-settings": "Edge trigger settings", + "edge-list-rule-hint": "If the field is empty, the trigger will be applied to all edge instances", "edit-notification-recipients-group": "Edit notification recipients group", "edit-notification-template": "Edit notification template", "edit-rule": "Edit rule", @@ -3432,7 +3438,9 @@ "rule-engine-lifecycle-event": "Rule engine lifecycle event", "rule-node": "Rule node", "new-platform-version": "New platform version", - "rate-limits": "Exceeded rate limits" + "rate-limits": "Exceeded rate limits", + "edge-communication-failure": "Edge communication failure", + "edge-connection": "Edge connection" }, "templates": "Templates", "notification-templates": "Notifications / Templates", @@ -3453,6 +3461,8 @@ "rule-engine-lifecycle-event": "Rule engine lifecycle event", "new-platform-version": "New platform version", "rate-limits": "Exceeded rate limits", + "edge-connection": "Edge connection", + "edge-communication-failure": "Edge communication failure", "trigger": "Trigger", "trigger-required": "Trigger is required" }, @@ -4185,6 +4195,10 @@ "edit-tenant-entity-import-rate-limit-title": "Edit entity version load rate limits", "edit-tenant-notification-request-rate-limit-title": "Edit notification requests rate limits", "edit-tenant-notification-requests-per-rule-rate-limit-title": "Edit notification requests per notification rule rate limits", + "edit-edge-events-rate-limit": "Edit edge events rate limits", + "edit-edge-events-per-edge-rate-limit": "Edit edge events per edge rate limits", + "edge-events-rate-limit": "Edge events", + "edge-events-per-edge-rate-limit": "Edge events per edge", "messages-per": "messages per", "not-set": "Not set", "number-of-messages": "Number of messages",