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 a66d1e090c..0f3efb6f05 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 @@ -22,17 +22,11 @@ import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Service; import org.thingsboard.rule.engine.api.RpcError; import org.thingsboard.server.actors.ActorSystemContext; -import org.thingsboard.server.common.data.id.RuleNodeId; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.msg.TbActorMsg; import org.thingsboard.server.common.msg.TbMsg; -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.ServiceQueue; -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.*; +import org.thingsboard.server.common.stats.StatsFactory; import org.thingsboard.server.gen.transport.TransportProtos; import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineNotificationMsg; @@ -42,40 +36,25 @@ import org.thingsboard.server.queue.discovery.PartitionChangeEvent; import org.thingsboard.server.queue.provider.TbRuleEngineQueueFactory; import org.thingsboard.server.queue.settings.TbQueueRuleEngineSettings; import org.thingsboard.server.queue.settings.TbRuleEngineQueueConfiguration; -import org.thingsboard.server.common.stats.StatsFactory; import org.thingsboard.server.queue.util.TbRuleEngineComponent; import org.thingsboard.server.service.encoding.DataDecodingEncodingService; -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.processing.*; import org.thingsboard.server.service.rpc.FromDeviceRpcResponse; import org.thingsboard.server.service.rpc.TbRuleEngineDeviceRpcService; import org.thingsboard.server.service.stats.RuleEngineStatisticsService; import javax.annotation.PostConstruct; import javax.annotation.PreDestroy; -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.TimeUnit; +import java.util.*; +import java.util.concurrent.*; @Service @TbRuleEngineComponent @Slf4j public class DefaultTbRuleEngineConsumerService extends AbstractConsumerService implements TbRuleEngineConsumerService { + public static final String SUCCESSFUL_STATUS = "successful"; + public static final String FAILED_STATUS = "failed"; @Value("${queue.rule-engine.poll-interval}") private long pollDuration; @Value("${queue.rule-engine.pack-processing-timeout}") @@ -170,7 +149,9 @@ public class DefaultTbRuleEngineConsumerService extends AbstractConsumerService< log.trace("[{}] Creating callback for message: {}", id, msg.getValue()); ToRuleEngineMsg toRuleEngineMsg = msg.getValue(); TenantId tenantId = new TenantId(new UUID(toRuleEngineMsg.getTenantIdMSB(), toRuleEngineMsg.getTenantIdLSB())); - TbMsgCallback callback = new TbMsgPackCallback(id, tenantId, ctx); + TbMsgCallback callback = statsEnabled ? + 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); diff --git a/application/src/main/java/org/thingsboard/server/service/queue/TbMsgPackCallback.java b/application/src/main/java/org/thingsboard/server/service/queue/TbMsgPackCallback.java index c7925fd759..4d5be5b883 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/TbMsgPackCallback.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/TbMsgPackCallback.java @@ -15,6 +15,7 @@ */ package org.thingsboard.server.service.queue; +import io.micrometer.core.instrument.Timer; import lombok.extern.slf4j.Slf4j; import org.thingsboard.server.common.data.id.RuleNodeId; import org.thingsboard.server.common.data.id.TenantId; @@ -23,28 +24,45 @@ import org.thingsboard.server.common.msg.queue.RuleNodeInfo; import org.thingsboard.server.common.msg.queue.TbMsgCallback; import java.util.UUID; +import java.util.concurrent.TimeUnit; @Slf4j public class TbMsgPackCallback implements TbMsgCallback { private final UUID id; private final TenantId tenantId; private final TbMsgPackProcessingContext ctx; + private final long startMsgProcessing; + private final Timer successfulMsgTimer; + private final Timer failedMsgTimer; public TbMsgPackCallback(UUID id, TenantId tenantId, TbMsgPackProcessingContext ctx) { + this(id, tenantId, ctx, null, null); + } + + public TbMsgPackCallback(UUID id, TenantId tenantId, TbMsgPackProcessingContext ctx, Timer successfulMsgTimer, Timer failedMsgTimer) { this.id = id; this.tenantId = tenantId; this.ctx = ctx; + this.successfulMsgTimer = successfulMsgTimer; + this.failedMsgTimer = failedMsgTimer; + startMsgProcessing = System.currentTimeMillis(); } @Override public void onSuccess() { log.trace("[{}] ON SUCCESS", id); + if (successfulMsgTimer != null) { + successfulMsgTimer.record(System.currentTimeMillis() - startMsgProcessing, TimeUnit.MILLISECONDS); + } ctx.onSuccess(id); } @Override public void onFailure(RuleEngineException e) { log.trace("[{}] ON FAILURE", id, e); + if (failedMsgTimer != null) { + failedMsgTimer.record(System.currentTimeMillis() - startMsgProcessing, TimeUnit.MILLISECONDS); + } ctx.onFailure(tenantId, id, e); } diff --git a/application/src/main/java/org/thingsboard/server/service/queue/TbRuleEngineConsumerStats.java b/application/src/main/java/org/thingsboard/server/service/queue/TbRuleEngineConsumerStats.java index 41004b41f5..011e3c0d59 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/TbRuleEngineConsumerStats.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/TbRuleEngineConsumerStats.java @@ -15,6 +15,8 @@ */ package org.thingsboard.server.service.queue; +import io.micrometer.core.instrument.DistributionSummary; +import io.micrometer.core.instrument.Timer; import lombok.extern.slf4j.Slf4j; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.msg.queue.RuleEngineException; @@ -41,6 +43,8 @@ public class TbRuleEngineConsumerStats { public static final String SUCCESSFUL_ITERATIONS = "successfulIterations"; public static final String FAILED_ITERATIONS = "failedIterations"; + private final StatsFactory statsFactory; + private final StatsCounter totalMsgCounter; private final StatsCounter successMsgCounter; private final StatsCounter tmpTimeoutMsgCounter; @@ -54,12 +58,14 @@ public class TbRuleEngineConsumerStats { private final List counters = new ArrayList<>(); private final ConcurrentMap tenantStats = new ConcurrentHashMap<>(); + private final ConcurrentMap tenantMsgProcessTimers = new ConcurrentHashMap<>(); private final ConcurrentMap tenantExceptions = new ConcurrentHashMap<>(); private final String queueName; public TbRuleEngineConsumerStats(String queueName, StatsFactory statsFactory) { this.queueName = queueName; + this.statsFactory = statsFactory; String statsKey = StatsType.RULE_ENGINE.getName() + "." + queueName; this.totalMsgCounter = statsFactory.createStatsCounter(statsKey, TOTAL_MSGS); @@ -82,6 +88,14 @@ public class TbRuleEngineConsumerStats { counters.add(failedIterationsCounter); } + public Timer getTimer(TenantId tenantId, String status){ + return tenantMsgProcessTimers.computeIfAbsent(tenantId, + id -> statsFactory.createTimer(StatsType.RULE_ENGINE.getName() + "." + queueName, + "tenantId", tenantId.getId().toString(), + "status", status + )); + } + public void log(TbRuleEngineProcessingResult msg, boolean finalIterationForPack) { int success = msg.getSuccessMap().size(); int pending = msg.getPendingMap().size(); diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml index 54a3dac870..a98305543d 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/application/src/main/resources/thingsboard.yml @@ -785,6 +785,10 @@ service: metrics: # Enable/disable actuator metrics. enabled: "${METRICS_ENABLED:false}" + timer: + # Metrics percentiles returned by actuator for timer metrics. List of double values (divided by ,). + percentiles: "${METRICS_TIMER_PERCENTILES:0.5}" + management: endpoints: diff --git a/common/stats/src/main/java/org/thingsboard/server/common/stats/DefaultStatsFactory.java b/common/stats/src/main/java/org/thingsboard/server/common/stats/DefaultStatsFactory.java index 8ee75c859e..acd66ff2f3 100644 --- a/common/stats/src/main/java/org/thingsboard/server/common/stats/DefaultStatsFactory.java +++ b/common/stats/src/main/java/org/thingsboard/server/common/stats/DefaultStatsFactory.java @@ -15,13 +15,13 @@ */ package org.thingsboard.server.common.stats; -import io.micrometer.core.instrument.Counter; -import io.micrometer.core.instrument.MeterRegistry; -import io.micrometer.core.instrument.Tags; +import io.micrometer.core.instrument.*; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Service; +import org.springframework.util.StringUtils; +import javax.annotation.PostConstruct; import java.util.concurrent.atomic.AtomicInteger; @Service @@ -40,6 +40,23 @@ public class DefaultStatsFactory implements StatsFactory { @Value("${metrics.enabled:false}") private Boolean metricsEnabled; + @Value("${metrics.timer.percentiles:0.5}") + private String timerPercentilesStr; + + private double[] timerPercentiles; + + @PostConstruct + public void init() { + if (!StringUtils.isEmpty(timerPercentilesStr)) { + String[] split = timerPercentilesStr.split(","); + timerPercentiles = new double[split.length]; + for (int i = 0; i < split.length; i++) { + timerPercentiles[i] = Double.parseDouble(split[i]); + } + } + } + + @Override public StatsCounter createStatsCounter(String key, String statsName) { return new StatsCounter( @@ -74,9 +91,21 @@ public class DefaultStatsFactory implements StatsFactory { return new DefaultMessagesStats(totalCounter, successfulCounter, failedCounter); } + @Override + public Timer createTimer(String key, String... tags) { + Timer.Builder timerBuilder = Timer.builder(key) + .tags(tags) + .publishPercentiles(); + if (timerPercentiles != null && timerPercentiles.length > 0) { + timerBuilder.publishPercentiles(timerPercentiles); + } + return timerBuilder.register(meterRegistry); + } + private static class StubCounter implements Counter { @Override - public void increment(double amount) {} + public void increment(double amount) { + } @Override public double count() { diff --git a/common/stats/src/main/java/org/thingsboard/server/common/stats/StatsFactory.java b/common/stats/src/main/java/org/thingsboard/server/common/stats/StatsFactory.java index b979d61f35..416e94980c 100644 --- a/common/stats/src/main/java/org/thingsboard/server/common/stats/StatsFactory.java +++ b/common/stats/src/main/java/org/thingsboard/server/common/stats/StatsFactory.java @@ -15,6 +15,8 @@ */ package org.thingsboard.server.common.stats; +import io.micrometer.core.instrument.Timer; + public interface StatsFactory { StatsCounter createStatsCounter(String key, String statsName); @@ -23,4 +25,6 @@ public interface StatsFactory { T createGauge(String key, T number, String... tags); MessagesStats createMessagesStats(String key); + + Timer createTimer(String key, String... tags); }