This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch optimize_prometheus_report
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/optimize_prometheus_report by
this push:
new 63295e6c8bb Fix Prometheus scheduler lifecycle and duplicate allocation
63295e6c8bb is described below
commit 63295e6c8bb8d4b974d1f5b868190e22cb538441
Author: Tian Jiang <[email protected]>
AuthorDate: Wed Sep 2 16:36:19 2026 +0800
Fix Prometheus scheduler lifecycle and duplicate allocation
---
.../reporter/prometheus/PrometheusReporter.java | 43 ++++++++-----
.../prometheus/PrometheusReporterTest.java | 75 ++++++++++++++++++++++
.../commons/service/metric/MetricService.java | 7 +-
3 files changed, 108 insertions(+), 17 deletions(-)
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 fa5695e9621..00ae6c6b988 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
@@ -70,6 +70,7 @@ import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
public class PrometheusReporter implements Reporter {
private static final Logger LOGGER =
LoggerFactory.getLogger(PrometheusReporter.class);
@@ -77,6 +78,7 @@ public class PrometheusReporter implements Reporter {
MetricConfigDescriptor.getInstance().getMetricConfig();
private static final long PROMETHEUS_DEFAULT_SCRAPE_INTERVAL_SECONDS = 15;
private final AbstractMetricManager metricManager;
+ private final Supplier<ScheduledExecutorService>
snapshotUpdateExecutorSupplier;
private volatile ScheduledExecutorService snapshotUpdateExecutor;
private volatile DisposableServer httpServer;
@@ -91,16 +93,30 @@ public class PrometheusReporter implements Reporter {
/**
* Creates a reporter with a self-managed scheduler for compatibility with
standalone users.
- * Server-side code should use the constructor accepting the IoTDB thread
pool.
+ * Server-side code should use the constructor accepting a scheduler factory.
*/
public PrometheusReporter(AbstractMetricManager metricManager) {
- this(metricManager, null);
+ this(metricManager,
PrometheusReporter::newStandaloneSnapshotUpdateExecutor);
}
+ /**
+ * Creates a reporter with a scheduler factory. The factory is invoked on
every start to obtain a
+ * fresh executor for the reporter lifecycle.
+ */
public PrometheusReporter(
- AbstractMetricManager metricManager, ScheduledExecutorService
snapshotUpdateExecutor) {
+ AbstractMetricManager metricManager,
+ Supplier<ScheduledExecutorService> snapshotUpdateExecutorSupplier) {
this.metricManager = metricManager;
- this.snapshotUpdateExecutor = snapshotUpdateExecutor;
+ this.snapshotUpdateExecutorSupplier =
Objects.requireNonNull(snapshotUpdateExecutorSupplier);
+ }
+
+ private static ScheduledExecutorService
newStandaloneSnapshotUpdateExecutor() {
+ return Executors.newSingleThreadScheduledExecutor(
+ runnable -> {
+ Thread thread = new Thread(runnable,
"prometheus-reporter-snapshot-updater");
+ thread.setDaemon(true);
+ return thread;
+ });
}
@Override
@@ -176,14 +192,10 @@ public class PrometheusReporter implements Reporter {
@SuppressWarnings("unsafeThreadSchedule")
private void startSnapshotUpdater() {
// Keep metric collection off Reactor HTTP threads and avoid overlapping
scrapes.
- if (snapshotUpdateExecutor == null) {
- snapshotUpdateExecutor =
- Executors.newSingleThreadScheduledExecutor(
- runnable -> {
- Thread thread = new Thread(runnable,
"prometheus-reporter-snapshot-updater");
- thread.setDaemon(true);
- return thread;
- });
+ if (snapshotUpdateExecutor == null || snapshotUpdateExecutor.isShutdown())
{
+ // Create a fresh executor for every start so a stopped reporter can be
started again with
+ // the same managed thread-pool factory.
+ snapshotUpdateExecutor =
Objects.requireNonNull(snapshotUpdateExecutorSupplier.get());
}
// Delay the first background scrape until metric sets have been bound by
the metric service.
snapshotUpdateFuture =
@@ -227,9 +239,10 @@ public class PrometheusReporter implements Reporter {
snapshotUpdateFuture.cancel(false);
snapshotUpdateFuture = null;
}
- if (snapshotUpdateExecutor != null) {
- snapshotUpdateExecutor.shutdownNow();
- snapshotUpdateExecutor = null;
+ ScheduledExecutorService executor = snapshotUpdateExecutor;
+ snapshotUpdateExecutor = null;
+ if (executor != null) {
+ executor.shutdownNow();
}
}
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
new file mode 100644
index 00000000000..7072777ac66
--- /dev/null
+++
b/iotdb-core/metrics/interface/src/test/java/org/apache/iotdb/metrics/reporter/prometheus/PrometheusReporterTest.java
@@ -0,0 +1,75 @@
+/*
+ * 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.reporter.prometheus;
+
+import org.apache.iotdb.metrics.config.MetricConfig;
+import org.apache.iotdb.metrics.config.MetricConfigDescriptor;
+import org.apache.iotdb.metrics.impl.DoNothingMetricManager;
+
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+public class PrometheusReporterTest {
+
+ @Test
+ public void testManagedExecutorRecreatedAfterRestart() {
+ MetricConfig metricConfig =
MetricConfigDescriptor.getInstance().getMetricConfig();
+ boolean originalAsyncUpdate =
metricConfig.isPrometheusReporterAsyncUpdate();
+ Integer originalPort = metricConfig.getPrometheusReporterPort();
+ metricConfig.setPrometheusReporterAsyncUpdate(true);
+ metricConfig.setPrometheusReporterPort(0);
+
+ AtomicInteger factoryCalls = new AtomicInteger();
+ List<ScheduledExecutorService> executors = new ArrayList<>();
+ PrometheusReporter reporter =
+ new PrometheusReporter(
+ new DoNothingMetricManager(),
+ () -> {
+ factoryCalls.incrementAndGet();
+ ScheduledExecutorService executor =
Executors.newSingleThreadScheduledExecutor();
+ executors.add(executor);
+ return executor;
+ });
+ try {
+ assertTrue(reporter.start());
+ assertEquals(1, factoryCalls.get());
+ assertTrue(reporter.stop());
+ assertTrue(executors.get(0).isShutdown());
+
+ assertTrue(reporter.start());
+ assertEquals(2, factoryCalls.get());
+ assertTrue(reporter.stop());
+ assertTrue(executors.get(1).isShutdown());
+ } finally {
+ reporter.stop();
+ metricConfig.setPrometheusReporterAsyncUpdate(originalAsyncUpdate);
+ metricConfig.setPrometheusReporterPort(originalPort);
+ executors.forEach(ScheduledExecutorService::shutdownNow);
+ }
+ }
+}
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/MetricService.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/MetricService.java
index a54d4ca483d..4af78e3d9cf 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/MetricService.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/MetricService.java
@@ -77,11 +77,14 @@ public class MetricService extends AbstractMetricService
implements MetricServic
break;
case PROMETHEUS:
if (METRIC_CONFIG.isPrometheusReporterAsyncUpdate()) {
+ // Defer pool creation until start so duplicate reporters rejected
below do not
+ // register an unused pool.
reporter =
new PrometheusReporter(
metricManager,
- IoTDBThreadPoolFactory.newSingleThreadScheduledExecutor(
-
ThreadName.PROMETHEUS_REPORTER_SNAPSHOT_UPDATER.getName()));
+ () ->
+
IoTDBThreadPoolFactory.newSingleThreadScheduledExecutor(
+
ThreadName.PROMETHEUS_REPORTER_SNAPSHOT_UPDATER.getName()));
} else {
reporter = new PrometheusReporter(metricManager);
}
