Fix #1923, Fix chaos in collecting SSM node info

Intel-bigdata · Sep 12, 2018 · ea6f083 · ea6f083
1 parent 6ebd540
commit ea6f083
Show file tree

Hide file tree

Showing 6 changed files with 29 additions and 25 deletions.
diff --git a/smart-agent/src/main/java/org/smartdata/agent/SmartAgent.java b/smart-agent/src/main/java/org/smartdata/agent/SmartAgent.java
@@ -90,7 +90,7 @@ public static void main(String[] args) throws IOException {
     }
     String agentAddress = AgentUtils.getAgentAddress(conf);
     LOG.info("Agent address: " + agentAddress);
-    RegisterNewAgent.getInstance(agentAddress + "-" + getDateString());
+    RegisterNewAgent.getInstance("SSMAgent@" + agentAddress.replaceAll(":.*$", ""));
 
     HadoopUtil.setSmartConfByHadoop(conf);
     agent.authentication(conf);

diff --git a/smart-engine/src/main/java/org/smartdata/server/engine/ActiveServerInfo.java b/smart-engine/src/main/java/org/smartdata/server/engine/ActiveServerInfo.java
@@ -23,17 +23,16 @@
 public class ActiveServerInfo extends NodeInfo {
   private static ActiveServerInfo inst;
 
-  private ActiveServerInfo(String id, String location, ExecutorType executorType) {
-    super(id, location, executorType);
+  private ActiveServerInfo(String id, String location) {
+    super(id, location, ExecutorType.LOCAL);
   }
 
   public static ActiveServerInfo getInstance() {
     assert inst != null;
     return inst;
   }
 
-  public static void setInstance(String id, String location,
-      ExecutorType executorType) {
-    inst = new ActiveServerInfo(id, location, executorType);
+  public static void setInstance(String id, String location) {
+    inst = new ActiveServerInfo(id, location);
   }
 }
diff --git a/smart-engine/src/main/java/org/smartdata/server/engine/cmdlet/CmdletDispatcher.java b/smart-engine/src/main/java/org/smartdata/server/engine/cmdlet/CmdletDispatcher.java
@@ -357,19 +357,15 @@ private boolean dispatch(LaunchCmdlet cmdlet) {
       }
 
       NodeCmdletMetrics metrics = regNodeInfos.get(nodeId);
-      String host = "";
       if (metrics != null) {
-        host = metrics.getNodeInfo().getHost();
         metrics.incCmdletsInExecution();
       }
-      updateCmdActionStatus(cmdlet, host);
+      updateCmdActionStatus(cmdlet, nodeId);
       dispatchedToSrvs.put(cmdlet.getCmdletId(), selected.getExecutorType());
 
       if (logDispResult) {
-        LOG.info(
-            String.format(
-                "Dispatching cmdlet->[%s] to executor service %s : %s",
-                cmdlet.getCmdletId(), selected.getExecutorType(), host));
+        LOG.info(String.format("Dispatching cmdlet->[%s] to executor: %s",
+            cmdlet.getCmdletId(), nodeId));
       }
       return true;
     }

diff --git a/smart-engine/src/main/java/org/smartdata/server/engine/cmdlet/HazelcastExecutorService.java b/smart-engine/src/main/java/org/smartdata/server/engine/cmdlet/HazelcastExecutorService.java
@@ -76,14 +76,15 @@ public HazelcastExecutorService(CmdletManager cmdletManager) {
   private void initChannels() {
     for (Member worker : HazelcastUtil.getWorkerMembers(instance)) {
       ITopic<Serializable> topic = instance.getTopic(WORKER_TOPIC_PREFIX + worker.getUuid());
-      this.masterToWorkers.put(worker.getUuid(), topic);
+      this.masterToWorkers.put(getMemberNodeId(worker), topic);
     }
   }
 
   public List<StandbyServerInfo> getStandbyServers() {
     List<StandbyServerInfo> infos = new ArrayList<>();
     for (Member worker : HazelcastUtil.getWorkerMembers(instance)) {
-      infos.add(new StandbyServerInfo(worker.getUuid(), worker.getAddress().toString()));
+      infos.add(new StandbyServerInfo(getMemberNodeId(worker),
+          worker.getAddress().getHost() + ":" + worker.getAddress().getPort()));
     }
     return infos;
   }
@@ -102,7 +103,12 @@ public List<NodeInfo> getNodesInfo() {
   }
 
   private NodeInfo memberToNodeInfo(Member member) {
-    return new StandbyServerInfo(member.getUuid(), member.getAddress().toString());
+    return new StandbyServerInfo(getMemberNodeId(member),
+        member.getAddress().getHost() + ":" + member.getAddress().getPort());
+  }
+
+  private String getMemberNodeId(Member member) {
+    return "StandbySSMServer@" + member.getAddress().getHost();
   }
 
   @Override
@@ -155,20 +161,22 @@ public ClusterMembershipListener(HazelcastInstance instance) {
     @Override
     public void memberAdded(MembershipEvent membershipEvent) {
       Member worker = membershipEvent.getMember();
-      if (!masterToWorkers.containsKey(worker.getUuid())) {
+      String id = getMemberNodeId(worker);
+      if (!masterToWorkers.containsKey(id)) {
         ITopic<Serializable> topic = instance.getTopic(WORKER_TOPIC_PREFIX + worker.getUuid());
-        masterToWorkers.put(worker.getUuid(), topic);
-        members.put(worker.getUuid(), worker);
+        masterToWorkers.put(id, topic);
+        members.put(id, worker);
         EngineEventBus.post(new AddNodeMessage(memberToNodeInfo(worker)));
       }
     }
 
     @Override
     public void memberRemoved(MembershipEvent membershipEvent) {
       Member member = membershipEvent.getMember();
-      if (masterToWorkers.containsKey(member.getUuid())) {
-        masterToWorkers.get(member.getUuid()).destroy();
-        members.remove(member.getUuid());
+      String id = getMemberNodeId(member);
+      if (masterToWorkers.containsKey(id)) {
+        masterToWorkers.get(id).destroy();
+        members.remove(id);
         EngineEventBus.post(new RemoveNodeMessage(memberToNodeInfo(member)));
       }
       //Todo: recover

diff --git a/...t-engine/src/main/java/org/smartdata/server/engine/cmdlet/LocalCmdletExecutorService.java b/...t-engine/src/main/java/org/smartdata/server/engine/cmdlet/LocalCmdletExecutorService.java
@@ -42,7 +42,7 @@
 
 public class LocalCmdletExecutorService extends CmdletExecutorService implements StatusReporter {
   private static final Logger LOG = LoggerFactory.getLogger(LocalCmdletExecutorService.class);
-  private static final String ACTIVE_SERVER_ID = "ActiveSSMServer";
+  private static final String ACTIVE_SERVER_ID = "ActiveSSMServer@";
   private SmartConf conf;
   private CmdletFactory cmdletFactory;
   private CmdletExecutor cmdletExecutor;
@@ -61,7 +61,8 @@ public LocalCmdletExecutorService(SmartConf smartConf, CmdletManager cmdletManag
     this.executorService.scheduleAtFixedRate(
         statusReportTask, 1000, reportPeriod, TimeUnit.MILLISECONDS);
 
-    ActiveServerInfo.setInstance(ACTIVE_SERVER_ID, getActiveServerAddress(), ExecutorType.LOCAL);
+    ActiveServerInfo.setInstance(ACTIVE_SERVER_ID + getActiveServerAddress(),
+        getActiveServerAddress());
     EngineEventBus.post(new AddNodeMessage(ActiveServerInfo.getInstance()));
   }
 

diff --git a/smart-engine/src/main/java/org/smartdata/server/engine/cmdlet/agent/AgentUtils.java b/smart-engine/src/main/java/org/smartdata/server/engine/cmdlet/agent/AgentUtils.java
@@ -47,7 +47,7 @@ public static String getFullPath(ActorSystem system, ActorPath path) {
   }
 
   public static String getHostPort(ActorRef ref) {
-    return ref.path().address().hostPort();
+    return ref.path().address().hostPort().replaceFirst("^.*@", "");
   }
 
   public static Cancellable repeatActionUntil(ActorSystem system,