ZOOKEEPER-3321: Add metrics for Leader

Author: Jie Huang <jiehuang@fb.com>

Reviewers: andor@apache.org, fangmin@apache.org, eolivelli@apache.org

Closes #858 from jhuan31/ZOOKEEPER-3321

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

@@ -210,6 +210,11 @@ public final class ServerMetrics {
         SYNC_PROCESS_TIME = metricsContext.getSummary("sync_process_time", DetailLevel.BASIC);
 
         BATCH_SIZE = metricsContext.getSummary("sync_processor_batch_size", DetailLevel.BASIC);
+
+        QUORUM_ACK_LATENCY = metricsContext.getSummary("quorum_ack_latency", DetailLevel.ADVANCED);
+        ACK_LATENCY = metricsContext.getSummarySet("ack_latency", DetailLevel.ADVANCED);
+        PROPOSAL_COUNT = metricsContext.getCounter("proposal_count");
+        QUIT_LEADING_DUE_TO_DISLOYAL_VOTER = metricsContext.getCounter("quit_leading_due_to_disloyal_voter");
     }
 
     /**
@@ -300,6 +305,11 @@ public final class ServerMetrics {
 
     public final Summary BATCH_SIZE;
 
+    public final Summary QUORUM_ACK_LATENCY;
+    public final SummarySet ACK_LATENCY;
+    public final Counter PROPOSAL_COUNT;
+    public final Counter QUIT_LEADING_DUE_TO_DISLOYAL_VOTER;
+
     /**
      * Fired watcher stats.
      */

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

@@ -101,6 +101,22 @@ public class Leader implements LearnerMaster {
 
     private final LearnerSnapshotThrottler learnerSnapshotThrottler;
 
+    // log ack latency if zxid is a multiple of ackLoggingFrequency. If <=0, disable logging.
+    protected static final String ACK_LOGGING_FREQUENCY = "zookeeper.leader.ackLoggingFrequency";
+    private static int ackLoggingFrequency;
+    static {
+        ackLoggingFrequency = Integer.getInteger(ACK_LOGGING_FREQUENCY, 1000);
+        LOG.info(ACK_LOGGING_FREQUENCY + " = " + ackLoggingFrequency);
+    }
+
+    public static void setAckLoggingFrequency(int frequency) {
+        ackLoggingFrequency = frequency;
+    }
+
+    public static int getAckLoggingFrequency() {
+        return ackLoggingFrequency;
+    }
+
     final LeaderZooKeeperServer zk;
 
     final QuorumPeer self;
@@ -865,6 +881,7 @@ public class Leader implements LearnerMaster {
             informAndActivate(p, designatedLeader);
             //turnOffFollowers();
         } else {
+            p.request.logLatency(ServerMetrics.getMetrics().QUORUM_ACK_LATENCY);
             commit(zxid);
             inform(p);
         }
@@ -931,6 +948,10 @@ public class Leader implements LearnerMaster {
             return;
         }
 
+        if (ackLoggingFrequency > 0 && (zxid % ackLoggingFrequency == 0)) {
+            p.request.logLatency(ServerMetrics.getMetrics().ACK_LATENCY, Long.toString(sid));
+        }
+
         p.addAck(sid);
         /*if (LOG.isDebugEnabled()) {
             LOG.debug("Count for zxid: 0x{} is {}",
@@ -1171,6 +1192,7 @@ public class Leader implements LearnerMaster {
             outstandingProposals.put(lastProposed, p);
             sendPacket(pp);
         }
+        ServerMetrics.getMetrics().PROPOSAL_COUNT.add(1);
         return p;
     }
 
@@ -1296,6 +1318,7 @@ public class Leader implements LearnerMaster {
             quitWaitForEpoch = true;
             connectingFollowers.notifyAll();
         }
+        ServerMetrics.getMetrics().QUIT_LEADING_DUE_TO_DISLOYAL_VOTER.add(1);
         LOG.info("Quit leading due to voter changed mind.");
     }
 

zookeeper-server/src/test/java/org/apache/zookeeper/server/quorum/LeaderMetricsTest.java

@@ -0,0 +1,99 @@
+/**
+ * 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.quorum;
+
+import org.apache.zookeeper.*;
+import org.apache.zookeeper.metrics.MetricsUtils;
+import org.apache.zookeeper.server.ServerMetrics;
+import org.apache.zookeeper.test.ClientBase;
+import org.apache.zookeeper.test.QuorumUtil;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.lang.reflect.Field;
+import java.lang.reflect.Modifier;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+
+import static org.hamcrest.number.OrderingComparison.greaterThan;
+import static org.hamcrest.number.OrderingComparison.greaterThanOrEqualTo;
+
+public class LeaderMetricsTest extends ZKTestCase {
+    CountDownLatch createdLatch;
+    int oldLoggingFeq;
+
+    private class MyWatcher implements Watcher {
+        @Override
+        public void process(WatchedEvent e){
+            createdLatch.countDown();
+        }
+    }
+
+    @Before
+    public void setup() {
+        oldLoggingFeq = Leader.getAckLoggingFrequency();
+    }
+
+    @After
+    public void teardown() {
+        Leader.setAckLoggingFrequency(oldLoggingFeq);
+    }
+    
+    @Test
+    public void testLeaderMetrics() throws Exception {
+        // set the logging frequency to one so we log the ack latency for every ack
+        Leader.setAckLoggingFrequency(1);
+
+        ServerMetrics.getMetrics().resetAll();
+
+        QuorumUtil util = new QuorumUtil(1); //creating a quorum of 3 servers
+        util.startAll();
+
+        ZooKeeper zk = ClientBase.createZKClient(util.getConnString());
+        createdLatch = new CountDownLatch(1);
+        zk.exists("/test", new MyWatcher());
+        zk.create("/test", new byte[2], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
+        createdLatch.await();
+
+        Map<String, Object> values = MetricsUtils.currentServerMetrics();
+
+        Assert.assertEquals(2L, values.get("proposal_count"));
+        // Quorum ack latency is per txn
+        Assert.assertEquals(2L, values.get("cnt_quorum_ack_latency"));
+        Assert.assertThat((long)values.get("min_quorum_ack_latency"), greaterThan(0L));
+
+        int numberOfAckServers = 0;
+        // ack latency is per server
+        for (int sid = 1; sid<=3; sid++) {
+            String metricName = "min_" + sid + "_ack_latency";
+            if (values.get(metricName) != null) {
+                numberOfAckServers++;
+                Assert.assertThat((long) values.get("min_" + sid + "_ack_latency"), greaterThanOrEqualTo(0L));
+            }
+        }
+
+        // at least two servers should have send ACKs
+        Assert.assertThat(numberOfAckServers, greaterThanOrEqualTo(2));
+
+        zk.close();
+        util.shutdownAll();
+    }
+}