incubator-ratis git commit: RATIS-94. Expose basic information over JMX. Contributed by Elek, Marton

[ji...@apache.org]

Repository: incubator-ratis
Updated Branches:
  refs/heads/master e2bdc2478 -> 812d7dbeb


RATIS-94. Expose basic information over JMX. Contributed by Elek, Marton


Project: http://git-wip-us.apache.org/repos/asf/incubator-ratis/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ratis/commit/812d7dbe
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ratis/tree/812d7dbe
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ratis/diff/812d7dbe

Branch: refs/heads/master
Commit: 812d7dbeb5361911e269fa68995a2a7179cdf6be
Parents: e2bdc24
Author: Jing Zhao <ji...@apache.org>
Authored: Thu Aug 10 22:30:50 2017 -0700
Committer: Jing Zhao <ji...@apache.org>
Committed: Thu Aug 10 22:30:50 2017 -0700

----------------------------------------------------------------------
 .../apache/ratis/server/RaftServerMXBean.java   |  60 +++++++++++
 .../apache/ratis/server/impl/LeaderState.java   |  12 ++-
 .../ratis/server/impl/RaftServerImpl.java       |  58 +++++++++++
 .../apache/ratis/server/TestRaftServerJmx.java  | 100 +++++++++++++++++++
 4 files changed, 229 insertions(+), 1 deletion(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-ratis/blob/812d7dbe/ratis-server/src/main/java/org/apache/ratis/server/RaftServerMXBean.java
----------------------------------------------------------------------
diff --git a/ratis-server/src/main/java/org/apache/ratis/server/RaftServerMXBean.java b/ratis-server/src/main/java/org/apache/ratis/server/RaftServerMXBean.java
new file mode 100644
index 0000000..9e6dc1b
--- /dev/null
+++ b/ratis-server/src/main/java/org/apache/ratis/server/RaftServerMXBean.java
@@ -0,0 +1,60 @@
+/**
+ * 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.ratis.server;
+
+import org.apache.ratis.protocol.RaftPeer;
+
+import java.util.List;
+
+/**
+ * JMX information about the state of the current raft cluster.
+ */
+public interface RaftServerMXBean {
+
+  /**
+   * Identifier of the current server.
+   */
+  String getId();
+
+  /**
+   * Identifier of the leader node.
+   */
+  String getLeaderId();
+
+  /**
+   * Latest RAFT term.
+   */
+  long getCurrentTerm();
+
+  /**
+   * Cluster identifier.
+   */
+  String getGroupId();
+
+  /**
+   * RAFT Role of the server.
+   */
+  String getRole();
+
+  /**
+   * Addresses of the followers, only for leaders
+   */
+  List<String> getFollowers();
+
+
+}

http://git-wip-us.apache.org/repos/asf/incubator-ratis/blob/812d7dbe/ratis-server/src/main/java/org/apache/ratis/server/impl/LeaderState.java
----------------------------------------------------------------------
diff --git a/ratis-server/src/main/java/org/apache/ratis/server/impl/LeaderState.java b/ratis-server/src/main/java/org/apache/ratis/server/impl/LeaderState.java
index 0415aab..de88382 100644
--- a/ratis-server/src/main/java/org/apache/ratis/server/impl/LeaderState.java
+++ b/ratis-server/src/main/java/org/apache/ratis/server/impl/LeaderState.java
@@ -32,6 +32,7 @@ import java.io.IOException;
 import java.util.*;
 import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 
@@ -109,7 +110,7 @@ public class LeaderState {
     Collection<RaftPeer> others = conf.getOtherPeers(state.getSelfId());
     final Timestamp t = new Timestamp().addTimeMs(-server.getMaxTimeoutMs());
     placeHolderIndex = raftLog.getNextIndex();
-    senders = new ArrayList<>(others.size());
+    senders = new CopyOnWriteArrayList<LogAppender>();
 
     for (RaftPeer p : others) {
       senders.add(server.newLogAppender(this, p, t, placeHolderIndex, true));
@@ -588,4 +589,13 @@ public class LeaderState {
               + newConf + ". Timeout when bootstrapping new peers."));
     }
   }
+
+  /**
+   * @return the RaftPeer (address and id) information of the followers.
+   */
+  List<RaftPeer> getFollowers() {
+    return Collections.unmodifiableList(senders.stream()
+        .map(sender -> sender.getFollower().getPeer())
+        .collect(Collectors.toList()));
+  }
 }

http://git-wip-us.apache.org/repos/asf/incubator-ratis/blob/812d7dbe/ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerImpl.java
----------------------------------------------------------------------
diff --git a/ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerImpl.java b/ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerImpl.java
index 9a5e8bc..9d7788e 100644
--- a/ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerImpl.java
+++ b/ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerImpl.java
@@ -20,6 +20,7 @@ package org.apache.ratis.server.impl;
 import org.apache.ratis.conf.RaftProperties;
 import org.apache.ratis.protocol.*;
 import org.apache.ratis.server.RaftServerConfigKeys;
+import org.apache.ratis.server.RaftServerMXBean;
 import org.apache.ratis.server.RaftServerRpc;
 import org.apache.ratis.server.protocol.RaftServerProtocol;
 import org.apache.ratis.server.protocol.TermIndex;
@@ -33,14 +34,18 @@ import org.apache.ratis.util.*;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import javax.management.MBeanServer;
+import javax.management.ObjectName;
 import java.io.IOException;
 import java.io.InterruptedIOException;
+import java.lang.management.ManagementFactory;
 import java.util.*;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ThreadLocalRandom;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Supplier;
