HADOOP-1000. Fix so that log messages in task subprocesses are not written to a task's standard error. Contributed by Arun.

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

CHANGES.txt

@@ -171,6 +171,9 @@ Trunk (unreleased changes)
 51. HADOOP-941.  Enhance record facility.
     (Milind Bhandarkar via cutting)
 
+52. HADOOP-1000.  Fix so that log messages in task subprocesses are
+    not written to a task's standard error.  (Arun C Murthy via cutting)
+
 
 Release 0.11.2 - 2007-02-16
 

conf/log4j.properties

@@ -39,6 +39,27 @@ log4j.appender.console.target=System.err
 log4j.appender.console.layout=org.apache.log4j.PatternLayout
 log4j.appender.console.layout.ConversionPattern=%d{yy/MM/dd HH:mm:ss} %p %c{2}: %m%n
 
+#
+# TaskLog Appender
+#
+
+#Default values
+hadoop.tasklog.taskid=null
+hadoop.tasklog.noKeepSplits=4
+hadoop.tasklog.totalLogFileSize=100
+hadoop.tasklog.purgeLogSplits=true
+hadoop.tasklog.logsRetainHours=12
+
+log4j.appender.TLA=org.apache.hadoop.mapred.TaskLogAppender
+log4j.appender.TLA.taskId=${hadoop.tasklog.taskid}
+log4j.appender.TLA.noKeepSplits=${hadoop.tasklog.noKeepSplits}
+log4j.appender.TLA.totalLogFileSize=${hadoop.tasklog.totalLogFileSize}
+log4j.appender.TLA.purgeLogSplits=${hadoop.tasklog.purgeLogSplits}
+log4j.appender.TLA.logsRetainHours=${hadoop.tasklog.logsRetainHours}
+
+log4j.appender.TLA.layout=org.apache.log4j.PatternLayout
+log4j.appender.TLA.layout.ConversionPattern=%d{ISO8601} %p %c: %m%n
+
 #
 # Rolling File Appender
 #

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

@@ -45,6 +45,10 @@ class TaskLog {
     }
   }
 
+  private static File getTaskLogDir(String taskid, LogFilter filter) {
+    return new File(new File(LOG_DIR, taskid), filter.getPrefix());
+  }
+  
   /**
    * The filter for userlogs.
    */
