ZOOKEEPER-1659. Add JMX support for dynamic reconfiguration (Rakesh R via michim)

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

CHANGES.txt

@@ -921,6 +921,9 @@ IMPROVEMENTS:
 
   ZOOKEEPER-1930. A typo in zookeeper recipes.html (Chengwei Yang via michim)
 
+  ZOOKEEPER-1659. Add JMX support for dynamic reconfiguration (Rakesh R via
+  michim)
+
 headers
 
 Release 3.4.0 - 

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

@@ -18,6 +18,7 @@
 
 package org.apache.zookeeper.server.quorum;
 
+import org.apache.zookeeper.common.HostNameUtils;
 
 /**
  * Implementation of the local peer MBean interface.
@@ -76,4 +77,32 @@ public class LocalPeerBean extends ServerBean implements LocalPeerMXBean {
     public int getElectionType() {
         return peer.getElectionType();
     }
+
+    public String getElectionAddress() {
+        return HostNameUtils.getHostString(peer.getElectionAddress()) + ":"
+                + peer.getElectionAddress().getPort();
+    }
+
+    public String getClientAddress() {
+        return HostNameUtils.getHostString(peer.getClientAddress()) + ":"
+                + peer.getClientAddress().getPort();
+    }
+
+    public String getLearnerType(){
+        return peer.getLearnerType().toString();
+    }
+
+    public long getConfigVersion(){
+        return peer.getQuorumVerifier().getVersion();
+    }
+
+    @Override
+    public String getQuorumSystemInfo() {
+        return peer.getQuorumVerifier().toString();
+    }
+
+    @Override
+    public boolean isPartOfEnsemble() {
+        return peer.getView().containsKey(peer.getId());
+    }
 }

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

@@ -74,4 +74,34 @@ public interface LocalPeerMXBean extends ServerMXBean {
      * @return the election type
      */
     public int getElectionType();
+
+    /**
+     * @return the election address
+     */
+    public String getElectionAddress();
+
+    /**
+     * @return the client address
+     */
+    public String getClientAddress();
+
+    /**
+     * @return the learner type
+     */
+    public String getLearnerType();
+
+    /**
+     * @return the config version
+     */
+    public long getConfigVersion();
+
+    /**
+     * @return the quorum system information
+     */
+    public String getQuorumSystemInfo();
+
+    /**
+     * @return true if quorum peer is part of the ensemble, false otherwise
+     */
+    public boolean isPartOfEnsemble();
 }

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

@@ -95,7 +95,7 @@ public class QuorumPeer extends ZooKeeperThread implements QuorumStats.Provider
 
     private QuorumBean jmxQuorumBean;
     LocalPeerBean jmxLocalPeerBean;
-    private Set<RemotePeerBean> jmxRemotePeerBean;
+    private Map<Long, RemotePeerBean> jmxRemotePeerBean;
     LeaderElectionBean jmxLeaderElectionBean;
     private QuorumCnxManager qcm;
 
@@ -585,7 +585,7 @@ public class QuorumPeer extends ZooKeeperThread implements QuorumStats.Provider
     public QuorumPeer() {
         super("QuorumPeer");
         quorumStats = new QuorumStats(this);
-        jmxRemotePeerBean = new HashSet<RemotePeerBean>();
+        jmxRemotePeerBean = new HashMap<Long, RemotePeerBean>();
     }
 
 
