Solve #1888, Add statistic info for ssm nodes

smart-engine/src/main/java/org/smartdata/server/cluster/ActiveServerNodeCmdletMetrics.java

@@ -0,0 +1,43 @@
+/**
+ * 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.smartdata.server.cluster;
+
+/**
+ * Contains metrics specific for active SSM server related with cmdlet execution.
+ *
+ */
+public class ActiveServerNodeCmdletMetrics extends NodeCmdletMetrics {
+  private int numPendingSchedule;
+  private int numPendingDispatch;
+
+  public int getNumPendingSchedule() {
+    return numPendingSchedule;
+  }
+
+  public void setNumPendingSchedule(int numPendingSchedule) {
+    this.numPendingSchedule = numPendingSchedule;
+  }
+
+  public int getNumPendingDispatch() {
+    return numPendingDispatch;
+  }
+
+  public void setNumPendingDispatch(int numPendingDispatch) {
+    this.numPendingDispatch = numPendingDispatch;
+  }
+}

smart-engine/src/main/java/org/smartdata/server/cluster/NodeCmdletMetrics.java

@@ -0,0 +1,88 @@
+/**
+ * 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.smartdata.server.cluster;
+
+/**
+ * Contains metrics for SSM nodes related with cmdlet execution.
+ *
+ */
+public class NodeCmdletMetrics {
+  private NodeInfo nodeInfo;
+
+  private long registTime;
+  private int numExecutors;
+
+  private long cmdletsExecuted;
+  private int cmdletsInExecution;
+
+
+  public NodeInfo getNodeInfo() {
+    return nodeInfo;
+  }
+
+  public void setNodeInfo(NodeInfo nodeInfo) {
+    this.nodeInfo = nodeInfo;
+  }
+
+  public long getRegistTime() {
+    return registTime;
+  }
+
+  public void setRegistTime(long registTime) {
+    this.registTime = registTime;
+  }
+
+  public int getNumExecutors() {
+    return numExecutors;
+  }
+
+  public void setNumExecutors(int numExecutors) {
+    this.numExecutors = numExecutors;
+  }
+
+  public long getCmdletsExecuted() {
+    return cmdletsExecuted;
+  }
+
+  public void setCmdletsExecuted(long cmdletsExecuted) {
+    this.cmdletsExecuted = cmdletsExecuted;
+  }
+
+  public synchronized void incCmdletsExecuted() {
+    cmdletsExecuted++;
+  }
+
+  public int getCmdletsInExecution() {
+    return cmdletsInExecution;
+  }
+
+  public void setCmdletsInExecution(int cmdletsInExecution) {
+    this.cmdletsInExecution = cmdletsInExecution;
+  }
+
+  public synchronized void incCmdletsInExecution() {
+    cmdletsInExecution++;
+  }
+
+  public synchronized void finishCmdlet() {
+    cmdletsExecuted++;
+    if (cmdletsInExecution > 0) { // TODO: restore
+      cmdletsInExecution--;
+    }
+  }
+}

smart-engine/src/main/java/org/smartdata/server/engine/CmdletManager.java

@@ -47,6 +47,7 @@
 import org.smartdata.protocol.message.CmdletStatusUpdate;
 import org.smartdata.protocol.message.StatusMessage;
 import org.smartdata.protocol.message.StatusReport;
+import org.smartdata.server.cluster.NodeCmdletMetrics;
 import org.smartdata.server.engine.cmdlet.CmdletDispatcher;
 import org.smartdata.server.engine.cmdlet.CmdletExecutorService;
 import org.smartdata.server.engine.cmdlet.message.LaunchCmdlet;
@@ -55,6 +56,7 @@
 import java.io.IOException;
 import java.text.ParseException;
 import java.util.ArrayList;
+import java.util.Collection;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Iterator;
@@ -453,6 +455,14 @@ private boolean shouldStopSchedule() {
     return false;
   }
 
