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 */
018
019package org.apache.hadoop.yarn.applications.distributedshell;
020
021import java.io.IOException;
022import java.nio.ByteBuffer;
023import java.util.ArrayList;
024import java.util.HashMap;
025import java.util.List;
026import java.util.Map;
027import java.util.Vector;
028
029import org.apache.commons.cli.CommandLine;
030import org.apache.commons.cli.GnuParser;
031import org.apache.commons.cli.HelpFormatter;
032import org.apache.commons.cli.Option;
033import org.apache.commons.cli.Options;
034import org.apache.commons.cli.ParseException;
035import org.apache.commons.io.IOUtils;
036import org.apache.commons.lang.StringUtils;
037import org.apache.commons.logging.Log;
038import org.apache.commons.logging.LogFactory;
039import org.apache.hadoop.classification.InterfaceAudience;
040import org.apache.hadoop.classification.InterfaceStability;
041import org.apache.hadoop.conf.Configuration;
042import org.apache.hadoop.fs.FSDataOutputStream;
043import org.apache.hadoop.fs.FileStatus;
044import org.apache.hadoop.fs.FileSystem;
045import org.apache.hadoop.fs.Path;
046import org.apache.hadoop.fs.permission.FsPermission;
047import org.apache.hadoop.io.DataOutputBuffer;
048import org.apache.hadoop.security.Credentials;
049import org.apache.hadoop.security.UserGroupInformation;
050import org.apache.hadoop.security.token.Token;
051import org.apache.hadoop.yarn.api.ApplicationClientProtocol;
052import org.apache.hadoop.yarn.api.ApplicationConstants;
053import org.apache.hadoop.yarn.api.ApplicationConstants.Environment;
054import org.apache.hadoop.yarn.api.protocolrecords.GetNewApplicationResponse;
055import org.apache.hadoop.yarn.api.protocolrecords.KillApplicationRequest;
056import org.apache.hadoop.yarn.api.records.ApplicationId;
057import org.apache.hadoop.yarn.api.records.ApplicationReport;
058import org.apache.hadoop.yarn.api.records.ApplicationSubmissionContext;
059import org.apache.hadoop.yarn.api.records.ContainerLaunchContext;
060import org.apache.hadoop.yarn.api.records.FinalApplicationStatus;
061import org.apache.hadoop.yarn.api.records.LocalResource;
062import org.apache.hadoop.yarn.api.records.LocalResourceType;
063import org.apache.hadoop.yarn.api.records.LocalResourceVisibility;
064import org.apache.hadoop.yarn.api.records.NodeReport;
065import org.apache.hadoop.yarn.api.records.NodeState;
066import org.apache.hadoop.yarn.api.records.Priority;
067import org.apache.hadoop.yarn.api.records.QueueACL;
068import org.apache.hadoop.yarn.api.records.QueueInfo;
069import org.apache.hadoop.yarn.api.records.QueueUserACLInfo;
070import org.apache.hadoop.yarn.api.records.Resource;
071import org.apache.hadoop.yarn.api.records.YarnApplicationState;
072import org.apache.hadoop.yarn.api.records.YarnClusterMetrics;
073import org.apache.hadoop.yarn.api.records.timeline.TimelineDomain;
074import org.apache.hadoop.yarn.client.api.TimelineClient;
075import org.apache.hadoop.yarn.client.api.YarnClient;
076import org.apache.hadoop.yarn.client.api.YarnClientApplication;
077import org.apache.hadoop.yarn.conf.YarnConfiguration;
078import org.apache.hadoop.yarn.exceptions.YarnException;
079import org.apache.hadoop.yarn.util.ConverterUtils;
080import org.apache.hadoop.yarn.util.timeline.TimelineUtils;
081
082/**
083 * Client for Distributed Shell application submission to YARN.
084 * 
085 * <p> The distributed shell client allows an application master to be launched that in turn would run 
086 * the provided shell command on a set of containers. </p>
087 * 
088 * <p>This client is meant to act as an example on how to write yarn-based applications. </p>
089 * 
090 * <p> To submit an application, a client first needs to connect to the <code>ResourceManager</code> 
091 * aka ApplicationsManager or ASM via the {@link ApplicationClientProtocol}. The {@link ApplicationClientProtocol} 
092 * provides a way for the client to get access to cluster information and to request for a
093 * new {@link ApplicationId}. <p>
094 * 
095 * <p> For the actual job submission, the client first has to create an {@link ApplicationSubmissionContext}. 
096 * The {@link ApplicationSubmissionContext} defines the application details such as {@link ApplicationId} 
097 * and application name, the priority assigned to the application and the queue
098 * to which this application needs to be assigned. In addition to this, the {@link ApplicationSubmissionContext}
099 * also defines the {@link ContainerLaunchContext} which describes the <code>Container</code> with which 
100 * the {@link ApplicationMaster} is launched. </p>
101 * 
102 * <p> The {@link ContainerLaunchContext} in this scenario defines the resources to be allocated for the 
103 * {@link ApplicationMaster}'s container, the local resources (jars, configuration files) to be made available 
104 * and the environment to be set for the {@link ApplicationMaster} and the commands to be executed to run the 
105 * {@link ApplicationMaster}. <p>
106 * 
107 * <p> Using the {@link ApplicationSubmissionContext}, the client submits the application to the 
108 * <code>ResourceManager</code> and then monitors the application by requesting the <code>ResourceManager</code> 
109 * for an {@link ApplicationReport} at regular time intervals. In case of the application taking too long, the client 
110 * kills the application by submitting a {@link KillApplicationRequest} to the <code>ResourceManager</code>. </p>
111 *
112 */
113@InterfaceAudience.Public
114@InterfaceStability.Unstable
115public class Client {
116
117  private static final Log LOG = LogFactory.getLog(Client.class);
118  
119  // Configuration
120  private Configuration conf;
121  private YarnClient yarnClient;
122  // Application master specific info to register a new Application with RM/ASM
123  private String appName = "";
124  // App master priority
125  private int amPriority = 0;
126  // Queue for App master
127  private String amQueue = "";
128  // Amt. of memory resource to request for to run the App Master
129  private int amMemory = 10; 
130  // Amt. of virtual core resource to request for to run the App Master
131  private int amVCores = 1;
132
133  // Application master jar file
134  private String appMasterJar = ""; 
135  // Main class to invoke application master
136  private final String appMasterMainClass;
137
138  // Shell command to be executed 
139  private String shellCommand = ""; 
140  // Location of shell script 
141  private String shellScriptPath = ""; 
142  // Args to be passed to the shell command
143  private String[] shellArgs = new String[] {};
144  // Env variables to be setup for the shell command 
145  private Map<String, String> shellEnv = new HashMap<String, String>();
146  // Shell Command Container priority 
147  private int shellCmdPriority = 0;
148
149  // Amt of memory to request for container in which shell script will be executed
150  private int containerMemory = 10; 
151  // Amt. of virtual cores to request for container in which shell script will be executed
152  private int containerVirtualCores = 1;
153  // No. of containers in which the shell script needs to be executed
154  private int numContainers = 1;
155  private String nodeLabelExpression = null;
156
157  // log4j.properties file 
158  // if available, add to local resources and set into classpath 
159  private String log4jPropFile = "";    
160
161  // Start time for client
162  private final long clientStartTime = System.currentTimeMillis();
163  // Timeout threshold for client. Kill app after time interval expires.
164  private long clientTimeout = 600000;
165
166  // flag to indicate whether to keep containers across application attempts.
167  private boolean keepContainers = false;
168
169  private long attemptFailuresValidityInterval = -1;
170
171  // Debug flag
172  boolean debugFlag = false;
173
174  // Timeline domain ID
175  private String domainId = null;
176
177  // Flag to indicate whether to create the domain of the given ID
178  private boolean toCreateDomain = false;
179
180  // Timeline domain reader access control
181  private String viewACLs = null;
182
183  // Timeline domain writer access control
184  private String modifyACLs = null;
185
186  // Command line options
187  private Options opts;
188
189  private static final String shellCommandPath = "shellCommands";
190  private static final String shellArgsPath = "shellArgs";
191  private static final String appMasterJarPath = "AppMaster.jar";
192  // Hardcoded path to custom log_properties
193  private static final String log4jPath = "log4j.properties";
194
195  public static final String SCRIPT_PATH = "ExecScript";
196
197  /**
198   * @param args Command line arguments 
199   */
200  public static void main(String[] args) {
201    boolean result = false;
202    try {
203      Client client = new Client();
204      LOG.info("Initializing Client");
205      try {
206        boolean doRun = client.init(args);
207        if (!doRun) {
208          System.exit(0);
209        }
210      } catch (IllegalArgumentException e) {
211        System.err.println(e.getLocalizedMessage());
212        client.printUsage();
213        System.exit(-1);
214      }
215      result = client.run();
216    } catch (Throwable t) {
217      LOG.fatal("Error running Client", t);
218      System.exit(1);
219    }
220    if (result) {
221      LOG.info("Application completed successfully");
222      System.exit(0);                   
223    } 
224    LOG.error("Application failed to complete successfully");
225    System.exit(2);
226  }
227
228  /**
229   */
230  public Client(Configuration conf) throws Exception  {
231    this(
232      "org.apache.hadoop.yarn.applications.distributedshell.ApplicationMaster",
233      conf);
234  }
235
236  Client(String appMasterMainClass, Configuration conf) {
237    this.conf = conf;
238    this.appMasterMainClass = appMasterMainClass;
239    yarnClient = YarnClient.createYarnClient();
240    yarnClient.init(conf);
241    opts = new Options();
242    opts.addOption("appname", true, "Application Name. Default value - DistributedShell");
243    opts.addOption("priority", true, "Application Priority. Default 0");
244    opts.addOption("queue", true, "RM Queue in which this application is to be submitted");
245    opts.addOption("timeout", true, "Application timeout in milliseconds");
246    opts.addOption("master_memory", true, "Amount of memory in MB to be requested to run the application master");
247    opts.addOption("master_vcores", true, "Amount of virtual cores to be requested to run the application master");
248    opts.addOption("jar", true, "Jar file containing the application master");
249    opts.addOption("shell_command", true, "Shell command to be executed by " +
250        "the Application Master. Can only specify either --shell_command " +
251        "or --shell_script");
252    opts.addOption("shell_script", true, "Location of the shell script to be " +
253        "executed. Can only specify either --shell_command or --shell_script");
254    opts.addOption("shell_args", true, "Command line args for the shell script." +
255        "Multiple args can be separated by empty space.");
256    opts.getOption("shell_args").setArgs(Option.UNLIMITED_VALUES);
257    opts.addOption("shell_env", true, "Environment for shell script. Specified as env_key=env_val pairs");
258    opts.addOption("shell_cmd_priority", true, "Priority for the shell command containers");
259    opts.addOption("container_memory", true, "Amount of memory in MB to be requested to run the shell command");
260    opts.addOption("container_vcores", true, "Amount of virtual cores to be requested to run the shell command");
261    opts.addOption("num_containers", true, "No. of containers on which the shell command needs to be executed");
262    opts.addOption("log_properties", true, "log4j.properties file");
263    opts.addOption("keep_containers_across_application_attempts", false,
264      "Flag to indicate whether to keep containers across application attempts." +
265      " If the flag is true, running containers will not be killed when" +
266      " application attempt fails and these containers will be retrieved by" +
267      " the new application attempt ");
268    opts.addOption("attempt_failures_validity_interval", true,
269      "when attempt_failures_validity_interval in milliseconds is set to > 0," +
270      "the failure number will not take failures which happen out of " +
271      "the validityInterval into failure count. " +
272      "If failure count reaches to maxAppAttempts, " +
273      "the application will be failed.");
274    opts.addOption("debug", false, "Dump out debug information");
275    opts.addOption("domain", true, "ID of the timeline domain where the "
276        + "timeline entities will be put");
277    opts.addOption("view_acls", true, "Users and groups that allowed to "
278        + "view the timeline entities in the given domain");
279    opts.addOption("modify_acls", true, "Users and groups that allowed to "
280        + "modify the timeline entities in the given domain");
281    opts.addOption("create", false, "Flag to indicate whether to create the "
282        + "domain specified with -domain.");
283    opts.addOption("help", false, "Print usage");
284    opts.addOption("node_label_expression", true,
285        "Node label expression to determine the nodes"
286            + " where all the containers of this application"
287            + " will be allocated, \"\" means containers"
288            + " can be allocated anywhere, if you don't specify the option,"
289            + " default node_label_expression of queue will be used.");
290  }
291
292  /**
293   */
294  public Client() throws Exception  {
295    this(new YarnConfiguration());
296  }
297
298  /**
299   * Helper function to print out usage
300   */
301  private void printUsage() {
302    new HelpFormatter().printHelp("Client", opts);
303  }
304
305  /**
306   * Parse command line options
307   * @param args Parsed command line options 
308   * @return Whether the init was successful to run the client
309   * @throws ParseException
310   */
311  public boolean init(String[] args) throws ParseException {
312
313    CommandLine cliParser = new GnuParser().parse(opts, args);
314
315    if (args.length == 0) {
316      throw new IllegalArgumentException("No args specified for client to initialize");
317    }
318
319    if (cliParser.hasOption("log_properties")) {
320      String log4jPath = cliParser.getOptionValue("log_properties");
321      try {
322        Log4jPropertyHelper.updateLog4jConfiguration(Client.class, log4jPath);
323      } catch (Exception e) {
324        LOG.warn("Can not set up custom log4j properties. " + e);
325      }
326    }
327
328    if (cliParser.hasOption("help")) {
329      printUsage();
330      return false;
331    }
332
333    if (cliParser.hasOption("debug")) {
334      debugFlag = true;
335
336    }
337
338    if (cliParser.hasOption("keep_containers_across_application_attempts")) {
339      LOG.info("keep_containers_across_application_attempts");
340      keepContainers = true;
341    }
342
343    appName = cliParser.getOptionValue("appname", "DistributedShell");
344    amPriority = Integer.parseInt(cliParser.getOptionValue("priority", "0"));
345    amQueue = cliParser.getOptionValue("queue", "default");
346    amMemory = Integer.parseInt(cliParser.getOptionValue("master_memory", "10"));               
347    amVCores = Integer.parseInt(cliParser.getOptionValue("master_vcores", "1"));
348
349    if (amMemory < 0) {
350      throw new IllegalArgumentException("Invalid memory specified for application master, exiting."
351          + " Specified memory=" + amMemory);
352    }
353    if (amVCores < 0) {
354      throw new IllegalArgumentException("Invalid virtual cores specified for application master, exiting."
355          + " Specified virtual cores=" + amVCores);
356    }
357
358    if (!cliParser.hasOption("jar")) {
359      throw new IllegalArgumentException("No jar file specified for application master");
360    }           
361
362    appMasterJar = cliParser.getOptionValue("jar");
363
364    if (!cliParser.hasOption("shell_command") && !cliParser.hasOption("shell_script")) {
365      throw new IllegalArgumentException(
366          "No shell command or shell script specified to be executed by application master");
367    } else if (cliParser.hasOption("shell_command") && cliParser.hasOption("shell_script")) {
368      throw new IllegalArgumentException("Can not specify shell_command option " +
369          "and shell_script option at the same time");
370    } else if (cliParser.hasOption("shell_command")) {
371      shellCommand = cliParser.getOptionValue("shell_command");
372    } else {
373      shellScriptPath = cliParser.getOptionValue("shell_script");
374    }
375    if (cliParser.hasOption("shell_args")) {
376      shellArgs = cliParser.getOptionValues("shell_args");
377    }
378    if (cliParser.hasOption("shell_env")) { 
379      String envs[] = cliParser.getOptionValues("shell_env");
380      for (String env : envs) {
381        env = env.trim();
382        int index = env.indexOf('=');
383        if (index == -1) {
384          shellEnv.put(env, "");
385          continue;
386        }
387        String key = env.substring(0, index);
388        String val = "";
389        if (index < (env.length()-1)) {
390          val = env.substring(index+1);
391        }
392        shellEnv.put(key, val);
393      }
394    }
395    shellCmdPriority = Integer.parseInt(cliParser.getOptionValue("shell_cmd_priority", "0"));
396
397    containerMemory = Integer.parseInt(cliParser.getOptionValue("container_memory", "10"));
398    containerVirtualCores = Integer.parseInt(cliParser.getOptionValue("container_vcores", "1"));
399    numContainers = Integer.parseInt(cliParser.getOptionValue("num_containers", "1"));
400    
401
402    if (containerMemory < 0 || containerVirtualCores < 0 || numContainers < 1) {
403      throw new IllegalArgumentException("Invalid no. of containers or container memory/vcores specified,"
404          + " exiting."
405          + " Specified containerMemory=" + containerMemory
406          + ", containerVirtualCores=" + containerVirtualCores
407          + ", numContainer=" + numContainers);
408    }
409    
410    nodeLabelExpression = cliParser.getOptionValue("node_label_expression", null);
411
412    clientTimeout = Integer.parseInt(cliParser.getOptionValue("timeout", "600000"));
413
414    attemptFailuresValidityInterval =
415        Long.parseLong(cliParser.getOptionValue(
416          "attempt_failures_validity_interval", "-1"));
417
418    log4jPropFile = cliParser.getOptionValue("log_properties", "");
419
420    // Get timeline domain options
421    if (cliParser.hasOption("domain")) {
422      domainId = cliParser.getOptionValue("domain");
423      toCreateDomain = cliParser.hasOption("create");
424      if (cliParser.hasOption("view_acls")) {
425        viewACLs = cliParser.getOptionValue("view_acls");
426      }
427      if (cliParser.hasOption("modify_acls")) {
428        modifyACLs = cliParser.getOptionValue("modify_acls");
429      }
430    }
431
432    return true;
433  }
434
435  /**
436   * Main run function for the client
437   * @return true if application completed successfully
438   * @throws IOException
439   * @throws YarnException
440   */
441  public boolean run() throws IOException, YarnException {
442
443    LOG.info("Running Client");
444    yarnClient.start();
445
446    YarnClusterMetrics clusterMetrics = yarnClient.getYarnClusterMetrics();
447    LOG.info("Got Cluster metric info from ASM" 
448        + ", numNodeManagers=" + clusterMetrics.getNumNodeManagers());
449
450    List<NodeReport> clusterNodeReports = yarnClient.getNodeReports(
451        NodeState.RUNNING);
452    LOG.info("Got Cluster node info from ASM");
453    for (NodeReport node : clusterNodeReports) {
454      LOG.info("Got node report from ASM for"
455          + ", nodeId=" + node.getNodeId() 
456          + ", nodeAddress" + node.getHttpAddress()
457          + ", nodeRackName" + node.getRackName()
458          + ", nodeNumContainers" + node.getNumContainers());
459    }
460
461    QueueInfo queueInfo = yarnClient.getQueueInfo(this.amQueue);
462    LOG.info("Queue info"
463        + ", queueName=" + queueInfo.getQueueName()
464        + ", queueCurrentCapacity=" + queueInfo.getCurrentCapacity()
465        + ", queueMaxCapacity=" + queueInfo.getMaximumCapacity()
466        + ", queueApplicationCount=" + queueInfo.getApplications().size()
467        + ", queueChildQueueCount=" + queueInfo.getChildQueues().size());               
468
469    List<QueueUserACLInfo> listAclInfo = yarnClient.getQueueAclsInfo();
470    for (QueueUserACLInfo aclInfo : listAclInfo) {
471      for (QueueACL userAcl : aclInfo.getUserAcls()) {
472        LOG.info("User ACL Info for Queue"
473            + ", queueName=" + aclInfo.getQueueName()                   
474            + ", userAcl=" + userAcl.name());
475      }
476    }           
477
478    if (domainId != null && domainId.length() > 0 && toCreateDomain) {
479      prepareTimelineDomain();
480    }
481
482    // Get a new application id
483    YarnClientApplication app = yarnClient.createApplication();
484    GetNewApplicationResponse appResponse = app.getNewApplicationResponse();
485    // TODO get min/max resource capabilities from RM and change memory ask if needed
486    // If we do not have min/max, we may not be able to correctly request 
487    // the required resources from the RM for the app master
488    // Memory ask has to be a multiple of min and less than max. 
489    // Dump out information about cluster capability as seen by the resource manager
490    int maxMem = appResponse.getMaximumResourceCapability().getMemory();
491    LOG.info("Max mem capabililty of resources in this cluster " + maxMem);
492
493    // A resource ask cannot exceed the max. 
494    if (amMemory > maxMem) {
495      LOG.info("AM memory specified above max threshold of cluster. Using max value."
496          + ", specified=" + amMemory
497          + ", max=" + maxMem);
498      amMemory = maxMem;
499    }                           
500
501    int maxVCores = appResponse.getMaximumResourceCapability().getVirtualCores();
502    LOG.info("Max virtual cores capabililty of resources in this cluster " + maxVCores);
503    
504    if (amVCores > maxVCores) {
505      LOG.info("AM virtual cores specified above max threshold of cluster. " 
506          + "Using max value." + ", specified=" + amVCores 
507          + ", max=" + maxVCores);
508      amVCores = maxVCores;
509    }
510    
511    // set the application name
512    ApplicationSubmissionContext appContext = app.getApplicationSubmissionContext();
513    ApplicationId appId = appContext.getApplicationId();
514
515    appContext.setKeepContainersAcrossApplicationAttempts(keepContainers);
516    appContext.setApplicationName(appName);
517
518    if (attemptFailuresValidityInterval >= 0) {
519      appContext
520        .setAttemptFailuresValidityInterval(attemptFailuresValidityInterval);
521    }
522
523    // set local resources for the application master
524    // local files or archives as needed
525    // In this scenario, the jar file for the application master is part of the local resources                 
526    Map<String, LocalResource> localResources = new HashMap<String, LocalResource>();
527
528    LOG.info("Copy App Master jar from local filesystem and add to local environment");
529    // Copy the application master jar to the filesystem 
530    // Create a local resource to point to the destination jar path 
531    FileSystem fs = FileSystem.get(conf);
532    addToLocalResources(fs, appMasterJar, appMasterJarPath, appId.toString(),
533        localResources, null);
534
535    // Set the log4j properties if needed 
536    if (!log4jPropFile.isEmpty()) {
537      addToLocalResources(fs, log4jPropFile, log4jPath, appId.toString(),
538          localResources, null);
539    }                   
540
541    // The shell script has to be made available on the final container(s)
542    // where it will be executed. 
543    // To do this, we need to first copy into the filesystem that is visible 
544    // to the yarn framework. 
545    // We do not need to set this as a local resource for the application 
546    // master as the application master does not need it.               
547    String hdfsShellScriptLocation = ""; 
548    long hdfsShellScriptLen = 0;
549    long hdfsShellScriptTimestamp = 0;
550    if (!shellScriptPath.isEmpty()) {
551      Path shellSrc = new Path(shellScriptPath);
552      String shellPathSuffix =
553          appName + "/" + appId.toString() + "/" + SCRIPT_PATH;
554      Path shellDst =
555          new Path(fs.getHomeDirectory(), shellPathSuffix);
556      fs.copyFromLocalFile(false, true, shellSrc, shellDst);
557      hdfsShellScriptLocation = shellDst.toUri().toString(); 
558      FileStatus shellFileStatus = fs.getFileStatus(shellDst);
559      hdfsShellScriptLen = shellFileStatus.getLen();
560      hdfsShellScriptTimestamp = shellFileStatus.getModificationTime();
561    }
562
563    if (!shellCommand.isEmpty()) {
564      addToLocalResources(fs, null, shellCommandPath, appId.toString(),
565          localResources, shellCommand);
566    }
567
568    if (shellArgs.length > 0) {
569      addToLocalResources(fs, null, shellArgsPath, appId.toString(),
570          localResources, StringUtils.join(shellArgs, " "));
571    }
572
573    // Set the necessary security tokens as needed
574    //amContainer.setContainerTokens(containerToken);
575
576    // Set the env variables to be setup in the env where the application master will be run
577    LOG.info("Set the environment for the application master");
578    Map<String, String> env = new HashMap<String, String>();
579
580    // put location of shell script into env
581    // using the env info, the application master will create the correct local resource for the 
582    // eventual containers that will be launched to execute the shell scripts
583    env.put(DSConstants.DISTRIBUTEDSHELLSCRIPTLOCATION, hdfsShellScriptLocation);
584    env.put(DSConstants.DISTRIBUTEDSHELLSCRIPTTIMESTAMP, Long.toString(hdfsShellScriptTimestamp));
585    env.put(DSConstants.DISTRIBUTEDSHELLSCRIPTLEN, Long.toString(hdfsShellScriptLen));
586    if (domainId != null && domainId.length() > 0) {
587      env.put(DSConstants.DISTRIBUTEDSHELLTIMELINEDOMAIN, domainId);
588    }
589
590    // Add AppMaster.jar location to classpath          
591    // At some point we should not be required to add 
592    // the hadoop specific classpaths to the env. 
593    // It should be provided out of the box. 
594    // For now setting all required classpaths including
595    // the classpath to "." for the application jar
596    StringBuilder classPathEnv = new StringBuilder(Environment.CLASSPATH.$$())
597      .append(ApplicationConstants.CLASS_PATH_SEPARATOR).append("./*");
598    for (String c : conf.getStrings(
599        YarnConfiguration.YARN_APPLICATION_CLASSPATH,
600        YarnConfiguration.DEFAULT_YARN_CROSS_PLATFORM_APPLICATION_CLASSPATH)) {
601      classPathEnv.append(ApplicationConstants.CLASS_PATH_SEPARATOR);
602      classPathEnv.append(c.trim());
603    }
604    classPathEnv.append(ApplicationConstants.CLASS_PATH_SEPARATOR).append(
605      "./log4j.properties");
606
607    // add the runtime classpath needed for tests to work
608    if (conf.getBoolean(YarnConfiguration.IS_MINI_YARN_CLUSTER, false)) {
609      classPathEnv.append(':');
610      classPathEnv.append(System.getProperty("java.class.path"));
611    }
612
613    env.put("CLASSPATH", classPathEnv.toString());
614
615    // Set the necessary command to execute the application master 
616    Vector<CharSequence> vargs = new Vector<CharSequence>(30);
617
618    // Set java executable command 
619    LOG.info("Setting up app master command");
620    vargs.add(Environment.JAVA_HOME.$$() + "/bin/java");
621    // Set Xmx based on am memory size
622    vargs.add("-Xmx" + amMemory + "m");
623    // Set class name 
624    vargs.add(appMasterMainClass);
625    // Set params for Application Master
626    vargs.add("--container_memory " + String.valueOf(containerMemory));
627    vargs.add("--container_vcores " + String.valueOf(containerVirtualCores));
628    vargs.add("--num_containers " + String.valueOf(numContainers));
629    if (null != nodeLabelExpression) {
630      appContext.setNodeLabelExpression(nodeLabelExpression);
631    }
632    vargs.add("--priority " + String.valueOf(shellCmdPriority));
633
634    for (Map.Entry<String, String> entry : shellEnv.entrySet()) {
635      vargs.add("--shell_env " + entry.getKey() + "=" + entry.getValue());
636    }                   
637    if (debugFlag) {
638      vargs.add("--debug");
639    }
640
641    vargs.add("1>" + ApplicationConstants.LOG_DIR_EXPANSION_VAR + "/AppMaster.stdout");
642    vargs.add("2>" + ApplicationConstants.LOG_DIR_EXPANSION_VAR + "/AppMaster.stderr");
643
644    // Get final commmand
645    StringBuilder command = new StringBuilder();
646    for (CharSequence str : vargs) {
647      command.append(str).append(" ");
648    }
649
650    LOG.info("Completed setting up app master command " + command.toString());     
651    List<String> commands = new ArrayList<String>();
652    commands.add(command.toString());           
653
654    // Set up the container launch context for the application master
655    ContainerLaunchContext amContainer = ContainerLaunchContext.newInstance(
656      localResources, env, commands, null, null, null);
657
658    // Set up resource type requirements
659    // For now, both memory and vcores are supported, so we set memory and 
660    // vcores requirements
661    Resource capability = Resource.newInstance(amMemory, amVCores);
662    appContext.setResource(capability);
663
664    // Service data is a binary blob that can be passed to the application
665    // Not needed in this scenario
666    // amContainer.setServiceData(serviceData);
667
668    // Setup security tokens
669    if (UserGroupInformation.isSecurityEnabled()) {
670      // Note: Credentials class is marked as LimitedPrivate for HDFS and MapReduce
671      Credentials credentials = new Credentials();
672      String tokenRenewer = conf.get(YarnConfiguration.RM_PRINCIPAL);
673      if (tokenRenewer == null || tokenRenewer.length() == 0) {
674        throw new IOException(
675          "Can't get Master Kerberos principal for the RM to use as renewer");
676      }
677
678      // For now, only getting tokens for the default file-system.
679      final Token<?> tokens[] =
680          fs.addDelegationTokens(tokenRenewer, credentials);
681      if (tokens != null) {
682        for (Token<?> token : tokens) {
683          LOG.info("Got dt for " + fs.getUri() + "; " + token);
684        }
685      }
686      DataOutputBuffer dob = new DataOutputBuffer();
687      credentials.writeTokenStorageToStream(dob);
688      ByteBuffer fsTokens = ByteBuffer.wrap(dob.getData(), 0, dob.getLength());
689      amContainer.setTokens(fsTokens);
690    }
691
692    appContext.setAMContainerSpec(amContainer);
693
694    // Set the priority for the application master
695    // TODO - what is the range for priority? how to decide? 
696    Priority pri = Priority.newInstance(amPriority);
697    appContext.setPriority(pri);
698
699    // Set the queue to which this application is to be submitted in the RM
700    appContext.setQueue(amQueue);
701
702    // Submit the application to the applications manager
703    // SubmitApplicationResponse submitResp = applicationsManager.submitApplication(appRequest);
704    // Ignore the response as either a valid response object is returned on success 
705    // or an exception thrown to denote some form of a failure
706    LOG.info("Submitting application to ASM");
707
708    yarnClient.submitApplication(appContext);
709
710    // TODO
711    // Try submitting the same request again
712    // app submission failure?
713
714    // Monitor the application
715    return monitorApplication(appId);
716
717  }
718
719  /**
720   * Monitor the submitted application for completion. 
721   * Kill application if time expires. 
722   * @param appId Application Id of application to be monitored
723   * @return true if application completed successfully
724   * @throws YarnException
725   * @throws IOException
726   */
727  private boolean monitorApplication(ApplicationId appId)
728      throws YarnException, IOException {
729
730    while (true) {
731
732      // Check app status every 1 second.
733      try {
734        Thread.sleep(1000);
735      } catch (InterruptedException e) {
736        LOG.debug("Thread sleep in monitoring loop interrupted");
737      }
738
739      // Get application report for the appId we are interested in 
740      ApplicationReport report = yarnClient.getApplicationReport(appId);
741
742      LOG.info("Got application report from ASM for"
743          + ", appId=" + appId.getId()
744          + ", clientToAMToken=" + report.getClientToAMToken()
745          + ", appDiagnostics=" + report.getDiagnostics()
746          + ", appMasterHost=" + report.getHost()
747          + ", appQueue=" + report.getQueue()
748          + ", appMasterRpcPort=" + report.getRpcPort()
749          + ", appStartTime=" + report.getStartTime()
750          + ", yarnAppState=" + report.getYarnApplicationState().toString()
751          + ", distributedFinalState=" + report.getFinalApplicationStatus().toString()
752          + ", appTrackingUrl=" + report.getTrackingUrl()
753          + ", appUser=" + report.getUser());
754
755      YarnApplicationState state = report.getYarnApplicationState();
756      FinalApplicationStatus dsStatus = report.getFinalApplicationStatus();
757      if (YarnApplicationState.FINISHED == state) {
758        if (FinalApplicationStatus.SUCCEEDED == dsStatus) {
759          LOG.info("Application has completed successfully. Breaking monitoring loop");
760          return true;        
761        }
762        else {
763          LOG.info("Application did finished unsuccessfully."
764              + " YarnState=" + state.toString() + ", DSFinalStatus=" + dsStatus.toString()
765              + ". Breaking monitoring loop");
766          return false;
767        }                         
768      }
769      else if (YarnApplicationState.KILLED == state     
770          || YarnApplicationState.FAILED == state) {
771        LOG.info("Application did not finish."
772            + " YarnState=" + state.toString() + ", DSFinalStatus=" + dsStatus.toString()
773            + ". Breaking monitoring loop");
774        return false;
775      }                 
776
777      if (System.currentTimeMillis() > (clientStartTime + clientTimeout)) {
778        LOG.info("Reached client specified timeout for application. Killing application");
779        forceKillApplication(appId);
780        return false;                           
781      }
782    }                   
783
784  }
785
786  /**
787   * Kill a submitted application by sending a call to the ASM
788   * @param appId Application Id to be killed. 
789   * @throws YarnException
790   * @throws IOException
791   */
792  private void forceKillApplication(ApplicationId appId)
793      throws YarnException, IOException {
794    // TODO clarify whether multiple jobs with the same app id can be submitted and be running at 
795    // the same time. 
796    // If yes, can we kill a particular attempt only?
797
798    // Response can be ignored as it is non-null on success or 
799    // throws an exception in case of failures
800    yarnClient.killApplication(appId);  
801  }
802
803  private void addToLocalResources(FileSystem fs, String fileSrcPath,
804      String fileDstPath, String appId, Map<String, LocalResource> localResources,
805      String resources) throws IOException {
806    String suffix =
807        appName + "/" + appId + "/" + fileDstPath;
808    Path dst =
809        new Path(fs.getHomeDirectory(), suffix);
810    if (fileSrcPath == null) {
811      FSDataOutputStream ostream = null;
812      try {
813        ostream = FileSystem
814            .create(fs, dst, new FsPermission((short) 0710));
815        ostream.writeUTF(resources);
816      } finally {
817        IOUtils.closeQuietly(ostream);
818      }
819    } else {
820      fs.copyFromLocalFile(new Path(fileSrcPath), dst);
821    }
822    FileStatus scFileStatus = fs.getFileStatus(dst);
823    LocalResource scRsrc =
824        LocalResource.newInstance(
825            ConverterUtils.getYarnUrlFromURI(dst.toUri()),
826            LocalResourceType.FILE, LocalResourceVisibility.APPLICATION,
827            scFileStatus.getLen(), scFileStatus.getModificationTime());
828    localResources.put(fileDstPath, scRsrc);
829  }
830
831  private void prepareTimelineDomain() {
832    TimelineClient timelineClient = null;
833    if (conf.getBoolean(YarnConfiguration.TIMELINE_SERVICE_ENABLED,
834        YarnConfiguration.DEFAULT_TIMELINE_SERVICE_ENABLED)) {
835      timelineClient = TimelineClient.createTimelineClient();
836      timelineClient.init(conf);
837      timelineClient.start();
838    } else {
839      LOG.warn("Cannot put the domain " + domainId +
840          " because the timeline service is not enabled");
841      return;
842    }
843    try {
844      //TODO: we need to check and combine the existing timeline domain ACLs,
845      //but let's do it once we have client java library to query domains.
846      TimelineDomain domain = new TimelineDomain();
847      domain.setId(domainId);
848      domain.setReaders(
849          viewACLs != null && viewACLs.length() > 0 ? viewACLs : " ");
850      domain.setWriters(
851          modifyACLs != null && modifyACLs.length() > 0 ? modifyACLs : " ");
852      timelineClient.putDomain(domain);
853      LOG.info("Put the timeline domain: " +
854          TimelineUtils.dumpTimelineRecordtoJSON(domain));
855    } catch (Exception e) {
856      LOG.error("Error when putting the timeline domain", e);
857    } finally {
858      timelineClient.stop();
859    }
860  }
861}