YARN-1497. Command line additions for moving apps between queues (Sandy Ryza)

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

hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/main/java/org/apache/hadoop/mapred/ResourceMgrDelegate.java

@@ -400,4 +400,10 @@ public List<ContainerReport> getContainers(
       IOException {
     return client.getContainers(applicationAttemptId);
   }
+
+  @Override
+  public void moveApplicationAcrossQueues(ApplicationId appId, String queue)
+      throws YarnException, IOException {
+    client.moveApplicationAcrossQueues(appId, queue);
+  }
 }

hadoop-yarn-project/CHANGES.txt

@@ -17,6 +17,9 @@ Trunk - Unreleased
     YARN-1499. Fair Scheduler changes for moving apps between queues (Sandy
     Ryza)
 
+    YARN-1497. Command line additions for moving apps between queues (Sandy
+    Ryza)
+
   IMPROVEMENTS
 
   OPTIMIZATIONS

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/YarnClient.java

@@ -436,4 +436,19 @@ public abstract ContainerReport getContainerReport(ContainerId containerId)
   public abstract List<ContainerReport> getContainers(
       ApplicationAttemptId applicationAttemptId) throws YarnException,
       IOException;
+  
+  /**
+   * <p>
+   * Attempts to move the given application to the given queue.
+   * </p>
+   * 
+   * @param appId
+   *    Application to move.
+   * @param queue
+   *    Queue to place it in to.
+   * @throws YarnException
+   * @throws IOException
+   */
+  public abstract void moveApplicationAcrossQueues(ApplicationId appId,
+      String queue) throws YarnException, IOException;
 }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/impl/YarnClientImpl.java

@@ -48,6 +48,7 @@
 import org.apache.hadoop.yarn.api.protocolrecords.GetQueueUserAclsInfoRequest;
 import org.apache.hadoop.yarn.api.protocolrecords.KillApplicationRequest;
 import org.apache.hadoop.yarn.api.protocolrecords.KillApplicationResponse;
+import org.apache.hadoop.yarn.api.protocolrecords.MoveApplicationAcrossQueuesRequest;
 import org.apache.hadoop.yarn.api.protocolrecords.SubmitApplicationRequest;
 import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
 import org.apache.hadoop.yarn.api.records.ApplicationAttemptReport;
@@ -478,4 +479,12 @@ public List<ContainerReport> getContainers(
     }
     throw new YarnException("History service is not enabled.");
   }
+  
+  @Override
+  public void moveApplicationAcrossQueues(ApplicationId appId,
+      String queue) throws YarnException, IOException {
+    MoveApplicationAcrossQueuesRequest request =
+        MoveApplicationAcrossQueuesRequest.newInstance(appId, queue);
+    rmClient.moveApplicationAcrossQueues(request);
+  }
 }

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

@@ -61,6 +61,7 @@ public class ApplicationCLI extends YarnCLI {
   private static final String APP_TYPE_CMD = "appTypes";
   private static final String APP_STATE_CMD = "appStates";
   private static final String ALLSTATES_OPTION = "ALL";
+  private static final String QUEUE_CMD = "queue";
   public static final String APPLICATION = "application";
   public static final String APPLICATION_ATTEMPT = "applicationattempt";
   public static final String CONTAINER = "container";
@@ -96,6 +97,10 @@ public int run(String[] args) throws Exception {
           + "and -appStates to filter applications based on application state");
     }
     opts.addOption(KILL_CMD, true, "Kills the application.");
