ZOOKEEPER-2940: Deal with maxbuffer as it relates to large requests from clients

https://issues.apache.org/jira/browse/ZOOKEEPER-2940

This PR covers the other part of Jute buffer monitoring which relates to client response sizes.

`jute.maxbuffer` is often set too small or the default (4MB) value is not enough to receive large responses from the server which causes the client unable to operate. e.g. `getChildren()` on a node which has lots of children or execution of a large multi.

These new metrics lets operators to monitor the size of generated responses to get an idea on how to properly set client's `jute.maxbuffer`.

The other part of monitoring which relates to proposal size has already been merged apache#415

Author: Andor Molnar <[email protected]>

Reviewers: [email protected]

Closes apache#466 from anmolnar/ZOOKEEPER-2940 and squashes the following commits:

29224d0 [Andor Molnar] ZOOKEEPER-2940. Test fixes
486d496 [Andor Molnar] ZOOKEEPER-2940. Removed support of 4lw commands, because master is targeted
60916e0 [Andor Molnar] ZOOKEEPER-2940. Code review fixes
b39c4c8 [Andor Molnar] ZOOKEEPER-2940. Code review: variable names renamed to proposalStats
87e72ff [Andor Molnar] ZOOKEEPER-2940. Some code review cleanups
6f0adb1 [Andor Molnar] ZOOKEEPER-2940. Code review fixes: stats class renamed, added comments to unit test asserts, updated docs
bcb6d4c [Andor Molnar] ZOOKEEPER-2940. Added new metrics to python files
77a5f0a [Andor Molnar] ZOOKEEPER-2940. Fixed failing unit tests
2def61c [Andor Molnar] ZOOKEEPER-2940. Track size of client responses

Change-Id: I68df371097cc355b1f41b8dbb0168da4fa5dee43

Author: anmolnar

docs/index.pdf

@@ -638,7 +638,9 @@ <h3 class="h4">Clustered (Multi-Server) Setup</h3>
           consists of a single line containing only the text of that machine's
           id. So <span class="codefrag filename">myid</span> of server 1 would contain the text
           "1" and nothing else. The id must be unique within the
-          ensemble and should have a value between 1 and 255.</p>
+          ensemble and should have a value between 1 and 255. <strong>IMPORTANT:</strong> if you
+          enable extended features such as TTL Nodes (see below) the id must be
+          between 1 and 254 due to internal limitations.</p>
 
 </li>
 
@@ -1345,6 +1347,37 @@ <h4>Advanced Configuration</h4>
               feature. Default is "true"</p>
 </dd>
 
+          
+<dt>
+<term>zookeeper.extendedTypesEnabled</term>
+</dt>
+<dd>
+<p>(Java system property only: <strong>zookeeper.extendedTypesEnabled</strong>)</p>
+<p>
+<strong>New in 3.5.4, 3.6.0:</strong> Define to "true" to enable
+              extended features such as the creation of <a href="zookeeperProgrammers.html#TTL+Nodes">TTL Nodes</a>.
+              They are disabled by default. IMPORTANT: when enabled server IDs must
+              be less than 255 due to internal limitations.
+              </p>
+</dd>
+
+          
+<dt>
+<term>zookeeper.emulate353TTLNodes</term>
+</dt>
+<dd>
+<p>(Java system property only: <strong>zookeeper.emulate353TTLNodes</strong>)</p>
+<p>
+<strong>New in 3.5.4, 3.6.0:</strong> Due to
+                <a href="https://issues.apache.org/jira/browse/ZOOKEEPER-2901">ZOOKEEPER-2901</a> TTL nodes
+                created in version 3.5.3 are not supported in 3.5.4/3.6.0. However, a workaround is provided via the
+                zookeeper.emulate353TTLNodes system property. If you used TTL nodes in ZooKeeper 3.5.3 and need to maintain
+                compatibility set <strong>zookeeper.emulate353TTLNodes</strong> to "true" in addition to
+                <strong>zookeeper.extendedTypesEnabled</strong>. NOTE: due to the bug, server IDs
+                must be 127 or less. Additionally, the maximum support TTL value is 1099511627775 which is smaller
+                than what was allowed in 3.5.3 (1152921504606846975)</p>
+</dd>
+
 
 </dl>
 <a name="sc_clusterOptions"></a>
@@ -2216,6 +2249,7 @@ <h4>The Four Letter Words</h4>
               zk_min_latency  0
               zk_packets_received 70
               zk_packets_sent 69
+              zk_num_alive_connections	1
               zk_outstanding_requests 0
               zk_server_state leader
               zk_znode_count   4