+import java.util.stream.Collectors;
 
 import static org.apache.ratis.server.impl.ServerProtoUtils.toRaftConfiguration;
 import static org.apache.ratis.shaded.proto.RaftProtos.AppendEntriesReplyProto.AppendResult.*;
@@ -156,6 +161,20 @@ public class RaftServerImpl implements RaftServerProtocol,
       LOG.debug("{} starts with initializing state", getId());
       startInitializing();
     }
+    registerMBean();
+
+  }
+
+  private void registerMBean() {
+    try {
+      final MBeanServer mbs = ManagementFactory.getPlatformMBeanServer();
+      ObjectName name =
+          new ObjectName("Ratis:service=RaftServer,group=" + getGroupId() + ",id=" + getId());
+      mbs.registerMBean(new RaftServerJmxAdapter(), name);
+    } catch (Exception ex) {
+      LOG.error("RaftServer JMX bean can't be registered", ex);
+    }
+
   }
 
   /**
@@ -926,4 +945,43 @@ public class RaftServerImpl implements RaftServerProtocol,
       replyPendingRequest(next, stateMachineFuture);
     }
   }
+
+  private class RaftServerJmxAdapter implements RaftServerMXBean {
+
+    @Override
+    public String getId() {
+      return getState().getSelfId().toString();
+    }
+
+    @Override
+    public String getLeaderId() {
+      return getState().getLeaderId().toString();
+    }
+
+    @Override
+    public long getCurrentTerm() {
+      return getState().getCurrentTerm();
+    }
+
+    @Override
+    public String getGroupId() {
+      return RaftServerImpl.this.getGroupId().toString();
+    }
+
+    @Override
+    public String getRole() {
+      return role.toString();
+    }
+
+    @Override
+    public List<String> getFollowers() {
+      return Optional.ofNullable(leaderState)
+          .map(leader ->
+              leader.getFollowers().stream()
+                  .map(RaftPeer::toString)
+                  .collect(Collectors.toList()))
+          .orElse(new ArrayList<>());
+    }
+
+  }
 }

http://git-wip-us.apache.org/repos/asf/incubator-ratis/blob/812d7dbe/ratis-server/src/test/java/org/apache/ratis/server/TestRaftServerJmx.java
----------------------------------------------------------------------
diff --git a/ratis-server/src/test/java/org/apache/ratis/server/TestRaftServerJmx.java b/ratis-server/src/test/java/org/apache/ratis/server/TestRaftServerJmx.java
new file mode 100644
index 0000000..ae96c02
--- /dev/null
+++ b/ratis-server/src/test/java/org/apache/ratis/server/TestRaftServerJmx.java
@@ -0,0 +1,100 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.ratis.server;
+
+import org.apache.log4j.Level;
+import org.apache.ratis.MiniRaftCluster;
+import org.apache.ratis.RaftBasicTests;
+import org.apache.ratis.client.RaftClient;
+import org.apache.ratis.conf.RaftProperties;
+import org.apache.ratis.server.impl.RaftServerImpl;
+import org.apache.ratis.server.simulation.MiniRaftClusterWithSimulatedRpc;
+import org.apache.ratis.util.LogUtils;
+import org.junit.*;
+import org.junit.rules.Timeout;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.management.MBeanServer;
+import javax.management.ObjectInstance;
+import javax.management.ObjectName;
+import java.io.IOException;
+import java.lang.management.ManagementFactory;
+import java.util.Set;
+
+import static org.apache.ratis.RaftTestUtil.waitForLeader;
+
+public class TestRaftServerJmx {
+  static {
+    LogUtils.setLogLevel(RaftServerImpl.LOG, Level.DEBUG);
+    LogUtils.setLogLevel(RaftClient.LOG, Level.DEBUG);
+  }
+
+  public static final Logger LOG = LoggerFactory.getLogger(RaftBasicTests.class);
+
+  public static final int NUM_SERVERS = 5;
+
+  protected static final RaftProperties properties = new RaftProperties();
+
+  private final MiniRaftClusterWithSimulatedRpc cluster;
+
+  public TestRaftServerJmx() throws IOException {
+    cluster = MiniRaftClusterWithSimulatedRpc.FACTORY.newCluster(
+        NUM_SERVERS, getProperties());
+  }
+
+  public RaftProperties getProperties() {
+    return properties;
+  }
+
+  @Rule
+  public Timeout globalTimeout = new Timeout(120 * 1000);
+
+  @Before
+  public void setup() throws IOException {
+    Assert.assertNull(getCluster().getLeader());
+    getCluster().start();
+  }
+
+  @After
+  public void tearDown() {
+    final MiniRaftCluster cluster = getCluster();
+    if (cluster != null) {
+      cluster.shutdown();
+    }
+  }
+
+  public MiniRaftClusterWithSimulatedRpc getCluster() {
+    return cluster;
+  }
+
+  @Test
+  public void testJmxBeans() throws Exception {
+    RaftServerImpl leader = waitForLeader(cluster);
+    System.out.println(cluster.getLeader());
+    MBeanServer platformMBeanServer = ManagementFactory.getPlatformMBeanServer();
+    Set<ObjectInstance> objectInstances = platformMBeanServer.queryMBeans(new ObjectName("Ratis:*"), null);
+    Assert.assertEquals(NUM_SERVERS, objectInstances.size());
+
+    for (ObjectInstance instance : objectInstances) {
+      Object groupId = platformMBeanServer.getAttribute(instance.getObjectName(), "GroupId");
+      Assert.assertEquals(cluster.getGroupId().toString(), groupId);
+    }
+  }
+
+}