From 07d144fb080db742160b932332df74e9557ef32b Mon Sep 17 00:00:00 2001 From: ViacheslavKlimov Date: Wed, 17 Apr 2024 15:03:02 +0300 Subject: [PATCH] Processing time stats for Housekeeper --- .../HousekeeperReprocessingService.java | 4 +- .../housekeeper/HousekeeperService.java | 2 +- .../stats/HousekeeperStatsService.java | 18 ++++-- .../common/stats/DefaultStatsFactory.java | 28 +++++++--- .../server/common/stats/StatsFactory.java | 2 + .../server/common/stats/StatsTimer.java | 55 +++++++++++++++++++ 6 files changed, 92 insertions(+), 17 deletions(-) create mode 100644 common/stats/src/main/java/org/thingsboard/server/common/stats/StatsTimer.java diff --git a/application/src/main/java/org/thingsboard/server/service/housekeeper/HousekeeperReprocessingService.java b/application/src/main/java/org/thingsboard/server/service/housekeeper/HousekeeperReprocessingService.java index cdc0d3f880..ae16f925a0 100644 --- a/application/src/main/java/org/thingsboard/server/service/housekeeper/HousekeeperReprocessingService.java +++ b/application/src/main/java/org/thingsboard/server/service/housekeeper/HousekeeperReprocessingService.java @@ -77,6 +77,8 @@ public class HousekeeperReprocessingService { } private void processMsgs(List> msgs, TbQueueConsumer> consumer) throws Exception { + Thread.sleep(config.getTaskReprocessingDelay()); + for (TbProtoQueueMsg 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) { diff --git a/application/src/main/java/org/thingsboard/server/service/housekeeper/HousekeeperService.java b/application/src/main/java/org/thingsboard/server/service/housekeeper/HousekeeperService.java index 1f5ed1be58..898ae74d3f 100644 --- a/application/src/main/java/org/thingsboard/server/service/housekeeper/HousekeeperService.java +++ b/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) { diff --git a/application/src/main/java/org/thingsboard/server/service/housekeeper/stats/HousekeeperStatsService.java b/application/src/main/java/org/thingsboard/server/service/housekeeper/stats/HousekeeperStatsService.java index da586aa820..6723d51625 100644 --- a/application/src/main/java/org/thingsboard/server/service/housekeeper/stats/HousekeeperStatsService.java +++ b/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(); } + } } 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 be0c4fe19f..ca97792186 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 @@ -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) { 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 b9bf0b97fb..a6cc4aa1d4 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 @@ -29,4 +29,6 @@ public interface StatsFactory { Timer createTimer(String key, String... tags); + StatsTimer createTimer(StatsType type, String name, String... tags); + } diff --git a/common/stats/src/main/java/org/thingsboard/server/common/stats/StatsTimer.java b/common/stats/src/main/java/org/thingsboard/server/common/stats/StatsTimer.java new file mode 100644 index 0000000000..ca74444619 --- /dev/null +++ b/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; + } + +}