+    opts.addOption(MOVE_TO_QUEUE_CMD, true, "Moves the application to a "
+        + "different queue.");
+    opts.addOption(QUEUE_CMD, true, "Works with the movetoqueue command to"
+        + " specify which queue to move an application to.");
     opts.addOption(HELP_CMD, false, "Displays help for all commands.");
     Option appTypeOpt = new Option(APP_TYPE_CMD, true, "Works with -list to "
         + "filter applications based on "
@@ -112,6 +117,8 @@ public int run(String[] args) throws Exception {
     appStateOpt.setArgName("States");
     opts.addOption(appStateOpt);
     opts.getOption(KILL_CMD).setArgName("Application ID");
+    opts.getOption(MOVE_TO_QUEUE_CMD).setArgName("Application ID");
+    opts.getOption(QUEUE_CMD).setArgName("Queue Name");
     opts.getOption(STATUS_CMD).setArgName("Application ID");
 
     int exitCode = -1;
@@ -202,6 +209,13 @@ public int run(String[] args) throws Exception {
         return exitCode;
       }
       killApplication(cliParser.getOptionValue(KILL_CMD));
+    } else if (cliParser.hasOption(MOVE_TO_QUEUE_CMD)) {
+      if (!cliParser.hasOption(QUEUE_CMD)) {
+        printUsage(opts);
+        return exitCode;
+      }
+      moveApplicationAcrossQueues(cliParser.getOptionValue(MOVE_TO_QUEUE_CMD),
+          cliParser.getOptionValue(QUEUE_CMD));
     } else if (cliParser.hasOption(HELP_CMD)) {
       printUsage(opts);
       return 0;
@@ -366,6 +380,28 @@ private void killApplication(String applicationId) throws YarnException,
       client.killApplication(appId);
     }
   }
