Java 类org.apache.hadoop.util.Shell.ShellCommandExecutor 实例源码

/**
 * Change the permissions on a file / directory, recursively, if
 * needed.
 * @param filename name of the file whose permissions are to change
 * @param perm permission string
 * @param recursive true, if permissions should be changed recursively
 * @return the exit code from the command.
 * @throws IOException
 */
public static int chmod(String filename, String perm, boolean recursive)
                          throws IOException {
  String [] cmd = Shell.getSetPermissionCommand(perm, recursive);
  String[] args = new String[cmd.length + 1];
  System.arraycopy(cmd, 0, args, 0, cmd.length);
  args[cmd.length] = new File(filename).getPath();
  ShellCommandExecutor shExec = new ShellCommandExecutor(args);
  try {
    shExec.execute();
  }catch(IOException e) {
    if(LOG.isDebugEnabled()) {
      LOG.debug("Error while changing permission : " + filename
                +" Exception: " + StringUtils.stringifyException(e));
    }
  }
  return shExec.getExitCode();
}

/**
 * Create a ShellCommandExecutor object which returns exit code 1,
 * emulating the case that the user does not exist.
 *
 * @param userName not used
 * @return a mock ShellCommandExecutor object
 */
@Override
protected ShellCommandExecutor createGroupExecutor(String userName) {
  ShellCommandExecutor executor = mock(ShellCommandExecutor.class);

  try {
    doThrow(new ExitCodeException(1,
        "id: foobarusernotexist: No such user")).
        when(executor).execute();

    when(executor.getOutput()).thenReturn("");
  } catch (IOException e) {
    LOG.warn(e.getMessage());
  }
  return executor;
}

/**
 * Create a ShellCommandExecutor object which returns partially resolved
 * group names for a user.
 *
 * @param userName not used
 * @return a mock ShellCommandExecutor object
 */
@Override
protected ShellCommandExecutor createGroupExecutor(String userName) {
  ShellCommandExecutor executor = mock(ShellCommandExecutor.class);

  try {
    // There is both a group name 9999 and a group ID 9999.
    // This is treated as unresolvable group.
    doThrow(new ExitCodeException(1, "cannot find name for group ID 9999")).
        when(executor).execute();

    when(executor.getOutput()).thenReturn("9999\n9999 abc def");
  } catch (IOException e) {
    LOG.warn(e.getMessage());
  }
  return executor;
}

public static boolean isAvailable() {
  if (Shell.WINDOWS) {
    ShellCommandExecutor shellExecutor = new ShellCommandExecutor(
        new String[] { Shell.WINUTILS, "help" });
    try {
      shellExecutor.execute();
    } catch (IOException e) {
      LOG.error(StringUtils.stringifyException(e));
    } finally {
      String output = shellExecutor.getOutput();
      if (output != null &&
          output.contains("Prints to stdout a list of processes in the task")) {
        return true;
      }
    }
  }
  return false;
}

public void run() {
  try {
    Vector<String> args = new Vector<String>();
    if (isSetsidAvailable()) {
      args.add("setsid");
    }
    args.add("bash");
    args.add("-c");
    args.add(" echo $$ > " + pidFile + "; sh " + shellScript + " " + N
        + ";");
    shexec = new ShellCommandExecutor(args.toArray(new String[0]));
    shexec.execute();
  } catch (ExitCodeException ee) {
    LOG.info("Shell Command exit with a non-zero exit code. This is"
        + " expected as we are killing the subprocesses of the"
        + " task intentionally. " + ee);
  } catch (IOException ioe) {
    LOG.info("Error executing shell command " + ioe);
  } finally {
    LOG.info("Exit code: " + shexec.getExitCode());
  }
}

/**
 * Is perl supported on this machine ?
 * @return true if perl is available and is working as expected
 */
