HADOOP-975. Separate stdout and stderr from tasks. Contributed by Arun.

git-svn-id: https://svn.apache.org/repos/asf/lucene/hadoop/trunk@504678 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -1,6 +1,12 @@
 Hadoop Change Log
 
 
+Trunk (unreleased changes)
+
+ 1. HADOOP-975.  Separate stdout and stderr from tasks.
+    (Arun C Murthy via cutting)
+
+
 Branch 0.11 - unreleased
 
  1. HADOOP-976.  Make SequenceFile.Metadata public.  (Runping Qi via cutting)

src/java/org/apache/hadoop/mapred/TaskLog.java

@@ -45,6 +45,27 @@ class TaskLog {
     }
   }
 
+  /**
+   * The filter for userlogs.
+   */
+  public static enum LogFilter {
+    /** Log on the stdout of the task. */
+    STDOUT ("stdout"),
+
+    /** Log on the stderr of the task. */
+    STDERR ("stderr");
+    
+    private String prefix;
+    
+    private LogFilter(String prefix) {
+      this.prefix = prefix;
+    }
+    
+    String getPrefix() {
+      return prefix;
+    }
+  }
+  
   /**
    * The log-writer responsible for handling writing user-logs
    * and maintaining splits and ensuring job-specifc limits 
@@ -55,6 +76,7 @@ class TaskLog {
   static class Writer {
     private String taskId;
     private JobConf conf;
+    private LogFilter filter;
 
     private final File taskLogDir;
     private final int noKeepSplits;
@@ -78,11 +100,15 @@ class TaskLog {
      * Creates a new TaskLog writer.
      * @param conf configuration of the task
      * @param taskId taskid of the task
+     * @param filter the {@link LogFilter} to apply on userlogs.
      */
