This is an automated email from the ASF dual-hosted git repository. JackieTien97 pushed a commit to branch ty/fix-prometheus-scrape-race in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit dcef0c99f3ae359aca1d8b684eca21f17c0a0629 Author: JackieTien97 <[email protected]> AuthorDate: Wed Oct 21 14:56:59 2026 +0800 [Metrics] Fix concurrent Prometheus scrapes and disk sampling --- .../iotdb/db/it/metric/IoTDBMetricFormatIT.java | 52 +++++++ .../apache/iotdb/db/it/metric/IoTDBMetricIT.java | 4 +- .../metricsets/disk/LinuxDiskMetricsManager.java | 120 +++++++++++------ .../reporter/prometheus/PrometheusReporter.java | 33 +++-- .../disk/LinuxDiskMetricsManagerTest.java | 149 +++++++++++++++++++++ .../prometheus/PrometheusReporterTest.java | 131 ++++++++++++++++++ 6 files changed, 434 insertions(+), 55 deletions(-) diff --git a/integration-test/src/test/java/org/apache/iotdb/db/it/metric/IoTDBMetricFormatIT.java b/integration-test/src/test/java/org/apache/iotdb/db/it/metric/IoTDBMetricFormatIT.java new file mode 100644 index 00000000000..4d1944cbe07 --- /dev/null +++ b/integration-test/src/test/java/org/apache/iotdb/db/it/metric/IoTDBMetricFormatIT.java @@ -0,0 +1,52 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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.apache.iotdb.db.it.metric; + +import org.apache.iotdb.it.framework.IoTDBTestRunner; +import org.apache.iotdb.itbase.category.ClusterIT; +import org.apache.iotdb.itbase.category.LocalStandaloneIT; + +import org.junit.Test; +import org.junit.experimental.categories.Category; +import org.junit.runner.RunWith; + +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +@RunWith(IoTDBTestRunner.class) +@Category({LocalStandaloneIT.class, ClusterIT.class}) +public class IoTDBMetricFormatIT { + private static final String SAMPLE = + "disk_io_busy_percentage{cluster=\"defaultCluster\",nodeType=\"CONFIGNODE\",nodeId=\"0\",disk_id=\"vda\",} "; + + @Test + public void testValidSampleValues() { + for (String value : new String[] {"NaN", "+Inf", "-Inf", "0", "-0.25", "1.0e-3"}) { + assertTrue(value, IoTDBMetricIT.isValidPrometheusTextFormat(SAMPLE + value)); + } + } + + @Test + public void testInvalidSampleValues() { + for (String value : new String[] {"", "Infinity", "null", "1.0garbage"}) { + assertFalse(value, IoTDBMetricIT.isValidPrometheusTextFormat(SAMPLE + value)); + } + } +} diff --git a/integration-test/src/test/java/org/apache/iotdb/db/it/metric/IoTDBMetricIT.java b/integration-test/src/test/java/org/apache/iotdb/db/it/metric/IoTDBMetricIT.java index 3e6f660d4f2..fc377dac082 100644 --- a/integration-test/src/test/java/org/apache/iotdb/db/it/metric/IoTDBMetricIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/db/it/metric/IoTDBMetricIT.java @@ -59,7 +59,7 @@ public class IoTDBMetricIT { + LABEL_NAME_REGEX + "=\"" + LABEL_VALUE_REGEX - + "\",)*})? [+-]?[0-9]*\\.?[0-9]+([eE][+-]?[0-9]+)?"; + + "\",)*})? ([+-]?[0-9]*\\.?[0-9]+([eE][+-]?[0-9]+)?|NaN|[+-]Inf)"; private static final String HELP_PREFIX = "# HELP "; private static final String HELP_REGEX = HELP_PREFIX + METRIC_NAME_REGEX; private static final String TYPE_PREFIX = "# TYPE "; @@ -74,7 +74,7 @@ public class IoTDBMetricIT { private static final String WRONG_USERNAME = "bad"; private static final String WRONG_PASSWORD = "!!"; - private static boolean isValidPrometheusTextFormat(String metrics) { + static boolean isValidPrometheusTextFormat(String metrics) { String[] lines = metrics.split("\\n"); boolean valid = true; diff --git a/iotdb-core/metrics/interface/src/main/java/org/apache/iotdb/metrics/metricsets/disk/LinuxDiskMetricsManager.java b/iotdb-core/metrics/interface/src/main/java/org/apache/iotdb/metrics/metricsets/disk/LinuxDiskMetricsManager.java index cb8ea23a821..b9c65f69091 100644 --- a/iotdb-core/metrics/interface/src/main/java/org/apache/iotdb/metrics/metricsets/disk/LinuxDiskMetricsManager.java +++ b/iotdb-core/metrics/interface/src/main/java/org/apache/iotdb/metrics/metricsets/disk/LinuxDiskMetricsManager.java @@ -28,13 +28,17 @@ import org.slf4j.LoggerFactory; import java.io.File; import java.io.IOException; import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; import java.util.ArrayList; import java.util.Arrays; +import java.util.Collections; import java.util.HashMap; import java.util.Map; import java.util.Objects; import java.util.Scanner; import java.util.Set; +import java.util.function.LongSupplier; import java.util.stream.Collectors; /** @@ -59,10 +63,10 @@ public class LinuxDiskMetricsManager implements IDiskMetricsManager { @SuppressWarnings("squid:S1075") private static final String DISK_ID_PATH = "/sys/block"; - @SuppressWarnings("squid:S1075") - private static final String DISK_SECTOR_SIZE_PATH = "/sys/block/%s/queue/hw_sector_size"; - - private final String processIoStatusPath; + private final Path diskStatsPath; + private final Path diskIdPath; + private final Path processIoStatusPath; + private final LongSupplier clock; private static final int DISK_ID_OFFSET = 3; private static final int DISK_READ_COUNT_OFFSET = 4; private static final int DISK_MERGED_READ_COUNT_OFFSET = 5; @@ -77,7 +81,7 @@ public class LinuxDiskMetricsManager implements IDiskMetricsManager { private static final int DEFAULT_SECTOR_SIZE = 512; private static final double BYTES_PER_KB = 1024.0; private static final long UPDATE_SMALLEST_INTERVAL = 10000L; - private Set<String> diskIdSet; + private Set<String> diskIdSet = Collections.emptySet(); private final Map<String, Integer> diskSectorSizeMap; private long lastUpdateTime = 0L; private long updateInterval = 1L; @@ -113,9 +117,21 @@ public class LinuxDiskMetricsManager implements IDiskMetricsManager { private long lastWriteOpsCountForProcess = 0L; public LinuxDiskMetricsManager() { - processIoStatusPath = - String.format( - "/proc/%s/io", MetricConfigDescriptor.getInstance().getMetricConfig().getPid()); + this( + Paths.get(DISK_STATUS_FILE_PATH), + Paths.get(DISK_ID_PATH), + Paths.get( + String.format( + "/proc/%s/io", MetricConfigDescriptor.getInstance().getMetricConfig().getPid())), + System::currentTimeMillis); + } + + LinuxDiskMetricsManager( + Path diskStatsPath, Path diskIdPath, Path processIoStatusPath, LongSupplier clock) { + this.diskStatsPath = diskStatsPath; + this.diskIdPath = diskIdPath; + this.processIoStatusPath = processIoStatusPath; + this.clock = clock; collectDiskId(); // leave one entry to avoid hashmap resizing diskSectorSizeMap = new HashMap<>(diskIdSet.size() + 1, 1); @@ -143,7 +159,7 @@ public class LinuxDiskMetricsManager implements IDiskMetricsManager { } @Override - public Map<String, Double> getReadDataSizeForDisk() { + public synchronized Map<String, Double> getReadDataSizeForDisk() { checkUpdate(); Map<String, Double> readDataMap = new HashMap<>(diskIdSet.size()); for (Map.Entry<String, Long> entry : lastReadSectorCountForDisk.entrySet()) { @@ -154,7 +170,7 @@ public class LinuxDiskMetricsManager implements IDiskMetricsManager { } @Override - public Map<String, Double> getWriteDataSizeForDisk() { + public synchronized Map<String, Double> getWriteDataSizeForDisk() { checkUpdate(); Map<String, Double> writeDataMap = new HashMap<>(diskIdSet.size()); for (Map.Entry<String, Long> entry : lastWriteSectorCountForDisk.entrySet()) { @@ -165,28 +181,32 @@ public class LinuxDiskMetricsManager implements IDiskMetricsManager { } @Override - public Map<String, Long> getReadOperationCountForDisk() { + public synchronized Map<String, Long> getReadOperationCountForDisk() { checkUpdate(); - return lastReadOperationCountForDisk; + return new HashMap<>(lastReadOperationCountForDisk); } @Override - public Map<String, Long> getWriteOperationCountForDisk() { - return lastWriteOperationCountForDisk; + public synchronized Map<String, Long> getWriteOperationCountForDisk() { + checkUpdate(); + return new HashMap<>(lastWriteOperationCountForDisk); } @Override - public Map<String, Long> getReadCostTimeForDisk() { - return lastReadTimeCostForDisk; + public synchronized Map<String, Long> getReadCostTimeForDisk() { + checkUpdate(); + return new HashMap<>(lastReadTimeCostForDisk); } @Override - public Map<String, Long> getWriteCostTimeForDisk() { - return lastWriteTimeCostForDisk; + public synchronized Map<String, Long> getWriteCostTimeForDisk() { + checkUpdate(); + return new HashMap<>(lastWriteTimeCostForDisk); } @Override - public Map<String, Double> getIoUtilsPercentage() { + public synchronized Map<String, Double> getIoUtilsPercentage() { + checkUpdate(); Map<String, Double> utilsMap = new HashMap<>(diskIdSet.size()); for (Map.Entry<String, Long> entry : incrementIoBusyTimeForDisk.entrySet()) { utilsMap.put(entry.getKey(), ((double) entry.getValue()) / updateInterval); @@ -195,7 +215,8 @@ public class LinuxDiskMetricsManager implements IDiskMetricsManager { } @Override - public Map<String, Double> getAvgReadCostTimeOfEachOpsForDisk() { + public synchronized Map<String, Double> getAvgReadCostTimeOfEachOpsForDisk() { + checkUpdate(); Map<String, Double> avgReadTimeCostMap = new HashMap<>(diskIdSet.size()); for (Map.Entry<String, Long> readCostEntry : incrementReadTimeCostForDisk.entrySet()) { // use Long.max to avoid NaN @@ -212,7 +233,8 @@ public class LinuxDiskMetricsManager implements IDiskMetricsManager { } @Override - public Map<String, Double> getAvgWriteCostTimeOfEachOpsForDisk() { + public synchronized Map<String, Double> getAvgWriteCostTimeOfEachOpsForDisk() { + checkUpdate(); Map<String, Double> avgWriteTimeCostMap = new HashMap<>(diskIdSet.size()); for (Map.Entry<String, Long> writeCostEntry : incrementWriteTimeCostForDisk.entrySet()) { // use Long.max to avoid NaN @@ -229,7 +251,8 @@ public class LinuxDiskMetricsManager implements IDiskMetricsManager { } @Override - public Map<String, Double> getAvgSizeOfEachReadForDisk() { + public synchronized Map<String, Double> getAvgSizeOfEachReadForDisk() { + checkUpdate(); Map<String, Double> avgSizeOfReadMap = new HashMap<>(diskIdSet.size()); for (Map.Entry<String, Long> readSectorSizeEntry : incrementReadSectorCountForDisk.entrySet()) { // use Long.max to avoid NaN @@ -249,7 +272,8 @@ public class LinuxDiskMetricsManager implements IDiskMetricsManager { } @Override - public Map<String, Double> getAvgSizeOfEachWriteForDisk() { + public synchronized Map<String, Double> getAvgSizeOfEachWriteForDisk() { + checkUpdate(); Map<String, Double> avgSizeOfWriteMap = new HashMap<>(diskIdSet.size()); for (Map.Entry<String, Long> writeSectorSizeEntry : incrementWriteSectorCountForDisk.entrySet()) { @@ -270,17 +294,20 @@ public class LinuxDiskMetricsManager implements IDiskMetricsManager { } @Override - public Map<String, Long> getMergedWriteOperationForDisk() { - return lastMergedWriteCountForDisk; + public synchronized Map<String, Long> getMergedWriteOperationForDisk() { + checkUpdate(); + return new HashMap<>(lastMergedWriteCountForDisk); } @Override - public Map<String, Long> getMergedReadOperationForDisk() { - return lastMergedReadCountForDisk; + public synchronized Map<String, Long> getMergedReadOperationForDisk() { + checkUpdate(); + return new HashMap<>(lastMergedReadCountForDisk); } @Override - public Map<String, Double> getQueueSizeForDisk() { + public synchronized Map<String, Double> getQueueSizeForDisk() { + checkUpdate(); Map<String, Double> avgQueueSizeMap = new HashMap<>(diskIdSet.size()); for (Map.Entry<String, Long> entry : incrementTimeInQueueForDisk.entrySet()) { avgQueueSizeMap.put(entry.getKey(), (((double) entry.getValue()) / updateInterval)); @@ -289,32 +316,38 @@ public class LinuxDiskMetricsManager implements IDiskMetricsManager { } @Override - public double getActualReadDataSizeForProcess() { + public synchronized double getActualReadDataSizeForProcess() { + checkUpdate(); return lastReallyReadSizeForProcess / BYTES_PER_KB; } @Override - public double getActualWriteDataSizeForProcess() { + public synchronized double getActualWriteDataSizeForProcess() { + checkUpdate(); return lastReallyWriteSizeForProcess / BYTES_PER_KB; } @Override - public long getReadOpsCountForProcess() { + public synchronized long getReadOpsCountForProcess() { + checkUpdate(); return lastReadOpsCountForProcess; } @Override - public long getWriteOpsCountForProcess() { + public synchronized long getWriteOpsCountForProcess() { + checkUpdate(); return lastWriteOpsCountForProcess; } @Override - public double getAttemptReadSizeForProcess() { + public synchronized double getAttemptReadSizeForProcess() { + checkUpdate(); return lastAttemptReadSizeForProcess / BYTES_PER_KB; } @Override - public double getAttemptWriteSizeForProcess() { + public synchronized double getAttemptWriteSizeForProcess() { + checkUpdate(); return lastAttemptWriteSizeForProcess / BYTES_PER_KB; } @@ -324,7 +357,7 @@ public class LinuxDiskMetricsManager implements IDiskMetricsManager { } private void collectDiskId() { - File diskIdFolder = new File(DISK_ID_PATH); + File diskIdFolder = diskIdPath.toFile(); if (!diskIdFolder.exists()) { return; } @@ -338,8 +371,7 @@ public class LinuxDiskMetricsManager implements IDiskMetricsManager { private void collectDiskInfo() { for (String diskId : diskIdSet) { - String diskSectorSizePath = String.format(DISK_SECTOR_SIZE_PATH, diskId); - File diskSectorSizeFile = new File(diskSectorSizePath); + File diskSectorSizeFile = diskIdPath.resolve(diskId).resolve("queue/hw_sector_size").toFile(); try (Scanner scanner = new Scanner(Files.newInputStream(diskSectorSizeFile.toPath()))) { if (scanner.hasNext()) { int sectorSize = Integer.parseInt(scanner.nextLine()); @@ -356,8 +388,7 @@ public class LinuxDiskMetricsManager implements IDiskMetricsManager { } } - private void updateInfo() { - long currentTime = System.currentTimeMillis(); + private void updateInfo(long currentTime) { updateInterval = currentTime - lastUpdateTime; lastUpdateTime = currentTime; updateDiskInfo(); @@ -365,7 +396,7 @@ public class LinuxDiskMetricsManager implements IDiskMetricsManager { } private void updateDiskInfo() { - File diskStatsFile = new File(DISK_STATUS_FILE_PATH); + File diskStatsFile = diskStatsPath.toFile(); if (!diskStatsFile.exists()) { LOGGER.warn(MetricsMessages.CANNOT_FIND_DISK_IO_STATUS_FILE, DISK_STATUS_FILE_PATH); return; @@ -443,7 +474,7 @@ public class LinuxDiskMetricsManager implements IDiskMetricsManager { } private void updateProcessInfo() { - File processStatInfoFile = new File(processIoStatusPath); + File processStatInfoFile = processIoStatusPath.toFile(); if (!processStatInfoFile.exists()) { LOGGER.warn(MetricsMessages.CANNOT_FIND_PROCESS_IO_STATUS_FILE, processIoStatusPath); } @@ -472,8 +503,11 @@ public class LinuxDiskMetricsManager implements IDiskMetricsManager { } private void checkUpdate() { - if (System.currentTimeMillis() - lastUpdateTime > UPDATE_SMALLEST_INTERVAL) { - updateInfo(); + // Accessors hold this monitor throughout refresh and snapshot construction. Reuse the checked + // timestamp so another clock read cannot turn a valid interval into zero or a negative value. + long currentTime = clock.getAsLong(); + if (currentTime - lastUpdateTime > UPDATE_SMALLEST_INTERVAL) { + updateInfo(currentTime); } } } diff --git a/iotdb-core/metrics/interface/src/main/java/org/apache/iotdb/metrics/reporter/prometheus/PrometheusReporter.java b/iotdb-core/metrics/interface/src/main/java/org/apache/iotdb/metrics/reporter/prometheus/PrometheusReporter.java index 00ae6c6b988..f852cc8b40b 100644 --- a/iotdb-core/metrics/interface/src/main/java/org/apache/iotdb/metrics/reporter/prometheus/PrometheusReporter.java +++ b/iotdb-core/metrics/interface/src/main/java/org/apache/iotdb/metrics/reporter/prometheus/PrometheusReporter.java @@ -78,6 +78,8 @@ public class PrometheusReporter implements Reporter { MetricConfigDescriptor.getInstance().getMetricConfig(); private static final long PROMETHEUS_DEFAULT_SCRAPE_INTERVAL_SECONDS = 15; private final AbstractMetricManager metricManager; + // Use a separate lock so stopping the HTTP server cannot wait on its own request threads. + private final Object scrapeLock = new Object(); private final Supplier<ScheduledExecutorService> snapshotUpdateExecutorSupplier; private volatile ScheduledExecutorService snapshotUpdateExecutor; private volatile DisposableServer httpServer; @@ -208,13 +210,7 @@ public class PrometheusReporter implements Reporter { private void updateSnapshot() { try { - String snapshot = scrape(); - // Do not publish an empty scrape taken before metric sets are bound. The request path will - // synchronously scrape until the first complete snapshot is available. Empty snapshots are - // published after initialization so removed metrics are not kept in the cache indefinitely. - if (!snapshot.isEmpty() || metricsSnapshot != null) { - metricsSnapshot = snapshot; - } + scrape(); } catch (Throwable t) { LOGGER.error( MetricsMessages @@ -226,9 +222,12 @@ public class PrometheusReporter implements Reporter { private String getMetricsSnapshot() { String snapshot = metricsSnapshot; if (snapshot == null) { - snapshot = scrape(); - if (!snapshot.isEmpty()) { - metricsSnapshot = snapshot; + synchronized (scrapeLock) { + // The background updater or another first request may have populated the cache meanwhile. + snapshot = metricsSnapshot; + if (snapshot == null) { + snapshot = scrape(); + } } } return snapshot; @@ -287,6 +286,20 @@ public class PrometheusReporter implements Reporter { } private String scrape() { + synchronized (scrapeLock) { + // Metric suppliers may update shared sampling state. Serialize all scrapes, including the + // first-request fallback and synchronous mode, and publish in the same order as collection. + String snapshot = collectMetrics(); + // Do not cache an empty scrape before metric sets are bound. After initialization, publish + // empty snapshots too so removed metrics are not retained indefinitely. + if (!snapshot.isEmpty() || metricsSnapshot != null) { + metricsSnapshot = snapshot; + } + return snapshot; + } + } + + private String collectMetrics() { Writer writer = new StringWriter(); PrometheusTextWriter prometheusTextWriter = new PrometheusTextWriter(writer); diff --git a/iotdb-core/metrics/interface/src/test/java/org/apache/iotdb/metrics/metricsets/disk/LinuxDiskMetricsManagerTest.java b/iotdb-core/metrics/interface/src/test/java/org/apache/iotdb/metrics/metricsets/disk/LinuxDiskMetricsManagerTest.java new file mode 100644 index 00000000000..3463fcf4d0a --- /dev/null +++ b/iotdb-core/metrics/interface/src/test/java/org/apache/iotdb/metrics/metricsets/disk/LinuxDiskMetricsManagerTest.java @@ -0,0 +1,149 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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.apache.iotdb.metrics.metricsets.disk; + +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.TemporaryFolder; + +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.FutureTask; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.locks.LockSupport; +import java.util.function.LongSupplier; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; + +public class LinuxDiskMetricsManagerTest { + @Rule public TemporaryFolder temporaryFolder = new TemporaryFolder(); + + private Path diskStats; + private Path diskIds; + private Path processIo; + private final AtomicLong clock = new AtomicLong(10001); + + @Before + public void setUp() throws Exception { + diskStats = temporaryFolder.newFile("diskstats").toPath(); + diskIds = temporaryFolder.newFolder("block").toPath(); + Path queue = Files.createDirectories(diskIds.resolve("vda/queue")); + Files.write(queue.resolve("hw_sector_size"), "512\n".getBytes(StandardCharsets.UTF_8)); + processIo = temporaryFolder.newFile("io").toPath(); + Files.write(processIo, "syscr: 1\nsyscw: 2\n".getBytes(StandardCharsets.UTF_8)); + writeDiskStats(10, 100, 200); + } + + @Test + public void testFirstSampleAndUnchangedOrBackwardClock() throws Exception { + LinuxDiskMetricsManager manager = newManager(clock::get); + assertEquals(0.0, manager.getIoUtilsPercentage().get("vda"), 0.0); + assertEquals(0.0, manager.getQueueSizeForDisk().get("vda"), 0.0); + + clock.addAndGet(20000); + writeDiskStats(20, 5100, 10200); + assertEquals(0.25, manager.getIoUtilsPercentage().get("vda"), 0.0); + assertEquals(0.5, manager.getQueueSizeForDisk().get("vda"), 0.0); + Map<String, Long> previousCounts = manager.getReadOperationCountForDisk(); + + writeDiskStats(30, 15100, 30200); + // Neither a zero interval nor a backward wall clock replaces the last valid sample. + assertEquals(0.25, manager.getIoUtilsPercentage().get("vda"), 0.0); + clock.set(1); + assertEquals(0.5, manager.getQueueSizeForDisk().get("vda"), 0.0); + clock.set(50001); + assertEquals(0.5, manager.getIoUtilsPercentage().get("vda"), 0.0); + assertEquals(1.0, manager.getQueueSizeForDisk().get("vda"), 0.0); + assertEquals(Long.valueOf(20), previousCounts.get("vda")); + assertEquals(Long.valueOf(30), manager.getReadOperationCountForDisk().get("vda")); + } + + @Test + public void testRefreshUsesTheTimestampCheckedForEligibility() { + // A second clock read would return the previous sample time and cause a zero denominator. + AtomicLong nextTime = new AtomicLong(10001); + LinuxDiskMetricsManager manager = newManager(() -> nextTime.getAndSet(10001)); + manager.getReadDataSizeForDisk(); + nextTime.set(30001); + manager.getReadDataSizeForDisk(); + assertEquals(0.0, manager.getIoUtilsPercentage().get("vda"), 0.0); + assertEquals(0.0, manager.getQueueSizeForDisk().get("vda"), 0.0); + } + + @Test + public void testConcurrentReadersWaitForCompleteSample() throws Exception { + CountDownLatch sampling = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + LinuxDiskMetricsManager manager = + newManager( + () -> { + sampling.countDown(); + try { + assertTrue(release.await(10, TimeUnit.SECONDS)); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new AssertionError(e); + } + return clock.get(); + }); + FutureTask<Map<String, Double>> refresh = new FutureTask<>(manager::getReadDataSizeForDisk); + FutureTask<Map<String, Double>> read = new FutureTask<>(manager::getIoUtilsPercentage); + Thread refreshing = new Thread(refresh, "disk-refresh"); + Thread reading = new Thread(read, "disk-read"); + try { + refreshing.start(); + assertTrue(sampling.await(10, TimeUnit.SECONDS)); + reading.start(); + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(10); + while (reading.getState() != Thread.State.BLOCKED && !read.isDone()) { + if (System.nanoTime() > deadline) { + fail("The metric reader did not wait for the active disk sample"); + } + LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(1)); + } + assertEquals(Thread.State.BLOCKED, reading.getState()); + release.countDown(); + refresh.get(10, TimeUnit.SECONDS); + assertEquals(0.0, read.get(10, TimeUnit.SECONDS).get("vda"), 0.0); + } finally { + release.countDown(); + refresh.cancel(true); + read.cancel(true); + refreshing.join(TimeUnit.SECONDS.toMillis(10)); + reading.join(TimeUnit.SECONDS.toMillis(10)); + } + } + + private LinuxDiskMetricsManager newManager(LongSupplier timeSupplier) { + return new LinuxDiskMetricsManager(diskStats, diskIds, processIo, timeSupplier); + } + + private void writeDiskStats(long reads, long busyTime, long queueTime) throws Exception { + String stats = " 8 0 vda " + reads + " 0 40 0 0 0 0 0 0 " + busyTime + " " + queueTime + "\n"; + Files.write(diskStats, stats.getBytes(StandardCharsets.UTF_8)); + } +} diff --git a/iotdb-core/metrics/interface/src/test/java/org/apache/iotdb/metrics/reporter/prometheus/PrometheusReporterTest.java b/iotdb-core/metrics/interface/src/test/java/org/apache/iotdb/metrics/reporter/prometheus/PrometheusReporterTest.java index 7072777ac66..94bb66bd7af 100644 --- a/iotdb-core/metrics/interface/src/test/java/org/apache/iotdb/metrics/reporter/prometheus/PrometheusReporterTest.java +++ b/iotdb-core/metrics/interface/src/test/java/org/apache/iotdb/metrics/reporter/prometheus/PrometheusReporterTest.java @@ -21,18 +21,33 @@ package org.apache.iotdb.metrics.reporter.prometheus; import org.apache.iotdb.metrics.config.MetricConfig; import org.apache.iotdb.metrics.config.MetricConfigDescriptor; +import org.apache.iotdb.metrics.impl.DoNothingAutoGauge; import org.apache.iotdb.metrics.impl.DoNothingMetricManager; +import org.apache.iotdb.metrics.type.AutoGauge; +import org.apache.iotdb.metrics.type.IMetric; +import org.apache.iotdb.metrics.utils.MetricInfo; +import org.apache.iotdb.metrics.utils.MetricType; import org.junit.Test; +import java.lang.reflect.Method; import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; import java.util.List; +import java.util.Map; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.Executors; +import java.util.concurrent.FutureTask; import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.locks.LockSupport; +import java.util.function.DoubleSupplier; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; public class PrometheusReporterTest { @@ -72,4 +87,120 @@ public class PrometheusReporterTest { executors.forEach(ScheduledExecutorService::shutdownNow); } } + + @Test + public void testFirstRequestWaitsForBackgroundSnapshot() throws Exception { + assertSerializedScrapes("updateSnapshot", "getMetricsSnapshot", 1); + } + + @Test + public void testConcurrentFirstRequestsReuseSnapshot() throws Exception { + assertSerializedScrapes("getMetricsSnapshot", "getMetricsSnapshot", 1); + } + + @Test + public void testBackgroundRefreshWaitsForFirstRequest() throws Exception { + assertSerializedScrapes("getMetricsSnapshot", "updateSnapshot", 2); + } + + @Test + public void testSynchronousScrapeWaitsForFirstRequest() throws Exception { + assertSerializedScrapes("getMetricsSnapshot", "scrape", 2); + } + + @Test + public void testEmptySnapshotBeforeAndAfterMetricRegistration() throws Exception { + Map<MetricInfo, IMetric> registeredMetrics = new HashMap<>(); + PrometheusReporter reporter = + new PrometheusReporter( + new DoNothingMetricManager() { + @Override + public Map<MetricInfo, IMetric> getAllMetrics() { + return registeredMetrics; + } + }); + assertEquals("", invoke(reporter, "getMetricsSnapshot")); + registeredMetrics.put(new MetricInfo(MetricType.AUTO_GAUGE, "sample"), sampleGauge(() -> 1.0)); + String snapshot = (String) invoke(reporter, "getMetricsSnapshot"); + assertTrue(snapshot.contains(" 1.0\n")); + registeredMetrics.clear(); + assertEquals(snapshot, invoke(reporter, "getMetricsSnapshot")); + invoke(reporter, "updateSnapshot"); + assertEquals("", invoke(reporter, "getMetricsSnapshot")); + } + + private void assertSerializedScrapes(String firstMethod, String secondMethod, int expectedReads) + throws Exception { + CountDownLatch collecting = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + AtomicInteger reads = new AtomicInteger(); + DoubleSupplier supplier = + () -> { + int value = reads.incrementAndGet(); + if (value == 1) { + collecting.countDown(); + try { + assertTrue(release.await(10, TimeUnit.SECONDS)); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new AssertionError(e); + } + } + return value; + }; + PrometheusReporter reporter = + new PrometheusReporter( + new DoNothingMetricManager() { + @Override + public Map<MetricInfo, IMetric> getAllMetrics() { + return Collections.singletonMap( + new MetricInfo(MetricType.AUTO_GAUGE, "sample"), sampleGauge(supplier)); + } + }); + FutureTask<Object> first = new FutureTask<>(() -> invoke(reporter, firstMethod)); + FutureTask<Object> second = new FutureTask<>(() -> invoke(reporter, secondMethod)); + Thread firstThread = new Thread(first, "first-scrape"); + Thread secondThread = new Thread(second, "second-scrape"); + try { + firstThread.start(); + assertTrue(collecting.await(10, TimeUnit.SECONDS)); + secondThread.start(); + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(10); + while (secondThread.getState() != Thread.State.BLOCKED && !second.isDone()) { + if (System.nanoTime() > deadline) { + fail("The second scrape did not wait for the active collection"); + } + LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(1)); + } + assertEquals(Thread.State.BLOCKED, secondThread.getState()); + assertEquals(1, reads.get()); + release.countDown(); + first.get(10, TimeUnit.SECONDS); + second.get(10, TimeUnit.SECONDS); + String snapshot = (String) invoke(reporter, "getMetricsSnapshot"); + assertTrue(snapshot.endsWith(" " + expectedReads + ".0\n")); + assertEquals(expectedReads, reads.get()); + } finally { + release.countDown(); + first.cancel(true); + second.cancel(true); + firstThread.join(TimeUnit.SECONDS.toMillis(10)); + secondThread.join(TimeUnit.SECONDS.toMillis(10)); + } + } + + private static AutoGauge sampleGauge(DoubleSupplier supplier) { + return new DoNothingAutoGauge() { + @Override + public double getValue() { + return supplier.getAsDouble(); + } + }; + } + + private static Object invoke(PrometheusReporter reporter, String methodName) throws Exception { + Method method = PrometheusReporter.class.getDeclaredMethod(methodName); + method.setAccessible(true); + return method.invoke(reporter); + } }
