This is an automated email from the ASF dual-hosted git repository.
JackieTien97 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 49336d47731 [Metrics] Fix concurrent Prometheus scrapes and disk
sampling (#18957)
49336d47731 is described below
commit 49336d477313b504182ff9627b3ebcf28d1f623a
Author: Jackie Tien <[email protected]>
AuthorDate: Wed Oct 21 19:32:06 2026 +0800
[Metrics] Fix concurrent Prometheus scrapes and disk sampling (#18957)
---
.../iotdb/db/it/metric/IoTDBMetricFormatIT.java | 109 +++++++++++++++
.../apache/iotdb/db/it/metric/IoTDBMetricIT.java | 5 +-
.../metricsets/disk/LinuxDiskMetricsManager.java | 120 +++++++++++------
.../reporter/prometheus/PrometheusReporter.java | 33 +++--
.../disk/LinuxDiskMetricsManagerTest.java | 149 +++++++++++++++++++++
.../prometheus/PrometheusReporterTest.java | 131 ++++++++++++++++++
6 files changed, 492 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..a4cf8a59208
--- /dev/null
+++
b/integration-test/src/test/java/org/apache/iotdb/db/it/metric/IoTDBMetricFormatIT.java
@@ -0,0 +1,109 @@
+/*
+ * 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.apache.iotdb.metrics.impl.DoNothingAutoGauge;
+import org.apache.iotdb.metrics.impl.DoNothingMetricManager;
+import org.apache.iotdb.metrics.reporter.prometheus.PrometheusReporter;
+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 org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+import java.lang.reflect.Method;
+import java.util.Collections;
+import java.util.Map;
+
+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",
+ "nan",
+ "Inf",
+ "+Inf",
+ "-Inf",
+ "Infinity",
+ "+Infinity",
+ "-Infinity",
+ "iNfInItY",
+ "+inf",
+ "-infinity",
+ "0",
+ "-0.25",
+ "1.0e-3"
+ }) {
+ assertTrue(value, IoTDBMetricIT.isValidPrometheusTextFormat(SAMPLE +
value));
+ }
+ }
+
+ @Test
+ public void testInvalidSampleValues() {
+ for (String value :
+ new String[] {
+ "", "Infinite", "Infinitygarbage", "--Inf", "+NaN", "-NaN", "null",
"1.0garbage"
+ }) {
+ assertFalse(value, IoTDBMetricIT.isValidPrometheusTextFormat(SAMPLE +
value));
+ }
+ }
+
+ @Test
+ public void testReporterSpecialValuesRoundTrip() throws Exception {
+ Method scrape = PrometheusReporter.class.getDeclaredMethod("scrape");
+ scrape.setAccessible(true);
+ for (double sampleValue :
+ new double[] {Double.POSITIVE_INFINITY, Double.NEGATIVE_INFINITY,
Double.NaN}) {
+ DoNothingAutoGauge sampleGauge =
+ new DoNothingAutoGauge() {
+ @Override
+ public double getValue() {
+ return sampleValue;
+ }
+ };
+ PrometheusReporter reporter =
+ new PrometheusReporter(
+ new DoNothingMetricManager() {
+ @Override
+ public Map<MetricInfo, IMetric> getAllMetrics() {
+ return Collections.singletonMap(
+ new MetricInfo(MetricType.AUTO_GAUGE, "special_value"),
sampleGauge);
+ }
+ });
+ String metrics = (String) scrape.invoke(reporter);
+ assertTrue(metrics, metrics.contains("} " + Double.toString(sampleValue)
+ "\n"));
+ assertTrue(metrics, IoTDBMetricIT.isValidPrometheusTextFormat(metrics));
+ }
+ }
+}
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..1d1b06b11ad 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
@@ -49,6 +49,7 @@ public class IoTDBMetricIT {
private static final String METRIC_NAME_REGEX = "[a-zA-Z_:][a-zA-Z0-9_:]*";
private static final String LABEL_NAME_REGEX = "[a-zA-Z_][a-zA-Z0-9_]*";
private static final String LABEL_VALUE_REGEX = "[^\",]+";
+ // Go ParseFloat accepts case-insensitive NaN, Inf and Infinity, with signs
only for infinities.
private static final String METRIC_LINE_REGEX =
METRIC_NAME_REGEX
+ "(\\{"
@@ -59,7 +60,7 @@ public class IoTDBMetricIT {
+ LABEL_NAME_REGEX
+ "=\""
+ LABEL_VALUE_REGEX
- + "\",)*})? [+-]?[0-9]*\\.?[0-9]+([eE][+-]?[0-9]+)?";
+ + "\",)*})?
([+-]?[0-9]*\\.?[0-9]+([eE][+-]?[0-9]+)?|(?i:NaN|[+-]?Inf(?:inity)?))";
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 +75,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 30ce9573637..eb048dafbc8 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
@@ -80,6 +80,8 @@ public class PrometheusReporter implements Reporter {
private static final long PROMETHEUS_DEFAULT_SCRAPE_INTERVAL_SECONDS = 15;
private static final int PROMETHEUS_RESPONSE_CHUNK_SIZE = 8 * 1024;
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;
@@ -232,13 +234,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
@@ -250,9 +246,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;
@@ -311,6 +310,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 5b19d7f5f31..4ad8ddc1350 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,24 +21,39 @@ 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 reactor.netty.DisposableServer;
import java.io.InputStream;
import java.lang.reflect.Field;
+import java.lang.reflect.Method;
import java.net.HttpURLConnection;
import java.net.URI;
import java.nio.charset.StandardCharsets;
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 {
@@ -79,6 +94,47 @@ public class PrometheusReporterTest {
}
}
+ @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"));
+ }
+
@Test
public void testMetricsResponseChunksPreserveContentAndSurrogatePairs() {
String metrics = "x".repeat(8 * 1024 - 1) + "\uD83D\uDE00" + "y".repeat(8
* 1024 + 1);
@@ -129,4 +185,79 @@ public class PrometheusReporterTest {
metricConfig.setPrometheusReporterPort(originalPort);
}
}
+
+ 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);
+ }
}
