ZOOKEEPER-3098: Add additional server metrics

This patch adds several new server-side metrics as well as makes it easier to add new metrics in the future. This patch also includes a handful of other minor metrics-related changes.

Here's a high-level summary of the changes.

1. This patch extends the request latency tracked in `ServerStats` to
   track `read` and `update` latency separately. Updates are any request
   that must be voted on and can change data, reads are all requests that
   can be handled locally and don't change data.

2. This patch adds the `ServerMetrics` logic and the related `AvgMinMaxCounter`
   and `SimpleCounter` classes. This code is designed to make it incredibly easy to
   add new metrics. To add a new metric you just add one line to `ServerMetrics` and
   then directly reference that new metric anywhere in the code base. The `ServerMetrics`
   logic handles creating the metric, properly adding the metric to the JSON output of
   the `/monitor` admin command, and properly resetting the metric when necessary.

   The motivation behind `ServerMetrics` is to make things easy enough that it encourages
   new metrics to be added liberally. Lack of in-depth metrics/visibility is a long-standing
   ZooKeeper weakness. At Facebook, most of our internal changes build on `ServerMetrics` and
   we have nearly 100 internal metrics at this time -- all of which we'll be upstreaming
   in the coming months as we publish more internal patches.

3. This patch adds 20 new metrics, 14 which are handled by `ServerMetrics`.

4. This patch replaces some uses of `synchronized` in `ServerStats` with atomic operations.

Here's a list of new metrics added in this patch:

- `uptime`: time that a peer has been in a stable leading/following/observing state
- `leader_uptime`: uptime for peer in leading state
- `global_sessions`: count of global sessions
- `local_sessions`: count of local sessions
- `quorum_size`: configured ensemble size
- `synced_observers`: similar to existing `synced_followers` but for observers
- `fsynctime`: time to fsync transaction log (avg/min/max)
- `snapshottime`: time to write a snapshot (avg/min/max)
- `dbinittime`: time to reload database -- read snapshot + apply transactions (avg/min/max)
- `readlatency`: read request latency (avg/min/max)
- `updatelatency`: update request latency (avg/min/max)
- `propagation_latency`: end-to-end latency for updates, from proposal on leader to committed-to-datatree on a given host (avg/min/max)
- `follower_sync_time`: time for follower to sync with leader (avg/min/max)
- `election_time`: time between entering and leaving election (avg/min/max)
- `looking_count`: number of transitions into looking state
- `diff_count`: number of diff syncs performed
- `snap_count`: number of snap syncs performed
- `commit_count`: number of commits performed on leader
- `connection_request_count`: number of incoming client connection requests
- `bytes_received_count`: similar to existing `packets_received` but tracks bytes

Author: Joseph Blomstedt <jdb@fb.com>

Reviewers: Allan Lyu <fangmin@apache.org>, Andor Molnár <andor@apache.org>, Enrico Olivelli <eolivelli@gmail.com>, Michael Han <hanm@apache.org>

Closes #580 from jtuple/ZOOKEEPER-3098

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