+  public int getNumPendingScheduleCmdlets() {
+    return pendingCmdlet.size() + schedulingCmdlet.size();
+  }
+
+  public Collection<NodeCmdletMetrics> getAllNodeCmdletMetrics() {
+    return dispatcher.getNodeCmdletMetrics();
+  }
+
   private int scheduleCmdlet() throws IOException {
     int nScheduled = 0;
 

smart-engine/src/main/java/org/smartdata/server/engine/cmdlet/CmdletDispatcher.java

@@ -30,13 +30,16 @@
 import org.smartdata.model.action.ActionScheduler;
 import org.smartdata.protocol.message.ActionStatus;
 import org.smartdata.protocol.message.CmdletStatus;
-import org.smartdata.server.cluster.NodeInfo;
+import org.smartdata.server.cluster.ActiveServerNodeCmdletMetrics;
+import org.smartdata.server.cluster.NodeCmdletMetrics;
+import org.smartdata.server.engine.ActiveServerInfo;
 import org.smartdata.server.engine.CmdletManager;
 import org.smartdata.server.engine.cmdlet.message.LaunchCmdlet;
 import org.smartdata.server.engine.message.NodeMessage;
 
 import java.io.IOException;
 import java.util.ArrayList;
+import java.util.Collection;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -73,7 +76,7 @@ public class CmdletDispatcher {
   private AtomicInteger index = new AtomicInteger(0);
 
   private Map<String, AtomicInteger> regNodes = new HashMap<>();
-  private Map<String, NodeInfo> regNodeInfos = new HashMap<>();
+  private Map<String, NodeCmdletMetrics> regNodeInfos = new HashMap<>();
 
   private List<List<String>> cmdExecSrvNodeIds = new ArrayList<>();
   private String[] completeOn = new String[ExecutorType.values().length];
@@ -353,8 +356,12 @@ private boolean dispatch(LaunchCmdlet cmdlet) {
         return false;
       }
 
-      NodeInfo nodeInfo = regNodeInfos.get(nodeId);
-      String host = nodeInfo == null ? "" : nodeInfo.getHost();
+      NodeCmdletMetrics metrics = regNodeInfos.get(nodeId);
+      String host = "";
+      if (metrics != null) {
+        host = metrics.getNodeInfo().getHost();
+        metrics.incCmdletsInExecution();
+      }
       updateCmdActionStatus(cmdlet, host);
       dispatchedToSrvs.put(cmdlet.getCmdletId(), selected.getExecutorType());
 
@@ -418,6 +425,12 @@ public void onCmdletFinished(long cmdletId) {
         if (regNodes.get(cmdlet.getNodeId()) != null) {
           regNodes.get(cmdlet.getNodeId()).incrementAndGet();
         }
+
+        NodeCmdletMetrics metrics = regNodeInfos.get(cmdlet.getNodeId());
+        if (metrics != null) {
+          metrics.finishCmdlet();
+        }
+
         ExecutorType t = dispatchedToSrvs.remove(cmdletId);
         updateSlotsLeft(t.ordinal(), 1);
         completeOn[t.ordinal()] = cmdlet.getNodeId();
@@ -438,7 +451,16 @@ public void onNodeMessage(NodeMessage msg, boolean isAdd) {
           return;
         } else {
           regNodes.put(nodeId, new AtomicInteger(defaultSlots));
-          regNodeInfos.put(nodeId, msg.getNodeInfo());
+          NodeCmdletMetrics metrics;
+          if (msg.getNodeInfo().getExecutorType() == ExecutorType.LOCAL) {
+            metrics = new ActiveServerNodeCmdletMetrics();
+          } else {
+            metrics = new NodeCmdletMetrics();
+          }
+          metrics.setNumExecutors(defaultSlots);
+          metrics.setRegistTime(System.currentTimeMillis());
+          metrics.setNodeInfo(msg.getNodeInfo());
+          regNodeInfos.put(nodeId, metrics);
           cmdExecSrvNodeIds.get(msg.getNodeInfo().getExecutorType().ordinal()).add(nodeId);
         }
       } else {
@@ -486,7 +508,24 @@ public int getTotalSlots() {
     return cmdExecSrvTotalInsts * defaultSlots;
   }
 
+  public Collection<NodeCmdletMetrics> getNodeCmdletMetrics() {
+    ActiveServerNodeCmdletMetrics metrics = (ActiveServerNodeCmdletMetrics) regNodeInfos.get(
+        ActiveServerInfo.getInstance().getId());
+    if (metrics != null) {
+      metrics.setNumPendingSchedule(cmdletManager.getNumPendingScheduleCmdlets());
+      metrics.setNumPendingDispatch(pendingCmdlets.size());
+    }
+    return regNodeInfos.values();
+  }
+
   public void start() {
+    if (disableLocalExec) {
+      ActiveServerNodeCmdletMetrics metrics = new ActiveServerNodeCmdletMetrics();
+      metrics.setNumExecutors(defaultSlots);
+      metrics.setRegistTime(System.currentTimeMillis());
+      metrics.setNodeInfo(ActiveServerInfo.getInstance());
+      regNodeInfos.put(ActiveServerInfo.getInstance().getId(), metrics);
+    }
     CmdletDispatcherHelper.getInst().register(this);
     int idx = 0;
     for (DispatchTask task : dispatchTasks) {

smart-zeppelin/zeppelin-server/src/main/java/org/smartdata/server/rest/ActionRestApi.java

@@ -17,16 +17,10 @@
  */
 package org.smartdata.server.rest;
 
-import com.google.gson.Gson;
-import com.google.gson.JsonArray;
-import com.google.gson.JsonElement;
-import com.google.gson.JsonParser;
-import com.google.gson.reflect.TypeToken;
 import org.apache.commons.lang.exception.ExceptionUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.smartdata.action.ActionRegistry;
-import org.smartdata.model.ActionInfo;
 import org.smartdata.server.SmartEngine;
 import org.smartdata.server.rest.message.JsonResponse;
 

smart-zeppelin/zeppelin-server/src/main/java/org/smartdata/server/rest/ClusterRestApi.java

@@ -176,6 +176,19 @@ public Response ssmNodesInfo() {
     }
   }
 
+  @GET
+  @Path("/primary/ssmnodescmdletmetrics")
+  public Response ssmNodesCmdletMetrics() {
+    try {
+      return new JsonResponse<>(Response.Status.OK,
+          smartEngine.getCmdletManager().getAllNodeCmdletMetrics()).build();
+    } catch (Exception e) {
+      logger.error("Exception in ClusterRestApi while listing metrics related with cmdlets", e);
+      return new JsonResponse<>(Response.Status.INTERNAL_SERVER_ERROR,
+          e.getMessage(), ExceptionUtils.getStackTrace(e)).build();
+    }
+  }
+
 //  @GET
 //  @Path("/alluxio/{clusterName}")
 //  public void alluxio() {