Merge pull request #3283 from vzikratyi-tb/feature/ack-time-metrics

Track time to acknowledge messages (in actuator/prometheus)

@@ -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);

@@ -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);
     }
 

@@ -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();

@@ -783,6 +783,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:

@@ -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() {

@@ -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);
 }