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 ae04de456b..b37e5abff3 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 @@ -15,44 +15,25 @@ */ package org.thingsboard.server.service.queue; -import com.google.protobuf.ProtocolStringList; import lombok.extern.slf4j.Slf4j; -import org.springframework.beans.factory.annotation.Value; import org.springframework.context.ApplicationEventPublisher; import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Service; -import org.thingsboard.common.util.ThingsBoardThreadFactory; import org.thingsboard.server.actors.ActorSystemContext; -import org.thingsboard.server.common.data.id.EntityId; -import org.thingsboard.server.common.data.id.EntityIdFactory; import org.thingsboard.server.common.data.id.QueueId; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.queue.Queue; import org.thingsboard.server.common.data.rpc.RpcError; -import org.thingsboard.server.common.msg.TbMsg; -import org.thingsboard.server.common.msg.gen.MsgProtos; -import org.thingsboard.server.common.msg.queue.QueueToRuleEngineMsg; -import org.thingsboard.server.common.msg.queue.RuleEngineException; -import org.thingsboard.server.common.msg.queue.RuleNodeInfo; import org.thingsboard.server.common.msg.queue.ServiceType; import org.thingsboard.server.common.msg.queue.TbCallback; -import org.thingsboard.server.common.msg.queue.TbMsgCallback; -import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; import org.thingsboard.server.common.msg.rpc.FromDeviceRpcResponse; -import org.thingsboard.server.common.stats.StatsFactory; -import org.thingsboard.server.dao.queue.QueueService; import org.thingsboard.server.dao.tenant.TbTenantProfileCache; import org.thingsboard.server.gen.transport.TransportProtos; -import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineNotificationMsg; -import org.thingsboard.server.queue.TbQueueAdmin; -import org.thingsboard.server.queue.TbQueueConsumer; import org.thingsboard.server.queue.common.TbProtoQueueMsg; import org.thingsboard.server.queue.discovery.PartitionService; import org.thingsboard.server.queue.discovery.QueueKey; -import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; import org.thingsboard.server.queue.discovery.event.PartitionChangeEvent; -import org.thingsboard.server.queue.provider.TbQueueProducerProvider; import org.thingsboard.server.queue.provider.TbRuleEngineQueueFactory; import org.thingsboard.server.queue.util.DataDecodingEncodingService; import org.thingsboard.server.queue.util.TbRuleEngineComponent; @@ -60,102 +41,48 @@ import org.thingsboard.server.service.apiusage.TbApiUsageStateService; import org.thingsboard.server.service.profile.TbAssetProfileCache; import org.thingsboard.server.service.profile.TbDeviceProfileCache; import org.thingsboard.server.service.queue.processing.AbstractConsumerService; -import org.thingsboard.server.service.queue.processing.TbRuleEngineProcessingDecision; -import org.thingsboard.server.service.queue.processing.TbRuleEngineProcessingResult; -import org.thingsboard.server.service.queue.processing.TbRuleEngineProcessingStrategy; -import org.thingsboard.server.service.queue.processing.TbRuleEngineProcessingStrategyFactory; -import org.thingsboard.server.service.queue.processing.TbRuleEngineSubmitStrategy; -import org.thingsboard.server.service.queue.processing.TbRuleEngineSubmitStrategyFactory; +import org.thingsboard.server.service.queue.ruleengine.TbRuleEngineConsumerContext; +import org.thingsboard.server.service.queue.ruleengine.TbRuleEngineQueueConsumerManager; import org.thingsboard.server.service.rpc.TbRuleEngineDeviceRpcService; -import org.thingsboard.server.service.stats.RuleEngineStatisticsService; -import org.threadly.concurrent.wrapper.KeyDistributedExecutor; import javax.annotation.PostConstruct; -import javax.annotation.PreDestroy; import java.util.ArrayList; -import java.util.Collections; -import java.util.HashSet; import java.util.List; -import java.util.Map; import java.util.Optional; -import java.util.Set; import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; -import java.util.concurrent.ScheduledExecutorService; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.locks.ReentrantLock; @Service @TbRuleEngineComponent @Slf4j public class DefaultTbRuleEngineConsumerService extends AbstractConsumerService implements TbRuleEngineConsumerService { - public static final String SUCCESSFUL_STATUS = "successful"; - public static final String FAILED_STATUS = "failed"; - public static final String THREAD_TOPIC_SEPARATOR = " | "; - @Value("${queue.rule-engine.poll-interval}") - private long pollDuration; - @Value("${queue.rule-engine.pack-processing-timeout}") - private long packProcessingTimeout; - @Value("${queue.rule-engine.stats.enabled:true}") - private boolean statsEnabled; - @Value("${queue.rule-engine.prometheus-stats.enabled:false}") - boolean prometheusStatsEnabled; - @Value("${queue.rule-engine.topic-deletion-delay:30}") - private int topicDeletionDelayInSec; - - private final StatsFactory statsFactory; - private final TbRuleEngineSubmitStrategyFactory submitStrategyFactory; - private final TbRuleEngineProcessingStrategyFactory processingStrategyFactory; - private final TbRuleEngineQueueFactory tbRuleEngineQueueFactory; - private final RuleEngineStatisticsService statisticsService; + private final TbRuleEngineConsumerContext ctx; private final TbRuleEngineDeviceRpcService tbDeviceRpcService; - private final TbServiceInfoProvider serviceInfoProvider; - private final QueueService queueService; - private final TbQueueProducerProvider producerProvider; - private final TbQueueAdmin queueAdmin; - - private final ConcurrentMap consumerMap = new ConcurrentHashMap<>(); - private final ExecutorService consumersExecutor = Executors.newCachedThreadPool(ThingsBoardThreadFactory.forName("tb-rule-engine-consumer")); - private final ExecutorService submitExecutor = Executors.newSingleThreadExecutor(ThingsBoardThreadFactory.forName("tb-rule-engine-consumer-submit")); - private final ScheduledExecutorService repartitionExecutor = Executors.newSingleThreadScheduledExecutor(ThingsBoardThreadFactory.forName("tb-rule-engine-consumer-repartition")); + private final ConcurrentMap consumers = new ConcurrentHashMap<>(); - public DefaultTbRuleEngineConsumerService(TbRuleEngineProcessingStrategyFactory processingStrategyFactory, - TbRuleEngineSubmitStrategyFactory submitStrategyFactory, + public DefaultTbRuleEngineConsumerService(TbRuleEngineConsumerContext ctx, TbRuleEngineQueueFactory tbRuleEngineQueueFactory, - RuleEngineStatisticsService statisticsService, ActorSystemContext actorContext, DataDecodingEncodingService encodingService, TbRuleEngineDeviceRpcService tbDeviceRpcService, - StatsFactory statsFactory, TbDeviceProfileCache deviceProfileCache, TbAssetProfileCache assetProfileCache, TbTenantProfileCache tenantProfileCache, TbApiUsageStateService apiUsageStateService, - PartitionService partitionService, ApplicationEventPublisher eventPublisher, - TbServiceInfoProvider serviceInfoProvider, QueueService queueService, - TbQueueProducerProvider producerProvider, TbQueueAdmin queueAdmin) { - super(actorContext, encodingService, tenantProfileCache, deviceProfileCache, assetProfileCache, apiUsageStateService, partitionService, eventPublisher, tbRuleEngineQueueFactory.createToRuleEngineNotificationsMsgConsumer(), Optional.empty()); - this.statisticsService = statisticsService; - this.tbRuleEngineQueueFactory = tbRuleEngineQueueFactory; - this.submitStrategyFactory = submitStrategyFactory; - this.processingStrategyFactory = processingStrategyFactory; + PartitionService partitionService, ApplicationEventPublisher eventPublisher) { + super(actorContext, encodingService, tenantProfileCache, deviceProfileCache, assetProfileCache, apiUsageStateService, partitionService, + eventPublisher, tbRuleEngineQueueFactory.createToRuleEngineNotificationsMsgConsumer(), Optional.empty()); + this.ctx = ctx; this.tbDeviceRpcService = tbDeviceRpcService; - this.statsFactory = statsFactory; - this.serviceInfoProvider = serviceInfoProvider; - this.queueService = queueService; - this.producerProvider = producerProvider; - this.queueAdmin = queueAdmin; } @PostConstruct public void init() { super.init("tb-rule-engine-notifications-consumer"); // TODO: restore init of the main consumer? - List queues = queueService.findAllQueues(); + List queues = ctx.findAllQueues(); for (Queue configuration : queues) { if (partitionService.isManagedByCurrentService(configuration.getTenantId())) { initConsumer(configuration); @@ -164,22 +91,14 @@ public class DefaultTbRuleEngineConsumerService extends AbstractConsumerService< } private void initConsumer(Queue configuration) { - consumerMap.computeIfAbsent(new QueueKey(ServiceType.TB_RULE_ENGINE, configuration), - key -> new TbRuleEngineQueueConsumerManager(repartitionExecutor, consumersExecutor, statsFactory, tbRuleEngineQueueFactory, key)).init(configuration); - } - - @PreDestroy - public void stop() { - super.destroy(); - consumersExecutor.shutdownNow(); // TODO: shutdown or shutdownNow? - submitExecutor.shutdownNow(); - repartitionExecutor.shutdownNow(); + consumers.computeIfAbsent(new QueueKey(ServiceType.TB_RULE_ENGINE, configuration), + key -> new TbRuleEngineQueueConsumerManager(ctx, key)).init(configuration); } @Override protected void onTbApplicationEvent(PartitionChangeEvent event) { if (event.getServiceType().equals(getServiceType())) { - var consumer = consumerMap.get(event.getQueueKey()); + var consumer = consumers.get(event.getQueueKey()); if (consumer != null) { consumer.subscribe(event); } else { @@ -188,207 +107,14 @@ public class DefaultTbRuleEngineConsumerService extends AbstractConsumerService< } } -// void subscribeConsumerPerPartition(QueueKey queue, Set partitions) { -// topicsConsumerPerPartition.get(queue).getSubscribeQueue().add(partitions); -// scheduleTopicRepartition(queue); -// } -// -// private void scheduleTopicRepartition(QueueKey queue) { -// repartitionExecutor.schedule(() -> repartitionTopicWithConsumerPerPartition(queue), 1, TimeUnit.SECONDS); -// } -// -// void repartitionTopicWithConsumerPerPartition(final QueueKey queueKey) { -// if (stopped) { -// return; -// } -// TbTopicWithConsumerPerPartition tbTopicWithConsumerPerPartition = topicsConsumerPerPartition.get(queueKey); -// java.util.Queue> subscribeQueue = tbTopicWithConsumerPerPartition.getSubscribeQueue(); -// if (subscribeQueue.isEmpty()) { -// return; -// } -// if (tbTopicWithConsumerPerPartition.getLock().tryLock()) { -// try { -// Set partitions = null; -// while (!subscribeQueue.isEmpty()) { -// partitions = subscribeQueue.poll(); -// } -// if (partitions == null) { -// return; -// } -// -// Set addedPartitions = new HashSet<>(partitions); -// ConcurrentMap>> consumers = tbTopicWithConsumerPerPartition.getConsumers(); -// addedPartitions.removeAll(consumers.keySet()); -// log.info("calculated addedPartitions {}", addedPartitions); -// -// Set removedPartitions = new HashSet<>(consumers.keySet()); -// removedPartitions.removeAll(partitions); -// log.info("calculated removedPartitions {}", removedPartitions); -// -// removedPartitions.forEach((tpi) -> { -// removeConsumerForTopicByTpi(queueKey.getQueueName(), consumers, tpi); -// }); -// -// addedPartitions.forEach((tpi) -> { -// log.info("[{}] Adding consumer for topic: {}", queueKey, tpi); -// Queue configuration = consumerConfigurations.get(queueKey); -// TbQueueConsumer> consumer = tbRuleEngineQueueFactory.createToRuleEngineMsgConsumer(configuration); -// consumers.put(tpi, consumer); -// launchConsumer(consumer, queueKey, tpi.getFullTopicName(), queueKey + "-" + tpi.getPartition().orElse(-999999)); -// consumer.subscribe(Collections.singleton(tpi)); -// }); -// } finally { -// tbTopicWithConsumerPerPartition.getLock().unlock(); -// } -// } else { -// scheduleTopicRepartition(queueKey); //reschedule later -// } -// } - - void removeConsumerForTopicByTpi(String queue, ConcurrentMap>> consumers, TopicPartitionInfo tpi) { - log.info("[{}] Removing consumer for topic: {}", queue, tpi); - consumers.remove(tpi).stop(); - } - @Override protected void launchMainConsumers() { - consumers.forEach((queue, consumer) -> launchConsumer(consumer, queue, queue, queue.getQueueName())); + consumers.values().forEach(TbRuleEngineQueueConsumerManager::launchMainConsumer); } @Override protected void stopMainConsumers() { - consumerMap.values().forEach(TbRuleEngineQueueConsumerManager::stop); - } - - void launchConsumer(TbQueueConsumer> consumer, QueueKey queueKey, Object consumerKey, String threadSuffix) { - if (isReady) { - log.info("[{}] Launching consumer", consumerKey); - consumersExecutor.execute(consumerKey, () -> consumerLoop(consumer, queueKey, threadSuffix)); - } else { - scheduleLaunchConsumer(consumer, queueKey, consumerKey, threadSuffix); - } - } - - private void scheduleLaunchConsumer(TbQueueConsumer> consumer, QueueKey queueKey, Object consumerKey, String threadSuffix) { - repartitionExecutor.schedule(() -> { - launchConsumer(consumer, queueKey, consumerKey, threadSuffix); - }, 10, TimeUnit.SECONDS); - } - - void consumerLoop(TbQueueConsumer> consumer, QueueKey queueKey, String threadSuffix) { - Queue configuration = consumerConfigurations.get(queueKey); - TbRuleEngineConsumerStats stats = consumerStats.get(queueKey); - updateCurrentThreadName(threadSuffix); - - while (!stopped && !consumer.isStopped() && !consumer.isQueueDeleted()) { - try { - List> msgs = consumer.poll(configuration.getPollInterval()); - if (msgs.isEmpty()) { - continue; - } - final TbRuleEngineSubmitStrategy submitStrategy = getSubmitStrategy(configuration); - final TbRuleEngineProcessingStrategy ackStrategy = getAckStrategy(configuration); - submitStrategy.init(msgs); - while (!stopped && !consumer.isStopped()) { - TbMsgPackProcessingContext ctx = new TbMsgPackProcessingContext(configuration.getName(), submitStrategy, ackStrategy.isSkipTimeoutMsgs()); - submitStrategy.submitAttempt((id, msg) -> submitExecutor.submit(() -> submitMessage(configuration, stats, ctx, id, msg))); - - final boolean timeout = !ctx.await(configuration.getPackProcessingTimeout(), TimeUnit.MILLISECONDS); - - TbRuleEngineProcessingResult result = new TbRuleEngineProcessingResult(configuration.getName(), timeout, ctx); - if (timeout) { - printFirstOrAll(configuration, ctx, ctx.getPendingMap(), "Timeout"); - } - if (!ctx.getFailedMap().isEmpty()) { - printFirstOrAll(configuration, ctx, ctx.getFailedMap(), "Failed"); - } - ctx.printProfilerStats(); - - TbRuleEngineProcessingDecision decision = ackStrategy.analyze(result); - if (statsEnabled) { - stats.log(result, decision.isCommit()); - } - - ctx.cleanup(); - - if (decision.isCommit()) { - submitStrategy.stop(); - break; - } else { - submitStrategy.update(decision.getReprocessMap()); - } - } - consumer.commit(); - } catch (Exception e) { - if (!stopped) { - log.warn("Failed to process messages from queue.", e); - try { - Thread.sleep(pollDuration); - } catch (InterruptedException e2) { - log.trace("Failed to wait until the server has capacity to handle new requests", e2); - } - } - } - } - - if (consumer.isStopped()) { - consumer.unsubscribe(); - } else if (consumer.isQueueDeleted()) { - processQueueDeletion(configuration, consumer); - } - log.info("TB Rule Engine Consumer stopped."); - } - - void updateCurrentThreadName(String threadSuffix) { - String name = Thread.currentThread().getName(); - int spliteratorIndex = name.indexOf(THREAD_TOPIC_SEPARATOR); - if (spliteratorIndex > 0) { - name = name.substring(0, spliteratorIndex); - } - name = name + THREAD_TOPIC_SEPARATOR + threadSuffix; - Thread.currentThread().setName(name); - } - - TbRuleEngineProcessingStrategy getAckStrategy(Queue configuration) { - return processingStrategyFactory.newInstance(configuration.getName(), configuration.getProcessingStrategy()); - } - - TbRuleEngineSubmitStrategy getSubmitStrategy(Queue configuration) { - return submitStrategyFactory.newInstance(configuration.getName(), configuration.getSubmitStrategy()); - } - - void submitMessage(Queue configuration, TbRuleEngineConsumerStats stats, TbMsgPackProcessingContext ctx, UUID id, TbProtoQueueMsg msg) { - log.trace("[{}] Creating callback for topic {} message: {}", id, configuration.getName(), msg.getValue()); - ToRuleEngineMsg toRuleEngineMsg = msg.getValue(); - TenantId tenantId = TenantId.fromUUID(new UUID(toRuleEngineMsg.getTenantIdMSB(), toRuleEngineMsg.getTenantIdLSB())); - TbMsgCallback callback = prometheusStatsEnabled ? - new TbMsgPackCallback(id, tenantId, ctx, stats.getTimer(tenantId, SUCCESSFUL_STATUS), stats.getTimer(tenantId, FAILED_STATUS)) : - new TbMsgPackCallback(id, tenantId, ctx); - try { - if (toRuleEngineMsg.getTbMsg() != null && !toRuleEngineMsg.getTbMsg().isEmpty()) { - forwardToRuleEngineActor(configuration.getName(), tenantId, toRuleEngineMsg, callback); - } else { - callback.onSuccess(); - } - } catch (Exception e) { - callback.onFailure(new RuleEngineException(e.getMessage(), e)); - } - } - - private void printFirstOrAll(Queue configuration, TbMsgPackProcessingContext ctx, Map> map, String prefix) { - boolean printAll = log.isTraceEnabled(); - log.info("{} to process [{}] messages", prefix, map.size()); - for (Map.Entry> pending : map.entrySet()) { - ToRuleEngineMsg tmp = pending.getValue().getValue(); - TbMsg tmpMsg = TbMsg.fromBytes(configuration.getName(), tmp.getTbMsg().toByteArray(), TbMsgCallback.EMPTY); - RuleNodeInfo ruleNodeInfo = ctx.getLastVisitedRuleNode(pending.getKey()); - if (printAll) { - log.trace("[{}] {} to process message: {}, Last Rule Node: {}", TenantId.fromUUID(new UUID(tmp.getTenantIdMSB(), tmp.getTenantIdLSB())), prefix, tmpMsg, ruleNodeInfo); - } else { - log.info("[{}] {} to process message: {}, Last Rule Node: {}", TenantId.fromUUID(new UUID(tmp.getTenantIdMSB(), tmp.getTenantIdLSB())), prefix, tmpMsg, ruleNodeInfo); - break; - } - } + consumers.values().forEach(TbRuleEngineQueueConsumerManager::stop); } @Override @@ -398,18 +124,18 @@ public class DefaultTbRuleEngineConsumerService extends AbstractConsumerService< @Override protected long getNotificationPollDuration() { - return pollDuration; + return ctx.getPollDuration(); } @Override protected long getNotificationPackProcessingTimeout() { - return packProcessingTimeout; + return ctx.getPackProcessingTimeout(); } @Override protected void handleNotification(UUID id, TbProtoQueueMsg msg, TbCallback callback) throws Exception { ToRuleEngineNotificationMsg nfMsg = msg.getValue(); - if (nfMsg.getComponentLifecycleMsg() != null && !nfMsg.getComponentLifecycleMsg().isEmpty()) { + if (!nfMsg.getComponentLifecycleMsg().isEmpty()) { handleComponentLifecycleMsg(id, nfMsg.getComponentLifecycleMsg()); callback.onSuccess(); } else if (nfMsg.hasFromDeviceRpcResponse()) { @@ -420,10 +146,10 @@ public class DefaultTbRuleEngineConsumerService extends AbstractConsumerService< tbDeviceRpcService.processRpcResponseFromDevice(response); callback.onSuccess(); } else if (nfMsg.hasQueueUpdateMsg()) { - repartitionExecutor.execute(() -> updateQueue(nfMsg.getQueueUpdateMsg())); + ctx.getScheduler().execute(() -> updateQueue(nfMsg.getQueueUpdateMsg())); callback.onSuccess(); } else if (nfMsg.hasQueueDeleteMsg()) { - repartitionExecutor.execute(() -> deleteQueue(nfMsg.getQueueDeleteMsg())); + ctx.getScheduler().execute(() -> deleteQueue(nfMsg.getQueueDeleteMsg())); callback.onSuccess(); } else { log.trace("Received notification with missing handler"); @@ -438,33 +164,13 @@ public class DefaultTbRuleEngineConsumerService extends AbstractConsumerService< QueueId queueId = new QueueId(new UUID(queueUpdateMsg.getQueueIdMSB(), queueUpdateMsg.getQueueIdLSB())); String queueName = queueUpdateMsg.getQueueName(); QueueKey queueKey = new QueueKey(ServiceType.TB_RULE_ENGINE, queueName, tenantId); - Queue queue = queueService.findQueueById(tenantId, queueId); - Queue oldQueue = consumerConfigurations.remove(queueKey); - if (oldQueue != null) { - if (oldQueue.isConsumerPerPartition()) { - TbTopicWithConsumerPerPartition consumerPerPartition = topicsConsumerPerPartition.remove(queueKey); - ReentrantLock lock = consumerPerPartition.getLock(); - try { - lock.lock(); - consumerPerPartition.getConsumers().values().forEach(TbQueueConsumer::stop); - } finally { - lock.unlock(); - } - } else { - TbQueueConsumer> consumer = consumers.remove(queueKey); - consumer.stop(); - } - } - - initConsumer(queue); - - if (!queue.isConsumerPerPartition()) { - launchConsumer(consumers.get(queueKey), queueKey, queueKey, queueName); - } + Queue queue = ctx.getQueueService().findQueueById(tenantId, queueId); + consumers.computeIfAbsent(queueKey, key -> new TbRuleEngineQueueConsumerManager(ctx, key)).update(queue); } partitionService.updateQueue(queueUpdateMsg); - partitionService.recalculatePartitions(serviceInfoProvider.getServiceInfo(), new ArrayList<>(partitionService.getOtherServices(ServiceType.TB_RULE_ENGINE))); + partitionService.recalculatePartitions(ctx.getServiceInfoProvider().getServiceInfo(), + new ArrayList<>(partitionService.getOtherServices(ServiceType.TB_RULE_ENGINE))); } private void deleteQueue(TransportProtos.QueueDeleteMsg queueDeleteMsg) { @@ -473,90 +179,18 @@ public class DefaultTbRuleEngineConsumerService extends AbstractConsumerService< QueueKey queueKey = new QueueKey(ServiceType.TB_RULE_ENGINE, queueDeleteMsg.getQueueName(), tenantId); partitionService.removeQueue(queueDeleteMsg); - Queue queue = consumerConfigurations.remove(queueKey); - if (queue != null) { - if (queue.isConsumerPerPartition()) { - TbTopicWithConsumerPerPartition tbTopicWithConsumerPerPartition = topicsConsumerPerPartition.remove(queueKey); - if (tbTopicWithConsumerPerPartition != null) { - tbTopicWithConsumerPerPartition.getConsumers().values().forEach(TbQueueConsumer::onQueueDelete); - tbTopicWithConsumerPerPartition.getConsumers().clear(); - } - } else { - TbQueueConsumer> consumer = consumers.remove(queueKey); - if (consumer != null) { - consumer.onQueueDelete(); - } - } - } - partitionService.recalculatePartitions(serviceInfoProvider.getServiceInfo(), new ArrayList<>(partitionService.getOtherServices(ServiceType.TB_RULE_ENGINE))); - } - - private void forwardToRuleEngineActor(String queueName, TenantId tenantId, ToRuleEngineMsg toRuleEngineMsg, TbMsgCallback callback) { - TbMsg tbMsg = TbMsg.fromBytes(queueName, toRuleEngineMsg.getTbMsg().toByteArray(), callback); - QueueToRuleEngineMsg msg; - ProtocolStringList relationTypesList = toRuleEngineMsg.getRelationTypesList(); - Set relationTypes = null; - if (relationTypesList != null) { - if (relationTypesList.size() == 1) { - relationTypes = Collections.singleton(relationTypesList.get(0)); - } else { - relationTypes = new HashSet<>(relationTypesList); - } - } - msg = new QueueToRuleEngineMsg(tenantId, tbMsg, relationTypes, toRuleEngineMsg.getFailureMessage()); - actorContext.tell(msg); - } - - private void processQueueDeletion(Queue queue, TbQueueConsumer> consumer) { - long finishTs = System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(topicDeletionDelayInSec); - try { - int n = 0; - while (System.currentTimeMillis() <= finishTs) { - List> msgs = consumer.poll(queue.getPollInterval()); - if (msgs.isEmpty()) { - continue; - } - for (TbProtoQueueMsg msg : msgs) { - try { - MsgProtos.TbMsgProto tbMsgProto = MsgProtos.TbMsgProto.parseFrom(msg.getValue().getTbMsg().toByteArray()); - EntityId originator = EntityIdFactory.getByTypeAndUuid(tbMsgProto.getEntityType(), new UUID(tbMsgProto.getEntityIdMSB(), tbMsgProto.getEntityIdLSB())); - - TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_RULE_ENGINE, queue.getName(), TenantId.SYS_TENANT_ID, originator); - producerProvider.getRuleEngineMsgProducer().send(tpi, msg, null); - n++; - } catch (Throwable e) { - log.debug("Failed to move message to system {}: {}", consumer.getTopic(), msg, e); - } - } - consumer.commit(); - } - if (n > 0) { - log.info("Moved {} messages from {} to system {}", n, consumer.getFullTopicNames(), consumer.getTopic()); - } - - consumer.unsubscribe(); - for (String topic : consumer.getFullTopicNames()) { - try { - queueAdmin.deleteTopic(topic); - log.info("Deleted topic {}", topic); - } catch (Exception e) { - log.error("Failed to delete topic {} after unsubscribing", topic, e); - } - } - } catch (Exception e) { - log.error("Failed to process deletion of {} ({})", consumer.getTopic(), queue.getTenantId(), e); + var manager = consumers.remove(queueKey); + if (manager != null) { + manager.delete(); } + partitionService.recalculatePartitions(ctx.getServiceInfoProvider().getServiceInfo(), new ArrayList<>(partitionService.getOtherServices(ServiceType.TB_RULE_ENGINE))); } @Scheduled(fixedDelayString = "${queue.rule-engine.stats.print-interval-ms}") public void printStats() { - if (statsEnabled) { + if (ctx.isStatsEnabled()) { long ts = System.currentTimeMillis(); - consumerStats.forEach((queue, stats) -> { - stats.printStats(); - statisticsService.reportQueueStats(ts, stats); - stats.reset(); - }); + consumers.values().forEach(manager -> manager.printStats(ts)); } } diff --git a/application/src/main/java/org/thingsboard/server/service/queue/TbQueueConsumerLauncher.java b/application/src/main/java/org/thingsboard/server/service/queue/TbQueueConsumerLauncher.java deleted file mode 100644 index f5688eb207..0000000000 --- a/application/src/main/java/org/thingsboard/server/service/queue/TbQueueConsumerLauncher.java +++ /dev/null @@ -1,37 +0,0 @@ -package org.thingsboard.server.service.queue; - -import lombok.Data; -import org.thingsboard.server.common.data.plugin.ComponentLifecycleEvent; -import org.thingsboard.server.common.data.queue.Queue; -import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; -import org.thingsboard.server.gen.transport.TransportProtos; -import org.thingsboard.server.queue.TbQueueConsumer; -import org.thingsboard.server.queue.common.TbProtoQueueMsg; -import org.thingsboard.server.queue.discovery.QueueKey; - -import java.util.Set; -import java.util.concurrent.ExecutionException; -import java.util.concurrent.Future; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.TimeoutException; - -@Data -public class TbQueueConsumerLauncher { - - private final TbQueueConsumer> consumer; - private volatile Future task; - - public void stop() { - this.consumer.stop(); - } - - public void awaitStopped() throws ExecutionException, InterruptedException, TimeoutException { - if (task != null) { - this.task.get(3, TimeUnit.MINUTES); - } - } - - public void subscribe(Set partitions) { - this.consumer.subscribe(partitions); - } -} diff --git a/application/src/main/java/org/thingsboard/server/service/queue/TbQueueConsumerManagerTask.java b/application/src/main/java/org/thingsboard/server/service/queue/TbQueueConsumerManagerTask.java deleted file mode 100644 index 11316a9e05..0000000000 --- a/application/src/main/java/org/thingsboard/server/service/queue/TbQueueConsumerManagerTask.java +++ /dev/null @@ -1,17 +0,0 @@ -package org.thingsboard.server.service.queue; - -import lombok.Data; -import org.thingsboard.server.common.data.plugin.ComponentLifecycleEvent; -import org.thingsboard.server.common.data.queue.Queue; -import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; - -import java.util.Set; - -@Data -public class TbQueueConsumerManagerTask { - - private final ComponentLifecycleEvent event; - private final Queue queue; - private final Set partitions; - -} diff --git a/application/src/main/java/org/thingsboard/server/service/queue/TbRuleEngineQueueConsumerManager.java b/application/src/main/java/org/thingsboard/server/service/queue/TbRuleEngineQueueConsumerManager.java deleted file mode 100644 index 2c6d191c0a..0000000000 --- a/application/src/main/java/org/thingsboard/server/service/queue/TbRuleEngineQueueConsumerManager.java +++ /dev/null @@ -1,205 +0,0 @@ -package org.thingsboard.server.service.queue; - -import lombok.Data; -import lombok.extern.slf4j.Slf4j; -import org.thingsboard.server.common.data.plugin.ComponentLifecycleEvent; -import org.thingsboard.server.common.data.queue.Queue; -import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; -import org.thingsboard.server.common.stats.StatsFactory; -import org.thingsboard.server.gen.transport.TransportProtos; -import org.thingsboard.server.queue.TbQueueConsumer; -import org.thingsboard.server.queue.common.TbProtoQueueMsg; -import org.thingsboard.server.queue.discovery.QueueKey; -import org.thingsboard.server.queue.discovery.event.PartitionChangeEvent; -import org.thingsboard.server.queue.provider.TbRuleEngineQueueFactory; - -import java.util.Collections; -import java.util.HashSet; -import java.util.Set; -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ConcurrentLinkedQueue; -import java.util.concurrent.ConcurrentMap; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.ScheduledExecutorService; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.locks.ReentrantLock; - -@Data -@Slf4j -public class TbRuleEngineQueueConsumerManager { - - private final ScheduledExecutorService scheduler; - private final ExecutorService consumerExecutor; - private final StatsFactory statsFactory; - private final TbRuleEngineQueueFactory queueFactory; - private final QueueKey key; - private final ReentrantLock lock = new ReentrantLock(); //NonfairSync - private final ConcurrentMap consumers = new ConcurrentHashMap<>(); - private final TbRuleEngineConsumerStats stats; - - private volatile Set partitions = Collections.emptySet(); - private volatile Queue queue; - private volatile TbQueueConsumerLauncher mainConsumer; - - private final java.util.Queue tasks = new ConcurrentLinkedQueue<>(); - - public TbRuleEngineQueueConsumerManager(ScheduledExecutorService scheduler, ExecutorService consumerExecutor, StatsFactory statsFactory, TbRuleEngineQueueFactory queueFactory, QueueKey key) { - this.scheduler = scheduler; - this.consumerExecutor = consumerExecutor; - this.statsFactory = statsFactory; - this.queueFactory = queueFactory; - this.key = key; - this.stats = new TbRuleEngineConsumerStats(key, statsFactory); - } - - public void init(Queue queue) { - processTask(new TbQueueConsumerManagerTask(ComponentLifecycleEvent.CREATED, queue, null)); - } - - private void processTask(TbQueueConsumerManagerTask todo) { - tasks.add(todo); - log.info("[{}] Adding task: {}", key, todo); - tryProcessTasks(); - } - - private void tryProcessTasks() { - consumerExecutor.submit(() -> { - if (lock.tryLock()) { - try { - TbQueueConsumerManagerTask lastUpdateTask = null; - while (!tasks.isEmpty()) { - TbQueueConsumerManagerTask task = tasks.poll(); - switch (task.getEvent()) { - case CREATED: - doInit(task.getQueue()); - break; - case UPDATED: - lastUpdateTask = task; - break; - case DELETED: - lastUpdateTask = null; - doDelete(); - break; - } - } - if (lastUpdateTask != null) { - doUpdate(lastUpdateTask.getQueue(), lastUpdateTask.getPartitions()); - } - } finally { - lock.unlock(); - } - } else { - log.debug("[{}] Failed to acquire lock.", key); - scheduler.schedule(this::tryProcessTasks, 1, TimeUnit.SECONDS); - } - }); - } - - public void doInit(Queue queue) { - log.info("[{}] Init consumer with queue: {}", key, queue); - this.queue = queue; - if (!queue.isConsumerPerPartition()) { - mainConsumer = new TbQueueConsumerLauncher(queueFactory.createToRuleEngineMsgConsumer(queue)); - } - } - - private void doUpdate(Queue newQueue, Set partitions) { - if (newQueue.isConsumerPerPartition()) { - - } else { - for (var oldConsumer : consumers.values()) { - oldConsumer.stop(); - } - for (var oldConsumer : consumers.entrySet()) { - try { - oldConsumer.getValue().awaitStopped(); - } catch (Exception e) { - log.info("[{}][{}] Failed to stop the consumer during update", key, oldConsumer.getKey().getPartition().orElse(-1), e); - } - } - if (mainConsumer == null) { - mainConsumer = new TbQueueConsumerLauncher(queueFactory.createToRuleEngineMsgConsumer(queue)); - //TODO: launch - } - mainConsumer.subscribe(partitions); - - } - } - - private void doDelete() { - } - - public void subscribe(PartitionChangeEvent event) { - log.info("[{}] Subscribing to partitions: {}", key, event.getPartitions()); - if (!queue.isConsumerPerPartition()) { - mainConsumer.subscribe(event.getPartitions()); - } else { - log.info("[{}] Subscribing consumer per partition: {}", key, event.getPartitions()); - subscribeConsumerPerPartition(event.getQueueKey(), event.getPartitions()); - } - } - - void subscribeConsumerPerPartition(QueueKey queue, Set partitions) { - topicsConsumerPerPartition.get(queue).getSubscribeQueue().add(partitions); - scheduleTopicRepartition(queue); - } - - private void scheduleTopicRepartition(QueueKey queue) { - repartitionExecutor.schedule(() -> repartitionTopicWithConsumerPerPartition(queue), 1, TimeUnit.SECONDS); - } - - void repartitionTopicWithConsumerPerPartition(final QueueKey queueKey) { - if (stopped) { - return; - } - TbTopicWithConsumerPerPartition tbTopicWithConsumerPerPartition = topicsConsumerPerPartition.get(queueKey); - java.util.Queue> subscribeQueue = tbTopicWithConsumerPerPartition.getSubscribeQueue(); - if (subscribeQueue.isEmpty()) { - return; - } - if (tbTopicWithConsumerPerPartition.getLock().tryLock()) { - try { - Set partitions = null; - while (!subscribeQueue.isEmpty()) { - partitions = subscribeQueue.poll(); - } - if (partitions == null) { - return; - } - - Set addedPartitions = new HashSet<>(partitions); - ConcurrentMap>> consumers = tbTopicWithConsumerPerPartition.getConsumers(); - addedPartitions.removeAll(consumers.keySet()); - log.info("calculated addedPartitions {}", addedPartitions); - - Set removedPartitions = new HashSet<>(consumers.keySet()); - removedPartitions.removeAll(partitions); - log.info("calculated removedPartitions {}", removedPartitions); - - removedPartitions.forEach((tpi) -> { - removeConsumerForTopicByTpi(queueKey.getQueueName(), consumers, tpi); - }); - - addedPartitions.forEach((tpi) -> { - log.info("[{}] Adding consumer for topic: {}", queueKey, tpi); - Queue configuration = consumerConfigurations.get(queueKey); - TbQueueConsumer> consumer = tbRuleEngineQueueFactory.createToRuleEngineMsgConsumer(configuration); - consumers.put(tpi, consumer); - launchConsumer(consumer, queueKey, tpi.getFullTopicName(), queueKey + "-" + tpi.getPartition().orElse(-999999)); - consumer.subscribe(Collections.singleton(tpi)); - }); - } finally { - tbTopicWithConsumerPerPartition.getLock().unlock(); - } - } else { - scheduleTopicRepartition(queueKey); //reschedule later - } - } - - public void stop() { -// consumers.values().forEach(TbQueueConsumer::stop); -// topicsConsumerPerPartition.values().forEach(tbTopicWithConsumerPerPartition -> tbTopicWithConsumerPerPartition.getConsumers().keySet() -// .forEach((tpi) -> removeConsumerForTopicByTpi(tbTopicWithConsumerPerPartition.getTopic(), tbTopicWithConsumerPerPartition.getConsumers(), tpi))); - } - -} diff --git a/application/src/main/java/org/thingsboard/server/service/queue/ruleengine/QueueEvent.java b/application/src/main/java/org/thingsboard/server/service/queue/ruleengine/QueueEvent.java new file mode 100644 index 0000000000..4d01d15d9a --- /dev/null +++ b/application/src/main/java/org/thingsboard/server/service/queue/ruleengine/QueueEvent.java @@ -0,0 +1,9 @@ +package org.thingsboard.server.service.queue.ruleengine; + +import java.io.Serializable; + +public enum QueueEvent implements Serializable { + + CREATED, LAUNCHED, UPDATED, PARTITION_CHANGE, STOP, DELETED + +} diff --git a/application/src/main/java/org/thingsboard/server/service/queue/ruleengine/TbQueueConsumerManagerTask.java b/application/src/main/java/org/thingsboard/server/service/queue/ruleengine/TbQueueConsumerManagerTask.java new file mode 100644 index 0000000000..5b4f164a10 --- /dev/null +++ b/application/src/main/java/org/thingsboard/server/service/queue/ruleengine/TbQueueConsumerManagerTask.java @@ -0,0 +1,32 @@ +/** + * Copyright © 2016-2023 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.service.queue.ruleengine; + +import lombok.Data; +import org.thingsboard.server.common.data.plugin.ComponentLifecycleEvent; +import org.thingsboard.server.common.data.queue.Queue; +import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; + +import java.util.Set; + +@Data +public class TbQueueConsumerManagerTask { + + private final QueueEvent event; + private final Queue queue; + private final Set partitions; + +} diff --git a/application/src/main/java/org/thingsboard/server/service/queue/ruleengine/TbQueueConsumerTask.java b/application/src/main/java/org/thingsboard/server/service/queue/ruleengine/TbQueueConsumerTask.java new file mode 100644 index 0000000000..3f549b273c --- /dev/null +++ b/application/src/main/java/org/thingsboard/server/service/queue/ruleengine/TbQueueConsumerTask.java @@ -0,0 +1,70 @@ +/** + * Copyright © 2016-2023 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.service.queue.ruleengine; + +import lombok.Data; +import lombok.extern.slf4j.Slf4j; +import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; +import org.thingsboard.server.gen.transport.TransportProtos; +import org.thingsboard.server.queue.TbQueueConsumer; +import org.thingsboard.server.queue.common.TbProtoQueueMsg; +import org.thingsboard.server.queue.discovery.QueueKey; + +import java.util.Set; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; + +@Data +@Slf4j +public class TbQueueConsumerTask { + + private final QueueKey key; + private final Object id; + private final TbQueueConsumer> consumer; + private volatile Future task; + + public void stop() { + this.consumer.stop(); + } + + public boolean stopAndAwait() { + this.consumer.stop(); + return await(); + } + + public boolean await() { + if (task != null) { + try { + this.task.get(3, TimeUnit.MINUTES); + } catch (ExecutionException | InterruptedException | TimeoutException e) { + log.warn("[{}][{}] Failed to await for consumer to stop", key, id, e); + return false; + } + } + return true; + } + + public void subscribe(Set partitions) { + this.consumer.subscribe(partitions); + } + + + public void unsubscribe() { + this.consumer.unsubscribe(); + } +} diff --git a/application/src/main/java/org/thingsboard/server/service/queue/ruleengine/TbRuleEngineConsumerContext.java b/application/src/main/java/org/thingsboard/server/service/queue/ruleengine/TbRuleEngineConsumerContext.java new file mode 100644 index 0000000000..49f1ad1fab --- /dev/null +++ b/application/src/main/java/org/thingsboard/server/service/queue/ruleengine/TbRuleEngineConsumerContext.java @@ -0,0 +1,75 @@ +package org.thingsboard.server.service.queue.ruleengine; + +import lombok.Data; +import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.stereotype.Component; +import org.thingsboard.common.util.ThingsBoardThreadFactory; +import org.thingsboard.server.actors.ActorSystemContext; +import org.thingsboard.server.common.data.queue.Queue; +import org.thingsboard.server.common.stats.StatsFactory; +import org.thingsboard.server.dao.queue.QueueService; +import org.thingsboard.server.queue.TbQueueAdmin; +import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; +import org.thingsboard.server.queue.provider.TbQueueProducerProvider; +import org.thingsboard.server.queue.provider.TbRuleEngineQueueFactory; +import org.thingsboard.server.queue.util.TbRuleEngineComponent; +import org.thingsboard.server.service.queue.processing.TbRuleEngineProcessingStrategyFactory; +import org.thingsboard.server.service.queue.processing.TbRuleEngineSubmitStrategyFactory; +import org.thingsboard.server.service.rpc.TbRuleEngineDeviceRpcService; +import org.thingsboard.server.service.stats.RuleEngineStatisticsService; + +import javax.annotation.PreDestroy; +import java.util.List; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; + +@Component +@TbRuleEngineComponent +@Slf4j +@Data +public class TbRuleEngineConsumerContext { + + @Value("${queue.rule-engine.poll-interval}") + private long pollDuration; + @Value("${queue.rule-engine.pack-processing-timeout}") + private long packProcessingTimeout; + @Value("${queue.rule-engine.stats.enabled:true}") + private boolean statsEnabled; + @Value("${queue.rule-engine.prometheus-stats.enabled:false}") + boolean prometheusStatsEnabled; + @Value("${queue.rule-engine.topic-deletion-delay:30}") + private int topicDeletionDelayInSec; + + protected volatile boolean stopped = false; + protected volatile boolean isReady = false; + + private final ActorSystemContext actorContext; + private final StatsFactory statsFactory; + private final TbRuleEngineSubmitStrategyFactory submitStrategyFactory; + private final TbRuleEngineProcessingStrategyFactory processingStrategyFactory; + private final TbRuleEngineQueueFactory queueFactory; + private final RuleEngineStatisticsService statisticsService; + private final TbServiceInfoProvider serviceInfoProvider; + private final QueueService queueService; + private final TbQueueProducerProvider producerProvider; + private final TbQueueAdmin queueAdmin; + + //TODO: add reasonable limit for mgmt pool. + private final ExecutorService mgmtExecutor = Executors.newCachedThreadPool(ThingsBoardThreadFactory.forName("tb-rule-engine-mgmt")); + private final ExecutorService consumersExecutor = Executors.newCachedThreadPool(ThingsBoardThreadFactory.forName("tb-rule-engine-consumer")); + private final ExecutorService submitExecutor = Executors.newSingleThreadExecutor(ThingsBoardThreadFactory.forName("tb-rule-engine-consumer-submit")); + private final ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor(ThingsBoardThreadFactory.forName("tb-rule-engine-consumer-scheduler")); + + public List findAllQueues() { + return queueService.findAllQueues(); + } + + @PreDestroy + public void stop() { + consumersExecutor.shutdownNow(); // TODO: shutdown or shutdownNow? + submitExecutor.shutdownNow(); + scheduler.shutdownNow(); + } +} diff --git a/application/src/main/java/org/thingsboard/server/service/queue/ruleengine/TbRuleEngineQueueConsumerManager.java b/application/src/main/java/org/thingsboard/server/service/queue/ruleengine/TbRuleEngineQueueConsumerManager.java new file mode 100644 index 0000000000..2599a214f0 --- /dev/null +++ b/application/src/main/java/org/thingsboard/server/service/queue/ruleengine/TbRuleEngineQueueConsumerManager.java @@ -0,0 +1,431 @@ +/** + * Copyright © 2016-2023 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.service.queue.ruleengine; + +import com.google.protobuf.ProtocolStringList; +import lombok.Data; +import lombok.SneakyThrows; +import lombok.extern.slf4j.Slf4j; +import org.thingsboard.common.util.ThingsBoardThreadFactory; +import org.thingsboard.server.common.data.id.EntityId; +import org.thingsboard.server.common.data.id.EntityIdFactory; +import org.thingsboard.server.common.data.id.TenantId; +import org.thingsboard.server.common.data.queue.Queue; +import org.thingsboard.server.common.msg.TbMsg; +import org.thingsboard.server.common.msg.gen.MsgProtos; +import org.thingsboard.server.common.msg.queue.QueueToRuleEngineMsg; +import org.thingsboard.server.common.msg.queue.RuleEngineException; +import org.thingsboard.server.common.msg.queue.RuleNodeInfo; +import org.thingsboard.server.common.msg.queue.ServiceType; +import org.thingsboard.server.common.msg.queue.TbMsgCallback; +import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; +import org.thingsboard.server.gen.transport.TransportProtos; +import org.thingsboard.server.queue.TbQueueConsumer; +import org.thingsboard.server.queue.common.TbProtoQueueMsg; +import org.thingsboard.server.queue.discovery.QueueKey; +import org.thingsboard.server.queue.discovery.event.PartitionChangeEvent; +import org.thingsboard.server.service.queue.TbMsgPackCallback; +import org.thingsboard.server.service.queue.TbMsgPackProcessingContext; +import org.thingsboard.server.service.queue.TbRuleEngineConsumerStats; +import org.thingsboard.server.service.queue.processing.TbRuleEngineProcessingDecision; +import org.thingsboard.server.service.queue.processing.TbRuleEngineProcessingResult; +import org.thingsboard.server.service.queue.processing.TbRuleEngineProcessingStrategy; +import org.thingsboard.server.service.queue.processing.TbRuleEngineSubmitStrategy; + +import java.util.Collections; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.UUID; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.locks.ReentrantLock; + +@Data +@Slf4j +public class TbRuleEngineQueueConsumerManager { + + public static final String SUCCESSFUL_STATUS = "successful"; + public static final String FAILED_STATUS = "failed"; + + private final TbRuleEngineConsumerContext ctx; + private final QueueKey key; + + private final ReentrantLock lock = new ReentrantLock(); //NonfairSync + private final ConcurrentMap consumersPerPartition = new ConcurrentHashMap<>(); + private final TbRuleEngineConsumerStats stats; + + private volatile Set partitions = Collections.emptySet(); + private volatile Queue queue; + private volatile TbQueueConsumerTask mainConsumer; + + private final java.util.Queue tasks = new ConcurrentLinkedQueue<>(); + + public TbRuleEngineQueueConsumerManager(TbRuleEngineConsumerContext ctx, QueueKey key) { + this.ctx = ctx; + this.key = key; + this.stats = new TbRuleEngineConsumerStats(key, ctx.getStatsFactory()); + } + + public void init(Queue queue) { + processTask(new TbQueueConsumerManagerTask(QueueEvent.CREATED, queue, null)); + } + + public void update(Queue queue) { + processTask(new TbQueueConsumerManagerTask(QueueEvent.UPDATED, queue, null)); + } + + public void subscribe(PartitionChangeEvent event) { + processTask(new TbQueueConsumerManagerTask(QueueEvent.PARTITION_CHANGE, queue, event.getPartitions())); + } + + public void launchMainConsumer() { + processTask(new TbQueueConsumerManagerTask(QueueEvent.LAUNCHED, null, null)); + } + + public void stop() { + processTask(new TbQueueConsumerManagerTask(QueueEvent.STOP, null, null)); + } + + public void delete() { + processTask(new TbQueueConsumerManagerTask(QueueEvent.DELETED, null, null)); + } + + private void processTask(TbQueueConsumerManagerTask todo) { + tasks.add(todo); + log.info("[{}] Adding task: {}", key, todo); + tryProcessTasks(); + } + + private void tryProcessTasks() { + ctx.getMgmtExecutor().submit(() -> { + if (lock.tryLock()) { + try { + Queue newConfiguration = null; + Set newPartitions = null; + while (!tasks.isEmpty()) { + TbQueueConsumerManagerTask task = tasks.poll(); + switch (task.getEvent()) { + case CREATED: + doInit(task.getQueue()); + break; + case LAUNCHED: + if (!queue.isConsumerPerPartition()) { + doLaunchMainConsumer(); + } + break; + case UPDATED: + newConfiguration = task.getQueue(); + break; + case PARTITION_CHANGE: + newPartitions = task.getPartitions(); + break; + case STOP: + newConfiguration = null; + newPartitions = null; + doStop(); + break; + case DELETED: + newConfiguration = null; + newPartitions = null; + doDelete(); + break; + } + } + if (newConfiguration != null) { + doUpdate(newConfiguration); + } + if (newPartitions != null) { + doUpdate(newPartitions); + } + } finally { + lock.unlock(); + } + } else { + log.debug("[{}] Failed to acquire lock.", key); + ctx.getScheduler().schedule(this::tryProcessTasks, 1, TimeUnit.SECONDS); + } + }); + } + + public void doInit(Queue queue) { + log.info("[{}] Init consumer with queue: {}", key, queue); + this.queue = queue; + if (queue.isConsumerPerPartition()) { + log.debug("[{}] Ignore init event since isConsumerPerPartition is enabled.", key); + } else { + mainConsumer = new TbQueueConsumerTask(key, "main", ctx.getQueueFactory().createToRuleEngineMsgConsumer(queue)); + } + } + + private void doLaunchMainConsumer() { + if (mainConsumer != null) { + launchConsumer(mainConsumer, queue, mainConsumer.getId(), queue.getName()); + } else { + log.warn("[{}] Can't launch main consumer since it is empty!", key); + } + } + + private void doUpdate(Queue newQueue) { + log.info("[{}] Processing queue update: {}", key, newQueue); + var oldQueue = queue; + if (log.isTraceEnabled()) { + log.trace("[{}] Old queue configuration: {}", key, oldQueue); + log.trace("[{}] New queue configuration: {}", key, newQueue); + } + if (oldQueue != null) { + doStop(oldQueue); + } + doInit(newQueue); + if (!newQueue.isConsumerPerPartition()) { + doLaunchMainConsumer(); + } + } + + private void doUpdate(Set partitions) { + log.info("[{}] Subscribing to partitions: {}", key, partitions); + if (queue.isConsumerPerPartition()) { + log.debug("[{}] Subscribing consumers per partition separately: {}", key, partitions); + Set addedPartitions = new HashSet<>(partitions); + addedPartitions.removeAll(consumersPerPartition.keySet()); + log.info("calculated addedPartitions {}", addedPartitions); + + Set removedPartitions = new HashSet<>(consumersPerPartition.keySet()); + removedPartitions.removeAll(partitions); + log.info("calculated removedPartitions {}", removedPartitions); + + removedPartitions.forEach((tpi) -> { + log.info("[{}] Unsubscribing from topic: {}", queue, tpi); + consumersPerPartition.get(tpi).unsubscribe(); + }); + + removedPartitions.forEach((tpi) -> { + log.info("[{}] Removing consumer for topic: {}", queue, tpi); + consumersPerPartition.get(tpi).stopAndAwait(); + consumersPerPartition.remove(tpi); + }); + + addedPartitions.forEach((tpi) -> { + log.info("[{}] Adding consumer for topic: {}", key, tpi); + TbQueueConsumerTask consumerTask = new TbQueueConsumerTask(key, tpi, ctx.getQueueFactory().createToRuleEngineMsgConsumer(queue)); + consumersPerPartition.put(tpi, consumerTask); + //TODO: Is it ok to subscribe first? + consumerTask.subscribe(Collections.singleton(tpi)); + launchConsumer(consumerTask, queue, mainConsumer.getId(), key + "-" + tpi.getPartition().orElse(-999999)); + }); + } else { + mainConsumer.subscribe(partitions); + } + } + + private void doStop() { + doStop(queue); + } + + private void doStop(Queue queue) { + if (queue.isConsumerPerPartition()) { + consumersPerPartition.values().forEach(TbQueueConsumerTask::unsubscribe); + consumersPerPartition.values().forEach(TbQueueConsumerTask::stopAndAwait); + } else if (mainConsumer != null) { + mainConsumer.unsubscribe(); + mainConsumer.stopAndAwait(); + } + } + + private void doDelete() { + doStop(); + //TODO: repack messages + } + + @SneakyThrows + void launchConsumer(TbQueueConsumerTask consumerTask, Queue configuration, Object consumerKey, String threadSuffix) { + log.info("[{}] Launching consumer: [{}]", key, consumerKey); + while (!ctx.isReady) { + //TODO: Remember this task. Cancel previous task if needed. + log.debug("[{}][{}] Waiting for consumer to get ready..", key, consumerKey); + Thread.sleep(1000); + } + consumerTask.setTask(ctx.getConsumersExecutor().submit(() -> consumerLoop(consumerTask.getConsumer(), configuration, threadSuffix))); + } + + void consumerLoop(TbQueueConsumer> consumer, Queue configuration, String threadSuffix) { + ThingsBoardThreadFactory.updateCurrentThreadName(threadSuffix); + while (!ctx.stopped && !consumer.isStopped() + //TODO: remove this. + && !consumer.isQueueDeleted()) { + try { + List> msgs = consumer.poll(queue.getPollInterval()); + if (msgs.isEmpty()) { + continue; + } + final TbRuleEngineSubmitStrategy submitStrategy = getSubmitStrategy(queue); + final TbRuleEngineProcessingStrategy ackStrategy = getProcessingStrategy(queue); + submitStrategy.init(msgs); + while (!ctx.isStopped() && !consumer.isStopped()) { + TbMsgPackProcessingContext packCtx = new TbMsgPackProcessingContext(queue.getName(), submitStrategy, ackStrategy.isSkipTimeoutMsgs()); + submitStrategy.submitAttempt((id, msg) -> ctx.getSubmitExecutor().submit(() -> submitMessage(configuration, stats, packCtx, id, msg))); + + final boolean timeout = !packCtx.await(configuration.getPackProcessingTimeout(), TimeUnit.MILLISECONDS); + + TbRuleEngineProcessingResult result = new TbRuleEngineProcessingResult(configuration.getName(), timeout, packCtx); + if (timeout) { + printFirstOrAll(configuration, packCtx, packCtx.getPendingMap(), "Timeout"); + } + if (!packCtx.getFailedMap().isEmpty()) { + printFirstOrAll(configuration, packCtx, packCtx.getFailedMap(), "Failed"); + } + packCtx.printProfilerStats(); + + TbRuleEngineProcessingDecision decision = ackStrategy.analyze(result); + if (ctx.isStatsEnabled()) { + stats.log(result, decision.isCommit()); + } + + packCtx.cleanup(); + + if (decision.isCommit()) { + submitStrategy.stop(); + break; + } else { + submitStrategy.update(decision.getReprocessMap()); + } + } + consumer.commit(); + } catch (Exception e) { + if (!ctx.stopped) { + log.warn("Failed to process messages from queue.", e); + try { + Thread.sleep(ctx.getPollDuration()); + } catch (InterruptedException e2) { + log.trace("Failed to wait until the server has capacity to handle new requests", e2); + } + } + } + } + //TODO: refactor and move to the "doDelete" method. Use separate consumer if needed (it is still synchronous). + if (consumer.isQueueDeleted()) { + processQueueDeletion(configuration, consumer); + } + log.info("TB Rule Engine Consumer stopped."); + } + + private void processQueueDeletion(Queue queue, TbQueueConsumer> consumer) { +// long finishTs = System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(topicDeletionDelayInSec); +// try { +// int n = 0; +// while (System.currentTimeMillis() <= finishTs) { +// List> msgs = consumer.poll(queue.getPollInterval()); +// if (msgs.isEmpty()) { +// continue; +// } +// for (TbProtoQueueMsg msg : msgs) { +// try { +// MsgProtos.TbMsgProto tbMsgProto = MsgProtos.TbMsgProto.parseFrom(msg.getValue().getTbMsg().toByteArray()); +// EntityId originator = EntityIdFactory.getByTypeAndUuid(tbMsgProto.getEntityType(), new UUID(tbMsgProto.getEntityIdMSB(), tbMsgProto.getEntityIdLSB())); +// +// TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_RULE_ENGINE, queue.getName(), TenantId.SYS_TENANT_ID, originator); +// producerProvider.getRuleEngineMsgProducer().send(tpi, msg, null); +// n++; +// } catch (Throwable e) { +// log.debug("Failed to move message to system {}: {}", consumer.getTopic(), msg, e); +// } +// } +// consumer.commit(); +// } +// if (n > 0) { +// log.info("Moved {} messages from {} to system {}", n, consumer.getFullTopicNames(), consumer.getTopic()); +// } +// +// consumer.unsubscribe(); +// for (String topic : consumer.getFullTopicNames()) { +// try { +// queueAdmin.deleteTopic(topic); +// log.info("Deleted topic {}", topic); +// } catch (Exception e) { +// log.error("Failed to delete topic {} after unsubscribing", topic, e); +// } +// } +// } catch (Exception e) { +// log.error("Failed to process deletion of {} ({})", consumer.getTopic(), queue.getTenantId(), e); +// } + } + + TbRuleEngineSubmitStrategy getSubmitStrategy(Queue configuration) { + return ctx.getSubmitStrategyFactory().newInstance(configuration.getName(), configuration.getSubmitStrategy()); + } + + TbRuleEngineProcessingStrategy getProcessingStrategy(Queue configuration) { + return ctx.getProcessingStrategyFactory().newInstance(configuration.getName(), configuration.getProcessingStrategy()); + } + + void submitMessage(Queue configuration, TbRuleEngineConsumerStats stats, TbMsgPackProcessingContext packCtx, UUID id, TbProtoQueueMsg msg) { + log.trace("[{}] Creating callback for topic {} message: {}", id, configuration.getName(), msg.getValue()); + TransportProtos.ToRuleEngineMsg toRuleEngineMsg = msg.getValue(); + TenantId tenantId = TenantId.fromUUID(new UUID(toRuleEngineMsg.getTenantIdMSB(), toRuleEngineMsg.getTenantIdLSB())); + TbMsgCallback callback = ctx.prometheusStatsEnabled ? + new TbMsgPackCallback(id, tenantId, packCtx, stats.getTimer(tenantId, SUCCESSFUL_STATUS), stats.getTimer(tenantId, FAILED_STATUS)) : + new TbMsgPackCallback(id, tenantId, packCtx); + try { + if (!toRuleEngineMsg.getTbMsg().isEmpty()) { + forwardToRuleEngineActor(configuration.getName(), tenantId, toRuleEngineMsg, callback); + } else { + callback.onSuccess(); + } + } catch (Exception e) { + callback.onFailure(new RuleEngineException(e.getMessage(), e)); + } + } + + private void forwardToRuleEngineActor(String queueName, TenantId tenantId, TransportProtos.ToRuleEngineMsg toRuleEngineMsg, TbMsgCallback callback) { + TbMsg tbMsg = TbMsg.fromBytes(queueName, toRuleEngineMsg.getTbMsg().toByteArray(), callback); + QueueToRuleEngineMsg msg; + ProtocolStringList relationTypesList = toRuleEngineMsg.getRelationTypesList(); + Set relationTypes; + if (relationTypesList.size() == 1) { + relationTypes = Collections.singleton(relationTypesList.get(0)); + } else { + relationTypes = new HashSet<>(relationTypesList); + } + msg = new QueueToRuleEngineMsg(tenantId, tbMsg, relationTypes, toRuleEngineMsg.getFailureMessage()); + ctx.getActorContext().tell(msg); + } + + + private void printFirstOrAll(Queue configuration, TbMsgPackProcessingContext ctx, Map> map, String prefix) { + boolean printAll = log.isTraceEnabled(); + log.info("{} to process [{}] messages", prefix, map.size()); + for (Map.Entry> pending : map.entrySet()) { + TransportProtos.ToRuleEngineMsg tmp = pending.getValue().getValue(); + TbMsg tmpMsg = TbMsg.fromBytes(configuration.getName(), tmp.getTbMsg().toByteArray(), TbMsgCallback.EMPTY); + RuleNodeInfo ruleNodeInfo = ctx.getLastVisitedRuleNode(pending.getKey()); + if (printAll) { + log.trace("[{}] {} to process message: {}, Last Rule Node: {}", TenantId.fromUUID(new UUID(tmp.getTenantIdMSB(), tmp.getTenantIdLSB())), prefix, tmpMsg, ruleNodeInfo); + } else { + log.info("[{}] {} to process message: {}, Last Rule Node: {}", TenantId.fromUUID(new UUID(tmp.getTenantIdMSB(), tmp.getTenantIdLSB())), prefix, tmpMsg, ruleNodeInfo); + break; + } + } + } + + public void printStats(long ts) { + stats.printStats(); + ctx.getStatisticsService().reportQueueStats(ts, stats); + stats.reset(); + } +} diff --git a/common/util/src/main/java/org/thingsboard/common/util/ThingsBoardThreadFactory.java b/common/util/src/main/java/org/thingsboard/common/util/ThingsBoardThreadFactory.java index 1f41c48f43..9eaca3ca47 100644 --- a/common/util/src/main/java/org/thingsboard/common/util/ThingsBoardThreadFactory.java +++ b/common/util/src/main/java/org/thingsboard/common/util/ThingsBoardThreadFactory.java @@ -22,6 +22,7 @@ import java.util.concurrent.atomic.AtomicInteger; * Copy of Executors.DefaultThreadFactory but with ability to set name of the pool */ public class ThingsBoardThreadFactory implements ThreadFactory { + public static final String THREAD_TOPIC_SEPARATOR = " | "; private static final AtomicInteger poolNumber = new AtomicInteger(1); private final ThreadGroup group; private final AtomicInteger threadNumber = new AtomicInteger(1); @@ -40,6 +41,17 @@ public class ThingsBoardThreadFactory implements ThreadFactory { "-thread-"; } + public static void updateCurrentThreadName(String threadSuffix) { + String name = Thread.currentThread().getName(); + int spliteratorIndex = name.indexOf(THREAD_TOPIC_SEPARATOR); + if (spliteratorIndex > 0) { + name = name.substring(0, spliteratorIndex); + } + name = name + THREAD_TOPIC_SEPARATOR + threadSuffix; + Thread.currentThread().setName(name); + } + + @Override public Thread newThread(Runnable r) { Thread t = new Thread(group, r,