ZOOKEEPER-1321. Add number of client connections metric in JMX and srvr. (Neha N. via camille)

git-svn-id: https://svn.apache.org/repos/asf/zookeeper/trunk@1242982 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -219,6 +219,7 @@ IMPROVEMENTS:
   ZOOKEEPER-1322. Cleanup/fix logging in Quorum code. 
   (phunt via mahadev)
 
+  ZOOKEEPER-1321. Add number of client connections metric in JMX and srvr. (Neha Narkhede via camille)
 Release 3.4.0 - 
 
 Non-backward compatible changes:

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

@@ -64,7 +64,7 @@ public class NIOServerCnxn extends ServerCnxn {
 
     SocketChannel sock;
 
-    private SelectionKey sk;
+    private final SelectionKey sk;
 
     boolean initialized;
 
@@ -749,6 +749,7 @@ public class NIOServerCnxn extends ServerCnxn {
 
             print("packets_received", stats.getPacketsReceived());
             print("packets_sent", stats.getPacketsSent());
+            print("num_alive_connections", stats.getNumAliveClientConnections());
 
             print("outstanding_requests", stats.getOutstandingRequests());
 

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

@@ -32,14 +32,14 @@ import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Set;
 
+import javax.security.auth.login.Configuration;
+import javax.security.auth.login.LoginException;
+
 import org.apache.zookeeper.Login;
 import org.apache.zookeeper.server.auth.SaslServerCallbackHandler;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import javax.security.auth.login.Configuration;
-import javax.security.auth.login.LoginException;
-
 public class NIOServerCnxnFactory extends ServerCnxnFactory implements Runnable {
     private static final Logger LOG = LoggerFactory.getLogger(NIOServerCnxnFactory.class);
 
@@ -72,13 +72,11 @@ public class NIOServerCnxnFactory extends ServerCnxnFactory implements Runnable
     */
     final ByteBuffer directBuffer = ByteBuffer.allocateDirect(64 * 1024);
 
-    final HashSet<ServerCnxn> cnxns = new HashSet<ServerCnxn>();
     final HashMap<InetAddress, Set<NIOServerCnxn>> ipMap =
         new HashMap<InetAddress, Set<NIOServerCnxn>>( );
 
     int maxClientCnxns = 60;
 
-
     /**
      * Construct a new server connection factory which will accept an unlimited number
      * of concurrent connections from each client (up to the file descriptor
@@ -322,5 +320,4 @@ public class NIOServerCnxnFactory extends ServerCnxnFactory implements Runnable
     public Iterable<ServerCnxn> getConnections() {
         return cnxns;
     }
-
 }

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

@@ -560,6 +560,7 @@ public class NettyServerCnxn extends ServerCnxn {
 
             print("packets_received", stats.getPacketsReceived());
             print("packets_sent", stats.getPacketsSent());
+            print("num_alive_connections", stats.getNumAliveClientConnections());
 
             print("outstanding_requests", stats.getOutstandingRequests());
 

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

@@ -56,7 +56,6 @@ public class NettyServerCnxnFactory extends ServerCnxnFactory {
     ServerBootstrap bootstrap;
     Channel parentChannel;
     ChannelGroup allChannels = new DefaultChannelGroup("zkServerCnxns");
-    HashSet<ServerCnxn> cnxns = new HashSet<ServerCnxn>();
     HashMap<InetAddress, Set<NettyServerCnxn>> ipMap =
         new HashMap<InetAddress, Set<NettyServerCnxn>>( );
     InetSocketAddress localAddress;

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

@@ -22,13 +22,15 @@ import java.io.IOException;
 import java.net.InetSocketAddress;
 import java.nio.ByteBuffer;
 import java.util.HashMap;
+import java.util.HashSet;
 
 import javax.management.JMException;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.apache.zookeeper.jmx.MBeanRegistry;
+
 import org.apache.zookeeper.Login;
+import org.apache.zookeeper.jmx.MBeanRegistry;
 import org.apache.zookeeper.server.auth.SaslServerCallbackHandler;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 public abstract class ServerCnxnFactory {
 
@@ -49,6 +51,12 @@ public abstract class ServerCnxnFactory {
     
     public abstract Iterable<ServerCnxn> getConnections();
 
+    public int getNumAliveConnections() {
+        synchronized(cnxns) {
+            return cnxns.size();
+        }
+    }
+
     public abstract void closeSession(long sessionId);
 
     public abstract void configure(InetSocketAddress addr,
@@ -115,7 +123,9 @@ public abstract class ServerCnxnFactory {
 
     public abstract InetSocketAddress getLocalAddress();
 
-    private HashMap<ServerCnxn, ConnectionBean> connectionBeans = new HashMap<ServerCnxn, ConnectionBean>();
+    private final HashMap<ServerCnxn, ConnectionBean> connectionBeans = new HashMap<ServerCnxn, ConnectionBean>();
+
+    protected final HashSet<ServerCnxn> cnxns = new HashSet<ServerCnxn>();
     public void unregisterConnection(ServerCnxn serverCnxn) {
         ConnectionBean jmxConnectionBean = connectionBeans.remove(serverCnxn);
         if (jmxConnectionBean != null){

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

@@ -19,6 +19,7 @@
 package org.apache.zookeeper.server;
 
 
+
 /**
  * Basic Server Statistics
  */
@@ -36,6 +37,7 @@ public class ServerStats {
         public long getOutstandingRequests();
         public long getLastProcessedZxid();
         public String getState();
+        public int getNumAliveConnections();
     }
     
     public ServerStats(Provider provider) {
@@ -78,6 +80,11 @@ public class ServerStats {
         return provider.getState();
     }
     
+    /** The number of client connections alive to this server */
+    public int getNumAliveClientConnections() {
+    	return provider.getNumAliveConnections();
+    }
+
     @Override
     public String toString(){
         StringBuilder sb = new StringBuilder();
@@ -85,6 +92,8 @@ public class ServerStats {
                 + getAvgLatency() + "/" + getMaxLatency() + "\n");
         sb.append("Received: " + getPacketsReceived() + "\n");
         sb.append("Sent: " + getPacketsSent() + "\n");
+        sb.append("Connections: " + getNumAliveClientConnections() + "\n");
+
         if (provider != null) {
             sb.append("Outstanding: " + getOutstandingRequests() + "\n");
             sb.append("Zxid: 0x"+ Long.toHexString(getLastProcessedZxid())+ "\n");
@@ -123,7 +132,6 @@ public class ServerStats {
         packetsReceived = 0;
         packetsSent = 0;
     }
-    
     synchronized public void reset() {
         resetLatency();
         resetRequestCounters();

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

@@ -256,7 +256,6 @@ public class ZooKeeperServer implements SessionExpirer, ServerStats.Provider {
         }
     }
 
-
     /**
      * This should be called from a synchronized block on this!
      */
@@ -685,6 +684,14 @@ public class ZooKeeperServer implements SessionExpirer, ServerStats.Provider {
         return getInProcess();
     }
 
+    /**
+     * return the total number of client connections that are alive
+     * to this server
+     */
+    public int getNumAliveConnections() {
+        return serverCnxnFactory.getNumAliveConnections();
+    }
+
     /**
      * trunccate the log to get in sync with others
      * if in a quorum

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

@@ -140,4 +140,8 @@ public class ZooKeeperServerBean implements ZooKeeperServerMXBean, ZKMBeanInfo {
         serverStats.resetRequestCounters();
         serverStats.resetLatency();
     }
+
+    public long getNumAliveConnections() {
+        return zks.getNumAliveConnections();
+    }
 }

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

@@ -103,4 +103,8 @@ public interface ZooKeeperServerMXBean {
      * Reset max latency statistics only.
      */
     public void resetMaxLatency();
+    /**
+     * @return number of alive client connections
+     */
+    public long getNumAliveConnections();
 }

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

@@ -282,6 +282,10 @@ public class Zab1_0Test {
         }
         public void closeAll() {
         }
+        @Override
+        public int getNumAliveConnections() {
+            return 0;
+        }
     }
     static Socket[] getSocketPair() throws IOException {
         ServerSocket ss = new ServerSocket();

src/java/test/org/apache/zookeeper/test/FourLetterWordsTest.java

@@ -94,6 +94,9 @@ public class FourLetterWordsTest extends ClientBase {
         verify("srvr", "Outstanding");
         verify("cons", "queued");
         verify("mntr", "zk_server_state\tstandalone");
+        verify("mntr", "num_alive_connections");
+        verify("stat", "Connections");
+        verify("srvr", "Connections");
     }
 
     private String sendRequest(String cmd) throws IOException {
@@ -136,6 +139,8 @@ public class FourLetterWordsTest extends ClientBase {
         line = in.readLine();
         Assert.assertTrue(Pattern.matches("^Sent: \\d+$", line));
         line = in.readLine();
+        Assert.assertTrue(Pattern.matches("^Connections: \\d+$", line));
+        line = in.readLine();
         Assert.assertTrue(Pattern.matches("^Outstanding: \\d+$", line));
         line = in.readLine();
         Assert.assertTrue(Pattern.matches("^Zxid: 0x[\\da-fA-F]+$", line));