diff --git a/application/src/main/java/org/thingsboard/server/service/edge/DefaultEdgeNotificationService.java b/application/src/main/java/org/thingsboard/server/service/edge/DefaultEdgeNotificationService.java index 69852e6c05..84e1faa6cb 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/DefaultEdgeNotificationService.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/DefaultEdgeNotificationService.java @@ -16,16 +16,13 @@ package org.thingsboard.server.service.edge; import com.fasterxml.jackson.databind.node.ObjectNode; -import com.google.common.util.concurrent.FutureCallback; -import com.google.common.util.concurrent.Futures; -import com.google.common.util.concurrent.ListenableFuture; import lombok.extern.slf4j.Slf4j; -import org.checkerframework.checker.nullness.qual.Nullable; import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Value; import org.springframework.context.ApplicationEventPublisher; import org.springframework.stereotype.Service; import org.thingsboard.common.util.JacksonUtil; -import org.thingsboard.common.util.ThingsBoardThreadFactory; +import org.thingsboard.common.util.ThingsBoardExecutors; import org.thingsboard.server.common.data.audit.ActionType; import org.thingsboard.server.common.data.edge.Edge; import org.thingsboard.server.common.data.edge.EdgeEventType; @@ -59,7 +56,7 @@ import javax.annotation.PostConstruct; import javax.annotation.PreDestroy; import java.util.UUID; import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; @Service @TbCoreComponent @@ -128,17 +125,20 @@ public class DefaultEdgeNotificationService implements EdgeNotificationService { @Autowired protected ApplicationEventPublisher eventPublisher; - private ExecutorService dbCallBackExecutor; + @Value("${actors.system.edge_dispatcher_pool_size:4}") + private int edgeDispatcherSize; + + private ExecutorService executor; @PostConstruct public void initExecutor() { - dbCallBackExecutor = Executors.newSingleThreadExecutor(ThingsBoardThreadFactory.forName("edge-notifications")); + executor = ThingsBoardExecutors.newWorkStealingPool(edgeDispatcherSize, "edge-notifications"); } @PreDestroy public void shutdownExecutor() { - if (dbCallBackExecutor != null) { - dbCallBackExecutor.shutdownNow(); + if (executor != null) { + executor.shutdownNow(); } } @@ -157,79 +157,78 @@ public class DefaultEdgeNotificationService implements EdgeNotificationService { public void pushNotificationToEdge(TransportProtos.EdgeNotificationMsgProto edgeNotificationMsg, TbCallback callback) { TenantId tenantId = TenantId.fromUUID(new UUID(edgeNotificationMsg.getTenantIdMSB(), edgeNotificationMsg.getTenantIdLSB())); log.debug("[{}] Pushing notification to edge {}", tenantId, edgeNotificationMsg); + final long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(60); try { - EdgeEventType type = EdgeEventType.valueOf(edgeNotificationMsg.getType()); - ListenableFuture future; - switch (type) { - case EDGE: - future = edgeProcessor.processEdgeNotification(tenantId, edgeNotificationMsg); - break; - case ASSET: - future = assetProcessor.processEntityNotification(tenantId, edgeNotificationMsg); - break; - case DEVICE: - future = deviceProcessor.processEntityNotification(tenantId, edgeNotificationMsg); - break; - case ENTITY_VIEW: - future = entityViewProcessor.processEntityNotification(tenantId, edgeNotificationMsg); - break; - case DASHBOARD: - future = dashboardProcessor.processEntityNotification(tenantId, edgeNotificationMsg); - break; - case RULE_CHAIN: - future = ruleChainProcessor.processEntityNotification(tenantId, edgeNotificationMsg); - break; - case USER: - future = userProcessor.processEntityNotification(tenantId, edgeNotificationMsg); - break; - case CUSTOMER: - future = customerProcessor.processCustomerNotification(tenantId, edgeNotificationMsg); - break; - case DEVICE_PROFILE: - future = deviceProfileProcessor.processEntityNotification(tenantId, edgeNotificationMsg); - break; - case ASSET_PROFILE: - future = assetProfileProcessor.processEntityNotification(tenantId, edgeNotificationMsg); - break; - case OTA_PACKAGE: - future = otaPackageProcessor.processEntityNotification(tenantId, edgeNotificationMsg); - break; - case WIDGETS_BUNDLE: - future = widgetBundleProcessor.processEntityNotification(tenantId, edgeNotificationMsg); - break; - case WIDGET_TYPE: - future = widgetTypeProcessor.processEntityNotification(tenantId, edgeNotificationMsg); - break; - case QUEUE: - future = queueProcessor.processEntityNotification(tenantId, edgeNotificationMsg); - break; - case ALARM: - future = alarmProcessor.processAlarmNotification(tenantId, edgeNotificationMsg); - break; - case RELATION: - future = relationProcessor.processRelationNotification(tenantId, edgeNotificationMsg); - break; - case TENANT: - future = tenantEdgeProcessor.processEntityNotification(tenantId, edgeNotificationMsg); - break; - case TENANT_PROFILE: - future = tenantProfileEdgeProcessor.processEntityNotification(tenantId, edgeNotificationMsg); - break; - default: - log.warn("[{}] Edge event type [{}] is not designed to be pushed to edge", tenantId, type); - future = Futures.immediateFuture(null); - } - Futures.addCallback(future, new FutureCallback<>() { - @Override - public void onSuccess(@Nullable Void unused) { - callback.onSuccess(); - } - - @Override - public void onFailure(Throwable throwable) { - callBackFailure(tenantId, edgeNotificationMsg, callback, throwable); + executor.submit(() -> { + try { + if (deadline < System.nanoTime()) { + log.warn("[{}] Skipping notification message because deadline reached {}", tenantId, edgeNotificationMsg); + return; + } + EdgeEventType type = EdgeEventType.valueOf(edgeNotificationMsg.getType()); + switch (type) { + case EDGE: + edgeProcessor.processEdgeNotification(tenantId, edgeNotificationMsg); + break; + case ASSET: + assetProcessor.processEntityNotification(tenantId, edgeNotificationMsg); + break; + case DEVICE: + deviceProcessor.processEntityNotification(tenantId, edgeNotificationMsg); + break; + case ENTITY_VIEW: + entityViewProcessor.processEntityNotification(tenantId, edgeNotificationMsg); + break; + case DASHBOARD: + dashboardProcessor.processEntityNotification(tenantId, edgeNotificationMsg); + break; + case RULE_CHAIN: + ruleChainProcessor.processEntityNotification(tenantId, edgeNotificationMsg); + break; + case USER: + userProcessor.processEntityNotification(tenantId, edgeNotificationMsg); + break; + case CUSTOMER: + customerProcessor.processCustomerNotification(tenantId, edgeNotificationMsg); + break; + case DEVICE_PROFILE: + deviceProfileProcessor.processEntityNotification(tenantId, edgeNotificationMsg); + break; + case ASSET_PROFILE: + assetProfileProcessor.processEntityNotification(tenantId, edgeNotificationMsg); + break; + case OTA_PACKAGE: + otaPackageProcessor.processEntityNotification(tenantId, edgeNotificationMsg); + break; + case WIDGETS_BUNDLE: + widgetBundleProcessor.processEntityNotification(tenantId, edgeNotificationMsg); + break; + case WIDGET_TYPE: + widgetTypeProcessor.processEntityNotification(tenantId, edgeNotificationMsg); + break; + case QUEUE: + queueProcessor.processEntityNotification(tenantId, edgeNotificationMsg); + break; + case ALARM: + alarmProcessor.processAlarmNotification(tenantId, edgeNotificationMsg); + break; + case RELATION: + relationProcessor.processRelationNotification(tenantId, edgeNotificationMsg); + break; + case TENANT: + tenantEdgeProcessor.processEntityNotification(tenantId, edgeNotificationMsg); + break; + case TENANT_PROFILE: + tenantProfileEdgeProcessor.processEntityNotification(tenantId, edgeNotificationMsg); + break; + default: + log.warn("[{}] Edge event type [{}] is not designed to be pushed to edge", tenantId, type); + } + } catch (Exception e) { + callBackFailure(tenantId, edgeNotificationMsg, callback, e); } - }, dbCallBackExecutor); + }); + callback.onSuccess(); } catch (Exception e) { callBackFailure(tenantId, edgeNotificationMsg, callback, e); } diff --git a/application/src/main/java/org/thingsboard/server/service/notification/rule/DefaultNotificationRuleProcessor.java b/application/src/main/java/org/thingsboard/server/service/notification/rule/DefaultNotificationRuleProcessor.java index e9656f2ea6..70ba344b84 100644 --- a/application/src/main/java/org/thingsboard/server/service/notification/rule/DefaultNotificationRuleProcessor.java +++ b/application/src/main/java/org/thingsboard/server/service/notification/rule/DefaultNotificationRuleProcessor.java @@ -77,18 +77,17 @@ public class DefaultNotificationRuleProcessor implements NotificationRuleProcess public void process(NotificationRuleTrigger trigger) { NotificationRuleTriggerType triggerType = trigger.getType(); TenantId tenantId = triggerType.isTenantLevel() ? trigger.getTenantId() : TenantId.SYS_TENANT_ID; - - try { - List enabledRules = notificationRulesCache.getEnabled(tenantId, triggerType); - if (enabledRules.isEmpty()) { - return; - } - if (trigger.deduplicate()) { - enabledRules = new ArrayList<>(enabledRules); - enabledRules.removeIf(rule -> deduplicationService.alreadyProcessed(trigger, rule)); - } - final List rules = enabledRules; - notificationExecutor.submit(() -> { + notificationExecutor.submit(() -> { + try { + List enabledRules = notificationRulesCache.getEnabled(tenantId, triggerType); + if (enabledRules.isEmpty()) { + return; + } + if (trigger.deduplicate()) { + enabledRules = new ArrayList<>(enabledRules); + enabledRules.removeIf(rule -> deduplicationService.alreadyProcessed(trigger, rule)); + } + final List rules = enabledRules; for (NotificationRule rule : rules) { try { processNotificationRule(rule, trigger); @@ -96,10 +95,10 @@ public class DefaultNotificationRuleProcessor implements NotificationRuleProcess log.error("Failed to process notification rule {} for trigger type {} with trigger object {}", rule.getId(), rule.getTriggerType(), trigger, e); } } - }); - } catch (Throwable e) { - log.error("Failed to process notification rules for trigger: {}", trigger, e); - } + } catch (Throwable e) { + log.error("Failed to process notification rules for trigger: {}", trigger, e); + } + }); } private void processNotificationRule(NotificationRule rule, NotificationRuleTrigger trigger) { diff --git a/application/src/main/java/org/thingsboard/server/service/queue/TbCoreConsumerStats.java b/application/src/main/java/org/thingsboard/server/service/queue/TbCoreConsumerStats.java index 6f614957c4..4d195a0dec 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/TbCoreConsumerStats.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/TbCoreConsumerStats.java @@ -36,10 +36,22 @@ public class TbCoreConsumerStats { public static final String DEVICE_CLAIMS = "claimDevice"; public static final String DEVICE_STATES = "deviceState"; public static final String SUBSCRIPTION_MSGS = "subMsgs"; - public static final String TO_CORE_NOTIFICATIONS = "coreNfs"; public static final String EDGE_NOTIFICATIONS = "edgeNfs"; public static final String DEVICE_ACTIVITIES = "deviceActivity"; + public static final String TO_CORE_NF_OTHER = "coreNfOther"; // normally, there is no messages when codebase is fine + public static final String TO_CORE_NF_COMPONENT_LIFECYCLE = "coreNfCompLfcl"; + public static final String TO_CORE_NF_DEVICE_RPC_RESPONSE = "coreNfDevRpcRsp"; + public static final String TO_CORE_NF_EDGE_EVENT_UPDATE = "coreNfEdgeUpd"; + public static final String TO_CORE_NF_EDGE_SYNC_REQUEST = "coreNfEdgeSyncReq"; + public static final String TO_CORE_NF_EDGE_SYNC_RESPONSE = "coreNfEdgeSyncResp"; + public static final String TO_CORE_NF_NOTIFICATION_RULE_PROCESSOR = "coreNfNfRlProc"; + public static final String TO_CORE_NF_QUEUE_UPDATE = "coreNfQueueUpd"; + public static final String TO_CORE_NF_QUEUE_DELETE = "coreNfQueueDel"; + public static final String TO_CORE_NF_SUBSCRIPTION_SERVICE = "coreNfSubSvc"; + public static final String TO_CORE_NF_SUBSCRIPTION_MANAGER = "coreNfSubMgr"; + public static final String TO_CORE_NF_VC_RESPONSE = "coreNfVCRsp"; + private final StatsCounter totalCounter; private final StatsCounter sessionEventCounter; private final StatsCounter getAttributesCounter; @@ -48,14 +60,25 @@ public class TbCoreConsumerStats { private final StatsCounter toDeviceRPCCallResponseCounter; private final StatsCounter subscriptionInfoCounter; private final StatsCounter claimDeviceCounter; - private final StatsCounter deviceStateCounter; private final StatsCounter subscriptionMsgCounter; - private final StatsCounter toCoreNotificationsCounter; private final StatsCounter edgeNotificationsCounter; private final StatsCounter deviceActivitiesCounter; - private final List counters = new ArrayList<>(); + private final StatsCounter toCoreNfOtherCounter; + private final StatsCounter toCoreNfComponentLifecycleCounter; + private final StatsCounter toCoreNfDeviceRpcResponseCounter; + private final StatsCounter toCoreNfEdgeEventUpdateCounter; + private final StatsCounter toCoreNfEdgeSyncRequestCounter; + private final StatsCounter toCoreNfEdgeSyncResponseCounter; + private final StatsCounter toCoreNfNotificationRuleProcessorCounter; + private final StatsCounter toCoreNfQueueUpdateCounter; + private final StatsCounter toCoreNfQueueDeleteCounter; + private final StatsCounter toCoreNfSubscriptionServiceCounter; + private final StatsCounter toCoreNfSubscriptionManagerCounter; + private final StatsCounter toCoreNfVersionControlResponseCounter; + + private final List counters = new ArrayList<>(24); public TbCoreConsumerStats(StatsFactory statsFactory) { String statsKey = StatsType.CORE.getName(); @@ -70,9 +93,23 @@ public class TbCoreConsumerStats { this.claimDeviceCounter = register(statsFactory.createStatsCounter(statsKey, DEVICE_CLAIMS)); this.deviceStateCounter = register(statsFactory.createStatsCounter(statsKey, DEVICE_STATES)); this.subscriptionMsgCounter = register(statsFactory.createStatsCounter(statsKey, SUBSCRIPTION_MSGS)); - this.toCoreNotificationsCounter = register(statsFactory.createStatsCounter(statsKey, TO_CORE_NOTIFICATIONS)); this.edgeNotificationsCounter = register(statsFactory.createStatsCounter(statsKey, EDGE_NOTIFICATIONS)); this.deviceActivitiesCounter = register(statsFactory.createStatsCounter(statsKey, DEVICE_ACTIVITIES)); + + // Core notification counters + this.toCoreNfOtherCounter = register(statsFactory.createStatsCounter(statsKey, TO_CORE_NF_OTHER)); + this.toCoreNfComponentLifecycleCounter = register(statsFactory.createStatsCounter(statsKey, TO_CORE_NF_COMPONENT_LIFECYCLE)); + this.toCoreNfDeviceRpcResponseCounter = register(statsFactory.createStatsCounter(statsKey, TO_CORE_NF_DEVICE_RPC_RESPONSE)); + this.toCoreNfEdgeEventUpdateCounter = register(statsFactory.createStatsCounter(statsKey, TO_CORE_NF_EDGE_EVENT_UPDATE)); + this.toCoreNfEdgeSyncRequestCounter = register(statsFactory.createStatsCounter(statsKey, TO_CORE_NF_EDGE_SYNC_REQUEST)); + this.toCoreNfEdgeSyncResponseCounter = register(statsFactory.createStatsCounter(statsKey, TO_CORE_NF_EDGE_SYNC_RESPONSE)); + this.toCoreNfNotificationRuleProcessorCounter = register(statsFactory.createStatsCounter(statsKey, TO_CORE_NF_NOTIFICATION_RULE_PROCESSOR)); + this.toCoreNfQueueUpdateCounter = register(statsFactory.createStatsCounter(statsKey, TO_CORE_NF_QUEUE_UPDATE)); + this.toCoreNfQueueDeleteCounter = register(statsFactory.createStatsCounter(statsKey, TO_CORE_NF_QUEUE_DELETE)); + this.toCoreNfSubscriptionServiceCounter = register(statsFactory.createStatsCounter(statsKey, TO_CORE_NF_SUBSCRIPTION_SERVICE)); + this.toCoreNfSubscriptionManagerCounter = register(statsFactory.createStatsCounter(statsKey, TO_CORE_NF_SUBSCRIPTION_MANAGER)); + this.toCoreNfVersionControlResponseCounter = register(statsFactory.createStatsCounter(statsKey, TO_CORE_NF_VC_RESPONSE)); + } private StatsCounter register(StatsCounter counter){ @@ -127,7 +164,31 @@ public class TbCoreConsumerStats { public void log(TransportProtos.ToCoreNotificationMsg msg) { totalCounter.increment(); - toCoreNotificationsCounter.increment(); + if (msg.hasToLocalSubscriptionServiceMsg()) { + toCoreNfSubscriptionServiceCounter.increment(); + } else if (msg.hasFromDeviceRpcResponse()) { + toCoreNfDeviceRpcResponseCounter.increment(); + } else if (!msg.getComponentLifecycleMsg().isEmpty()) { + toCoreNfComponentLifecycleCounter.increment(); + } else if (!msg.getEdgeEventUpdateMsg().isEmpty()) { + toCoreNfEdgeEventUpdateCounter.increment(); + } else if (!msg.getToEdgeSyncRequestMsg().isEmpty()) { + toCoreNfEdgeSyncRequestCounter.increment(); + } else if (!msg.getFromEdgeSyncResponseMsg().isEmpty()) { + toCoreNfEdgeSyncResponseCounter.increment(); + } else if (msg.hasQueueUpdateMsg()) { + toCoreNfQueueUpdateCounter.increment(); + } else if (msg.hasQueueDeleteMsg()) { + toCoreNfQueueDeleteCounter.increment(); + } else if (msg.hasVcResponseMsg()) { + toCoreNfVersionControlResponseCounter.increment(); + } else if (msg.hasToSubscriptionMgrMsg()) { + toCoreNfSubscriptionManagerCounter.increment(); + } else if (msg.hasNotificationRuleProcessorMsg()) { + toCoreNfNotificationRuleProcessorCounter.increment(); + } else { + toCoreNfOtherCounter.increment(); + } } public void printStats() { diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml index b73c4f1fda..bc53175839 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/application/src/main/resources/thingsboard.yml @@ -418,6 +418,7 @@ actors: tenant_dispatcher_pool_size: "${ACTORS_SYSTEM_TENANT_DISPATCHER_POOL_SIZE:2}" # Thread pool size for actor system dispatcher that process messages for tenant actors device_dispatcher_pool_size: "${ACTORS_SYSTEM_DEVICE_DISPATCHER_POOL_SIZE:4}" # Thread pool size for actor system dispatcher that process messages for device actors rule_dispatcher_pool_size: "${ACTORS_SYSTEM_RULE_DISPATCHER_POOL_SIZE:8}" # Thread pool size for actor system dispatcher that process messages for rule engine (chain/node) actors + edge_dispatcher_pool_size: "${ACTORS_SYSTEM_EDGE_DISPATCHER_POOL_SIZE:4}" # Thread pool size for actor system dispatcher that process messages for edge actors tenant: create_components_on_init: "${ACTORS_TENANT_CREATE_COMPONENTS_ON_INIT:true}" # Create components in initialization session: diff --git a/application/src/test/resources/application-test.properties b/application/src/test/resources/application-test.properties index d8943b47e9..b01ffe0d72 100644 --- a/application/src/test/resources/application-test.properties +++ b/application/src/test/resources/application-test.properties @@ -12,7 +12,6 @@ transport.lwm2m.security.trust-credentials.keystore.store_file=lwm2m/credentials # Edge disabled to speed up the context init. Will be enabled by @TestPropertySource in respective tests edges.enabled=false actors.rpc.submit_strategy=BURST -queue.rule-engine.stats.enabled=true # Transports disabled to speed up the context init. Particular transport will be enabled with @TestPropertySource in respective tests transport.http.enabled=false @@ -59,7 +58,9 @@ queue.rule-engine.queues[2].processing-strategy.retries=1 queue.rule-engine.queues[2].processing-strategy.pause-between-retries=0 queue.rule-engine.queues[2].processing-strategy.max-pause-between-retries=0 +queue.rule-engine.stats.enabled=true usage.stats.report.enabled=false +queue.core.stats.enabled=true sql.audit_logs.partition_size=24 sql.ttl.audit_logs.ttl=2592000 diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaProducerTemplate.java b/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaProducerTemplate.java index 850f5c08ea..d67d2c4399 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaProducerTemplate.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaProducerTemplate.java @@ -85,8 +85,8 @@ public class TbKafkaProducerTemplate implements TbQueuePro if (log.isTraceEnabled()) { try { StackTraceElement[] stackTrace = Thread.currentThread().getStackTrace(); - int maxlevel = Math.min(stackTrace.length, 10); - for (int i = 2; i < maxlevel; i++) { // ignore two levels: getStackTrace and addAnalyticHeaders + int maxLevel = Math.min(stackTrace.length, 20); + for (int i = 2; i < maxLevel; i++) { // ignore two levels: getStackTrace and addAnalyticHeaders headers.add(new RecordHeader("_stackTrace" + i, stackTrace[i].toString().getBytes(StandardCharsets.UTF_8))); } } catch (Throwable t) { diff --git a/dao/src/main/java/org/thingsboard/server/dao/tenant/TenantServiceImpl.java b/dao/src/main/java/org/thingsboard/server/dao/tenant/TenantServiceImpl.java index 0f1b60ecb5..e10a80af2d 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/tenant/TenantServiceImpl.java +++ b/dao/src/main/java/org/thingsboard/server/dao/tenant/TenantServiceImpl.java @@ -197,7 +197,7 @@ public class TenantServiceImpl extends AbstractCachedEntityService this.commandForm.get('result').setValue(JSON.stringify(resp)), error: error => { console.log(error);