+  
+  /**
+   * Kills the application with the application id as appId
+   * 
+   * @param applicationId
+   * @throws YarnException
+   * @throws IOException
+   */
+  private void moveApplicationAcrossQueues(String applicationId, String queue)
+      throws YarnException, IOException {
+    ApplicationId appId = ConverterUtils.toApplicationId(applicationId);
+    ApplicationReport appReport = client.getApplicationReport(appId);
+    if (appReport.getYarnApplicationState() == YarnApplicationState.FINISHED
+        || appReport.getYarnApplicationState() == YarnApplicationState.KILLED
+        || appReport.getYarnApplicationState() == YarnApplicationState.FAILED) {
+      sysout.println("Application " + applicationId + " has already finished ");
+    } else {
+      sysout.println("Moving application " + applicationId + " to queue " + queue);
+      client.moveApplicationAcrossQueues(appId, queue);
+      sysout.println("Successfully completed move.");
+    }
+  }
 
   /**
    * Prints the application report for an application id.

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

@@ -33,6 +33,7 @@ public abstract class YarnCLI extends Configured implements Tool {
   public static final String STATUS_CMD = "status";
   public static final String LIST_CMD = "list";
   public static final String KILL_CMD = "kill";
+  public static final String MOVE_TO_QUEUE_CMD = "movetoqueue";
   public static final String HELP_CMD = "help";
   protected PrintStream sysout;
   protected PrintStream syserr;

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

@@ -675,6 +675,7 @@ public void testAppsHelpCommand() throws Exception {
     int result = spyCli.run(new String[] { "-help" });
     Assert.assertTrue(result == 0);
     verify(spyCli).printUsage(any(Options.class));
+    System.err.println(sysOutStream.toString()); //todo sandyt remove this hejfkdsl
     Assert.assertEquals(createApplicationCLIHelpMessage(),
         sysOutStream.toString());
 
@@ -748,6 +749,56 @@ public void testKillApplication() throws Exception {
           "' doesn't exist in RM.", ex.getMessage());
     }
   }
+  
+  @Test
+  public void testMoveApplicationAcrossQueues() throws Exception {
+    ApplicationCLI cli = createAndGetAppCLI();
+    ApplicationId applicationId = ApplicationId.newInstance(1234, 5);
+
+    ApplicationReport newApplicationReport2 = ApplicationReport.newInstance(
+        applicationId, ApplicationAttemptId.newInstance(applicationId, 1),
+        "user", "queue", "appname", "host", 124, null,
+        YarnApplicationState.FINISHED, "diagnostics", "url", 0, 0,
+        FinalApplicationStatus.SUCCEEDED, null, "N/A", 0.53789f, "YARN", null);
+    when(client.getApplicationReport(any(ApplicationId.class))).thenReturn(
+        newApplicationReport2);
+    int result = cli.run(new String[] { "-movetoqueue", applicationId.toString(),
+        "-queue", "targetqueue"});
+    assertEquals(0, result);
+    verify(client, times(0)).moveApplicationAcrossQueues(
+        any(ApplicationId.class), any(String.class));
+    verify(sysOut).println(
+        "Application " + applicationId + " has already finished ");
+
+    ApplicationReport newApplicationReport = ApplicationReport.newInstance(
+        applicationId, ApplicationAttemptId.newInstance(applicationId, 1),
+        "user", "queue", "appname", "host", 124, null,
+        YarnApplicationState.RUNNING, "diagnostics", "url", 0, 0,
+        FinalApplicationStatus.SUCCEEDED, null, "N/A", 0.53789f, "YARN", null);
+    when(client.getApplicationReport(any(ApplicationId.class))).thenReturn(
+        newApplicationReport);
+    result = cli.run(new String[] { "-movetoqueue", applicationId.toString(),
+        "-queue", "targetqueue"});
+    assertEquals(0, result);
+    verify(client).moveApplicationAcrossQueues(any(ApplicationId.class),
+        any(String.class));
+    verify(sysOut).println("Moving application application_1234_0005 to queue targetqueue");
+    verify(sysOut).println("Successfully completed move.");
+
+    doThrow(new ApplicationNotFoundException("Application with id '"
+        + applicationId + "' doesn't exist in RM.")).when(client)
+        .moveApplicationAcrossQueues(applicationId, "targetqueue");
+    cli = createAndGetAppCLI();
+    try {
+      result = cli.run(new String[] { "-movetoqueue", applicationId.toString(),
+          "-queue", "targetqueue"});
+      Assert.fail();
+    } catch (Exception ex) {
+      Assert.assertTrue(ex instanceof ApplicationNotFoundException);
+      Assert.assertEquals("Application with id '" + applicationId +
+          "' doesn't exist in RM.", ex.getMessage());
+    }
+  }
 
   @Test
   public void testListClusterNodes() throws Exception {
@@ -1087,23 +1138,28 @@ private String createApplicationCLIHelpMessage() throws IOException {
     ByteArrayOutputStream baos = new ByteArrayOutputStream();
     PrintWriter pw = new PrintWriter(baos);
     pw.println("usage: application");
-    pw.println(" -appStates <States>        Works with -list to filter applications based");
+    pw.println(" -appStates <States>             Works with -list to filter applications");
-    pw.println("                            on input comma-separated list of application");
+    pw.println("                                 based on input comma-separated list of");
-    pw.println("                            states. The valid application state can be one");
+    pw.println("                                 application states. The valid application");
-    pw.println("                            of the following:");
+    pw.println("                                 state can be one of the following:");
-    pw.println("                            ALL,NEW,NEW_SAVING,SUBMITTED,ACCEPTED,RUNNING,");
+    pw.println("                                 ALL,NEW,NEW_SAVING,SUBMITTED,ACCEPTED,RUN");
-    pw.println("                            FINISHED,FAILED,KILLED");
+    pw.println("                                 NING,FINISHED,FAILED,KILLED");
-    pw.println(" -appTypes <Types>          Works with -list to filter applications based");
+    pw.println(" -appTypes <Types>               Works with -list to filter applications");
-    pw.println("                            on input comma-separated list of application");
+    pw.println("                                 based on input comma-separated list of");
-    pw.println("                            types.");
+    pw.println("                                 application types.");
-    pw.println(" -help                      Displays help for all commands.");
+    pw.println(" -help                           Displays help for all commands.");
-    pw.println(" -kill <Application ID>     Kills the application.");
+    pw.println(" -kill <Application ID>          Kills the application.");
-    pw.println(" -list                      List applications from the RM. Supports");
+    pw.println(" -list                           List applications from the RM. Supports");
-    pw.println("                            optional use of -appTypes to filter");
+    pw.println("                                 optional use of -appTypes to filter");
-    pw.println("                            applications based on application type, and");
+    pw.println("                                 applications based on application type,");
-    pw.println("                            -appStates to filter applications based on");
+    pw.println("                                 and -appStates to filter applications");
-    pw.println("                            application state");
+    pw.println("                                 based on application state");
-    pw.println(" -status <Application ID>   Prints the status of the application.");
+    pw.println(" -movetoqueue <Application ID>   Moves the application to a different");
+    pw.println("                                 queue.");
+    pw.println(" -queue <Queue Name>             Works with the movetoqueue command to");
+    pw.println("                                 specify which queue to move an");
+    pw.println("                                 application to.");
+    pw.println(" -status <Application ID>        Prints the status of the application.");
     pw.close();
     String appsHelpStr = baos.toString("UTF-8");
     return appsHelpStr;