@@ -872,10 +872,10 @@ public class QuorumPeer extends ZooKeeperThread implements QuorumStats.Provider
                         jmxLocalPeerBean = null;
                     }
                 } else {
-                    p = new RemotePeerBean(s);
-                    jmxRemotePeerBean.add((RemotePeerBean) p);
+                    RemotePeerBean rBean = new RemotePeerBean(s);
                     try {
-                        MBeanRegistry.getInstance().register(p, jmxQuorumBean);
+                        MBeanRegistry.getInstance().register(rBean, jmxQuorumBean);
+                        jmxRemotePeerBean.put(s.id, rBean);
                     } catch (Exception e) {
                         LOG.warn("Failed to register with JMX", e);
                     }
@@ -1005,7 +1005,7 @@ public class QuorumPeer extends ZooKeeperThread implements QuorumStats.Provider
             instance.unregister(jmxQuorumBean);
             instance.unregister(jmxLocalPeerBean);
 
-            for (RemotePeerBean remotePeerBean : jmxRemotePeerBean) {
+            for (RemotePeerBean remotePeerBean : jmxRemotePeerBean.values()) {
                 instance.unregister(remotePeerBean);
             }
 
@@ -1557,9 +1557,11 @@ public class QuorumPeer extends ZooKeeperThread implements QuorumStats.Provider
        initConfigInZKDatabase();
 
        if (prevQV.getVersion() < qv.getVersion() && !prevQV.equals(qv)) {
+           Map<Long, QuorumServer> newMembers = qv.getAllMembers();
+           updateRemotePeerMXBeans(newMembers);
            if (restartLE) restartLeaderElection(prevQV, qv);
 
-           QuorumServer myNewQS = qv.getAllMembers().get(getId());
+           QuorumServer myNewQS = newMembers.get(getId());
            if (myNewQS != null && myNewQS.clientAddr != null
                    && !myNewQS.clientAddr.equals(oldClientAddr)) {
                cnxnFactory.reconfigure(myNewQS.clientAddr);
@@ -1588,7 +1590,41 @@ public class QuorumPeer extends ZooKeeperThread implements QuorumStats.Provider
        return false;
 
    }
-    
+
+    private void updateRemotePeerMXBeans(Map<Long, QuorumServer> newMembers) {
+        Set<Long> existingMembers = new HashSet<Long>(newMembers.keySet());
+        existingMembers.retainAll(jmxRemotePeerBean.keySet());
+        for (Long id : existingMembers) {
+            RemotePeerBean rBean = jmxRemotePeerBean.get(id);
+            rBean.setQuorumServer(newMembers.get(id));
+        }
+
+        Set<Long> joiningMembers = new HashSet<Long>(newMembers.keySet());
+        joiningMembers.removeAll(jmxRemotePeerBean.keySet());
+        joiningMembers.remove(getId()); // remove self as it is local bean
+        for (Long id : joiningMembers) {
+            QuorumServer qs = newMembers.get(id);
+            RemotePeerBean rBean = new RemotePeerBean(qs);
+            try {
+                MBeanRegistry.getInstance().register(rBean, jmxQuorumBean);
+                jmxRemotePeerBean.put(qs.id, rBean);
+            } catch (Exception e) {
+                LOG.warn("Failed to register with JMX", e);
+            }
+        }
+
+        Set<Long> leavingMembers = new HashSet<Long>(jmxRemotePeerBean.keySet());
+        leavingMembers.removeAll(newMembers.keySet());
+        for (Long id : leavingMembers) {
+            RemotePeerBean rBean = jmxRemotePeerBean.remove(id);
+            try {
+                MBeanRegistry.getInstance().unregister(rBean);
+            } catch (Exception e) {
+                LOG.warn("Failed to unregister with JMX", e);
+            }
+        }
+    }
+
    private boolean updateLearnerType(QuorumVerifier newQV) {        
        //check if I'm an observer in new config
        if (newQV.getObservingMembers().containsKey(getId())) {

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

@@ -18,8 +18,8 @@
 
 package org.apache.zookeeper.server.quorum;
 
+import org.apache.zookeeper.common.HostNameUtils;
 import org.apache.zookeeper.jmx.ZKMBeanInfo;
-import org.apache.zookeeper.server.quorum.QuorumPeer;
 
 /**
  * A remote peer bean only provides limited information about the remote peer,
@@ -31,6 +31,11 @@ public class RemotePeerBean implements RemotePeerMXBean,ZKMBeanInfo {
     public RemotePeerBean(QuorumPeer.QuorumServer peer){
         this.peer=peer;
     }
+
+    public void setQuorumServer(QuorumPeer.QuorumServer peer) {
+        this.peer = peer;
+    }
+
     public String getName() {
         return "replica."+peer.id;
     }
@@ -42,4 +47,17 @@ public class RemotePeerBean implements RemotePeerMXBean,ZKMBeanInfo {
         return peer.addr.getHostName()+":"+peer.addr.getPort();
     }
 
+    public String getElectionAddress() {
+        return HostNameUtils.getHostString(peer.electionAddr) + ":"
+                + peer.electionAddr.getPort();
+    }
+
+    public String getClientAddress() {
+        return HostNameUtils.getHostString(peer.clientAddr) + ":"
+                + peer.clientAddr.getPort();
+    }
+
+    public String getLearnerType() {
+        return peer.type.toString();
+    }
 }

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

@@ -30,4 +30,19 @@ public interface RemotePeerMXBean {
      * @return IP address of the quorum peer 
      */
     public String getQuorumAddress();
+
+    /**
+     * @return the election address
+     */
+    public String getElectionAddress();
+
+    /**
+     * @return the client address
+     */
+    public String getClientAddress();
+
+    /**
+     * @return the learner type
+     */
+    public String getLearnerType();
 }

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

@@ -30,6 +30,7 @@ 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.QuorumServer;
 import org.apache.zookeeper.server.quorum.flexible.QuorumHierarchical;
@@ -256,6 +257,16 @@ public class HierarchicalQuorumTest extends ClientBase {
             ensureNames.add("name0=ReplicatedServer_id" + i);
         }
         JMXEnv.ensureAll(ensureNames.toArray(new String[ensureNames.size()]));
+
+        for (int i = 1; i <= 5; i++) {
+            String bean = CommonNames.DOMAIN + ":name0=ReplicatedServer_id" + i
+                    + ",name1=replica." + i;
+            JMXEnv.ensureBeanAttribute(bean, "ConfigVersion");
+            JMXEnv.ensureBeanAttribute(bean, "LearnerType");
+            JMXEnv.ensureBeanAttribute(bean, "ClientAddress");
+            JMXEnv.ensureBeanAttribute(bean, "ElectionAddress");
+            JMXEnv.ensureBeanAttribute(bean, "QuorumSystemInfo");
+        }
     }
 
     @Override

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

@@ -255,6 +255,52 @@ public class JMXEnv {
         return beans;
     }
 
+    /**
+     * 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);
+        TestCase.fail("Failed to find bean:" + expectedName + ", attribute:"
+                + expectedAttribute);
+        return value;
+    }
+
     /**
      * Comparing that the given name exists in the bean. For component beans,
      * the component name will be present at the end of the bean name

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

@@ -32,7 +32,10 @@ import org.apache.zookeeper.ZKTestCase;
 import org.apache.zookeeper.ZooDefs;
 import org.apache.zookeeper.ZooKeeper;
 import org.apache.zookeeper.AsyncCallback.DataCallback;
+import org.apache.zookeeper.common.HostNameUtils;
 import org.apache.zookeeper.data.Stat;
+import org.apache.zookeeper.jmx.CommonNames;
+import org.apache.zookeeper.server.quorum.QuorumPeer;
 import org.apache.zookeeper.server.quorum.QuorumStats;
 import org.apache.zookeeper.server.quorum.QuorumPeer.QuorumServer;
 import org.apache.zookeeper.server.quorum.QuorumPeer.ServerState;
@@ -778,4 +781,193 @@ public class ReconfigTest extends ZKTestCase implements DataCallback{
             Assert.assertTrue(version == 0x100000000L);
         }
     }
+
+    /**
+     * Tests verifies the jmx attributes of local and remote peer bean - remove
+     * one quorum peer and again adding it back
+     */
+    @Test
+    public void testJMXBeanAfterRemoveAddOne() throws Exception {
+        qu = new QuorumUtil(1); // create 3 servers
+        qu.disableJMXTest = true;
+        qu.startAll();
+        ZooKeeper[] zkArr = createHandles(qu);
+
+        List<String> leavingServers = new ArrayList<String>();
+        List<String> joiningServers = new ArrayList<String>();
+
+        // assert remotePeerBean.1 of ReplicatedServer_2
+        int leavingIndex = 1;
+        int replica2 = 2;
+        QuorumPeer peer2 = qu.getPeer(replica2).peer;
+        QuorumServer leavingQS2 = peer2.getView().get(new Long(leavingIndex));
+        String remotePeerBean2 = CommonNames.DOMAIN
+                + ":name0=ReplicatedServer_id" + replica2 + ",name1=replica."
+                + leavingIndex;
+        assertRemotePeerMXBeanAttributes(leavingQS2, remotePeerBean2);
+
+        // assert remotePeerBean.1 of ReplicatedServer_3
+        int replica3 = 3;
+        QuorumPeer peer3 = qu.getPeer(replica3).peer;
+        QuorumServer leavingQS3 = peer3.getView().get(new Long(leavingIndex));
+        String remotePeerBean3 = CommonNames.DOMAIN
+                + ":name0=ReplicatedServer_id" + replica3 + ",name1=replica."
+                + leavingIndex;
+        assertRemotePeerMXBeanAttributes(leavingQS3, remotePeerBean3);
+
+        ZooKeeper zk = zkArr[leavingIndex];
+
+        leavingServers.add(Integer.toString(leavingIndex));
+
+        // remember this server so we can add it back later
+        joiningServers.add("server." + leavingIndex + "=127.0.0.1:"
+                + qu.getPeer(leavingIndex).peer.getQuorumAddress().getPort()
+                + ":"
+                + qu.getPeer(leavingIndex).peer.getElectionAddress().getPort()
+                + ":participant;127.0.0.1:"
+                + qu.getPeer(leavingIndex).peer.getClientPort());
+
+        // Remove ReplicatedServer_1 from the ensemble
+        reconfig(zk, null, leavingServers, null, -1);
+
+        // localPeerBean.1 of ReplicatedServer_1
+        QuorumPeer removedPeer = qu.getPeer(leavingIndex).peer;
+        String localPeerBean = CommonNames.DOMAIN
+                + ":name0=ReplicatedServer_id" + leavingIndex
+                + ",name1=replica." + leavingIndex;
+        assertLocalPeerMXBeanAttributes(removedPeer, localPeerBean, false);
+
+        // remotePeerBean.1 shouldn't exists in ReplicatedServer_2
+        JMXEnv.ensureNone(remotePeerBean2);
+        // remotePeerBean.1 shouldn't exists in ReplicatedServer_3
+        JMXEnv.ensureNone(remotePeerBean3);
+
+        // Add ReplicatedServer_1 back to the ensemble
+        reconfig(zk, joiningServers, null, null, -1);
+
+        // localPeerBean.1 of ReplicatedServer_1
+        assertLocalPeerMXBeanAttributes(removedPeer, localPeerBean, true);
+
+        // assert remotePeerBean.1 of ReplicatedServer_2
+        leavingQS2 = peer2.getView().get(new Long(leavingIndex));
+        assertRemotePeerMXBeanAttributes(leavingQS2, remotePeerBean2);
+
+        // assert remotePeerBean.1 of ReplicatedServer_3
+        leavingQS3 = peer3.getView().get(new Long(leavingIndex));
+        assertRemotePeerMXBeanAttributes(leavingQS3, remotePeerBean3);
+
+        closeAllHandles(zkArr);
+    }
+
+    /**
+     * Tests verifies the jmx attributes of local and remote peer bean - change
+     * participant to observer role
+     */
+    @Test
+    public void testJMXBeanAfterRoleChange() throws Exception {
+        qu = new QuorumUtil(1); // create 3 servers
+        qu.disableJMXTest = true;
+        qu.startAll();
+        ZooKeeper[] zkArr = createHandles(qu);
+
+        // changing a server's role / port is done by "adding" it with the same
+        // id but different role / port
+        List<String> joiningServers = new ArrayList<String>();
+
+        // assert remotePeerBean.1 of ReplicatedServer_2
+        int changingIndex = 1;
+        int replica2 = 2;
+        QuorumPeer peer2 = qu.getPeer(replica2).peer;
+        QuorumServer changingQS2 = peer2.getView().get(new Long(changingIndex));
+        String remotePeerBean2 = CommonNames.DOMAIN
+                + ":name0=ReplicatedServer_id" + replica2 + ",name1=replica."
+                + changingIndex;
+        assertRemotePeerMXBeanAttributes(changingQS2, remotePeerBean2);
+
+        // assert remotePeerBean.1 of ReplicatedServer_3
+        int replica3 = 3;
+        QuorumPeer peer3 = qu.getPeer(replica3).peer;
+        QuorumServer changingQS3 = peer3.getView().get(new Long(changingIndex));
+        String remotePeerBean3 = CommonNames.DOMAIN
+                + ":name0=ReplicatedServer_id" + replica3 + ",name1=replica."
+                + changingIndex;
+        assertRemotePeerMXBeanAttributes(changingQS3, remotePeerBean3);
+
+        String newRole = "observer";
+
+        ZooKeeper zk = zkArr[changingIndex];
+
+        // exactly as it is now, except for role change
+        joiningServers.add("server." + changingIndex + "=127.0.0.1:"
+                + qu.getPeer(changingIndex).peer.getQuorumAddress().getPort()
+                + ":"
+                + qu.getPeer(changingIndex).peer.getElectionAddress().getPort()
+                + ":" + newRole + ";127.0.0.1:"
+                + qu.getPeer(changingIndex).peer.getClientPort());
+
+        reconfig(zk, joiningServers, null, null, -1);
+        testNormalOperation(zkArr[changingIndex], zk);
+
+        Assert.assertTrue(qu.getPeer(changingIndex).peer.observer != null
+                && qu.getPeer(changingIndex).peer.follower == null
+                && qu.getPeer(changingIndex).peer.leader == null);
+        Assert.assertTrue(qu.getPeer(changingIndex).peer.getPeerState() == ServerState.OBSERVING);
+
+        QuorumPeer qp = qu.getPeer(changingIndex).peer;
+        String localPeerBeanName = CommonNames.DOMAIN
+                + ":name0=ReplicatedServer_id" + changingIndex
+                + ",name1=replica." + changingIndex;
+
+        // localPeerBean.1 of ReplicatedServer_1
+        assertLocalPeerMXBeanAttributes(qp, localPeerBeanName, true);
+
+        // assert remotePeerBean.1 of ReplicatedServer_2
+        changingQS2 = peer2.getView().get(new Long(changingIndex));
+        assertRemotePeerMXBeanAttributes(changingQS2, remotePeerBean2);
+
+        // assert remotePeerBean.1 of ReplicatedServer_3
+        changingQS3 = peer3.getView().get(new Long(changingIndex));
+        assertRemotePeerMXBeanAttributes(changingQS3, remotePeerBean3);
+
+        closeAllHandles(zkArr);
+    }
+
+    private void assertLocalPeerMXBeanAttributes(QuorumPeer qp,
+            String beanName, Boolean isPartOfEnsemble) throws Exception {
+        Assert.assertEquals("Mismatches LearnerType!", qp.getLearnerType()
+                .name(), JMXEnv.ensureBeanAttribute(beanName, "LearnerType"));
+        Assert.assertEquals("Mismatches ClientAddress!",
+                HostNameUtils.getHostString(qp.getClientAddress()) + ":"
+                        + qp.getClientAddress().getPort(),
+                JMXEnv.ensureBeanAttribute(beanName, "ClientAddress"));
+        Assert.assertEquals("Mismatches LearnerType!",
+                HostNameUtils.getHostString(qp.getElectionAddress()) + ":"
+                        + qp.getElectionAddress().getPort(),
+                JMXEnv.ensureBeanAttribute(beanName, "ElectionAddress"));
+        Assert.assertEquals("Mismatches PartOfEnsemble!", isPartOfEnsemble,
+                JMXEnv.ensureBeanAttribute(beanName, "PartOfEnsemble"));
+        Assert.assertEquals("Mismatches ConfigVersion!", qp.getQuorumVerifier()
+                .getVersion(), JMXEnv.ensureBeanAttribute(beanName,
+                "ConfigVersion"));
+        Assert.assertEquals("Mismatches QuorumSystemInfo!", qp
+                .getQuorumVerifier().toString(), JMXEnv.ensureBeanAttribute(
+                beanName, "QuorumSystemInfo"));
+    }
+
+    private void assertRemotePeerMXBeanAttributes(QuorumServer qs,
+            String beanName) throws Exception {
+        Assert.assertEquals("Mismatches LearnerType!", qs.type.name(),
+                JMXEnv.ensureBeanAttribute(beanName, "LearnerType"));
+        Assert.assertEquals("Mismatches ClientAddress!",
+                HostNameUtils.getHostString(qs.clientAddr) + ":"
+                        + qs.clientAddr.getPort(),
+                JMXEnv.ensureBeanAttribute(beanName, "ClientAddress"));
+        Assert.assertEquals("Mismatches ElectionAddress!",
+                HostNameUtils.getHostString(qs.electionAddr) + ":"
+                        + qs.electionAddr.getPort(),
+                JMXEnv.ensureBeanAttribute(beanName, "ElectionAddress"));
+        Assert.assertEquals("Mismatches QuorumAddress!", qs.addr.getHostName()
+                + ":" + qs.addr.getPort(),
+                JMXEnv.ensureBeanAttribute(beanName, "QuorumAddress"));
+    }
 }