This is an automated email from the ASF dual-hosted git repository. spricoder pushed a commit to branch fix/flush_1.1 in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit b524ee04c00baf8b585cd5d20a879cd4ecb9ce2a Author: spricoder <[email protected]> AuthorDate: Thu Aug 24 13:54:17 2023 +0800 [Metric] Fix flush point statistics (#10915) (cherry picked from commit 1dce787c39e14973918ee390c470b901b68be73e) --- .../iotdb/metrics/AbstractMetricService.java | 45 +++++++++-------- .../metrics/reporter/iotdb/IoTDBReporter.java | 6 +-- .../iotdb/db/engine/flush/MemTableFlushTask.java | 58 +++++++++++++++------- 3 files changed, 69 insertions(+), 40 deletions(-) diff --git a/metrics/interface/src/main/java/org/apache/iotdb/metrics/AbstractMetricService.java b/metrics/interface/src/main/java/org/apache/iotdb/metrics/AbstractMetricService.java index 31e1e352ed1..cc9452dc1f5 100644 --- a/metrics/interface/src/main/java/org/apache/iotdb/metrics/AbstractMetricService.java +++ b/metrics/interface/src/main/java/org/apache/iotdb/metrics/AbstractMetricService.java @@ -271,9 +271,9 @@ public abstract class AbstractMetricService { /** GetOrCreateCounter with internal report. */ public Counter getOrCreateCounterWithInternalReport( - String metric, MetricLevel metricLevel, String... tags) { + String metric, MetricLevel metricLevel, long time, String... tags) { Counter counter = metricManager.getOrCreateCounter(metric, metricLevel, tags); - internalReporter.writeMetricToIoTDB(counter, metric, tags); + internalReporter.writeMetricToIoTDB(counter, metric, time, tags); return counter; } @@ -287,69 +287,74 @@ public abstract class AbstractMetricService { /** GetOrCreateGauge with internal report. */ public Gauge getOrCreateGaugeWithInternalReport( - String metric, MetricLevel metricLevel, String... tags) { + String metric, MetricLevel metricLevel, long time, String... tags) { Gauge gauge = metricManager.getOrCreateGauge(metric, metricLevel, tags); - internalReporter.writeMetricToIoTDB(gauge, metric, tags); + internalReporter.writeMetricToIoTDB(gauge, metric, time, tags); return gauge; } /** GetOrCreateRate with internal report. */ public Rate getOrCreateRateWithInternalReport( - String metric, MetricLevel metricLevel, String... tags) { + String metric, MetricLevel metricLevel, long time, String... tags) { Rate rate = metricManager.getOrCreateRate(metric, metricLevel, tags); - internalReporter.writeMetricToIoTDB(rate, metric, tags); + internalReporter.writeMetricToIoTDB(rate, metric, time, tags); return rate; } /** GetOrCreateHistogram with internal report. */ public Histogram getOrCreateHistogramWithInternalReport( - String metric, MetricLevel metricLevel, String... tags) { + String metric, MetricLevel metricLevel, long time, String... tags) { Histogram histogram = metricManager.getOrCreateHistogram(metric, metricLevel, tags); - internalReporter.writeMetricToIoTDB(histogram, metric, tags); + internalReporter.writeMetricToIoTDB(histogram, metric, time, tags); return histogram; } /** GetOrCreateTimer with internal report. */ public Timer getOrCreateTimerWithInternalReport( - String metric, MetricLevel metricLevel, String... tags) { + String metric, MetricLevel metricLevel, long time, String... tags) { Timer timer = metricManager.getOrCreateTimer(metric, metricLevel, tags); - internalReporter.writeMetricToIoTDB(timer, metric, tags); + internalReporter.writeMetricToIoTDB(timer, metric, time, tags); return timer; } /** Count with internal report. */ public void countWithInternalReportAsync( - long delta, String metric, MetricLevel metricLevel, String... tags) { + long delta, String metric, MetricLevel metricLevel, long time, String... tags) { internalReporter.writeMetricToIoTDB( - metricManager.count(delta, metric, metricLevel, tags), metric, tags); + metricManager.count(delta, metric, metricLevel, tags), metric, time, tags); } /** Gauge value with internal report */ public void gaugeWithInternalReportAsync( - long value, String metric, MetricLevel metricLevel, String... tags) { + long value, String metric, MetricLevel metricLevel, long time, String... tags) { internalReporter.writeMetricToIoTDB( - metricManager.gauge(value, metric, metricLevel, tags), metric, tags); + metricManager.gauge(value, metric, metricLevel, tags), metric, time, tags); } /** Rate with internal report. */ public void rateWithInternalReportAsync( - long value, String metric, MetricLevel metricLevel, String... tags) { + long value, String metric, MetricLevel metricLevel, long time, String... tags) { internalReporter.writeMetricToIoTDB( - metricManager.rate(value, metric, metricLevel, tags), metric, tags); + metricManager.rate(value, metric, metricLevel, tags), metric, time, tags); } /** Histogram with internal report. */ public void histogramWithInternalReportAsync( - long value, String metric, MetricLevel metricLevel, String... tags) { + long value, String metric, MetricLevel metricLevel, long time, String... tags) { internalReporter.writeMetricToIoTDB( - metricManager.histogram(value, metric, metricLevel, tags), metric, tags); + metricManager.histogram(value, metric, metricLevel, tags), metric, time, tags); } /** Timer with internal report. */ public void timerWithInternalReportAsync( - long delta, TimeUnit timeUnit, String metric, MetricLevel metricLevel, String... tags) { + long delta, + TimeUnit timeUnit, + String metric, + MetricLevel metricLevel, + long time, + String... tags) { internalReporter.writeMetricToIoTDB( - metricManager.timer(delta, timeUnit, metric, metricLevel, tags), metric, tags); + metricManager.timer(delta, timeUnit, metric, metricLevel, tags), metric, time, tags); } public List<Pair<String, String[]>> getAllMetricKeys() { diff --git a/metrics/interface/src/main/java/org/apache/iotdb/metrics/reporter/iotdb/IoTDBReporter.java b/metrics/interface/src/main/java/org/apache/iotdb/metrics/reporter/iotdb/IoTDBReporter.java index 2cc67f13a31..db1e915c8e9 100644 --- a/metrics/interface/src/main/java/org/apache/iotdb/metrics/reporter/iotdb/IoTDBReporter.java +++ b/metrics/interface/src/main/java/org/apache/iotdb/metrics/reporter/iotdb/IoTDBReporter.java @@ -36,14 +36,14 @@ public abstract class IoTDBReporter implements Reporter { * * @param metric the target metric * @param name the name of metric + * @param time the target time of metric * @param tags the tags of metric */ - public void writeMetricToIoTDB(IMetric metric, String name, String... tags) { + public void writeMetricToIoTDB(IMetric metric, String name, long time, String... tags) { if (!(metric instanceof DoNothingMetric)) { Map<String, Object> values = new HashMap<>(); metric.constructValueMap(values); - writeMetricToIoTDB( - values, IoTDBMetricsUtils.generatePath(name, tags), System.currentTimeMillis()); + writeMetricToIoTDB(values, IoTDBMetricsUtils.generatePath(name, tags), time); } } diff --git a/server/src/main/java/org/apache/iotdb/db/engine/flush/MemTableFlushTask.java b/server/src/main/java/org/apache/iotdb/db/engine/flush/MemTableFlushTask.java index 35199617d42..26212ff2563 100644 --- a/server/src/main/java/org/apache/iotdb/db/engine/flush/MemTableFlushTask.java +++ b/server/src/main/java/org/apache/iotdb/db/engine/flush/MemTableFlushTask.java @@ -30,6 +30,7 @@ import org.apache.iotdb.db.engine.memtable.IWritableMemChunkGroup; import org.apache.iotdb.db.exception.runtime.FlushRunTimeException; import org.apache.iotdb.db.metadata.idtable.entry.IDeviceID; import org.apache.iotdb.db.rescon.SystemInfo; +import org.apache.iotdb.db.utils.DateTimeUtils; import org.apache.iotdb.metrics.utils.IoTDBMetricsUtils; import org.apache.iotdb.metrics.utils.MetricLevel; import org.apache.iotdb.tsfile.write.chunk.IChunkWriter; @@ -43,6 +44,7 @@ import java.util.ArrayList; import java.util.Comparator; import java.util.List; import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutionException; import java.util.concurrent.Future; import java.util.concurrent.LinkedBlockingQueue; @@ -58,6 +60,8 @@ public class MemTableFlushTask { private static final FlushSubTaskPoolManager SUB_TASK_POOL_MANAGER = FlushSubTaskPoolManager.getInstance(); private static IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig(); + /* storage group name -> last time */ + private static final Map<String, Long> flushPointsCache = new ConcurrentHashMap<>(); private final Future<?> encodingTaskFuture; private final Future<?> ioTaskFuture; private RestorableTsFileIOWriter writer; @@ -257,23 +261,7 @@ public class MemTableFlushTask { Thread.currentThread().interrupt(); } - if (!storageGroup.startsWith(IoTDBMetricsUtils.DATABASE)) { - int lastIndex = storageGroup.lastIndexOf("-"); - if (lastIndex == -1) { - lastIndex = storageGroup.length(); - } - MetricService.getInstance() - .gaugeWithInternalReportAsync( - memTable.getTotalPointsNum(), - Metric.POINTS.toString(), - MetricLevel.CORE, - Tag.DATABASE.toString(), - storageGroup.substring(0, lastIndex), - Tag.TYPE.toString(), - "flush", - Tag.REGION.toString(), - dataRegionId); - } + recordFlushPointsMetric(); LOGGER.info( "Database {}, flushing memtable {} into disk: Encoding data cost " + "{} ms.", @@ -283,6 +271,42 @@ public class MemTableFlushTask { } }; + private void recordFlushPointsMetric() { + if (storageGroup.startsWith(IoTDBMetricsUtils.DATABASE)) { + return; + } + int lastIndex = storageGroup.lastIndexOf("-"); + if (lastIndex == -1) { + lastIndex = storageGroup.length(); + } + String storageGroupName = storageGroup.substring(0, lastIndex); + long currentTime = DateTimeUtils.currentTime(); + // compute the flush points + long writeTime = + flushPointsCache.compute( + storageGroupName, + (storageGroup, lastTime) -> { + if (lastTime == null || lastTime != currentTime) { + return currentTime; + } else { + return currentTime + 1; + } + }); + // record the flush points + MetricService.getInstance() + .gaugeWithInternalReportAsync( + memTable.getTotalPointsNum(), + Metric.POINTS.toString(), + MetricLevel.CORE, + writeTime, + Tag.DATABASE.toString(), + storageGroup.substring(0, lastIndex), + Tag.TYPE.toString(), + "flush", + Tag.REGION.toString(), + dataRegionId); + } + /** io task (third task of pipeline) */ @SuppressWarnings("squid:S135") private Runnable ioTask =
