DragonCLI

DragonCLI.java

/**
 * 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.hadoop.realtime;

import java.io.IOException;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.classification.InterfaceStability;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.ipc.RemoteException;
import org.apache.hadoop.realtime.conf.DragonConfiguration;
import org.apache.hadoop.realtime.job.JobPriority;
import org.apache.hadoop.realtime.records.JobId;
import org.apache.hadoop.realtime.records.JobReport;
import org.apache.hadoop.security.AccessControlException;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import org.apache.hadoop.yarn.api.records.ApplicationId;
import org.apache.hadoop.yarn.logaggregation.LogDumper;
import org.apache.hadoop.yarn.util.Records;

@InterfaceAudience.Public
@InterfaceStability.Evolving
/**
 * Dragon command line interface that interprets the dragon job options 
 */
public final class DragonCLI extends Configured implements Tool {
  private static final Log LOG = LogFactory.getLog(DragonCLI.class);

  private String getJobPriorityNames() {
    StringBuffer sb = new StringBuffer();
    for (JobPriority p : JobPriority.values()) {
      sb.append(p.name()).append(" ");
    }
    return sb.substring(0, sb.length()-1);
  }

  /**
   * Display usage of the command-line tool and terminate execution.
   */
  private void displayUsage(String cmd) {
    String prefix = "Usage: CLI ";
    String jobPriorityValues = getJobPriorityNames();
    String taskStates = "running, completed";
    if ("-submit".equals(cmd)) {
      System.err.println(prefix + "[" + cmd + " <job-file>]");
    } else if ("-status".equals(cmd) || "-kill".equals(cmd)) {
      System.err.println(prefix + "[" + cmd + " <job-id>]");
    } else if ("-counter".equals(cmd)) {
      System.err.println(prefix + "[" + cmd + 
        " <job-id> <group-name> <counter-name>]");
    } else if ("-events".equals(cmd)) {
      System.err.println(prefix + "[" + cmd + 
        " <job-id> <from-event-#> <#-of-events>]. Event #s start from 1.");
    } else if ("-history".equals(cmd)) {
      System.err.println(prefix + "[" + cmd + " <jobHistoryFile>]");
    } else if ("-list".equals(cmd)) {
      System.err.println(prefix + "[" + cmd + " [all]]");
    } else if ("-kill-task".equals(cmd) || "-fail-task".equals(cmd)) {
      System.err.println(prefix + "[" + cmd + " <task-attempt-id>]");
    } else if ("-set-priority".equals(cmd)) {
      System.err.println(prefix + "[" + cmd + " <job-id> <priority>]. " +
          "Valid values for priorities are: " 
          + jobPriorityValues); 
    } else if ("-list-attempt-ids".equals(cmd)) {
      System.err.println(prefix + "[" + cmd + 
          " <job-id> <task-type> <task-state>]. " +
          "Valid values for <task-state> are " + taskStates);
    } else if ("-logs".equals(cmd)) {
      System.err.println(prefix + "[" + cmd +
          " <job-id> <task-attempt-id>]. " +
          " <task-attempt-id> is optional to get task attempt logs.");      
    } else {
      System.err.printf(prefix + "<command> <args>\n");
      System.err.printf("\t[-submit <job-file>]\n");
      System.err.printf("\t[-status <job-id>]\n");
      System.err.printf("\t[-counter <job-id> <group-name> <counter-name>]\n");
      System.err.printf("\t[-kill <job-id>]\n");
      System.err.printf("\t[-set-priority <job-id> <priority>]. " +
        "Valid values for priorities are: " + jobPriorityValues + "\n");
      System.err.printf("\t[-events <job-id> <from-event-#> <#-of-events>]\n");
      System.err.printf("\t[-history <jobHistoryFile>]\n");
      System.err.printf("\t[-list [all]]\n");
      System.err.println("\t[-list-attempt-ids <job-id> <task-type> " +
        "<task-state>]. " +
        "Valid values for <task-state> are " + taskStates);
      System.err.printf("\t[-kill-task <task-attempt-id>]\n");
      System.err.printf("\t[-fail-task <task-attempt-id>]\n");
      System.err.printf("\t[-logs <job-id> <task-attempt-id>]\n\n");
      ToolRunner.printGenericCommandUsage(System.out);
    }
  }
  
  /** Construct a JobId object from given string */
  private JobId jobIdForName(String name) throws IllegalArgumentException {
    if (name == null)
      return null;
    try {
      String[] parts = name.split("_");
      if (parts.length == 3) {
        if (parts[0].equals("job")) {
          long clusterTimeStamp = Long.parseLong(parts[1]);
          int id = Integer.parseInt(parts[2]);
          JobId jobId = Records.newRecord(JobId.class);
          jobId.setId(id);
          ApplicationId appId = Records.newRecord(ApplicationId.class);
          appId.setId(id);
          appId.setClusterTimestamp(clusterTimeStamp);
          jobId.setAppId(appId);
          return jobId;
        }
      }
    } catch (Exception ex) {// fall below
    }
    throw new IllegalArgumentException("JobId string : " + name
        + " is not properly formed");
  }

