Add various operational metrics

Add the following metrics:
* Max/Min/Avg latency for read and write requests.
* Total request count for read and write requests.
* Connection/session create and close count.

Author: Jiajun Wang

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

@@ -201,15 +201,19 @@ public void processRequest(Request request) {
             AuditHelper.addAuditLog(request, rc);
             switch (request.type) {
             case OpCode.ping: {
+                zks.serverStats().updateLatency(request.type, request, Time.currentElapsedTime());
+
                 lastOp = "PING";
-                updateStats(request, lastOp, lastZxid);
+                updateStats(request.type, request, lastOp, lastZxid);
 
                 cnxn.sendResponse(new ReplyHeader(ClientCnxn.PING_XID, lastZxid, 0), null, "response");
                 return;
             }
             case OpCode.createSession: {
+                zks.serverStats().updateLatency(request.type, request, Time.currentElapsedTime());
+
                 lastOp = "SESS";
-                updateStats(request, lastOp, lastZxid);
+                updateStats(request.type, request, lastOp, lastZxid);
 
                 zks.finishSessionInit(request.cnxn, true);
                 return;
@@ -615,7 +619,28 @@ public void processRequest(Request request) {
 
         ReplyHeader hdr = new ReplyHeader(request.cxid, lastZxid, err.intValue());
 
-        updateStats(request, lastOp, lastZxid);
+        int type = request.type;
+        if (type == OpCode.multi) {
+            // check if contains only read operation
+            boolean containsWrite = false;
+            for (ProcessTxnResult subTxnResult : rc.multiResult) {
+                switch (subTxnResult.type) {
+                    case OpCode.create:
+                    case OpCode.delete:
+                    case OpCode.setData:
+                        containsWrite = true;
+                        break;
+                    case OpCode.check:
+                    case OpCode.error:
+                    default:
+                        break;
+                }
+            }
+            if (!containsWrite) {
+                type = OpCode.check;
+            }
+        }
+        updateStats(type, request, lastOp, lastZxid);
 
         try {
             if (path == null || rsp == null) {
@@ -694,13 +719,12 @@ public void shutdown() {
         LOG.info("shutdown of request processor complete");
     }
 
-    private void updateStats(Request request, String lastOp, long lastZxid) {
+    private void updateStats(int type, Request request, String lastOp, long lastZxid) {
         if (request.cnxn == null) {
             return;
         }
         long currentTime = Time.currentElapsedTime();
-        zks.serverStats().updateLatency(request, currentTime);
+        zks.serverStats().updateLatency(type, request, currentTime);
         request.cnxn.updateStatsForResponse(request.cxid, lastZxid, lastOp, request.createTime, currentTime);
     }
-
 }

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

@@ -830,6 +830,7 @@ private void addCnxn(NIOServerCnxn cnxn) throws IOException {
         set.add(cnxn);
 
         cnxns.add(cnxn);
+        increaseConnectionCreateCount();
         touchCnxn(cnxn);
     }
 

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

@@ -722,15 +722,18 @@ public InetSocketAddress getLocalAddress() {
 
     private void addCnxn(final NettyServerCnxn cnxn) {
         cnxns.add(cnxn);
-        InetAddress addr = ((InetSocketAddress) cnxn.getChannel().remoteAddress()).getAddress();
-
-        ipMap.compute(addr, (a, cnxnCount) -> {
-            if (cnxnCount == null) {
-                cnxnCount = new AtomicInteger();
-            }
-            cnxnCount.incrementAndGet();
-            return cnxnCount;
-        });
+        increaseConnectionCreateCount();
+        synchronized (ipMap) {
+            InetAddress addr =
+                ((InetSocketAddress) cnxn.getChannel().remoteAddress()).getAddress();
+            ipMap.compute(addr, (a, cnxnCount) -> {
+                if (cnxnCount == null) {
+                    cnxnCount = new AtomicInteger();
+                }
+                cnxnCount.incrementAndGet();
+                return cnxnCount;
+            });
+        }
     }
 
     void removeCnxnFromIpMap(NettyServerCnxn cnxn, InetAddress remoteAddress) {

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

@@ -200,13 +200,42 @@ public static ServerCnxnFactory createFactory(InetSocketAddress addr, int maxCli
 
     private final ConcurrentHashMap<ServerCnxn, ConnectionBean> connectionBeans = new ConcurrentHashMap<ServerCnxn, ConnectionBean>();
 
+    // Connection create count is increased after connection established. But this connection may not establish a session.
+    // In this case, registerConnection won't be called.
+    // So the following method need to be called in the implementation class.
+    public void increaseConnectionCreateCount() {
+        if (zkServer != null) {
+            zkServer.serverStats().incrementConnectionCreated();
+        }
+    }
+
+    private void increaseConnectionCloseCount() {
+        if (zkServer != null) {
+            zkServer.serverStats().incrementConnectionClosed();
+        }
+    }
+
+    private void increaseSessionCreateCount() {
+        if (zkServer != null) {
+            zkServer.serverStats().incrementSessionCreated();
+        }
+    }
+
+    private void increaseSessionCloseCount() {
+        if (zkServer != null) {
+            zkServer.serverStats().incrementSessionClosed();
+        }
+    }
+
     // Connection set is relied on heavily by four letter commands
     // Construct a ConcurrentHashSet using a ConcurrentHashMap
     protected final Set<ServerCnxn> cnxns = Collections.newSetFromMap(new ConcurrentHashMap<ServerCnxn, Boolean>());
     public void unregisterConnection(ServerCnxn serverCnxn) {
+        increaseConnectionCloseCount();
         ConnectionBean jmxConnectionBean = connectionBeans.remove(serverCnxn);
         if (jmxConnectionBean != null) {
             MBeanRegistry.getInstance().unregister(jmxConnectionBean);
+            increaseSessionCloseCount();
         }
     }
 
@@ -216,6 +245,7 @@ public void registerConnection(ServerCnxn serverCnxn) {
             try {
                 MBeanRegistry.getInstance().register(jmxConnectionBean, zkServer.jmxServerBean);
                 connectionBeans.put(serverCnxn, jmxConnectionBean);
+                increaseSessionCreateCount();
             } catch (JMException e) {
                 LOG.warn("Could not register connection", e);
             }