ZOOKEEPER-3324: Add read/write metrics for top level znodes

another example of Set metrics.

Author: Jie Huang <jiehuang@fb.com>

Reviewers: andor@apache.org

Closes #860 from jhuan31/ZOOKEEPER-3324

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

@@ -134,6 +134,11 @@ public class DataTree {
      */
     private final PathTrie pTrie = new PathTrie();
 
+    /**
+     * over-the-wire size of znode's stat. Counting the fields of Stat class
+     */
+    public static final int STAT_OVERHEAD_BYTES = (6 * 8) + (5 * 4);
+
     /**
      * This hashtable lists the paths of the ephemeral nodes of a session.
      */
@@ -509,6 +514,7 @@ public class DataTree {
             // ok we have some match and need to update
             updateCountBytes(lastPrefix, bytes, 1);
         }
+        updateWriteStat(path, bytes);
         dataWatches.triggerWatch(path, Event.EventType.NodeCreated);
         childWatches.triggerWatch(parentName.equals("") ? "/" : parentName,
                 Event.EventType.NodeChildrenChanged);
@@ -592,6 +598,9 @@ public class DataTree {
             }
             updateCountBytes(lastPrefix, bytes,-1);
         }
+
+        updateWriteStat(path, 0L);
+
         if (LOG.isTraceEnabled()) {
             ZooTrace.logTraceMessage(LOG, ZooTrace.EVENT_DELIVERY_TRACE_MASK,
                     "dataWatches.triggerWatch " + path);
@@ -623,12 +632,14 @@ public class DataTree {
         }
         // now update if the path is in a quota subtree.
         String lastPrefix = getMaxPrefixWithQuota(path);
+        long dataBytes = data == null ? 0 : data.length;
         if(lastPrefix != null) {
-            long dataBytes = data == null ? 0 : data.length;
             this.updateCountBytes(lastPrefix, dataBytes
                     - (lastdata == null ? 0 : lastdata.length), 0);
         }
         nodeDataSize.addAndGet(getNodeSize(path, data) - getNodeSize(path, lastdata));
+
+        updateWriteStat(path, dataBytes);
         dataWatches.triggerWatch(path, EventType.NodeDataChanged);
         return s;
     }
@@ -656,6 +667,7 @@ public class DataTree {
     public byte[] getData(String path, Stat stat, Watcher watcher)
             throws KeeperException.NoNodeException {
         DataNode n = nodes.get(path);
+        byte[] data = null;
         if (n == null) {
             throw new KeeperException.NoNodeException();
         }
@@ -664,8 +676,10 @@ public class DataTree {
             if (watcher != null) {
                 dataWatches.addWatch(path, watcher);
             }
-            return n.data;
+            data = n.data;
         }
+        updateReadStat(path, data == null ? 0 : data.length);
+        return data;
     }
 
     public Stat statNode(String path, Watcher watcher)
@@ -680,8 +694,9 @@ public class DataTree {
         }
         synchronized (n) {
             n.copyStat(stat);
-            return stat;
         }
+        updateReadStat(path, 0L);
+        return stat;
     }
 
     public List<String> getChildren(String path, Stat stat, Watcher watcher)