@@ -2227,6 +2261,9 @@ <h4>The Four Letter Words</h4>
               zk_pending_syncs    0               - only exposed by the Leader
               zk_open_file_descriptor_count 23    - only available on Unix platforms
               zk_max_file_descriptor_count 1024   - only available on Unix platforms
+              zk_last_proposal_size 23
+              zk_min_proposal_size 23
+              zk_max_proposal_size 64
               </pre>
 <p>The output is compatible with java properties format and the content 
               may change over time (new keys added). Your scripts should expect changes.</p>

docs/javaExample.pdf

@@ -596,6 +596,10 @@ <h4>TTL Nodes</h4>
           you can optionally set a TTL in milliseconds for the znode. If the znode
           is not modified within the TTL and has no children it will become a candidate
           to be deleted by the server at some point in the future.</p>
+<p>Note: TTL Nodes must be enabled via System property as
+        they are disabled by default. See the <a href="zookeeperAdmin.html#sc_configuration">Administrator's
+        Guide</a> for details. If you attempt to create TTL Nodes without the proper System property set the server
+        will throw <em>KeeperException.UnimplementedException</em>.</p>
 <a name="sc_timeInZk"></a>
 <h3 class="h4">Time in ZooKeeper</h3>
 <p>ZooKeeper tracks time multiple ways:</p>

docs/linkmap.pdf

@@ -293,7 +293,14 @@ def _parse_stat(self, data):
                 result['zk_zxid_epoch']   = int(m.group(1), 16) >>32 # high 32 bits
                 continue
 
-        return result 
+            m = re.match('Proposal sizes last/min/max: (\d+)/(\d+)/(\d+)', line)
+            if m is not None:
+                result['zk_last_proposal_size'] = int(m.group(1))
+                result['zk_min_proposal_size'] = int(m.group(2))
+                result['zk_max_proposal_size'] = int(m.group(3))
+                continue
+
+        return result
 
     def _parse_line(self, line):
         try:

docs/recipes.pdf

@@ -46,5 +46,8 @@ collection_group {
   metric { name = "zk_followers" }
   metric { name = "zk_synced_followers" }
   metric { name = "zk_pending_syncs" }
+  metric { name = "zk_last_proposal_size" }
+  metric { name = "zk_min_proposal_size" }
+  metric { name = "zk_max_proposal_size" }
 }
 

docs/zookeeperAdmin.html

@@ -214,7 +214,10 @@ def metric_init(params=None):
         'zk_max_file_descriptor_count': {'units': 'descriptors'},
         'zk_followers': {'units': 'nodes'},
         'zk_synced_followers': {'units': 'nodes'},
-        'zk_pending_syncs': {'units': 'syncs'}
+        'zk_pending_syncs': {'units': 'syncs'},
+        'zk_last_proposal_size': {'units': 'bytes'},
+        'zk_min_proposal_size': {'units': 'bytes'},
+        'zk_max_proposal_size': {'units': 'bytes'}
     }
     metric_handler.descriptors = {}
     for name, updates in metrics.iteritems():

docs/zookeeperAdmin.pdf

@@ -115,7 +115,7 @@ void updateLastSendAndHeard() {
         this.lastHeard = now;
     }
 
