mapreduce/src/java/org/apache/hadoop/mapred/TaskLog.java - hadoop - Git at Google

 /**
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
  * regarding copyright ownership.  The ASF licenses this file
  * to you under the Apache License, Version 2.0 (the
  * "License"); you may not use this file except in compliance
  * with the License.  You may obtain a copy of the License at
  *
  *     http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */

 package org.apache.hadoop.mapred;

 import java.io.BufferedOutputStream;
 import java.io.BufferedReader;
 import java.io.DataOutputStream;
 import java.io.File;
 import java.io.FileInputStream;
 import java.io.FileOutputStream;
 import java.io.IOException;
 import java.io.InputStream;
 import java.util.ArrayList;
 import java.util.Enumeration;
 import java.util.List;

 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;

 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.LocalFileSystem;
 import org.apache.hadoop.fs.FileUtil;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.mapreduce.JobID;
 import org.apache.hadoop.mapreduce.util.ProcessTree;
 import org.apache.hadoop.util.Shell;
 import org.apache.log4j.Appender;
 import org.apache.log4j.LogManager;
 import org.apache.log4j.Logger;

 /**
  * A simple logger to handle the task-specific user logs.
  * This class uses the system property <code>hadoop.log.dir</code>.
  *
  */
 @InterfaceAudience.Private
 public class TaskLog {
   private static final Log LOG =
     LogFactory.getLog(TaskLog.class);

   static final String USERLOGS_DIR_NAME = "userlogs";

   private static final File LOG_DIR =
     new File(getBaseLogDir(), USERLOGS_DIR_NAME).getAbsoluteFile();

   // localFS is set in (and used by) writeToIndexFile()
   static LocalFileSystem localFS = null;
   static {
     if (!LOG_DIR.exists()) {
       boolean b = LOG_DIR.mkdirs();
       if (!b) {
         LOG.warn("mkdirs failed. Ignoring.");
       }
     }
   }

   public static File getTaskLogFile(TaskAttemptID taskid, boolean isCleanup,
       LogName filter) {
     return new File(getAttemptDir(taskid, isCleanup), filter.toString());
   }

   static File getRealTaskLogFileLocation(TaskAttemptID taskid,
       boolean isCleanup, LogName filter) {
     LogFileDetail l;
     try {
       l = getLogFileDetail(taskid, filter, isCleanup);
     } catch (IOException ie) {
       LOG.error("getTaskLogFileDetail threw an exception " + ie);
       return null;
     }
     return new File(l.location, filter.toString());
   }
   private static class LogFileDetail {
     final static String LOCATION = "LOG_DIR:";
     String location;
     long start;
     long length;
   }

   private static LogFileDetail getLogFileDetail(TaskAttemptID taskid,
                                                 LogName filter,
                                                 boolean isCleanup)
   throws IOException {
     File indexFile = getIndexFile(taskid, isCleanup);
     BufferedReader fis = new BufferedReader(new java.io.FileReader(indexFile));
     //the format of the index file is
     //LOG_DIR: <the dir where the task logs are really stored>
     //stdout:<start-offset in the stdout file> <length>
     //stderr:<start-offset in the stderr file> <length>
     //syslog:<start-offset in the syslog file> <length>
     LogFileDetail l = new LogFileDetail();
     String str = fis.readLine();
     if (str == null) { //the file doesn't have anything
       throw new IOException ("Index file for the log of " + taskid+" doesn't exist.");
     }
     l.location = str.substring(str.indexOf(LogFileDetail.LOCATION)+
         LogFileDetail.LOCATION.length());
     //special cases are the debugout and profile.out files. They are guaranteed
     //to be associated with each task attempt since jvm reuse is disabled
     //when profiling/debugging is enabled
     if (filter.equals(LogName.DEBUGOUT) || filter.equals(LogName.PROFILE)) {
       l.length = new File(l.location, filter.toString()).length();
       l.start = 0;
       fis.close();
       return l;
     }
     str = fis.readLine();
     while (str != null) {
       //look for the exact line containing the logname
       if (str.contains(filter.toString())) {
         str = str.substring(filter.toString().length()+1);
         String[] startAndLen = str.split(" ");
         l.start = Long.parseLong(startAndLen[0]);
         l.length = Long.parseLong(startAndLen[1]);
         break;
       }
       str = fis.readLine();
     }
     fis.close();
     return l;
   }

   private static File getTmpIndexFile(TaskAttemptID taskid, boolean isCleanup) {
     return new File(getAttemptDir(taskid, isCleanup), "log.tmp");
   }

   static File getIndexFile(TaskAttemptID taskid, boolean isCleanup) {
     return new File(getAttemptDir(taskid, isCleanup), "log.index");
   }

   static String getBaseLogDir() {
     return System.getProperty("hadoop.log.dir");
   }

   static File getAttemptDir(TaskAttemptID taskid, boolean isCleanup) {
     String cleanupSuffix = isCleanup ? ".cleanup" : "";
     return new File(getJobDir(taskid.getJobID()), taskid + cleanupSuffix);
   }
   private static long prevOutLength;
   private static long prevErrLength;
   private static long prevLogLength;

   private static void writeToIndexFile(String logLocation,
                                        boolean isCleanup)
   throws IOException {
     // To ensure atomicity of updates to index file, write to temporary index
     // file first and then rename.
     File tmpIndexFile = getTmpIndexFile(currentTaskid, isCleanup);

     BufferedOutputStream bos =
       new BufferedOutputStream(new FileOutputStream(tmpIndexFile,false));
     DataOutputStream dos = new DataOutputStream(bos);
     //the format of the index file is
     //LOG_DIR: <the dir where the task logs are really stored>
     //STDOUT: <start-offset in the stdout file> <length>
     //STDERR: <start-offset in the stderr file> <length>
     //SYSLOG: <start-offset in the syslog file> <length>
     dos.writeBytes(LogFileDetail.LOCATION + logLocation + "\n"
         + LogName.STDOUT.toString() + ":");
     dos.writeBytes(Long.toString(prevOutLength) + " ");
     dos.writeBytes(Long.toString(new File(logLocation, LogName.STDOUT
         .toString()).length() - prevOutLength)
         + "\n" + LogName.STDERR + ":");
     dos.writeBytes(Long.toString(prevErrLength) + " ");
     dos.writeBytes(Long.toString(new File(logLocation, LogName.STDERR
         .toString()).length() - prevErrLength)
         + "\n" + LogName.SYSLOG.toString() + ":");
     dos.writeBytes(Long.toString(prevLogLength) + " ");
     dos.writeBytes(Long.toString(new File(logLocation, LogName.SYSLOG
         .toString()).length() - prevLogLength)
         + "\n");
     dos.close();

     File indexFile = getIndexFile(currentTaskid, isCleanup);
     Path indexFilePath = new Path(indexFile.getAbsolutePath());
     Path tmpIndexFilePath = new Path(tmpIndexFile.getAbsolutePath());

     if (localFS == null) {// set localFS once
       localFS = FileSystem.getLocal(new Configuration());
     }
     localFS.rename (tmpIndexFilePath, indexFilePath);
   }
   private static void resetPrevLengths(String logLocation) {
     prevOutLength = new File(logLocation, LogName.STDOUT.toString()).length();
     prevErrLength = new File(logLocation, LogName.STDERR.toString()).length();
     prevLogLength = new File(logLocation, LogName.SYSLOG.toString()).length();
   }
   private volatile static TaskAttemptID currentTaskid = null;

   @SuppressWarnings("unchecked")
   public synchronized static void syncLogs(String logLocation,
                                            TaskAttemptID taskid,
                                            boolean isCleanup)
   throws IOException {
     System.out.flush();
     System.err.flush();
     Enumeration<Logger> allLoggers = LogManager.getCurrentLoggers();
     while (allLoggers.hasMoreElements()) {
       Logger l = allLoggers.nextElement();
       Enumeration<Appender> allAppenders = l.getAllAppenders();
       while (allAppenders.hasMoreElements()) {
         Appender a = allAppenders.nextElement();
         if (a instanceof TaskLogAppender) {
           ((TaskLogAppender)a).flush();
         }
       }
     }
     if (currentTaskid != taskid) {
       currentTaskid = taskid;
       resetPrevLengths(logLocation);
     }
     writeToIndexFile(logLocation, isCleanup);
   }

   /**
    * The filter for userlogs.
    */
   @InterfaceAudience.Private
   public static enum LogName {
     /** Log on the stdout of the task. */
     STDOUT ("stdout"),

     /** Log on the stderr of the task. */
     STDERR ("stderr"),

     /** Log on the map-reduce system logs of the task. */
     SYSLOG ("syslog"),

     /** The java profiler information. */
     PROFILE ("profile.out"),

     /** Log the debug script's stdout  */
     DEBUGOUT ("debugout");

     private String prefix;

     private LogName(String prefix) {
       this.prefix = prefix;
     }

     @Override
     public String toString() {
       return prefix;
     }
   }

   public static class Reader extends InputStream {
     private long bytesRemaining;
     private FileInputStream file;

     /**
      * Read a log file from start to end positions. The offsets may be negative,
      * in which case they are relative to the end of the file. For example,
      * Reader(taskid, kind, 0, -1) is the entire file and
      * Reader(taskid, kind, -4197, -1) is the last 4196 bytes.
      * @param taskid the id of the task to read the log file for
      * @param kind the kind of log to read
      * @param start the offset to read from (negative is relative to tail)
      * @param end the offset to read upto (negative is relative to tail)
      * @param isCleanup whether the attempt is cleanup attempt or not
      * @throws IOException
      */
     public Reader(TaskAttemptID taskid, LogName kind,
                   long start, long end, boolean isCleanup) throws IOException {
       // find the right log file
       LogFileDetail fileDetail = getLogFileDetail(taskid, kind, isCleanup);
       // calculate the start and stop
       long size = fileDetail.length;
       if (start < 0) {
         start += size + 1;
       }
       if (end < 0) {
         end += size + 1;
       }
       start = Math.max(0, Math.min(start, size));
       end = Math.max(0, Math.min(end, size));
       start += fileDetail.start;
       end += fileDetail.start;
       bytesRemaining = end - start;
       file = new FileInputStream(new File(fileDetail.location, kind.toString()));
       // skip upto start
       long pos = 0;
       while (pos < start) {
         long result = file.skip(start - pos);
         if (result < 0) {
           bytesRemaining = 0;
           break;
         }
         pos += result;
       }
     }

     @Override
     public int read() throws IOException {
       int result = -1;
       if (bytesRemaining > 0) {
         bytesRemaining -= 1;
         result = file.read();
       }
       return result;
     }

     @Override
     public int read(byte[] buffer, int offset, int length) throws IOException {
       length = (int) Math.min(length, bytesRemaining);
       int bytes = file.read(buffer, offset, length);
       if (bytes > 0) {
         bytesRemaining -= bytes;
       }
       return bytes;
     }

     @Override
     public int available() throws IOException {
       return (int) Math.min(bytesRemaining, file.available());
     }

     @Override
     public void close() throws IOException {
       file.close();
     }
   }

   private static final String bashCommand = "bash";
   private static final String tailCommand = "tail";

   /**
    * Get the desired maximum length of task's logs.
    * @param conf the job to look in
    * @return the number of bytes to cap the log files at
    */
   public static long getTaskLogLength(JobConf conf) {
     return conf.getLong(JobContext.TASK_USERLOG_LIMIT, 0) * 1024;
   }

   /**
    * Wrap a command in a shell to capture stdout and stderr to files.
    * If the tailLength is 0, the entire output will be saved.
    * @param cmd The command and the arguments that should be run
    * @param stdoutFilename The filename that stdout should be saved to
    * @param stderrFilename The filename that stderr should be saved to
    * @param tailLength The length of the tail to be saved.
    * @return the modified command that should be run
    */
   public static List<String> captureOutAndError(List<String> cmd,
                                                 File stdoutFilename,
                                                 File stderrFilename,
                                                 long tailLength
                                                ) throws IOException {
     return captureOutAndError(null, cmd, stdoutFilename,
                               stderrFilename, tailLength, false);
   }

   /**
    * Wrap a command in a shell to capture stdout and stderr to files.
    * Setup commands such as setting memory limit can be passed which
    * will be executed before exec.
    * If the tailLength is 0, the entire output will be saved.
    * @param setup The setup commands for the execed process.
    * @param cmd The command and the arguments that should be run
    * @param stdoutFilename The filename that stdout should be saved to
    * @param stderrFilename The filename that stderr should be saved to
    * @param tailLength The length of the tail to be saved.
    * @return the modified command that should be run
    */
   public static List<String> captureOutAndError(List<String> setup,
                                                 List<String> cmd,
                                                 File stdoutFilename,
                                                 File stderrFilename,
                                                 long tailLength
                                                ) throws IOException {
     return captureOutAndError(setup, cmd, stdoutFilename, stderrFilename,
                               tailLength, false);
   }

   /**
    * Wrap a command in a shell to capture stdout and stderr to files.
    * Setup commands such as setting memory limit can be passed which
    * will be executed before exec.
    * If the tailLength is 0, the entire output will be saved.
    * @param setup The setup commands for the execed process.
    * @param cmd The command and the arguments that should be run
    * @param stdoutFilename The filename that stdout should be saved to
    * @param stderrFilename The filename that stderr should be saved to
    * @param tailLength The length of the tail to be saved.
    * @param pidFileName The name of the pid-file. pid-file's usage is deprecated
    * @return the modified command that should be run
    *
    * @deprecated     pidFiles are no more used. Instead pid is exported to
    *                 env variable JVM_PID.
    */
   @Deprecated
   public static List<String> captureOutAndError(List<String> setup,
                                                 List<String> cmd,
                                                 File stdoutFilename,
                                                 File stderrFilename,
                                                 long tailLength,
                                                 String pidFileName
                                                ) throws IOException {
     return captureOutAndError(setup, cmd, stdoutFilename, stderrFilename,
         tailLength, false);
   }

   /**
    * Wrap a command in a shell to capture stdout and stderr to files.
    * Setup commands such as setting memory limit can be passed which
    * will be executed before exec.
    * If the tailLength is 0, the entire output will be saved.
    * @param setup The setup commands for the execed process.
    * @param cmd The command and the arguments that should be run
    * @param stdoutFilename The filename that stdout should be saved to
    * @param stderrFilename The filename that stderr should be saved to
    * @param tailLength The length of the tail to be saved.
    * @param useSetsid Should setsid be used in the command or not.
    * @return the modified command that should be run
    */
   public static List<String> captureOutAndError(List<String> setup,
                                                 List<String> cmd,
                                                 File stdoutFilename,
                                                 File stderrFilename,
                                                 long tailLength,
                                                 boolean useSetsid
                                                ) throws IOException {
     List<String> result = new ArrayList<String>(3);
     result.add(bashCommand);
     result.add("-c");
     String mergedCmd = buildCommandLine(setup, cmd, stdoutFilename,
                                                     stderrFilename, tailLength,
                                                     useSetsid);
     result.add(mergedCmd);
     return result;
   }

   /**
    * Construct the command line for running the task JVM
    * @param setup The setup commands for the execed process.
    * @param cmd The command and the arguments that should be run
    * @param stdoutFilename The filename that stdout should be saved to
    * @param stderrFilename The filename that stderr should be saved to
    * @param tailLength The length of the tail to be saved.
    * @return the command line as a String
    * @throws IOException
    */
   static String buildCommandLine(List<String> setup, List<String> cmd,
                                       File stdoutFilename,
                                       File stderrFilename,
                                       long tailLength,
                                       boolean useSetsid)
                                 throws IOException {

     String stdout = FileUtil.makeShellPath(stdoutFilename);
     String stderr = FileUtil.makeShellPath(stderrFilename);
     StringBuffer mergedCmd = new StringBuffer();

     // Export the pid of taskJvm to env variable JVM_PID.
     // Currently pid is not used on Windows
     if (!Shell.WINDOWS) {
       mergedCmd.append(" export JVM_PID=`echo $$` ; ");
     }

     if (setup != null && setup.size() > 0) {
       mergedCmd.append(addCommand(setup, false));
       mergedCmd.append(";");
     }
     if (tailLength > 0) {
       mergedCmd.append("(");
     } else if(ProcessTree.isSetsidAvailable && useSetsid &&
         !Shell.WINDOWS) {
       mergedCmd.append("exec setsid ");
     } else {
       mergedCmd.append("exec ");
     }
     mergedCmd.append(addCommand(cmd, true));
     mergedCmd.append(" < /dev/null ");
     if (tailLength > 0) {
       mergedCmd.append(" | ");
       mergedCmd.append(tailCommand);
       mergedCmd.append(" -c ");
       mergedCmd.append(tailLength);
       mergedCmd.append(" >> ");
       mergedCmd.append(stdout);
       mergedCmd.append(" ; exit $PIPESTATUS ) 2>&1 | ");
       mergedCmd.append(tailCommand);
       mergedCmd.append(" -c ");
       mergedCmd.append(tailLength);
       mergedCmd.append(" >> ");
       mergedCmd.append(stderr);
       mergedCmd.append(" ; exit $PIPESTATUS");
     } else {
       mergedCmd.append(" 1>> ");
       mergedCmd.append(stdout);
       mergedCmd.append(" 2>> ");
       mergedCmd.append(stderr);
     }
     return mergedCmd.toString();
   }

   /**
    * Construct the command line for running the debug script
    * @param cmd The command and the arguments that should be run
    * @param stdoutFilename The filename that stdout should be saved to
    * @param stderrFilename The filename that stderr should be saved to
    * @param tailLength The length of the tail to be saved.
    * @return the command line as a String
    * @throws IOException
    */
   static String buildDebugScriptCommandLine(List<String> cmd, String debugout)
   throws IOException {
     StringBuilder mergedCmd = new StringBuilder();
     mergedCmd.append("exec ");
     boolean isExecutable = true;
     for(String s: cmd) {
       if (isExecutable) {
         // the executable name needs to be expressed as a shell path for the
         // shell to find it.
         mergedCmd.append(FileUtil.makeShellPath(new File(s)));
         isExecutable = false;
       } else {
         mergedCmd.append(s);
       }
       mergedCmd.append(" ");
     }
     mergedCmd.append(" < /dev/null ");
     mergedCmd.append(" >");
     mergedCmd.append(debugout);
     mergedCmd.append(" 2>&1 ");
     return mergedCmd.toString();
   }
   /**
    * Add quotes to each of the command strings and
    * return as a single string
    * @param cmd The command to be quoted
    * @param isExecutable makes shell path if the first
    * argument is executable
    * @return returns The quoted string.
    * @throws IOException
    */
   public static String addCommand(List<String> cmd, boolean isExecutable)
   throws IOException {
     StringBuffer command = new StringBuffer();
     for(String s: cmd) {
     	command.append('\'');
       if (isExecutable) {
         // the executable name needs to be expressed as a shell path for the
         // shell to find it.
     	  command.append(FileUtil.makeShellPath(new File(s)));
         isExecutable = false;
       } else {
     	  command.append(s);
       }
       command.append('\'');
       command.append(" ");
     }
     return command.toString();
   }

   /**
    * Wrap a command in a shell to capture debug script's
    * stdout and stderr to debugout.
    * @param cmd The command and the arguments that should be run
    * @param debugoutFilename The filename that stdout and stderr
    *  should be saved to.
    * @return the modified command that should be run
    * @throws IOException
    */
   public static List<String> captureDebugOut(List<String> cmd,
                                              File debugoutFilename
                                             ) throws IOException {
     String debugout = FileUtil.makeShellPath(debugoutFilename);
     List<String> result = new ArrayList<String>(3);
     result.add(bashCommand);
     result.add("-c");
     result.add(buildDebugScriptCommandLine(cmd, debugout));
     return result;
   }

   /**
    * Method to return the location of user log directory.
    *
    * @return base log directory
    */
   static File getUserLogDir() {
     return LOG_DIR;
   }

   /**
    * Get the user log directory for the job jobid.
    *
    * @param jobid
    * @return user log directory for the job
    */
   public static File getJobDir(JobID jobid) {
     return new File(getUserLogDir(), jobid.toString());
   }

 } // TaskLog
	/**
	* Licensed to the Apache Software Foundation (ASF) under one
	* or more contributor license agreements. See the NOTICE file
	* distributed with this work for additional information
	* regarding copyright ownership. The ASF licenses this file
	* to you under the Apache License, Version 2.0 (the
	* "License"); you may not use this file except in compliance
	* with the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing, software
	* distributed under the License is distributed on an "AS IS" BASIS,
	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	* See the License for the specific language governing permissions and
	* limitations under the License.
	*/

	package org.apache.hadoop.mapred;

	import java.io.BufferedOutputStream;
	import java.io.BufferedReader;
	import java.io.DataOutputStream;
	import java.io.File;
	import java.io.FileInputStream;
	import java.io.FileOutputStream;
	import java.io.IOException;
	import java.io.InputStream;
	import java.util.ArrayList;
	import java.util.Enumeration;
	import java.util.List;

	import org.apache.commons.logging.Log;
	import org.apache.commons.logging.LogFactory;

	import org.apache.hadoop.classification.InterfaceAudience;
	import org.apache.hadoop.conf.Configuration;
	import org.apache.hadoop.fs.FileSystem;
	import org.apache.hadoop.fs.LocalFileSystem;
	import org.apache.hadoop.fs.FileUtil;
	import org.apache.hadoop.fs.Path;
	import org.apache.hadoop.mapreduce.JobID;
	import org.apache.hadoop.mapreduce.util.ProcessTree;
	import org.apache.hadoop.util.Shell;
	import org.apache.log4j.Appender;
	import org.apache.log4j.LogManager;
	import org.apache.log4j.Logger;

	/**
	* A simple logger to handle the task-specific user logs.
	* This class uses the system property <code>hadoop.log.dir</code>.
	*
	*/
	@InterfaceAudience.Private
	public class TaskLog {
	private static final Log LOG =
	LogFactory.getLog(TaskLog.class);

	static final String USERLOGS_DIR_NAME = "userlogs";

	private static final File LOG_DIR =
	new File(getBaseLogDir(), USERLOGS_DIR_NAME).getAbsoluteFile();

	// localFS is set in (and used by) writeToIndexFile()
	static LocalFileSystem localFS = null;
	static {
	if (!LOG_DIR.exists()) {
	boolean b = LOG_DIR.mkdirs();
	if (!b) {
	LOG.warn("mkdirs failed. Ignoring.");
	}
	}
	}

	public static File getTaskLogFile(TaskAttemptID taskid, boolean isCleanup,
	LogName filter) {
	return new File(getAttemptDir(taskid, isCleanup), filter.toString());
	}

	static File getRealTaskLogFileLocation(TaskAttemptID taskid,
	boolean isCleanup, LogName filter) {
	LogFileDetail l;
	try {
	l = getLogFileDetail(taskid, filter, isCleanup);
	} catch (IOException ie) {
	LOG.error("getTaskLogFileDetail threw an exception " + ie);
	return null;
	}
	return new File(l.location, filter.toString());
	}
	private static class LogFileDetail {
	final static String LOCATION = "LOG_DIR:";
	String location;
	long start;
	long length;
	}

	private static LogFileDetail getLogFileDetail(TaskAttemptID taskid,
	LogName filter,
	boolean isCleanup)
	throws IOException {
	File indexFile = getIndexFile(taskid, isCleanup);
	BufferedReader fis = new BufferedReader(new java.io.FileReader(indexFile));
	//the format of the index file is
	//LOG_DIR: <the dir where the task logs are really stored>
	//stdout:<start-offset in the stdout file> <length>
	//stderr:<start-offset in the stderr file> <length>
	//syslog:<start-offset in the syslog file> <length>
	LogFileDetail l = new LogFileDetail();
	String str = fis.readLine();
	if (str == null) { //the file doesn't have anything
	throw new IOException ("Index file for the log of " + taskid+" doesn't exist.");
	}
	l.location = str.substring(str.indexOf(LogFileDetail.LOCATION)+
	LogFileDetail.LOCATION.length());
	//special cases are the debugout and profile.out files. They are guaranteed
	//to be associated with each task attempt since jvm reuse is disabled
	//when profiling/debugging is enabled
	if (filter.equals(LogName.DEBUGOUT) \|\| filter.equals(LogName.PROFILE)) {
	l.length = new File(l.location, filter.toString()).length();
	l.start = 0;
	fis.close();
	return l;
	}
	str = fis.readLine();
	while (str != null) {
	//look for the exact line containing the logname
	if (str.contains(filter.toString())) {
	str = str.substring(filter.toString().length()+1);
	String[] startAndLen = str.split(" ");
	l.start = Long.parseLong(startAndLen[0]);
	l.length = Long.parseLong(startAndLen[1]);
	break;
	}
	str = fis.readLine();
	}
	fis.close();
	return l;
	}

	private static File getTmpIndexFile(TaskAttemptID taskid, boolean isCleanup) {
	return new File(getAttemptDir(taskid, isCleanup), "log.tmp");
	}

	static File getIndexFile(TaskAttemptID taskid, boolean isCleanup) {
	return new File(getAttemptDir(taskid, isCleanup), "log.index");
	}

	static String getBaseLogDir() {
	return System.getProperty("hadoop.log.dir");
	}

	static File getAttemptDir(TaskAttemptID taskid, boolean isCleanup) {
	String cleanupSuffix = isCleanup ? ".cleanup" : "";
	return new File(getJobDir(taskid.getJobID()), taskid + cleanupSuffix);
	}
	private static long prevOutLength;
	private static long prevErrLength;
	private static long prevLogLength;

	private static void writeToIndexFile(String logLocation,
	boolean isCleanup)
	throws IOException {
	// To ensure atomicity of updates to index file, write to temporary index
	// file first and then rename.
	File tmpIndexFile = getTmpIndexFile(currentTaskid, isCleanup);

	BufferedOutputStream bos =
	new BufferedOutputStream(new FileOutputStream(tmpIndexFile,false));
	DataOutputStream dos = new DataOutputStream(bos);
	//the format of the index file is
	//LOG_DIR: <the dir where the task logs are really stored>
	//STDOUT: <start-offset in the stdout file> <length>
	//STDERR: <start-offset in the stderr file> <length>
	//SYSLOG: <start-offset in the syslog file> <length>
	dos.writeBytes(LogFileDetail.LOCATION + logLocation + "\n"
	+ LogName.STDOUT.toString() + ":");
	dos.writeBytes(Long.toString(prevOutLength) + " ");
	dos.writeBytes(Long.toString(new File(logLocation, LogName.STDOUT
	.toString()).length() - prevOutLength)
	+ "\n" + LogName.STDERR + ":");
	dos.writeBytes(Long.toString(prevErrLength) + " ");
	dos.writeBytes(Long.toString(new File(logLocation, LogName.STDERR
	.toString()).length() - prevErrLength)
	+ "\n" + LogName.SYSLOG.toString() + ":");
	dos.writeBytes(Long.toString(prevLogLength) + " ");
	dos.writeBytes(Long.toString(new File(logLocation, LogName.SYSLOG
	.toString()).length() - prevLogLength)
	+ "\n");
	dos.close();

	File indexFile = getIndexFile(currentTaskid, isCleanup);
	Path indexFilePath = new Path(indexFile.getAbsolutePath());
	Path tmpIndexFilePath = new Path(tmpIndexFile.getAbsolutePath());

	if (localFS == null) {// set localFS once
	localFS = FileSystem.getLocal(new Configuration());
	}
	localFS.rename (tmpIndexFilePath, indexFilePath);
	}
	private static void resetPrevLengths(String logLocation) {
	prevOutLength = new File(logLocation, LogName.STDOUT.toString()).length();
	prevErrLength = new File(logLocation, LogName.STDERR.toString()).length();
	prevLogLength = new File(logLocation, LogName.SYSLOG.toString()).length();
	}
	private volatile static TaskAttemptID currentTaskid = null;

	@SuppressWarnings("unchecked")
	public synchronized static void syncLogs(String logLocation,
	TaskAttemptID taskid,
	boolean isCleanup)
	throws IOException {
	System.out.flush();
	System.err.flush();
	Enumeration<Logger> allLoggers = LogManager.getCurrentLoggers();
	while (allLoggers.hasMoreElements()) {
	Logger l = allLoggers.nextElement();
	Enumeration<Appender> allAppenders = l.getAllAppenders();
	while (allAppenders.hasMoreElements()) {
	Appender a = allAppenders.nextElement();
	if (a instanceof TaskLogAppender) {
	((TaskLogAppender)a).flush();
	}
	}
	}
	if (currentTaskid != taskid) {
	currentTaskid = taskid;
	resetPrevLengths(logLocation);
	}
	writeToIndexFile(logLocation, isCleanup);
	}

	/**
	* The filter for userlogs.
	*/
	@InterfaceAudience.Private
	public static enum LogName {
	/** Log on the stdout of the task. */
	STDOUT ("stdout"),

	/** Log on the stderr of the task. */
	STDERR ("stderr"),

	/** Log on the map-reduce system logs of the task. */
	SYSLOG ("syslog"),

	/** The java profiler information. */
	PROFILE ("profile.out"),

	/** Log the debug script's stdout */
	DEBUGOUT ("debugout");

	private String prefix;

	private LogName(String prefix) {
	this.prefix = prefix;
	}

	@Override
	public String toString() {
	return prefix;
	}
	}

	public static class Reader extends InputStream {
	private long bytesRemaining;
	private FileInputStream file;

	/**
	* Read a log file from start to end positions. The offsets may be negative,
	* in which case they are relative to the end of the file. For example,
	* Reader(taskid, kind, 0, -1) is the entire file and
	* Reader(taskid, kind, -4197, -1) is the last 4196 bytes.
	* @param taskid the id of the task to read the log file for
	* @param kind the kind of log to read
	* @param start the offset to read from (negative is relative to tail)
	* @param end the offset to read upto (negative is relative to tail)
	* @param isCleanup whether the attempt is cleanup attempt or not
	* @throws IOException
	*/
	public Reader(TaskAttemptID taskid, LogName kind,
	long start, long end, boolean isCleanup) throws IOException {
	// find the right log file
	LogFileDetail fileDetail = getLogFileDetail(taskid, kind, isCleanup);
	// calculate the start and stop
	long size = fileDetail.length;
	if (start < 0) {
	start += size + 1;
	}
	if (end < 0) {
	end += size + 1;
	}
	start = Math.max(0, Math.min(start, size));
	end = Math.max(0, Math.min(end, size));
	start += fileDetail.start;
	end += fileDetail.start;
	bytesRemaining = end - start;
	file = new FileInputStream(new File(fileDetail.location, kind.toString()));
	// skip upto start
	long pos = 0;
	while (pos < start) {
	long result = file.skip(start - pos);
	if (result < 0) {
	bytesRemaining = 0;
	break;
	}
	pos += result;
	}
	}

	@Override
	public int read() throws IOException {
	int result = -1;
	if (bytesRemaining > 0) {
	bytesRemaining -= 1;
	result = file.read();
	}
	return result;
	}

	@Override
	public int read(byte[] buffer, int offset, int length) throws IOException {
	length = (int) Math.min(length, bytesRemaining);
	int bytes = file.read(buffer, offset, length);
	if (bytes > 0) {
	bytesRemaining -= bytes;
	}
	return bytes;
	}

	@Override
	public int available() throws IOException {
	return (int) Math.min(bytesRemaining, file.available());
	}

	@Override
	public void close() throws IOException {
	file.close();
	}
	}

	private static final String bashCommand = "bash";
	private static final String tailCommand = "tail";

	/**
	* Get the desired maximum length of task's logs.
	* @param conf the job to look in
	* @return the number of bytes to cap the log files at
	*/
	public static long getTaskLogLength(JobConf conf) {
	return conf.getLong(JobContext.TASK_USERLOG_LIMIT, 0) * 1024;
	}

	/**
	* Wrap a command in a shell to capture stdout and stderr to files.
	* If the tailLength is 0, the entire output will be saved.
	* @param cmd The command and the arguments that should be run
	* @param stdoutFilename The filename that stdout should be saved to
	* @param stderrFilename The filename that stderr should be saved to
	* @param tailLength The length of the tail to be saved.
	* @return the modified command that should be run
	*/
	public static List<String> captureOutAndError(List<String> cmd,
	File stdoutFilename,
	File stderrFilename,
	long tailLength
	) throws IOException {
	return captureOutAndError(null, cmd, stdoutFilename,
	stderrFilename, tailLength, false);
	}

	/**
	* Wrap a command in a shell to capture stdout and stderr to files.
	* Setup commands such as setting memory limit can be passed which
	* will be executed before exec.
	* If the tailLength is 0, the entire output will be saved.
	* @param setup The setup commands for the execed process.
	* @param cmd The command and the arguments that should be run
	* @param stdoutFilename The filename that stdout should be saved to
	* @param stderrFilename The filename that stderr should be saved to
	* @param tailLength The length of the tail to be saved.
	* @return the modified command that should be run
	*/
	public static List<String> captureOutAndError(List<String> setup,
	List<String> cmd,
	File stdoutFilename,
	File stderrFilename,
	long tailLength
	) throws IOException {
	return captureOutAndError(setup, cmd, stdoutFilename, stderrFilename,
	tailLength, false);
	}

	/**
	* Wrap a command in a shell to capture stdout and stderr to files.
	* Setup commands such as setting memory limit can be passed which
	* will be executed before exec.
	* If the tailLength is 0, the entire output will be saved.
	* @param setup The setup commands for the execed process.
	* @param cmd The command and the arguments that should be run
	* @param stdoutFilename The filename that stdout should be saved to
	* @param stderrFilename The filename that stderr should be saved to
	* @param tailLength The length of the tail to be saved.
	* @param pidFileName The name of the pid-file. pid-file's usage is deprecated
	* @return the modified command that should be run
	*
	* @deprecated pidFiles are no more used. Instead pid is exported to
	* env variable JVM_PID.
	*/
	@Deprecated
	public static List<String> captureOutAndError(List<String> setup,
	List<String> cmd,
	File stdoutFilename,
	File stderrFilename,
	long tailLength,
	String pidFileName
	) throws IOException {
	return captureOutAndError(setup, cmd, stdoutFilename, stderrFilename,
	tailLength, false);
	}

	/**
	* Wrap a command in a shell to capture stdout and stderr to files.
	* Setup commands such as setting memory limit can be passed which
	* will be executed before exec.
	* If the tailLength is 0, the entire output will be saved.
	* @param setup The setup commands for the execed process.
	* @param cmd The command and the arguments that should be run
	* @param stdoutFilename The filename that stdout should be saved to
	* @param stderrFilename The filename that stderr should be saved to
	* @param tailLength The length of the tail to be saved.
	* @param useSetsid Should setsid be used in the command or not.
	* @return the modified command that should be run
	*/
	public static List<String> captureOutAndError(List<String> setup,
	List<String> cmd,
	File stdoutFilename,
	File stderrFilename,
	long tailLength,
	boolean useSetsid
	) throws IOException {
	List<String> result = new ArrayList<String>(3);
	result.add(bashCommand);
	result.add("-c");
	String mergedCmd = buildCommandLine(setup, cmd, stdoutFilename,
	stderrFilename, tailLength,
	useSetsid);
	result.add(mergedCmd);
	return result;
	}

	/**
	* Construct the command line for running the task JVM
	* @param setup The setup commands for the execed process.
	* @param cmd The command and the arguments that should be run
	* @param stdoutFilename The filename that stdout should be saved to
	* @param stderrFilename The filename that stderr should be saved to
	* @param tailLength The length of the tail to be saved.
	* @return the command line as a String
	* @throws IOException
	*/
	static String buildCommandLine(List<String> setup, List<String> cmd,
	File stdoutFilename,
	File stderrFilename,
	long tailLength,
	boolean useSetsid)
	throws IOException {

	String stdout = FileUtil.makeShellPath(stdoutFilename);
	String stderr = FileUtil.makeShellPath(stderrFilename);
	StringBuffer mergedCmd = new StringBuffer();

	// Export the pid of taskJvm to env variable JVM_PID.
	// Currently pid is not used on Windows
	if (!Shell.WINDOWS) {
	mergedCmd.append(" export JVM_PID=`echo $$` ; ");
	}

	if (setup != null && setup.size() > 0) {
	mergedCmd.append(addCommand(setup, false));
	mergedCmd.append(";");
	}
	if (tailLength > 0) {
	mergedCmd.append("(");
	} else if(ProcessTree.isSetsidAvailable && useSetsid &&
	!Shell.WINDOWS) {
	mergedCmd.append("exec setsid ");
	} else {
	mergedCmd.append("exec ");
	}
	mergedCmd.append(addCommand(cmd, true));
	mergedCmd.append(" < /dev/null ");
	if (tailLength > 0) {
	mergedCmd.append(" \| ");
	mergedCmd.append(tailCommand);
	mergedCmd.append(" -c ");
	mergedCmd.append(tailLength);
	mergedCmd.append(" >> ");
	mergedCmd.append(stdout);
	mergedCmd.append(" ; exit $PIPESTATUS ) 2>&1 \| ");
	mergedCmd.append(tailCommand);
	mergedCmd.append(" -c ");
	mergedCmd.append(tailLength);
	mergedCmd.append(" >> ");
	mergedCmd.append(stderr);
	mergedCmd.append(" ; exit $PIPESTATUS");
	} else {
	mergedCmd.append(" 1>> ");
	mergedCmd.append(stdout);
	mergedCmd.append(" 2>> ");
	mergedCmd.append(stderr);
	}
	return mergedCmd.toString();
	}

	/**
	* Construct the command line for running the debug script
	* @param cmd The command and the arguments that should be run
	* @param stdoutFilename The filename that stdout should be saved to
	* @param stderrFilename The filename that stderr should be saved to
	* @param tailLength The length of the tail to be saved.
	* @return the command line as a String
	* @throws IOException
	*/
	static String buildDebugScriptCommandLine(List<String> cmd, String debugout)
	throws IOException {
	StringBuilder mergedCmd = new StringBuilder();
	mergedCmd.append("exec ");
	boolean isExecutable = true;
	for(String s: cmd) {
	if (isExecutable) {
	// the executable name needs to be expressed as a shell path for the
	// shell to find it.
	mergedCmd.append(FileUtil.makeShellPath(new File(s)));
	isExecutable = false;
	} else {
	mergedCmd.append(s);
	}
	mergedCmd.append(" ");
	}
	mergedCmd.append(" < /dev/null ");
	mergedCmd.append(" >");
	mergedCmd.append(debugout);
	mergedCmd.append(" 2>&1 ");
	return mergedCmd.toString();
	}
	/**
	* Add quotes to each of the command strings and
	* return as a single string
	* @param cmd The command to be quoted
	* @param isExecutable makes shell path if the first
	* argument is executable
	* @return returns The quoted string.
	* @throws IOException
	*/
	public static String addCommand(List<String> cmd, boolean isExecutable)
	throws IOException {
	StringBuffer command = new StringBuffer();
	for(String s: cmd) {
	command.append('\'');
	if (isExecutable) {
	// the executable name needs to be expressed as a shell path for the
	// shell to find it.
	command.append(FileUtil.makeShellPath(new File(s)));
	isExecutable = false;
	} else {
	command.append(s);
	}
	command.append('\'');
	command.append(" ");
	}
	return command.toString();
	}

	/**
	* Wrap a command in a shell to capture debug script's
	* stdout and stderr to debugout.
	* @param cmd The command and the arguments that should be run
	* @param debugoutFilename The filename that stdout and stderr
	* should be saved to.
	* @return the modified command that should be run
	* @throws IOException
	*/
	public static List<String> captureDebugOut(List<String> cmd,
	File debugoutFilename
	) throws IOException {
	String debugout = FileUtil.makeShellPath(debugoutFilename);
	List<String> result = new ArrayList<String>(3);
	result.add(bashCommand);
	result.add("-c");
	result.add(buildDebugScriptCommandLine(cmd, debugout));
	return result;
	}

	/**
	* Method to return the location of user log directory.
	*
	* @return base log directory
	*/
	static File getUserLogDir() {
	return LOG_DIR;
	}

	/**
	* Get the user log directory for the job jobid.
	*
	* @param jobid
	* @return user log directory for the job
	*/
	public static File getJobDir(JobID jobid) {
	return new File(getUserLogDir(), jobid.toString());
	}

	} // TaskLog