  @Override
  public int run(String[] args) throws Exception {
    int exitCode = -1;
    if (args.length < 1) {
      displayUsage("");
      return exitCode;
    }    
    // process arguments
    String cmd = args[0];
    String submitJobFile = null;
    String jobid = null;
    String taskid = null;
    String historyFile = null;
    String counterGroupName = null;
    String counterName = null;
    JobPriority jp = null;
    String taskState = null;
    int fromEvent = 0;
    int nEvents = 0;
    boolean getStatus = false;
    boolean getCounter = false;
    boolean killJob = false;
    boolean listEvents = false;
    boolean viewHistory = false;
    boolean viewAllHistory = false;
    boolean listJobs = false;
    boolean listAllJobs = false;
    boolean displayTasks = false;
    boolean killTask = false;
    boolean failTask = false;
    boolean setJobPriority = false;
    boolean logs = false;

    if ("-submit".equals(cmd)) {
      if (args.length != 2) {
        displayUsage(cmd);
        return exitCode;
      }
      submitJobFile = args[1];
    } else if ("-status".equals(cmd)) {
      if (args.length != 2) {
        displayUsage(cmd);
        return exitCode;
      }
      jobid = args[1];
      getStatus = true;
    } else if("-counter".equals(cmd)) {
      if (args.length != 4) {
        displayUsage(cmd);
        return exitCode;
      }
      getCounter = true;
      jobid = args[1];
      counterGroupName = args[2];
      counterName = args[3];
    } else if ("-kill".equals(cmd)) {
      if (args.length != 2) {
        displayUsage(cmd);
        return exitCode;
      }
      jobid = args[1];
      killJob = true;
    } else if ("-set-priority".equals(cmd)) {
      if (args.length != 3) {
        displayUsage(cmd);
        return exitCode;
      }
      jobid = args[1];
      try {
        jp = JobPriority.valueOf(args[2]); 
      } catch (IllegalArgumentException iae) {
        LOG.info(iae);
        displayUsage(cmd);
        return exitCode;
      }
      setJobPriority = true; 
    } else if ("-events".equals(cmd)) {
      if (args.length != 4) {
        displayUsage(cmd);
        return exitCode;
      }
      jobid = args[1];
      fromEvent = Integer.parseInt(args[2]);
      nEvents = Integer.parseInt(args[3]);
      listEvents = true;
    } else if ("-history".equals(cmd)) {
      if (args.length != 2 && !(args.length == 3 && "all".equals(args[1]))) {
         displayUsage(cmd);
         return exitCode;
      }
      viewHistory = true;
      if (args.length == 3 && "all".equals(args[1])) {
        viewAllHistory = true;
        historyFile = args[2];
      } else {
        historyFile = args[1];
      }
    } else if ("-list".equals(cmd)) {
      if (args.length != 1 && !(args.length == 2 && "all".equals(args[1]))) {
        displayUsage(cmd);
        return exitCode;
      }
      if (args.length == 2 && "all".equals(args[1])) {
        listAllJobs = true;
      } else {
        listJobs = true;
      }
    } else if("-kill-task".equals(cmd)) {
      if (args.length != 2) {
        displayUsage(cmd);
        return exitCode;
      }
      killTask = true;
      taskid = args[1];
    } else if("-fail-task".equals(cmd)) {
      if (args.length != 2) {
        displayUsage(cmd);
        return exitCode;
      }
      failTask = true;
      taskid = args[1];
    } else if ("-list-attempt-ids".equals(cmd)) {
      if (args.length != 3) {
        displayUsage(cmd);
        return exitCode;
      }
      jobid = args[1];
      taskState = args[2];
      displayTasks = true;
    } else if ("-logs".equals(cmd)) {
      if (args.length == 2 || args.length ==3) {
        logs = true;
        jobid = args[1];
        if (args.length == 3) {
          taskid = args[2];
        }  else {
          taskid = null;
        }
      } else {
        displayUsage(cmd);
        return exitCode;
      }
    } else {
      displayUsage(cmd);
      return exitCode;
    }

    // initialize cluster
    DragonJobServiceFactory factory = new DragonJobServiceFactory();
    DragonJobService service = factory.create(getConf());
        
    // Submit the request
    try {
      if (submitJobFile != null) {
        DragonJob job =
            DragonJob.getInstance(new DragonConfiguration(submitJobFile));
        job.submit();
        System.out.println("Created job " + job.getID());
        exitCode = 0;
      } else if (getStatus) {
        JobId jobId = jobIdForName(jobid);
        JobReport report = service.getJobReport(jobId);
        if (report == null) {
          System.out.println("Could not find job " + jobid);
        } else {
          exitCode = 0;
        }
      } else if (getCounter) {

      } else if (killJob) {
        JobId jobId = jobIdForName(jobid);
        JobReport report = service.getJobReport(jobId);
        if (report == null) {
          System.out.println("Could not find job " + jobid);
        } else {
          service.killJob(jobId);
          System.out.println("Killed job " + jobid);
          exitCode = 0;
        }
      } else if (setJobPriority) {
        JobId jobId = jobIdForName(jobid);
        JobReport report = service.getJobReport(jobId);
        if (report == null) {
          System.out.println("Could not find job " + jobid);
        } else {
          System.out.println("Changed job priority.");
          exitCode = 0;
        } 
      } else if (viewHistory) {
        exitCode = 0;
      } else if (listEvents) {
        exitCode = 0;
      } else if (listJobs) {
        exitCode = 0;
      } else if (listAllJobs) {
        exitCode = 0;
      } else if (displayTasks) {

      } else if(killTask) {

      } else if(failTask) {

      } else if (logs) {
      }
    } catch (RemoteException re) {
      IOException unwrappedException = re.unwrapRemoteException();
      if (unwrappedException instanceof AccessControlException) {
        System.out.println(unwrappedException.getMessage());
      } else {
        throw re;
      }
    } finally {
    }
    return exitCode;
  }

  /**
   * @param args
   */
  public static void main(String[] args) throws Exception {
    int res = ToolRunner.run(new DragonCLI(), args);
    System.exit(res);
  }
}