Upgrade metrics library and remove depricated metrics

Patch by tjake; reviewed by aleksey for CASSANDRA-5657

Author: tjake

CHANGES.txt

@@ -1,4 +1,5 @@
 3.0
+ * Upgrade Metrics library and remove depricated metrics (CASSANDRA-5657)
  * Serializing Row cache alternative, fully off heap (CASSANDRA-7438)
  * Duplicate rows returned when in clause has repeated values (CASSANDRA-6707)
  * Make CassandraException unchecked, extend RuntimeException (CASSANDRA-8560)

build.xml

@@ -368,7 +368,7 @@
 
           <dependency groupId="org.apache.cassandra" artifactId="cassandra-all" version="${version}" />
           <dependency groupId="org.apache.cassandra" artifactId="cassandra-thrift" version="${version}" />
-          <dependency groupId="com.yammer.metrics" artifactId="metrics-core" version="2.2.0" />
+          <dependency groupId="io.dropwizard.metrics" artifactId="metrics-core" version="3.1.0" />
           <dependency groupId="com.addthis.metrics" artifactId="reporter-config" version="2.1.0" />
           <dependency groupId="org.mindrot" artifactId="jbcrypt" version="0.3m" />
           <dependency groupId="io.airlift" artifactId="airline" version="0.6" />
@@ -483,7 +483,7 @@
         <dependency groupId="com.boundary" artifactId="high-scale-lib"/>
         <dependency groupId="org.yaml" artifactId="snakeyaml"/>
         <dependency groupId="org.mindrot" artifactId="jbcrypt"/>
-        <dependency groupId="com.yammer.metrics" artifactId="metrics-core"/>
+        <dependency groupId="io.dropwizard.metrics" artifactId="metrics-core"/>
         <dependency groupId="com.addthis.metrics" artifactId="reporter-config"/>
         <dependency groupId="com.thinkaurelius.thrift" artifactId="thrift-server" version="0.3.5"/>
         <dependency groupId="com.clearspring.analytics" artifactId="stream" version="2.5.2" />

lib/licenses/metrics-core-2.2.0.txt renamed to lib/licenses/metrics-core-3.1.0.txt

@@ -107,12 +107,12 @@ public long getPendingTasks()
 
     public int getTotalBlockedTasks()
     {
-        return (int) metrics.totalBlocked.count();
+        return (int) metrics.totalBlocked.getCount();
     }
 
     public int getCurrentlyBlockedTasks()
     {
-        return (int) metrics.currentBlocked.count();
+        return (int) metrics.currentBlocked.getCount();
     }
 
     public int getCoreThreads()

lib/metrics-core-2.2.0.jar

@@ -36,7 +36,7 @@
 public class JMXEnabledThreadPoolExecutor extends DebuggableThreadPoolExecutor implements JMXEnabledThreadPoolExecutorMBean
 {
     private final String mbeanName;
-    private final ThreadPoolMetrics metrics;
+    public final ThreadPoolMetrics metrics;
 
     public JMXEnabledThreadPoolExecutor(String threadPoolName)
     {
@@ -132,30 +132,17 @@ public synchronized List<Runnable> shutdownNow()
         return super.shutdownNow();
     }
 
-    /**
-     * Get the number of completed tasks
-     */
-    public long getCompletedTasks()
-    {
-        return getCompletedTaskCount();
-    }
 
-    /**
-     * Get the number of tasks waiting to be executed
-     */
-    public long getPendingTasks()
-    {
-        return getTaskCount() - getCompletedTaskCount();
-    }
+
 
     public int getTotalBlockedTasks()
     {
-        return (int) metrics.totalBlocked.count();
+        return (int) metrics.totalBlocked.getCount();
     }
 
     public int getCurrentlyBlockedTasks()
     {
-        return (int) metrics.currentBlocked.count();
+        return (int) metrics.currentBlocked.getCount();
     }
 
     public int getCoreThreads()

lib/metrics-core-3.1.0.jar

@@ -17,24 +17,9 @@
  */
 package org.apache.cassandra.concurrent;
 
-/**
- * @see org.apache.cassandra.metrics.ThreadPoolMetrics
- */
-@Deprecated
-public interface JMXEnabledThreadPoolExecutorMBean extends IExecutorMBean
-{
-    /**
-     * Get the number of tasks that had blocked before being accepted (or
-     * rejected).
-     */
-    public int getTotalBlockedTasks();
-
-    /**
-     * Get the number of tasks currently blocked, waiting to be accepted by
-     * the executor (because all threads are busy and the backing queue is full).
-     */
-    public int getCurrentlyBlockedTasks();
 
+public interface JMXEnabledThreadPoolExecutorMBean
+{
     /**
      * Returns core pool size of thread pool.
      */

src/java/org/apache/cassandra/concurrent/JMXEnabledScheduledThreadPoolExecutor.java

@@ -24,6 +24,7 @@
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
 
+import org.apache.cassandra.metrics.SEPMetrics;
 import org.apache.cassandra.utils.concurrent.SimpleCondition;
 import org.apache.cassandra.utils.concurrent.WaitQueue;
 
@@ -35,6 +36,7 @@ public class SEPExecutor extends AbstractTracingAwareExecutorService
 
     public final int maxWorkers;
     private final int maxTasksQueued;
+    private final SEPMetrics metrics;
 
     // stores both a set of work permits and task permits:
     //  bottom 32 bits are number of queued tasks, in the range [0..maxTasksQueued]   (initially 0)
@@ -43,8 +45,6 @@ public class SEPExecutor extends AbstractTracingAwareExecutorService
 
     // producers wait on this when there is no room on the queue
     private final WaitQueue hasRoom = new WaitQueue();
-    private final AtomicLong totalBlocked = new AtomicLong();
-    private final AtomicInteger currentlyBlocked = new AtomicInteger();
     private final AtomicLong completedTasks = new AtomicLong();
 
     volatile boolean shuttingDown = false;
@@ -53,12 +53,13 @@ public class SEPExecutor extends AbstractTracingAwareExecutorService
     // TODO: see if other queue implementations might improve throughput
     protected final ConcurrentLinkedQueue<FutureTask<?>> tasks = new ConcurrentLinkedQueue<>();
 
-    SEPExecutor(SharedExecutorPool pool, int maxWorkers, int maxTasksQueued)
+    SEPExecutor(SharedExecutorPool pool, int maxWorkers, int maxTasksQueued, String jmxPath, String name)
     {
         this.pool = pool;
         this.maxWorkers = maxWorkers;
         this.maxTasksQueued = maxTasksQueued;
         this.permits.set(combine(0, maxWorkers));
+        this.metrics = new SEPMetrics(this, jmxPath, name);
     }
 
     protected void onCompletion()
@@ -116,10 +117,11 @@ else if (taskPermits >= maxTasksQueued)
                 // if we're blocking, we might as well directly schedule a worker if we aren't already at max
                 if (takeWorkPermit(true))
                     pool.schedule(new Work(this));
-                totalBlocked.incrementAndGet();
-                currentlyBlocked.incrementAndGet();
+
+                metrics.totalBlocked.inc();
+                metrics.currentBlocked.inc();
                 s.awaitUninterruptibly();
-                currentlyBlocked.decrementAndGet();
+                metrics.currentBlocked.dec();
             }
             else // don't propagate our signal when we cancel, just cancel
                 s.cancel();
@@ -207,6 +209,9 @@ public synchronized void shutdown()
         pool.executors.remove(this);
         if (getActiveCount() == 0)
             shutdown.signalAll();
+
+        // release metrics
+        metrics.release();
     }
 
     public synchronized List<Runnable> shutdownNow()
@@ -249,21 +254,6 @@ public int getActiveCount()
         return maxWorkers - workPermits(permits.get());
     }
 
