Processing time stats for Housekeeper

application/src/main/java/org/thingsboard/server/service/housekeeper/HousekeeperReprocessingService.java

@@ -77,6 +77,8 @@ public class HousekeeperReprocessingService {
     }
 
     private void processMsgs(List<TbProtoQueueMsg<ToHousekeeperServiceMsg>> msgs, TbQueueConsumer<TbProtoQueueMsg<ToHousekeeperServiceMsg>> consumer) throws Exception {
+        Thread.sleep(config.getTaskReprocessingDelay());
+
         for (TbProtoQueueMsg<ToHousekeeperServiceMsg> msg : msgs) {
             log.trace("Reprocessing task: {}", msg);
             try {
@@ -89,8 +91,6 @@ public class HousekeeperReprocessingService {
             }
         }
         consumer.commit();
-
-        Thread.sleep(config.getTaskReprocessingDelay());
     }
 
     public void submitForReprocessing(ToHousekeeperServiceMsg msg, Throwable error) {

application/src/main/java/org/thingsboard/server/service/housekeeper/HousekeeperService.java

@@ -130,7 +130,7 @@ public class HousekeeperService {
             if (log.isDebugEnabled()) {
                 log.debug("[{}] Processed {} in {} ms (attempt {})", task.getTenantId(), task.getDescription(), timing, msg.getTask().getAttempt());
             }
-            statsService.ifPresent(statsService -> statsService.reportProcessed(taskType, msg));
+            statsService.ifPresent(statsService -> statsService.reportProcessed(taskType, msg, timing));
         } catch (InterruptedException e) {
             throw e;
         } catch (Throwable e) {

application/src/main/java/org/thingsboard/server/service/housekeeper/stats/HousekeeperStatsService.java

@@ -24,6 +24,7 @@ import org.thingsboard.server.common.data.housekeeper.HousekeeperTaskType;
 import org.thingsboard.server.common.stats.DefaultCounter;
 import org.thingsboard.server.common.stats.StatsCounter;
 import org.thingsboard.server.common.stats.StatsFactory;
+import org.thingsboard.server.common.stats.StatsTimer;
 import org.thingsboard.server.common.stats.StatsType;
 import org.thingsboard.server.gen.transport.TransportProtos.ToHousekeeperServiceMsg;
 
@@ -53,14 +54,14 @@ public class HousekeeperStatsService {
         String statsStr = stats.values().stream().map(stats -> {
             String countersStr = stats.getCounters().stream()
                     .filter(counter -> counter.get() > 0)
-                    .map(counter -> counter.getName() + " = " + counter.get())
-                    .collect(Collectors.joining(", "));
+                    .map(counter -> counter.getName() + " = [" + counter.get() + "]")
+                    .collect(Collectors.joining(" "));
             if (countersStr.isEmpty()) {
                 return null;
             } else {
-                return stats.getTaskType() + " {" + countersStr + "}";
+                return stats.getTaskType() + " " + countersStr + " avgProcessingTime [" + stats.getProcessingTimer().getAvg() + " ms]";
             }
-        }).filter(Objects::nonNull).collect(Collectors.joining("; "));
+        }).filter(Objects::nonNull).collect(Collectors.joining(", "));
 
         if (!statsStr.isEmpty()) {
             stats.values().forEach(HousekeeperStats::reset);
@@ -68,14 +69,14 @@ public class HousekeeperStatsService {
         }
     }
 
-    public void reportProcessed(HousekeeperTaskType taskType, ToHousekeeperServiceMsg msg) {
-        // todo: report timings
+    public void reportProcessed(HousekeeperTaskType taskType, ToHousekeeperServiceMsg msg, long timing) {
         HousekeeperStats stats = this.stats.get(taskType);
         if (msg.getTask().getErrorsCount() == 0) {
             stats.getProcessedCounter().increment();
         } else {
             stats.getReprocessedCounter().increment();
         }
+        stats.getProcessingTimer().record(timing);
     }
 
     public void reportFailure(HousekeeperTaskType taskType, ToHousekeeperServiceMsg msg) {
@@ -97,12 +98,15 @@ public class HousekeeperStatsService {
         private final StatsCounter reprocessedCounter;
         private final StatsCounter failedReprocessingCounter;
 
+        private final StatsTimer processingTimer;
+
         public HousekeeperStats(HousekeeperTaskType taskType, StatsFactory statsFactory) {
             this.taskType = taskType;
             this.processedCounter = register("processed", statsFactory);
             this.failedProcessingCounter = register("failedProcessing", statsFactory);
             this.reprocessedCounter = register("reprocessed", statsFactory);
             this.failedReprocessingCounter = register("failedReprocessing", statsFactory);
+            this.processingTimer = statsFactory.createTimer(StatsType.HOUSEKEEPER, "processingTime", "taskType", taskType.name());
         }
 
         private StatsCounter register(String statsName, StatsFactory statsFactory) {
@@ -113,7 +117,9 @@ public class HousekeeperStatsService {
 
         public void reset() {
             counters.forEach(DefaultCounter::clear);
+            processingTimer.reset();
         }
+
     }
 
 }

common/stats/src/main/java/org/thingsboard/server/common/stats/DefaultStatsFactory.java

@@ -19,13 +19,13 @@ import io.micrometer.core.instrument.Counter;
 import io.micrometer.core.instrument.MeterRegistry;
 import io.micrometer.core.instrument.Tags;
 import io.micrometer.core.instrument.Timer;
+import jakarta.annotation.PostConstruct;
 import org.apache.commons.lang3.ArrayUtils;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Service;
 import org.thingsboard.server.common.data.StringUtils;
 
-import jakarta.annotation.PostConstruct;
 import java.util.concurrent.atomic.AtomicInteger;
 
 @Service
@@ -63,13 +63,7 @@ public class DefaultStatsFactory implements StatsFactory {
 
     @Override
     public StatsCounter createStatsCounter(String key, String statsName, String... otherTags) {
-        String[] tags = new String[]{STATS_NAME_TAG, statsName};
-        if (otherTags.length > 0) {
-            if (otherTags.length % 2 != 0) {
-                throw new IllegalArgumentException("Invalid tags array size");
-            }
-            tags = ArrayUtils.addAll(tags, otherTags);
-        }
+        String[] tags = getTags(statsName, otherTags);
         return new StatsCounter(
                 new AtomicInteger(0),
                 metricsEnabled ? meterRegistry.counter(key, tags) : STUB_COUNTER,
@@ -111,6 +105,24 @@ public class DefaultStatsFactory implements StatsFactory {
         return timerBuilder.register(meterRegistry);
     }
 
+    @Override
+    public StatsTimer createTimer(StatsType type, String name, String... tags) {
+        return new StatsTimer(name, Timer.builder(type.getName())
+                .tags(getTags(name, tags))
+                .register(meterRegistry));
+    }
+
+    private static String[] getTags(String statsName, String[] otherTags) {
+        String[] tags = new String[]{STATS_NAME_TAG, statsName};
+        if (otherTags.length > 0) {
+            if (otherTags.length % 2 != 0) {
+                throw new IllegalArgumentException("Invalid tags array size");
+            }
+            tags = ArrayUtils.addAll(tags, otherTags);
+        }
+        return tags;
+    }
+
     private static class StubCounter implements Counter {
         @Override
         public void increment(double amount) {

common/stats/src/main/java/org/thingsboard/server/common/stats/StatsFactory.java

@@ -29,4 +29,6 @@ public interface StatsFactory {
 
     Timer createTimer(String key, String... tags);
 
+    StatsTimer createTimer(StatsType type, String name, String... tags);
+
 }

common/stats/src/main/java/org/thingsboard/server/common/stats/StatsTimer.java

@@ -0,0 +1,55 @@
+/**
+ * Copyright © 2016-2024 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.common.stats;
+
+import io.micrometer.core.instrument.Timer;
+import lombok.Getter;
+
+import java.util.concurrent.TimeUnit;
+
+public class StatsTimer {
+
+    @Getter
+    private final String name;
+    private final Timer timer;
+
+    private int count;
+    private long totalTime;
+
+    public StatsTimer(String name, Timer micrometerTimer) {
+        this.name = name;
+        this.timer = micrometerTimer;
+    }
+
+    public  void record(long timeMs) {
+        count++;
+        totalTime += timeMs;
+        timer.record(timeMs, TimeUnit.MILLISECONDS);
+    }
+
+    public double getAvg() {
+        if (count == 0) {
+            return 0.0;
+        }
+        return (double) totalTime / count;
+    }
+
+    public  void reset() {
+        count = 0;
+        totalTime = 0;
+    }
+
+}