public static boolean hasPerlSupport() {
  boolean hasPerl = false;
  ShellCommandExecutor shexec = new ShellCommandExecutor(
    new String[] { "perl", "-e", "print 42" });
  try {
    shexec.execute();
    if (shexec.getOutput().equals("42")) {
      hasPerl = true;
    }
    else {
      LOG.warn("Perl is installed, but isn't behaving as expected.");
    }
  } catch (Exception e) {
    LOG.warn("Could not run perl: " + e);
  }
  return hasPerl;
}

/**
 * Change the permissions on a file / directory, recursively, if
 * needed.
 * @param filename name of the file whose permissions are to change
 * @param perm permission string
 * @param recursive true, if permissions should be changed recursively
 * @return the exit code from the command.
 * @throws IOException
 */
public static int chmod(String filename, String perm, boolean recursive)
                          throws IOException {
  String [] cmd = Shell.getSetPermissionCommand(perm, recursive);
  String[] args = new String[cmd.length + 1];
  System.arraycopy(cmd, 0, args, 0, cmd.length);
  args[cmd.length] = new File(filename).getPath();
  ShellCommandExecutor shExec = new ShellCommandExecutor(args);
  try {
    shExec.execute();
  }catch(IOException e) {
    if(LOG.isDebugEnabled()) {
      LOG.debug("Error while changing permission : " + filename 
                +" Exception: " + StringUtils.stringifyException(e));
    }
  }
  return shExec.getExitCode();
}

public static boolean isAvailable() {
  if (Shell.WINDOWS) {
    if (!Shell.hasWinutilsPath()) {
      return false;
    }
    ShellCommandExecutor shellExecutor = new ShellCommandExecutor(
        new String[] { Shell.getWinUtilsPath(), "help" });
    try {
      shellExecutor.execute();
    } catch (IOException e) {
      LOG.error(StringUtils.stringifyException(e));
    } finally {
      String output = shellExecutor.getOutput();
      if (output != null &&
          output.contains("Prints to stdout a list of processes in the task")) {
        return true;
      }
    }
  }
  return false;
}

public void run() {
  try {
    Vector<String> args = new Vector<String>();
    if (isSetsidAvailable()) {
      args.add("setsid");
    }
    args.add("bash");
    args.add("-c");
    args.add(" echo $$ > " + pidFile + "; sh " + shellScript + " " + N
        + ";");
    shexec = new ShellCommandExecutor(args.toArray(new String[0]));
    shexec.execute();
  } catch (ExitCodeException ee) {
    LOG.info("Shell Command exit with a non-zero exit code. This is"
        + " expected as we are killing the subprocesses of the"
        + " task intentionally. " + ee);
  } catch (IOException ioe) {
    LOG.info("Error executing shell command " + ioe);
  } finally {
    LOG.info("Exit code: " + shexec.getExitCode());
  }
}

/**
 * Is perl supported on this machine ?
 * @return true if perl is available and is working as expected
 */
public static boolean hasPerlSupport() {
  boolean hasPerl = false;
  ShellCommandExecutor shexec = new ShellCommandExecutor(
    new String[] { "perl", "-e", "print 42" });
  try {
    shexec.execute();
    if (shexec.getOutput().equals("42")) {
      hasPerl = true;
    }
    else {
      LOG.warn("Perl is installed, but isn't behaving as expected.");
    }
  } catch (Exception e) {
    LOG.warn("Could not run perl: " + e);
  }
  return hasPerl;
}

/**
 * Change the permissions on a file / directory, recursively, if
 * needed.
 * @param filename name of the file whose permissions are to change
 * @param perm permission string
 * @param recursive true, if permissions should be changed recursively
 * @return the exit code from the command.
 * @throws IOException
 */
