Repository: dropwizard/metrics
Pull request #4479 (2024-10-03 17:14:40)

Add check in ScheduledReporter's stop method
Refs #4246
Refs #4135


Patch:
diff --git a/metrics-core/src/main/java/io/dropwizard/metrics5/EWMA.java b/metrics-core/src/main/java/io/dropwizard/metrics5/EWMA.java
index 7c79ac71f..0b0d11b95 100644
--- a/metrics-core/src/main/java/io/dropwizard/metrics5/EWMA.java
+++ b/metrics-core/src/main/java/io/dropwizard/metrics5/EWMA.java
@@ -81,6 +81,14 @@ public class EWMA {
         uncounted.add(n);
     }
 
+    /**
+     * Set the rate to the smallest possible positive value. Used to avoid calling tick a large number of times.
+     */
+    public void reset() {
+        uncounted.reset();
+        rate = Double.MIN_NORMAL;
+    }
+
     /**
      * Mark the passage of time and decay the current rate accordingly.
      */
diff --git a/metrics-core/src/main/java/io/dropwizard/metrics5/ExponentialMovingAverages.java b/metrics-core/src/main/java/io/dropwizard/metrics5/ExponentialMovingAverages.java
index f18126e49..085981c55 100644
--- a/metrics-core/src/main/java/io/dropwizard/metrics5/ExponentialMovingAverages.java
+++ b/metrics-core/src/main/java/io/dropwizard/metrics5/ExponentialMovingAverages.java
@@ -12,8 +12,28 @@ import java.util.concurrent.atomic.LongAdder;
  */
 public class ExponentialMovingAverages implements MovingAverages {
 
+    /**
+     * If ticking would reduce even Long.MAX_VALUE in the 15 minute EWMA below this target then don't bother
+     * ticking in a loop and instead reset all the EWMAs.
+     */
+    private static final double maxTickZeroTarget = 0.0001;
+    private static final int maxTicks;
     private static final long TICK_INTERVAL = TimeUnit.SECONDS.toNanos(5);
 
+    static
+    {
+        int m3Ticks = 1;
+        final EWMA m3 = EWMA.fifteenMinuteEWMA();
+        m3.update(Long.MAX_VALUE);
+        do
+        {
+            m3.tick();
+            m3Ticks++;
+        }
+        while (m3.getRate(TimeUnit.SECONDS) > maxTickZeroTarget);
+        maxTicks = m3Ticks;
+    }
+
     private final EWMA m1Rate = EWMA.oneMinuteEWMA();
     private final EWMA m5Rate = EWMA.fiveMinuteEWMA();
     private final EWMA m15Rate = EWMA.fifteenMinuteEWMA();
@@ -54,10 +74,19 @@ public class ExponentialMovingAverages implements MovingAverages {
             if (lastTick.compareAndSet(oldTick, newIntervalStartTick)) {
                 sum.add(age);
                 final long requiredTicks = age / TICK_INTERVAL;
-                for (long i = 0; i < requiredTicks; i++) {
-                    m1Rate.tick();
-                    m5Rate.tick();
-                    m15Rate.tick();
+                if (requiredTicks >= maxTicks) {
+                    m1Rate.reset();
+                    m5Rate.reset();
+                    m15Rate.reset();
+                }
+                else
+                {
+                    for (long i = 0; i < requiredTicks; i++)
+                    {
+                        m1Rate.tick();
+                        m5Rate.tick();
+                        m15Rate.tick();
+                    }
                 }
             }
         }
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
diff --git a/metrics-core/src/main/java/io/dropwizard/metrics5/ScheduledReporter.java b/metrics-core/src/main/java/io/dropwizard/metrics5/ScheduledReporter.java
index f25257c41..a3ed5548a 100644
--- a/metrics-core/src/main/java/io/dropwizard/metrics5/ScheduledReporter.java
+++ b/metrics-core/src/main/java/io/dropwizard/metrics5/ScheduledReporter.java
@@ -220,10 +220,13 @@ public abstract class ScheduledReporter implements Closeable, Reporter {
             executor.shutdown(); // Disable new tasks from being submitted
         }
 
-        try {
-            report(); // Report metrics one last time
-        } catch (Exception e) {
-            LOG.warn("Final reporting of metrics failed.", e);
+        if (this.scheduledFuture != null) {
+            // Reporter started, try to report metrics one last time
+            try {
+                report();
+            } catch (Exception e) {
+                LOG.warn("Final reporting of metrics failed.", e);
+            }
         }
 
         if (shutdownExecutorOnStop) {


Test patch:
diff --git a/metrics-core/src/test/java/io/dropwizard/metrics5/ExponentialMovingAveragesTest.java b/metrics-core/src/test/java/io/dropwizard/metrics5/ExponentialMovingAveragesTest.java
new file mode 100644
index 000000000..7d6bbbd60
--- /dev/null
+++ b/metrics-core/src/test/java/io/dropwizard/metrics5/ExponentialMovingAveragesTest.java
@@ -0,0 +1,24 @@
+package io.dropwizard.metrics5;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+public class ExponentialMovingAveragesTest {
+    @Test
+    public void testMaxTicks() {
+        final Clock clock = mock(Clock.class);
+        when(clock.getTick()).thenReturn(0L, Long.MAX_VALUE);
+        final ExponentialMovingAverages ema = new ExponentialMovingAverages(clock);
+        ema.update(Long.MAX_VALUE);
+        ema.tickIfNecessary();
+        final long secondNanos = TimeUnit.SECONDS.toNanos(1);
+        assertEquals(ema.getM1Rate(), Double.MIN_NORMAL * secondNanos, 0.0);
+        assertEquals(ema.getM5Rate(), Double.MIN_NORMAL * secondNanos, 0.0);
+        assertEquals(ema.getM15Rate(), Double.MIN_NORMAL * secondNanos, 0.0);
+    }
+}
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")
+        );
+    }
 }
diff --git a/metrics-graphite/src/test/java/io/dropwizard/metrics5/graphite/GraphiteReporterTest.java b/metrics-graphite/src/test/java/io/dropwizard/metrics5/graphite/GraphiteReporterTest.java
index 492456c6b..4d96f3c09 100644
--- a/metrics-graphite/src/test/java/io/dropwizard/metrics5/graphite/GraphiteReporterTest.java
+++ b/metrics-graphite/src/test/java/io/dropwizard/metrics5/graphite/GraphiteReporterTest.java
@@ -458,6 +458,7 @@ class GraphiteReporterTest {
 
     @Test
     void closesConnectionOnReporterStop() throws Exception {
+        reporter.start(1, TimeUnit.SECONDS);
         reporter.stop();
 
         final InOrder inOrder = inOrder(graphite);