ZOOKEEPER-3143: Pluggable metrics system for ZooKeeper - Data Collection on Server

- Make ServerMetrics a class, not an enum.
- Make ServerMetrics use MetricsProvider
- After the bootstrap of the MetricsProvider reconfigure ServerMetrics in order to push data to the provider.
- Introduce a default implementation of MetricsProvider, based on current implementation
- Change MetricsContext interface in order to support several types of Metrics, in order to cover current metrics facilities,

The MetricsProvider API will allow ZooKeeper users to integrate ZooKeeper with other metrics systems, like Prometheus.io/Dropwizard.....

Author: Enrico Olivelli <eolivelli@apache.org>

Reviewers: fangmin@apache.org, andor@apache.org

Closes #854 from eolivelli/fix/metrics-provider-port-static and squashes the following commits:

c5714ee66 [Enrico Olivelli] Fix dump and reset for SummarySets
ed6230e51 [Enrico Olivelli] Fix build
8e8d87bdd [Enrico Olivelli] Merge branch 'master' into fix/metrics-provider-port-static
02722b55b [Enrico Olivelli] Fix tests
7ebc091e6 [Enrico Olivelli] Use toString in Monitor Command
49a5c801a [Enrico Olivelli] Add Metrics to 4lw interface - mntr command
137bda23a [Enrico Olivelli] A few nits
80396915f [Enrico Olivelli] Introduce MetricsUtils
9f28747bc [Enrico Olivelli] Fix after merge master
e380607ae [Enrico Olivelli] Fix imports
c2d0c358c [Enrico Olivelli] Introduce SummarySet
4674d2350 [Enrico Olivelli] Fix trailing spaces
6580b4a53 [Enrico Olivelli] Use a global static variable
ad8d44241 [Enrico Olivelli] Make DefaultMetricsProvider the default
d455a1c8c [Enrico Olivelli] Fix test
f997c30f3 [Enrico Olivelli] Fix spotbugs
228c1d268 [Enrico Olivelli] ZOOKEEPER-3143 Pluggable metrics system for ZooKeeper - Data Collection on Server

zookeeper-server/src/main/java/org/apache/zookeeper/metrics/Counter.java

@@ -29,7 +29,7 @@ public interface Counter {
      * <p>This method is thread safe, The MetricsProvider will take care of synchronization.</p>
      */
     default void inc() {
-        inc(1);
+        add(1);
     }
 
     /**
@@ -38,7 +38,7 @@ public interface Counter {
      *
      * @param delta amount to increment, this cannot be a negative number.
      */
-    void inc(long delta);
+    void add(long delta);
 
     /**
      * Get the current value held by the counter.

zookeeper-server/src/main/java/org/apache/zookeeper/metrics/MetricsContext.java

@@ -15,15 +15,15 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-
 package org.apache.zookeeper.metrics;
 
 /**
- * A MetricsContext is like a namespace for metrics.
- * Each component/submodule will have its own MetricsContext.
- * <p>In some cases it is possible to have a separate MetricsContext
- * for each instance of a component, for instance on the server side
- * a possible usecase it to gather metrics for every other peer.
+ * A MetricsContext is like a namespace for metrics. Each component/submodule
+ * will have its own MetricsContext.
+ * <p>
+ * In some cases it is possible to have a separate MetricsContext for each
+ * instance of a component, for instance on the server side a possible usecase
+ * it to gather metrics for every other peer.
  * </p>
  * <p>
  * Contexts are organized in a hierarchy.
@@ -50,22 +50,48 @@ public interface MetricsContext {
     Counter getCounter(String name);
 
     /**
-     * Registers an user provided {@link Gauge} which will be called by the MetricsProvider in order to sample
-     * an integer value.
+     * Registers an user provided {@link Gauge} which will be called by the
+     * MetricsProvider in order to sample an integer value.
      *
      * @param name unique name of the Gauge in this context
      * @param gauge the implementation of the Gauge
      *
-     * @return true if the Gauge was successfully registered, false if the Gauge was already registered.
+     * @return true if the Gauge was successfully registered, false if the Gauge
+     * was already registered.
      */
     boolean registerGauge(String name, Gauge gauge);
 
+    static enum DetailLevel {
+        /**
+         * The returned Summary is expected to track only simple aggregated
+         * values, like min/max/avg
+         */
+        BASIC,
+        /**
+         * It is expected that the returned Summary performs expensive
+         * aggregations, like percentiles.
+         */
+        ADVANCED
+    }
+
     /**
      * Returns a summary.
      *
      * @param name
+     * @param detailLevel
+     * @return the summary identified by name in this context.
+     * @see #getSummary(java.lang.String)
+     */
+    Summary getSummary(String name, DetailLevel detailLevel);
+
+    /**
+     * Returns a set of summaries.
+     *
+     * @param name
+     * @param detailLevel
      * @return the summary identified by name in this context.
+     * @see #getSummary(java.lang.String)
      */
-    Summary getSummary(String name);
+    SummarySet getSummarySet(String name, DetailLevel detailLevel);
 
 }

zookeeper-server/src/main/java/org/apache/zookeeper/metrics/MetricsProvider.java

@@ -19,6 +19,7 @@
 package org.apache.zookeeper.metrics;
 
 import java.util.Properties;
+import java.util.function.BiConsumer;
 
 /**
  * A MetricsProvider is a system which collects Metrics and publishes current values to external facilities.
@@ -61,4 +62,18 @@ public interface MetricsProvider {
      * This method can be called more than once.
      */
     void stop();
+
+    /**
+     * Dumps all metrics as a key-value pair.
+     * This method will be used in legacy monitor command.
+     * @param sink the receiver of all of the current values.
+     */
+    public void dump(BiConsumer<String, Object> sink);
+
+    /**
+     * Reset all values.
+     * This method is optional and can be noop, depending
+     * on the underlying implementation.
+     */
+    public void resetAllValues();
 }

zookeeper-server/src/main/java/org/apache/zookeeper/metrics/Summary.java

@@ -30,6 +30,6 @@ public interface Summary {
       *
       * @param value current value
       */
-     void registerValue(long value);
+     void add(long value);
 
 }

zookeeper-server/src/main/java/org/apache/zookeeper/metrics/SummarySet.java

@@ -0,0 +1,37 @@
+/**
+ * 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.zookeeper.metrics;
+
+/**
+ * Summaries track the size and number of events.
+ * They are able to publish minumum, maximum, average values, depending on the capabilities of the MetricsProvider.
+ * A SummarySet is a set of {@link Summary}.
+ */
+public interface SummarySet {
+
+     /**
+      * Register a value.
+      * <p>This method is thread safe, The MetricsProvider will take care of synchronization.</p>
+      *
+      * @param key the key to access the Summary for the given key
+      * @param value current value
+      */
+     void add(String key, long value);
+
+}

zookeeper-server/src/main/java/org/apache/zookeeper/metrics/impl/DefaultMetricsProvider.java