public static int chmod(String filename, String perm, boolean recursive)
                          throws IOException {
  String [] cmd = Shell.getSetPermissionCommand(perm, recursive);
  String[] args = new String[cmd.length + 1];
  System.arraycopy(cmd, 0, args, 0, cmd.length);
  args[cmd.length] = new File(filename).getPath();
  ShellCommandExecutor shExec = new ShellCommandExecutor(args);
  try {
    shExec.execute();
  }catch(IOException e) {
    if(LOG.isDebugEnabled()) {
      LOG.debug("Error while changing permission : " + filename
                +" Exception: " + StringUtils.stringifyException(e));
    }
  }
  return shExec.getExitCode();
}

/**
 * Create a ShellCommandExecutor object which returns exit code 1,
 * emulating the case that the user does not exist.
 *
 * @param userName not used
 * @return a mock ShellCommandExecutor object
 */
@Override
protected ShellCommandExecutor createGroupExecutor(String userName) {
  ShellCommandExecutor executor = mock(ShellCommandExecutor.class);

  try {
    doThrow(new ExitCodeException(1,
        "id: foobarusernotexist: No such user")).
        when(executor).execute();

    when(executor.getOutput()).thenReturn("");
  } catch (IOException e) {
    LOG.warn(e.getMessage());
  }
  return executor;
}

/**
 * Create a ShellCommandExecutor object which returns partially resolved
 * group names for a user.
 *
 * @param userName not used
 * @return a mock ShellCommandExecutor object
 */
@Override
protected ShellCommandExecutor createGroupExecutor(String userName) {
  ShellCommandExecutor executor = mock(ShellCommandExecutor.class);

  try {
    // There is both a group name 9999 and a group ID 9999.
    // This is treated as unresolvable group.
    doThrow(new ExitCodeException(1, "cannot find name for group ID 9999")).
        when(executor).execute();

    when(executor.getOutput()).thenReturn("9999\n9999 abc def");
  } catch (IOException e) {
    LOG.warn(e.getMessage());
  }
  return executor;
}

public static boolean isAvailable() {
  if (Shell.WINDOWS) {
    ShellCommandExecutor shellExecutor = new ShellCommandExecutor(
        new String[] { Shell.WINUTILS, "help" });
    try {
      shellExecutor.execute();
    } catch (IOException e) {
      LOG.error(StringUtils.stringifyException(e));
    } finally {
      String output = shellExecutor.getOutput();
      if (output != null &&
          output.contains("Prints to stdout a list of processes in the task")) {
        return true;
      }
    }
  }
  return false;
}

public void run() {
  try {
    Vector<String> args = new Vector<String>();
    if (isSetsidAvailable()) {
      args.add("setsid");
    }
    args.add("bash");
    args.add("-c");
    args.add(" echo $$ > " + pidFile + "; sh " + shellScript + " " + N
        + ";");
    shexec = new ShellCommandExecutor(args.toArray(new String[0]));
    shexec.execute();
  } catch (ExitCodeException ee) {
    LOG.info("Shell Command exit with a non-zero exit code. This is"
        + " expected as we are killing the subprocesses of the"
        + " task intentionally. " + ee);
  } catch (IOException ioe) {
    LOG.info("Error executing shell command " + ioe);
  } finally {
    LOG.info("Exit code: " + shexec.getExitCode());
  }
}

@Override 
public void init() throws IOException {        
  // Send command to executor which will just start up, 
  // verify configuration/permissions and exit
  List<String> command = new ArrayList<String>(
      Arrays.asList(containerExecutorExe,
          "--checksetup"));
  String[] commandArray = command.toArray(new String[command.size()]);
  ShellCommandExecutor shExec = new ShellCommandExecutor(commandArray);
  if (LOG.isDebugEnabled()) {
    LOG.debug("checkLinuxExecutorSetup: " + Arrays.toString(commandArray));
  }
  try {
    shExec.execute();
  } catch (ExitCodeException e) {
    int exitCode = shExec.getExitCode();
    LOG.warn("Exit code from container executor initialization is : "
        + exitCode, e);
    logOutput(shExec.getOutput());
    throw new IOException("Linux container executor not configured properly"
        + " (error=" + exitCode + ")", e);
  }

  resourcesHandler.init(this);
}

