From 70fe28a5ac172bbc7aa093fdee04e8151118485e Mon Sep 17 00:00:00 2001 From: YevhenBondarenko Date: Tue, 20 Oct 2020 13:59:14 +0300 Subject: [PATCH] added usage stats consumer and producer to queue providers --- .../queue/DefaultTbCoreConsumerService.java | 20 +++++++---- .../common/msg/queue/TopicPartitionInfo.java | 4 +++ .../provider/TbCoreQueueProducerProvider.java | 34 ++++++++++++------ .../provider/TbQueueProducerProvider.java | 8 ++--- .../TbRuleEngineProducerProvider.java | 35 ++++++++++++------- .../TbTransportQueueProducerProvider.java | 25 +++++++------ .../usagestats/DefaultTbUsageStatsClient.java | 2 +- 7 files changed, 81 insertions(+), 47 deletions(-) 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 51fce853db..929d5000f0 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 @@ -30,28 +30,29 @@ import org.thingsboard.server.common.msg.MsgType; import org.thingsboard.server.common.msg.TbActorMsg; import org.thingsboard.server.common.msg.queue.ServiceType; import org.thingsboard.server.common.msg.queue.TbCallback; +import org.thingsboard.server.common.stats.StatsFactory; +import org.thingsboard.server.common.transport.util.DataDecodingEncodingService; import org.thingsboard.server.dao.util.mapping.JacksonUtil; -import org.thingsboard.server.gen.transport.TransportProtos.ToUsageStatsServiceMsg; import org.thingsboard.server.gen.transport.TransportProtos.DeviceStateServiceMsgProto; import org.thingsboard.server.gen.transport.TransportProtos.FromDeviceRPCResponseProto; import org.thingsboard.server.gen.transport.TransportProtos.LocalSubscriptionServiceMsgProto; import org.thingsboard.server.gen.transport.TransportProtos.SubscriptionMgrMsgProto; -import org.thingsboard.server.gen.transport.TransportProtos.TbAttributeUpdateProto; -import org.thingsboard.server.gen.transport.TransportProtos.TbAttributeDeleteProto; -import org.thingsboard.server.gen.transport.TransportProtos.TbAlarmUpdateProto; import org.thingsboard.server.gen.transport.TransportProtos.TbAlarmDeleteProto; +import org.thingsboard.server.gen.transport.TransportProtos.TbAlarmUpdateProto; +import org.thingsboard.server.gen.transport.TransportProtos.TbAttributeDeleteProto; +import org.thingsboard.server.gen.transport.TransportProtos.TbAttributeUpdateProto; import org.thingsboard.server.gen.transport.TransportProtos.TbSubscriptionCloseProto; import org.thingsboard.server.gen.transport.TransportProtos.TbTimeSeriesUpdateProto; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg; +import org.thingsboard.server.gen.transport.TransportProtos.ToUsageStatsServiceMsg; import org.thingsboard.server.gen.transport.TransportProtos.TransportToDeviceActorMsg; import org.thingsboard.server.queue.TbQueueConsumer; import org.thingsboard.server.queue.common.TbProtoQueueMsg; import org.thingsboard.server.queue.discovery.PartitionChangeEvent; import org.thingsboard.server.queue.provider.TbCoreQueueFactory; -import org.thingsboard.server.common.stats.StatsFactory; import org.thingsboard.server.queue.util.TbCoreComponent; -import org.thingsboard.server.common.transport.util.DataDecodingEncodingService; +import org.thingsboard.server.service.apiusage.TbApiUsageStateService; import org.thingsboard.server.service.profile.TbDeviceProfileCache; import org.thingsboard.server.service.queue.processing.AbstractConsumerService; import org.thingsboard.server.service.rpc.FromDeviceRpcResponse; @@ -62,7 +63,6 @@ import org.thingsboard.server.service.subscription.SubscriptionManagerService; import org.thingsboard.server.service.subscription.TbLocalSubscriptionService; import org.thingsboard.server.service.subscription.TbSubscriptionUtils; import org.thingsboard.server.service.transport.msg.TransportToDeviceActorMsgWrapper; -import org.thingsboard.server.service.apiusage.TbApiUsageStateService; import javax.annotation.PostConstruct; import javax.annotation.PreDestroy; @@ -143,6 +143,12 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService tpi.newByTopic(usageStatsConsumer.getTopic())) + .collect(Collectors.toSet())); } } diff --git a/common/message/src/main/java/org/thingsboard/server/common/msg/queue/TopicPartitionInfo.java b/common/message/src/main/java/org/thingsboard/server/common/msg/queue/TopicPartitionInfo.java index bf0fffe404..c2dec02612 100644 --- a/common/message/src/main/java/org/thingsboard/server/common/msg/queue/TopicPartitionInfo.java +++ b/common/message/src/main/java/org/thingsboard/server/common/msg/queue/TopicPartitionInfo.java @@ -50,6 +50,10 @@ public class TopicPartitionInfo { this.fullTopicName = tmp; } + public TopicPartitionInfo newByTopic(String topic) { + return new TopicPartitionInfo(topic, this.tenantId, this.partition, this.myPartition); + } + public String getTopic() { return topic; } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbCoreQueueProducerProvider.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbCoreQueueProducerProvider.java index 9fe5de0bcc..e620259960 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbCoreQueueProducerProvider.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbCoreQueueProducerProvider.java @@ -17,7 +17,12 @@ package org.thingsboard.server.queue.provider; import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; import org.springframework.stereotype.Service; -import org.thingsboard.server.gen.transport.TransportProtos; +import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; +import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg; +import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg; +import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineNotificationMsg; +import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg; +import org.thingsboard.server.gen.transport.TransportProtos.ToUsageStatsServiceMsg; import org.thingsboard.server.queue.TbQueueProducer; import org.thingsboard.server.queue.common.TbProtoQueueMsg; @@ -28,11 +33,12 @@ import javax.annotation.PostConstruct; public class TbCoreQueueProducerProvider implements TbQueueProducerProvider { private final TbCoreQueueFactory tbQueueProvider; - private TbQueueProducer> toTransport; - private TbQueueProducer> toRuleEngine; - private TbQueueProducer> toTbCore; - private TbQueueProducer> toRuleEngineNotifications; - private TbQueueProducer> toTbCoreNotifications; + private TbQueueProducer> toTransport; + private TbQueueProducer> toRuleEngine; + private TbQueueProducer> toTbCore; + private TbQueueProducer> toRuleEngineNotifications; + private TbQueueProducer> toTbCoreNotifications; + private TbQueueProducer> toUsageStats; public TbCoreQueueProducerProvider(TbCoreQueueFactory tbQueueProvider) { this.tbQueueProvider = tbQueueProvider; @@ -45,30 +51,36 @@ public class TbCoreQueueProducerProvider implements TbQueueProducerProvider { this.toRuleEngine = tbQueueProvider.createRuleEngineMsgProducer(); this.toRuleEngineNotifications = tbQueueProvider.createRuleEngineNotificationsMsgProducer(); this.toTbCoreNotifications = tbQueueProvider.createTbCoreNotificationsMsgProducer(); + this.toUsageStats = tbQueueProvider.createToUsageStatsServiceMsgProducer(); } @Override - public TbQueueProducer> getTransportNotificationsMsgProducer() { + public TbQueueProducer> getTransportNotificationsMsgProducer() { return toTransport; } @Override - public TbQueueProducer> getRuleEngineMsgProducer() { + public TbQueueProducer> getRuleEngineMsgProducer() { return toRuleEngine; } @Override - public TbQueueProducer> getRuleEngineNotificationsMsgProducer() { + public TbQueueProducer> getRuleEngineNotificationsMsgProducer() { return toRuleEngineNotifications; } @Override - public TbQueueProducer> getTbCoreMsgProducer() { + public TbQueueProducer> getTbCoreMsgProducer() { return toTbCore; } @Override - public TbQueueProducer> getTbCoreNotificationsMsgProducer() { + public TbQueueProducer> getTbCoreNotificationsMsgProducer() { return toTbCoreNotifications; } + + @Override + public TbQueueProducer> getTbUsageStatsMsgProducer() { + return toUsageStats; + } } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbQueueProducerProvider.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbQueueProducerProvider.java index fa20f11960..8c3a0b6565 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbQueueProducerProvider.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbQueueProducerProvider.java @@ -15,12 +15,12 @@ */ package org.thingsboard.server.queue.provider; -import org.thingsboard.server.gen.transport.TransportProtos.ToUsageStatsServiceMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineNotificationMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg; +import org.thingsboard.server.gen.transport.TransportProtos.ToUsageStatsServiceMsg; import org.thingsboard.server.queue.TbQueueProducer; import org.thingsboard.server.queue.common.TbProtoQueueMsg; @@ -69,9 +69,5 @@ public interface TbQueueProducerProvider { * * @return */ - default TbQueueProducer> getTbUsageStatsMsgProducer() { - //TODO: implement - return null; - } - + TbQueueProducer> getTbUsageStatsMsgProducer(); } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbRuleEngineProducerProvider.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbRuleEngineProducerProvider.java index 69de7ca3e2..309def1269 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbRuleEngineProducerProvider.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbRuleEngineProducerProvider.java @@ -17,7 +17,12 @@ package org.thingsboard.server.queue.provider; import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; import org.springframework.stereotype.Service; -import org.thingsboard.server.gen.transport.TransportProtos; +import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; +import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg; +import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg; +import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineNotificationMsg; +import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg; +import org.thingsboard.server.gen.transport.TransportProtos.ToUsageStatsServiceMsg; import org.thingsboard.server.queue.TbQueueProducer; import org.thingsboard.server.queue.common.TbProtoQueueMsg; @@ -28,12 +33,12 @@ import javax.annotation.PostConstruct; public class TbRuleEngineProducerProvider implements TbQueueProducerProvider { private final TbRuleEngineQueueFactory tbQueueProvider; - private TbQueueProducer> toTransport; - private TbQueueProducer> toRuleEngine; - private TbQueueProducer> toTbCore; - private TbQueueProducer> toRuleEngineNotifications; - private TbQueueProducer> toTbCoreNotifications; - + private TbQueueProducer> toTransport; + private TbQueueProducer> toRuleEngine; + private TbQueueProducer> toTbCore; + private TbQueueProducer> toRuleEngineNotifications; + private TbQueueProducer> toTbCoreNotifications; + private TbQueueProducer> toUsageStats; public TbRuleEngineProducerProvider(TbRuleEngineQueueFactory tbQueueProvider) { this.tbQueueProvider = tbQueueProvider; @@ -46,30 +51,36 @@ public class TbRuleEngineProducerProvider implements TbQueueProducerProvider { this.toRuleEngine = tbQueueProvider.createRuleEngineMsgProducer(); this.toRuleEngineNotifications = tbQueueProvider.createRuleEngineNotificationsMsgProducer(); this.toTbCoreNotifications = tbQueueProvider.createTbCoreNotificationsMsgProducer(); + this.toUsageStats = tbQueueProvider.createToUsageStatsServiceMsgProducer(); } @Override - public TbQueueProducer> getTransportNotificationsMsgProducer() { + public TbQueueProducer> getTransportNotificationsMsgProducer() { return toTransport; } @Override - public TbQueueProducer> getRuleEngineMsgProducer() { + public TbQueueProducer> getRuleEngineMsgProducer() { return toRuleEngine; } @Override - public TbQueueProducer> getRuleEngineNotificationsMsgProducer() { + public TbQueueProducer> getRuleEngineNotificationsMsgProducer() { return toRuleEngineNotifications; } @Override - public TbQueueProducer> getTbCoreMsgProducer() { + public TbQueueProducer> getTbCoreMsgProducer() { return toTbCore; } @Override - public TbQueueProducer> getTbCoreNotificationsMsgProducer() { + public TbQueueProducer> getTbCoreNotificationsMsgProducer() { return toTbCoreNotifications; } + + @Override + public TbQueueProducer> getTbUsageStatsMsgProducer() { + return toUsageStats; + } } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbTransportQueueProducerProvider.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbTransportQueueProducerProvider.java index 9d11aa727b..dee6508ef4 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbTransportQueueProducerProvider.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbTransportQueueProducerProvider.java @@ -17,7 +17,12 @@ package org.thingsboard.server.queue.provider; import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; import org.springframework.stereotype.Service; -import org.thingsboard.server.gen.transport.TransportProtos; +import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; +import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg; +import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg; +import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineNotificationMsg; +import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg; +import org.thingsboard.server.gen.transport.TransportProtos.ToUsageStatsServiceMsg; import org.thingsboard.server.queue.TbQueueProducer; import org.thingsboard.server.queue.common.TbProtoQueueMsg; @@ -28,9 +33,9 @@ import javax.annotation.PostConstruct; public class TbTransportQueueProducerProvider implements TbQueueProducerProvider { private final TbTransportQueueFactory tbQueueProvider; - private TbQueueProducer> toRuleEngine; - private TbQueueProducer> toTbCore; - private TbQueueProducer> toUsageStats; + private TbQueueProducer> toRuleEngine; + private TbQueueProducer> toTbCore; + private TbQueueProducer> toUsageStats; public TbTransportQueueProducerProvider(TbTransportQueueFactory tbQueueProvider) { this.tbQueueProvider = tbQueueProvider; @@ -44,32 +49,32 @@ public class TbTransportQueueProducerProvider implements TbQueueProducerProvider } @Override - public TbQueueProducer> getTransportNotificationsMsgProducer() { + public TbQueueProducer> getTransportNotificationsMsgProducer() { throw new RuntimeException("Not Implemented! Should not be used by Transport!"); } @Override - public TbQueueProducer> getRuleEngineMsgProducer() { + public TbQueueProducer> getRuleEngineMsgProducer() { return toRuleEngine; } @Override - public TbQueueProducer> getTbCoreMsgProducer() { + public TbQueueProducer> getTbCoreMsgProducer() { return toTbCore; } @Override - public TbQueueProducer> getRuleEngineNotificationsMsgProducer() { + public TbQueueProducer> getRuleEngineNotificationsMsgProducer() { throw new RuntimeException("Not Implemented! Should not be used by Transport!"); } @Override - public TbQueueProducer> getTbCoreNotificationsMsgProducer() { + public TbQueueProducer> getTbCoreNotificationsMsgProducer() { throw new RuntimeException("Not Implemented! Should not be used by Transport!"); } @Override - public TbQueueProducer> getTbUsageStatsMsgProducer() { + public TbQueueProducer> getTbUsageStatsMsgProducer() { return toUsageStats; } } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/usagestats/DefaultTbUsageStatsClient.java b/common/queue/src/main/java/org/thingsboard/server/queue/usagestats/DefaultTbUsageStatsClient.java index 35bb0c2a6a..5a94cd505e 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/usagestats/DefaultTbUsageStatsClient.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/usagestats/DefaultTbUsageStatsClient.java @@ -91,7 +91,7 @@ public class DefaultTbUsageStatsClient implements TbUsageStatsClient { report.forEach(((tenantId, builder) -> { //TODO: figure out how to minimize messages into the queue. Maybe group by 100s of messages? - TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_CORE, tenantId, tenantId); + TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_CORE, tenantId, tenantId).newByTopic(msgProducer.getDefaultTopic()); msgProducer.send(tpi, new TbProtoQueueMsg<>(UUID.randomUUID(), builder.build()), null); })); log.info("Report statistics for: {} tenants", report.size());