YARN-905. Add state filters to nodes CLI (Wei Yan via Sandy Ryza)

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

hadoop-yarn-project/CHANGES.txt

@@ -23,6 +23,8 @@ Release 2.3.0 - UNRELEASED
 
   IMPROVEMENTS
 
+    YARN-905. Add state filters to nodes CLI (Wei Yan via Sandy Ryza)
+
   OPTIMIZATIONS
 
   BUG FIXES

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/cli/NodeCLI.java

@@ -21,11 +21,14 @@ import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.io.PrintWriter;
 import java.util.Date;
+import java.util.HashSet;
 import java.util.List;
+import java.util.Set;
 
 import org.apache.commons.cli.CommandLine;
 import org.apache.commons.cli.GnuParser;
 import org.apache.commons.cli.HelpFormatter;
+import org.apache.commons.cli.Option;
 import org.apache.commons.cli.Options;
 import org.apache.commons.lang.time.DateFormatUtils;
 import org.apache.hadoop.classification.InterfaceAudience.Private;
@@ -40,9 +43,12 @@ import org.apache.hadoop.yarn.util.ConverterUtils;
 @Private
 @Unstable
 public class NodeCLI extends YarnCLI {
-  private static final String NODES_PATTERN = "%16s\t%10s\t%17s\t%18s" +
+  private static final String NODES_PATTERN = "%16s\t%15s\t%17s\t%18s" +
     System.getProperty("line.separator");
 
+  private static final String NODE_STATE_CMD = "states";
+  private static final String NODE_ALL = "all";
+
   public static void main(String[] args) throws Exception {
     NodeCLI cli = new NodeCLI();
     cli.setSysOutPrintStream(System.out);
@@ -57,7 +63,18 @@ public class NodeCLI extends YarnCLI {
 
     Options opts = new Options();
     opts.addOption(STATUS_CMD, true, "Prints the status report of the node.");
-    opts.addOption(LIST_CMD, false, "Lists all the nodes in the RUNNING state.");
+    opts.addOption(LIST_CMD, false, "List all running nodes. " +
+        "Supports optional use of --states to filter nodes " +
+        "based on node state, all --all to list all nodes.");
+    Option nodeStateOpt = new Option(NODE_STATE_CMD, true,
+        "Works with -list to filter nodes based on their states.");
+    nodeStateOpt.setValueSeparator(',');
+    nodeStateOpt.setArgs(Option.UNLIMITED_VALUES);
+    nodeStateOpt.setArgName("Comma-separated list of node states");
+    opts.addOption(nodeStateOpt);
+    Option allOpt = new Option(NODE_ALL, false,
+        "Works with -list to list all nodes.");
+    opts.addOption(allOpt);
     CommandLine cliParser = new GnuParser().parse(opts, args);
 
     int exitCode = -1;
@@ -68,7 +85,24 @@ public class NodeCLI extends YarnCLI {
       }
       printNodeStatus(cliParser.getOptionValue("status"));
     } else if (cliParser.hasOption("list")) {
-      listClusterNodes();
+      Set<NodeState> nodeStates = new HashSet<NodeState>();
+      if (cliParser.hasOption(NODE_ALL)) {
+        for (NodeState state : NodeState.values()) {
+          nodeStates.add(state);
+        }
+      } else if (cliParser.hasOption(NODE_STATE_CMD)) {
+        String[] types = cliParser.getOptionValues(NODE_STATE_CMD);
+        if (types != null) {
+          for (String type : types) {
+            if (!type.trim().isEmpty()) {
+              nodeStates.add(NodeState.valueOf(type.trim().toUpperCase()));
+            }
+          }
+        }
+      } else {
+        nodeStates.add(NodeState.RUNNING);
+      }
+      listClusterNodes(nodeStates);
     } else {
       syserr.println("Invalid Command Usage : ");
       printUsage(opts);
@@ -86,14 +120,17 @@ public class NodeCLI extends YarnCLI {
   }
 
   /**
-   * Lists all the nodes present in the cluster
+   * Lists the nodes matching the given node states
    * 
+   * @param nodeStates
    * @throws YarnException
    * @throws IOException
    */
-  private void listClusterNodes() throws YarnException, IOException {
+  private void listClusterNodes(Set<NodeState> nodeStates) 
+            throws YarnException, IOException {
     PrintWriter writer = new PrintWriter(sysout);
-    List<NodeReport> nodesReport = client.getNodeReports(NodeState.RUNNING);
+    List<NodeReport> nodesReport = client.getNodeReports(
+                                       nodeStates.toArray(new NodeState[0]));
     writer.println("Total Nodes:" + nodesReport.size());
     writer.printf(NODES_PATTERN, "Node-Id", "Node-State", "Node-Http-Address",
         "Running-Containers");

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/cli/TestYarnCLI.java

@@ -363,36 +363,239 @@ public class TestYarnCLI {
 
   @Test
   public void testListClusterNodes() throws Exception {
+    List<NodeReport> nodeReports = new ArrayList<NodeReport>();
+    nodeReports.addAll(getNodeReports(1, NodeState.NEW));
+    nodeReports.addAll(getNodeReports(2, NodeState.RUNNING));
+    nodeReports.addAll(getNodeReports(1, NodeState.UNHEALTHY));
+    nodeReports.addAll(getNodeReports(1, NodeState.DECOMMISSIONED));
+    nodeReports.addAll(getNodeReports(1, NodeState.REBOOTED));
+    nodeReports.addAll(getNodeReports(1, NodeState.LOST));
+
     NodeCLI cli = new NodeCLI();
-    when(client.getNodeReports(NodeState.RUNNING)).thenReturn(
-        getNodeReports(3));
     cli.setClient(client);
     cli.setSysOutPrintStream(sysOut);
-    int result = cli.run(new String[] { "-list" });
+
+    Set<NodeState> nodeStates = new HashSet<NodeState>();
+    nodeStates.add(NodeState.NEW);
+    NodeState[] states = nodeStates.toArray(new NodeState[0]);
+    when(client.getNodeReports(states))
+        .thenReturn(getNodeReports(nodeReports, nodeStates));
+    int result = cli.run(new String[] { "-list", "--states", "NEW" });
     assertEquals(0, result);
-    verify(client).getNodeReports(NodeState.RUNNING);
+    verify(client).getNodeReports(states);
     ByteArrayOutputStream baos = new ByteArrayOutputStream();
     PrintWriter pw = new PrintWriter(baos);
-    pw.println("Total Nodes:3");
+    pw.println("Total Nodes:1");
+    pw.print("         Node-Id\t     Node-State\tNode-Http-Address\t");
+    pw.println("Running-Containers");
+    pw.print("         host0:0\t            NEW\t       host1:8888");
+    pw.println("\t                 0");
+    pw.close();
+    String nodesReportStr = baos.toString("UTF-8");
+    Assert.assertEquals(nodesReportStr, sysOutStream.toString());
+    verify(sysOut, times(1)).write(any(byte[].class), anyInt(), anyInt());
+
+    sysOutStream.reset();
+    nodeStates.clear();
+    nodeStates.add(NodeState.RUNNING);
+    states = nodeStates.toArray(new NodeState[0]);
+    when(client.getNodeReports(states))
+        .thenReturn(getNodeReports(nodeReports, nodeStates));
+    result = cli.run(new String[] { "-list", "--states", "RUNNING" });
+    assertEquals(0, result);
+    verify(client).getNodeReports(states);
+    baos = new ByteArrayOutputStream();
+    pw = new PrintWriter(baos);
+    pw.println("Total Nodes:2");
     pw.print("         Node-Id\t     Node-State\tNode-Http-Address\t");
     pw.println("Running-Containers");
     pw.print("         host0:0\t        RUNNING\t       host1:8888");
     pw.println("\t                 0");
     pw.print("         host1:0\t        RUNNING\t       host1:8888");
     pw.println("\t                 0");
-    pw.print("         host2:0\t   RUNNING\t       host1:8888");
+    pw.close();
+    nodesReportStr = baos.toString("UTF-8");
+    Assert.assertEquals(nodesReportStr, sysOutStream.toString());
+    verify(sysOut, times(2)).write(any(byte[].class), anyInt(), anyInt());
+
+    sysOutStream.reset();
+    result = cli.run(new String[] { "-list" });
+    assertEquals(0, result);
+    Assert.assertEquals(nodesReportStr, sysOutStream.toString());
+    verify(sysOut, times(3)).write(any(byte[].class), anyInt(), anyInt());
+
+    sysOutStream.reset();
+    nodeStates.clear();
+    nodeStates.add(NodeState.UNHEALTHY);
+    states = nodeStates.toArray(new NodeState[0]);
+    when(client.getNodeReports(states))
+        .thenReturn(getNodeReports(nodeReports, nodeStates));
+    result = cli.run(new String[] { "-list", "--states", "UNHEALTHY" });
+    assertEquals(0, result);
+    verify(client).getNodeReports(states);
+    baos = new ByteArrayOutputStream();
+    pw = new PrintWriter(baos);
+    pw.println("Total Nodes:1");
+    pw.print("         Node-Id\t     Node-State\tNode-Http-Address\t");
+    pw.println("Running-Containers");
+    pw.print("         host0:0\t      UNHEALTHY\t       host1:8888");
     pw.println("\t                 0");
     pw.close();
-    String nodesReportStr = baos.toString("UTF-8");
+    nodesReportStr = baos.toString("UTF-8");
     Assert.assertEquals(nodesReportStr, sysOutStream.toString());
-    verify(sysOut, times(1)).write(any(byte[].class), anyInt(), anyInt());
+    verify(sysOut, times(4)).write(any(byte[].class), anyInt(), anyInt());
+
+    sysOutStream.reset();
+    nodeStates.clear();
+    nodeStates.add(NodeState.DECOMMISSIONED);
+    states = nodeStates.toArray(new NodeState[0]);
+    when(client.getNodeReports(states))
+        .thenReturn(getNodeReports(nodeReports, nodeStates));
+    result = cli.run(new String[] { "-list", "--states", "DECOMMISSIONED" });
+    assertEquals(0, result);
+    verify(client).getNodeReports(states);
+    baos = new ByteArrayOutputStream();
+    pw = new PrintWriter(baos);
+    pw.println("Total Nodes:1");
+    pw.print("         Node-Id\t     Node-State\tNode-Http-Address\t");
+    pw.println("Running-Containers");
+    pw.print("         host0:0\t DECOMMISSIONED\t       host1:8888");
+    pw.println("\t                 0");
+    pw.close();
+    nodesReportStr = baos.toString("UTF-8");
+    Assert.assertEquals(nodesReportStr, sysOutStream.toString());
+    verify(sysOut, times(5)).write(any(byte[].class), anyInt(), anyInt());
+
+    sysOutStream.reset();
+    nodeStates.clear();
+    nodeStates.add(NodeState.REBOOTED);
+    states = nodeStates.toArray(new NodeState[0]);
+    when(client.getNodeReports(states))
+        .thenReturn(getNodeReports(nodeReports, nodeStates));
+    result = cli.run(new String[] { "-list", "--states", "REBOOTED" });
+    assertEquals(0, result);
+    verify(client).getNodeReports(states);
+    baos = new ByteArrayOutputStream();
+    pw = new PrintWriter(baos);
+    pw.println("Total Nodes:1");
+    pw.print("         Node-Id\t     Node-State\tNode-Http-Address\t");
+    pw.println("Running-Containers");
+    pw.print("         host0:0\t       REBOOTED\t       host1:8888");
+    pw.println("\t                 0");
+    pw.close();
+    nodesReportStr = baos.toString("UTF-8");
+    Assert.assertEquals(nodesReportStr, sysOutStream.toString());
+    verify(sysOut, times(6)).write(any(byte[].class), anyInt(), anyInt());
+
+    sysOutStream.reset();
+    nodeStates.clear();
+    nodeStates.add(NodeState.LOST);
+    states = nodeStates.toArray(new NodeState[0]);
+    when(client.getNodeReports(states))
+        .thenReturn(getNodeReports(nodeReports, nodeStates));
+    result = cli.run(new String[] { "-list", "--states", "LOST" });
+    assertEquals(0, result);
+    verify(client).getNodeReports(states);
+    baos = new ByteArrayOutputStream();
+    pw = new PrintWriter(baos);
+    pw.println("Total Nodes:1");
+    pw.print("         Node-Id\t     Node-State\tNode-Http-Address\t");
+    pw.println("Running-Containers");
+    pw.print("         host0:0\t           LOST\t       host1:8888");
+    pw.println("\t                 0");
+    pw.close();
+    nodesReportStr = baos.toString("UTF-8");
+    Assert.assertEquals(nodesReportStr, sysOutStream.toString());
+    verify(sysOut, times(7)).write(any(byte[].class), anyInt(), anyInt());
+
+    sysOutStream.reset();
+    nodeStates.clear();
+    nodeStates.add(NodeState.NEW);
+    nodeStates.add(NodeState.RUNNING);
+    nodeStates.add(NodeState.LOST);
+    nodeStates.add(NodeState.REBOOTED);
+    states = nodeStates.toArray(new NodeState[0]);
+    when(client.getNodeReports(states))
+        .thenReturn(getNodeReports(nodeReports, nodeStates));
+    result = cli.run(new String[] { "-list", "--states", 
+                                        "NEW,RUNNING,LOST,REBOOTED" });
+    assertEquals(0, result);
+    verify(client).getNodeReports(states);
+    baos = new ByteArrayOutputStream();
+    pw = new PrintWriter(baos);
+    pw.println("Total Nodes:5");
+    pw.print("         Node-Id\t     Node-State\tNode-Http-Address\t");
+    pw.println("Running-Containers");
+    pw.print("         host0:0\t            NEW\t       host1:8888");
+    pw.println("\t                 0");
+    pw.print("         host0:0\t        RUNNING\t       host1:8888");
+    pw.println("\t                 0");
+    pw.print("         host1:0\t        RUNNING\t       host1:8888");
+    pw.println("\t                 0");
+    pw.print("         host0:0\t       REBOOTED\t       host1:8888");
+    pw.println("\t                 0");
+    pw.print("         host0:0\t           LOST\t       host1:8888");
+    pw.println("\t                 0");
+    pw.close();
+    nodesReportStr = baos.toString("UTF-8");
+    Assert.assertEquals(nodesReportStr, sysOutStream.toString());
+    verify(sysOut, times(8)).write(any(byte[].class), anyInt(), anyInt());
+
+    sysOutStream.reset();
+    nodeStates.clear();
+    for (NodeState s : NodeState.values()) {
+      nodeStates.add(s);
+    }
+    states = nodeStates.toArray(new NodeState[0]);
+    when(client.getNodeReports(states))
+        .thenReturn(getNodeReports(nodeReports, nodeStates));
+    result = cli.run(new String[] { "-list", "--all" });
+    assertEquals(0, result);
+    verify(client).getNodeReports(states);
+    baos = new ByteArrayOutputStream();
+    pw = new PrintWriter(baos);
+    pw.println("Total Nodes:7");
+    pw.print("         Node-Id\t     Node-State\tNode-Http-Address\t");
+    pw.println("Running-Containers");
+    pw.print("         host0:0\t            NEW\t       host1:8888");
+    pw.println("\t                 0");
+    pw.print("         host0:0\t        RUNNING\t       host1:8888");
+    pw.println("\t                 0");
+    pw.print("         host1:0\t        RUNNING\t       host1:8888");
+    pw.println("\t                 0");
+    pw.print("         host0:0\t      UNHEALTHY\t       host1:8888");
+    pw.println("\t                 0");
+    pw.print("         host0:0\t DECOMMISSIONED\t       host1:8888");
+    pw.println("\t                 0");
+    pw.print("         host0:0\t       REBOOTED\t       host1:8888");
+    pw.println("\t                 0");
+    pw.print("         host0:0\t           LOST\t       host1:8888");
+    pw.println("\t                 0");
+    pw.close();
+    nodesReportStr = baos.toString("UTF-8");
+    Assert.assertEquals(nodesReportStr, sysOutStream.toString());
+    verify(sysOut, times(9)).write(any(byte[].class), anyInt(), anyInt());
+  }
+
+  private List<NodeReport> getNodeReports(
+      List<NodeReport> nodeReports,
+      Set<NodeState> nodeStates) {
+    List<NodeReport> reports = new ArrayList<NodeReport>();
+
+    for (NodeReport nodeReport : nodeReports) {
+      if (nodeStates.contains(nodeReport.getNodeState())) {
+        reports.add(nodeReport);
+      }
+    }
+    return reports;
   }
 
   @Test
   public void testNodeStatus() throws Exception {
     NodeId nodeId = NodeId.newInstance("host0", 0);
     NodeCLI cli = new NodeCLI();
-    when(client.getNodeReports()).thenReturn(getNodeReports(3));
+    when(client.getNodeReports()).thenReturn(
+                    getNodeReports(3, NodeState.RUNNING));
     cli.setClient(client);
     cli.setSysOutPrintStream(sysOut);
     cli.setSysErrPrintStream(sysErr);
@@ -424,7 +627,8 @@ public class TestYarnCLI {
   public void testAbsentNodeStatus() throws Exception {
     NodeId nodeId = NodeId.newInstance("Absenthost0", 0);
     NodeCLI cli = new NodeCLI();
-    when(client.getNodeReports()).thenReturn(getNodeReports(0));
+    when(client.getNodeReports()).thenReturn(
+                getNodeReports(0, NodeState.RUNNING));
     cli.setClient(client);
     cli.setSysOutPrintStream(sysOut);
     cli.setSysErrPrintStream(sysErr);
@@ -452,12 +656,12 @@ public class TestYarnCLI {
     verify(sysErr).println("Invalid Command Usage : ");
   }
 
-  private List<NodeReport> getNodeReports(int noOfNodes) {
+  private List<NodeReport> getNodeReports(int noOfNodes, NodeState state) {
     List<NodeReport> nodeReports = new ArrayList<NodeReport>();
 
     for (int i = 0; i < noOfNodes; i++) {
       NodeReport nodeReport = NodeReport.newInstance(NodeId
-        .newInstance("host" + i, 0), NodeState.RUNNING, "host" + 1 + ":8888",
+        .newInstance("host" + i, 0), state, "host" + 1 + ":8888",
           "rack1", Records.newRecord(Resource.class), Records
               .newRecord(Resource.class), 0, "", 0);
       nodeReports.add(nodeReport);