public void mountCgroups(List<String> cgroupKVs, String hierarchy)
       throws IOException {
  List<String> command = new ArrayList<String>(
          Arrays.asList(containerExecutorExe, "--mount-cgroups", hierarchy));
  command.addAll(cgroupKVs);

  String[] commandArray = command.toArray(new String[command.size()]);
  ShellCommandExecutor shExec = new ShellCommandExecutor(commandArray);

  if (LOG.isDebugEnabled()) {
      LOG.debug("mountCgroups: " + Arrays.toString(commandArray));
  }

  try {
      shExec.execute();
  } catch (IOException e) {
      int ret_code = shExec.getExitCode();
      LOG.warn("Exception in LinuxContainerExecutor mountCgroups ", e);
      logOutput(shExec.getOutput());
      throw new IOException("Problem mounting cgroups " + cgroupKVs + 
        "; exit code = " + ret_code + " and output: " + shExec.getOutput(), e);
  }
}

/**
 * Is perl supported on this machine ?
 * @return true if perl is available and is working as expected
 */
public static boolean hasPerlSupport() {
  boolean hasPerl = false;
  ShellCommandExecutor shexec = new ShellCommandExecutor(
    new String[] { "perl", "-e", "print 42" });
  try {
    shexec.execute();
    if (shexec.getOutput().equals("42")) {
      hasPerl = true;
    }
    else {
      LOG.warn("Perl is installed, but isn't behaving as expected.");
    }
  } catch (Exception e) {
    LOG.warn("Could not run perl: " + e);
  }
  return hasPerl;
}

/**
 * Change the permissions on a file / directory, recursively, if
 * needed.
 * @param filename name of the file whose permissions are to change
 * @param perm permission string
 * @param recursive true, if permissions should be changed recursively
 * @return the exit code from the command.
 * @throws IOException
 */
public static int chmod(String filename, String perm, boolean recursive)
                          throws IOException {
  String [] cmd = Shell.getSetPermissionCommand(perm, recursive);
  String[] args = new String[cmd.length + 1];
  System.arraycopy(cmd, 0, args, 0, cmd.length);
  args[cmd.length] = new File(filename).getPath();
  ShellCommandExecutor shExec = new ShellCommandExecutor(args);
  try {
    shExec.execute();
  }catch(IOException e) {
    if(LOG.isDebugEnabled()) {
      LOG.debug("Error while changing permission : " + filename 
                +" Exception: " + StringUtils.stringifyException(e));
    }
  }
  return shExec.getExitCode();
}

public static boolean isAvailable() {
  if (Shell.WINDOWS) {
    ShellCommandExecutor shellExecutor = new ShellCommandExecutor(
        new String[] { Shell.WINUTILS, "help" });
    try {
      shellExecutor.execute();
    } catch (IOException e) {
      LOG.error(StringUtils.stringifyException(e));
    } finally {
      String output = shellExecutor.getOutput();
      if (output != null &&
          output.contains("Prints to stdout a list of processes in the task")) {
        return true;
      }
    }
  }
  return false;
}

public void run() {
  try {
    Vector<String> args = new Vector<String>();
    if (isSetsidAvailable()) {
      args.add("setsid");
    }
    args.add("bash");
    args.add("-c");
    args.add(" echo $$ > " + pidFile + "; sh " + shellScript + " " + N
        + ";");
    shexec = new ShellCommandExecutor(args.toArray(new String[0]));
    shexec.execute();
  } catch (ExitCodeException ee) {
    LOG.info("Shell Command exit with a non-zero exit code. This is"
        + " expected as we are killing the subprocesses of the"
        + " task intentionally. " + ee);
  } catch (IOException ioe) {
    LOG.info("Error executing shell command " + ioe);
  } finally {
    LOG.info("Exit code: " + shexec.getExitCode());
  }
}