-    Writer(JobConf conf, String taskId) {
+    Writer(JobConf conf, String taskId, LogFilter filter) {
       this.conf = conf;
       this.taskId = taskId;
-      this.taskLogDir = new File(LOG_DIR, this.taskId);
+      this.filter = filter;
+      
+      this.taskLogDir = new File(new File(LOG_DIR, this.taskId), 
+                                this.filter.getPrefix());
       
       noKeepSplits = this.conf.getInt("mapred.userlog.num.splits", 4);
       splitFileSize = 
@@ -114,11 +140,14 @@ class TaskLog {
       File[] files = dir.listFiles();
       if (files != null) {
         for (int i=0; i < files.length; ++i) {
+          if (files[i].isDirectory()) {
+            deleteDir(files[i]);
+          }
           files[i].delete();
         }
       }
       boolean del = dir.delete();
-      LOG.debug("Deleted " + del + ": " + del);
+      LOG.debug("Deleted " + dir + ": " + del);
     }
     
     /**
@@ -257,6 +286,8 @@ class TaskLog {
    */
   static class Reader {
     private String taskId;
+    private LogFilter filter;
+    
     private File taskLogDir;
     private boolean initialized = false;
     
@@ -269,10 +300,14 @@ class TaskLog {
      * Create a new task log reader.
      * 
      * @param taskId task id of the task.
+     * @param filter the {@link LogFilter} to apply on userlogs.
      */
-    public Reader(String taskId) {
+    public Reader(String taskId, LogFilter filter) {
       this.taskId = taskId;
-      this.taskLogDir = new File(LOG_DIR, this.taskId);
+      this.filter = filter;
+      
+      this.taskLogDir = new File(new File(LOG_DIR, this.taskId), 
+                                this.filter.getPrefix());
     }
 
     private static class IndexRecord {
@@ -289,8 +324,8 @@ class TaskLog {
     
     private synchronized void init() throws IOException {
       this.splitIndex = new BufferedReader(new InputStreamReader(
-                          new FileInputStream(new File(taskLogDir, SPLIT_INDEX_NAME))
-                          ));
+                          new FileInputStream(new File(taskLogDir, 
+                                  SPLIT_INDEX_NAME))));
 
       // Parse the split-index and store the offsets/lengths
       ArrayList<IndexRecord> records = new ArrayList<IndexRecord>();
@@ -336,6 +371,7 @@ class TaskLog {
       
       return logFileSize;
     }
+    
     /**
      * Return the entire user-log (remaining splits).
      * 
@@ -427,7 +463,7 @@ class TaskLog {
     public synchronized int read(byte[] b, int off, int len, 
         long logOffset, long logLength) 
     throws IOException {
-      LOG.debug("TaskeLog.Reader.read: logOffset: " + logOffset + " - logLength: " + logLength);
+      LOG.debug("TaskLog.Reader.read: logOffset: " + logOffset + " - logLength: " + logLength);
 
       // Sanity check
       if (logLength == 0) {

src/java/org/apache/hadoop/mapred/TaskRunner.java

@@ -41,13 +41,17 @@ abstract class TaskRunner extends Thread {
 
   protected JobConf conf;
 
-  private TaskLog.Writer taskLogWriter;
+  private TaskLog.Writer taskStdOutLogWriter;
+  private TaskLog.Writer taskStdErrLogWriter;
   
   public TaskRunner(Task t, TaskTracker tracker, JobConf conf) {
     this.t = t;
     this.tracker = tracker;
     this.conf = conf;
-    this.taskLogWriter = new TaskLog.Writer(conf, t.getTaskId());
+    this.taskStdOutLogWriter = 
+      new TaskLog.Writer(conf, t.getTaskId(), TaskLog.LogFilter.STDOUT);
+    this.taskStdErrLogWriter = 
+      new TaskLog.Writer(conf, t.getTaskId(), TaskLog.LogFilter.STDERR);
   }
 
   public Task getTask() { return t; }
@@ -57,7 +61,8 @@ abstract class TaskRunner extends Thread {
    * process before the child is spawned.  It should not execute user code,
    * only system code. */
   public boolean prepare() throws IOException {
-    taskLogWriter.init();                   // initialize the child task's log
+    taskStdOutLogWriter.init();       // initialize the child task's stdout log
+    taskStdErrLogWriter.init();       // initialize the child task's stderr log
     return true;
   }
 
@@ -323,11 +328,13 @@ abstract class TaskRunner extends Thread {
     try {
       new Thread() {
         public void run() {
-          logStream(process.getErrorStream());    // copy log output
+          // Copy stderr of the process
+          logStream(process.getErrorStream(), taskStdErrLogWriter); 
         }
       }.start();
         
-      logStream(process.getInputStream());		  // normally empty
+      // Copy stderr of the process; normally empty
+      logStream(process.getInputStream(), taskStdOutLogWriter);		  
       
       int exit_code = process.waitFor();
      
@@ -340,7 +347,8 @@ abstract class TaskRunner extends Thread {
       throw new IOException(e.toString());
     } finally {
       kill();
-      taskLogWriter.close();
+      taskStdOutLogWriter.close();
+      taskStdErrLogWriter.close();
     }
   }
 
@@ -356,13 +364,13 @@ abstract class TaskRunner extends Thread {
 
   /**
    */
-  private void logStream(InputStream output) {
+  private void logStream(InputStream output, TaskLog.Writer taskLog) {
     try {
       byte[] buf = new byte[512];
       int n = 0;
       while ((n = output.read(buf, 0, buf.length)) != -1) {
         // Write out to the task's log
-        taskLogWriter.write(buf, 0, n);
+        taskLog.write(buf, 0, n);
       }
     } catch (IOException e) {
       LOG.warn(t.getTaskId()+" Error reading child output", e);

src/webapps/task/tasklog.jsp

@@ -10,7 +10,8 @@
   long tailSize = 1024;
   int tailWindow = 1;
   boolean entireLog = false;
-  boolean plainText = false; 
+  boolean plainText = false;
+  TaskLog.LogFilter filter = null;
   
   String taskId = request.getParameter("taskid");
   if (taskId == null) {
@@ -18,6 +19,19 @@
   	return;
   }
   
+  String logFilter = request.getParameter("filter");
+  if (logFilter == null) {
+  	logFilter = "stderr";
+  }
+  
+  try {
+    filter = TaskLog.LogFilter.valueOf(TaskLog.LogFilter.class, 
+                                      logFilter.toUpperCase());
+  } catch (IllegalArgumentException iae) {
+    out.println("<h2>Illegal 'filter': " + logFilter + "</h2>");
+    return;
+  }
+  
   String sLogOff = request.getParameter("off");
   if (sLogOff != null) {
   	logOffset = Long.valueOf(sLogOff).longValue();
@@ -64,10 +78,10 @@
   
   if( !plainText ) {
     out.println("<html>");
-    out.println("<title>" + taskId + "Task Logs </title>"); 
+    out.println("<title>Task Logs: '" + taskId + "' (" + logFilter + ")</title>"); 
     out.println("<body>");
-    out.println("<h1>" +  taskId + "Task Logs</h1><br>"); 
-    out.println("<h2>Task Logs</h2>");
+    out.println("<h1>Task Logs from " +  taskId + "'s " + logFilter + "</h1><br>"); 
+    out.println("<h2>'" + logFilter + "':</h2>");
     out.println("<pre>");
 
   }
@@ -76,7 +90,7 @@
 <%
   boolean gotRequiredData = true;
   try {
-  	TaskLog.Reader taskLogReader = new TaskLog.Reader(taskId);
+  	TaskLog.Reader taskLogReader = new TaskLog.Reader(taskId, filter);
     byte[] b = null;
   	int bytesRead = 0;
   	int targetLength = 0;
@@ -113,7 +127,7 @@
 	String logData = new String(b, 0, bytesRead);
 	out.println(logData);
   } catch (IOException ioe) {
-  	out.println("Failed to retrieve logs for task: " + taskId);
+  	out.println("Failed to retrieve '" + logFilter + "' logs for task: " + taskId);
   }
   
   if( !plainText ) {
@@ -126,25 +140,29 @@
       if (gotRequiredData) {
   	  	out.println("<a href='/tasklog.jsp?taskid=" + taskId + 
   		    "&tail=true&tailsize=" + tailSize + "&tailwindow=" + (tailWindow+1) + 
-  		    "'>Earlier</a>");
+  		    "&filter=" + logFilter + "'>Earlier</a>");
   	  }
   	  if (tailWindow > 1) {
         out.println("<a href='/tasklog.jsp?taskid=" + taskId + 
   	  	    "&tail=true&tailsize=" + tailSize + "&tailwindow=" + (tailWindow-1) 
-  	  	    + "'>Later</a>");
+  	  	    + "&filter=" + logFilter + "'>Later</a>");
   	  }
     } else {
       if (gotRequiredData) {
       	out.println("<a href='/tasklog.jsp?taskid=" + taskId + 
     		"&tail=false&off=" + Math.max(0, (logOffset-logLength)) +
-  		  	"&len=" + logLength + "'>Earlier</a>");
+  		  	"&len=" + logLength + "&filter=" + logFilter + "'>Earlier</a>");
   	  }
   	  out.println("<a href='/tasklog.jsp?taskid=" + taskId + 
   		  "&tail=false&off=" + (logOffset+logLength) +
-  		  "&len=" + logLength + "'>Later</a>");
+  		  "&len=" + logLength + "&filter=" + logFilter + "'>Later</a>");
     }
   }
   if( !plainText ) {
+    String otherFilter = (logFilter.equals("stdout") ? "stderr" : "stdout");
+    out.println("<br><br>See <a href='/tasklog.jsp?taskid=" + taskId + 
+        "&all=true&filter=" + otherFilter + "'>" + otherFilter + "</a>" +
+        " logs of this task");
     out.println("<hr>");
     out.println("<a href='http://lucene.apache.org/hadoop'>Hadoop</a>, 2006.<br>");
     out.println("</body>");