Repository: dropwizard/metrics
Pull request #4478 (2024-10-03 17:09:50)

Clean up InstrumentedExecutorService metrics after shutdown
Refs #3202
Refs #2920


Patch:
diff --git a/metrics-core/src/main/java/io/dropwizard/metrics5/InstrumentedExecutorService.java b/metrics-core/src/main/java/io/dropwizard/metrics5/InstrumentedExecutorService.java
index 8ba76104f..23be352ed 100644
--- a/metrics-core/src/main/java/io/dropwizard/metrics5/InstrumentedExecutorService.java
+++ b/metrics-core/src/main/java/io/dropwizard/metrics5/InstrumentedExecutorService.java
@@ -25,6 +25,8 @@ public class InstrumentedExecutorService implements ExecutorService {
     private static final AtomicLong NAME_COUNTER = new AtomicLong();
 
     private final ExecutorService delegate;
+    private final MetricRegistry registry;
+    private final String name;
     private final Meter submitted;
     private final Counter running;
     private final Meter completed;
@@ -50,12 +52,18 @@ public class InstrumentedExecutorService implements ExecutorService {
      */
     public InstrumentedExecutorService(ExecutorService delegate, MetricRegistry registry, String name) {
         this.delegate = delegate;
+        this.registry = registry;
+        this.name = name;
         this.submitted = registry.meter(MetricRegistry.name(name, "submitted"));
         this.running = registry.counter(MetricRegistry.name(name, "running"));
         this.completed = registry.meter(MetricRegistry.name(name, "completed"));
         this.idle = registry.timer(MetricRegistry.name(name, "idle"));
         this.duration = registry.timer(MetricRegistry.name(name, "duration"));
 
+        registerInternalMetrics();
+    }
+
+    private void registerInternalMetrics() {
         if (delegate instanceof ThreadPoolExecutor) {
             ThreadPoolExecutor executor = (ThreadPoolExecutor) delegate;
             registry.registerGauge(MetricRegistry.name(name, "pool.size"),
@@ -86,6 +94,23 @@ public class InstrumentedExecutorService implements ExecutorService {
         }
     }
 
+    private void removeInternalMetrics() {
+        if (delegate instanceof ThreadPoolExecutor) {
+            registry.remove(MetricRegistry.name(name, "pool.size"));
+            registry.remove(MetricRegistry.name(name, "pool.core"));
+            registry.remove(MetricRegistry.name(name, "pool.max"));
+            registry.remove(MetricRegistry.name(name, "tasks.active"));
+            registry.remove(MetricRegistry.name(name, "tasks.completed"));
+            registry.remove(MetricRegistry.name(name, "tasks.queued"));
+            registry.remove(MetricRegistry.name(name, "tasks.capacity"));
+        } else if (delegate instanceof ForkJoinPool) {
+            registry.remove(MetricRegistry.name(name, "tasks.stolen"));
+            registry.remove(MetricRegistry.name(name, "tasks.queued"));
+            registry.remove(MetricRegistry.name(name, "threads.active"));
+            registry.remove(MetricRegistry.name(name, "threads.running"));
+        }
+    }
+
     /**
      * {@inheritDoc}
      */
@@ -173,11 +198,14 @@ public class InstrumentedExecutorService implements ExecutorService {
     @Override
     public void shutdown() {
         delegate.shutdown();
+        removeInternalMetrics();
     }
 
     @Override
     public List<Runnable> shutdownNow() {
-        return delegate.shutdownNow();
+        List<Runnable> remainingTasks = delegate.shutdownNow();
+        removeInternalMetrics();
+        return remainingTasks;
     }
 
     @Override


Test patch:
diff --git a/metrics-core/src/test/java/io/dropwizard/metrics5/InstrumentedExecutorServiceTest.java b/metrics-core/src/test/java/io/dropwizard/metrics5/InstrumentedExecutorServiceTest.java
index 1c66c409b..29a750cf8 100644
--- a/metrics-core/src/test/java/io/dropwizard/metrics5/InstrumentedExecutorServiceTest.java
+++ b/metrics-core/src/test/java/io/dropwizard/metrics5/InstrumentedExecutorServiceTest.java
@@ -10,7 +10,6 @@ import java.time.Duration;
 import java.util.concurrent.Callable;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
-import java.util.concurrent.ForkJoinPool;
 import java.util.concurrent.Future;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.ThreadPoolExecutor;
@@ -167,6 +166,36 @@ class InstrumentedExecutorServiceTest {
         assertThat(poolSize.getValue()).isEqualTo(1);
     }
 
+    @Test
+    public void removesMetricsAfterShutdownForThreadPoolExecutor() {
+        executor = new ThreadPoolExecutor(4, 16,
+                0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(32));
+        instrumentedExecutorService = new InstrumentedExecutorService(executor, registry, "stp");
+
+        MetricName prefix = MetricName.build("stp");
+        assertThat(registry.getMetrics()).containsKeys(
+                MetricRegistry.name("stp", "pool.size"),
+                MetricRegistry.name("stp", "pool.core"),
+                MetricRegistry.name("stp", "pool.max"),
+                MetricRegistry.name("stp", "tasks.active"),
+                MetricRegistry.name("stp", "tasks.completed"),
+                MetricRegistry.name("stp", "tasks.queued"),
+                MetricRegistry.name("stp", "tasks.capacity")
+        );
+
+        instrumentedExecutorService.shutdown();
+
+        assertThat(registry.getMetrics()).doesNotContainKeys(
+                MetricRegistry.name("stp", "pool.size"),
+                MetricRegistry.name("stp", "pool.core"),
+                MetricRegistry.name("stp", "pool.max"),
+                MetricRegistry.name("stp", "tasks.active"),
+                MetricRegistry.name("stp", "tasks.completed"),
+                MetricRegistry.name("stp", "tasks.queued"),
+                MetricRegistry.name("stp", "tasks.capacity")
+        );
+    }
+
     @Test
     @SuppressWarnings("unchecked")
     void reportsTasksInformationForForkJoinPool() throws Exception {
@@ -215,4 +244,26 @@ class InstrumentedExecutorServiceTest {
         assertThat(idle.getCount()).isEqualTo(1);
         assertThat(idle.getSnapshot().size()).isEqualTo(1);
     }
+
+    @Test
+    public void removesMetricsAfterShutdownForForkJoinPool() {
+        executor = Executors.newWorkStealingPool(4);
+        instrumentedExecutorService = new InstrumentedExecutorService(executor, registry, "sfjp");
+
+        assertThat(registry.getMetrics()).containsKeys(
+                MetricRegistry.name("sfjp", "tasks.stolen"),
+                MetricRegistry.name("sfjp", "tasks.queued"),
+                MetricRegistry.name("sfjp", "threads.active"),
+                MetricRegistry.name("sfjp", "threads.running")
+        );
+
+        instrumentedExecutorService.shutdown();
+
+        assertThat(registry.getMetrics()).doesNotContainKeys(
+                MetricRegistry.name("sfjp", "tasks.stolen"),
+                MetricRegistry.name("sfjp", "tasks.queued"),
+                MetricRegistry.name("sfjp", "threads.active"),
+                MetricRegistry.name("sfjp", "threads.running")
+        );
+    }
 }