/**
 * Create a ShellCommandExecutor object which returns exit code 1,
 * emulating the case that the user does not exist.
 *
 * @param userName not used
 * @return a mock ShellCommandExecutor object
 */
@Override
protected ShellCommandExecutor createGroupExecutor(String userName) {
  ShellCommandExecutor executor = mock(ShellCommandExecutor.class);

  try {
    doThrow(new ExitCodeException(1,
        "id: foobarusernotexist: No such user")).
        when(executor).execute();

    when(executor.getOutput()).thenReturn("");
  } catch (IOException e) {
    LOG.warn(e.getMessage());
  }
  return executor;
}

public void mountCgroups(List<String> cgroupKVs, String hierarchy)
       throws IOException {
  List<String> command = new ArrayList<String>(
          Arrays.asList(containerExecutorExe, "--mount-cgroups", hierarchy));
  command.addAll(cgroupKVs);

  String[] commandArray = command.toArray(new String[command.size()]);
  ShellCommandExecutor shExec = new ShellCommandExecutor(commandArray);

  if (LOG.isDebugEnabled()) {
      LOG.debug("mountCgroups: " + Arrays.toString(commandArray));
  }

  try {
      shExec.execute();
  } catch (IOException e) {
      int ret_code = shExec.getExitCode();
      LOG.warn("Exception in LinuxContainerExecutor mountCgroups ", e);
      logOutput(shExec.getOutput());
      throw new IOException("Problem mounting cgroups " + cgroupKVs + 
        "; exit code = " + ret_code + " and output: " + shExec.getOutput(), e);
  }
}

/**
 * Is perl supported on this machine ?
 * @return true if perl is available and is working as expected
 */
public static boolean hasPerlSupport() {
  boolean hasPerl = false;
  ShellCommandExecutor shexec = new ShellCommandExecutor(
    new String[] { "perl", "-e", "print 42" });
  try {
    shexec.execute();
    if (shexec.getOutput().equals("42")) {
      hasPerl = true;
    }
    else {
      LOG.warn("Perl is installed, but isn't behaving as expected.");
    }
  } catch (Exception e) {
    LOG.warn("Could not run perl: " + e);
  }
  return hasPerl;
}

/**
 * Creates a hardlink 
 * @param file - existing source file
 * @param linkName - desired target link file
 */
public static void createHardLink(File file, File linkName) 
throws IOException {
  if (file == null) {
    throw new IOException(
        "invalid arguments to createHardLink: source file is null");
  }
  if (linkName == null) {
    throw new IOException(
        "invalid arguments to createHardLink: link name is null");
  }
 // construct and execute shell command
  String[] hardLinkCommand = getHardLinkCommand.linkOne(file, linkName);
  ShellCommandExecutor shexec = new ShellCommandExecutor(hardLinkCommand);
  try {
    shexec.execute();
  } catch (ExitCodeException e) {
    throw new IOException("Failed to execute command " +
        Arrays.toString(hardLinkCommand) +
        "; command output: \"" + shexec.getOutput() + "\"" +
        "; WrappedException: \"" + e.getMessage() + "\"");
  }
}

/**
 * Change the permissions on a file / directory, recursively, if
 * needed.
 * @param filename name of the file whose permissions are to change
 * @param perm permission string
 * @param recursive true, if permissions should be changed recursively
 * @return the exit code from the command.
 * @throws IOException
 */
public static int chmod(String filename, String perm, boolean recursive)
                          throws IOException {
  String [] cmd = Shell.getSetPermissionCommand(perm, recursive);
  String[] args = new String[cmd.length + 1];
  System.arraycopy(cmd, 0, args, 0, cmd.length);
  args[cmd.length] = new File(filename).getPath();
  ShellCommandExecutor shExec = new ShellCommandExecutor(args);
  try {
    shExec.execute();
  }catch(IOException e) {
    if(LOG.isDebugEnabled()) {
      LOG.debug("Error while changing permission : " + filename 
                +" Exception: " + StringUtils.stringifyException(e));
    }
  }
  return shExec.getExitCode();
}

