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    
019    package org.apache.hadoop.yarn.applications.distributedshell;
020    
021    import java.io.File;
022    import java.io.IOException;
023    import java.nio.ByteBuffer;
024    import java.util.ArrayList;
025    import java.util.HashMap;
026    import java.util.List;
027    import java.util.Map;
028    import java.util.Vector;
029    
030    import org.apache.commons.cli.CommandLine;
031    import org.apache.commons.cli.GnuParser;
032    import org.apache.commons.cli.HelpFormatter;
033    import org.apache.commons.cli.Option;
034    import org.apache.commons.cli.Options;
035    import org.apache.commons.cli.ParseException;
036    import org.apache.commons.io.IOUtils;
037    import org.apache.commons.lang.StringUtils;
038    import org.apache.commons.logging.Log;
039    import org.apache.commons.logging.LogFactory;
040    import org.apache.hadoop.classification.InterfaceAudience;
041    import org.apache.hadoop.classification.InterfaceStability;
042    import org.apache.hadoop.conf.Configuration;
043    import org.apache.hadoop.fs.FSDataOutputStream;
044    import org.apache.hadoop.fs.FileStatus;
045    import org.apache.hadoop.fs.FileSystem;
046    import org.apache.hadoop.fs.Path;
047    import org.apache.hadoop.fs.permission.FsPermission;
048    import org.apache.hadoop.io.DataOutputBuffer;
049    import org.apache.hadoop.security.Credentials;
050    import org.apache.hadoop.security.UserGroupInformation;
051    import org.apache.hadoop.security.token.Token;
052    import org.apache.hadoop.util.Shell;
053    import org.apache.hadoop.yarn.api.ApplicationClientProtocol;
054    import org.apache.hadoop.yarn.api.ApplicationConstants;
055    import org.apache.hadoop.yarn.api.ApplicationConstants.Environment;
056    import org.apache.hadoop.yarn.api.protocolrecords.GetNewApplicationResponse;
057    import org.apache.hadoop.yarn.api.protocolrecords.KillApplicationRequest;
058    import org.apache.hadoop.yarn.api.records.ApplicationId;
059    import org.apache.hadoop.yarn.api.records.ApplicationReport;
060    import org.apache.hadoop.yarn.api.records.ApplicationSubmissionContext;
061    import org.apache.hadoop.yarn.api.records.ContainerLaunchContext;
062    import org.apache.hadoop.yarn.api.records.FinalApplicationStatus;
063    import org.apache.hadoop.yarn.api.records.LocalResource;
064    import org.apache.hadoop.yarn.api.records.LocalResourceType;
065    import org.apache.hadoop.yarn.api.records.LocalResourceVisibility;
066    import org.apache.hadoop.yarn.api.records.NodeReport;
067    import org.apache.hadoop.yarn.api.records.NodeState;
068    import org.apache.hadoop.yarn.api.records.Priority;
069    import org.apache.hadoop.yarn.api.records.QueueACL;
070    import org.apache.hadoop.yarn.api.records.QueueInfo;
071    import org.apache.hadoop.yarn.api.records.QueueUserACLInfo;
072    import org.apache.hadoop.yarn.api.records.Resource;
073    import org.apache.hadoop.yarn.api.records.YarnApplicationState;
074    import org.apache.hadoop.yarn.api.records.YarnClusterMetrics;
075    import org.apache.hadoop.yarn.client.api.YarnClient;
076    import org.apache.hadoop.yarn.client.api.YarnClientApplication;
077    import org.apache.hadoop.yarn.conf.YarnConfiguration;
078    import org.apache.hadoop.yarn.exceptions.YarnException;
079    import org.apache.hadoop.yarn.util.ConverterUtils;
080    import org.apache.hadoop.yarn.util.Records;
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
115    public 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    
156      // log4j.properties file 
157      // if available, add to local resources and set into classpath 
158      private String log4jPropFile = "";    
159    
160      // Start time for client
161      private final long clientStartTime = System.currentTimeMillis();
162      // Timeout threshold for client. Kill app after time interval expires.
163      private long clientTimeout = 600000;
164    
165      // Debug flag
166      boolean debugFlag = false;    
167    
168      // Command line options
169      private Options opts;
170    
171      private static final String shellCommandPath = "shellCommands";
172      private static final String shellArgsPath = "shellArgs";
173      private static final String appMasterJarPath = "AppMaster.jar";
174      // Hardcoded path to custom log_properties
175      private static final String log4jPath = "log4j.properties";
176    
177      private static final String linuxShellPath = "ExecShellScript.sh";
178      private static final String windowBatPath = "ExecBatScript.bat";
179    
180      /**
181       * @param args Command line arguments 
182       */
183      public static void main(String[] args) {
184        boolean result = false;
185        try {
186          Client client = new Client();
187          LOG.info("Initializing Client");
188          try {
189            boolean doRun = client.init(args);
190            if (!doRun) {
191              System.exit(0);
192            }
193          } catch (IllegalArgumentException e) {
194            System.err.println(e.getLocalizedMessage());
195            client.printUsage();
196            System.exit(-1);
197          }
198          result = client.run();
199        } catch (Throwable t) {
200          LOG.fatal("Error running CLient", t);
201          System.exit(1);
202        }
203        if (result) {
204          LOG.info("Application completed successfully");
205          System.exit(0);                   
206        } 
207        LOG.error("Application failed to complete successfully");
208        System.exit(2);
209      }
210    
211      /**
212       */
213      public Client(Configuration conf) throws Exception  {
214        this(
215          "org.apache.hadoop.yarn.applications.distributedshell.ApplicationMaster",
216          conf);
217      }
218    
219      Client(String appMasterMainClass, Configuration conf) {
220        this.conf = conf;
221        this.appMasterMainClass = appMasterMainClass;
222        yarnClient = YarnClient.createYarnClient();
223        yarnClient.init(conf);
224        opts = new Options();
225        opts.addOption("appname", true, "Application Name. Default value - DistributedShell");
226        opts.addOption("priority", true, "Application Priority. Default 0");
227        opts.addOption("queue", true, "RM Queue in which this application is to be submitted");
228        opts.addOption("timeout", true, "Application timeout in milliseconds");
229        opts.addOption("master_memory", true, "Amount of memory in MB to be requested to run the application master");
230        opts.addOption("master_vcores", true, "Amount of virtual cores to be requested to run the application master");
231        opts.addOption("jar", true, "Jar file containing the application master");
232        opts.addOption("shell_command", true, "Shell command to be executed by " +
233            "the Application Master. Can only specify either --shell_command " +
234            "or --shell_script");
235        opts.addOption("shell_script", true, "Location of the shell script to be " +
236            "executed. Can only specify either --shell_command or --shell_script");
237        opts.addOption("shell_args", true, "Command line args for the shell script." +
238            "Multiple args can be separated by empty space.");
239        opts.getOption("shell_args").setArgs(Option.UNLIMITED_VALUES);
240        opts.addOption("shell_env", true, "Environment for shell script. Specified as env_key=env_val pairs");
241        opts.addOption("shell_cmd_priority", true, "Priority for the shell command containers");
242        opts.addOption("container_memory", true, "Amount of memory in MB to be requested to run the shell command");
243        opts.addOption("container_vcores", true, "Amount of virtual cores to be requested to run the shell command");
244        opts.addOption("num_containers", true, "No. of containers on which the shell command needs to be executed");
245        opts.addOption("log_properties", true, "log4j.properties file");
246        opts.addOption("debug", false, "Dump out debug information");
247        opts.addOption("help", false, "Print usage");
248    
249      }
250    
251      /**
252       */
253      public Client() throws Exception  {
254        this(new YarnConfiguration());
255      }
256    
257      /**
258       * Helper function to print out usage
259       */
260      private void printUsage() {
261        new HelpFormatter().printHelp("Client", opts);
262      }
263    
264      /**
265       * Parse command line options
266       * @param args Parsed command line options 
267       * @return Whether the init was successful to run the client
268       * @throws ParseException
269       */
270      public boolean init(String[] args) throws ParseException {
271    
272        CommandLine cliParser = new GnuParser().parse(opts, args);
273    
274        if (args.length == 0) {
275          throw new IllegalArgumentException("No args specified for client to initialize");
276        }
277    
278        if (cliParser.hasOption("log_properties")) {
279          String log4jPath = cliParser.getOptionValue("log_properties");
280          try {
281            Log4jPropertyHelper.updateLog4jConfiguration(Client.class, log4jPath);
282          } catch (Exception e) {
283            LOG.warn("Can not set up custom log4j properties. " + e);
284          }
285        }
286    
287        if (cliParser.hasOption("help")) {
288          printUsage();
289          return false;
290        }
291    
292        if (cliParser.hasOption("debug")) {
293          debugFlag = true;
294    
295        }
296    
297        appName = cliParser.getOptionValue("appname", "DistributedShell");
298        amPriority = Integer.parseInt(cliParser.getOptionValue("priority", "0"));
299        amQueue = cliParser.getOptionValue("queue", "default");
300        amMemory = Integer.parseInt(cliParser.getOptionValue("master_memory", "10"));               
301        amVCores = Integer.parseInt(cliParser.getOptionValue("master_vcores", "1"));
302        
303        if (amMemory < 0) {
304          throw new IllegalArgumentException("Invalid memory specified for application master, exiting."
305              + " Specified memory=" + amMemory);
306        }
307        if (amVCores < 0) {
308          throw new IllegalArgumentException("Invalid virtual cores specified for application master, exiting."
309              + " Specified virtual cores=" + amVCores);
310        }
311    
312        if (!cliParser.hasOption("jar")) {
313          throw new IllegalArgumentException("No jar file specified for application master");
314        }           
315    
316        appMasterJar = cliParser.getOptionValue("jar");
317    
318        if (!cliParser.hasOption("shell_command") && !cliParser.hasOption("shell_script")) {
319          throw new IllegalArgumentException(
320              "No shell command or shell script specified to be executed by application master");
321        } else if (cliParser.hasOption("shell_command") && cliParser.hasOption("shell_script")) {
322          throw new IllegalArgumentException("Can not specify shell_command option " +
323              "and shell_script option at the same time");
324        } else if (cliParser.hasOption("shell_command")) {
325          shellCommand = cliParser.getOptionValue("shell_command");
326        } else {
327          shellScriptPath = cliParser.getOptionValue("shell_script");
328        }
329        if (cliParser.hasOption("shell_args")) {
330          shellArgs = cliParser.getOptionValues("shell_args");
331        }
332        if (cliParser.hasOption("shell_env")) { 
333          String envs[] = cliParser.getOptionValues("shell_env");
334          for (String env : envs) {
335            env = env.trim();
336            int index = env.indexOf('=');
337            if (index == -1) {
338              shellEnv.put(env, "");
339              continue;
340            }
341            String key = env.substring(0, index);
342            String val = "";
343            if (index < (env.length()-1)) {
344              val = env.substring(index+1);
345            }
346            shellEnv.put(key, val);
347          }
348        }
349        shellCmdPriority = Integer.parseInt(cliParser.getOptionValue("shell_cmd_priority", "0"));
350    
351        containerMemory = Integer.parseInt(cliParser.getOptionValue("container_memory", "10"));
352        containerVirtualCores = Integer.parseInt(cliParser.getOptionValue("container_vcores", "1"));
353        numContainers = Integer.parseInt(cliParser.getOptionValue("num_containers", "1"));
354    
355        if (containerMemory < 0 || containerVirtualCores < 0 || numContainers < 1) {
356          throw new IllegalArgumentException("Invalid no. of containers or container memory/vcores specified,"
357              + " exiting."
358              + " Specified containerMemory=" + containerMemory
359              + ", containerVirtualCores=" + containerVirtualCores
360              + ", numContainer=" + numContainers);
361        }
362    
363        clientTimeout = Integer.parseInt(cliParser.getOptionValue("timeout", "600000"));
364    
365        log4jPropFile = cliParser.getOptionValue("log_properties", "");
366    
367        return true;
368      }
369    
370      /**
371       * Main run function for the client
372       * @return true if application completed successfully
373       * @throws IOException
374       * @throws YarnException
375       */
376      public boolean run() throws IOException, YarnException {
377    
378        LOG.info("Running Client");
379        yarnClient.start();
380    
381        YarnClusterMetrics clusterMetrics = yarnClient.getYarnClusterMetrics();
382        LOG.info("Got Cluster metric info from ASM" 
383            + ", numNodeManagers=" + clusterMetrics.getNumNodeManagers());
384    
385        List<NodeReport> clusterNodeReports = yarnClient.getNodeReports(
386            NodeState.RUNNING);
387        LOG.info("Got Cluster node info from ASM");
388        for (NodeReport node : clusterNodeReports) {
389          LOG.info("Got node report from ASM for"
390              + ", nodeId=" + node.getNodeId() 
391              + ", nodeAddress" + node.getHttpAddress()
392              + ", nodeRackName" + node.getRackName()
393              + ", nodeNumContainers" + node.getNumContainers());
394        }
395    
396        QueueInfo queueInfo = yarnClient.getQueueInfo(this.amQueue);
397        LOG.info("Queue info"
398            + ", queueName=" + queueInfo.getQueueName()
399            + ", queueCurrentCapacity=" + queueInfo.getCurrentCapacity()
400            + ", queueMaxCapacity=" + queueInfo.getMaximumCapacity()
401            + ", queueApplicationCount=" + queueInfo.getApplications().size()
402            + ", queueChildQueueCount=" + queueInfo.getChildQueues().size());               
403    
404        List<QueueUserACLInfo> listAclInfo = yarnClient.getQueueAclsInfo();
405        for (QueueUserACLInfo aclInfo : listAclInfo) {
406          for (QueueACL userAcl : aclInfo.getUserAcls()) {
407            LOG.info("User ACL Info for Queue"
408                + ", queueName=" + aclInfo.getQueueName()                   
409                + ", userAcl=" + userAcl.name());
410          }
411        }           
412    
413        // Get a new application id
414        YarnClientApplication app = yarnClient.createApplication();
415        GetNewApplicationResponse appResponse = app.getNewApplicationResponse();
416        // TODO get min/max resource capabilities from RM and change memory ask if needed
417        // If we do not have min/max, we may not be able to correctly request 
418        // the required resources from the RM for the app master
419        // Memory ask has to be a multiple of min and less than max. 
420        // Dump out information about cluster capability as seen by the resource manager
421        int maxMem = appResponse.getMaximumResourceCapability().getMemory();
422        LOG.info("Max mem capabililty of resources in this cluster " + maxMem);
423    
424        // A resource ask cannot exceed the max. 
425        if (amMemory > maxMem) {
426          LOG.info("AM memory specified above max threshold of cluster. Using max value."
427              + ", specified=" + amMemory
428              + ", max=" + maxMem);
429          amMemory = maxMem;
430        }                           
431    
432        int maxVCores = appResponse.getMaximumResourceCapability().getVirtualCores();
433        LOG.info("Max virtual cores capabililty of resources in this cluster " + maxVCores);
434        
435        if (amVCores > maxVCores) {
436          LOG.info("AM virtual cores specified above max threshold of cluster. " 
437              + "Using max value." + ", specified=" + amVCores 
438              + ", max=" + maxVCores);
439          amVCores = maxVCores;
440        }
441        
442        // set the application name
443        ApplicationSubmissionContext appContext = app.getApplicationSubmissionContext();
444        ApplicationId appId = appContext.getApplicationId();
445        appContext.setApplicationName(appName);
446    
447        // Set up the container launch context for the application master
448        ContainerLaunchContext amContainer = Records.newRecord(ContainerLaunchContext.class);
449    
450        // set local resources for the application master
451        // local files or archives as needed
452        // In this scenario, the jar file for the application master is part of the local resources                 
453        Map<String, LocalResource> localResources = new HashMap<String, LocalResource>();
454    
455        LOG.info("Copy App Master jar from local filesystem and add to local environment");
456        // Copy the application master jar to the filesystem 
457        // Create a local resource to point to the destination jar path 
458        FileSystem fs = FileSystem.get(conf);
459        addToLocalResources(fs, appMasterJar, appMasterJarPath, appId.getId(),
460            localResources, null);
461    
462        // Set the log4j properties if needed 
463        if (!log4jPropFile.isEmpty()) {
464          addToLocalResources(fs, log4jPropFile, log4jPath, appId.getId(),
465              localResources, null);
466        }                   
467    
468        // The shell script has to be made available on the final container(s)
469        // where it will be executed. 
470        // To do this, we need to first copy into the filesystem that is visible 
471        // to the yarn framework. 
472        // We do not need to set this as a local resource for the application 
473        // master as the application master does not need it.               
474        String hdfsShellScriptLocation = ""; 
475        long hdfsShellScriptLen = 0;
476        long hdfsShellScriptTimestamp = 0;
477        if (!shellScriptPath.isEmpty()) {
478          Path shellSrc = new Path(shellScriptPath);
479          String shellPathSuffix =
480              appName + "/" + appId.getId() + "/"
481                  + (Shell.WINDOWS ? windowBatPath : linuxShellPath);
482          Path shellDst =
483              new Path(fs.getHomeDirectory(), shellPathSuffix);
484          fs.copyFromLocalFile(false, true, shellSrc, shellDst);
485          hdfsShellScriptLocation = shellDst.toUri().toString(); 
486          FileStatus shellFileStatus = fs.getFileStatus(shellDst);
487          hdfsShellScriptLen = shellFileStatus.getLen();
488          hdfsShellScriptTimestamp = shellFileStatus.getModificationTime();
489        }
490    
491        if (!shellCommand.isEmpty()) {
492          addToLocalResources(fs, null, shellCommandPath, appId.getId(),
493              localResources, shellCommand);
494        }
495    
496        if (shellArgs.length > 0) {
497          addToLocalResources(fs, null, shellArgsPath, appId.getId(),
498              localResources, StringUtils.join(shellArgs, " "));
499        }
500        // Set local resource info into app master container launch context
501        amContainer.setLocalResources(localResources);
502    
503        // Set the necessary security tokens as needed
504        //amContainer.setContainerTokens(containerToken);
505    
506        // Set the env variables to be setup in the env where the application master will be run
507        LOG.info("Set the environment for the application master");
508        Map<String, String> env = new HashMap<String, String>();
509    
510        // put location of shell script into env
511        // using the env info, the application master will create the correct local resource for the 
512        // eventual containers that will be launched to execute the shell scripts
513        env.put(DSConstants.DISTRIBUTEDSHELLSCRIPTLOCATION, hdfsShellScriptLocation);
514        env.put(DSConstants.DISTRIBUTEDSHELLSCRIPTTIMESTAMP, Long.toString(hdfsShellScriptTimestamp));
515        env.put(DSConstants.DISTRIBUTEDSHELLSCRIPTLEN, Long.toString(hdfsShellScriptLen));
516    
517        // Add AppMaster.jar location to classpath          
518        // At some point we should not be required to add 
519        // the hadoop specific classpaths to the env. 
520        // It should be provided out of the box. 
521        // For now setting all required classpaths including
522        // the classpath to "." for the application jar
523        StringBuilder classPathEnv = new StringBuilder(Environment.CLASSPATH.$())
524          .append(File.pathSeparatorChar).append("./*");
525        for (String c : conf.getStrings(
526            YarnConfiguration.YARN_APPLICATION_CLASSPATH,
527            YarnConfiguration.DEFAULT_YARN_APPLICATION_CLASSPATH)) {
528          classPathEnv.append(File.pathSeparatorChar);
529          classPathEnv.append(c.trim());
530        }
531        classPathEnv.append(File.pathSeparatorChar).append("./log4j.properties");
532    
533        // add the runtime classpath needed for tests to work
534        if (conf.getBoolean(YarnConfiguration.IS_MINI_YARN_CLUSTER, false)) {
535          classPathEnv.append(':');
536          classPathEnv.append(System.getProperty("java.class.path"));
537        }
538    
539        env.put("CLASSPATH", classPathEnv.toString());
540    
541        amContainer.setEnvironment(env);
542    
543        // Set the necessary command to execute the application master 
544        Vector<CharSequence> vargs = new Vector<CharSequence>(30);
545    
546        // Set java executable command 
547        LOG.info("Setting up app master command");
548        vargs.add(Environment.JAVA_HOME.$() + "/bin/java");
549        // Set Xmx based on am memory size
550        vargs.add("-Xmx" + amMemory + "m");
551        // Set class name 
552        vargs.add(appMasterMainClass);
553        // Set params for Application Master
554        vargs.add("--container_memory " + String.valueOf(containerMemory));
555        vargs.add("--container_vcores " + String.valueOf(containerVirtualCores));
556        vargs.add("--num_containers " + String.valueOf(numContainers));
557        vargs.add("--priority " + String.valueOf(shellCmdPriority));
558    
559        for (Map.Entry<String, String> entry : shellEnv.entrySet()) {
560          vargs.add("--shell_env " + entry.getKey() + "=" + entry.getValue());
561        }                   
562        if (debugFlag) {
563          vargs.add("--debug");
564        }
565    
566        vargs.add("1>" + ApplicationConstants.LOG_DIR_EXPANSION_VAR + "/AppMaster.stdout");
567        vargs.add("2>" + ApplicationConstants.LOG_DIR_EXPANSION_VAR + "/AppMaster.stderr");
568    
569        // Get final commmand
570        StringBuilder command = new StringBuilder();
571        for (CharSequence str : vargs) {
572          command.append(str).append(" ");
573        }
574    
575        LOG.info("Completed setting up app master command " + command.toString());     
576        List<String> commands = new ArrayList<String>();
577        commands.add(command.toString());           
578        amContainer.setCommands(commands);
579    
580        // Set up resource type requirements
581        // For now, both memory and vcores are supported, so we set memory and 
582        // vcores requirements
583        Resource capability = Records.newRecord(Resource.class);
584        capability.setMemory(amMemory);
585        capability.setVirtualCores(amVCores);
586        appContext.setResource(capability);
587    
588        // Service data is a binary blob that can be passed to the application
589        // Not needed in this scenario
590        // amContainer.setServiceData(serviceData);
591    
592        // Setup security tokens
593        if (UserGroupInformation.isSecurityEnabled()) {
594          Credentials credentials = new Credentials();
595          String tokenRenewer = conf.get(YarnConfiguration.RM_PRINCIPAL);
596          if (tokenRenewer == null || tokenRenewer.length() == 0) {
597            throw new IOException(
598              "Can't get Master Kerberos principal for the RM to use as renewer");
599          }
600    
601          // For now, only getting tokens for the default file-system.
602          final Token<?> tokens[] =
603              fs.addDelegationTokens(tokenRenewer, credentials);
604          if (tokens != null) {
605            for (Token<?> token : tokens) {
606              LOG.info("Got dt for " + fs.getUri() + "; " + token);
607            }
608          }
609          DataOutputBuffer dob = new DataOutputBuffer();
610          credentials.writeTokenStorageToStream(dob);
611          ByteBuffer fsTokens = ByteBuffer.wrap(dob.getData(), 0, dob.getLength());
612          amContainer.setTokens(fsTokens);
613        }
614    
615        appContext.setAMContainerSpec(amContainer);
616    
617        // Set the priority for the application master
618        Priority pri = Records.newRecord(Priority.class);
619        // TODO - what is the range for priority? how to decide? 
620        pri.setPriority(amPriority);
621        appContext.setPriority(pri);
622    
623        // Set the queue to which this application is to be submitted in the RM
624        appContext.setQueue(amQueue);
625    
626        // Submit the application to the applications manager
627        // SubmitApplicationResponse submitResp = applicationsManager.submitApplication(appRequest);
628        // Ignore the response as either a valid response object is returned on success 
629        // or an exception thrown to denote some form of a failure
630        LOG.info("Submitting application to ASM");
631    
632        yarnClient.submitApplication(appContext);
633    
634        // TODO
635        // Try submitting the same request again
636        // app submission failure?
637    
638        // Monitor the application
639        return monitorApplication(appId);
640    
641      }
642    
643      /**
644       * Monitor the submitted application for completion. 
645       * Kill application if time expires. 
646       * @param appId Application Id of application to be monitored
647       * @return true if application completed successfully
648       * @throws YarnException
649       * @throws IOException
650       */
651      private boolean monitorApplication(ApplicationId appId)
652          throws YarnException, IOException {
653    
654        while (true) {
655    
656          // Check app status every 1 second.
657          try {
658            Thread.sleep(1000);
659          } catch (InterruptedException e) {
660            LOG.debug("Thread sleep in monitoring loop interrupted");
661          }
662    
663          // Get application report for the appId we are interested in 
664          ApplicationReport report = yarnClient.getApplicationReport(appId);
665    
666          LOG.info("Got application report from ASM for"
667              + ", appId=" + appId.getId()
668              + ", clientToAMToken=" + report.getClientToAMToken()
669              + ", appDiagnostics=" + report.getDiagnostics()
670              + ", appMasterHost=" + report.getHost()
671              + ", appQueue=" + report.getQueue()
672              + ", appMasterRpcPort=" + report.getRpcPort()
673              + ", appStartTime=" + report.getStartTime()
674              + ", yarnAppState=" + report.getYarnApplicationState().toString()
675              + ", distributedFinalState=" + report.getFinalApplicationStatus().toString()
676              + ", appTrackingUrl=" + report.getTrackingUrl()
677              + ", appUser=" + report.getUser());
678    
679          YarnApplicationState state = report.getYarnApplicationState();
680          FinalApplicationStatus dsStatus = report.getFinalApplicationStatus();
681          if (YarnApplicationState.FINISHED == state) {
682            if (FinalApplicationStatus.SUCCEEDED == dsStatus) {
683              LOG.info("Application has completed successfully. Breaking monitoring loop");
684              return true;        
685            }
686            else {
687              LOG.info("Application did finished unsuccessfully."
688                  + " YarnState=" + state.toString() + ", DSFinalStatus=" + dsStatus.toString()
689                  + ". Breaking monitoring loop");
690              return false;
691            }                         
692          }
693          else if (YarnApplicationState.KILLED == state     
694              || YarnApplicationState.FAILED == state) {
695            LOG.info("Application did not finish."
696                + " YarnState=" + state.toString() + ", DSFinalStatus=" + dsStatus.toString()
697                + ". Breaking monitoring loop");
698            return false;
699          }                 
700    
701          if (System.currentTimeMillis() > (clientStartTime + clientTimeout)) {
702            LOG.info("Reached client specified timeout for application. Killing application");
703            forceKillApplication(appId);
704            return false;                           
705          }
706        }                   
707    
708      }
709    
710      /**
711       * Kill a submitted application by sending a call to the ASM
712       * @param appId Application Id to be killed. 
713       * @throws YarnException
714       * @throws IOException
715       */
716      private void forceKillApplication(ApplicationId appId)
717          throws YarnException, IOException {
718        // TODO clarify whether multiple jobs with the same app id can be submitted and be running at 
719        // the same time. 
720        // If yes, can we kill a particular attempt only?
721    
722        // Response can be ignored as it is non-null on success or 
723        // throws an exception in case of failures
724        yarnClient.killApplication(appId);  
725      }
726    
727      private void addToLocalResources(FileSystem fs, String fileSrcPath,
728          String fileDstPath, int appId, Map<String, LocalResource> localResources,
729          String resources) throws IOException {
730        String suffix =
731            appName + "/" + appId + "/" + fileDstPath;
732        Path dst =
733            new Path(fs.getHomeDirectory(), suffix);
734        if (fileSrcPath == null) {
735          FSDataOutputStream ostream = null;
736          try {
737            ostream = FileSystem
738                .create(fs, dst, new FsPermission((short) 0710));
739            ostream.writeUTF(resources);
740          } finally {
741            IOUtils.closeQuietly(ostream);
742          }
743        } else {
744          fs.copyFromLocalFile(new Path(fileSrcPath), dst);
745        }
746        FileStatus scFileStatus = fs.getFileStatus(dst);
747        LocalResource scRsrc =
748            LocalResource.newInstance(
749                ConverterUtils.getYarnUrlFromURI(dst.toUri()),
750                LocalResourceType.FILE, LocalResourceVisibility.APPLICATION,
751                scFileStatus.getLen(), scFileStatus.getModificationTime());
752        localResources.put(fileDstPath, scRsrc);
753      }
754    }