diff --git a/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainActorMessageProcessor.java b/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainActorMessageProcessor.java index cc0c9da8f2..830714b431 100644 --- a/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainActorMessageProcessor.java +++ b/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainActorMessageProcessor.java @@ -35,6 +35,7 @@ import org.thingsboard.server.common.data.rule.RuleChain; import org.thingsboard.server.common.data.rule.RuleNode; import org.thingsboard.server.common.msg.TbMsg; import org.thingsboard.server.common.msg.plugin.ComponentLifecycleMsg; +import org.thingsboard.server.common.msg.plugin.RuleNodeUpdatedMsg; import org.thingsboard.server.common.msg.queue.PartitionChangeMsg; import org.thingsboard.server.common.msg.queue.QueueToRuleEngineMsg; import org.thingsboard.server.common.msg.queue.RuleEngineException; @@ -131,7 +132,7 @@ public class RuleChainActorMessageProcessor extends ComponentMsgProcessor implements ActorService { public static final String APP_DISPATCHER_NAME = "app-dispatcher"; public static final String TENANT_DISPATCHER_NAME = "tenant-dispatcher"; @@ -120,10 +121,10 @@ public class DefaultActorService implements ActorService { appActor.tellWithHighPriority(new AppInitMsg()); } - @EventListener(PartitionChangeEvent.class) - public void onApplicationEvent(PartitionChangeEvent partitionChangeEvent) { + @Override + protected void onTbApplicationEvent(PartitionChangeEvent event) { log.info("Received partition change event."); - this.appActor.tellWithHighPriority(new PartitionChangeMsg(partitionChangeEvent.getServiceQueueKey(), partitionChangeEvent.getPartitions())); + this.appActor.tellWithHighPriority(new PartitionChangeMsg(event.getServiceQueueKey(), event.getPartitions())); } @PreDestroy diff --git a/application/src/main/java/org/thingsboard/server/controller/BaseController.java b/application/src/main/java/org/thingsboard/server/controller/BaseController.java index bc724f4492..d6c04eb8a5 100644 --- a/application/src/main/java/org/thingsboard/server/controller/BaseController.java +++ b/application/src/main/java/org/thingsboard/server/controller/BaseController.java @@ -27,7 +27,22 @@ import org.springframework.beans.factory.annotation.Value; import org.springframework.security.core.Authentication; import org.springframework.security.core.context.SecurityContextHolder; import org.springframework.web.bind.annotation.ExceptionHandler; -import org.thingsboard.server.common.data.*; +import org.thingsboard.server.common.data.Customer; +import org.thingsboard.server.common.data.Dashboard; +import org.thingsboard.server.common.data.DashboardInfo; +import org.thingsboard.server.common.data.DataConstants; +import org.thingsboard.server.common.data.Device; +import org.thingsboard.server.common.data.DeviceInfo; +import org.thingsboard.server.common.data.DeviceProfile; +import org.thingsboard.server.common.data.EntityType; +import org.thingsboard.server.common.data.EntityView; +import org.thingsboard.server.common.data.EntityViewInfo; +import org.thingsboard.server.common.data.HasName; +import org.thingsboard.server.common.data.HasTenantId; +import org.thingsboard.server.common.data.Tenant; +import org.thingsboard.server.common.data.TenantInfo; +import org.thingsboard.server.common.data.TenantProfile; +import org.thingsboard.server.common.data.User; import org.thingsboard.server.common.data.alarm.Alarm; import org.thingsboard.server.common.data.alarm.AlarmInfo; import org.thingsboard.server.common.data.asset.Asset; @@ -84,6 +99,7 @@ import org.thingsboard.server.dao.oauth2.OAuth2ConfigTemplateService; import org.thingsboard.server.dao.oauth2.OAuth2Service; import org.thingsboard.server.dao.relation.RelationService; import org.thingsboard.server.dao.rule.RuleChainService; +import org.thingsboard.server.dao.tenant.TbTenantProfileCache; import org.thingsboard.server.dao.tenant.TenantProfileService; import org.thingsboard.server.dao.tenant.TenantService; import org.thingsboard.server.dao.user.UserService; @@ -96,7 +112,6 @@ import org.thingsboard.server.queue.util.TbCoreComponent; import org.thingsboard.server.service.component.ComponentDiscoveryService; import org.thingsboard.server.service.lwm2m.LwM2MModelsRepository; import org.thingsboard.server.service.profile.TbDeviceProfileCache; -import org.thingsboard.server.dao.tenant.TbTenantProfileCache; import org.thingsboard.server.service.queue.TbClusterService; import org.thingsboard.server.service.security.model.SecurityUser; import org.thingsboard.server.service.security.permission.AccessControlService; @@ -124,6 +139,9 @@ public abstract class BaseController { public static final String INCORRECT_TENANT_ID = "Incorrect tenantId "; public static final String YOU_DON_T_HAVE_PERMISSION_TO_PERFORM_THIS_OPERATION = "You don't have permission to perform this operation!"; + protected static final String DEFAULT_DASHBOARD = "defaultDashboardId"; + protected static final String HOME_DASHBOARD = "homeDashboardId"; + private static final ObjectMapper json = new ObjectMapper(); @Autowired @@ -766,7 +784,7 @@ public abstract class BaseController { String scope = extractParameter(String.class, 0, additionalInfo); @SuppressWarnings("unchecked") List attributes = extractParameter(List.class, 1, additionalInfo); - metaData.putValue("scope", scope); + metaData.putValue(DataConstants.SCOPE, scope); if (attributes != null) { for (AttributeKvEntry attr : attributes) { addKvEntry(entityNode, attr); @@ -776,7 +794,7 @@ public abstract class BaseController { String scope = extractParameter(String.class, 0, additionalInfo); @SuppressWarnings("unchecked") List keys = extractParameter(List.class, 1, additionalInfo); - metaData.putValue("scope", scope); + metaData.putValue(DataConstants.SCOPE, scope); ArrayNode attrsArrayNode = entityNode.putArray("attributes"); if (keys != null) { keys.forEach(attrsArrayNode::add); @@ -862,4 +880,14 @@ public abstract class BaseController { } } } + + protected void processDashboardIdFromAdditionalInfo(ObjectNode additionalInfo, String requiredFields) throws ThingsboardException { + String dashboardId = additionalInfo.has(requiredFields) ? additionalInfo.get(requiredFields).asText() : null; + if(dashboardId != null && !dashboardId.equals("null")) { + if(dashboardService.findDashboardById(getTenantId(), new DashboardId(UUID.fromString(dashboardId))) == null) { + additionalInfo.remove(requiredFields); + } + } + } + } diff --git a/application/src/main/java/org/thingsboard/server/controller/CustomerController.java b/application/src/main/java/org/thingsboard/server/controller/CustomerController.java index 1a2357450c..dd0312169b 100644 --- a/application/src/main/java/org/thingsboard/server/controller/CustomerController.java +++ b/application/src/main/java/org/thingsboard/server/controller/CustomerController.java @@ -55,7 +55,11 @@ public class CustomerController extends BaseController { checkParameter(CUSTOMER_ID, strCustomerId); try { CustomerId customerId = new CustomerId(toUUID(strCustomerId)); - return checkCustomerId(customerId, Operation.READ); + Customer customer = checkCustomerId(customerId, Operation.READ); + if(!customer.getAdditionalInfo().isNull()) { + processDashboardIdFromAdditionalInfo((ObjectNode) customer.getAdditionalInfo(), HOME_DASHBOARD); + } + return customer; } catch (Exception e) { throw handleException(e); } diff --git a/application/src/main/java/org/thingsboard/server/controller/TenantController.java b/application/src/main/java/org/thingsboard/server/controller/TenantController.java index 2bd8a04578..70b5efe40d 100644 --- a/application/src/main/java/org/thingsboard/server/controller/TenantController.java +++ b/application/src/main/java/org/thingsboard/server/controller/TenantController.java @@ -15,6 +15,7 @@ */ package org.thingsboard.server.controller; +import com.fasterxml.jackson.databind.node.ObjectNode; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.http.HttpStatus; @@ -59,7 +60,11 @@ public class TenantController extends BaseController { checkParameter("tenantId", strTenantId); try { TenantId tenantId = new TenantId(toUUID(strTenantId)); - return checkTenantId(tenantId, Operation.READ); + Tenant tenant = checkTenantId(tenantId, Operation.READ); + if(!tenant.getAdditionalInfo().isNull()) { + processDashboardIdFromAdditionalInfo((ObjectNode) tenant.getAdditionalInfo(), HOME_DASHBOARD); + } + return tenant; } catch (Exception e) { throw handleException(e); } diff --git a/application/src/main/java/org/thingsboard/server/controller/UserController.java b/application/src/main/java/org/thingsboard/server/controller/UserController.java index 47be552c2f..5c6e7e9fe8 100644 --- a/application/src/main/java/org/thingsboard/server/controller/UserController.java +++ b/application/src/main/java/org/thingsboard/server/controller/UserController.java @@ -53,7 +53,6 @@ import org.thingsboard.server.service.security.model.token.JwtTokenFactory; import org.thingsboard.server.service.security.permission.Operation; import org.thingsboard.server.service.security.permission.Resource; import org.thingsboard.server.service.security.system.SystemSecurityService; -import org.thingsboard.server.utils.MiscUtils; import javax.servlet.http.HttpServletRequest; @@ -90,7 +89,12 @@ public class UserController extends BaseController { checkParameter(USER_ID, strUserId); try { UserId userId = new UserId(toUUID(strUserId)); - return checkUserId(userId, Operation.READ); + User user = checkUserId(userId, Operation.READ); + if(!user.getAdditionalInfo().isNull()) { + processDashboardIdFromAdditionalInfo((ObjectNode) user.getAdditionalInfo(), DEFAULT_DASHBOARD); + processDashboardIdFromAdditionalInfo((ObjectNode) user.getAdditionalInfo(), HOME_DASHBOARD); + } + return user; } catch (Exception e) { throw handleException(e); } @@ -329,4 +333,5 @@ public class UserController extends BaseController { throw handleException(e); } } + } diff --git a/application/src/main/java/org/thingsboard/server/service/apiusage/DefaultTbApiUsageStateService.java b/application/src/main/java/org/thingsboard/server/service/apiusage/DefaultTbApiUsageStateService.java index d4a5d42320..d0a3984660 100644 --- a/application/src/main/java/org/thingsboard/server/service/apiusage/DefaultTbApiUsageStateService.java +++ b/application/src/main/java/org/thingsboard/server/service/apiusage/DefaultTbApiUsageStateService.java @@ -54,6 +54,7 @@ import org.thingsboard.server.gen.transport.TransportProtos.UsageStatsKVProto; import org.thingsboard.server.queue.common.TbProtoQueueMsg; import org.thingsboard.server.queue.discovery.PartitionChangeEvent; import org.thingsboard.server.queue.discovery.PartitionService; +import org.thingsboard.server.queue.discovery.TbApplicationEventListener; import org.thingsboard.server.queue.scheduler.SchedulerComponent; import org.thingsboard.server.service.queue.TbClusterService; import org.thingsboard.server.service.telemetry.InternalTelemetryService; @@ -78,7 +79,7 @@ import java.util.stream.Collectors; @Slf4j @Service -public class DefaultTbApiUsageStateService implements TbApiUsageStateService { +public class DefaultTbApiUsageStateService extends TbApplicationEventListener implements TbApiUsageStateService { public static final String HOURLY = "Hourly"; public static final FutureCallback VOID_CALLBACK = new FutureCallback() { @@ -188,7 +189,7 @@ public class DefaultTbApiUsageStateService implements TbApiUsageStateService { } @Override - public void onApplicationEvent(PartitionChangeEvent partitionChangeEvent) { + protected void onTbApplicationEvent(PartitionChangeEvent partitionChangeEvent) { if (partitionChangeEvent.getServiceType().equals(ServiceType.TB_CORE)) { myTenantStates.entrySet().removeIf(entry -> !partitionService.resolve(ServiceType.TB_CORE, entry.getKey(), entry.getKey()).isMyPartition()); otherTenantStates.entrySet().removeIf(entry -> partitionService.resolve(ServiceType.TB_CORE, entry.getKey(), entry.getKey()).isMyPartition()); diff --git a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java index df0e7f86b5..af9239d298 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java @@ -151,12 +151,12 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService tpi.newByTopic(usageStatsConsumer.getTopic())) diff --git a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbRuleEngineConsumerService.java b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbRuleEngineConsumerService.java index 106ca40c94..390798a3e2 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbRuleEngineConsumerService.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbRuleEngineConsumerService.java @@ -140,11 +140,11 @@ public class DefaultTbRuleEngineConsumerService extends AbstractConsumerService< } @Override - public void onApplicationEvent(PartitionChangeEvent partitionChangeEvent) { - if (partitionChangeEvent.getServiceType().equals(getServiceType())) { - ServiceQueue serviceQueue = partitionChangeEvent.getServiceQueueKey().getServiceQueue(); - log.info("[{}] Subscribing to partitions: {}", serviceQueue.getQueue(), partitionChangeEvent.getPartitions()); - consumers.get(serviceQueue.getQueue()).subscribe(partitionChangeEvent.getPartitions()); + protected void onTbApplicationEvent(PartitionChangeEvent event) { + if (event.getServiceType().equals(getServiceType())) { + ServiceQueue serviceQueue = event.getServiceQueueKey().getServiceQueue(); + log.info("[{}] Subscribing to partitions: {}", serviceQueue.getQueue(), event.getPartitions()); + consumers.get(serviceQueue.getQueue()).subscribe(event.getPartitions()); } } diff --git a/application/src/main/java/org/thingsboard/server/service/queue/processing/AbstractConsumerService.java b/application/src/main/java/org/thingsboard/server/service/queue/processing/AbstractConsumerService.java index 31d5cf47c3..02378eb557 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/processing/AbstractConsumerService.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/processing/AbstractConsumerService.java @@ -36,6 +36,7 @@ import org.thingsboard.server.queue.TbQueueConsumer; import org.thingsboard.server.queue.common.TbProtoQueueMsg; import org.thingsboard.server.queue.discovery.PartitionChangeEvent; import org.thingsboard.server.common.transport.util.DataDecodingEncodingService; +import org.thingsboard.server.queue.discovery.TbApplicationEventListener; import org.thingsboard.server.service.apiusage.TbApiUsageStateService; import org.thingsboard.server.service.profile.TbDeviceProfileCache; import org.thingsboard.server.dao.tenant.TbTenantProfileCache; @@ -56,7 +57,7 @@ import java.util.function.Function; import java.util.stream.Collectors; @Slf4j -public abstract class AbstractConsumerService implements ApplicationListener { +public abstract class AbstractConsumerService extends TbApplicationEventListener { protected volatile ExecutorService consumersExecutor; protected volatile ExecutorService notificationsConsumerExecutor; diff --git a/application/src/main/java/org/thingsboard/server/service/state/DefaultDeviceStateService.java b/application/src/main/java/org/thingsboard/server/service/state/DefaultDeviceStateService.java index b1d47c9864..be31a0df45 100644 --- a/application/src/main/java/org/thingsboard/server/service/state/DefaultDeviceStateService.java +++ b/application/src/main/java/org/thingsboard/server/service/state/DefaultDeviceStateService.java @@ -56,6 +56,7 @@ import org.thingsboard.server.dao.util.mapping.JacksonUtil; import org.thingsboard.server.gen.transport.TransportProtos; import org.thingsboard.server.queue.discovery.PartitionChangeEvent; import org.thingsboard.server.queue.discovery.PartitionService; +import org.thingsboard.server.queue.discovery.TbApplicationEventListener; import org.thingsboard.server.queue.util.TbCoreComponent; import org.thingsboard.server.service.queue.TbClusterService; import org.thingsboard.server.service.telemetry.TelemetrySubscriptionService; @@ -90,7 +91,7 @@ import static org.thingsboard.server.common.data.DataConstants.SERVER_SCOPE; @Service @TbCoreComponent @Slf4j -public class DefaultDeviceStateService implements DeviceStateService { +public class DefaultDeviceStateService extends TbApplicationEventListener implements DeviceStateService { public static final String ACTIVITY_STATE = "active"; public static final String LAST_CONNECT_TIME = "lastConnectTime"; @@ -206,7 +207,6 @@ public class DefaultDeviceStateService implements DeviceStateService { if (!state.isActive()) { state.setActive(true); save(deviceId, ACTIVITY_STATE, state.isActive()); - stateData.getMetaData().putValue("scope", SERVER_SCOPE); pushRuleEngineMessage(stateData, ACTIVITY_EVENT); } } @@ -295,7 +295,7 @@ public class DefaultDeviceStateService implements DeviceStateService { } @Override - public void onApplicationEvent(PartitionChangeEvent partitionChangeEvent) { + protected void onTbApplicationEvent(PartitionChangeEvent partitionChangeEvent) { if (ServiceType.TB_CORE.equals(partitionChangeEvent.getServiceType())) { deduplicationExecutor.submit(partitionChangeEvent.getPartitions()); } @@ -447,7 +447,7 @@ public class DefaultDeviceStateService implements DeviceStateService { } private Function, DeviceStateData> extractDeviceStateData(Device device) { - return new Function, DeviceStateData>() { + return new Function<>() { @Nullable @Override public DeviceStateData apply(@Nullable List data) { @@ -503,7 +503,11 @@ public class DefaultDeviceStateService implements DeviceStateService { } else { data = JacksonUtil.toString(state); } - TbMsg tbMsg = TbMsg.newMsg(msgType, stateData.getDeviceId(), stateData.getMetaData().copy(), TbMsgDataType.JSON, data); + TbMsgMetaData md = stateData.getMetaData().copy(); + if(!persistToTelemetry){ + md.putValue(DataConstants.SCOPE, SERVER_SCOPE); + } + TbMsg tbMsg = TbMsg.newMsg(msgType, stateData.getDeviceId(), md, TbMsgDataType.JSON, data); clusterService.pushMsgToRuleEngine(stateData.getTenantId(), stateData.getDeviceId(), tbMsg, null); } catch (Exception e) { log.warn("[{}] Failed to push inactivity alarm: {}", stateData.getDeviceId(), state, e); diff --git a/application/src/main/java/org/thingsboard/server/service/subscription/DefaultSubscriptionManagerService.java b/application/src/main/java/org/thingsboard/server/service/subscription/DefaultSubscriptionManagerService.java index bafbb45ba8..844db92de8 100644 --- a/application/src/main/java/org/thingsboard/server/service/subscription/DefaultSubscriptionManagerService.java +++ b/application/src/main/java/org/thingsboard/server/service/subscription/DefaultSubscriptionManagerService.java @@ -48,6 +48,7 @@ import org.thingsboard.server.queue.TbQueueProducer; import org.thingsboard.server.queue.common.TbProtoQueueMsg; import org.thingsboard.server.queue.discovery.PartitionChangeEvent; import org.thingsboard.server.queue.discovery.PartitionService; +import org.thingsboard.server.queue.discovery.TbApplicationEventListener; import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; import org.thingsboard.server.queue.provider.TbQueueProducerProvider; import org.thingsboard.server.queue.util.TbCoreComponent; @@ -76,7 +77,7 @@ import java.util.function.Predicate; @Slf4j @TbCoreComponent @Service -public class DefaultSubscriptionManagerService implements SubscriptionManagerService { +public class DefaultSubscriptionManagerService extends TbApplicationEventListener implements SubscriptionManagerService { @Autowired private AttributesService attrService; @@ -178,7 +179,7 @@ public class DefaultSubscriptionManagerService implements SubscriptionManagerSer } @Override - public void onApplicationEvent(PartitionChangeEvent partitionChangeEvent) { + protected void onTbApplicationEvent(PartitionChangeEvent partitionChangeEvent) { if (ServiceType.TB_CORE.equals(partitionChangeEvent.getServiceType())) { Set removedPartitions = new HashSet<>(currentPartitions); removedPartitions.removeAll(partitionChangeEvent.getPartitions()); diff --git a/application/src/main/java/org/thingsboard/server/service/subscription/DefaultTbEntityDataSubscriptionService.java b/application/src/main/java/org/thingsboard/server/service/subscription/DefaultTbEntityDataSubscriptionService.java index fe2fbce573..689f80f757 100644 --- a/application/src/main/java/org/thingsboard/server/service/subscription/DefaultTbEntityDataSubscriptionService.java +++ b/application/src/main/java/org/thingsboard/server/service/subscription/DefaultTbEntityDataSubscriptionService.java @@ -476,7 +476,7 @@ public class DefaultTbEntityDataSubscriptionService implements TbEntityDataSubsc public void cancelAllSessionSubscriptions(String sessionId) { Map sessionSubs = subscriptionsBySessionId.remove(sessionId); if (sessionSubs != null) { - sessionSubs.values().stream().filter(sub -> sub instanceof TbEntityDataSubCtx).map(sub -> (TbEntityDataSubCtx) sub).forEach(this::cleanupAndCancel); + sessionSubs.values().forEach(this::cleanupAndCancel); } } diff --git a/application/src/main/java/org/thingsboard/server/service/subscription/DefaultTbLocalSubscriptionService.java b/application/src/main/java/org/thingsboard/server/service/subscription/DefaultTbLocalSubscriptionService.java index ee00b28562..0220a94964 100644 --- a/application/src/main/java/org/thingsboard/server/service/subscription/DefaultTbLocalSubscriptionService.java +++ b/application/src/main/java/org/thingsboard/server/service/subscription/DefaultTbLocalSubscriptionService.java @@ -28,6 +28,7 @@ import org.thingsboard.server.queue.discovery.PartitionService; import org.thingsboard.server.common.msg.queue.ServiceType; import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; import org.thingsboard.server.common.msg.queue.TbCallback; +import org.thingsboard.server.queue.discovery.TbApplicationEventListener; import org.thingsboard.server.queue.util.TbCoreComponent; import org.thingsboard.server.service.queue.TbClusterService; import org.thingsboard.server.service.telemetry.sub.AlarmSubscriptionUpdate; @@ -62,6 +63,34 @@ public class DefaultTbLocalSubscriptionService implements TbLocalSubscriptionSer private SubscriptionManagerService subscriptionManagerService; private ExecutorService subscriptionUpdateExecutor; + + private TbApplicationEventListener partitionChangeListener = new TbApplicationEventListener<>() { + @Override + protected void onTbApplicationEvent(PartitionChangeEvent event) { + if (ServiceType.TB_CORE.equals(event.getServiceType())) { + currentPartitions.clear(); + currentPartitions.addAll(event.getPartitions()); + } + } + }; + + private TbApplicationEventListener clusterTopologyChangeListener = new TbApplicationEventListener<>() { + @Override + protected void onTbApplicationEvent(ClusterTopologyChangeEvent event) { + if (event.getServiceQueueKeys().stream().anyMatch(key -> ServiceType.TB_CORE.equals(key.getServiceType()))) { + /* + * If the cluster topology has changed, we need to push all current subscriptions to SubscriptionManagerService again. + * Otherwise, the SubscriptionManagerService may "forget" those subscriptions in case of restart. + * Although this is resource consuming operation, it is cheaper than sending ping/pong commands periodically + * It is also cheaper then caching the subscriptions by entity id and then lookup of those caches every time we have new telemetry in SubscriptionManagerService. + * Even if we cache locally the list of active subscriptions by entity id, it is still time consuming operation to get them from cache + * Since number of subscriptions is usually much less then number of devices that are pushing data. + */ + subscriptionsBySessionId.values().forEach(map -> map.values() + .forEach(sub -> pushSubscriptionToManagerService(sub, true))); + } + } + }; @PostConstruct public void initExecutor() { @@ -77,28 +106,14 @@ public class DefaultTbLocalSubscriptionService implements TbLocalSubscriptionSer @Override @EventListener(PartitionChangeEvent.class) - public void onApplicationEvent(PartitionChangeEvent partitionChangeEvent) { - if (ServiceType.TB_CORE.equals(partitionChangeEvent.getServiceType())) { - currentPartitions.clear(); - currentPartitions.addAll(partitionChangeEvent.getPartitions()); - } + public void onApplicationEvent(PartitionChangeEvent event) { + partitionChangeListener.onApplicationEvent(event); } @Override @EventListener(ClusterTopologyChangeEvent.class) public void onApplicationEvent(ClusterTopologyChangeEvent event) { - if (event.getServiceQueueKeys().stream().anyMatch(key -> ServiceType.TB_CORE.equals(key.getServiceType()))) { - /* - * If the cluster topology has changed, we need to push all current subscriptions to SubscriptionManagerService again. - * Otherwise, the SubscriptionManagerService may "forget" those subscriptions in case of restart. - * Although this is resource consuming operation, it is cheaper than sending ping/pong commands periodically - * It is also cheaper then caching the subscriptions by entity id and then lookup of those caches every time we have new telemetry in SubscriptionManagerService. - * Even if we cache locally the list of active subscriptions by entity id, it is still time consuming operation to get them from cache - * Since number of subscriptions is usually much less then number of devices that are pushing data. - */ - subscriptionsBySessionId.values().forEach(map -> map.values() - .forEach(sub -> pushSubscriptionToManagerService(sub, true))); - } + clusterTopologyChangeListener.onApplicationEvent(event); } //TODO 3.1: replace null callbacks with callbacks from websocket service. diff --git a/application/src/main/java/org/thingsboard/server/service/telemetry/AbstractSubscriptionService.java b/application/src/main/java/org/thingsboard/server/service/telemetry/AbstractSubscriptionService.java index 8827a71f70..168e1271fd 100644 --- a/application/src/main/java/org/thingsboard/server/service/telemetry/AbstractSubscriptionService.java +++ b/application/src/main/java/org/thingsboard/server/service/telemetry/AbstractSubscriptionService.java @@ -41,6 +41,7 @@ import org.thingsboard.server.dao.timeseries.TimeseriesService; import org.thingsboard.server.gen.transport.TransportProtos; import org.thingsboard.server.queue.discovery.PartitionChangeEvent; import org.thingsboard.server.queue.discovery.PartitionService; +import org.thingsboard.server.queue.discovery.TbApplicationEventListener; import org.thingsboard.server.service.queue.TbClusterService; import org.thingsboard.server.service.subscription.SubscriptionManagerService; import org.thingsboard.server.service.subscription.TbSubscriptionUtils; @@ -61,7 +62,7 @@ import java.util.function.Consumer; * Created by ashvayka on 27.03.18. */ @Slf4j -public abstract class AbstractSubscriptionService implements ApplicationListener { +public abstract class AbstractSubscriptionService extends TbApplicationEventListener{ protected final Set currentPartitions = ConcurrentHashMap.newKeySet(); @@ -97,8 +98,7 @@ public abstract class AbstractSubscriptionService implements ApplicationListener } @Override - @EventListener(PartitionChangeEvent.class) - public void onApplicationEvent(PartitionChangeEvent partitionChangeEvent) { + protected void onTbApplicationEvent(PartitionChangeEvent partitionChangeEvent) { if (ServiceType.TB_CORE.equals(partitionChangeEvent.getServiceType())) { currentPartitions.clear(); currentPartitions.addAll(partitionChangeEvent.getPartitions()); diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml index 1453015101..93fff15df2 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/application/src/main/resources/thingsboard.yml @@ -118,6 +118,15 @@ security: githubMapper: emailUrl: "${SECURITY_OAUTH2_GITHUB_MAPPER_EMAIL_URL_KEY:https://api.github.com/user/emails}" +# Usage statistics parameters +usage: + stats: + report: + enabled: "${USAGE_STATS_REPORT_ENABLED:true}" + interval: "${USAGE_STATS_REPORT_INTERVAL:10}" + check: + cycle: "${USAGE_STATS_CHECK_CYCLE:60000}" + # Dashboard parameters dashboard: # Maximum allowed datapoints fetched by widgets @@ -675,6 +684,10 @@ queue: transport-api: "${TB_QUEUE_KAFKA_TA_TOPIC_PROPERTIES:retention.ms:604800000;segment.bytes:26214400;retention.bytes:1048576000;partitions:1}" notifications: "${TB_QUEUE_KAFKA_NOTIFICATIONS_TOPIC_PROPERTIES:retention.ms:604800000;segment.bytes:26214400;retention.bytes:1048576000;partitions:1}" js-executor: "${TB_QUEUE_KAFKA_JE_TOPIC_PROPERTIES:retention.ms:604800000;segment.bytes:26214400;retention.bytes:104857600;partitions:100}" + consumer-stats: + enabled: "${TB_QUEUE_KAFKA_CONSUMER_STATS_ENABLED:true}" + print-interval-ms: "${TB_QUEUE_KAFKA_CONSUMER_STATS_MIN_PRINT_INTERVAL_MS:60000}" + kafka-response-timeout-ms: "${TB_QUEUE_KAFKA_CONSUMER_STATS_RESPONSE_TIMEOUT_MS:1000}" aws_sqs: use_default_credential_provider_chain: "${TB_QUEUE_AWS_SQS_USE_DEFAULT_CREDENTIAL_PROVIDER_CHAIN:false}" access_key_id: "${TB_QUEUE_AWS_SQS_ACCESS_KEY_ID:YOUR_KEY}" diff --git a/common/actor/src/main/java/org/thingsboard/server/actors/TbActorMailbox.java b/common/actor/src/main/java/org/thingsboard/server/actors/TbActorMailbox.java index 689a657d6d..25dd8694f8 100644 --- a/common/actor/src/main/java/org/thingsboard/server/actors/TbActorMailbox.java +++ b/common/actor/src/main/java/org/thingsboard/server/actors/TbActorMailbox.java @@ -17,6 +17,7 @@ package org.thingsboard.server.actors; import lombok.Data; import lombok.extern.slf4j.Slf4j; +import org.thingsboard.server.common.msg.MsgType; import org.thingsboard.server.common.msg.TbActorMsg; import org.thingsboard.server.common.msg.TbActorStopReason; @@ -73,7 +74,7 @@ public final class TbActorMailbox implements TbActorCtx { if (strategy.isStop() || (settings.getMaxActorInitAttempts() > 0 && attemptIdx > settings.getMaxActorInitAttempts())) { log.info("[{}] Failed to init actor, attempt {}, going to stop attempts.", selfId, attempt, t); stopReason = TbActorStopReason.INIT_FAILED; - system.stop(selfId); + destroy(); } else if (strategy.getRetryDelay() > 0) { log.info("[{}] Failed to init actor, attempt {}, going to retry in attempts in {}ms", selfId, attempt, strategy.getRetryDelay()); log.debug("[{}] Error", selfId, t); @@ -95,7 +96,19 @@ public final class TbActorMailbox implements TbActorCtx { } tryProcessQueue(true); } else { - msg.onTbActorStopped(stopReason); + if (highPriority && msg.getMsgType().equals(MsgType.RULE_NODE_UPDATED_MSG)) { + synchronized (this) { + if (stopReason == TbActorStopReason.INIT_FAILED) { + destroyInProgress.set(false); + stopReason = null; + initActor(); + } else { + msg.onTbActorStopped(stopReason); + } + } + } else { + msg.onTbActorStopped(stopReason); + } } } @@ -126,6 +139,9 @@ public final class TbActorMailbox implements TbActorCtx { try { log.debug("[{}] Going to process message: {}", selfId, msg); actor.process(msg); + } catch (TbRuleNodeUpdateException updateException){ + stopReason = TbActorStopReason.INIT_FAILED; + destroy(); } catch (Throwable t) { log.debug("[{}] Failed to process message: {}", selfId, msg, t); ProcessFailureStrategy strategy = actor.onProcessFailure(t); diff --git a/common/actor/src/main/java/org/thingsboard/server/actors/TbRuleNodeUpdateException.java b/common/actor/src/main/java/org/thingsboard/server/actors/TbRuleNodeUpdateException.java new file mode 100644 index 0000000000..7e3cca863d --- /dev/null +++ b/common/actor/src/main/java/org/thingsboard/server/actors/TbRuleNodeUpdateException.java @@ -0,0 +1,26 @@ +/** + * Copyright © 2016-2021 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.actors; + +public class TbRuleNodeUpdateException extends RuntimeException { + + private static final long serialVersionUID = 8209771144711980882L; + + public TbRuleNodeUpdateException(String message, Throwable cause) { + super(message, cause); + } +} + diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/DataConstants.java b/common/data/src/main/java/org/thingsboard/server/common/data/DataConstants.java index 12cc17270c..cc78e2cb45 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/DataConstants.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/DataConstants.java @@ -24,6 +24,7 @@ public class DataConstants { public static final String CUSTOMER = "CUSTOMER"; public static final String DEVICE = "DEVICE"; + public static final String SCOPE = "scope"; public static final String CLIENT_SCOPE = "CLIENT_SCOPE"; public static final String SERVER_SCOPE = "SERVER_SCOPE"; public static final String SHARED_SCOPE = "SHARED_SCOPE"; diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/query/DynamicValue.java b/common/data/src/main/java/org/thingsboard/server/common/data/query/DynamicValue.java index 3d46f8d2ab..2fa0a4d36a 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/query/DynamicValue.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/query/DynamicValue.java @@ -17,17 +17,23 @@ package org.thingsboard.server.common.data.query; import com.fasterxml.jackson.annotation.JsonIgnore; import lombok.Data; -import lombok.Getter; +import lombok.RequiredArgsConstructor; @Data +@RequiredArgsConstructor public class DynamicValue { @JsonIgnore private T resolvedValue; - @Getter private final DynamicValueSourceType sourceType; - @Getter private final String sourceAttribute; + private final boolean inherit; + + public DynamicValue(DynamicValueSourceType sourceType, String sourceAttribute) { + this.sourceAttribute = sourceAttribute; + this.sourceType = sourceType; + this.inherit = false; + } } diff --git a/common/message/src/main/java/org/thingsboard/server/common/msg/MsgType.java b/common/message/src/main/java/org/thingsboard/server/common/msg/MsgType.java index ffc574d5c2..13d0a6e9eb 100644 --- a/common/message/src/main/java/org/thingsboard/server/common/msg/MsgType.java +++ b/common/message/src/main/java/org/thingsboard/server/common/msg/MsgType.java @@ -39,6 +39,11 @@ public enum MsgType { */ COMPONENT_LIFE_CYCLE_MSG, + /** + * Special message to indicate rule node update request + */ + RULE_NODE_UPDATED_MSG, + /** * Misc messages consumed from the Queue and forwarded to Rule Engine Actor. * diff --git a/common/message/src/main/java/org/thingsboard/server/common/msg/plugin/RuleNodeUpdatedMsg.java b/common/message/src/main/java/org/thingsboard/server/common/msg/plugin/RuleNodeUpdatedMsg.java new file mode 100644 index 0000000000..ca4ab1c949 --- /dev/null +++ b/common/message/src/main/java/org/thingsboard/server/common/msg/plugin/RuleNodeUpdatedMsg.java @@ -0,0 +1,40 @@ +/** + * Copyright © 2016-2021 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.msg.plugin; + +import lombok.ToString; +import org.thingsboard.server.common.data.id.EntityId; +import org.thingsboard.server.common.data.id.TenantId; +import org.thingsboard.server.common.data.plugin.ComponentLifecycleEvent; +import org.thingsboard.server.common.msg.MsgType; + +import java.util.Optional; + +/** + * @author Andrew Shvayka + */ +@ToString +public class RuleNodeUpdatedMsg extends ComponentLifecycleMsg { + + public RuleNodeUpdatedMsg(TenantId tenantId, EntityId entityId) { + super(tenantId, entityId, ComponentLifecycleEvent.UPDATED); + } + + @Override + public MsgType getMsgType() { + return MsgType.RULE_NODE_UPDATED_MSG; + } +} \ No newline at end of file diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/discovery/ClusterTopologyChangeEvent.java b/common/queue/src/main/java/org/thingsboard/server/queue/discovery/ClusterTopologyChangeEvent.java index 0602dee3ae..1e5b90b5fe 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/discovery/ClusterTopologyChangeEvent.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/discovery/ClusterTopologyChangeEvent.java @@ -22,7 +22,9 @@ import org.thingsboard.server.common.msg.queue.ServiceQueueKey; import java.util.Set; -public class ClusterTopologyChangeEvent extends ApplicationEvent { +public class ClusterTopologyChangeEvent extends TbApplicationEvent { + + private static final long serialVersionUID = -2441739930040282254L; @Getter private final Set serviceQueueKeys; diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/discovery/HashPartitionService.java b/common/queue/src/main/java/org/thingsboard/server/queue/discovery/HashPartitionService.java index d8164f4be8..2da438417a 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/discovery/HashPartitionService.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/discovery/HashPartitionService.java @@ -126,7 +126,7 @@ public class HashPartitionService implements PartitionService { } @Override - public void recalculatePartitions(ServiceInfo currentService, List otherServices) { + public synchronized void recalculatePartitions(ServiceInfo currentService, List otherServices) { logServiceInfo(currentService); otherServices.forEach(this::logServiceInfo); Map> queueServicesMap = new HashMap<>(); @@ -134,7 +134,7 @@ public class HashPartitionService implements PartitionService { for (ServiceInfo other : otherServices) { addNode(queueServicesMap, other); } - queueServicesMap.values().forEach(list -> list.sort((a, b) -> a.getServiceId().compareTo(b.getServiceId()))); + queueServicesMap.values().forEach(list -> list.sort(Comparator.comparing(ServiceInfo::getServiceId))); ConcurrentMap> oldPartitions = myPartitions; TenantId myIsolatedOrSystemTenantId = getSystemOrIsolatedTenantId(currentService); diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/discovery/PartitionChangeEvent.java b/common/queue/src/main/java/org/thingsboard/server/queue/discovery/PartitionChangeEvent.java index e4edabbe19..2edcd2ceca 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/discovery/PartitionChangeEvent.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/discovery/PartitionChangeEvent.java @@ -24,7 +24,9 @@ import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; import java.util.Set; -public class PartitionChangeEvent extends ApplicationEvent { +public class PartitionChangeEvent extends TbApplicationEvent { + + private static final long serialVersionUID = -8731788167026510559L; @Getter private final ServiceQueueKey serviceQueueKey; diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/discovery/TbApplicationEvent.java b/common/queue/src/main/java/org/thingsboard/server/queue/discovery/TbApplicationEvent.java new file mode 100644 index 0000000000..face2d36d6 --- /dev/null +++ b/common/queue/src/main/java/org/thingsboard/server/queue/discovery/TbApplicationEvent.java @@ -0,0 +1,37 @@ +/** + * Copyright © 2016-2021 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.queue.discovery; + +import lombok.Getter; +import org.springframework.context.ApplicationEvent; + +import java.util.concurrent.atomic.AtomicInteger; + +public class TbApplicationEvent extends ApplicationEvent { + + private static final long serialVersionUID = 3884264064887765146L; + + private static final AtomicInteger sequence = new AtomicInteger(); + + @Getter + private final int sequenceNumber; + + public TbApplicationEvent(Object source) { + super(source); + sequenceNumber = sequence.incrementAndGet(); + } + +} diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/discovery/TbApplicationEventListener.java b/common/queue/src/main/java/org/thingsboard/server/queue/discovery/TbApplicationEventListener.java new file mode 100644 index 0000000000..9158d8f0c8 --- /dev/null +++ b/common/queue/src/main/java/org/thingsboard/server/queue/discovery/TbApplicationEventListener.java @@ -0,0 +1,52 @@ +/** + * Copyright © 2016-2021 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.queue.discovery; + +import lombok.extern.slf4j.Slf4j; +import org.springframework.context.ApplicationListener; + +import java.util.concurrent.locks.Lock; +import java.util.concurrent.locks.ReentrantLock; + +@Slf4j +public abstract class TbApplicationEventListener implements ApplicationListener { + + private int lastProcessedSequenceNumber = Integer.MIN_VALUE; + private final Lock seqNumberLock = new ReentrantLock(); + + @Override + public void onApplicationEvent(T event) { + boolean validUpdate = false; + seqNumberLock.lock(); + try { + if (event.getSequenceNumber() > lastProcessedSequenceNumber) { + validUpdate = true; + lastProcessedSequenceNumber = event.getSequenceNumber(); + } + } finally { + seqNumberLock.unlock(); + } + if (validUpdate) { + onTbApplicationEvent(event); + } else { + log.info("Application event ignored due to invalid sequence number ({} > {}). Event: {}", lastProcessedSequenceNumber, event.getSequenceNumber(), event); + } + } + + protected abstract void onTbApplicationEvent(T event); + + +} diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaConsumerStatisticConfig.java b/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaConsumerStatisticConfig.java new file mode 100644 index 0000000000..bfe7d29bb6 --- /dev/null +++ b/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaConsumerStatisticConfig.java @@ -0,0 +1,37 @@ +/** + * Copyright © 2016-2021 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.queue.kafka; + +import lombok.AllArgsConstructor; +import lombok.Getter; +import lombok.NoArgsConstructor; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.stereotype.Component; + +@Component +@ConditionalOnProperty(prefix = "queue", value = "type", havingValue = "kafka") +@Getter +@AllArgsConstructor +@NoArgsConstructor +public class TbKafkaConsumerStatisticConfig { + @Value("${queue.kafka.consumer-stats.enabled:true}") + private Boolean enabled; + @Value("${queue.kafka.consumer-stats.print-interval-ms:60000}") + private Long printIntervalMs; + @Value("${queue.kafka.consumer-stats.kafka-response-timeout-ms:1000}") + private Long kafkaResponseTimeoutMs; +} diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaConsumerStatsService.java b/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaConsumerStatsService.java new file mode 100644 index 0000000000..0ccf9ba42c --- /dev/null +++ b/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaConsumerStatsService.java @@ -0,0 +1,184 @@ +/** + * Copyright © 2016-2021 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.queue.kafka; + +import lombok.Builder; +import lombok.Data; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.apache.kafka.clients.admin.AdminClient; +import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.clients.consumer.KafkaConsumer; +import org.apache.kafka.clients.consumer.OffsetAndMetadata; +import org.apache.kafka.common.TopicPartition; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.stereotype.Component; +import org.springframework.util.StringUtils; +import org.thingsboard.common.util.ThingsBoardThreadFactory; +import org.thingsboard.server.common.data.id.TenantId; +import org.thingsboard.server.common.msg.queue.ServiceType; +import org.thingsboard.server.queue.discovery.PartitionService; + +import javax.annotation.PostConstruct; +import javax.annotation.PreDestroy; +import java.time.Duration; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.Properties; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; + +@Slf4j +@Component +@RequiredArgsConstructor +@ConditionalOnProperty(prefix = "queue", value = "type", havingValue = "kafka") +public class TbKafkaConsumerStatsService { + private final Set monitoredGroups = ConcurrentHashMap.newKeySet(); + + private final TbKafkaSettings kafkaSettings; + private final TbKafkaConsumerStatisticConfig statsConfig; + private final PartitionService partitionService; + + private AdminClient adminClient; + private Consumer consumer; + private ScheduledExecutorService statsPrintScheduler; + + @PostConstruct + public void init() { + if (!statsConfig.getEnabled()) { + return; + } + this.adminClient = AdminClient.create(kafkaSettings.toAdminProps()); + this.statsPrintScheduler = Executors.newSingleThreadScheduledExecutor(ThingsBoardThreadFactory.forName("kafka-consumer-stats")); + + Properties consumerProps = kafkaSettings.toConsumerProps(); + consumerProps.put(ConsumerConfig.CLIENT_ID_CONFIG, "consumer-stats-loader-client"); + consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, "consumer-stats-loader-client-group"); + this.consumer = new KafkaConsumer<>(consumerProps); + + startLogScheduling(); + } + + private void startLogScheduling() { + Duration timeoutDuration = Duration.ofMillis(statsConfig.getKafkaResponseTimeoutMs()); + statsPrintScheduler.scheduleWithFixedDelay(() -> { + if (!isStatsPrintRequired()) { + return; + } + for (String groupId : monitoredGroups) { + try { + Map groupOffsets = adminClient.listConsumerGroupOffsets(groupId).partitionsToOffsetAndMetadata() + .get(statsConfig.getKafkaResponseTimeoutMs(), TimeUnit.MILLISECONDS); + Map endOffsets = consumer.endOffsets(groupOffsets.keySet(), timeoutDuration); + + List lagTopicsStats = getTopicsStatsWithLag(groupOffsets, endOffsets); + if (!lagTopicsStats.isEmpty()) { + StringBuilder builder = new StringBuilder(); + for (int i = 0; i < lagTopicsStats.size(); i++) { + builder.append(lagTopicsStats.get(i).toString()); + if (i != lagTopicsStats.size() - 1) { + builder.append(", "); + } + } + log.info("[{}] Topic partitions with lag: [{}].", groupId, builder.toString()); + } + } catch (Exception e) { + log.warn("[{}] Failed to get consumer group stats. Reason - {}.", groupId, e.getMessage()); + log.trace("Detailed error: ", e); + } + } + + }, statsConfig.getPrintIntervalMs(), statsConfig.getPrintIntervalMs(), TimeUnit.MILLISECONDS); + } + + private boolean isStatsPrintRequired() { + boolean isMyRuleEnginePartition = partitionService.resolve(ServiceType.TB_RULE_ENGINE, TenantId.SYS_TENANT_ID, TenantId.SYS_TENANT_ID).isMyPartition(); + boolean isMyCorePartition = partitionService.resolve(ServiceType.TB_CORE, TenantId.SYS_TENANT_ID, TenantId.SYS_TENANT_ID).isMyPartition(); + return log.isInfoEnabled() && (isMyRuleEnginePartition || isMyCorePartition); + } + + private List getTopicsStatsWithLag(Map groupOffsets, Map endOffsets) { + List consumerGroupStats = new ArrayList<>(); + for (TopicPartition topicPartition : groupOffsets.keySet()) { + long endOffset = endOffsets.get(topicPartition); + long committedOffset = groupOffsets.get(topicPartition).offset(); + long lag = endOffset - committedOffset; + if (lag != 0) { + GroupTopicStats groupTopicStats = GroupTopicStats.builder() + .topic(topicPartition.topic()) + .partition(topicPartition.partition()) + .committedOffset(committedOffset) + .endOffset(endOffset) + .lag(lag) + .build(); + consumerGroupStats.add(groupTopicStats); + } + } + return consumerGroupStats; + } + + public void registerClientGroup(String groupId) { + if (statsConfig.getEnabled() && !StringUtils.isEmpty(groupId)) { + monitoredGroups.add(groupId); + } + } + + public void unregisterClientGroup(String groupId) { + if (statsConfig.getEnabled() && !StringUtils.isEmpty(groupId)) { + monitoredGroups.remove(groupId); + } + } + + @PreDestroy + public void destroy() { + if (statsPrintScheduler != null) { + statsPrintScheduler.shutdownNow(); + } + if (adminClient != null) { + adminClient.close(); + } + if (consumer != null) { + consumer.close(); + } + } + + + @Builder + @Data + private static class GroupTopicStats { + private String topic; + private int partition; + private long committedOffset; + private long endOffset; + private long lag; + + @Override + public String toString() { + return "[" + + "topic=[" + topic + ']' + + ", partition=[" + partition + "]" + + ", committedOffset=[" + committedOffset + "]" + + ", endOffset=[" + endOffset + "]" + + ", lag=[" + lag + "]" + + "]"; + } + } +} diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaConsumerTemplate.java b/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaConsumerTemplate.java index 8743f4d6cc..b1b3d5ce05 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaConsumerTemplate.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaConsumerTemplate.java @@ -42,10 +42,13 @@ public class TbKafkaConsumerTemplate extends AbstractTbQue private final KafkaConsumer consumer; private final TbKafkaDecoder decoder; + private final TbKafkaConsumerStatsService statsService; + private final String groupId; + @Builder private TbKafkaConsumerTemplate(TbKafkaSettings settings, TbKafkaDecoder decoder, String clientId, String groupId, String topic, - TbQueueAdmin admin) { + TbQueueAdmin admin, TbKafkaConsumerStatsService statsService) { super(topic); Properties props = settings.toConsumerProps(); props.put(ConsumerConfig.CLIENT_ID_CONFIG, clientId); @@ -53,6 +56,13 @@ public class TbKafkaConsumerTemplate extends AbstractTbQue props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId); } + this.statsService = statsService; + this.groupId = groupId; + + if (statsService != null) { + statsService.registerClientGroup(groupId); + } + this.admin = admin; this.consumer = new KafkaConsumer<>(props); this.decoder = decoder; @@ -96,6 +106,8 @@ public class TbKafkaConsumerTemplate extends AbstractTbQue consumer.unsubscribe(); consumer.close(); } + if (statsService != null) { + statsService.unregisterClientGroup(groupId); + } } - } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaMonolithQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaMonolithQueueFactory.java index eebffc37b0..e07f3e43d3 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaMonolithQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaMonolithQueueFactory.java @@ -39,6 +39,7 @@ import org.thingsboard.server.queue.common.TbProtoQueueMsg; import org.thingsboard.server.queue.discovery.PartitionService; import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; import org.thingsboard.server.queue.kafka.TbKafkaAdmin; +import org.thingsboard.server.queue.kafka.TbKafkaConsumerStatsService; import org.thingsboard.server.queue.kafka.TbKafkaConsumerTemplate; import org.thingsboard.server.queue.kafka.TbKafkaProducerTemplate; import org.thingsboard.server.queue.kafka.TbKafkaSettings; @@ -65,6 +66,7 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi private final TbQueueTransportApiSettings transportApiSettings; private final TbQueueTransportNotificationSettings transportNotificationSettings; private final TbQueueRemoteJsInvokeSettings jsInvokeSettings; + private final TbKafkaConsumerStatsService consumerStatsService; private final TbQueueAdmin coreAdmin; private final TbQueueAdmin ruleEngineAdmin; @@ -79,6 +81,7 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi TbQueueTransportApiSettings transportApiSettings, TbQueueTransportNotificationSettings transportNotificationSettings, TbQueueRemoteJsInvokeSettings jsInvokeSettings, + TbKafkaConsumerStatsService consumerStatsService, TbKafkaTopicConfigs kafkaTopicConfigs) { this.partitionService = partitionService; this.kafkaSettings = kafkaSettings; @@ -88,6 +91,7 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi this.transportApiSettings = transportApiSettings; this.transportNotificationSettings = transportNotificationSettings; this.jsInvokeSettings = jsInvokeSettings; + this.consumerStatsService = consumerStatsService; this.coreAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getCoreConfigs()); this.ruleEngineAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getRuleEngineConfigs()); @@ -156,6 +160,7 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi consumerBuilder.groupId("re-" + queueName + "-consumer"); consumerBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), ToRuleEngineMsg.parseFrom(msg.getData()), msg.getHeaders())); consumerBuilder.admin(ruleEngineAdmin); + consumerBuilder.statsService(consumerStatsService); return consumerBuilder.build(); } @@ -168,6 +173,7 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi consumerBuilder.groupId("monolith-rule-engine-notifications-consumer-" + serviceInfoProvider.getServiceId()); consumerBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), ToRuleEngineNotificationMsg.parseFrom(msg.getData()), msg.getHeaders())); consumerBuilder.admin(notificationAdmin); + consumerBuilder.statsService(consumerStatsService); return consumerBuilder.build(); } @@ -180,6 +186,7 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi consumerBuilder.groupId("monolith-core-consumer"); consumerBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), ToCoreMsg.parseFrom(msg.getData()), msg.getHeaders())); consumerBuilder.admin(coreAdmin); + consumerBuilder.statsService(consumerStatsService); return consumerBuilder.build(); } @@ -192,6 +199,7 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi consumerBuilder.groupId("monolith-core-notifications-consumer-" + serviceInfoProvider.getServiceId()); consumerBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), ToCoreNotificationMsg.parseFrom(msg.getData()), msg.getHeaders())); consumerBuilder.admin(notificationAdmin); + consumerBuilder.statsService(consumerStatsService); return consumerBuilder.build(); } @@ -204,6 +212,7 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi consumerBuilder.groupId("monolith-transport-api-consumer"); consumerBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportApiRequestMsg.parseFrom(msg.getData()), msg.getHeaders())); consumerBuilder.admin(transportApiAdmin); + consumerBuilder.statsService(consumerStatsService); return consumerBuilder.build(); } @@ -237,6 +246,7 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi return new TbProtoQueueMsg<>(msg.getKey(), builder.build(), msg.getHeaders()); } ); + responseBuilder.statsService(consumerStatsService); responseBuilder.admin(jsExecutorAdmin); DefaultTbQueueRequestTemplate.DefaultTbQueueRequestTemplateBuilder @@ -259,6 +269,7 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi consumerBuilder.groupId("monolith-us-consumer"); consumerBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), ToUsageStatsServiceMsg.parseFrom(msg.getData()), msg.getHeaders())); consumerBuilder.admin(coreAdmin); + consumerBuilder.statsService(consumerStatsService); return consumerBuilder.build(); } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbCoreQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbCoreQueueFactory.java index 15771a3b78..24e80adec5 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbCoreQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbCoreQueueFactory.java @@ -39,6 +39,7 @@ import org.thingsboard.server.queue.common.TbProtoQueueMsg; import org.thingsboard.server.queue.discovery.PartitionService; import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; import org.thingsboard.server.queue.kafka.TbKafkaAdmin; +import org.thingsboard.server.queue.kafka.TbKafkaConsumerStatsService; import org.thingsboard.server.queue.kafka.TbKafkaConsumerTemplate; import org.thingsboard.server.queue.kafka.TbKafkaProducerTemplate; import org.thingsboard.server.queue.kafka.TbKafkaSettings; @@ -62,6 +63,7 @@ public class KafkaTbCoreQueueFactory implements TbCoreQueueFactory { private final TbQueueRuleEngineSettings ruleEngineSettings; private final TbQueueTransportApiSettings transportApiSettings; private final TbQueueRemoteJsInvokeSettings jsInvokeSettings; + private final TbKafkaConsumerStatsService consumerStatsService; private final TbQueueAdmin coreAdmin; private final TbQueueAdmin ruleEngineAdmin; @@ -75,6 +77,7 @@ public class KafkaTbCoreQueueFactory implements TbCoreQueueFactory { TbQueueRuleEngineSettings ruleEngineSettings, TbQueueTransportApiSettings transportApiSettings, TbQueueRemoteJsInvokeSettings jsInvokeSettings, + TbKafkaConsumerStatsService consumerStatsService, TbKafkaTopicConfigs kafkaTopicConfigs) { this.partitionService = partitionService; this.kafkaSettings = kafkaSettings; @@ -83,6 +86,7 @@ public class KafkaTbCoreQueueFactory implements TbCoreQueueFactory { this.ruleEngineSettings = ruleEngineSettings; this.transportApiSettings = transportApiSettings; this.jsInvokeSettings = jsInvokeSettings; + this.consumerStatsService = consumerStatsService; this.coreAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getCoreConfigs()); this.ruleEngineAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getRuleEngineConfigs()); @@ -150,6 +154,7 @@ public class KafkaTbCoreQueueFactory implements TbCoreQueueFactory { consumerBuilder.groupId("tb-core-node"); consumerBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), ToCoreMsg.parseFrom(msg.getData()), msg.getHeaders())); consumerBuilder.admin(coreAdmin); + consumerBuilder.statsService(consumerStatsService); return consumerBuilder.build(); } @@ -162,6 +167,7 @@ public class KafkaTbCoreQueueFactory implements TbCoreQueueFactory { consumerBuilder.groupId("tb-core-notifications-node-" + serviceInfoProvider.getServiceId()); consumerBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), ToCoreNotificationMsg.parseFrom(msg.getData()), msg.getHeaders())); consumerBuilder.admin(notificationAdmin); + consumerBuilder.statsService(consumerStatsService); return consumerBuilder.build(); } @@ -174,6 +180,7 @@ public class KafkaTbCoreQueueFactory implements TbCoreQueueFactory { consumerBuilder.groupId("tb-core-transport-api-consumer"); consumerBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportApiRequestMsg.parseFrom(msg.getData()), msg.getHeaders())); consumerBuilder.admin(transportApiAdmin); + consumerBuilder.statsService(consumerStatsService); return consumerBuilder.build(); } @@ -208,6 +215,7 @@ public class KafkaTbCoreQueueFactory implements TbCoreQueueFactory { } ); responseBuilder.admin(jsExecutorAdmin); + responseBuilder.statsService(consumerStatsService); DefaultTbQueueRequestTemplate.DefaultTbQueueRequestTemplateBuilder , TbProtoQueueMsg> builder = DefaultTbQueueRequestTemplate.builder(); @@ -229,6 +237,7 @@ public class KafkaTbCoreQueueFactory implements TbCoreQueueFactory { consumerBuilder.groupId("tb-core-us-consumer"); consumerBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), ToUsageStatsServiceMsg.parseFrom(msg.getData()), msg.getHeaders())); consumerBuilder.admin(coreAdmin); + consumerBuilder.statsService(consumerStatsService); return consumerBuilder.build(); } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbRuleEngineQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbRuleEngineQueueFactory.java index b45ec62b04..a8247dc11e 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbRuleEngineQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbRuleEngineQueueFactory.java @@ -37,6 +37,7 @@ import org.thingsboard.server.queue.common.TbProtoQueueMsg; import org.thingsboard.server.queue.discovery.PartitionService; import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; import org.thingsboard.server.queue.kafka.TbKafkaAdmin; +import org.thingsboard.server.queue.kafka.TbKafkaConsumerStatsService; import org.thingsboard.server.queue.kafka.TbKafkaConsumerTemplate; import org.thingsboard.server.queue.kafka.TbKafkaProducerTemplate; import org.thingsboard.server.queue.kafka.TbKafkaSettings; @@ -59,6 +60,7 @@ public class KafkaTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory { private final TbQueueCoreSettings coreSettings; private final TbQueueRuleEngineSettings ruleEngineSettings; private final TbQueueRemoteJsInvokeSettings jsInvokeSettings; + private final TbKafkaConsumerStatsService consumerStatsService; private final TbQueueAdmin coreAdmin; private final TbQueueAdmin ruleEngineAdmin; @@ -70,6 +72,7 @@ public class KafkaTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory { TbQueueCoreSettings coreSettings, TbQueueRuleEngineSettings ruleEngineSettings, TbQueueRemoteJsInvokeSettings jsInvokeSettings, + TbKafkaConsumerStatsService consumerStatsService, TbKafkaTopicConfigs kafkaTopicConfigs) { this.partitionService = partitionService; this.kafkaSettings = kafkaSettings; @@ -77,6 +80,7 @@ public class KafkaTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory { this.coreSettings = coreSettings; this.ruleEngineSettings = ruleEngineSettings; this.jsInvokeSettings = jsInvokeSettings; + this.consumerStatsService = consumerStatsService; this.coreAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getCoreConfigs()); this.ruleEngineAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getRuleEngineConfigs()); @@ -145,6 +149,7 @@ public class KafkaTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory { consumerBuilder.groupId("re-" + queueName + "-consumer"); consumerBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), ToRuleEngineMsg.parseFrom(msg.getData()), msg.getHeaders())); consumerBuilder.admin(ruleEngineAdmin); + consumerBuilder.statsService(consumerStatsService); return consumerBuilder.build(); } @@ -157,6 +162,7 @@ public class KafkaTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory { consumerBuilder.groupId("tb-rule-engine-notifications-node-" + serviceInfoProvider.getServiceId()); consumerBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), ToRuleEngineNotificationMsg.parseFrom(msg.getData()), msg.getHeaders())); consumerBuilder.admin(notificationAdmin); + consumerBuilder.statsService(consumerStatsService); return consumerBuilder.build(); } @@ -181,6 +187,7 @@ public class KafkaTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory { } ); responseBuilder.admin(jsExecutorAdmin); + responseBuilder.statsService(consumerStatsService); DefaultTbQueueRequestTemplate.DefaultTbQueueRequestTemplateBuilder , TbProtoQueueMsg> builder = DefaultTbQueueRequestTemplate.builder(); diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbTransportQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbTransportQueueFactory.java index 0c64a57304..8c0c6999d5 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbTransportQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbTransportQueueFactory.java @@ -32,6 +32,7 @@ import org.thingsboard.server.queue.common.DefaultTbQueueRequestTemplate; import org.thingsboard.server.queue.common.TbProtoQueueMsg; import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; import org.thingsboard.server.queue.kafka.TbKafkaAdmin; +import org.thingsboard.server.queue.kafka.TbKafkaConsumerStatsService; import org.thingsboard.server.queue.kafka.TbKafkaConsumerTemplate; import org.thingsboard.server.queue.kafka.TbKafkaProducerTemplate; import org.thingsboard.server.queue.kafka.TbKafkaSettings; @@ -54,6 +55,7 @@ public class KafkaTbTransportQueueFactory implements TbTransportQueueFactory { private final TbQueueRuleEngineSettings ruleEngineSettings; private final TbQueueTransportApiSettings transportApiSettings; private final TbQueueTransportNotificationSettings transportNotificationSettings; + private final TbKafkaConsumerStatsService consumerStatsService; private final TbQueueAdmin coreAdmin; private final TbQueueAdmin ruleEngineAdmin; @@ -66,6 +68,7 @@ public class KafkaTbTransportQueueFactory implements TbTransportQueueFactory { TbQueueRuleEngineSettings ruleEngineSettings, TbQueueTransportApiSettings transportApiSettings, TbQueueTransportNotificationSettings transportNotificationSettings, + TbKafkaConsumerStatsService consumerStatsService, TbKafkaTopicConfigs kafkaTopicConfigs) { this.kafkaSettings = kafkaSettings; this.serviceInfoProvider = serviceInfoProvider; @@ -73,6 +76,7 @@ public class KafkaTbTransportQueueFactory implements TbTransportQueueFactory { this.ruleEngineSettings = ruleEngineSettings; this.transportApiSettings = transportApiSettings; this.transportNotificationSettings = transportNotificationSettings; + this.consumerStatsService = consumerStatsService; this.coreAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getCoreConfigs()); this.ruleEngineAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getRuleEngineConfigs()); @@ -95,6 +99,7 @@ public class KafkaTbTransportQueueFactory implements TbTransportQueueFactory { responseBuilder.groupId("transport-node-" + serviceInfoProvider.getServiceId()); responseBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportApiResponseMsg.parseFrom(msg.getData()), msg.getHeaders())); responseBuilder.admin(transportApiAdmin); + responseBuilder.statsService(consumerStatsService); DefaultTbQueueRequestTemplate.DefaultTbQueueRequestTemplateBuilder , TbProtoQueueMsg> templateBuilder = DefaultTbQueueRequestTemplate.builder(); @@ -136,6 +141,7 @@ public class KafkaTbTransportQueueFactory implements TbTransportQueueFactory { responseBuilder.groupId("transport-node-" + serviceInfoProvider.getServiceId()); responseBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), ToTransportMsg.parseFrom(msg.getData()), msg.getHeaders())); responseBuilder.admin(notificationAdmin); + responseBuilder.statsService(consumerStatsService); return responseBuilder.build(); } diff --git a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/adaptor/JsonConverter.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/adaptor/JsonConverter.java index 1a162d3c3b..2c5dfafe4e 100644 --- a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/adaptor/JsonConverter.java +++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/adaptor/JsonConverter.java @@ -51,6 +51,7 @@ import org.thingsboard.server.gen.transport.TransportProtos.ValidateBasicMqttCre import org.thingsboard.server.gen.transport.TransportProtos.ValidateDeviceTokenRequestMsg; import org.thingsboard.server.gen.transport.TransportProtos.ValidateDeviceX509CertRequestMsg; +import java.math.BigDecimal; import java.util.ArrayList; import java.util.HashMap; import java.util.HashSet; @@ -244,12 +245,25 @@ public class JsonConverter { } private static void parseNumericValue(List result, Entry valueEntry, JsonPrimitive value) { - if (value.getAsString().contains(".")) { - result.add(new DoubleDataEntry(valueEntry.getKey(), value.getAsDouble())); + String valueAsString = value.getAsString(); + String key = valueEntry.getKey(); + if (valueAsString.contains("e") || valueAsString.contains("E")) { + var bd = new BigDecimal(valueAsString); + if (bd.stripTrailingZeros().scale() <= 0) { + try { + result.add(new LongDataEntry(key, bd.longValueExact())); + } catch (ArithmeticException e) { + result.add(new DoubleDataEntry(key, bd.doubleValue())); + } + } else { + result.add(new DoubleDataEntry(key, bd.doubleValue())); + } + } else if (valueAsString.contains(".")) { + result.add(new DoubleDataEntry(key, value.getAsDouble())); } else { try { long longValue = Long.parseLong(value.getAsString()); - result.add(new LongDataEntry(valueEntry.getKey(), longValue)); + result.add(new LongDataEntry(key, longValue)); } catch (NumberFormatException e) { throw new JsonSyntaxException("Big integer values are not supported!"); } @@ -284,7 +298,8 @@ public class JsonConverter { return result; } - public static JsonObject getJsonObjectForGateway(String deviceName, TransportProtos.GetAttributeResponseMsg responseMsg) { + public static JsonObject getJsonObjectForGateway(String deviceName, TransportProtos.GetAttributeResponseMsg + responseMsg) { JsonObject result = new JsonObject(); result.addProperty("id", responseMsg.getRequestId()); result.addProperty(DEVICE_PROPERTY, deviceName); @@ -297,7 +312,8 @@ public class JsonConverter { return result; } - public static JsonObject getJsonObjectForGateway(String deviceName, AttributeUpdateNotificationMsg notificationMsg) { + public static JsonObject getJsonObjectForGateway(String deviceName, AttributeUpdateNotificationMsg + notificationMsg) { JsonObject result = new JsonObject(); result.addProperty(DEVICE_PROPERTY, deviceName); result.add("data", toJson(notificationMsg)); @@ -447,7 +463,8 @@ public class JsonConverter { return result; } - public static JsonElement toGatewayJson(String deviceName, TransportProtos.ProvisionDeviceResponseMsg responseRequest) { + public static JsonElement toGatewayJson(String deviceName, TransportProtos.ProvisionDeviceResponseMsg + responseRequest) { JsonObject result = new JsonObject(); result.addProperty(DEVICE_PROPERTY, deviceName); result.add("data", JsonConverter.toJson(responseRequest)); @@ -497,15 +514,18 @@ public class JsonConverter { return result; } - public static Map> convertToTelemetry(JsonElement jsonElement, long systemTs) throws JsonSyntaxException { + public static Map> convertToTelemetry(JsonElement jsonElement, long systemTs) throws + JsonSyntaxException { return convertToTelemetry(jsonElement, systemTs, false); } - public static Map> convertToSortedTelemetry(JsonElement jsonElement, long systemTs) throws JsonSyntaxException { + public static Map> convertToSortedTelemetry(JsonElement jsonElement, long systemTs) throws + JsonSyntaxException { return convertToTelemetry(jsonElement, systemTs, true); } - public static Map> convertToTelemetry(JsonElement jsonElement, long systemTs, boolean sorted) throws JsonSyntaxException { + public static Map> convertToTelemetry(JsonElement jsonElement, long systemTs, boolean sorted) throws + JsonSyntaxException { Map> result = sorted ? new TreeMap<>() : new HashMap<>(); convertToTelemetry(jsonElement, systemTs, result, null); return result; @@ -575,7 +595,8 @@ public class JsonConverter { .build(); } - private static TransportProtos.ProvisionDeviceCredentialsMsg buildProvisionDeviceCredentialsMsg(String provisionKey, String provisionSecret) { + private static TransportProtos.ProvisionDeviceCredentialsMsg buildProvisionDeviceCredentialsMsg(String + provisionKey, String provisionSecret) { return TransportProtos.ProvisionDeviceCredentialsMsg.newBuilder() .setProvisionDeviceKey(provisionKey) .setProvisionDeviceSecret(provisionSecret) 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 893a93e31a..473eeea42c 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 @@ -784,8 +784,8 @@ public class DefaultTransportService implements TransportService { wrappedCallback); } - protected void sendToRuleEngine(TenantId tenantId, TbMsg tbMsg, TbQueueCallback callback) { - TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_RULE_ENGINE, tenantId, tbMsg.getOriginator()); + private void sendToRuleEngine(TenantId tenantId, TbMsg tbMsg, TbQueueCallback callback) { + TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_RULE_ENGINE, tbMsg.getQueueName(), tenantId, tbMsg.getOriginator()); if (log.isTraceEnabled()) { log.trace("[{}][{}] Pushing to topic {} message {}", tenantId, tbMsg.getOriginator(), tpi.getFullTopicName(), tbMsg); } @@ -797,7 +797,7 @@ public class DefaultTransportService implements TransportService { ruleEngineMsgProducer.send(tpi, new TbProtoQueueMsg<>(tbMsg.getId(), msg), wrappedCallback); } - protected void sendToRuleEngine(TenantId tenantId, DeviceId deviceId, TransportProtos.SessionInfoProto sessionInfo, JsonObject json, + private void sendToRuleEngine(TenantId tenantId, DeviceId deviceId, TransportProtos.SessionInfoProto sessionInfo, JsonObject json, TbMsgMetaData metaData, SessionMsgType sessionMsgType, TbQueueCallback callback) { DeviceProfileId deviceProfileId = new DeviceProfileId(new UUID(sessionInfo.getDeviceProfileIdMSB(), sessionInfo.getDeviceProfileIdLSB())); DeviceProfile deviceProfile = deviceProfileCache.get(deviceProfileId); diff --git a/common/transport/transport-api/src/test/java/JsonConverterTest.java b/common/transport/transport-api/src/test/java/JsonConverterTest.java new file mode 100644 index 0000000000..cedbef50c9 --- /dev/null +++ b/common/transport/transport-api/src/test/java/JsonConverterTest.java @@ -0,0 +1,53 @@ +/** + * Copyright © 2016-2021 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. + */ + +import com.google.gson.JsonParser; +import org.junit.Assert; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.junit.MockitoJUnitRunner; +import org.thingsboard.server.common.transport.adaptor.JsonConverter; + +@RunWith(MockitoJUnitRunner.class) +public class JsonConverterTest { + + private static final JsonParser JSON_PARSER = new JsonParser(); + + @Test + public void testParseBigDecimalAsLong() { + var result = JsonConverter.convertToTelemetry(JSON_PARSER.parse("{\"meterReadingDelta\": 1E+1}"), 0L); + Assert.assertEquals(10L, result.get(0L).get(0).getLongValue().get().longValue()); + } + + @Test + public void testParseBigDecimalAsDouble() { + var result = JsonConverter.convertToTelemetry(JSON_PARSER.parse("{\"meterReadingDelta\": 101E-1}"), 0L); + Assert.assertEquals(10.1, result.get(0L).get(0).getDoubleValue().get(), 0.0); + } + + @Test + public void testParseAsDouble() { + var result = JsonConverter.convertToTelemetry(JSON_PARSER.parse("{\"meterReadingDelta\": 1.1}"), 0L); + Assert.assertEquals(1.1, result.get(0L).get(0).getDoubleValue().get(), 0.0); + } + + @Test + public void testParseAsLong() { + var result = JsonConverter.convertToTelemetry(JSON_PARSER.parse("{\"meterReadingDelta\": 11}"), 0L); + Assert.assertEquals(11L, result.get(0L).get(0).getLongValue().get().longValue()); + } + +} diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbCopyAttributesToEntityViewNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbCopyAttributesToEntityViewNode.java index db95a5986c..152f8fbe49 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbCopyAttributesToEntityViewNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbCopyAttributesToEntityViewNode.java @@ -76,7 +76,7 @@ public class TbCopyAttributesToEntityViewNode implements TbNode { if (!msg.getMetaData().getData().isEmpty()) { long now = System.currentTimeMillis(); String scope = msg.getType().equals(SessionMsgType.POST_ATTRIBUTES_REQUEST.name()) ? - DataConstants.CLIENT_SCOPE : msg.getMetaData().getValue("scope"); + DataConstants.CLIENT_SCOPE : msg.getMetaData().getValue(DataConstants.SCOPE); ListenableFuture> entityViewsFuture = ctx.getEntityViewService().findEntityViewsByTenantIdAndEntityIdAsync(ctx.getTenantId(), msg.getOriginator()); diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/debug/TbMsgGeneratorNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/debug/TbMsgGeneratorNode.java index 69482a7b89..6047370193 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/debug/TbMsgGeneratorNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/debug/TbMsgGeneratorNode.java @@ -104,9 +104,10 @@ public class TbMsgGeneratorNode implements TbNode { } }, t -> { - if (initialized) { + if (initialized && (config.getMsgCount() == TbMsgGeneratorNodeConfiguration.UNLIMITED_MSG_COUNT || currentMsgCount < config.getMsgCount())) { ctx.tellFailure(msg, t); scheduleTickMsg(ctx); + currentMsgCount++; } }); } diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/CalculateDeltaNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/CalculateDeltaNode.java index 0cd8a8b08a..63b326a816 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/CalculateDeltaNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/CalculateDeltaNode.java @@ -33,6 +33,7 @@ import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.kv.TsKvEntry; import org.thingsboard.server.common.data.plugin.ComponentType; import org.thingsboard.server.common.msg.TbMsg; +import org.thingsboard.server.common.msg.session.SessionMsgType; import org.thingsboard.server.dao.timeseries.TimeseriesService; import org.thingsboard.server.dao.util.mapping.JacksonUtil; @@ -44,11 +45,13 @@ import java.util.concurrent.ConcurrentHashMap; @Slf4j @RuleNode(type = ComponentType.ENRICHMENT, - name = "calculate delta", + name = "calculate delta", relationTypes = {"Success", "Failure", "Other"}, configClazz = CalculateDeltaNodeConfiguration.class, nodeDescription = "Calculates and adds 'delta' value into message based on the incoming and previous value", nodeDetails = "Calculates delta and period based on the previous time-series reading and current data. " + - "Delta calculation is done in scope of the message originator, e.g. device, asset or customer.", + "Delta calculation is done in scope of the message originator, e.g. device, asset or customer. " + + "If there is input key, the output relation will be 'Success' unless delta is negative and corresponding configuration parameter is set. " + + "If there is no input value key in the incoming message, the output relation will be 'Other'.", uiResources = {"static/rulenode/rulenode-core-config.js"}, configDirective = "tbEnrichmentNodeCalculateDeltaConfig") public class CalculateDeltaNode implements TbNode { @@ -72,43 +75,50 @@ public class CalculateDeltaNode implements TbNode { @Override public void onMsg(TbContext ctx, TbMsg msg) { - JsonNode json = JacksonUtil.toJsonNode(msg.getData()); - String inputKey = config.getInputValueKey(); - if (json.has(inputKey)) { - DonAsynchron.withCallback(getLastValue(msg.getOriginator()), - previousData -> { - double currentValue = json.get(inputKey).asDouble(); - long currentTs = TbMsgTimeseriesNode.getTs(msg); - - if (useCache) { - cache.put(msg.getOriginator(), new ValueWithTs(currentTs, currentValue)); - } - - BigDecimal delta = BigDecimal.valueOf(previousData != null ? currentValue - previousData.value : 0.0); - - if (config.isTellFailureIfDeltaIsNegative() && delta.doubleValue() < 0) { - ctx.tellNext(msg, TbRelationTypes.FAILURE); - return; - } - - if (config.getRound() != null) { - delta = delta.setScale(config.getRound(), RoundingMode.HALF_UP); - } - - ObjectNode result = (ObjectNode) json; - result.put(config.getOutputValueKey(), delta); - - if (config.isAddPeriodBetweenMsgs()) { - long period = previousData != null ? currentTs - previousData.ts : 0; - result.put(config.getPeriodValueKey(), period); - } - ctx.tellSuccess(TbMsg.transformMsg(msg, msg.getType(), msg.getOriginator(), msg.getMetaData(), JacksonUtil.toString(result))); - }, - t -> ctx.tellFailure(msg, t), ctx.getDbCallbackExecutor()); - } else if (config.isTellFailureIfInputValueKeyIsAbsent()) { - ctx.tellNext(msg, TbRelationTypes.FAILURE); + if (msg.getType().equals(SessionMsgType.POST_TELEMETRY_REQUEST.name())) { + JsonNode json = JacksonUtil.toJsonNode(msg.getData()); + String inputKey = config.getInputValueKey(); + if (json.has(inputKey)) { + DonAsynchron.withCallback(getLastValue(msg.getOriginator()), + previousData -> { + double currentValue = json.get(inputKey).asDouble(); + long currentTs = TbMsgTimeseriesNode.getTs(msg); + + if (useCache) { + cache.put(msg.getOriginator(), new ValueWithTs(currentTs, currentValue)); + } + + BigDecimal delta = BigDecimal.valueOf(previousData != null ? currentValue - previousData.value : 0.0); + + if (config.isTellFailureIfDeltaIsNegative() && delta.doubleValue() < 0) { + ctx.tellNext(msg, TbRelationTypes.FAILURE); + return; + } + + + if (config.getRound() != null) { + delta = delta.setScale(config.getRound(), RoundingMode.HALF_UP); + } + + ObjectNode result = (ObjectNode) json; + if (delta.stripTrailingZeros().scale() > 0) { + result.put(config.getOutputValueKey(), delta.doubleValue()); + } else { + result.put(config.getOutputValueKey(), delta.longValueExact()); + } + + if (config.isAddPeriodBetweenMsgs()) { + long period = previousData != null ? currentTs - previousData.ts : 0; + result.put(config.getPeriodValueKey(), period); + } + ctx.tellSuccess(TbMsg.transformMsg(msg, msg.getType(), msg.getOriginator(), msg.getMetaData(), JacksonUtil.toString(result))); + }, + t -> ctx.tellFailure(msg, t), ctx.getDbCallbackExecutor()); + } else { + ctx.tellNext(msg, "Other"); + } } else { - ctx.tellSuccess(msg); + ctx.tellNext(msg, "Other"); } } diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/CalculateDeltaNodeConfiguration.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/CalculateDeltaNodeConfiguration.java index b16a97caec..bec93a1d90 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/CalculateDeltaNodeConfiguration.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/CalculateDeltaNodeConfiguration.java @@ -15,10 +15,12 @@ */ package org.thingsboard.rule.engine.metadata; +import com.fasterxml.jackson.annotation.JsonIgnoreProperties; import lombok.Data; import org.thingsboard.rule.engine.api.NodeConfiguration; @Data +@JsonIgnoreProperties(ignoreUnknown = true) public class CalculateDeltaNodeConfiguration implements NodeConfiguration { private String inputValueKey; private String outputValueKey; @@ -26,7 +28,6 @@ public class CalculateDeltaNodeConfiguration implements NodeConfiguration attributes = JsonConverter.convertToAttributes(new JsonParser().parse(msg.getData())); - String scope = msg.getMetaData().getValue("scope"); + String scope = msg.getMetaData().getValue(DataConstants.SCOPE); if (StringUtils.isEmpty(scope)) { scope = DataConstants.CLIENT_SCOPE; } - return processAttributesUpdate(ctx, msg, attributes, scope); + return processAttributes(ctx, msg, scope); } private boolean processAttributesDeleteNotification(TbContext ctx, TbMsg msg) throws ExecutionException, InterruptedException { boolean stateChanged = false; List keys = new ArrayList<>(); new JsonParser().parse(msg.getData()).getAsJsonObject().get("attributes").getAsJsonArray().forEach(e -> keys.add(e.getAsString())); - String scope = msg.getMetaData().getValue("scope"); + String scope = msg.getMetaData().getValue(DataConstants.SCOPE); if (StringUtils.isEmpty(scope)) { scope = DataConstants.CLIENT_SCOPE; } @@ -211,12 +221,12 @@ class DeviceState { } protected boolean processAttributesUpdateRequest(TbContext ctx, TbMsg msg) throws ExecutionException, InterruptedException { - Set attributes = JsonConverter.convertToAttributes(new JsonParser().parse(msg.getData())); - return processAttributesUpdate(ctx, msg, attributes, DataConstants.CLIENT_SCOPE); + return processAttributes(ctx, msg, DataConstants.CLIENT_SCOPE); } - private boolean processAttributesUpdate(TbContext ctx, TbMsg msg, Set attributes, String scope) throws ExecutionException, InterruptedException { + private boolean processAttributes(TbContext ctx, TbMsg msg, String scope) throws ExecutionException, InterruptedException { boolean stateChanged = false; + Set attributes = JsonConverter.convertToAttributes(new JsonParser().parse(msg.getData())); if (!attributes.isEmpty()) { SnapshotUpdate update = merge(latestValues, attributes, scope); for (DeviceProfileAlarm alarm : deviceProfile.getAlarmSettings()) { diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/TbDeviceProfileNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/TbDeviceProfileNode.java index 3faac79420..9b891ea8eb 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/TbDeviceProfileNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/TbDeviceProfileNode.java @@ -134,7 +134,8 @@ public class TbDeviceProfileNode implements TbNode { if (deviceState != null) { deviceState.process(ctx, msg); } else { - ctx.tellFailure(msg, new IllegalStateException("Device profile for device [" + deviceId + "] not found!")); + log.info("Device was not found! Most probably device [" + deviceId + "] has been removed from the database. Acknowledging msg."); + ctx.ack(msg); } } } else { diff --git a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/profile/TbDeviceProfileNodeTest.java b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/profile/TbDeviceProfileNodeTest.java index 9282d2c357..3cdac17b42 100644 --- a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/profile/TbDeviceProfileNodeTest.java +++ b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/profile/TbDeviceProfileNodeTest.java @@ -434,11 +434,152 @@ public class TbDeviceProfileNodeTest { verify(ctx, Mockito.never()).tellFailure(Mockito.any(), Mockito.any()); } - private void init() throws TbNodeException { + @Test + public void testTenantInheritModeForDynamicValues() throws Exception { + init(); + + DeviceProfile deviceProfile = new DeviceProfile(); + DeviceProfileData deviceProfileData = new DeviceProfileData(); + + AttributeKvCompositeKey compositeKey = new AttributeKvCompositeKey( + EntityType.TENANT, deviceId.getId(), "SERVER_SCOPE", "tenantAttribute" + ); + + AttributeKvEntity attributeKvEntity = new AttributeKvEntity(); + attributeKvEntity.setId(compositeKey); + attributeKvEntity.setLongValue(100L); + attributeKvEntity.setLastUpdateTs(0L); + + AttributeKvEntry entry = attributeKvEntity.toData(); + ListenableFuture> listListenableFutureWithLess = + Futures.immediateFuture(Collections.singletonList(entry)); + + KeyFilter lowTempFilter = new KeyFilter(); + lowTempFilter.setKey(new EntityKey(EntityKeyType.TIME_SERIES, "temperature")); + lowTempFilter.setValueType(EntityKeyValueType.NUMERIC); + NumericFilterPredicate lowTempPredicate = new NumericFilterPredicate(); + lowTempPredicate.setOperation(NumericFilterPredicate.NumericOperation.GREATER); + lowTempPredicate.setValue( + new FilterPredicateValue<>( + 0.0, + null, + new DynamicValue<>(DynamicValueSourceType.CURRENT_DEVICE, "tenantAttribute", true)) + ); + lowTempFilter.setPredicate(lowTempPredicate); + AlarmCondition alarmCondition = new AlarmCondition(); + alarmCondition.setCondition(Collections.singletonList(lowTempFilter)); + AlarmRule alarmRule = new AlarmRule(); + alarmRule.setCondition(alarmCondition); + DeviceProfileAlarm dpa = new DeviceProfileAlarm(); + dpa.setId("lesstempID"); + dpa.setAlarmType("lessTemperatureAlarm"); + dpa.setCreateRules(new TreeMap<>(Collections.singletonMap(AlarmSeverity.CRITICAL, alarmRule))); + + deviceProfileData.setAlarms(Collections.singletonList(dpa)); + deviceProfile.setProfileData(deviceProfileData); - UUID uuid = new UUID(6041557255264276971L, -9019477126543226049L); - System.out.println(uuid); + Mockito.when(cache.get(tenantId, deviceId)).thenReturn(deviceProfile); + Mockito.when(timeseriesService.findLatest(tenantId, deviceId, Collections.singleton("temperature"))) + .thenReturn(Futures.immediateFuture(Collections.emptyList())); + Mockito.when(alarmService.findLatestByOriginatorAndType(tenantId, deviceId, "lessTemperatureAlarm")) + .thenReturn(Futures.immediateFuture(null)); + Mockito.when(alarmService.createOrUpdateAlarm(Mockito.any())) + .thenAnswer(AdditionalAnswers.returnsFirstArg()); + Mockito.when(ctx.getAttributesService()).thenReturn(attributesService); + Mockito.when(attributesService.find(eq(tenantId), eq(deviceId), Mockito.anyString(), Mockito.anySet())) + .thenReturn(listListenableFutureWithLess); + + TbMsg theMsg = TbMsg.newMsg("ALARM", deviceId, new TbMsgMetaData(), ""); + Mockito.when(ctx.newMsg(Mockito.anyString(), Mockito.anyString(), Mockito.any(), Mockito.any(), Mockito.anyString())) + .thenReturn(theMsg); + + ObjectNode data = mapper.createObjectNode(); + data.put("temperature", 150L); + TbMsg msg = TbMsg.newMsg(SessionMsgType.POST_TELEMETRY_REQUEST.name(), deviceId, new TbMsgMetaData(), + TbMsgDataType.JSON, mapper.writeValueAsString(data), null, null); + + node.onMsg(ctx, msg); + verify(ctx).tellSuccess(msg); + verify(ctx).tellNext(theMsg, "Alarm Created"); + verify(ctx, Mockito.never()).tellFailure(Mockito.any(), Mockito.any()); + + } + + @Test + public void testCustomerInheritModeForDynamicValues() throws Exception { + init(); + DeviceProfile deviceProfile = new DeviceProfile(); + DeviceProfileData deviceProfileData = new DeviceProfileData(); + + AttributeKvCompositeKey compositeKey = new AttributeKvCompositeKey( + EntityType.TENANT, deviceId.getId(), "SERVER_SCOPE", "customerAttribute" + ); + + AttributeKvEntity attributeKvEntity = new AttributeKvEntity(); + attributeKvEntity.setId(compositeKey); + attributeKvEntity.setLongValue(100L); + attributeKvEntity.setLastUpdateTs(0L); + + AttributeKvEntry entry = attributeKvEntity.toData(); + ListenableFuture> listListenableFutureWithLess = + Futures.immediateFuture(Collections.singletonList(entry)); + ListenableFuture> optionalListenableFutureWithLess = + Futures.immediateFuture(Optional.of(entry)); + + KeyFilter lowTempFilter = new KeyFilter(); + lowTempFilter.setKey(new EntityKey(EntityKeyType.TIME_SERIES, "temperature")); + lowTempFilter.setValueType(EntityKeyValueType.NUMERIC); + NumericFilterPredicate lowTempPredicate = new NumericFilterPredicate(); + lowTempPredicate.setOperation(NumericFilterPredicate.NumericOperation.GREATER); + lowTempPredicate.setValue( + new FilterPredicateValue<>( + 0.0, + null, + new DynamicValue<>(DynamicValueSourceType.CURRENT_CUSTOMER, "customerAttribute", true)) + ); + lowTempFilter.setPredicate(lowTempPredicate); + AlarmCondition alarmCondition = new AlarmCondition(); + alarmCondition.setCondition(Collections.singletonList(lowTempFilter)); + AlarmRule alarmRule = new AlarmRule(); + alarmRule.setCondition(alarmCondition); + DeviceProfileAlarm dpa = new DeviceProfileAlarm(); + dpa.setId("lesstempID"); + dpa.setAlarmType("lessTemperatureAlarm"); + dpa.setCreateRules(new TreeMap<>(Collections.singletonMap(AlarmSeverity.CRITICAL, alarmRule))); + + deviceProfileData.setAlarms(Collections.singletonList(dpa)); + deviceProfile.setProfileData(deviceProfileData); + + Mockito.when(cache.get(tenantId, deviceId)).thenReturn(deviceProfile); + Mockito.when(timeseriesService.findLatest(tenantId, deviceId, Collections.singleton("temperature"))) + .thenReturn(Futures.immediateFuture(Collections.emptyList())); + Mockito.when(alarmService.findLatestByOriginatorAndType(tenantId, deviceId, "lessTemperatureAlarm")) + .thenReturn(Futures.immediateFuture(null)); + Mockito.when(alarmService.createOrUpdateAlarm(Mockito.any())) + .thenAnswer(AdditionalAnswers.returnsFirstArg()); + Mockito.when(ctx.getAttributesService()).thenReturn(attributesService); + Mockito.when(attributesService.find(eq(tenantId), eq(deviceId), Mockito.anyString(), Mockito.anySet())) + .thenReturn(listListenableFutureWithLess); + Mockito.when(attributesService.find(eq(tenantId), eq(tenantId), eq(DataConstants.SERVER_SCOPE), Mockito.anyString())) + .thenReturn(optionalListenableFutureWithLess); + + TbMsg theMsg = TbMsg.newMsg("ALARM", deviceId, new TbMsgMetaData(), ""); + Mockito.when(ctx.newMsg(Mockito.anyString(), Mockito.anyString(), Mockito.any(), Mockito.any(), Mockito.anyString())) + .thenReturn(theMsg); + + ObjectNode data = mapper.createObjectNode(); + data.put("temperature", 150L); + TbMsg msg = TbMsg.newMsg(SessionMsgType.POST_TELEMETRY_REQUEST.name(), deviceId, new TbMsgMetaData(), + TbMsgDataType.JSON, mapper.writeValueAsString(data), null, null); + + node.onMsg(ctx, msg); + verify(ctx).tellSuccess(msg); + verify(ctx).tellNext(theMsg, "Alarm Created"); + verify(ctx, Mockito.never()).tellFailure(Mockito.any(), Mockito.any()); + } + + private void init() throws TbNodeException { Mockito.when(ctx.getTenantId()).thenReturn(tenantId); Mockito.when(ctx.getDeviceProfileCache()).thenReturn(cache); Mockito.when(ctx.getTimeseriesService()).thenReturn(timeseriesService); diff --git a/ui-ngx/src/app/modules/home/components/alarm/alarm-table-config.ts b/ui-ngx/src/app/modules/home/components/alarm/alarm-table-config.ts index db12d8ff65..743b49c286 100644 --- a/ui-ngx/src/app/modules/home/components/alarm/alarm-table-config.ts +++ b/ui-ngx/src/app/modules/home/components/alarm/alarm-table-config.ts @@ -43,6 +43,7 @@ import { AlarmDetailsDialogComponent, AlarmDetailsDialogData } from '@home/components/alarm/alarm-details-dialog.component'; +import { DAY, historyInterval } from '@shared/models/time/time.models'; export class AlarmTableConfig extends EntityTableConfig { @@ -59,6 +60,7 @@ export class AlarmTableConfig extends EntityTableConfig this.loadDataOnInit = false; this.tableTitle = ''; this.useTimePageLink = true; + this.defaultTimewindowInterval = historyInterval(DAY * 30); this.detailsPanelEnabled = false; this.selectionEnabled = false; this.searchEnabled = true; diff --git a/ui-ngx/src/app/modules/home/components/entity/entities-table.component.ts b/ui-ngx/src/app/modules/home/components/entity/entities-table.component.ts index dedaced162..da735c1567 100644 --- a/ui-ngx/src/app/modules/home/components/entity/entities-table.component.ts +++ b/ui-ngx/src/app/modules/home/components/entity/entities-table.component.ts @@ -55,11 +55,11 @@ import { EntityTypeTranslation } from '@shared/models/entity-type.models'; import { DialogService } from '@core/services/dialog.service'; import { AddEntityDialogComponent } from './add-entity-dialog.component'; import { AddEntityDialogData, EntityAction } from '@home/models/entity/entity-component.models'; -import { DAY, historyInterval, HistoryWindowType, Timewindow } from '@shared/models/time/time.models'; +import { HistoryWindowType, Timewindow } from '@shared/models/time/time.models'; import { DomSanitizer, SafeHtml } from '@angular/platform-browser'; import { TbAnchorComponent } from '@shared/components/tb-anchor.component'; import { isDefined, isUndefined } from '@core/utils'; -import { HasUUID } from '../../../../shared/models/id/has-uuid'; +import { HasUUID } from '@shared/models/id/has-uuid'; @Component({ selector: 'tb-entities-table', @@ -202,7 +202,7 @@ export class EntitiesTableComponent extends PageComponent implements AfterViewIn this.pageSizeOptions = [this.defaultPageSize, this.defaultPageSize * 2, this.defaultPageSize * 3]; if (this.entitiesTableConfig.useTimePageLink) { - this.timewindow = historyInterval(DAY); + this.timewindow = this.entitiesTableConfig.defaultTimewindowInterval; const currentTime = Date.now(); this.pageLink = new TimePageLink(10, 0, null, sortOrder, currentTime - this.timewindow.history.timewindowMs, currentTime); @@ -446,7 +446,7 @@ export class EntitiesTableComponent extends PageComponent implements AfterViewIn resetSortAndFilter(update: boolean = true, preserveTimewindow: boolean = false) { this.pageLink.textSearch = null; if (this.entitiesTableConfig.useTimePageLink && !preserveTimewindow) { - this.timewindow = historyInterval(DAY); + this.timewindow = this.entitiesTableConfig.defaultTimewindowInterval; } if (this.displayPagination) { this.paginator.pageIndex = 0; diff --git a/ui-ngx/src/app/modules/home/components/filter/boolean-filter-predicate.component.html b/ui-ngx/src/app/modules/home/components/filter/boolean-filter-predicate.component.html index 6194581625..253f7e2138 100644 --- a/ui-ngx/src/app/modules/home/components/filter/boolean-filter-predicate.component.html +++ b/ui-ngx/src/app/modules/home/components/filter/boolean-filter-predicate.component.html @@ -16,7 +16,7 @@ -->
- + @@ -25,7 +25,7 @@ diff --git a/ui-ngx/src/app/modules/home/components/filter/filter-predicate-list.component.html b/ui-ngx/src/app/modules/home/components/filter/filter-predicate-list.component.html index 41c04ea238..c6aee52976 100644 --- a/ui-ngx/src/app/modules/home/components/filter/filter-predicate-list.component.html +++ b/ui-ngx/src/app/modules/home/components/filter/filter-predicate-list.component.html @@ -26,12 +26,12 @@
-
+
- +
diff --git a/ui-ngx/src/app/modules/home/components/filter/filter-predicate-value.component.html b/ui-ngx/src/app/modules/home/components/filter/filter-predicate-value.component.html index adda4bea1a..dd1b92cf7f 100644 --- a/ui-ngx/src/app/modules/home/components/filter/filter-predicate-value.component.html +++ b/ui-ngx/src/app/modules/home/components/filter/filter-predicate-value.component.html @@ -47,7 +47,7 @@
-
+
@@ -68,6 +68,14 @@
filter.source-attribute
+
+ + {{ 'filter.inherit-owner' | translate}} + +
filter.source-attribute-not-set
+