@Override
public void setup(LocalDirAllocator allocator, LocalStorage localStorage)
    throws IOException {

  // Check the permissions of the task-controller binary by running
  // it plainly.  If permissions are correct, it returns an error
  // code 1, else it returns 24 or something else if some other bugs
  // are also present.
  String[] taskControllerCmd =
      new String[] { taskControllerExe };
  ShellCommandExecutor shExec = new ShellCommandExecutor(taskControllerCmd);
  try {
    shExec.execute();
  } catch (ExitCodeException e) {
    int exitCode = shExec.getExitCode();
    if (exitCode != 1) {
      LOG.warn("Exit code from checking binary permissions is : " + exitCode);
      logOutput(shExec.getOutput());
      throw new IOException("Task controller setup failed because of invalid"
        + "permissions/ownership with exit code " + exitCode, e);
    }
  }
  this.allocator = allocator;
  this.localStorage = localStorage;
}

@Override
public void createLogDir(TaskAttemptID taskID,
                         boolean isCleanup) throws IOException {
  // Log dirs are created during attempt dir creation when running the task
  String[] command = 
    new String[]{taskControllerExe, 
        jobUserMap.get(taskID.getJobID().toString()),
                 localStorage.getDirsString(),
                 Integer.toString(Commands.INITIALIZE_TASK.getValue()),
                 taskID.getJobID().toString(),
                 taskID.toString()};
  ShellCommandExecutor shExec = new ShellCommandExecutor(command);
  if (LOG.isDebugEnabled()) {
    LOG.debug("createLogDir: " + Arrays.toString(command));
  }
  shExec.execute();
}

@Override
public void signalTask(String user, int taskPid, 
                       Signal signal) throws IOException {
  String[] command = 
    new String[]{taskControllerExe, 
                 user,
                 localStorage.getDirsString(),
                 Integer.toString(Commands.SIGNAL_TASK.getValue()),
                 Integer.toString(taskPid),
                 Integer.toString(signal.getValue())};
  ShellCommandExecutor shExec = new ShellCommandExecutor(command);
  if (LOG.isDebugEnabled()) {
    LOG.debug("signalTask: " + Arrays.toString(command));
  }
  try {
    shExec.execute();
  } catch (ExitCodeException e) {
    int ret_code = shExec.getExitCode();
    if (ret_code != ResultCode.INVALID_TASK_PID.getValue()) {
      logOutput(shExec.getOutput());
      throw new IOException("Problem signalling task " + taskPid + " with " +
                            signal + "; exit = " + ret_code);
    }
  }
}

/**
 * Sends signal to process, forcefully terminating the process.
 * 
 * @param pid process id
 * @param signal the signal number to send
 */
public static void killProcess(String pid, Signal signal) {

  //If process tree is not alive then return immediately.
  if(!ProcessTree.isAlive(pid)) {
    return;
  }
  String[] args = { "kill", "-" + signal.getValue(), pid };
  ShellCommandExecutor shexec = new ShellCommandExecutor(args);
  try {
    shexec.execute();
  } catch (IOException e) {
    LOG.warn("Error sending signal " + signal + " to process "+ pid + " ."+ 
        StringUtils.stringifyException(e));
  } finally {
    LOG.info("Killing process " + pid + " with signal " + signal + 
             ". Exit code " + shexec.getExitCode());
  }
}

/**
 * Sends signal to all process belonging to same process group,
 * forcefully terminating the process group.
 * 
 * @param pgrpId process group id
 * @param signal the signal number to send
 */