@@ -53,7 +57,10 @@ class TaskLog {
     STDOUT ("stdout"),
 
     /** Log on the stderr of the task. */
-    STDERR ("stderr");
+    STDERR ("stderr"),
+    
+    /** Log on the map-reduce system logs of the task. */
+    SYSLOG ("syslog");
     
     private String prefix;
     
@@ -102,20 +109,17 @@ class TaskLog {
      * @param taskId taskid of the task
      * @param filter the {@link LogFilter} to apply on userlogs.
      */
-    Writer(JobConf conf, String taskId, LogFilter filter) {
-      this.conf = conf;
+    Writer(String taskId, LogFilter filter, 
+            int noKeepSplits, long totalLogSize, boolean purgeLogSplits, int logsRetainHours) {
       this.taskId = taskId;
       this.filter = filter;
       
-      this.taskLogDir = new File(new File(LOG_DIR, this.taskId), 
-                                this.filter.getPrefix());
+      this.taskLogDir = getTaskLogDir(this.taskId, this.filter);
       
-      noKeepSplits = this.conf.getInt("mapred.userlog.num.splits", 4);
-      splitFileSize = 
-        (this.conf.getInt("mapred.userlog.limit.kb", 100) * 1024) / noKeepSplits; 
-      purgeLogSplits = this.conf.getBoolean("mapred.userlog.purgesplits", 
-                                      true);
-      logsRetainHours = this.conf.getInt("mapred.userlog.retain.hours", 12);
+      this.noKeepSplits = noKeepSplits;
+      this.splitFileSize = (totalLogSize / noKeepSplits);
+      this.purgeLogSplits = purgeLogSplits;
+      this.logsRetainHours = logsRetainHours;
     }
     
     private static class TaskLogsPurgeFilter implements FileFilter {
@@ -220,7 +224,7 @@ class TaskLog {
      * 
      * @throws IOException
      */
-    public void close() throws IOException {
+    public synchronized void close() throws IOException {
       // Close the final split
       if (out != null) {
         out.close();
@@ -306,8 +310,7 @@ class TaskLog {
       this.taskId = taskId;
       this.filter = filter;
       
-      this.taskLogDir = new File(new File(LOG_DIR, this.taskId), 
-                                this.filter.getPrefix());
+      this.taskLogDir = getTaskLogDir(this.taskId, this.filter);
     }
 
     private static class IndexRecord {

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

@@ -0,0 +1,120 @@
+package org.apache.hadoop.mapred;
+
+import java.io.IOException;
+
+import org.apache.hadoop.util.StringUtils;
+import org.apache.log4j.AppenderSkeleton;
+import org.apache.log4j.spi.ErrorCode;
+import org.apache.log4j.spi.LoggingEvent;
+
+/**
+ * A simple log4j-appender for the task child's 
+ * map-reduce system logs.
+ * 
+ * @author Arun C Murthy
+ */
+public class TaskLogAppender extends AppenderSkeleton {
+  private TaskLog.Writer taskLogWriter = null;
+  private String taskId;
+  private int noKeepSplits;
+  private long totalLogFileSize;
+  private boolean purgeLogSplits;
+  private int logsRetainHours;
+
+  public void activateOptions() {
+    taskLogWriter = 
+      new TaskLog.Writer(taskId, TaskLog.LogFilter.SYSLOG, 
+              noKeepSplits, totalLogFileSize, purgeLogSplits, logsRetainHours);
+    try {
+      taskLogWriter.init();
+    } catch (IOException ioe) {
+      taskLogWriter = null;
+      errorHandler.error("Failed to initialize the task's logging " +
+              "infrastructure: " + StringUtils.stringifyException(ioe));
+    }
+  }
+  
+  protected synchronized void append(LoggingEvent event) {
+    if (taskLogWriter == null) {
+      errorHandler.error("Calling 'append' on uninitialize/closed logger");
+      return;
+    }
+
+    if (this.layout == null) {
+      errorHandler.error("No layout for appender " + name , 
+              null, ErrorCode.MISSING_LAYOUT );
+    }
+    
+    // Log the message to the task's log
+    String logMessage = this.layout.format(event);
+    try {
+      taskLogWriter.write(logMessage.getBytes(), 0, logMessage.length());
+    } catch (IOException ioe) {
+      errorHandler.error("Failed to log: '" + logMessage + 
+              "' to the task's logging infrastructure with the exception: " + 
+              StringUtils.stringifyException(ioe));
+    }
+  }
+
+  public boolean requiresLayout() {
+    return true;
+  }
+
+  public synchronized void close() {
+    if (taskLogWriter != null) {
+      try {
+        taskLogWriter.close();
+      } catch (IOException ioe) {
+        errorHandler.error("Failed to close the task's log with the exception: " 
+                + StringUtils.stringifyException(ioe));
+      }
+    } else {
+      errorHandler.error("Calling 'close' on uninitialize/closed logger");
+    }
+  }
+
+  /**
+   * Getter/Setter methods for log4j.
+   */
+  
+  public String getTaskId() {
+    return taskId;
+  }
+
+  public void setTaskId(String taskId) {
+    this.taskId = taskId;
+  }
+
+  public int getNoKeepSplits() {
+    return noKeepSplits;
+  }
+
+  public void setNoKeepSplits(int noKeepSplits) {
+    this.noKeepSplits = noKeepSplits;
+  }
+
+  public int getLogsRetainHours() {
+    return logsRetainHours;
+  }
+
+  public void setLogsRetainHours(int logsRetainHours) {
+    this.logsRetainHours = logsRetainHours;
+  }
+
+  public boolean isPurgeLogSplits() {
+    return purgeLogSplits;
+  }
+
+  public void setPurgeLogSplits(boolean purgeLogSplits) {
+    this.purgeLogSplits = purgeLogSplits;
+  }
+
+  public long getTotalLogFileSize() {
+    return totalLogFileSize;
+  }
+
+  public void setTotalLogFileSize(long splitFileSize) {
+    this.totalLogFileSize = splitFileSize;
+  }
+
+}

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

@@ -49,9 +49,17 @@ abstract class TaskRunner extends Thread {
     this.tracker = tracker;
     this.conf = conf;
     this.taskStdOutLogWriter = 
-      new TaskLog.Writer(conf, t.getTaskId(), TaskLog.LogFilter.STDOUT);
+      new TaskLog.Writer(t.getTaskId(), TaskLog.LogFilter.STDOUT, 
+              this.conf.getInt("mapred.userlog.num.splits", 4), 
+              this.conf.getInt("mapred.userlog.limit.kb", 100) * 1024, 
+              this.conf.getBoolean("mapred.userlog.purgesplits", true),
+              this.conf.getInt("mapred.userlog.retain.hours", 12));
     this.taskStdErrLogWriter = 
-      new TaskLog.Writer(conf, t.getTaskId(), TaskLog.LogFilter.STDERR);
+      new TaskLog.Writer(t.getTaskId(), TaskLog.LogFilter.STDERR, 
+              this.conf.getInt("mapred.userlog.num.splits", 4), 
+              this.conf.getInt("mapred.userlog.limit.kb", 100) * 1024, 
+              this.conf.getBoolean("mapred.userlog.purgesplits", true),
+              this.conf.getInt("mapred.userlog.retain.hours", 12));
   }
 
   public Task getTask() { return t; }
@@ -166,7 +174,7 @@ abstract class TaskRunner extends Thread {
       classPath.append(sep);
       classPath.append(workDir);
       //  Build exec child jmv args.
-      Vector vargs = new Vector(8);
+      Vector<String> vargs = new Vector<String>(8);
       File jvm =                                  // use same jvm as parent
         new File(new File(System.getProperty("java.home"), "bin"), "java");
 
@@ -210,6 +218,15 @@ abstract class TaskRunner extends Thread {
       vargs.add("-classpath");
       vargs.add(classPath.toString());
 
+      // Setup the log4j prop
+      vargs.add("-Dhadoop.log.dir=" + System.getProperty("hadoop.log.dir"));
+      vargs.add("-Dhadoop.root.logger=INFO,TLA");
+      vargs.add("-Dhadoop.tasklog.taskid=" + t.getTaskId());
+      vargs.add("-Dhadoop.tasklog.noKeepSplits=" + conf.getInt("mapred.userlog.num.splits", 4)); 
+      vargs.add("-Dhadoop.tasklog.totalLogFileSize=" + (conf.getInt("mapred.userlog.limit.kb", 100) * 1024));
+      vargs.add("-Dhadoop.tasklog.purgeLogSplits=" + conf.getBoolean("mapred.userlog.purgesplits", true));
+      vargs.add("-Dhadoop.tasklog.logsRetainHours=" + conf.getInt("mapred.userlog.retain.hours", 12)); 
+
       // Add java.library.path; necessary for native-hadoop libraries
       String libraryPath = System.getProperty("java.library.path");
       if (libraryPath != null) {

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

@@ -63,6 +63,7 @@ import org.apache.hadoop.util.ReflectionUtils;
 import org.apache.hadoop.util.RunJar;
 import org.apache.hadoop.util.StringUtils;
 import org.apache.hadoop.util.DiskChecker.DiskErrorException;
+import org.apache.log4j.LogManager;
 
 /*******************************************************
  * TaskTracker is a process that starts and tracks MR Tasks
@@ -1451,6 +1452,11 @@ public class TaskTracker
               ByteArrayOutputStream baos = new ByteArrayOutputStream();
               throwable.printStackTrace(new PrintStream(baos));
               umbilical.reportDiagnosticInfo(taskid, baos.toString());
+          } finally {
+            // Shutting down log4j of the child-vm... 
+            // This assumes that on return from Task.run() 
+            // there is no more logging done.
+            LogManager.shutdown();
           }
         }
 

src/webapps/task/tasklog.jsp

@@ -4,7 +4,8 @@
   import="javax.servlet.http.*"
   import="java.io.*"
 %>
-<%
+<%!
+  String taskId = null;
   long logOffset = -1, logLength = -1;
   boolean tailLog = false;
   long tailSize = 1024;
@@ -12,8 +13,93 @@
   boolean entireLog = false;
   boolean plainText = false;
   TaskLog.LogFilter filter = null;
+
+  private void printTaskLog(JspWriter out, TaskLog.LogFilter filter) 
+  throws IOException {
+    if (!plainText) {
+      out.println("<br><b><u>" + filter + " logs</u></b><br>");
+      out.println("<table border=2 cellpadding=\"2\">");
+    }
+    
+    boolean gotRequiredData = true;
+    try {
+  	  TaskLog.Reader taskLogReader = new TaskLog.Reader(taskId, filter);
+      byte[] b = null;
+  	  int bytesRead = 0;
+  	  int targetLength = 0;
+
+  	  if (entireLog) {
+  	    b = taskLogReader.fetchAll();
+  	    targetLength = bytesRead = b.length;
+  	  } else {
+  	    if (tailLog) {
+  		  b = new byte[(int)tailSize];
+  		  targetLength = (int)tailSize;
+  		  bytesRead = taskLogReader.tail(b, 0, b.length, tailSize, tailWindow);
+  	    } else {
+  		  b = new byte[(int)logLength];
+  		  targetLength = (int)logLength;
+  		  bytesRead = taskLogReader.read(b, 0, b.length, logOffset, logLength);
+    	}
+  	  }
+  	  
+  	  if (bytesRead != targetLength && 
+  	    targetLength <= taskLogReader.getTotalLogSize()) {
+  	    if( !plainText) {
+	  	  out.println("<b>Warning: Could not fetch " + targetLength + 
+	  		  " bytes from the task-logs; probably purged!</b><br/>");
+  	    }else{
+	  	  out.println("Warning: Could not fetch " + targetLength + 
+  		  " bytes from the task-logs; probably purged!");
+  	    }
+  	    gotRequiredData = false;
+  	  }
+	  String logData = new String(b, 0, bytesRead);
+	  if (!plainText) {
+	    out.print("<tr><td><pre>" + logData + "</pre></td></tr>");
+	  } else {
+	    out.print(logData);
+	  }
+    } catch (IOException ioe) {
+  	  out.println("Failed to retrieve '" + filter + "' logs for task: " + taskId);
+    }
   
-  String taskId = request.getParameter("taskid");
+    if( !plainText ) {
+     out.println("</table>\n");
+    } 
+
+    if (!entireLog && !plainText) {
+      if (tailLog) {
+        if (gotRequiredData) {
+  	  	  out.println("<a href='/tasklog.jsp?taskid=" + taskId + 
+  		    "&tail=true&tailsize=" + tailSize + "&tailwindow=" + (tailWindow+1) + 
+  		    "&filter=" + filter + "'>Earlier</a>");
+  	    }
+  	    if (tailWindow > 1) {
+          out.println("<a href='/tasklog.jsp?taskid=" + taskId + 
+  	  	    "&tail=true&tailsize=" + tailSize + "&tailwindow=" + (tailWindow-1) 
+  	  	    + "&filter=" + filter + "'>Later</a>");
+  	    }
+      } else {
+        if (gotRequiredData) {
+      	  out.println("<a href='/tasklog.jsp?taskid=" + taskId + 
+    		"&tail=false&off=" + Math.max(0, (logOffset-logLength)) +
+  		  	"&len=" + logLength + "&filter=" + filter + "'>Earlier</a>");
+  	    }
+  	    out.println("<a href='/tasklog.jsp?taskid=" + taskId + 
+  		  "&tail=false&off=" + (logOffset+logLength) +
+  		  "&len=" + logLength + "&filter=" + filter + "'>Later</a>");
+      }
+    }
+    
+    if (!plainText) {
+      out.println("<hr><br>");
+    }
+  }
+%>
+
+<%  
+  taskId = request.getParameter("taskid");
   if (taskId == null) {
   	out.println("<h2>Missing 'taskid' for fetching logs!</h2>");
   	return;
@@ -78,91 +164,18 @@
   
   if( !plainText ) {
     out.println("<html>");
-    out.println("<title>Task Logs: '" + taskId + "' (" + logFilter + ")</title>"); 
+    out.println("<title>Task Logs: '" + taskId + "'</title>"); 
     out.println("<body>");
-    out.println("<h1>Task Logs from " +  taskId + "'s " + logFilter + "</h1><br>"); 
-    out.println("<h2>'" + logFilter + "':</h2>");
-    out.println("<pre>");
-
-  }
-%>
-
-<%
-  boolean gotRequiredData = true;
-  try {
-  	TaskLog.Reader taskLogReader = new TaskLog.Reader(taskId, filter);
-    byte[] b = null;
-  	int bytesRead = 0;
-  	int targetLength = 0;
-
-  	if (entireLog) {
-  	  b = taskLogReader.fetchAll();
-  	  targetLength = bytesRead = b.length;
-  	} else {
-  	  if (tailLog) {
-  		b = new byte[(int)tailSize];
-  		targetLength = (int)tailSize;
-  		bytesRead = taskLogReader.tail(b, 0, b.length, tailSize, tailWindow);
-  	  } else {
-  		b = new byte[(int)logLength];
-  		targetLength = (int)logLength;
-  		bytesRead = taskLogReader.read(b, 0, b.length, logOffset, logLength);
-   	  }
-  	}
-  	  
-  	if (bytesRead != targetLength && 
-  	  targetLength <= taskLogReader.getTotalLogSize()) {
-  	  if( !plainText) {
-	  	  out.println("<b>Warning: Could not fetch " + targetLength + 
-	  		  " bytes from the task-logs; probably purged!</b><br/>");
-  	  }else{
-	  	  out.println("Warning: Could not fetch " + targetLength + 
-  		  " bytes from the task-logs; probably purged!");
-  	  }
-  	  gotRequiredData = false;
-  	}
-	String logData = new String(b, 0, bytesRead);
-	out.println(logData);
-  } catch (IOException ioe) {
-  	out.println("Failed to retrieve '" + logFilter + "' logs for task: " + taskId);
-  }
-  
-  if( !plainText ) {
-    out.println("</pre>");
-  }
-%>
-<%
-  if (!entireLog && !plainText) {
-    if (tailLog) {
-      if (gotRequiredData) {
-  	  	out.println("<a href='/tasklog.jsp?taskid=" + taskId + 
-  		    "&tail=true&tailsize=" + tailSize + "&tailwindow=" + (tailWindow+1) + 
-  		    "&filter=" + logFilter + "'>Earlier</a>");
-  	  }
-  	  if (tailWindow > 1) {
-        out.println("<a href='/tasklog.jsp?taskid=" + taskId + 
-  	  	    "&tail=true&tailsize=" + tailSize + "&tailwindow=" + (tailWindow-1) 
-  	  	    + "&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 + "&filter=" + logFilter + "'>Earlier</a>");
-  	  }
-  	  out.println("<a href='/tasklog.jsp?taskid=" + taskId + 
-  		  "&tail=false&off=" + (logOffset+logLength) +
-  		  "&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("<h1>Task Logs: '" +  taskId +  "'</h1><br>"); 
+    
+    printTaskLog(out, TaskLog.LogFilter.STDOUT);
+    printTaskLog(out, TaskLog.LogFilter.STDERR);
+    printTaskLog(out, TaskLog.LogFilter.SYSLOG);
+    
     out.println("<a href='http://lucene.apache.org/hadoop'>Hadoop</a>, 2006.<br>");
     out.println("</body>");
     out.println("</html>");
-  }
+  } else {
+    printTaskLog(out, filter);
+  } 
 %>