@@ -0,0 +1,176 @@
+/**
+ * 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.zookeeper.metrics.impl;
+
+import java.util.Properties;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.function.BiConsumer;
+import org.apache.zookeeper.metrics.Counter;
+import org.apache.zookeeper.metrics.Gauge;
+import org.apache.zookeeper.metrics.MetricsContext;
+import org.apache.zookeeper.metrics.MetricsProvider;
+import org.apache.zookeeper.metrics.MetricsProviderLifeCycleException;
+import org.apache.zookeeper.metrics.Summary;
+import org.apache.zookeeper.metrics.SummarySet;
+import org.apache.zookeeper.server.metric.AvgMinMaxCounter;
+import org.apache.zookeeper.server.metric.AvgMinMaxCounterSet;
+import org.apache.zookeeper.server.metric.AvgMinMaxPercentileCounter;
+import org.apache.zookeeper.server.metric.AvgMinMaxPercentileCounterSet;
+import org.apache.zookeeper.server.metric.SimpleCounter;
+
+/**
+ * Default implementation of {@link MetricsProvider}.<br>
+ * It does not implement a real hierarchy of contexts, but metrics are flattened
+ * in a single namespace.<br>
+ * It is mostly useful to make the legacy 4 letter words interface work as
+ * expected.
+ */
+public class DefaultMetricsProvider implements MetricsProvider {
+
+    private final DefaultMetricsContext rootMetricsContext = new DefaultMetricsContext();
+
+    @Override
+    public void configure(Properties configuration) throws MetricsProviderLifeCycleException {
+    }
+
+    @Override
+    public void start() throws MetricsProviderLifeCycleException {
+    }
+
+    @Override
+    public MetricsContext getRootContext() {
+        return rootMetricsContext;
+    }
+
+    @Override
+    public void stop() {
+    }
+
+    @Override
+    public void dump(BiConsumer<String, Object> sink) {
+        rootMetricsContext.dump(sink);
+    }
+
+    @Override
+    public void resetAllValues() {
+        rootMetricsContext.reset();
+    }
+
+    private static final class DefaultMetricsContext implements MetricsContext {
+
+        private final ConcurrentMap<String, SimpleCounter> counters = new ConcurrentHashMap<>();
+        private final ConcurrentMap<String, AvgMinMaxCounter> basicSummaries = new ConcurrentHashMap<>();
+        private final ConcurrentMap<String, AvgMinMaxPercentileCounter> summaries = new ConcurrentHashMap<>();
+        private final ConcurrentMap<String, AvgMinMaxCounterSet> basicSummarySets = new ConcurrentHashMap<>();
+        private final ConcurrentMap<String, AvgMinMaxPercentileCounterSet> summarySets = new ConcurrentHashMap<>();
+
+        @Override
+        public MetricsContext getContext(String name) {
+            // no hierarchy
+            return this;
+        }
+
+        @Override
+        public Counter getCounter(String name) {
+            return counters.computeIfAbsent(name, (n) -> {
+                return new SimpleCounter(n);
+            });
+        }
+
+        @Override
+        public boolean registerGauge(String name, Gauge gauge) {
+            // Not supported
+            return false;
+        }
+
+        @Override
+        public Summary getSummary(String name, DetailLevel detailLevel) {
+            if (detailLevel == DetailLevel.BASIC) {
+                return basicSummaries.computeIfAbsent(name, (n) -> {
+                    if (summaries.containsKey(n)) {
+                        throw new IllegalArgumentException("Already registered a non basic summary as " + n);
+                    }
+                    return new AvgMinMaxCounter(name);
+                });
+            } else {
+                return summaries.computeIfAbsent(name, (n) -> {
+                    if (basicSummaries.containsKey(n)) {
+                        throw new IllegalArgumentException("Already registered a basic summary as " + n);
+                    }
+                    return new AvgMinMaxPercentileCounter(name);
+                });
+            }
+        }
+
+        @Override
+        public SummarySet getSummarySet(String name, DetailLevel detailLevel) {
+            if (detailLevel == DetailLevel.BASIC) {
+                return basicSummarySets.computeIfAbsent(name, (n) -> {
+                    if (summarySets.containsKey(n)) {
+                        throw new IllegalArgumentException("Already registered a non basic summary set as " + n);
+                    }
+                    return new AvgMinMaxCounterSet(name);
+                });
+            } else {
+                return summarySets.computeIfAbsent(name, (n) -> {
+                    if (basicSummarySets.containsKey(n)) {
+                        throw new IllegalArgumentException("Already registered a basic summary set as " + n);
+                    }
+                    return new AvgMinMaxPercentileCounterSet(name);
+                });
+            }
+        }
+
+        void dump(BiConsumer<String, Object> sink) {
+            counters.values().forEach(metric -> {
+                metric.values().forEach(sink);
+            });
+            basicSummaries.values().forEach(metric -> {
+                metric.values().forEach(sink);
+            });
+            summaries.values().forEach(metric -> {
+                metric.values().forEach(sink);
+            });
+            basicSummarySets.values().forEach(metric -> {
+                metric.values().forEach(sink);
+            });
+            summarySets.values().forEach(metric -> {
+                metric.values().forEach(sink);
+            });
+        }
+
+        void reset() {
+            counters.values().forEach(metric -> {
+                metric.reset();
+            });
+            basicSummaries.values().forEach(metric -> {
+                metric.reset();
+            });
+            summaries.values().forEach(metric -> {
+                metric.reset();
+            });
+            basicSummarySets.values().forEach(metric -> {
+                metric.reset();
+            });
+            summarySets.values().forEach(metric -> {
+                metric.reset();
+            });
+        }
+    }
+}

zookeeper-server/src/main/java/org/apache/zookeeper/metrics/impl/NullMetricsProvider.java

@@ -18,18 +18,25 @@
 package org.apache.zookeeper.metrics.impl;
 
 import java.util.Properties;
+import java.util.function.BiConsumer;
 import org.apache.zookeeper.metrics.Counter;
 import org.apache.zookeeper.metrics.Gauge;
 import org.apache.zookeeper.metrics.MetricsContext;
 import org.apache.zookeeper.metrics.MetricsProvider;
 import org.apache.zookeeper.metrics.MetricsProviderLifeCycleException;
 import org.apache.zookeeper.metrics.Summary;