public static void killProcessGroup(String pgrpId, Signal signal) {

  //If process tree is not alive then return immediately.
  if(!ProcessTree.isProcessGroupAlive(pgrpId)) {
    return;
  }

  String[] args = { "kill", "-" + signal.getValue() , "-"+pgrpId };
  ShellCommandExecutor shexec = new ShellCommandExecutor(args);
  try {
    shexec.execute();
  } catch (IOException e) {
    LOG.warn("Error sending signal " + signal + " to process group "+ 
             pgrpId + " ."+ 
        StringUtils.stringifyException(e));
  } finally {
    LOG.info("Killing process group" + pgrpId + " with signal " + signal + 
             ". Exit code " + shexec.getExitCode());
  }
}

public void run() {
  try {
    Vector<String> args = new Vector<String>();
    if(ProcessTree.isSetsidAvailable) {
      args.add("setsid");
    }
    args.add("bash");
    args.add("-c");
    args.add(" echo $$ > " + pidFile + "; sh " +
        shellScript + " " + N + ";") ;
    shexec = new ShellCommandExecutor(args.toArray(new String[0]));
    shexec.execute();
  } catch (ExitCodeException ee) {
    LOG.info("Shell Command exit with a non-zero exit code. This is" +
        " expected as we are killing the subprocesses of the" +
        " task intentionally. " + ee);
  } catch (IOException ioe) {
    LOG.info("Error executing shell command " + ioe);
  } finally {
    LOG.info("Exit code: " + shexec.getExitCode());
  }
}

/**
 * Helper method that runs a LinuxTaskController command
 * 
 * @param taskCommand
 * @param user
 * @param cmdArgs
 * @param env
 * @throws IOException
 */
private void runCommand(TaskCommands taskCommand, String user,
    List<String> cmdArgs, File workDir, Map<String, String> env)
    throws IOException {

  ShellCommandExecutor shExec =
      buildTaskControllerExecutor(taskCommand, user, cmdArgs, workDir, env);
  try {
    shExec.execute();
  } catch (Exception e) {
    LOG.warn("Exit code from " + taskCommand.toString() + " is : "
        + shExec.getExitCode());
    LOG.warn("Exception thrown by " + taskCommand.toString() + " : "
        + StringUtils.stringifyException(e));
    LOG.info("Output from LinuxTaskController's " + taskCommand.toString()
        + " follows:");
    logOutput(shExec.getOutput());
    throw new IOException(e);
  }
  if (LOG.isDebugEnabled()) {
    LOG.info("Output from LinuxTaskController's " + taskCommand.toString()
        + " follows:");
    logOutput(shExec.getOutput());
  }
}

/**
 * Convenience method used to sending appropriate Kill signal to the task 
 * VM
 * @param context
 * @param command
 * @throws IOException
 */
private void finishTask(TaskControllerContext context,
    TaskCommands command) throws IOException{
  if(context.task == null) {
    LOG.info("Context task null not killing the JVM");
    return;
  }
  ShellCommandExecutor shExec = buildTaskControllerExecutor(
      command, context.env.conf.getUser(), 
      buildKillTaskCommandArgs(context), context.env.workDir,
      context.env.env);
  try {
    shExec.execute();
  } catch (Exception e) {
    LOG.warn("Output from task-contoller is : " + shExec.getOutput());
    throw new IOException(e);
  }
}

@Override
void terminateTask(TaskControllerContext context) {
  ShellCommandExecutor shexec = context.shExec;
  if (shexec != null) {
    Process process = shexec.getProcess();
    if (Shell.WINDOWS) {
      // Currently we don't use setsid on WINDOWS. 
      //So kill the process alone.
      if (process != null) {
        process.destroy();
      }
    }
    else { // In addition to the task JVM, kill its subprocesses also.
      String pid = context.pid;
      if (pid != null) {
        if(ProcessTree.isSetsidAvailable) {
          ProcessTree.terminateProcessGroup(pid);
        }else {
          ProcessTree.terminateProcess(pid);
        }
      }
    }
  }
}

