Fix Prometheus scheduler lifecycle and duplicate allocation
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 fa5695e..00ae6c6 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.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 @@ 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 @@ /** * 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 @@ @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 @@ 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 0000000..7072777 --- /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 a54d4ca..4af78e3 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 @@ 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); }