ZOOKEEPER-2479: Add 'electionTimeTaken' value in LeaderMXBean and Fol…

…lowerMXBean

This PR is against branch-3.4, please review. Thanks!

Author: Rakesh Radhakrishnan <rakeshr@apache.org>

Reviewers: fpj <fpj@apache.org>

Closes #130 from rakeshadr/ZK-2479-br-3-4

src/java/main/org/apache/zookeeper/server/quorum/Follower.java

@@ -59,8 +59,9 @@ public String toString() {
      */
     void followLeader() throws InterruptedException {
         self.end_fle = System.currentTimeMillis();
-        LOG.info("FOLLOWING - LEADER ELECTION TOOK - " +
-              (self.end_fle - self.start_fle));
+        long electionTimeTaken = self.end_fle - self.start_fle;
+        self.setElectionTimeTaken(electionTimeTaken);
+        LOG.info("FOLLOWING - LEADER ELECTION TOOK - {}", electionTimeTaken);
         self.start_fle = 0;
         self.end_fle = 0;
         fzk.registerJMX(new FollowerBean(this, zk), self.jmxLocalPeerBean);

src/java/main/org/apache/zookeeper/server/quorum/FollowerBean.java

@@ -22,7 +22,7 @@
 import org.apache.zookeeper.server.ZooKeeperServerBean;
 
 /**
- * Follower MBean inteface implementation
+ * Follower MBean interface implementation.
  */
 public class FollowerBean extends ZooKeeperServerBean implements FollowerMXBean {
     private final Follower follower;
@@ -47,4 +47,9 @@ public String getLastQueuedZxid() {
     public int getPendingRevalidationCount() {
         return follower.getPendingRevalidationsCount();
     }
+
+    @Override
+    public long getElectionTimeTaken() {
+        return follower.self.getElectionTimeTaken();
+    }
 }

src/java/main/org/apache/zookeeper/server/quorum/FollowerMXBean.java

@@ -38,4 +38,9 @@ public interface FollowerMXBean extends ZooKeeperServerMXBean {
      * @return count of pending revalidations
      */
     public int getPendingRevalidationCount();
+
+    /**
+     * @return time taken for leader election in milliseconds.
+     */
+    public long getElectionTimeTaken();
 }

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

@@ -366,8 +366,9 @@ public void halt() {
      */
     void lead() throws IOException, InterruptedException {
         self.end_fle = System.currentTimeMillis();
-        LOG.info("LEADING - LEADER ELECTION TOOK - " +
-              (self.end_fle - self.start_fle));
+        long electionTimeTaken = self.end_fle - self.start_fle;
+        self.setElectionTimeTaken(electionTimeTaken);
+        LOG.info("LEADING - LEADER ELECTION TOOK - {}", electionTimeTaken);
         self.start_fle = 0;
         self.end_fle = 0;
 

src/java/main/org/apache/zookeeper/server/quorum/LeaderBean.java

@@ -50,4 +50,8 @@ public String followerInfo() {
         return sb.toString();
     }
 
+    @Override
+    public long getElectionTimeTaken() {
+        return leader.self.getElectionTimeTaken();
+    }
 }

src/java/main/org/apache/zookeeper/server/quorum/LeaderMXBean.java

@@ -33,4 +33,9 @@ public interface LeaderMXBean extends ZooKeeperServerMXBean {
      * @return information on current followers
      */
     public String followerInfo();
+
+    /**
+     * @return time taken for leader election in milliseconds.
+     */
+    public long getElectionTimeTaken();
 }

src/java/main/org/apache/zookeeper/server/quorum/QuorumPeer.java

@@ -395,6 +395,12 @@ synchronized void setBCVote(Vote v) {
      */
     protected int quorumCnxnThreadsSize = QUORUM_CNXN_THREADS_SIZE_DEFAULT_VALUE;
 
+    /**
+     * Keeps time taken for leader election in milliseconds. Sets the value to
+     * this variable only after the completion of leader election.
+     */
+    private long electionTimeTaken = -1;
+
     /**
      * @deprecated As of release 3.4.0, this class has been deprecated, since
      * it is used with one of the udp-based versions of leader election, which
@@ -1459,4 +1465,21 @@ public QuorumCnxManager createCnxnManager() {
                                     this.quorumCnxnThreadsSize,
                                     this.isQuorumSaslAuthEnabled());
     }
+
+    /**
+     * Sets the time taken for leader election in milliseconds.
+     *
+     * @param electionTimeTaken
+     *            time taken for leader election
+     */
+    void setElectionTimeTaken(long electionTimeTaken) {
+        this.electionTimeTaken = electionTimeTaken;
+    }
+
+    /**
+     * @return the time taken for leader election in milliseconds.
+     */
+    long getElectionTimeTaken() {
+        return electionTimeTaken;
+    }
 }

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

@@ -20,18 +20,22 @@
 import java.io.ByteArrayInputStream;
 import java.io.File;
 import java.io.IOException;
-import java.net.InetSocketAddress;
+import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.LinkedHashSet;
+import java.util.List;
 import java.util.Properties;
 import java.util.Set;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.apache.zookeeper.PortAssignment;
 import org.apache.zookeeper.TestableZooKeeper;
+import org.apache.zookeeper.jmx.CommonNames;
 import org.apache.zookeeper.server.quorum.QuorumPeer;
+import org.apache.zookeeper.server.quorum.QuorumPeer.LearnerType;
 import org.apache.zookeeper.server.quorum.QuorumPeer.QuorumServer;
+import org.apache.zookeeper.server.quorum.QuorumPeer.ServerState;
 import org.apache.zookeeper.server.quorum.flexible.QuorumHierarchical;
 import org.junit.Assert;
 import org.junit.Test;
@@ -139,23 +143,28 @@ void startServers(boolean withObservers) throws Exception {
                         : QuorumPeer.LearnerType.PARTICIPANT));
 
         LOG.info("creating QuorumPeer 1 port " + port1);
+        List <QuorumPeer> qps = new ArrayList<>();
         QuorumHierarchical hq1 = new QuorumHierarchical(qp); 
         s1 = new QuorumPeer(peers, s1dir, s1dir, port1, 3, 1, tickTime, initLimit, syncLimit, hq1);
+        qps.add(s1);
         Assert.assertEquals(port1, s1.getClientPort());
 
         LOG.info("creating QuorumPeer 2 port " + port2);
         QuorumHierarchical hq2 = new QuorumHierarchical(qp); 
         s2 = new QuorumPeer(peers, s2dir, s2dir, port2, 3, 2, tickTime, initLimit, syncLimit, hq2);
+        qps.add(s2);
         Assert.assertEquals(port2, s2.getClientPort());
 
         LOG.info("creating QuorumPeer 3 port " + port3);
         QuorumHierarchical hq3 = new QuorumHierarchical(qp); 
         s3 = new QuorumPeer(peers, s3dir, s3dir, port3, 3, 3, tickTime, initLimit, syncLimit, hq3);
+        qps.add(s3);
         Assert.assertEquals(port3, s3.getClientPort());
 
         LOG.info("creating QuorumPeer 4 port " + port4);
         QuorumHierarchical hq4 = new QuorumHierarchical(qp); 
         s4 = new QuorumPeer(peers, s4dir, s4dir, port4, 3, 4, tickTime, initLimit, syncLimit, hq4);
+        qps.add(s4);
         if (withObservers) {
             s4.setLearnerType(QuorumPeer.LearnerType.OBSERVER);
         }
@@ -164,6 +173,7 @@ void startServers(boolean withObservers) throws Exception {
         LOG.info("creating QuorumPeer 5 port " + port5);
         QuorumHierarchical hq5 = new QuorumHierarchical(qp); 
         s5 = new QuorumPeer(peers, s5dir, s5dir, port5, 3, 5, tickTime, initLimit, syncLimit, hq5);
+        qps.add(s5);
         if (withObservers) {
             s5.setLearnerType(QuorumPeer.LearnerType.OBSERVER);
         }
@@ -219,6 +229,7 @@ void startServers(boolean withObservers) throws Exception {
             ensureNames.add("name0=ReplicatedServer_id" + i);
         }
         JMXEnv.ensureAll(ensureNames.toArray(new String[ensureNames.size()]));
+        verifyElectionTimeTakenJMXAttribute(qps);
     }
 
     @Override
@@ -264,6 +275,33 @@ protected TestableZooKeeper createClient(String hp)
         return createClient(watcher, hp);
     }
 
+    private void verifyElectionTimeTakenJMXAttribute(List<QuorumPeer> peers)
+            throws Exception {
+        LOG.info("Verify QuorumPeer#electionTimeTaken jmx bean attribute");
+
+        for (int i = 1; i <= peers.size(); i++) {
+            QuorumPeer qp = peers.get(i - 1);
+            if (qp.getLearnerType() == LearnerType.OBSERVER) {
+                continue; // Observer don't have electionTimeTaken attribute.
+            }
+            Long electionTimeTaken = -1L;
+            String bean = "";
+            if (qp.getPeerState() == ServerState.FOLLOWING) {
+                bean = String.format(
+                        "%s:name0=ReplicatedServer_id%d,name1=replica.%d,name2=Follower",
+                        CommonNames.DOMAIN, i, i);
+            } else if (qp.getPeerState() == ServerState.LEADING) {
+                bean = String.format(
+                        "%s:name0=ReplicatedServer_id%d,name1=replica.%d,name2=Leader",
+                        CommonNames.DOMAIN, i, i);
+            }
+            electionTimeTaken = (Long) JMXEnv.ensureBeanAttribute(bean,
+                    "ElectionTimeTaken");
+            Assert.assertTrue("Wrong electionTimeTaken value!",
+                    electionTimeTaken >= 0);
+        }
+    }
+
     @Test
     public void testHierarchicalQuorum() throws Throwable {
         cht.runHammer(5, 10);

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

@@ -37,6 +37,7 @@
 
 import org.apache.zookeeper.jmx.CommonNames;
 import org.apache.zookeeper.jmx.MBeanRegistry;
+import org.junit.Assert;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -274,4 +275,51 @@ private static boolean compare(String bean, String name) {
         }
         return false;
     }
+
+    /**
+     * Ensure that the specified bean name and its attribute is registered. Note
+     * that these are components of the name. It waits in a loop up to 60
+     * seconds before failing if there is a mismatch. This will return the beans
+     * which are not matched.
+     *
+     * @param expectedName
+     *            - expected bean
+     * @param expectedAttribute
+     *            - expected attribute
+     * @return the value of the attribute
+     *
+     * @throws Exception
+     */
+    public static Object ensureBeanAttribute(String expectedName,
+            String expectedAttribute) throws Exception {
+        String value = "";
+        LOG.info("ensure bean:{}, attribute:{}", new Object[] { expectedName,
+                expectedAttribute });
+
+        Set<ObjectName> beans;
+        int nTry = 0;
+        do {
+            if (nTry++ > 0) {
+                Thread.sleep(500);
+            }
+            try {
+                beans = conn().queryNames(
+                        new ObjectName(CommonNames.DOMAIN + ":*"), null);
+            } catch (MalformedObjectNameException e) {
+                throw new RuntimeException(e);
+            }
+            LOG.info("expect:" + expectedName);
+            for (ObjectName bean : beans) {
+                // check the existence of name in bean
+                if (bean.toString().equals(expectedName)) {
+                    LOG.info("found:{} {}", new Object[] { expectedName, bean });
+                    return conn().getAttribute(bean, expectedAttribute);
+                }
+            }
+        } while (nTry < 120);
+        Assert.fail("Failed to find bean:" + expectedName + ", attribute:"
+                + expectedAttribute);
+        return value;
+    }
+
 }