@Override
void killTask(TaskControllerContext context) {
  ShellCommandExecutor shexec = context.shExec;
  if (shexec != null) {
    if (Shell.WINDOWS) {
      //We don't do send kill process signal in case of windows as 
      //already we have done a process.destroy() in termintateTaskJVM()
      return;
    }
    String pid = context.pid;
    if (pid != null) {
      if(ProcessTree.isSetsidAvailable) {
        ProcessTree.killProcessGroup(pid);
      }else {
        ProcessTree.killProcess(pid);
      }
    } else {
      LOG.warn("killTask: Failed to get pid from task context " + context);
    }
  }
}

public void testShellCommandTimeout() throws Throwable {
  String rootDir = new File(System.getProperty(
      "test.build.data", "/tmp")).getAbsolutePath();
  File shellFile = new File(rootDir, "timeout.sh");
  String timeoutCommand = "sleep 4; echo \"hello\"";
  PrintWriter writer = new PrintWriter(new FileOutputStream(shellFile));
  writer.println(timeoutCommand);
  writer.close();
  shellFile.setExecutable(true);
  Shell.ShellCommandExecutor shexc 
  = new Shell.ShellCommandExecutor(new String[]{shellFile.getAbsolutePath()},
                                    null, null, 100);
  try {
    shexc.execute();
  } catch (Exception e) {
    //When timing out exception is thrown.
  }
  shellFile.delete();
  assertTrue("Script didnt not timeout" , shexc.isTimedOut());
}

public void run() {
  try {
    Vector<String> args = new Vector<String>();
    if(ProcessTree.isSetsidAvailable) {
      args.add("setsid");
    }
    args.add("bash");
    args.add("-c");
    args.add(" echo $$ > " + pidFile + "; sh " +
                      shellScript + " " + N + ";") ;
    shexec = new ShellCommandExecutor(args.toArray(new String[0]));
    shexec.execute();
  } catch (ExitCodeException ee) {
    LOG.info("Shell Command exit with a non-zero exit code. This is" +
             " expected as we are killing the subprocesses of the" +
             " task intentionally. " + ee);
  } catch (IOException ioe) {
    LOG.info("Error executing shell command " + ioe);
  } finally {
    LOG.info("Exit code: " + shexec.getExitCode());
  }
}

/**
 * Create a ShellCommandExecutor object which returns partially resolved
 * group names for a user.
 *
 * @param userName not used
 * @return a mock ShellCommandExecutor object
 */
@Override
protected ShellCommandExecutor createGroupExecutor(String userName) {
  ShellCommandExecutor executor = mock(ShellCommandExecutor.class);

  try {
    // There is both a group name 9999 and a group ID 9999.
    // This is treated as unresolvable group.
    doThrow(new ExitCodeException(1, "cannot find name for group ID 9999")).
        when(executor).execute();

    when(executor.getOutput()).thenReturn("9999\n9999 abc def");
  } catch (IOException e) {
    LOG.warn(e.getMessage());
  }
  return executor;
}

public List<TaskTrackerLoadInfo> releaseTrackers(int numTrackers)
        throws IOException {
  List<TaskTrackerLoadInfo> releasedTrackers =
          new ArrayList<TaskTrackerLoadInfo>();
  TaskTrackerLoadInfoIterator iterator = new WastedTimeTTLIIterator();
  iterator.setTrackers(trackers);
  while (releasedTrackers.size() < numTrackers && iterator.hasNext()) {
    TaskTrackerLoadInfo tracker = iterator.next();
    String host = tracker.getTaskTrackerHost();
    if (trackers.contains(host)) {
      continue;
    }
    ShellCommandExecutor removeHostCommand = new ShellCommandExecutor(
            new String[]{"ssh", hostName,
              "cd " + hadoopHome + " && " + "bin/hadoop " +
              TTMover.class.getCanonicalName() + " -remove " + host});
    try {
      removeHostCommand.execute();
      releasedTrackers.add(tracker);
    } catch (IOException ex) {
      DynamicCloudsDaemon.LOG.error("Error removing tracker " +
              tracker.getTaskTrackerName(), ex);
    }
  }
  return releasedTrackers;
}