-    public int getTotalBlockedTasks()
-    {
-        return (int) totalBlocked.get();
-    }
-
-    public int getMaximumThreads()
-    {
-        return maxWorkers;
-    }
-
-    public int getCurrentlyBlockedTasks()
-    {
-        return currentlyBlocked.get();
-    }
-
     private static int taskPermits(long both)
     {
         return (int) both;

src/java/org/apache/cassandra/concurrent/JMXEnabledSharedExecutorPool.java

@@ -54,6 +54,8 @@
 public class SharedExecutorPool
 {
 
+    public static final SharedExecutorPool SHARED = new SharedExecutorPool("SharedPool");
+
     // the name assigned to workers in the pool, and the id suffix
     final String poolName;
     final AtomicLong workerId = new AtomicLong();
@@ -100,4 +102,11 @@ void maybeStartSpinningWorker()
         if (current == 0 && spinningCount.compareAndSet(0, 1))
             schedule(Work.SPINNING);
     }
+
+    public TracingAwareExecutorService newExecutor(int maxConcurrency, int maxQueuedTasks, String jmxPath, String name)
+    {
+        SEPExecutor executor = new SEPExecutor(this, maxConcurrency, maxQueuedTasks, jmxPath, name);
+        executors.add(executor);
+        return executor;
+    }
 }

src/java/org/apache/cassandra/concurrent/JMXEnabledThreadPoolExecutor.java

@@ -17,6 +17,11 @@
  */
 package org.apache.cassandra.concurrent;
 
+import java.util.Arrays;
+
+import com.google.common.base.Predicate;
+import com.google.common.collect.Iterables;
+
 public enum Stage
 {
     READ,
@@ -31,6 +36,17 @@ public enum Stage
     INTERNAL_RESPONSE,
     READ_REPAIR;
 
+    public static Iterable<Stage> jmxEnabledStages()
+    {
+        return Iterables.filter(Arrays.asList(values()), new Predicate<Stage>()
+        {
+            public boolean apply(Stage stage)
+            {
+                return stage != TRACING;
+            }
+        });
+    }
+
     public String getJmxType()
     {
         switch (this)

src/java/org/apache/cassandra/concurrent/JMXEnabledThreadPoolExecutorMBean.java

@@ -89,7 +89,7 @@ private static JMXEnabledThreadPoolExecutor multiThreadedStage(Stage stage, int
 
     private static TracingAwareExecutorService multiThreadedLowSignalStage(Stage stage, int numThreads)
     {
-        return JMXEnabledSharedExecutorPool.SHARED.newExecutor(numThreads, Integer.MAX_VALUE, stage.getJmxName(), stage.getJmxType());
+        return SharedExecutorPool.SHARED.newExecutor(numThreads, Integer.MAX_VALUE, stage.getJmxType(), stage.getJmxName());
     }
 
     /**