@@ -690,17 +705,25 @@ public class DataTree {
         if (n == null) {
             throw new KeeperException.NoNodeException();
         }
+        List<String> children;
         synchronized (n) {
             if (stat != null) {
                 n.copyStat(stat);
             }
-            List<String> children=new ArrayList<String>(n.getChildren());
+            children = new ArrayList<String>(n.getChildren());
 
             if (watcher != null) {
                 childWatches.addWatch(path, watcher);
             }
-            return children;
         }
+
+        int bytes = 0;
+        for (String child : children) {
+            bytes += child.length();
+        }
+        updateReadStat(path, bytes);
+
+        return children;
     }
 
     public int getAllChildrenNumber(String path) {
@@ -1504,4 +1527,26 @@ public class DataTree {
     public ReferenceCountedACLCache getReferenceCountedAclCache() {
         return aclCache;
     }
+
+    private String getTopNamespace(String path) {
+        String[] parts = path.split("/");
+        return parts.length > 1 ? parts[1] : null;
+    }
+
+    private void updateReadStat(String path, long bytes) {
+        String namespace = getTopNamespace(path);
+        if (namespace == null) {
+            return;
+        }
+        long totalBytes = path.length() + bytes + STAT_OVERHEAD_BYTES;
+        ServerMetrics.READ_PER_NAMESPACE.add(namespace, totalBytes);
+    }
+
+    private void updateWriteStat(String path, long bytes) {
+        String namespace = getTopNamespace(path);
+        if (namespace == null) {
+            return;
+        }
+        ServerMetrics.WRITE_PER_NAMESPACE.add(namespace, path.length() + bytes);
+    }
 }

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

@@ -76,6 +76,9 @@ public enum ServerMetrics {
 
     UNRECOVERABLE_ERROR_COUNT(new SimpleCounter("unrecoverable_error_count")),
 
+    WRITE_PER_NAMESPACE(new AvgMinMaxCounterSet("write_per_namespace")),
+    READ_PER_NAMESPACE(new AvgMinMaxCounterSet("read_per_namespace")),
+
     BYTES_RECEIVED_COUNT(new SimpleCounter("bytes_received_count")),
 
     /**

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

@@ -45,6 +45,8 @@ import org.apache.jute.BinaryOutputArchive;
 import org.apache.jute.Record;
 import org.apache.zookeeper.common.PathTrie;
 import java.lang.reflect.*;
+import java.util.HashMap;
+import java.util.Map;
 import java.util.concurrent.Semaphore;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -358,4 +360,62 @@ public class DataTreeTest extends ZKTestCase {
         //add these three init nodes:/zookeeper,/zookeeper/quota,/zookeeper/config,so the number is 8.
         Assert.assertEquals( 8, dt.getAllChildrenNumber("/"));
     }
+
+    @Test
+    public void testDataTreeMetrics() throws Exception {
+        ServerMetrics.resetAll();
+
+
+        long readBytes1 = 0;
+        long readBytes2 = 0;
+        long writeBytes1 = 0;
+        long writeBytes2 = 0;
+
+        final String TOP1 = "top1";
+        final String TOP2 = "ttop2";
+        final String TOP1PATH = "/" + TOP1;
+        final String TOP2PATH = "/" + TOP2;
+        final String CHILD1 = "child1";
+        final String CHILD2 = "springishere";
+        final String CHILD1PATH = TOP1PATH + "/" + CHILD1;
+        final String CHILD2PATH = TOP1PATH + "/" + CHILD2;
+
+        final int TOP2_LEN = 50;
+        final int CHILD1_LEN = 100;
+        final int CHILD2_LEN = 250;
+
+        DataTree dt = new DataTree();
+        dt.createNode(TOP1PATH, null, null, -1, 1, 1, 1);
+        writeBytes1 += TOP1PATH.length();
+        dt.createNode(TOP2PATH, new byte[TOP2_LEN], null, -1, 1, 1, 1);
+        writeBytes2 += TOP2PATH.length() + TOP2_LEN;
+        dt.createNode(CHILD1PATH, null, null, -1, 1, 1, 1);
+        writeBytes1 += CHILD1PATH.length();
+        dt.setData(CHILD1PATH, new byte[CHILD1_LEN], 1, -1, 1);
+        writeBytes1 += CHILD1PATH.length() + CHILD1_LEN;
+        dt.createNode(CHILD2PATH, new byte[CHILD2_LEN], null, -1, 1, 1, 1);
+        writeBytes1 += CHILD2PATH.length() + CHILD2_LEN;
+        dt.getData(TOP1PATH, new Stat(), null);
+        readBytes1 += TOP1PATH.length() + DataTree.STAT_OVERHEAD_BYTES;
+        dt.getData(TOP2PATH, new Stat(), null);
+        readBytes2 += TOP2PATH.length() + TOP2_LEN + DataTree.STAT_OVERHEAD_BYTES;
+        dt.statNode(CHILD2PATH, null);
+        readBytes1 += CHILD2PATH.length() + DataTree.STAT_OVERHEAD_BYTES;
+        dt.getChildren(TOP1PATH, new Stat(), null);
+        readBytes1 += TOP1PATH.length() + CHILD1.length() + CHILD2.length() + DataTree.STAT_OVERHEAD_BYTES;
+        dt.deleteNode(TOP1PATH, 1);
+        writeBytes1 += TOP1PATH.length();
+        
+        Map<String, Object> values = ServerMetrics.getAllValues();
+
+        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"));
+        Assert.assertEquals(1L, values.get("cnt_" + TOP2 + "_write_per_namespace"));
+
+        Assert.assertEquals(readBytes1, values.get("sum_" + TOP1+ "_read_per_namespace"));
+        Assert.assertEquals(3L, values.get("cnt_" + TOP1 + "_read_per_namespace"));
+        Assert.assertEquals(readBytes2, values.get("sum_" + TOP2+ "_read_per_namespace"));
+        Assert.assertEquals(1L, values.get("cnt_" + TOP2 + "_read_per_namespace"));
+    }
 }