Log time to acknowledge message

6 years ago · b957a0ce8b
6 changed files with 83 additions and 33 deletions
--- 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<ToRuleEngineNotificationMsg> 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);
--- 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);
    }

--- 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<StatsCounter> counters = new ArrayList<>();
    private final ConcurrentMap<UUID, TbTenantRuleEngineStats> tenantStats = new ConcurrentHashMap<>();
+    private final ConcurrentMap<TenantId, Timer> tenantMsgProcessTimers = new ConcurrentHashMap<>();
    private final ConcurrentMap<TenantId, RuleEngineException> 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();
--- 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:
--- 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() {
--- 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 extends Number> T createGauge(String key, T number, String... tags);

    MessagesStats createMessagesStats(String key);
+
+    Timer createTimer(String key, String... tags);
 }