-    protected void readLength() throws IOException {
+    void readLength() throws IOException {
         int len = incomingBuffer.getInt();
         if (len < 0 || len >= packetLen) {
             throw new IOException("Packet len " + len + " is out of range!");

docs/zookeeperHierarchicalQuorums.pdf

@@ -678,8 +678,6 @@ public static void closeSock(SocketChannel sock) {
         }
     }
 
-    private final static byte fourBytes[] = new byte[4];
-
     /*
      * (non-Javadoc)
      *
@@ -689,23 +687,7 @@ public static void closeSock(SocketChannel sock) {
     @Override
     public void sendResponse(ReplyHeader h, Record r, String tag) {
         try {
-            ByteArrayOutputStream baos = new ByteArrayOutputStream();
-            // Make space for length
-            BinaryOutputArchive bos = BinaryOutputArchive.getArchive(baos);
-            try {
-                baos.write(fourBytes);
-                bos.writeRecord(h, "header");
-                if (r != null) {
-                    bos.writeRecord(r, tag);
-                }
-                baos.close();
-            } catch (IOException e) {
-                LOG.error("Error serializing response");
-            }
-            byte b[] = baos.toByteArray();
-            ByteBuffer bb = ByteBuffer.wrap(b);
-            bb.putInt(b.length - 4).rewind();
-            sendBuffer(bb);
+            super.sendResponse(h, r, tag);
             if (h.getXid() > 0) {
                 // check throttling
                 if (outstandingRequests.decrementAndGet() < 1 ||

docs/zookeeperInternals.pdf

@@ -160,7 +160,6 @@ public void process(WatchedEvent event) {
         }
     }
 
-    private static final byte[] fourBytes = new byte[4];
     static class ResumeMessageEvent implements MessageEvent {
         Channel channel;
         ResumeMessageEvent(Channel channel) {
@@ -182,23 +181,7 @@ public void sendResponse(ReplyHeader h, Record r, String tag)
         if (closingChannel || !channel.isOpen()) {
             return;
         }
-        ByteArrayOutputStream baos = new ByteArrayOutputStream();
-        // Make space for length
-        BinaryOutputArchive bos = BinaryOutputArchive.getArchive(baos);
-        try {
-            baos.write(fourBytes);
-            bos.writeRecord(h, "header");
-            if (r != null) {
-                bos.writeRecord(r, tag);
-            }
-            baos.close();
-        } catch (IOException e) {
-            LOG.error("Error serializing response");
-        }
-        byte b[] = baos.toByteArray();
-        ByteBuffer bb = ByteBuffer.wrap(b);
-        bb.putInt(b.length - 4).rewind();
-        sendBuffer(bb);
+        super.sendResponse(h, r, tag);
         if (h.getXid() > 0) {
             // zks cannot be null otherwise we would not have gotten here!
             if (!zkServer.shouldThrottle(outstandingCount.decrementAndGet())) {

docs/zookeeperJMX.pdf

@@ -18,6 +18,7 @@
 
 package org.apache.zookeeper.server;
 
+import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.io.PrintWriter;
 import java.io.StringWriter;
@@ -34,6 +35,7 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicLong;
 
+import org.apache.jute.BinaryOutputArchive;
 import org.apache.jute.Record;
 import org.apache.zookeeper.WatchedEvent;
 import org.apache.zookeeper.Watcher;
@@ -55,6 +57,8 @@ public abstract class ServerCnxn implements Stats, Watcher {
 
     private Set<Id> authInfo = Collections.newSetFromMap(new ConcurrentHashMap<Id, Boolean>());
 
+    private static final byte[] fourBytes = new byte[4];
+
     /**
      * If the client is of old version, we don't send r-o mode info to it.
      * The reason is that if we would, old C client doesn't read it, which
@@ -66,8 +70,26 @@ public abstract class ServerCnxn implements Stats, Watcher {
 
     abstract void close();
 
-    public abstract void sendResponse(ReplyHeader h, Record r, String tag)
-        throws IOException;
+    public void sendResponse(ReplyHeader h, Record r, String tag) throws IOException {
+        ByteArrayOutputStream baos = new ByteArrayOutputStream();
+        // Make space for length
+        BinaryOutputArchive bos = BinaryOutputArchive.getArchive(baos);
+        try {
+            baos.write(fourBytes);
+            bos.writeRecord(h, "header");
+            if (r != null) {
+                bos.writeRecord(r, tag);
+            }
+            baos.close();
+        } catch (IOException e) {
+            LOG.error("Error serializing response");
+        }
+        byte b[] = baos.toByteArray();
+        serverStats().updateClientResponseSize(b.length - 4);
+        ByteBuffer bb = ByteBuffer.wrap(b);
+        bb.putInt(b.length - 4).rewind();
+        sendBuffer(bb);
+    }
 
     /* notify the client the session is closing and close/cleanup socket */
     abstract void sendCloseSession();
@@ -133,12 +155,12 @@ protected void packetSent() {
         incrPacketsSent();
         ServerStats serverStats = serverStats();
         if (serverStats != null) {
-            serverStats().incrementPacketsSent();
+            serverStats.incrementPacketsSent();
         }
     }
 
     protected abstract ServerStats serverStats();
-    
+
     protected final Date established = new Date();
 
     protected final AtomicLong packetsReceived = new AtomicLong();

docs/zookeeperObservers.pdf

@@ -21,13 +21,18 @@
 
 
 import org.apache.zookeeper.common.Time;
+import org.apache.zookeeper.server.quorum.BufferStats;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.util.concurrent.atomic.AtomicLong;
 
 /**
  * Basic Server Statistics
  */
 public class ServerStats {
+    private static final Logger LOG = LoggerFactory.getLogger(ServerStats.class);
+
     private long packetsSent;
     private long packetsReceived;
     private long maxLatency;
@@ -36,6 +41,8 @@ public class ServerStats {
     private long count = 0;
     private AtomicLong fsyncThresholdExceedCount = new AtomicLong(0);
 
+    private final BufferStats clientResponseStats = new BufferStats();
+
     private final Provider provider;
 
     public interface Provider {
@@ -167,6 +174,14 @@ public void resetFsyncThresholdExceedCount() {
     synchronized public void reset() {
         resetLatency();
         resetRequestCounters();
+        clientResponseStats.reset();
     }
 
+    public void updateClientResponseSize(int size) {
+        clientResponseStats.setLastBufferSize(size);
+    }
+
+    public BufferStats getClientResponseStats() {
+        return clientResponseStats;
+    }
 }

docs/zookeeperOver.pdf

@@ -182,4 +182,19 @@ public long getTxnLogElapsedSyncTime() {
     public int getJuteMaxBufferSize() {
         return BinaryInputArchive.maxBuffer;
     }
+
+    @Override
+    public int getLastClientResponseSize() {
+        return zks.serverStats().getClientResponseStats().getLastBufferSize();
+    }
+
+    @Override
+    public int getMinClientResponseSize() {
+        return zks.serverStats().getClientResponseStats().getMinBufferSize();
+    }
+
+    @Override
+    public int getMaxClientResponseSize() {
+        return zks.serverStats().getClientResponseStats().getMaxBufferSize();
+    }
 }

docs/zookeeperProgrammers.html

@@ -143,4 +143,19 @@ public interface ZooKeeperServerMXBean {
      * @return Returns the value of the following config setting: jute.maxbuffer
      */
     public int getJuteMaxBufferSize();
+
+    /**
+     * @return size of latest generated client response
+     */
+    public int getLastClientResponseSize();
+
+    /**
+     * @return size of smallest generated client response
+     */
+    public int getMinClientResponseSize();
+
+    /**
+     * @return size of largest generated client response
+     */
+    public int getMaxClientResponseSize();
 }

docs/zookeeperProgrammers.pdf

@@ -330,12 +330,20 @@ public CommandResponse run(ZooKeeperServer zkServer, Map<String, String> kwargs)
             response.put("open_file_descriptor_count", osMbean.getOpenFileDescriptorCount());
             response.put("max_file_descriptor_count", osMbean.getMaxFileDescriptorCount());
 
+            response.put("last_client_response_size", stats.getClientResponseStats().getLastBufferSize());
+            response.put("max_client_response_size", stats.getClientResponseStats().getMaxBufferSize());
+            response.put("min_client_response_size", stats.getClientResponseStats().getMinBufferSize());
+
             if (zkServer instanceof LeaderZooKeeperServer) {
                 Leader leader = ((LeaderZooKeeperServer) zkServer).getLeader();
 
                 response.put("followers", leader.getLearners().size());
                 response.put("synced_followers", leader.getForwardingFollowers().size());
                 response.put("pending_syncs", leader.getNumPendingSyncs());
+
+                response.put("last_proposal_size", leader.getProposalStats().getLastBufferSize());
+                response.put("max_proposal_size", leader.getProposalStats().getMaxBufferSize());
+                response.put("min_proposal_size", leader.getProposalStats().getMinBufferSize());
             }
 
             return response;
@@ -415,9 +423,13 @@ public CommandResponse run(ZooKeeperServer zkServer, Map<String, String> kwargs)
             response.put("version", Version.getFullVersion());
             response.put("read_only", zkServer instanceof ReadOnlyZooKeeperServer);
             response.put("server_stats", zkServer.serverStats());
+            response.put("client_response", zkServer.serverStats().getClientResponseStats());
+            if (zkServer instanceof LeaderZooKeeperServer) {
+                Leader leader = ((LeaderZooKeeperServer)zkServer).getLeader();
+                response.put("proposal_stats", leader.getProposalStats());
+            }
             response.put("node_count", zkServer.getZKDatabase().getNodeCount());
             return response;
-
         }
     }
 

docs/zookeeperQuotas.pdf

@@ -75,9 +75,9 @@ public void commandRun() {
             print("synced_followers", leader.getForwardingFollowers().size());
             print("pending_syncs", leader.getNumPendingSyncs());
 
-            print("last_proposal_size", leader.getProposalStats().getLastProposalSize());
-            print("max_proposal_size", leader.getProposalStats().getMaxProposalSize());
-            print("min_proposal_size", leader.getProposalStats().getMinProposalSize());
+            print("last_proposal_size", leader.getProposalStats().getLastBufferSize());
+            print("max_proposal_size", leader.getProposalStats().getMaxBufferSize());
+            print("min_proposal_size", leader.getProposalStats().getMinBufferSize());
         }
     }