001/**
002 * Licensed to the Apache Software Foundation (ASF) under one
003 * or more contributor license agreements.  See the NOTICE file
004 * distributed with this work for additional information
005 * regarding copyright ownership.  The ASF licenses this file
006 * to you under the Apache License, Version 2.0 (the
007 * "License"); you may not use this file except in compliance
008 * with the License.  You may obtain a copy of the License at
009 *
010 *     http://www.apache.org/licenses/LICENSE-2.0
011 *
012 * Unless required by applicable law or agreed to in writing, software
013 * distributed under the License is distributed on an "AS IS" BASIS,
014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
015 * See the License for the specific language governing permissions and
016 * limitations under the License.
017 */
018package org.apache.hadoop.mapreduce.tools;
019
020import java.io.IOException;
021import java.io.OutputStreamWriter;
022import java.io.PrintWriter;
023import java.util.ArrayList;
024import java.util.List;
025import java.util.Set;
026import java.util.HashSet;
027import java.util.Arrays;
028
029import org.apache.commons.lang.StringUtils;
030import org.apache.commons.logging.Log;
031import org.apache.commons.logging.LogFactory;
032import org.apache.hadoop.classification.InterfaceAudience;
033import org.apache.hadoop.classification.InterfaceStability;
034import org.apache.hadoop.classification.InterfaceAudience.Private;
035import org.apache.hadoop.conf.Configuration;
036import org.apache.hadoop.conf.Configured;
037import org.apache.hadoop.ipc.RemoteException;
038import org.apache.hadoop.mapred.JobConf;
039import org.apache.hadoop.mapred.TIPStatus;
040import org.apache.hadoop.mapreduce.Cluster;
041import org.apache.hadoop.mapreduce.Counters;
042import org.apache.hadoop.mapreduce.Job;
043import org.apache.hadoop.mapreduce.JobID;
044import org.apache.hadoop.mapreduce.JobPriority;
045import org.apache.hadoop.mapreduce.JobStatus;
046import org.apache.hadoop.mapreduce.TaskAttemptID;
047import org.apache.hadoop.mapreduce.TaskCompletionEvent;
048import org.apache.hadoop.mapreduce.TaskReport;
049import org.apache.hadoop.mapreduce.TaskTrackerInfo;
050import org.apache.hadoop.mapreduce.TaskType;
051import org.apache.hadoop.mapreduce.jobhistory.HistoryViewer;
052import org.apache.hadoop.mapreduce.v2.LogParams;
053import org.apache.hadoop.security.AccessControlException;
054import org.apache.hadoop.util.ExitUtil;
055import org.apache.hadoop.util.Tool;
056import org.apache.hadoop.util.ToolRunner;
057import org.apache.hadoop.yarn.logaggregation.LogCLIHelpers;
058
059import com.google.common.base.Charsets;
060
061/**
062 * Interprets the map reduce cli options 
063 */
064@InterfaceAudience.Public
065@InterfaceStability.Stable
066public class CLI extends Configured implements Tool {
067  private static final Log LOG = LogFactory.getLog(CLI.class);
068  protected Cluster cluster;
069  private final Set<String> taskStates = new HashSet<String>(
070              Arrays.asList("pending", "running", "completed", "failed", "killed"));
071  private static final Set<String> taskTypes = new HashSet<String>(
072      Arrays.asList("MAP", "REDUCE"));
073  
074  public CLI() {
075  }
076  
077  public CLI(Configuration conf) {
078    setConf(conf);
079  }
080  
081  public int run(String[] argv) throws Exception {
082    int exitCode = -1;
083    if (argv.length < 1) {
084      displayUsage("");
085      return exitCode;
086    }    
087    // process arguments
088    String cmd = argv[0];
089    String submitJobFile = null;
090    String jobid = null;
091    String taskid = null;
092    String historyFile = null;
093    String counterGroupName = null;
094    String counterName = null;
095    JobPriority jp = null;
096    String taskType = null;
097    String taskState = null;
098    int fromEvent = 0;
099    int nEvents = 0;
100    boolean getStatus = false;
101    boolean getCounter = false;
102    boolean killJob = false;
103    boolean listEvents = false;
104    boolean viewHistory = false;
105    boolean viewAllHistory = false;
106    boolean listJobs = false;
107    boolean listAllJobs = false;
108    boolean listActiveTrackers = false;
109    boolean listBlacklistedTrackers = false;
110    boolean displayTasks = false;
111    boolean killTask = false;
112    boolean failTask = false;
113    boolean setJobPriority = false;
114    boolean logs = false;
115
116    if ("-submit".equals(cmd)) {
117      if (argv.length != 2) {
118        displayUsage(cmd);
119        return exitCode;
120      }
121      submitJobFile = argv[1];
122    } else if ("-status".equals(cmd)) {
123      if (argv.length != 2) {
124        displayUsage(cmd);
125        return exitCode;
126      }
127      jobid = argv[1];
128      getStatus = true;
129    } else if("-counter".equals(cmd)) {
130      if (argv.length != 4) {
131        displayUsage(cmd);
132        return exitCode;
133      }
134      getCounter = true;
135      jobid = argv[1];
136      counterGroupName = argv[2];
137      counterName = argv[3];
138    } else if ("-kill".equals(cmd)) {
139      if (argv.length != 2) {
140        displayUsage(cmd);
141        return exitCode;
142      }
143      jobid = argv[1];
144      killJob = true;
145    } else if ("-set-priority".equals(cmd)) {
146      if (argv.length != 3) {
147        displayUsage(cmd);
148        return exitCode;
149      }
150      jobid = argv[1];
151      try {
152        jp = JobPriority.valueOf(argv[2]); 
153      } catch (IllegalArgumentException iae) {
154        LOG.info(iae);
155        displayUsage(cmd);
156        return exitCode;
157      }
158      setJobPriority = true; 
159    } else if ("-events".equals(cmd)) {
160      if (argv.length != 4) {
161        displayUsage(cmd);
162        return exitCode;
163      }
164      jobid = argv[1];
165      fromEvent = Integer.parseInt(argv[2]);
166      nEvents = Integer.parseInt(argv[3]);
167      listEvents = true;
168    } else if ("-history".equals(cmd)) {
169      if (argv.length != 2 && !(argv.length == 3 && "all".equals(argv[1]))) {
170         displayUsage(cmd);
171         return exitCode;
172      }
173      viewHistory = true;
174      if (argv.length == 3 && "all".equals(argv[1])) {
175        viewAllHistory = true;
176        historyFile = argv[2];
177      } else {
178        historyFile = argv[1];
179      }
180    } else if ("-list".equals(cmd)) {
181      if (argv.length != 1 && !(argv.length == 2 && "all".equals(argv[1]))) {
182        displayUsage(cmd);
183        return exitCode;
184      }
185      if (argv.length == 2 && "all".equals(argv[1])) {
186        listAllJobs = true;
187      } else {
188        listJobs = true;
189      }
190    } else if("-kill-task".equals(cmd)) {
191      if (argv.length != 2) {
192        displayUsage(cmd);
193        return exitCode;
194      }
195      killTask = true;
196      taskid = argv[1];
197    } else if("-fail-task".equals(cmd)) {
198      if (argv.length != 2) {
199        displayUsage(cmd);
200        return exitCode;
201      }
202      failTask = true;
203      taskid = argv[1];
204    } else if ("-list-active-trackers".equals(cmd)) {
205      if (argv.length != 1) {
206        displayUsage(cmd);
207        return exitCode;
208      }
209      listActiveTrackers = true;
210    } else if ("-list-blacklisted-trackers".equals(cmd)) {
211      if (argv.length != 1) {
212        displayUsage(cmd);
213        return exitCode;
214      }
215      listBlacklistedTrackers = true;
216    } else if ("-list-attempt-ids".equals(cmd)) {
217      if (argv.length != 4) {
218        displayUsage(cmd);
219        return exitCode;
220      }
221      jobid = argv[1];
222      taskType = argv[2];
223      taskState = argv[3];
224      displayTasks = true;
225      if (!taskTypes.contains(
226          org.apache.hadoop.util.StringUtils.toUpperCase(taskType))) {
227        System.out.println("Error: Invalid task-type: " + taskType);
228        displayUsage(cmd);
229        return exitCode;
230      }
231      if (!taskStates.contains(
232          org.apache.hadoop.util.StringUtils.toLowerCase(taskState))) {
233        System.out.println("Error: Invalid task-state: " + taskState);
234        displayUsage(cmd);
235        return exitCode;
236      }
237    } else if ("-logs".equals(cmd)) {
238      if (argv.length == 2 || argv.length ==3) {
239        logs = true;
240        jobid = argv[1];
241        if (argv.length == 3) {
242          taskid = argv[2];
243        }  else {
244          taskid = null;
245        }
246      } else {
247        displayUsage(cmd);
248        return exitCode;
249      }
250    } else {
251      displayUsage(cmd);
252      return exitCode;
253    }
254
255    // initialize cluster
256    cluster = createCluster();
257        
258    // Submit the request
259    try {
260      if (submitJobFile != null) {
261        Job job = Job.getInstance(new JobConf(submitJobFile));
262        job.submit();
263        System.out.println("Created job " + job.getJobID());
264        exitCode = 0;
265      } else if (getStatus) {
266        Job job = cluster.getJob(JobID.forName(jobid));
267        if (job == null) {
268          System.out.println("Could not find job " + jobid);
269        } else {
270          Counters counters = job.getCounters();
271          System.out.println();
272          System.out.println(job);
273          if (counters != null) {
274            System.out.println(counters);
275          } else {
276            System.out.println("Counters not available. Job is retired.");
277          }
278          exitCode = 0;
279        }
280      } else if (getCounter) {
281        Job job = cluster.getJob(JobID.forName(jobid));
282        if (job == null) {
283          System.out.println("Could not find job " + jobid);
284        } else {
285          Counters counters = job.getCounters();
286          if (counters == null) {
287            System.out.println("Counters not available for retired job " + 
288            jobid);
289            exitCode = -1;
290          } else {
291            System.out.println(getCounter(counters,
292              counterGroupName, counterName));
293            exitCode = 0;
294          }
295        }
296      } else if (killJob) {
297        Job job = cluster.getJob(JobID.forName(jobid));
298        if (job == null) {
299          System.out.println("Could not find job " + jobid);
300        } else {
301          JobStatus jobStatus = job.getStatus();
302          if (jobStatus.getState() == JobStatus.State.FAILED) {
303            System.out.println("Could not mark the job " + jobid
304                + " as killed, as it has already failed.");
305            exitCode = -1;
306          } else if (jobStatus.getState() == JobStatus.State.KILLED) {
307            System.out
308                .println("The job " + jobid + " has already been killed.");
309            exitCode = -1;
310          } else if (jobStatus.getState() == JobStatus.State.SUCCEEDED) {
311            System.out.println("Could not kill the job " + jobid
312                + ", as it has already succeeded.");
313            exitCode = -1;
314          } else {
315            job.killJob();
316            System.out.println("Killed job " + jobid);
317            exitCode = 0;
318          }
319        }
320      } else if (setJobPriority) {
321        Job job = cluster.getJob(JobID.forName(jobid));
322        if (job == null) {
323          System.out.println("Could not find job " + jobid);
324        } else {
325          job.setPriority(jp);
326          System.out.println("Changed job priority.");
327          exitCode = 0;
328        } 
329      } else if (viewHistory) {
330        viewHistory(historyFile, viewAllHistory);
331        exitCode = 0;
332      } else if (listEvents) {
333        listEvents(cluster.getJob(JobID.forName(jobid)), fromEvent, nEvents);
334        exitCode = 0;
335      } else if (listJobs) {
336        listJobs(cluster);
337        exitCode = 0;
338      } else if (listAllJobs) {
339        listAllJobs(cluster);
340        exitCode = 0;
341      } else if (listActiveTrackers) {
342        listActiveTrackers(cluster);
343        exitCode = 0;
344      } else if (listBlacklistedTrackers) {
345        listBlacklistedTrackers(cluster);
346        exitCode = 0;
347      } else if (displayTasks) {
348        displayTasks(cluster.getJob(JobID.forName(jobid)), taskType, taskState);
349        exitCode = 0;
350      } else if(killTask) {
351        TaskAttemptID taskID = TaskAttemptID.forName(taskid);
352        Job job = cluster.getJob(taskID.getJobID());
353        if (job == null) {
354          System.out.println("Could not find job " + jobid);
355        } else if (job.killTask(taskID, false)) {
356          System.out.println("Killed task " + taskid);
357          exitCode = 0;
358        } else {
359          System.out.println("Could not kill task " + taskid);
360          exitCode = -1;
361        }
362      } else if(failTask) {
363        TaskAttemptID taskID = TaskAttemptID.forName(taskid);
364        Job job = cluster.getJob(taskID.getJobID());
365        if (job == null) {
366            System.out.println("Could not find job " + jobid);
367        } else if(job.killTask(taskID, true)) {
368          System.out.println("Killed task " + taskID + " by failing it");
369          exitCode = 0;
370        } else {
371          System.out.println("Could not fail task " + taskid);
372          exitCode = -1;
373        }
374      } else if (logs) {
375        try {
376        JobID jobID = JobID.forName(jobid);
377        TaskAttemptID taskAttemptID = TaskAttemptID.forName(taskid);
378        LogParams logParams = cluster.getLogParams(jobID, taskAttemptID);
379        LogCLIHelpers logDumper = new LogCLIHelpers();
380        logDumper.setConf(getConf());
381        exitCode = logDumper.dumpAContainersLogs(logParams.getApplicationId(),
382            logParams.getContainerId(), logParams.getNodeId(),
383            logParams.getOwner());
384        } catch (IOException e) {
385          if (e instanceof RemoteException) {
386            throw e;
387          } 
388          System.out.println(e.getMessage());
389        }
390      }
391    } catch (RemoteException re) {
392      IOException unwrappedException = re.unwrapRemoteException();
393      if (unwrappedException instanceof AccessControlException) {
394        System.out.println(unwrappedException.getMessage());
395      } else {
396        throw re;
397      }
398    } finally {
399      cluster.close();
400    }
401    return exitCode;
402  }
403
404  Cluster createCluster() throws IOException {
405    return new Cluster(getConf());
406  }
407  
408  private String getJobPriorityNames() {
409    StringBuffer sb = new StringBuffer();
410    for (JobPriority p : JobPriority.values()) {
411      sb.append(p.name()).append(" ");
412    }
413    return sb.substring(0, sb.length()-1);
414  }
415
416  private String getTaskTypes() {
417    return StringUtils.join(taskTypes, " ");
418  }
419  
420  /**
421   * Display usage of the command-line tool and terminate execution.
422   */
423  private void displayUsage(String cmd) {
424    String prefix = "Usage: job ";
425    String jobPriorityValues = getJobPriorityNames();
426    String taskStates = "running, completed";
427    
428    if ("-submit".equals(cmd)) {
429      System.err.println(prefix + "[" + cmd + " <job-file>]");
430    } else if ("-status".equals(cmd) || "-kill".equals(cmd)) {
431      System.err.println(prefix + "[" + cmd + " <job-id>]");
432    } else if ("-counter".equals(cmd)) {
433      System.err.println(prefix + "[" + cmd + 
434        " <job-id> <group-name> <counter-name>]");
435    } else if ("-events".equals(cmd)) {
436      System.err.println(prefix + "[" + cmd + 
437        " <job-id> <from-event-#> <#-of-events>]. Event #s start from 1.");
438    } else if ("-history".equals(cmd)) {
439      System.err.println(prefix + "[" + cmd + " <jobHistoryFile>]");
440    } else if ("-list".equals(cmd)) {
441      System.err.println(prefix + "[" + cmd + " [all]]");
442    } else if ("-kill-task".equals(cmd) || "-fail-task".equals(cmd)) {
443      System.err.println(prefix + "[" + cmd + " <task-attempt-id>]");
444    } else if ("-set-priority".equals(cmd)) {
445      System.err.println(prefix + "[" + cmd + " <job-id> <priority>]. " +
446          "Valid values for priorities are: " 
447          + jobPriorityValues); 
448    } else if ("-list-active-trackers".equals(cmd)) {
449      System.err.println(prefix + "[" + cmd + "]");
450    } else if ("-list-blacklisted-trackers".equals(cmd)) {
451      System.err.println(prefix + "[" + cmd + "]");
452    } else if ("-list-attempt-ids".equals(cmd)) {
453      System.err.println(prefix + "[" + cmd + 
454          " <job-id> <task-type> <task-state>]. " +
455          "Valid values for <task-type> are " + getTaskTypes() + ". " +
456          "Valid values for <task-state> are " + taskStates);
457    } else if ("-logs".equals(cmd)) {
458      System.err.println(prefix + "[" + cmd +
459          " <job-id> <task-attempt-id>]. " +
460          " <task-attempt-id> is optional to get task attempt logs.");      
461    } else {
462      System.err.printf(prefix + "<command> <args>%n");
463      System.err.printf("\t[-submit <job-file>]%n");
464      System.err.printf("\t[-status <job-id>]%n");
465      System.err.printf("\t[-counter <job-id> <group-name> <counter-name>]%n");
466      System.err.printf("\t[-kill <job-id>]%n");
467      System.err.printf("\t[-set-priority <job-id> <priority>]. " +
468        "Valid values for priorities are: " + jobPriorityValues + "%n");
469      System.err.printf("\t[-events <job-id> <from-event-#> <#-of-events>]%n");
470      System.err.printf("\t[-history <jobHistoryFile>]%n");
471      System.err.printf("\t[-list [all]]%n");
472      System.err.printf("\t[-list-active-trackers]%n");
473      System.err.printf("\t[-list-blacklisted-trackers]%n");
474      System.err.println("\t[-list-attempt-ids <job-id> <task-type> " +
475        "<task-state>]. " +
476        "Valid values for <task-type> are " + getTaskTypes() + ". " +
477        "Valid values for <task-state> are " + taskStates);
478      System.err.printf("\t[-kill-task <task-attempt-id>]%n");
479      System.err.printf("\t[-fail-task <task-attempt-id>]%n");
480      System.err.printf("\t[-logs <job-id> <task-attempt-id>]%n%n");
481      ToolRunner.printGenericCommandUsage(System.out);
482    }
483  }
484    
485  private void viewHistory(String historyFile, boolean all) 
486    throws IOException {
487    HistoryViewer historyViewer = new HistoryViewer(historyFile,
488                                        getConf(), all);
489    historyViewer.print();
490  }
491
492  protected long getCounter(Counters counters, String counterGroupName,
493      String counterName) throws IOException {
494    return counters.findCounter(counterGroupName, counterName).getValue();
495  }
496  
497  /**
498   * List the events for the given job
499   * @param jobId the job id for the job's events to list
500   * @throws IOException
501   */
502  private void listEvents(Job job, int fromEventId, int numEvents)
503      throws IOException, InterruptedException {
504    TaskCompletionEvent[] events = job.
505      getTaskCompletionEvents(fromEventId, numEvents);
506    System.out.println("Task completion events for " + job.getJobID());
507    System.out.println("Number of events (from " + fromEventId + ") are: " 
508      + events.length);
509    for(TaskCompletionEvent event: events) {
510      System.out.println(event.getStatus() + " " + 
511        event.getTaskAttemptId() + " " + 
512        getTaskLogURL(event.getTaskAttemptId(), event.getTaskTrackerHttp()));
513    }
514  }
515
516  protected static String getTaskLogURL(TaskAttemptID taskId, String baseUrl) {
517    return (baseUrl + "/tasklog?plaintext=true&attemptid=" + taskId); 
518  }
519  
520
521  /**
522   * Dump a list of currently running jobs
523   * @throws IOException
524   */
525  private void listJobs(Cluster cluster) 
526      throws IOException, InterruptedException {
527    List<JobStatus> runningJobs = new ArrayList<JobStatus>();
528    for (JobStatus job : cluster.getAllJobStatuses()) {
529      if (!job.isJobComplete()) {
530        runningJobs.add(job);
531      }
532    }
533    displayJobList(runningJobs.toArray(new JobStatus[0]));
534  }
535    
536  /**
537   * Dump a list of all jobs submitted.
538   * @throws IOException
539   */
540  private void listAllJobs(Cluster cluster) 
541      throws IOException, InterruptedException {
542    displayJobList(cluster.getAllJobStatuses());
543  }
544  
545  /**
546   * Display the list of active trackers
547   */
548  private void listActiveTrackers(Cluster cluster) 
549      throws IOException, InterruptedException {
550    TaskTrackerInfo[] trackers = cluster.getActiveTaskTrackers();
551    for (TaskTrackerInfo tracker : trackers) {
552      System.out.println(tracker.getTaskTrackerName());
553    }
554  }
555
556  /**
557   * Display the list of blacklisted trackers
558   */
559  private void listBlacklistedTrackers(Cluster cluster) 
560      throws IOException, InterruptedException {
561    TaskTrackerInfo[] trackers = cluster.getBlackListedTaskTrackers();
562    if (trackers.length > 0) {
563      System.out.println("BlackListedNode \t Reason");
564    }
565    for (TaskTrackerInfo tracker : trackers) {
566      System.out.println(tracker.getTaskTrackerName() + "\t" + 
567        tracker.getReasonForBlacklist());
568    }
569  }
570
571  private void printTaskAttempts(TaskReport report) {
572    if (report.getCurrentStatus() == TIPStatus.COMPLETE) {
573      System.out.println(report.getSuccessfulTaskAttemptId());
574    } else if (report.getCurrentStatus() == TIPStatus.RUNNING) {
575      for (TaskAttemptID t : 
576        report.getRunningTaskAttemptIds()) {
577        System.out.println(t);
578      }
579    }
580  }
581
582  /**
583   * Display the information about a job's tasks, of a particular type and
584   * in a particular state
585   * 
586   * @param job the job
587   * @param type the type of the task (map/reduce/setup/cleanup)
588   * @param state the state of the task 
589   * (pending/running/completed/failed/killed)
590   * @throws IOException when there is an error communicating with the master
591   * @throws InterruptedException
592   * @throws IllegalArgumentException if an invalid type/state is passed
593   */
594  protected void displayTasks(Job job, String type, String state) 
595  throws IOException, InterruptedException {
596          
597    TaskReport[] reports=null;
598    reports = job.getTaskReports(TaskType.valueOf(
599        org.apache.hadoop.util.StringUtils.toUpperCase(type)));
600    for (TaskReport report : reports) {
601      TIPStatus status = report.getCurrentStatus();
602      if ((state.equalsIgnoreCase("pending") && status ==TIPStatus.PENDING) ||
603          (state.equalsIgnoreCase("running") && status ==TIPStatus.RUNNING) ||
604          (state.equalsIgnoreCase("completed") && status == TIPStatus.COMPLETE) ||
605          (state.equalsIgnoreCase("failed") && status == TIPStatus.FAILED) ||
606          (state.equalsIgnoreCase("killed") && status == TIPStatus.KILLED)) {
607        printTaskAttempts(report);
608      }
609    }
610  }
611
612  public void displayJobList(JobStatus[] jobs) 
613      throws IOException, InterruptedException {
614    displayJobList(jobs, new PrintWriter(new OutputStreamWriter(System.out,
615        Charsets.UTF_8)));
616  }
617
618  @Private
619  public static String headerPattern = "%23s\t%20s\t%10s\t%14s\t%12s\t%12s" +
620      "\t%10s\t%15s\t%15s\t%8s\t%8s\t%10s\t%10s\n";
621  @Private
622  public static String dataPattern   = "%23s\t%20s\t%10s\t%14d\t%12s\t%12s" +
623      "\t%10s\t%15s\t%15s\t%8s\t%8s\t%10s\t%10s\n";
624  private static String memPattern   = "%dM";
625  private static String UNAVAILABLE  = "N/A";
626
627  @Private
628  public void displayJobList(JobStatus[] jobs, PrintWriter writer) {
629    writer.println("Total jobs:" + jobs.length);
630    writer.printf(headerPattern, "JobId", "JobName", "State", "StartTime",
631      "UserName", "Queue", "Priority", "UsedContainers",
632      "RsvdContainers", "UsedMem", "RsvdMem", "NeededMem", "AM info");
633    for (JobStatus job : jobs) {
634      int numUsedSlots = job.getNumUsedSlots();
635      int numReservedSlots = job.getNumReservedSlots();
636      int usedMem = job.getUsedMem();
637      int rsvdMem = job.getReservedMem();
638      int neededMem = job.getNeededMem();
639      int jobNameLength = job.getJobName().length();
640      writer.printf(dataPattern, job.getJobID().toString(),
641          job.getJobName().substring(0, jobNameLength > 20 ? 20 : jobNameLength),
642          job.getState(), job.getStartTime(), job.getUsername(),
643          job.getQueue(), job.getPriority().name(),
644          numUsedSlots < 0 ? UNAVAILABLE : numUsedSlots,
645          numReservedSlots < 0 ? UNAVAILABLE : numReservedSlots,
646          usedMem < 0 ? UNAVAILABLE : String.format(memPattern, usedMem),
647          rsvdMem < 0 ? UNAVAILABLE : String.format(memPattern, rsvdMem),
648          neededMem < 0 ? UNAVAILABLE : String.format(memPattern, neededMem),
649          job.getSchedulingInfo());
650    }
651    writer.flush();
652  }
653  
654  public static void main(String[] argv) throws Exception {
655    int res = ToolRunner.run(new CLI(), argv);
656    ExitUtil.terminate(res);
657  }
658}