@@ -107,7 +107,6 @@ public class FinalRequestProcessor implements RequestProcessor {
             // that add to outstandingChanges.
             if (request.getHdr() != null) {
                 TxnHeader hdr = request.getHdr();
-                Record txn = request.getTxn();
                 long zxid = hdr.getZxid();
                 while (!zks.outstandingChanges.isEmpty()
                        && zks.outstandingChanges.peek().zxid <= zxid) {
@@ -143,11 +142,28 @@ public class FinalRequestProcessor implements RequestProcessor {
             }
         }
 
+        if (request.getHdr() != null) {
+            /*
+             * Request header is created only by the leader, so this must be
+             * a quorum request. Since we're comparing timestamps across hosts,
+             * this metric may be incorrect. However, it's still a very useful
+             * metric to track in the happy case. If there is clock drift,
+             * the latency can go negative. Note: headers use wall time, not
+             * CLOCK_MONOTONIC.
+             */
+            long propagationLatency = Time.currentWallTime() - request.getHdr().getTime();
+            if (propagationLatency > 0) {
+                ServerMetrics.PROPAGATION_LATENCY.add(propagationLatency);
+            }
+        }
+
         if (request.cnxn == null) {
             return;
         }
         ServerCnxn cnxn = request.cnxn;
 
+        long lastZxid = zks.getZKDatabase().getDataTreeLastProcessedZxid();
+
         String lastOp = "NA";
         zks.decInProcess();
         Code err = Code.OK;
@@ -182,22 +198,15 @@ public class FinalRequestProcessor implements RequestProcessor {
             }
             switch (request.type) {
             case OpCode.ping: {
-                zks.serverStats().updateLatency(request.createTime);
-
                 lastOp = "PING";
-                cnxn.updateStatsForResponse(request.cxid, request.zxid, lastOp,
-                        request.createTime, Time.currentElapsedTime());
+                updateStats(request, lastOp, lastZxid);
 
-                cnxn.sendResponse(new ReplyHeader(-2,
-                        zks.getZKDatabase().getDataTreeLastProcessedZxid(), 0), null, "response");
+                cnxn.sendResponse(new ReplyHeader(-2, lastZxid, 0), null, "response");
                 return;
             }
             case OpCode.createSession: {
-                zks.serverStats().updateLatency(request.createTime);
-
                 lastOp = "SESS";
-                cnxn.updateStatsForResponse(request.cxid, request.zxid, lastOp,
-                        request.createTime, Time.currentElapsedTime());
+                updateStats(request, lastOp, lastZxid);
 
                 zks.finishSessionInit(request.cnxn, true);
                 return;
@@ -453,13 +462,10 @@ public class FinalRequestProcessor implements RequestProcessor {
             err = Code.MARSHALLINGERROR;
         }
 
-        long lastZxid = zks.getZKDatabase().getDataTreeLastProcessedZxid();
         ReplyHeader hdr =
             new ReplyHeader(request.cxid, lastZxid, err.intValue());
 
-        zks.serverStats().updateLatency(request.createTime);
-        cnxn.updateStatsForResponse(request.cxid, lastZxid, lastOp,
-                    request.createTime, Time.currentElapsedTime());
+        updateStats(request, lastOp, lastZxid);
 
         try {
             cnxn.sendResponse(hdr, rsp, "response");
@@ -487,4 +493,13 @@ public class FinalRequestProcessor implements RequestProcessor {
         LOG.info("shutdown of request processor complete");
     }
 
+    private void updateStats(Request request, String lastOp, long lastZxid) {
+        if (request.cnxn == null) {
+            return;
+        }
+        long currentTime = Time.currentElapsedTime();
+        zks.serverStats().updateLatency(request, currentTime);
+        request.cnxn.updateStatsForResponse(request.cxid, lastZxid, lastOp,
+                request.createTime, currentTime);
+    }
 }

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

@@ -173,8 +173,8 @@ public class NIOServerCnxn extends ServerCnxn {
         }
 
         if (incomingBuffer.remaining() == 0) { // have we read length bytes?
-            packetReceived();
             incomingBuffer.flip();
+            packetReceived(4 + incomingBuffer.remaining());
             if (!initialized) {
                 readConnectRequest();
             } else {
@@ -484,7 +484,7 @@ public class NIOServerCnxn extends ServerCnxn {
         }
 
         String cmd = FourLetterCommands.getCommandString(len);
-        packetReceived();
+        packetReceived(4);
 
         /** cancel the selection key to remove the socket handling
          * from selector. This is to prevent netcat problem wherein

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

@@ -275,7 +275,7 @@ public class NettyServerCnxn extends ServerCnxn {
         String cmd = FourLetterCommands.getCommandString(len);
 
         channel.setInterestOps(0).awaitUninterruptibly();
-        packetReceived();
+        packetReceived(4);
 
         final PrintWriter pwriter = new PrintWriter(
                 new BufferedWriter(new SendBufferWriter()));
@@ -342,8 +342,8 @@ public class NettyServerCnxn extends ServerCnxn {
                                         ChannelBuffers.copiedBuffer(dat)));
                     }
                     if (bb.remaining() == 0) {
-                        packetReceived();
                         bb.flip();
+                        packetReceived(4 + bb.remaining());
 
                         ZooKeeperServer zks = this.zkServer;
                         if (zks == null || !zks.isRunning()) {

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

@@ -143,12 +143,13 @@ public abstract class ServerCnxn implements Stats, Watcher {
         }
     }
 
-    protected void packetReceived() {
+    protected void packetReceived(long bytes) {
         incrPacketsReceived();
         ServerStats serverStats = serverStats();
         if (serverStats != null) {
             serverStats().incrementPacketsReceived();
         }
+        ServerMetrics.BYTES_RECEIVED_COUNT.add(bytes);
     }
 
     protected void packetSent() {

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

@@ -0,0 +1,103 @@
+/**
+ * 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.server;
+
+import org.apache.zookeeper.server.metric.AvgMinMaxCounter;
+import org.apache.zookeeper.server.metric.Metric;
+import org.apache.zookeeper.server.metric.SimpleCounter;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+public enum ServerMetrics {
+    /**
+     * Txnlog fsync time
+     */
+    FSYNC_TIME(new AvgMinMaxCounter("fsynctime")),
+
+    /**
+     * Snapshot writing time
+     */
+    SNAPSHOT_TIME(new AvgMinMaxCounter("snapshottime")),
+
+    /**
+     * Db init time (snapshot loading + txnlog replay)
+     */
+    DB_INIT_TIME(new AvgMinMaxCounter("dbinittime")),
+
+    /**
+     * Stats for read request. The timing start from when the server see the
+     * request until it leave final request processor.
+     */
+    READ_LATENCY(new AvgMinMaxCounter("readlatency")),
+
+    /**
+     * 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 AvgMinMaxCounter("updatelatency")),
+
+    /**
+     * Stats for all quorum request. The timing start from when the leader
+     * see the request until it reach the learner.
+     */
+    PROPAGATION_LATENCY(new AvgMinMaxCounter("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")),
+    BYTES_RECEIVED_COUNT(new SimpleCounter("bytes_received_count"));
+
+    private final Metric metric;
+
+    ServerMetrics(Metric metric) {
+        this.metric = metric;
+    }
+
+    public void add(long value) {
+        metric.add(value);
+    }
+
+    public void reset() {
+        metric.reset();
+    }
+
+    Map<String, Long> getValues() {
+        return metric.values();
+    }
+
+    static public Map<String, Long> getAllValues() {
+        LinkedHashMap<String, Long> 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();
+        }
+    }
+}

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

@@ -21,6 +21,7 @@ package org.apache.zookeeper.server;
 
 
 import org.apache.zookeeper.common.Time;
+import org.apache.zookeeper.server.metric.AvgMinMaxCounter;
 import org.apache.zookeeper.server.quorum.BufferStats;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -33,17 +34,17 @@ import java.util.concurrent.atomic.AtomicLong;
 public class ServerStats {
     private static final Logger LOG = LoggerFactory.getLogger(ServerStats.class);
 
-    private long packetsSent;
-    private long packetsReceived;
-    private long maxLatency;
-    private long minLatency = Long.MAX_VALUE;
-    private long totalLatency = 0;
-    private long count = 0;
+    private final AtomicLong packetsSent = new AtomicLong();
+    private final AtomicLong packetsReceived = new AtomicLong();
+
+    private final AvgMinMaxCounter requestLatency = new AvgMinMaxCounter("request_latency");
+
     private 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 long getOutstandingRequests();
@@ -59,19 +60,16 @@ public class ServerStats {
     }
     
     // getters
-    synchronized public long getMinLatency() {
-        return minLatency == Long.MAX_VALUE ? 0 : minLatency;
+    public long getMinLatency() {
+        return requestLatency.getMin();
     }
 
-    synchronized public long getAvgLatency() {
-        if (count != 0) {
-            return totalLatency / count;
-        }
-        return 0;
+    public long getAvgLatency() {
+        return requestLatency.getAvg();
     }
 
-    synchronized public long getMaxLatency() {
-        return maxLatency;
+    public long getMaxLatency() {
+        return requestLatency.getMax();
     }
 
     public long getOutstandingRequests() {
@@ -90,12 +88,12 @@ public class ServerStats {
         return provider.getLogDirSize();
     }
     
-    synchronized public long getPacketsReceived() {
-        return packetsReceived;
+    public long getPacketsReceived() {
+        return packetsReceived.get();
     }
 
-    synchronized public long getPacketsSent() {
-        return packetsSent;
+    public long getPacketsSent() {
+        return packetsSent.get();
     }
 
     public String getServerState() {
@@ -107,6 +105,10 @@ public class ServerStats {
     	return provider.getNumAliveConnections();
     }
 
+    public long getUptime() {
+        return Time.currentElapsedTime() - startTime;
+    }
+
     public boolean isProviderNull() {
         return provider == null;
     }
@@ -127,36 +129,45 @@ public class ServerStats {
         sb.append("Mode: " + getServerState() + "\n");
         return sb.toString();
     }
-    // mutators
-    synchronized void updateLatency(long requestCreateTime) {
-        long latency = Time.currentElapsedTime() - requestCreateTime;
-        totalLatency += latency;
-        count++;
-        if (latency < minLatency) {
-            minLatency = latency;
+
+    /**
+     * Update request statistic. This should only be called from a request
+     * that originated from that machine.
+     */
+    public void updateLatency(Request request, long currentTime) {
+        long latency = currentTime - request.createTime;
+        if (latency < 0) {
+            return;
         }
-        if (latency > maxLatency) {
-            maxLatency = latency;
+        requestLatency.addDataPoint(latency);
+        if (request.getHdr() != null) {
+            // Only quorum request should have header
+            ServerMetrics.UPDATE_LATENCY.add(latency);
+        } else {
+            // All read request should goes here
+            ServerMetrics.READ_LATENCY.add(latency);
         }
     }
-    synchronized public void resetLatency(){
-        totalLatency = 0;
-        count = 0;
-        maxLatency = 0;
-        minLatency = Long.MAX_VALUE;
+
+    public void resetLatency() {
+        requestLatency.reset();
     }
-    synchronized public void resetMaxLatency(){
-        maxLatency = getMinLatency();
+
+    public void resetMaxLatency() {
+        requestLatency.resetMax();
     }
-    synchronized public void incrementPacketsReceived() {
-        packetsReceived++;
+
+    public void incrementPacketsReceived() {
+        packetsReceived.incrementAndGet();
     }
-    synchronized public void incrementPacketsSent() {
-        packetsSent++;
+
+    public void incrementPacketsSent() {
+        packetsSent.incrementAndGet();
     }
-    synchronized public void resetRequestCounters(){
-        packetsReceived = 0;
-        packetsSent = 0;
+
+    public void resetRequestCounters(){
+        packetsReceived.set(0);
+        packetsSent.set(0);
     }
 
     public long getFsyncThresholdExceedCount() {
@@ -171,10 +182,11 @@ public class ServerStats {
         fsyncThresholdExceedCount.set(0);
     }
 
-    synchronized public void reset() {
+    public void reset() {
         resetLatency();
         resetRequestCounters();
         clientResponseStats.reset();
+        ServerMetrics.resetAll();
     }
 
     public void updateClientResponseSize(int size) {

src/java/main/org/apache/zookeeper/server/SessionTracker.java

@@ -131,4 +131,10 @@ public interface SessionTracker {
      * Returns a mapping of time to session IDs that expire at that time.
      */
     Map<Long, Set<Long>> getSessionExpiryMap();
+
+    /**
+     * If this session tracker supports local sessions, return how many.
+     * otherwise returns 0;
+     */
+    public long getLocalSessionCount();
 }

src/java/main/org/apache/zookeeper/server/SessionTrackerImpl.java

@@ -327,4 +327,8 @@ public class SessionTrackerImpl extends ZooKeeperCriticalThread implements
             throw new KeeperException.SessionExpiredException();
         }
     }
+
+    public long getLocalSessionCount() {
+        return 0;
+    }
 }

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

@@ -216,6 +216,13 @@ public class ZKDatabase {
         return dataTree.getSessions();
     }
 
+    /**
+     * @return number of (global) sessions
+     */
+    public long getSessionCount() {
+        return sessionsWithTimeouts.size();
+    }
+
     /**
      * get sessions with timeouts
      * @return the hashmap of sessions with timeouts
@@ -237,8 +244,12 @@ public class ZKDatabase {
      * @throws IOException
      */
     public long loadDataBase() throws IOException {
+        long startTime = Time.currentElapsedTime();
         long zxid = snapLog.restore(dataTree, sessionsWithTimeouts, commitProposalPlaybackListener);
         initialized = true;
+        long loadTime = Time.currentElapsedTime() - startTime;
+        ServerMetrics.DB_INIT_TIME.add(loadTime);
+        LOG.info("Snapshot loaded in " + loadTime + " ms");
         return zxid;
     }
 

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

@@ -48,6 +48,7 @@ import org.apache.zookeeper.KeeperException;
 import org.apache.zookeeper.KeeperException.Code;
 import org.apache.zookeeper.KeeperException.SessionExpiredException;
 import org.apache.zookeeper.ZooDefs.OpCode;
+import org.apache.zookeeper.common.Time;
 import org.apache.zookeeper.data.ACL;
 import org.apache.zookeeper.data.Id;
 import org.apache.zookeeper.data.StatPersisted;
@@ -314,6 +315,7 @@ public class ZooKeeperServer implements SessionExpirer, ServerStats.Provider {
     }
 
     public void takeSnapshot(boolean syncSnap){
+        long start = Time.currentElapsedTime();
         try {
             txnLogFactory.save(zkDb.getDataTree(), zkDb.getSessionWithTimeOuts(), syncSnap);
         } catch (IOException e) {
@@ -322,6 +324,9 @@ public class ZooKeeperServer implements SessionExpirer, ServerStats.Provider {
             // so we need to exit
             System.exit(ExitCode.TXNLOG_ERROR_TAKING_SNAPSHOT.getValue());
         }
+        long elapsed = Time.currentElapsedTime() - start;
+        LOG.info("Snapshot taken in " + elapsed + " ms");
+        ServerMetrics.SNAPSHOT_TIME.add(elapsed);
     }
 
     @Override
@@ -1017,6 +1022,7 @@ public class ZooKeeperServer implements SessionExpirer, ServerStats.Provider {
                     + " client's lastZxid is 0x"
                     + Long.toHexString(connReq.getLastZxidSeen()));
         }
+        ServerMetrics.CONNECTION_REQUEST_COUNT.add(1);
         boolean readOnly = false;
         try {
             readOnly = bia.readBool("readOnly");

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

@@ -32,12 +32,15 @@ import org.apache.zookeeper.Environment.Entry;
 import org.apache.zookeeper.Version;
 import org.apache.zookeeper.server.DataTree;
 import org.apache.zookeeper.server.ServerCnxnFactory;
+import org.apache.zookeeper.server.ServerMetrics;
 import org.apache.zookeeper.server.ServerStats;
 import org.apache.zookeeper.server.ZKDatabase;
 import org.apache.zookeeper.server.ZooKeeperServer;
 import org.apache.zookeeper.server.ZooTrace;
 import org.apache.zookeeper.server.quorum.Leader;
 import org.apache.zookeeper.server.quorum.LeaderZooKeeperServer;
+import org.apache.zookeeper.server.quorum.QuorumPeer;
+import org.apache.zookeeper.server.quorum.QuorumZooKeeperServer;
 import org.apache.zookeeper.server.quorum.ReadOnlyZooKeeperServer;
 import org.apache.zookeeper.server.util.OSMXBean;
 import org.slf4j.Logger;
@@ -332,6 +335,7 @@ public class Commands {
             response.put("num_alive_connections", stats.getNumAliveClientConnections());
 
             response.put("outstanding_requests", stats.getOutstandingRequests());
+            response.put("uptime", stats.getUptime());
 
             response.put("server_state", stats.getServerState());
             response.put("znode_count", zkdb.getNodeCount());
@@ -340,6 +344,10 @@ public class Commands {
             response.put("ephemerals_count", zkdb.getDataTree().getEphemeralsCount());
             response.put("approximate_data_size", zkdb.getDataTree().cachedApproximateDataSize());
 
+            response.put("global_sessions", zkdb.getSessionCount());
+            response.put("local_sessions",
+                    zkServer.getSessionTracker().getLocalSessionCount());
+
             OSMXBean osMbean = new OSMXBean();
             response.put("open_file_descriptor_count", osMbean.getOpenFileDescriptorCount());
             response.put("max_file_descriptor_count", osMbean.getMaxFileDescriptorCount());
@@ -348,18 +356,27 @@ public class Commands {
             response.put("max_client_response_size", stats.getClientResponseStats().getMaxBufferSize());
             response.put("min_client_response_size", stats.getClientResponseStats().getMinBufferSize());
 
+            if (zkServer instanceof QuorumZooKeeperServer) {
+                QuorumPeer peer = ((QuorumZooKeeperServer) zkServer).self;
+                response.put("quorum_size", peer.getQuorumSize());
+            }
+
             if (zkServer instanceof LeaderZooKeeperServer) {
                 Leader leader = ((LeaderZooKeeperServer) zkServer).getLeader();
 
                 response.put("learners", leader.getLearners().size());
                 response.put("synced_followers", leader.getForwardingFollowers().size());
+                response.put("synced_observers", leader.getObservingLearners().size());
                 response.put("pending_syncs", leader.getNumPendingSyncs());
+                response.put("leader_uptime", leader.getUptime());
 
                 response.put("last_proposal_size", leader.getProposalStats().getLastBufferSize());
                 response.put("max_proposal_size", leader.getProposalStats().getMaxBufferSize());
                 response.put("min_proposal_size", leader.getProposalStats().getMinBufferSize());
             }
 
+            response.putAll(ServerMetrics.getAllValues());
+
             return response;
 
         }}

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

@@ -0,0 +1,113 @@
+/**
+ * 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.server.metric;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * Generic long counter that keep track of min/max/avg. The counter is
+ * thread-safe
+ */
+public class AvgMinMaxCounter implements 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 AvgMinMaxCounter(String name) {
+        this.name = name;
+    }
+
+    public void addDataPoint(long value) {
+        total.addAndGet(value);
+        count.incrementAndGet();
+        setMin(value);
+        setMax(value);
+    }
+
+    private void setMax(long value) {
+        long current;
+        while (value > (current = max.get())
+                && !max.compareAndSet(current, value))
+            ;
+    }
+
+    private void setMin(long value) {
+        long current;
+        while (value < (current = min.get())
+                && !min.compareAndSet(current, value))
+            ;
+    }
+
+    public long getAvg() {
+        // There is possible race-condition but we don't need the stats to be
+        // extremely accurate.
+        long currentCount = count.get();
+        long currentTotal = total.get();
+        if (currentCount > 0) {
+            return currentTotal / currentCount;
+        }
+        return 0;
+    }
+
+    public long getCount() {
+        return count.get();
+    }
+
+    public long getMax() {
+        long current = max.get();
+        return  (current == Long.MIN_VALUE) ? 0: current;
+    }
+
+    public long getMin() {
+        long current = min.get();
+        return  (current == Long.MAX_VALUE) ? 0: current;
+    }
+
+    public long getTotal() {
+        return total.get();
+    }
+
+    public void resetMax() {
+        max.set(getMin());
+    }
+
+    public void reset() {
+        count.set(0);
+        total.set(0);
+        min.set(Long.MAX_VALUE);
+        max.set(Long.MIN_VALUE);
+    }
+
+    public void add(long value) {
+        addDataPoint(value);
+    }
+
+    public Map<String, Long> values() {
+        Map<String, Long> m = new LinkedHashMap<String, Long>();
+        m.put("avg_" + name, this.getAvg());
+        m.put("min_" + name, this.getMin());
+        m.put("max_" + name, this.getMax());
+        m.put("cnt_" + name, this.getCount());
+        return m;
+    }
+}

src/java/main/org/apache/zookeeper/server/metric/Metric.java

@@ -0,0 +1,27 @@
+/**
+ * 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.server.metric;
+
+import java.util.Map;
+
+public interface Metric {
+    void add(long value);
+    void reset();
+    Map<String, Long> values();
+}

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

@@ -0,0 +1,53 @@
+/**
+ * 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.server.metric;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicLong;
+
+public class SimpleCounter implements Metric {
+    private final String name;
+    private final AtomicLong counter = new AtomicLong();
+
+    public SimpleCounter(String name) {
+        this.name = name;
+    }
+
+    @Override
+    public void add(long value) {
+        counter.addAndGet(value);
+    }
+
+    @Override
+    public void reset() {
+        counter.set(0);
+    }
+
+    public long getCount() {
+        return counter.get();
+    }
+
+    @Override
+    public Map<String, Long> values() {
+        Map<String, Long> m = new LinkedHashMap<String, Long>();
+        m.put(name, this.getCount());
+        return m;
+    }
+}

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

@@ -41,6 +41,7 @@ import org.apache.jute.BinaryOutputArchive;
 import org.apache.jute.InputArchive;
 import org.apache.jute.OutputArchive;
 import org.apache.jute.Record;
+import org.apache.zookeeper.server.ServerMetrics;
 import org.apache.zookeeper.server.ServerStats;
 import org.apache.zookeeper.server.util.SerializeUtils;
 import org.apache.zookeeper.txn.TxnHeader;
@@ -345,6 +346,7 @@ public class FileTxnLog implements TxnLog {
                             + "File size is " + channel.size() + " bytes. "
                             + "See the ZooKeeper troubleshooting guide");
                 }
+                ServerMetrics.FSYNC_TIME.add(syncElapsedMS);
             }
         }
         while (streamsToFlush.size() > 1) {

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

@@ -26,12 +26,14 @@ import org.apache.jute.Record;
 import org.apache.zookeeper.ZooDefs.OpCode;
 import org.apache.zookeeper.common.Time;
 import org.apache.zookeeper.server.Request;
+import org.apache.zookeeper.server.ServerMetrics;
 import org.apache.zookeeper.server.quorum.flexible.QuorumVerifier;
 import org.apache.zookeeper.server.quorum.QuorumPeer.QuorumServer;
 import org.apache.zookeeper.server.util.SerializeUtils;
 import org.apache.zookeeper.server.util.ZxidUtils;
 import org.apache.zookeeper.txn.SetDataTxn;
 import org.apache.zookeeper.txn.TxnHeader;
+
 /**
  * This class has the control logic for the Follower.
  */
@@ -66,6 +68,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);
         LOG.info("FOLLOWING - LEADER ELECTION TOOK - {} {}", electionTimeTaken,
                 QuorumPeer.FLE_TIME_UNIT);
         self.start_fle = 0;
@@ -86,7 +89,13 @@ public class Follower extends Learner{
                             + " is less than our accepted epoch " + ZxidUtils.zxidToString(self.getAcceptedEpoch()));
                     throw new IOException("Error: Epoch of leader is lower");
                 }
-                syncWithLeader(newEpochZxid);                
+                long startTime = Time.currentElapsedTime();
+                try {
+                    syncWithLeader(newEpochZxid);
+                } finally {
+                    long syncTime = Time.currentElapsedTime() - startTime;
+                    ServerMetrics.FOLLOWER_SYNC_TIME.add(syncTime);
+                }
                 QuorumPacket qp = new QuorumPacket();
                 while (this.isRunning()) {
                     readPacket(qp);

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

@@ -46,6 +46,7 @@ import org.apache.zookeeper.common.Time;
 import org.apache.zookeeper.server.FinalRequestProcessor;
 import org.apache.zookeeper.server.Request;
 import org.apache.zookeeper.server.RequestProcessor;
+import org.apache.zookeeper.server.ServerMetrics;
 import org.apache.zookeeper.server.ZooKeeperCriticalThread;
 import org.apache.zookeeper.server.quorum.QuorumPeer.LearnerType;
 import org.apache.zookeeper.server.quorum.flexible.QuorumVerifier;
@@ -425,6 +426,19 @@ public class Leader {
     // when a reconfig occurs where the leader is removed or becomes an observer,
    // it does not commit ops after committing the reconfig
     boolean allowedToCommit = true;
+
+    /**
+     * Timestamp when this leader started serving request (Quorum is running)
+     */
+    private long leaderStartTime;
+
+    public long getUptime() {
+        if (leaderStartTime > 0) {
+            return Time.currentElapsedTime() - leaderStartTime;
+        }
+        return 0;
+    }
+
     /**
      * This method is main function that is called to lead
      *
@@ -435,6 +449,7 @@ public class Leader {
         self.end_fle = Time.currentElapsedTime();
         long electionTimeTaken = self.end_fle - self.start_fle;
         self.setElectionTimeTaken(electionTimeTaken);
+        ServerMetrics.ELECTION_TIME.add(electionTimeTaken);
         LOG.info("LEADING - LEADER ELECTION TOOK - {} {}", electionTimeTaken,
                 QuorumPeer.FLE_TIME_UNIT);
         self.start_fle = 0;
@@ -988,6 +1003,7 @@ public class Leader {
         }
         QuorumPacket qp = new QuorumPacket(Leader.COMMIT, zxid, null, null);
         sendPacket(qp);
+        ServerMetrics.COMMIT_COUNT.add(1);
     }
 
     //commit and send some info
@@ -1345,6 +1361,7 @@ public class Leader {
             allowedToCommit = false;
         }
 
+        leaderStartTime = Time.currentElapsedTime();
         zk.startup();
         /*
          * Update the election vote here to ensure that all members of the

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

@@ -41,6 +41,7 @@ import org.apache.jute.Record;
 import org.apache.zookeeper.KeeperException.SessionExpiredException;
 import org.apache.zookeeper.ZooDefs.OpCode;
 import org.apache.zookeeper.server.Request;
+import org.apache.zookeeper.server.ServerMetrics;
 import org.apache.zookeeper.server.TxnLogProposalIterator;
 import org.apache.zookeeper.server.ZKDatabase;
 import org.apache.zookeeper.server.ZooKeeperThread;
@@ -474,8 +475,12 @@ public class LearnerHandler extends ZooKeeperThread {
                     bufferedOutput.flush();
                 } finally {
                     snapshot.close();
+                    ServerMetrics.SNAP_COUNT.add(1);
                 }
             }
+            else {
+                ServerMetrics.DIFF_COUNT.add(1);
+            }
 
             LOG.debug("Sending NEWLEADER message to " + sid);
             // the version of this quorumVerifier will be set by leader.lead() in case

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

@@ -51,6 +51,7 @@ import org.apache.zookeeper.common.Time;
 import org.apache.zookeeper.jmx.MBeanRegistry;
 import org.apache.zookeeper.jmx.ZKMBeanInfo;
 import org.apache.zookeeper.server.ServerCnxnFactory;
+import org.apache.zookeeper.server.ServerMetrics;
 import org.apache.zookeeper.server.ZKDatabase;
 import org.apache.zookeeper.server.ZooKeeperServer;
 import org.apache.zookeeper.server.ZooKeeperThread;
@@ -1115,6 +1116,7 @@ public class QuorumPeer extends ZooKeeperThread implements QuorumStats.Provider
                 switch (getPeerState()) {
                 case LOOKING:
                     LOG.info("LOOKING");
+                    ServerMetrics.LOOKING_COUNT.add(1);
 
                     if (Boolean.getBoolean("readonlymode.enabled")) {
                         LOG.info("Attempting to start ReadOnlyZooKeeperServer");

src/java/main/org/apache/zookeeper/server/quorum/UpgradeableSessionTracker.java

@@ -114,4 +114,11 @@ public abstract class UpgradeableSessionTracker implements SessionTracker {
             KeeperException.SessionMovedException {
         throw new UnsupportedOperationException();
     }
+
+    public long getLocalSessionCount() {
+        if (localSessionsWithTimeouts == null) {
+            return 0;
+        }
+        return localSessionsWithTimeouts.size();
+    }
 }

src/java/test/org/apache/zookeeper/server/PrepRequestProcessorTest.java

@@ -281,5 +281,9 @@ public class PrepRequestProcessorTest extends ClientBase {
         public Map<Long, Set<Long>> getSessionExpiryMap() {
             return new HashMap<Long, Set<Long>>();
         }
+        @Override
+        public long getLocalSessionCount() {
+            return 0;
+        }
     }
 }

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

@@ -0,0 +1,109 @@
+/**
+ * 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.server;
+
+import org.apache.zookeeper.ZKTestCase;
+import org.apache.zookeeper.server.metric.AvgMinMaxCounter;
+import org.apache.zookeeper.server.metric.SimpleCounter;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.Arrays;
+import java.util.Map;
+import java.util.concurrent.ThreadLocalRandom;
+
+public class ServerMetricsTest extends ZKTestCase {
+    private static final int RANDOM_TRIALS = 100;
+    private static final int RANDOM_SIZE = 100;
+
+    private long[] generateRandomValues(int size) {
+        // Clamp range to prevent overflow in metric aggregation
+        final long[] values = new long[size];
+        if (size == 0) {
+            return values;
+        }
+        final long rangeMin = Long.MIN_VALUE / size;
+        final long rangeMax = Long.MAX_VALUE / size;
+        for (int i = 0; i < size; ++i) {
+            values[i] = ThreadLocalRandom.current().nextLong(rangeMin, rangeMax);
+        }
+        return values;
+    }
+
+    @Test
+    public void testAvgMinMaxCounter() {
+        final AvgMinMaxCounter metric = new AvgMinMaxCounter("test");
+        testAvgMinMaxCounter(metric, 0);
+        testAvgMinMaxCounter(metric, 1);
+        for (int i = 0; i < RANDOM_TRIALS; ++i) {
+            testAvgMinMaxCounter(metric, RANDOM_SIZE);
+        }
+    }
+
+    private void testAvgMinMaxCounter(AvgMinMaxCounter metric, int size) {
+        final long[] values = generateRandomValues(size);
+        for (long value : values) {
+            metric.add(value);
+        }
+        long expectedMin = Arrays.stream(values).min().orElse(0);
+        long expectedMax = Arrays.stream(values).max().orElse(0);
+        long expectedSum = Arrays.stream(values).sum();
+        long expectedCnt = values.length;
+        long expectedAvg = expectedSum / Math.max(1, expectedCnt);
+
+        Assert.assertEquals(expectedAvg, metric.getAvg());
+        Assert.assertEquals(expectedMin, metric.getMin());
+        Assert.assertEquals(expectedMax, metric.getMax());
+        Assert.assertEquals(expectedCnt, metric.getCount());
+        Assert.assertEquals(expectedSum, metric.getTotal());
+
+        final Map<String, Long> results = metric.values();
+        Assert.assertEquals(expectedMax, (long)results.get("max_test"));
+        Assert.assertEquals(expectedMin, (long)results.get("min_test"));
+        Assert.assertEquals(expectedCnt, (long)results.get("cnt_test"));
+        Assert.assertEquals(expectedAvg, (long)results.get("avg_test"));
+
+        metric.reset();
+    }
+
+    @Test
+    public void testSimpleCounter() {
+        SimpleCounter metric = new SimpleCounter("test");
+        testSimpleCounter(metric, 0);
+        testSimpleCounter(metric, 1);
+        for (int i = 0; i < RANDOM_TRIALS; ++i) {
+            testSimpleCounter(metric, RANDOM_SIZE);
+        }
+    }
+
+    private void testSimpleCounter(SimpleCounter metric, int size) {
+        final long[] values = generateRandomValues(size);
+        for (long value : values) {
+            metric.add(value);
+        }
+
+        long expectedCount = Arrays.stream(values).sum();
+        Assert.assertEquals(expectedCount, metric.getCount());
+
+        final Map<String, Long> results = metric.values();
+        Assert.assertEquals(expectedCount, (long)results.get("test"));
+
+        metric.reset();
+    }
+}

src/java/test/org/apache/zookeeper/server/ServerStatsTest.java

@@ -68,8 +68,9 @@ public class ServerStatsTest extends ZKTestCase {
         ServerStats serverStats = new ServerStats(providerMock);
 
         // When incremented...
-        serverStats.updateLatency(Time.currentElapsedTime()-1000);
-        serverStats.updateLatency(Time.currentElapsedTime()-2000);
+        Request fakeRequest = new Request(0, 0, 0, null, null, 0);
+        serverStats.updateLatency(fakeRequest, fakeRequest.createTime + 1000);
+        serverStats.updateLatency(fakeRequest, fakeRequest.createTime + 2000);
 
         // Then ...
         assertThat("Max latency check", 2000L,
@@ -117,9 +118,10 @@ public class ServerStatsTest extends ZKTestCase {
         assertAllLatencyZero(serverStats);
 
         // When ...
+        Request fakeRequest = new Request(0, 0, 0, null, null, 0);
         serverStats.incrementPacketsSent();
         serverStats.incrementPacketsReceived();
-        serverStats.updateLatency(Time.currentElapsedTime()-1000);
+        serverStats.updateLatency(fakeRequest, fakeRequest.createTime + 1000);
 
         serverStats.reset();
 

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

@@ -27,10 +27,13 @@ import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
 import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.HashMap;
 import java.util.Map;
 
 import org.apache.zookeeper.server.ServerCnxnFactory;
+import org.apache.zookeeper.server.ServerMetrics;
 import org.apache.zookeeper.server.ServerStats;
 import org.apache.zookeeper.server.ZooKeeperServer;
 import org.apache.zookeeper.server.quorum.BufferStats;
@@ -166,25 +169,34 @@ public class CommandsTest extends ClientBase {
 
     @Test
     public void testMonitor() throws IOException, InterruptedException {
-        testCommand("monitor",
-                    new Field("version", String.class),
-                    new Field("avg_latency", Long.class),
-                    new Field("max_latency", Long.class),
-                    new Field("min_latency", Long.class),
-                    new Field("packets_received", Long.class),
-                    new Field("packets_sent", Long.class),
-                    new Field("num_alive_connections", Integer.class),
-                    new Field("outstanding_requests", Long.class),
-                    new Field("server_state", String.class),
-                    new Field("znode_count", Integer.class),
-                    new Field("watch_count", Integer.class),
-                    new Field("ephemerals_count", Integer.class),
-                    new Field("approximate_data_size", Long.class),
-                    new Field("open_file_descriptor_count", Long.class),
-                    new Field("max_file_descriptor_count", Long.class),
-                    new Field("last_client_response_size", Integer.class),
-                    new Field("max_client_response_size", Integer.class),
-                    new Field("min_client_response_size", Integer.class));
+        ArrayList<Field> fields = new ArrayList<>(Arrays.asList(
+                new Field("version", String.class),
+                new Field("avg_latency", Long.class),
+                new Field("max_latency", Long.class),
+                new Field("min_latency", Long.class),
+                new Field("packets_received", Long.class),
+                new Field("packets_sent", Long.class),
+                new Field("num_alive_connections", Integer.class),
+                new Field("outstanding_requests", Long.class),
+                new Field("server_state", String.class),
+                new Field("znode_count", Integer.class),
+                new Field("watch_count", Integer.class),
+                new Field("ephemerals_count", Integer.class),
+                new Field("approximate_data_size", Long.class),
+                new Field("open_file_descriptor_count", Long.class),
+                new Field("max_file_descriptor_count", Long.class),
+                new Field("last_client_response_size", Integer.class),
+                new Field("max_client_response_size", Integer.class),
+                new Field("min_client_response_size", Integer.class),
+                new Field("uptime", Long.class),
+                new Field("global_sessions", Long.class),
+                new Field("local_sessions", Long.class)
+        ));
+        for (String metric : ServerMetrics.getAllValues().keySet()) {
+            fields.add(new Field(metric, Long.class));
+        }
+        Field fieldsArray[] = fields.toArray(new Field[0]);
+        testCommand("monitor", fieldsArray);
     }
 
     @Test