+import org.apache.zookeeper.metrics.SummarySet;
 
 /**
  * This is a dummy MetricsProvider which does nothing.
  */
 public class NullMetricsProvider implements MetricsProvider {
 
+    /**
+     * Instance of NullMetricsProvider useful for tests.
+     */
+    public static final MetricsProvider INSTANCE = new NullMetricsProvider();
+
     @Override
     public void configure(Properties configuration) throws MetricsProviderLifeCycleException {
     }
@@ -43,6 +50,14 @@ public class NullMetricsProvider implements MetricsProvider {
         return NullMetricsContext.INSTANCE;
     }
 
+    @Override
+    public void dump(BiConsumer<String, Object> sink) {
+    }
+
+    @Override
+    public void resetAllValues() {
+    }
+
     @Override
     public void stop() {
     }
@@ -67,10 +82,15 @@ public class NullMetricsProvider implements MetricsProvider {
         }
 
         @Override
-        public Summary getSummary(String name) {
+        public Summary getSummary(String name, DetailLevel detailLevel) {
             return NullSummary.INSTANCE;
         }
 
+        @Override
+        public SummarySet getSummarySet(String name, DetailLevel detailLevel) {
+            return NullSummarySet.INSTANCE;
+        }
+
     }
 
     private static final class NullCounter implements Counter {
@@ -78,7 +98,7 @@ public class NullMetricsProvider implements MetricsProvider {
         private static final NullCounter INSTANCE = new NullCounter();
 
         @Override
-        public void inc(long delta) {
+        public void add(long delta) {
         }
 
         @Override
@@ -93,7 +113,17 @@ public class NullMetricsProvider implements MetricsProvider {
         private static final NullSummary INSTANCE = new NullSummary();
 
         @Override
-        public void registerValue(long value) {
+        public void add(long value) {
+        }
+
+    }
+
+    private static final class NullSummarySet implements SummarySet {
+
+        private static final NullSummarySet INSTANCE = new NullSummarySet();
+
+        @Override
+        public void add(String key, long value) {
         }
 
     }

zookeeper-server/src/main/java/org/apache/zookeeper/server/DataTree.java

@@ -1539,7 +1539,7 @@ public class DataTree {
             return;
         }
         long totalBytes = path.length() + bytes + STAT_OVERHEAD_BYTES;
-        ServerMetrics.READ_PER_NAMESPACE.add(namespace, totalBytes);
+        ServerMetrics.getMetrics().READ_PER_NAMESPACE.add(namespace, totalBytes);
     }
 
     private void updateWriteStat(String path, long bytes) {
@@ -1547,6 +1547,6 @@ public class DataTree {
         if (namespace == null) {
             return;
         }
-        ServerMetrics.WRITE_PER_NAMESPACE.add(namespace, path.length() + bytes);
+        ServerMetrics.getMetrics().WRITE_PER_NAMESPACE.add(namespace, path.length() + bytes);
     }
 }

zookeeper-server/src/main/java/org/apache/zookeeper/server/FinalRequestProcessor.java

@@ -168,7 +168,7 @@ public class FinalRequestProcessor implements RequestProcessor {
              */
             long propagationLatency = Time.currentWallTime() - request.getHdr().getTime();
             if (propagationLatency > 0) {
-                ServerMetrics.PROPAGATION_LATENCY.add(propagationLatency);
+                ServerMetrics.getMetrics().PROPAGATION_LATENCY.add(propagationLatency);
             }
         }
 

zookeeper-server/src/main/java/org/apache/zookeeper/server/NIOServerCnxn.java

@@ -362,7 +362,7 @@ public class NIOServerCnxn extends ServerCnxn {
             close();
         } catch (ClientCnxnLimitException e) {
             // Common case exception, print at debug level
-            ServerMetrics.CONNECTION_REJECTED.add(1);
+            ServerMetrics.getMetrics().CONNECTION_REJECTED.add(1);
             if (LOG.isDebugEnabled()) {
                 LOG.debug("Exception causing close of session 0x"
                           + Long.toHexString(sessionId) + ": " + e.getMessage());

zookeeper-server/src/main/java/org/apache/zookeeper/server/NIOServerCnxnFactory.java

@@ -310,7 +310,7 @@ public class NIOServerCnxnFactory extends ServerCnxnFactory {
                 acceptErrorLogger.flush();
             } catch (IOException e) {
                 // accept, maxClientCnxns, configureBlocking
-                ServerMetrics.CONNECTION_REJECTED.add(1);
+                ServerMetrics.getMetrics().CONNECTION_REJECTED.add(1);
                 acceptErrorLogger.rateLimitLog(
                     "Error accepting new connection: " + e.getMessage());
                 fastCloseSock(sc);

zookeeper-server/src/main/java/org/apache/zookeeper/server/NettyServerCnxn.java

@@ -531,7 +531,7 @@ public class NettyServerCnxn extends ServerCnxn {
             close();
         } catch(ClientCnxnLimitException e) {
             // Common case exception, print at debug level
-            ServerMetrics.CONNECTION_REJECTED.add(1);
+            ServerMetrics.getMetrics().CONNECTION_REJECTED.add(1);
             if (LOG.isDebugEnabled()) {
                 LOG.debug("Closing connection to " + getRemoteSocketAddress(), e);
             }

zookeeper-server/src/main/java/org/apache/zookeeper/server/NettyServerCnxnFactory.java

@@ -109,7 +109,7 @@ public class NettyServerCnxnFactory extends ServerCnxnFactory {
             InetAddress addr = ((InetSocketAddress) channel.remoteAddress())
                     .getAddress();
             if (maxClientCnxns > 0 && getClientCnxnCount(addr) >= maxClientCnxns) {
-                ServerMetrics.CONNECTION_REJECTED.add(1);
+                ServerMetrics.getMetrics().CONNECTION_REJECTED.add(1);
                 LOG.warn("Too many connections from {} - max is {}", addr,
                         maxClientCnxns);
                 channel.close();

zookeeper-server/src/main/java/org/apache/zookeeper/server/ServerCnxn.java

@@ -137,9 +137,9 @@ public abstract class ServerCnxn implements Stats, Watcher {
                     // Cache miss, serialize the response and put it in cache.
                     data = serializeRecord(r);
                     cache.put(cacheKey, data, stat);
-                    ServerMetrics.RESPONSE_PACKET_CACHE_MISSING.add(1);
+                    ServerMetrics.getMetrics().RESPONSE_PACKET_CACHE_MISSING.add(1);
                 } else {
-                    ServerMetrics.RESPONSE_PACKET_CACHE_HITS.add(1);
+                    ServerMetrics.getMetrics().RESPONSE_PACKET_CACHE_HITS.add(1);
                 }
             } else {
                 data = serializeRecord(r);
@@ -235,7 +235,7 @@ public abstract class ServerCnxn implements Stats, Watcher {
         if (serverStats != null) {
             serverStats().incrementPacketsReceived();
         }
-        ServerMetrics.BYTES_RECEIVED_COUNT.add(bytes);
+        ServerMetrics.getMetrics().BYTES_RECEIVED_COUNT.add(bytes);
     }
 
     protected void packetSent() {

zookeeper-server/src/main/java/org/apache/zookeeper/server/ServerConfig.java

@@ -24,7 +24,7 @@ import java.util.Arrays;
 import java.util.Properties;
 
 import org.apache.yetus.audience.InterfaceAudience;
-import org.apache.zookeeper.metrics.impl.NullMetricsProvider;
+import org.apache.zookeeper.metrics.impl.DefaultMetricsProvider;
 import org.apache.zookeeper.server.quorum.QuorumPeerConfig;
 import org.apache.zookeeper.server.quorum.QuorumPeerConfig.ConfigException;
 
@@ -50,7 +50,7 @@ public class ServerConfig {
     protected int minSessionTimeout = -1;
     /** defaults to -1 if not set explicitly */
     protected int maxSessionTimeout = -1;
-    protected String metricsProviderClassName = NullMetricsProvider.class.getName();
+    protected String metricsProviderClassName = DefaultMetricsProvider.class.getName();
     protected Properties metricsProviderConfiguration = new Properties();
     /** defaults to -1 if not set explicitly */
     protected int listenBacklog = -1;

zookeeper-server/src/main/java/org/apache/zookeeper/server/ServerMetrics.java

@@ -15,144 +15,201 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-
 package org.apache.zookeeper.server;
 
-import org.apache.zookeeper.server.metric.AvgMinMaxCounter;
-import org.apache.zookeeper.server.metric.AvgMinMaxCounterSet;
-import org.apache.zookeeper.server.metric.AvgMinMaxPercentileCounter;
-import org.apache.zookeeper.server.metric.AvgMinMaxPercentileCounterSet;
-import org.apache.zookeeper.server.metric.Metric;
-import org.apache.zookeeper.server.metric.SimpleCounter;
+import org.apache.zookeeper.metrics.Counter;
+import org.apache.zookeeper.metrics.MetricsContext;
+import org.apache.zookeeper.metrics.MetricsContext.DetailLevel;
+
+import org.apache.zookeeper.metrics.MetricsProvider;
+import org.apache.zookeeper.metrics.Summary;
+import org.apache.zookeeper.metrics.SummarySet;
+import org.apache.zookeeper.metrics.impl.DefaultMetricsProvider;
+import org.apache.zookeeper.metrics.impl.NullMetricsProvider;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
-import java.util.LinkedHashMap;
-import java.util.Map;
+public final class ServerMetrics {
+
+    private static final Logger LOG = LoggerFactory.getLogger(ServerMetrics.class);
+
+    /**
+     * Dummy instance useful for tests.
+     */
+    public static final ServerMetrics NULL_METRICS
+            = new ServerMetrics(NullMetricsProvider.INSTANCE);
+
+    /**
+     * Dummy instance useful for tests.
+     */
+    public static final ServerMetrics DEFAULT_METRICS_FOR_TESTS
+            = new ServerMetrics(new DefaultMetricsProvider());
+
+    /**
+     * Real instance used for tracking server side metrics. The final value is
+     * assigned after the {@link MetricsProvider} bootstrap.
+     */
+    private static volatile ServerMetrics CURRENT = DEFAULT_METRICS_FOR_TESTS;
+
+    /**
+     * Access current ServerMetrics.
+     *
+     * @return a reference to the current Metrics
+     */
+    public static ServerMetrics getMetrics() {
+        return CURRENT;
+    }
+
+    public static void metricsProviderInitialized(MetricsProvider metricsProvider) {
+        LOG.info("ServerMetrics initialized with provider {}", metricsProvider);
+        CURRENT = new ServerMetrics(metricsProvider);
+    }
+
+    private ServerMetrics(MetricsProvider metricsProvider) {
+        this.metricsProvider = metricsProvider;
+        MetricsContext metricsContext = this.metricsProvider.getRootContext();
+
+        FSYNC_TIME = metricsContext.getSummary("fsynctime", DetailLevel.BASIC);
+
+        SNAPSHOT_TIME = metricsContext.getSummary("snapshottime", DetailLevel.BASIC);
+        DB_INIT_TIME = metricsContext.getSummary("dbinittime", DetailLevel.BASIC);
+        READ_LATENCY = metricsContext.getSummary("readlatency", DetailLevel.ADVANCED);
+        UPDATE_LATENCY = metricsContext.getSummary("updatelatency", DetailLevel.ADVANCED);
+        PROPAGATION_LATENCY = metricsContext.getSummary("propagation_latency", DetailLevel.ADVANCED);
+        FOLLOWER_SYNC_TIME = metricsContext.getSummary("follower_sync_time", DetailLevel.BASIC);
+        ELECTION_TIME = metricsContext.getSummary("election_time", DetailLevel.BASIC);
+        LOOKING_COUNT = metricsContext.getCounter("looking_count");
+        DIFF_COUNT = metricsContext.getCounter("diff_count");
+        SNAP_COUNT = metricsContext.getCounter("snap_count");
+        COMMIT_COUNT = metricsContext.getCounter("commit_count");
+        CONNECTION_REQUEST_COUNT = metricsContext.getCounter("connection_request_count");
+        CONNECTION_TOKEN_DEFICIT = metricsContext.getSummary("connection_token_deficit", DetailLevel.BASIC);
+        CONNECTION_REJECTED = metricsContext.getCounter("connection_rejected");
+
+        WRITE_PER_NAMESPACE = metricsContext.getSummarySet("write_per_namespace", DetailLevel.BASIC);
+        READ_PER_NAMESPACE = metricsContext.getSummarySet("read_per_namespace", DetailLevel.BASIC);
+
+        BYTES_RECEIVED_COUNT = metricsContext.getCounter("bytes_received_count");
+        UNRECOVERABLE_ERROR_COUNT = metricsContext.getCounter("unrecoverable_error_count");
+
+        NODE_CREATED_WATCHER = metricsContext.getSummary("node_created_watch_count", DetailLevel.BASIC);
+        NODE_DELETED_WATCHER = metricsContext.getSummary("node_deleted_watch_count", DetailLevel.BASIC);
+        NODE_CHANGED_WATCHER = metricsContext.getSummary("node_changed_watch_count", DetailLevel.BASIC);
+        NODE_CHILDREN_WATCHER = metricsContext.getSummary("node_children_watch_count", DetailLevel.BASIC);
+
+
+        /*
+     * Number of dead watchers in DeadWatcherListener
+         */
+        ADD_DEAD_WATCHER_STALL_TIME = metricsContext.getCounter("add_dead_watcher_stall_time");
+        DEAD_WATCHERS_QUEUED = metricsContext.getCounter("dead_watchers_queued");
+        DEAD_WATCHERS_CLEARED = metricsContext.getCounter("dead_watchers_cleared");
+        DEAD_WATCHERS_CLEANER_LATENCY = metricsContext.getSummary("dead_watchers_cleaner_latency", DetailLevel.ADVANCED);
+
+        RESPONSE_PACKET_CACHE_HITS = metricsContext.getCounter("response_packet_cache_hits");
+        RESPONSE_PACKET_CACHE_MISSING = metricsContext.getCounter("response_packet_cache_misses");
+
+        ENSEMBLE_AUTH_SUCCESS = metricsContext.getCounter("ensemble_auth_success");
+
+        ENSEMBLE_AUTH_FAIL = metricsContext.getCounter("ensemble_auth_fail");
+
+        ENSEMBLE_AUTH_SKIP = metricsContext.getCounter("ensemble_auth_skip");
+
+    }
 
-public enum ServerMetrics {
     /**
      * Txnlog fsync time
      */
-    FSYNC_TIME(new AvgMinMaxCounter("fsynctime")),
+    public final Summary FSYNC_TIME;
 
     /**
      * Snapshot writing time
      */
-    SNAPSHOT_TIME(new AvgMinMaxCounter("snapshottime")),
+    public final Summary SNAPSHOT_TIME;
 
     /**
      * Db init time (snapshot loading + txnlog replay)
      */
-    DB_INIT_TIME(new AvgMinMaxCounter("dbinittime")),
+    public final Summary DB_INIT_TIME;
 
     /**
      * Stats for read request. The timing start from when the server see the
      * request until it leave final request processor.
      */
-    READ_LATENCY(new AvgMinMaxPercentileCounter("readlatency")),
+    public final Summary READ_LATENCY;
 
     /**
      * Stats for request that need quorum voting. Timing is the same as read
      * request. We only keep track of stats for request that originated from
      * this machine only.
      */
-    UPDATE_LATENCY(new AvgMinMaxPercentileCounter("updatelatency")),
+    public final Summary UPDATE_LATENCY;
 
     /**
-     * Stats for all quorum request. The timing start from when the leader
-     * see the request until it reach the learner.
+     * Stats for all quorum request. The timing start from when the leader see
+     * the request until it reach the learner.
      */
-    PROPAGATION_LATENCY(new AvgMinMaxPercentileCounter("propagation_latency")),
-
-    FOLLOWER_SYNC_TIME(new AvgMinMaxCounter("follower_sync_time")),
-    ELECTION_TIME(new AvgMinMaxCounter("election_time")),
-    LOOKING_COUNT(new SimpleCounter("looking_count")),
-    DIFF_COUNT(new SimpleCounter("diff_count")),
-    SNAP_COUNT(new SimpleCounter("snap_count")),
-    COMMIT_COUNT(new SimpleCounter("commit_count")),
-    CONNECTION_REQUEST_COUNT(new SimpleCounter("connection_request_count")),
-    // Connection throttling related
-    CONNECTION_TOKEN_DEFICIT(new AvgMinMaxCounter("connection_token_deficit")),
-    CONNECTION_REJECTED(new SimpleCounter("connection_rejected")),
+    public final Summary PROPAGATION_LATENCY;
 
-    UNRECOVERABLE_ERROR_COUNT(new SimpleCounter("unrecoverable_error_count")),
+    public final Summary FOLLOWER_SYNC_TIME;
 
-    WRITE_PER_NAMESPACE(new AvgMinMaxCounterSet("write_per_namespace")),
-    READ_PER_NAMESPACE(new AvgMinMaxCounterSet("read_per_namespace")),
+    public final Summary ELECTION_TIME;
 
-    BYTES_RECEIVED_COUNT(new SimpleCounter("bytes_received_count")),
+    public final Counter LOOKING_COUNT;
+    public final Counter DIFF_COUNT;
+    public final Counter SNAP_COUNT;
+    public final Counter COMMIT_COUNT;
+    public final Counter CONNECTION_REQUEST_COUNT;
+    // Connection throttling related
+    public final Summary CONNECTION_TOKEN_DEFICIT;
+    public final Counter CONNECTION_REJECTED;
+
+    public final Counter UNRECOVERABLE_ERROR_COUNT;
+    public final SummarySet WRITE_PER_NAMESPACE;
+    public final SummarySet READ_PER_NAMESPACE;
+    public final Counter BYTES_RECEIVED_COUNT;
 
     /**
      * Fired watcher stats.
      */
-    NODE_CREATED_WATCHER(new AvgMinMaxCounter("node_created_watch_count")),
-    NODE_DELETED_WATCHER(new AvgMinMaxCounter("node_deleted_watch_count")),
-    NODE_CHANGED_WATCHER(new AvgMinMaxCounter("node_changed_watch_count")),
-    NODE_CHILDREN_WATCHER(new AvgMinMaxCounter("node_children_watch_count")),
-
+    public final Summary NODE_CREATED_WATCHER;
+    public final Summary NODE_DELETED_WATCHER;
+    public final Summary NODE_CHANGED_WATCHER;
+    public final Summary NODE_CHILDREN_WATCHER;
 
     /*
      * Number of dead watchers in DeadWatcherListener
      */
-    ADD_DEAD_WATCHER_STALL_TIME(new SimpleCounter("add_dead_watcher_stall_time")),
-    DEAD_WATCHERS_QUEUED(new SimpleCounter("dead_watchers_queued")),
-    DEAD_WATCHERS_CLEARED(new SimpleCounter("dead_watchers_cleared")),
-    DEAD_WATCHERS_CLEANER_LATENCY(new AvgMinMaxPercentileCounter("dead_watchers_cleaner_latency")),
-
-    RESPONSE_PACKET_CACHE_HITS(new SimpleCounter("response_packet_cache_hits")),
-    RESPONSE_PACKET_CACHE_MISSING(new SimpleCounter("response_packet_cache_misses")),
-    
+    public final Counter ADD_DEAD_WATCHER_STALL_TIME;
+    public final Counter DEAD_WATCHERS_QUEUED;
+    public final Counter DEAD_WATCHERS_CLEARED;
+    public final Summary DEAD_WATCHERS_CLEANER_LATENCY;
+    public final Counter RESPONSE_PACKET_CACHE_HITS;
+    public final Counter RESPONSE_PACKET_CACHE_MISSING;
+
     /*
      * Number of successful matches of expected ensemble name in EnsembleAuthenticationProvider.
      */
-    ENSEMBLE_AUTH_SUCCESS(new SimpleCounter("ensemble_auth_success")),
+    public final Counter ENSEMBLE_AUTH_SUCCESS;
 
     /*
      * Number of unsuccessful matches of expected ensemble name in EnsembleAuthenticationProvider.
      */
-    ENSEMBLE_AUTH_FAIL(new SimpleCounter("ensemble_auth_fail")),
+    public final Counter ENSEMBLE_AUTH_FAIL;
 
     /*
      * Number of client auth requests with no ensemble set in EnsembleAuthenticationProvider.
      */
-    ENSEMBLE_AUTH_SKIP(new SimpleCounter("ensemble_auth_skip"));
-
-    private final Metric metric;
-
-    ServerMetrics(Metric metric) {
-        this.metric = metric;
-    }
-
-    public void add(long value) {
-        metric.add(value);
-    }
+    public final Counter ENSEMBLE_AUTH_SKIP;
 
-    public void add(int key, long value) {
-        metric.add(key, value);
-    }
-
-    public void add(String key, long value) {
-        metric.add(key, value);
-    }
+    private final MetricsProvider metricsProvider;
 
-    public void reset() {
-        metric.reset();
+    public void resetAll() {
+        metricsProvider.resetAllValues();
     }
 
-    Map<String, Object> getValues() {
-        return metric.values();
+    public MetricsProvider getMetricsProvider() {
+        return metricsProvider;
     }
 
-    static public Map<String, Object> getAllValues() {
-        LinkedHashMap<String, Object> m = new LinkedHashMap<>();
-        for (ServerMetrics metric : ServerMetrics.values()) {
-            m.putAll(metric.getValues());
-        }
-        return m;
-    }
-
-    static public void resetAll() {
-        for (ServerMetrics metric : ServerMetrics.values()) {
-            metric.reset();
-        }
-    }
 }

zookeeper-server/src/main/java/org/apache/zookeeper/server/ServerStats.java

@@ -39,14 +39,14 @@ public class ServerStats {
 
     private final AvgMinMaxCounter requestLatency = new AvgMinMaxCounter("request_latency");
 
-    private AtomicLong fsyncThresholdExceedCount = new AtomicLong(0);
+    private final AtomicLong fsyncThresholdExceedCount = new AtomicLong(0);
 
     private final BufferStats clientResponseStats = new BufferStats();
 
     private final Provider provider;
     private final long startTime = Time.currentElapsedTime();
 
-    public interface Provider {
+    public static interface Provider {
         public long getOutstandingRequests();
         public long getLastProcessedZxid();
         public String getState();
@@ -58,7 +58,7 @@ public class ServerStats {
     public ServerStats(Provider provider) {
         this.provider = provider;
     }
-    
+
     // getters
     public long getMinLatency() {
         return requestLatency.getMin();
@@ -142,10 +142,10 @@ public class ServerStats {
         requestLatency.addDataPoint(latency);
         if (request.getHdr() != null) {
             // Only quorum request should have header
-            ServerMetrics.UPDATE_LATENCY.add(latency);
+            ServerMetrics.getMetrics().UPDATE_LATENCY.add(latency);
         } else {
             // All read request should goes here
-            ServerMetrics.READ_LATENCY.add(latency);
+            ServerMetrics.getMetrics().READ_LATENCY.add(latency);
         }
     }
 
@@ -186,7 +186,7 @@ public class ServerStats {
         resetLatency();
         resetRequestCounters();
         clientResponseStats.reset();
-        ServerMetrics.resetAll();
+        ServerMetrics.getMetrics().resetAll();
     }
 
     public void updateClientResponseSize(int size) {

zookeeper-server/src/main/java/org/apache/zookeeper/server/ZKDatabase.java

@@ -255,7 +255,7 @@ public class ZKDatabase {
         long zxid = snapLog.restore(dataTree, sessionsWithTimeouts, commitProposalPlaybackListener);
         initialized = true;
         long loadTime = Time.currentElapsedTime() - startTime;
-        ServerMetrics.DB_INIT_TIME.add(loadTime);
+        ServerMetrics.getMetrics().DB_INIT_TIME.add(loadTime);
         LOG.info("Snapshot loaded in " + loadTime + " ms");
         return zxid;
     }

zookeeper-server/src/main/java/org/apache/zookeeper/server/ZooKeeperCriticalThread.java

@@ -47,6 +47,6 @@ public class ZooKeeperCriticalThread extends ZooKeeperThread {
     protected void handleException(String threadName, Throwable e) {
         LOG.error("Severe unrecoverable error, from thread : {}", threadName, e);
         listener.notifyStopping(threadName, ExitCode.UNEXPECTED_ERROR.getValue());
-        ServerMetrics.UNRECOVERABLE_ERROR_COUNT.add(1);
+        ServerMetrics.getMetrics().UNRECOVERABLE_ERROR_COUNT.add(1);
     }
 }

zookeeper-server/src/main/java/org/apache/zookeeper/server/ZooKeeperServer.java

@@ -53,8 +53,6 @@ import org.apache.zookeeper.data.ACL;
 import org.apache.zookeeper.data.Id;
 import org.apache.zookeeper.data.StatPersisted;
 import org.apache.zookeeper.jmx.MBeanRegistry;
-import org.apache.zookeeper.metrics.MetricsContext;
-import org.apache.zookeeper.metrics.impl.NullMetricsProvider;
 import org.apache.zookeeper.proto.AuthPacket;
 import org.apache.zookeeper.proto.ConnectRequest;
 import org.apache.zookeeper.proto.ConnectResponse;
@@ -140,7 +138,6 @@ public class ZooKeeperServer implements SessionExpirer, ServerStats.Provider {
 
     private final ServerStats serverStats;
     private final ZooKeeperServerListener listener;
-    private MetricsContext rootMetricsContext = NullMetricsProvider.NullMetricsContext.INSTANCE;
     private ZooKeeperServerShutdownHandler zkShutdownHandler;
     private volatile int createSessionTrackerServerId = 1;
 
@@ -183,8 +180,8 @@ public class ZooKeeperServer implements SessionExpirer, ServerStats.Provider {
      * @throws IOException
      */
     public ZooKeeperServer() {
-        serverStats = new ServerStats(this);
         listener = new ZooKeeperServerListenerImpl(this);
+        serverStats = new ServerStats(this);
     }
 
     /**
@@ -379,7 +376,7 @@ public class ZooKeeperServer implements SessionExpirer, ServerStats.Provider {
         }
         long elapsed = Time.currentElapsedTime() - start;
         LOG.info("Snapshot taken in " + elapsed + " ms");
-        ServerMetrics.SNAPSHOT_TIME.add(elapsed);
+        ServerMetrics.getMetrics().SNAPSHOT_TIME.add(elapsed);
     }
 
     @Override
@@ -938,14 +935,6 @@ public class ZooKeeperServer implements SessionExpirer, ServerStats.Provider {
         secureServerCnxnFactory = factory;
     }
 
-    public MetricsContext getRootMetricsContext() {
-        return rootMetricsContext;
-    }
-
-    public void setRootMetricsContext(MetricsContext rootMetricsContext) {
-        this.rootMetricsContext = rootMetricsContext;
-    }
-
     /**
      * return the last proceesed id from the
      * datatree
@@ -1084,7 +1073,7 @@ public class ZooKeeperServer implements SessionExpirer, ServerStats.Provider {
         if (connThrottle.checkLimit(1) == false) {
             throw new ClientCnxnLimitException();
         }
-        ServerMetrics.CONNECTION_TOKEN_DEFICIT.add(connThrottle.getDeficit());
+        ServerMetrics.getMetrics().CONNECTION_TOKEN_DEFICIT.add(connThrottle.getDeficit());
 
         BinaryInputArchive bia = BinaryInputArchive.getArchive(new ByteBufferInputStream(incomingBuffer));
         ConnectRequest connReq = new ConnectRequest();
@@ -1095,7 +1084,7 @@ public class ZooKeeperServer implements SessionExpirer, ServerStats.Provider {
                     + " client's lastZxid is 0x"
                     + Long.toHexString(connReq.getLastZxidSeen()));
         }
-        ServerMetrics.CONNECTION_REQUEST_COUNT.add(1);
+        ServerMetrics.getMetrics().CONNECTION_REQUEST_COUNT.add(1);
         boolean readOnly = false;
         try {
             readOnly = bia.readBool("readOnly");

zookeeper-server/src/main/java/org/apache/zookeeper/server/ZooKeeperServerMain.java

@@ -129,7 +129,7 @@ public class ZooKeeperServerMain {
                 throw new IOException("Cannot boot MetricsProvider "+config.getMetricsProviderClassName(),
                     error);
             }
-
+            ServerMetrics.metricsProviderInitialized(metricsProvider);
             // Note that this thread isn't going to be doing anything else,
             // so rather than spawning another thread, we will just call
             // run() in this thread.
@@ -138,7 +138,6 @@ public class ZooKeeperServerMain {
             final ZooKeeperServer zkServer = new ZooKeeperServer(txnLog,
                     config.tickTime, config.minSessionTimeout, config.maxSessionTimeout,
                     config.listenBacklog, null);
-            zkServer.setRootMetricsContext(metricsProvider.getRootContext());
             txnLog.setServerStats(zkServer.serverStats());
 
             // Registers shutdown handler which will be used to know the

zookeeper-server/src/main/java/org/apache/zookeeper/server/admin/Commands.java

@@ -392,8 +392,12 @@ public class Commands {
                 response.put("observer_master_id", ((ObserverZooKeeperServer)zkServer).getObserver().getLearnerMasterId());
             }
 
-            response.putAll(ServerMetrics.getAllValues());
-
+            ServerMetrics.getMetrics()
+                    .getMetricsProvider()
+                    .dump(
+                    (metric, value) -> {
+                        response.put(metric, value);
+                    });
             return response;
 
         }}

zookeeper-server/src/main/java/org/apache/zookeeper/server/auth/EnsembleAuthenticationProvider.java

@@ -34,6 +34,7 @@ import org.slf4j.LoggerFactory;
 import org.apache.zookeeper.jmx.ZKMBeanInfo;
 
 import javax.management.JMException;
+import org.apache.zookeeper.server.ServerStats;
 
 /**
  * This is not a true AuthenticationProvider in the strict sense. it does
@@ -83,19 +84,19 @@ public class EnsembleAuthenticationProvider implements AuthenticationProvider {
     handleAuthentication(ServerCnxn cnxn, byte[] authData)
     {
         if (authData == null || authData.length == 0) {
-            ServerMetrics.ENSEMBLE_AUTH_SKIP.add(1);
+            ServerMetrics.getMetrics().ENSEMBLE_AUTH_SKIP.add(1);
             return KeeperException.Code.OK;
         }
 
         String receivedEnsembleName = new String(authData);
 
         if (ensembleNames == null) {
-            ServerMetrics.ENSEMBLE_AUTH_SKIP.add(1);
+            ServerMetrics.getMetrics().ENSEMBLE_AUTH_SKIP.add(1);
             return KeeperException.Code.OK;
         }
 
         if (ensembleNames.contains(receivedEnsembleName)) {
-            ServerMetrics.ENSEMBLE_AUTH_SUCCESS.add(1);
+            ServerMetrics.getMetrics().ENSEMBLE_AUTH_SUCCESS.add(1);
             return KeeperException.Code.OK;
         }
 
@@ -111,7 +112,7 @@ public class EnsembleAuthenticationProvider implements AuthenticationProvider {
          * we return an error, the client will get a fatal auth error and
          * shutdown.
          */
-        ServerMetrics.ENSEMBLE_AUTH_FAIL.add(1);
+        ServerMetrics.getMetrics().ENSEMBLE_AUTH_FAIL.add(1);
         cnxn.close();
         return KeeperException.Code.BADARGUMENTS;
     }

zookeeper-server/src/main/java/org/apache/zookeeper/server/command/MonitorCommand.java

@@ -22,6 +22,7 @@ import java.io.PrintWriter;
 
 import org.apache.zookeeper.Version;
 import org.apache.zookeeper.server.ServerCnxn;
+import org.apache.zookeeper.server.ServerMetrics;
 import org.apache.zookeeper.server.ServerStats;
 import org.apache.zookeeper.server.ZKDatabase;
 import org.apache.zookeeper.server.quorum.Leader;
@@ -80,6 +81,22 @@ public class MonitorCommand extends AbstractFourLetterCommand {
             print("max_proposal_size", leader.getProposalStats().getMaxBufferSize());
             print("min_proposal_size", leader.getProposalStats().getMinBufferSize());
         }
+
+        ServerMetrics.getMetrics()
+                    .getMetricsProvider()
+                    .dump(
+                    (metric, value) -> {
+                        if (value == null) {
+                            print(metric, null);
+                        } else if (value instanceof Long
+                                || value instanceof Integer) {
+                            print(metric, ((Number) value).longValue());
+                        } else if (value instanceof Number) {
+                            print(metric, ((Number) value).doubleValue());                        
+                        } else {
+                            print(metric, value.toString());
+                        }
+                    });
     }
 
     private void print(String key, long number) {

zookeeper-server/src/main/java/org/apache/zookeeper/server/metric/AvgMinMaxCounter.java

@@ -22,17 +22,19 @@ import java.math.BigDecimal;
 import java.util.LinkedHashMap;
 import java.util.Map;
 import java.util.concurrent.atomic.AtomicLong;
+import org.apache.zookeeper.metrics.Summary;
 
 /**
  * Generic long counter that keep track of min/max/avg. The counter is
  * thread-safe
  */
-public class AvgMinMaxCounter extends Metric {
-    private String name;
-    private AtomicLong total = new AtomicLong();
-    private AtomicLong min = new AtomicLong(Long.MAX_VALUE);
-    private AtomicLong max = new AtomicLong(Long.MIN_VALUE);
-    private AtomicLong count = new AtomicLong();
+public class AvgMinMaxCounter extends Metric
+            implements Summary {
+    private final String name;
+    private final AtomicLong total = new AtomicLong();
+    private final AtomicLong min = new AtomicLong(Long.MAX_VALUE);
+    private final AtomicLong max = new AtomicLong(Long.MIN_VALUE);
+    private final AtomicLong count = new AtomicLong();
 
     public AvgMinMaxCounter(String name) {
         this.name = name;

zookeeper-server/src/main/java/org/apache/zookeeper/server/metric/AvgMinMaxCounterSet.java

@@ -18,19 +18,18 @@
 
 package org.apache.zookeeper.server.metric;
 
-import java.lang.Integer;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.LinkedHashMap;
 import java.util.Map;
-import java.util.concurrent.atomic.AtomicLong;
+import org.apache.zookeeper.metrics.SummarySet;
 
 /**
  * Generic set of long counters that keep track of min/max/avg
  * for different keys.
  * The counter is thread-safe
  */
-public class AvgMinMaxCounterSet extends Metric {
-    private String name;
+public class AvgMinMaxCounterSet extends Metric implements SummarySet {
+    private final String name;
 
     private ConcurrentHashMap<String, AvgMinMaxCounter> counters = new ConcurrentHashMap<>();
 

zookeeper-server/src/main/java/org/apache/zookeeper/server/metric/AvgMinMaxPercentileCounter.java

@@ -29,16 +29,18 @@ import java.util.Map;
 import java.util.concurrent.ThreadLocalRandom;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicLongArray;
+import org.apache.zookeeper.metrics.Summary;
 
 
 /**
  * Generic long counter that keep track of min/max/avg/percentiles.
  * The counter is thread-safe
  */
-public class AvgMinMaxPercentileCounter extends Metric  {
+public class AvgMinMaxPercentileCounter extends Metric 
+                implements Summary {
 
-    private String name;
-    private AvgMinMaxCounter counter;
+    private final String name;
+    private final AvgMinMaxCounter counter;
     private final ResettableUniformReservoir reservoir;
     private final Histogram histogram;
 

zookeeper-server/src/main/java/org/apache/zookeeper/server/metric/AvgMinMaxPercentileCounterSet.java

@@ -18,19 +18,18 @@
 
 package org.apache.zookeeper.server.metric;
 
-import java.lang.Integer;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.LinkedHashMap;
 import java.util.Map;
-import java.util.concurrent.atomic.AtomicLong;
+import org.apache.zookeeper.metrics.SummarySet;
 
 /**
  * Generic set of long counters that keep track of min/max/avg
  * for different keys.
  * The counter is thread-safe
  */
-public class AvgMinMaxPercentileCounterSet extends Metric {
-    private String name;
+public class AvgMinMaxPercentileCounterSet extends Metric implements SummarySet {
+    private final String name;
 
     private ConcurrentHashMap<String, AvgMinMaxPercentileCounter> counters = new ConcurrentHashMap<>();
 

zookeeper-server/src/main/java/org/apache/zookeeper/server/metric/SimpleCounter.java

@@ -22,8 +22,10 @@ import java.lang.Override;
 import java.util.LinkedHashMap;
 import java.util.Map;
 import java.util.concurrent.atomic.AtomicLong;
+import org.apache.zookeeper.metrics.Counter;
 
-public class SimpleCounter extends Metric {
+public class SimpleCounter extends Metric
+                    implements Counter {
     private final String name;
     private final AtomicLong counter = new AtomicLong();
 
@@ -41,14 +43,14 @@ public class SimpleCounter extends Metric {
         counter.set(0);
     }
 
-    public long getCount() {
+    public long get() {
         return counter.get();
     }
 
     @Override
     public Map<String, Object> values() {
         Map<String, Object> m = new LinkedHashMap<String, Object>();
-        m.put(name, this.getCount());
+        m.put(name, this.get());
         return m;
     }
 }

zookeeper-server/src/main/java/org/apache/zookeeper/server/persistence/FileTxnLog.java

@@ -388,7 +388,8 @@ public class FileTxnLog implements TxnLog {
                             + "File size is " + channel.size() + " bytes. "
                             + "See the ZooKeeper troubleshooting guide");
                 }
-                ServerMetrics.FSYNC_TIME.add(syncElapsedMS);
+                
+                ServerMetrics.getMetrics().FSYNC_TIME.add(syncElapsedMS);
             }
         }
         while (streamsToFlush.size() > 1) {

zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/Follower.java

@@ -70,7 +70,7 @@ public class Follower extends Learner{
         self.end_fle = Time.currentElapsedTime();
         long electionTimeTaken = self.end_fle - self.start_fle;
         self.setElectionTimeTaken(electionTimeTaken);
-        ServerMetrics.ELECTION_TIME.add(electionTimeTaken);
+        ServerMetrics.getMetrics().ELECTION_TIME.add(electionTimeTaken);
         LOG.info("FOLLOWING - LEADER ELECTION TOOK - {} {}", electionTimeTaken,
                 QuorumPeer.FLE_TIME_UNIT);
         self.start_fle = 0;
@@ -96,7 +96,7 @@ public class Follower extends Learner{
                     syncWithLeader(newEpochZxid);
                 } finally {
                     long syncTime = Time.currentElapsedTime() - startTime;
-                    ServerMetrics.FOLLOWER_SYNC_TIME.add(syncTime);
+                    ServerMetrics.getMetrics().FOLLOWER_SYNC_TIME.add(syncTime);
                 }
                 if (self.getObserverMasterPort() > 0) {
                     LOG.info("Starting ObserverMaster");

zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/Leader.java

@@ -428,7 +428,8 @@ public class Leader implements LearnerMaster {
 
                         BufferedInputStream is = new BufferedInputStream(
                                 s.getInputStream());
-                        LearnerHandler fh = new LearnerHandler(s, is, Leader.this);
+                        LearnerHandler fh = new LearnerHandler(s, is,
+                                Leader.this);
                         fh.start();
                     } catch (SocketException e) {
                         error = true;
@@ -502,7 +503,7 @@ public class Leader implements LearnerMaster {
         self.end_fle = Time.currentElapsedTime();
         long electionTimeTaken = self.end_fle - self.start_fle;
         self.setElectionTimeTaken(electionTimeTaken);
-        ServerMetrics.ELECTION_TIME.add(electionTimeTaken);
+        ServerMetrics.getMetrics().ELECTION_TIME.add(electionTimeTaken);
         LOG.info("LEADING - LEADER ELECTION TOOK - {} {}", electionTimeTaken,
                 QuorumPeer.FLE_TIME_UNIT);
         self.start_fle = 0;
@@ -1057,7 +1058,7 @@ public class Leader implements LearnerMaster {
         }
         QuorumPacket qp = new QuorumPacket(Leader.COMMIT, zxid, null, null);
         sendPacket(qp);
-        ServerMetrics.COMMIT_COUNT.add(1);
+        ServerMetrics.getMetrics().COMMIT_COUNT.add(1);
     }
 
     //commit and send some info

zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/LearnerHandler.java

@@ -481,11 +481,11 @@ public class LearnerHandler extends ZooKeeperThread {
                     bufferedOutput.flush();
                 } finally {
                     snapshot.close();
-                    ServerMetrics.SNAP_COUNT.add(1);
+                    ServerMetrics.getMetrics().SNAP_COUNT.add(1);
                 }
             }
             else {
-                ServerMetrics.DIFF_COUNT.add(1);
+                ServerMetrics.getMetrics().DIFF_COUNT.add(1);
             }
 
             LOG.debug("Sending NEWLEADER message to " + sid);

zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/QuorumPeer.java

@@ -79,6 +79,7 @@ import org.slf4j.LoggerFactory;
 
 import static org.apache.zookeeper.common.NetUtils.formatInetAddr;
 import org.apache.zookeeper.metrics.MetricsContext;
+import org.apache.zookeeper.metrics.MetricsProvider;
 import org.apache.zookeeper.metrics.impl.NullMetricsProvider;
 
 /**
@@ -832,8 +833,6 @@ public class QuorumPeer extends ZooKeeperThread implements QuorumStats.Provider
 
     AdminServer adminServer;
 
-    private MetricsContext rootMetricsContext = NullMetricsProvider.NullMetricsContext.INSTANCE;
-
     public static QuorumPeer testingQuorumPeer() throws SaslException {
         return new QuorumPeer();
     }
@@ -1178,7 +1177,7 @@ public class QuorumPeer extends ZooKeeperThread implements QuorumStats.Provider
                 switch (getPeerState()) {
                 case LOOKING:
                     LOG.info("LOOKING");
-                    ServerMetrics.LOOKING_COUNT.add(1);
+                    ServerMetrics.getMetrics().LOOKING_COUNT.add(1);
 
                     if (Boolean.getBoolean("readonlymode.enabled")) {
                         LOG.info("Attempting to start ReadOnlyZooKeeperServer");
@@ -1471,7 +1470,7 @@ public class QuorumPeer extends ZooKeeperThread implements QuorumStats.Provider
         }
         return -1;
     }
-
+    
     /** Whether local sessions are enabled */
     public boolean areLocalSessionsEnabled() {
         return localSessionsEnabled;
@@ -1780,10 +1779,6 @@ public class QuorumPeer extends ZooKeeperThread implements QuorumStats.Provider
         this.secureCnxnFactory = secureCnxnFactory;
     }
 
-    public void setRootMetricsContext(MetricsContext rootMetricsContext) {
-        this.rootMetricsContext = rootMetricsContext;
-    }
-
     public void setSslQuorum(boolean sslQuorum) {
         if (sslQuorum) {
             LOG.info("Using TLS encrypted quorum communication");

zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/QuorumPeerConfig.java

@@ -56,7 +56,7 @@ import org.apache.zookeeper.server.quorum.flexible.QuorumVerifier;
 import org.apache.zookeeper.server.util.VerifyingFileFactory;
 
 import static org.apache.zookeeper.common.NetUtils.formatInetAddr;
-import org.apache.zookeeper.metrics.impl.NullMetricsProvider;
+import org.apache.zookeeper.metrics.impl.DefaultMetricsProvider;
 
 @InterfaceAudience.Public
 public class QuorumPeerConfig {
@@ -83,7 +83,7 @@ public class QuorumPeerConfig {
     protected int minSessionTimeout = -1;
     /** defaults to -1 if not set explicitly */
     protected int maxSessionTimeout = -1;
-    protected String metricsProviderClassName = NullMetricsProvider.class.getName();
+    protected String metricsProviderClassName = DefaultMetricsProvider.class.getName();
     protected Properties metricsProviderConfiguration = new Properties();
     protected boolean localSessionsEnabled = false;
     protected boolean localSessionsUpgradingEnabled = false;

zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/QuorumPeerMain.java

@@ -34,6 +34,7 @@ import org.apache.zookeeper.server.ExitCode;
 import org.apache.zookeeper.server.ServerCnxnFactory;
 import org.apache.zookeeper.server.ZKDatabase;
 import org.apache.zookeeper.server.DatadirCleanupManager;
+import org.apache.zookeeper.server.ServerMetrics;
 import org.apache.zookeeper.server.ZooKeeperServerMain;
 import org.apache.zookeeper.server.admin.AdminServer.AdminServerException;
 import org.apache.zookeeper.server.persistence.FileTxnSnapLog;
@@ -154,7 +155,7 @@ public class QuorumPeerMain {
                       error);
       }
       try {
-
+          ServerMetrics.metricsProviderInitialized(metricsProvider);
           ServerCnxnFactory cnxnFactory = null;
           ServerCnxnFactory secureCnxnFactory = null;
 
@@ -173,7 +174,6 @@ public class QuorumPeerMain {
           }
 
           quorumPeer = getQuorumPeer();
-          quorumPeer.setRootMetricsContext(metricsProvider.getRootContext());
           quorumPeer.setTxnFactory(new FileTxnSnapLog(
                       config.getDataLogDir(),
                       config.getDataDir()));

zookeeper-server/src/main/java/org/apache/zookeeper/server/watch/WatchManager.java

@@ -48,7 +48,7 @@ public class WatchManager implements IWatchManager {
 
     private final Map<Watcher, Set<String>> watch2Paths =
         new HashMap<Watcher, Set<String>>();
-
+    
     @Override
     public synchronized int size(){
         int result = 0;
@@ -142,19 +142,19 @@ public class WatchManager implements IWatchManager {
 
         switch (type) {
         case NodeCreated:
-            ServerMetrics.NODE_CREATED_WATCHER.add(watchers.size());
+            ServerMetrics.getMetrics().NODE_CREATED_WATCHER.add(watchers.size());
             break;
 
         case NodeDeleted:
-            ServerMetrics.NODE_DELETED_WATCHER.add(watchers.size());
+            ServerMetrics.getMetrics().NODE_DELETED_WATCHER.add(watchers.size());
             break;
 
         case NodeDataChanged:
-            ServerMetrics.NODE_CHANGED_WATCHER.add(watchers.size());
+            ServerMetrics.getMetrics().NODE_CHANGED_WATCHER.add(watchers.size());
             break;
 
         case NodeChildrenChanged:
-            ServerMetrics.NODE_CHILDREN_WATCHER.add(watchers.size());
+            ServerMetrics.getMetrics().NODE_CHILDREN_WATCHER.add(watchers.size());
             break;
         default:
             // Other types not logged.
@@ -267,4 +267,5 @@ public class WatchManager implements IWatchManager {
 
     @Override
     public void shutdown() { /* do nothing */ }
+
 }

zookeeper-server/src/main/java/org/apache/zookeeper/server/watch/WatchManagerOptimized.java

@@ -277,19 +277,19 @@ public class WatchManagerOptimized
     void updateMetrics(final EventType type, int size) {
         switch (type) {
         case NodeCreated:
-            ServerMetrics.NODE_CREATED_WATCHER.add(size);
+            ServerMetrics.getMetrics().NODE_CREATED_WATCHER.add(size);
             break;
 
         case NodeDeleted:
-            ServerMetrics.NODE_DELETED_WATCHER.add(size);
+            ServerMetrics.getMetrics().NODE_DELETED_WATCHER.add(size);
             break;
 
         case NodeDataChanged:
-            ServerMetrics.NODE_CHANGED_WATCHER.add(size);
+            ServerMetrics.getMetrics().NODE_CHANGED_WATCHER.add(size);
             break;
 
         case NodeChildrenChanged:
-            ServerMetrics.NODE_CHILDREN_WATCHER.add(size);
+            ServerMetrics.getMetrics().NODE_CHILDREN_WATCHER.add(size);
             break;
         default:
             // Other types not logged.
@@ -414,4 +414,5 @@ public class WatchManagerOptimized
         sb.append("Total watches:").append(size());
         return sb.toString();
     }
+
 }

zookeeper-server/src/main/java/org/apache/zookeeper/server/watch/WatcherCleaner.java

@@ -109,7 +109,7 @@ public class WatcherCleaner extends Thread {
                     processingCompletedEvent.wait(100);
                 }
                 long latency = Time.currentElapsedTime() - startTime;
-                ServerMetrics.ADD_DEAD_WATCHER_STALL_TIME.add(latency);
+                ServerMetrics.getMetrics().ADD_DEAD_WATCHER_STALL_TIME.add(latency);
             } catch (InterruptedException e) {
                 LOG.info("Got interrupted while waiting for dead watches " +
                         "queue size");
@@ -119,7 +119,7 @@ public class WatcherCleaner extends Thread {
         synchronized (this) {
             if (deadWatchers.add(watcherBit)) {
                 totalDeadWatchers.incrementAndGet();
-                ServerMetrics.DEAD_WATCHERS_QUEUED.add(1);
+                ServerMetrics.getMetrics().DEAD_WATCHERS_QUEUED.add(1);
                 if (deadWatchers.size() >= watcherCleanThreshold) {
                     synchronized (cleanEvent) {
                         cleanEvent.notifyAll();
@@ -169,8 +169,8 @@ public class WatcherCleaner extends Thread {
                         listener.processDeadWatchers(snapshot);
                         long latency = Time.currentElapsedTime() - startTime;
                         LOG.info("Takes {} to process {} watches", latency, total);
-                        ServerMetrics.DEAD_WATCHERS_CLEANER_LATENCY.add(latency);
-                        ServerMetrics.DEAD_WATCHERS_CLEARED.add(total);
+                        ServerMetrics.getMetrics().DEAD_WATCHERS_CLEANER_LATENCY.add(latency);
+                        ServerMetrics.getMetrics().DEAD_WATCHERS_CLEARED.add(total);
                         totalDeadWatchers.addAndGet(-total);
                         synchronized(processingCompletedEvent) {
                             processingCompletedEvent.notifyAll();

zookeeper-server/src/test/java/org/apache/zookeeper/metrics/BaseTestMetricsProvider.java

@@ -20,6 +20,7 @@ package org.apache.zookeeper.metrics;
 import java.util.Properties;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.BiConsumer;
 import org.apache.zookeeper.metrics.impl.NullMetricsProvider;
 
 /**
@@ -44,6 +45,14 @@ public abstract class BaseTestMetricsProvider implements MetricsProvider {
     public void stop() {
     }
 
+    @Override
+    public void dump(BiConsumer<String, Object> sink) {
+    }
+
+    @Override
+    public void resetAllValues() {
+    }
+
     public static final class MetricsProviderCapturingLifecycle extends BaseTestMetricsProvider {
 
         public static final AtomicBoolean configureCalled = new AtomicBoolean();

zookeeper-server/src/test/java/org/apache/zookeeper/metrics/MetricsUtils.java

@@ -0,0 +1,58 @@
+/**
+ * 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.zookeeper.metrics;
+
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.zookeeper.server.ServerMetrics;
+
+/**
+ * Utility for Metrics in tests.
+ */
+public abstract class MetricsUtils {
+
+    private MetricsUtils() {
+    }
+
+    /**
+     * Collect all metrics from a {@link MetricsProvider}. A MetricsProvider
+     * provides a {@link MetricsProvider#dump(java.util.function.BiConsumer)
+     * }
+     * method, that method will in general be more efficient and it does not
+     * impose to the MetricsProvider to waste resources.
+     *
+     * @param metricsProvider
+     * @return a Map which collects one entry per each different key returned by
+     * {@link MetricsProvider#dump(java.util.function.BiConsumer) }
+     */
+    public static Map<String, Object> collect(MetricsProvider metricsProvider) {
+        Map<String, Object> res = new HashMap<>();
+        metricsProvider.dump(res::put);
+        return res;
+    }
+    
+    /**
+     * Collect current {@link ServerMetrics} as a Map.
+     *
+     * @return a flattened view of all metrics reported by the MetricsProvider
+     * in use by the current ServerMetrics static instance.
+     */
+    public static Map<String, Object> currentServerMetrics() {
+        return collect(ServerMetrics.getMetrics().getMetricsProvider());
+    }
+}

zookeeper-server/src/test/java/org/apache/zookeeper/server/DataTreeTest.java

@@ -52,6 +52,7 @@ import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.List;
 import java.util.ArrayList;
+import org.apache.zookeeper.metrics.MetricsUtils;
 
 public class DataTreeTest extends ZKTestCase {
     protected static final Logger LOG = LoggerFactory.getLogger(DataTreeTest.class);
@@ -363,7 +364,7 @@ public class DataTreeTest extends ZKTestCase {
 
     @Test
     public void testDataTreeMetrics() throws Exception {
-        ServerMetrics.resetAll();
+        ServerMetrics.getMetrics().resetAll();
 
 
         long readBytes1 = 0;
@@ -406,8 +407,8 @@ public class DataTreeTest extends ZKTestCase {
         dt.deleteNode(TOP1PATH, 1);
         writeBytes1 += TOP1PATH.length();
         
-        Map<String, Object> values = ServerMetrics.getAllValues();
-
+        Map<String, Object> values = MetricsUtils.currentServerMetrics();
+        System.out.println("values:"+values);
         Assert.assertEquals(writeBytes1, values.get("sum_" + TOP1+ "_write_per_namespace"));
         Assert.assertEquals(5L, values.get("cnt_" + TOP1 + "_write_per_namespace"));
         Assert.assertEquals(writeBytes2, values.get("sum_" + TOP2+ "_write_per_namespace"));

zookeeper-server/src/test/java/org/apache/zookeeper/server/ServerMetricsTest.java

@@ -99,7 +99,7 @@ public class ServerMetricsTest extends ZKTestCase {
         }
 
         long expectedCount = Arrays.stream(values).sum();
-        Assert.assertEquals(expectedCount, metric.getCount());
+        Assert.assertEquals(expectedCount, metric.get());
 
         final Map<String, Object> results = metric.values();
         Assert.assertEquals(expectedCount, (long)results.get("test"));

zookeeper-server/src/test/java/org/apache/zookeeper/server/ZooKeeperCriticalThreadMetricsTest.java

@@ -26,6 +26,7 @@ import org.junit.Test;
 import java.nio.ByteBuffer;
 import java.util.Map;
 import java.util.concurrent.CountDownLatch;
+import org.apache.zookeeper.metrics.MetricsUtils;
 
 public class ZooKeeperCriticalThreadMetricsTest extends ZKTestCase {
     CountDownLatch processed;
@@ -57,7 +58,7 @@ public class ZooKeeperCriticalThreadMetricsTest extends ZKTestCase {
 
     @Test
     public void testUnrecoverableErrorCountFromRequestProcessor() throws Exception{
-        ServerMetrics.resetAll();
+        ServerMetrics.getMetrics().resetAll();
 
         processed = new CountDownLatch(1);
         PrepRequestProcessor processor =new MyPrepRequestProcessor();
@@ -69,20 +70,20 @@ public class ZooKeeperCriticalThreadMetricsTest extends ZKTestCase {
 
         processor.shutdown();
 
-        Map<String, Object> values = ServerMetrics.getAllValues();
+        Map<String, Object> values = MetricsUtils.currentServerMetrics();
         Assert.assertEquals(1L, values.get("unrecoverable_error_count"));
     }
 
     @Test
     public void testUnrecoverableErrorCount() {
-        ServerMetrics.resetAll();
+        ServerMetrics.getMetrics().resetAll();
 
         ZooKeeperServer zks = new ZooKeeperServer();
         ZooKeeperCriticalThread thread = new ZooKeeperCriticalThread("test", zks.getZooKeeperServerListener());
 
         thread.handleException("test", new Exception());
 
-        Map<String, Object> values = ServerMetrics.getAllValues();
+        Map<String, Object> values = MetricsUtils.currentServerMetrics();
         Assert.assertEquals(1L, values.get("unrecoverable_error_count"));
     }
 }

zookeeper-server/src/test/java/org/apache/zookeeper/server/admin/CommandsTest.java

@@ -31,6 +31,7 @@ import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.HashMap;
 import java.util.Map;
+import org.apache.zookeeper.metrics.MetricsUtils;
 
 import org.apache.zookeeper.server.ServerCnxnFactory;
 import org.apache.zookeeper.server.ServerMetrics;
@@ -193,8 +194,10 @@ public class CommandsTest extends ClientBase {
                 new Field("global_sessions", Long.class),
                 new Field("local_sessions", Long.class),
                 new Field("connection_drop_probability", Double.class)
-        ));
-        for (String metric : ServerMetrics.getAllValues().keySet()) {
+        ));        
+        Map<String, Object> metrics = MetricsUtils.currentServerMetrics();
+        
+        for (String metric : metrics.keySet()) {
             if (metric.startsWith("avg_")) {
                 fields.add(new Field(metric, Double.class));  
             } else {

zookeeper-server/src/test/java/org/apache/zookeeper/server/quorum/Zab1_0Test.java

@@ -60,7 +60,6 @@ import org.apache.zookeeper.server.persistence.FileTxnSnapLog;
 import org.apache.zookeeper.server.quorum.QuorumPeer.QuorumServer;
 
 import org.apache.zookeeper.server.util.ZxidUtils;
-import org.apache.zookeeper.test.ClientBase;
 import org.apache.zookeeper.test.TestUtils;
 import org.apache.zookeeper.txn.CreateSessionTxn;
 import org.apache.zookeeper.txn.CreateTxn;

zookeeper-server/src/test/java/org/apache/zookeeper/server/watch/WatchManagerTest.java

@@ -19,6 +19,7 @@ package org.apache.zookeeper.server.watch;
 import java.io.IOException;
 import java.util.Arrays;
 import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Random;
@@ -33,10 +34,8 @@ import org.apache.zookeeper.server.DumbWatcher;
 import org.apache.zookeeper.server.ServerCnxn;
 
 import org.apache.zookeeper.ZKTestCase;
+import org.apache.zookeeper.metrics.MetricsUtils;
 import org.apache.zookeeper.server.ServerMetrics;
-import org.apache.zookeeper.server.metric.AvgMinMaxCounter;
-import org.apache.zookeeper.server.metric.Metric;
-import org.eclipse.jetty.util.IO;
 import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
@@ -70,6 +69,7 @@ public class WatchManagerTest extends ZKTestCase {
 
     @Before
     public void setUp() {
+        ServerMetrics.getMetrics().resetAll();
         watchers = new ConcurrentHashMap<Integer, DumbWatcher>();
         r = new Random(System.nanoTime());
     }
@@ -409,7 +409,7 @@ public class WatchManagerTest extends ZKTestCase {
     }
 
     private void checkMetrics(String metricName, long min, long max, double avg, long cnt, long sum){
-        Map<String, Object> values = ServerMetrics.getAllValues();
+        Map<String, Object> values = MetricsUtils.currentServerMetrics();
 
         Assert.assertEquals(min, values.get("min_" + metricName));
         Assert.assertEquals(max, values.get("max_" + metricName));
@@ -421,7 +421,7 @@ public class WatchManagerTest extends ZKTestCase {
     @Test
     public void testWatcherMetrics() throws IOException {
         IWatchManager manager = getWatchManager();
-        ServerMetrics.resetAll();
+        ServerMetrics.getMetrics().resetAll();
 
         DumbWatcher watcher1 = new DumbWatcher(1);
         DumbWatcher watcher2 = new DumbWatcher(2);

zookeeper-server/src/test/java/org/apache/zookeeper/server/watch/WatcherCleanerTest.java

@@ -16,6 +16,7 @@
  */
 package org.apache.zookeeper.server.watch;
 
+import java.util.HashMap;
 import java.util.Map;
 import java.util.Set;
 import java.util.HashSet;
@@ -24,6 +25,8 @@ import java.util.concurrent.TimeUnit;
 
 import org.apache.zookeeper.ZKTestCase;
 import org.apache.zookeeper.common.Time;
+import org.apache.zookeeper.metrics.MetricsUtils;
+import org.apache.zookeeper.metrics.impl.DefaultMetricsProvider;
 import org.apache.zookeeper.server.ServerMetrics;
 import org.junit.Test;
 import org.junit.Assert;
@@ -134,6 +137,7 @@ public class WatcherCleanerTest extends ZKTestCase {
 
     @Test
     public void testDeadWatcherMetrics() {
+        ServerMetrics.getMetrics().resetAll();
         MyDeadWatcherListener listener = new MyDeadWatcherListener();
         WatcherCleaner cleaner = new WatcherCleaner(listener, 1, 1, 1, 1);
         listener.setDelayMs(20);
@@ -146,9 +150,8 @@ public class WatcherCleanerTest extends ZKTestCase {
         cleaner.addDeadWatcher(3);
 
         Assert.assertTrue(listener.wait(5000));
-
-        Map<String, Object> values = ServerMetrics.getAllValues();
-
+        
+        Map<String, Object> values = MetricsUtils.currentServerMetrics();
         Assert.assertThat("Adding dead watcher should be stalled twice",
                           (Long)values.get("add_dead_watcher_stall_time"),
                            greaterThan(0L));

zookeeper-server/src/test/java/org/apache/zookeeper/test/NonRecoverableErrorTest.java

@@ -29,6 +29,7 @@ import org.apache.zookeeper.CreateMode;
 import org.apache.zookeeper.PortAssignment;
 import org.apache.zookeeper.ZooDefs.Ids;
 import org.apache.zookeeper.ZooKeeper;
+import org.apache.zookeeper.server.ServerMetrics;
 import org.apache.zookeeper.server.ZKDatabase;
 import org.apache.zookeeper.server.persistence.FileTxnSnapLog;
 import org.apache.zookeeper.server.quorum.QuorumPeer;

zookeeper-server/src/test/java/org/apache/zookeeper/test/ResponseCacheTest.java

@@ -24,6 +24,7 @@ import org.apache.zookeeper.CreateMode;
 import org.apache.zookeeper.ZooDefs;
 import org.apache.zookeeper.ZooKeeper;
 import org.apache.zookeeper.data.Stat;
+import org.apache.zookeeper.metrics.MetricsUtils;
 import org.apache.zookeeper.server.ServerMetrics;
 import org.junit.Assert;
 import org.junit.Test;
@@ -48,13 +49,14 @@ public class ResponseCacheTest extends ClientBase {
     }
 
     private void checkCacheStatus(long expectedHits, long expectedMisses) {
-        Map<String, Object> metrics = ServerMetrics.getAllValues();
+        
+        Map<String, Object> metrics = MetricsUtils.currentServerMetrics();
         Assert.assertEquals(expectedHits, metrics.get("response_packet_cache_hits"));
         Assert.assertEquals(expectedMisses, metrics.get("response_packet_cache_misses"));
     }
 
     public void performCacheTest(ZooKeeper zk, String path, boolean useCache) throws Exception {
-        ServerMetrics.resetAll();
+        ServerMetrics.getMetrics().resetAll();
         Stat writeStat = new Stat();
         Stat readStat = new Stat();
         byte[] readData = null;