MAPREDUCE-4006. History server container log web UI sometimes combines stderr/stdout/syslog contents together (Siddharth Seth via tgraves)

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

hadoop-mapreduce-project/CHANGES.txt

@@ -328,6 +328,9 @@ Release 0.23.2 - UNRELEASED
     MAPREDUCE-4005. Fixes broken AM container logs URL on ResourceManager
     Application Info page. (Jason Lowe via sseth)
 
+    MAPREDUCE-4006. History server container log web UI sometimes combines
+    stderr/stdout/syslog contents together (Siddharth Seth via tgraves)
+
 Release 0.23.1 - 2012-02-17
 
   INCOMPATIBLE CHANGES

hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/logaggregation/AggregatedLogFormat.java

@@ -430,7 +430,7 @@ public static void readAcontainerLogs(DataInputStream valueStream,
         int currentRead = 0;
         int totalRead = 0;
         while ((currentRead = reader.read(cbuf, 0, bufferSize)) != -1) {
-          writer.write(cbuf);
+          writer.write(cbuf, 0, currentRead);
           totalRead += currentRead;
         }
       }

hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-common/src/test/java/org/apache/hadoop/yarn/logaggregation/TestAggregatedLogFormat.java

@@ -0,0 +1,144 @@
+/**
+ * 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.yarn.logaggregation;
+
+import java.io.DataInputStream;
+import java.io.File;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.OutputStreamWriter;
+import java.io.StringWriter;
+import java.io.Writer;
+import java.util.Collections;
+
+import junit.framework.Assert;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hadoop.yarn.api.records.ContainerId;
+import org.apache.hadoop.yarn.logaggregation.AggregatedLogFormat.LogKey;
+import org.apache.hadoop.yarn.logaggregation.AggregatedLogFormat.LogReader;
+import org.apache.hadoop.yarn.logaggregation.AggregatedLogFormat.LogValue;
+import org.apache.hadoop.yarn.logaggregation.AggregatedLogFormat.LogWriter;
+import org.apache.hadoop.yarn.util.BuilderUtils;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+public class TestAggregatedLogFormat {
+
+  private static final File testWorkDir = new File("target",
+      "TestAggregatedLogFormat");
+  private static final Configuration conf = new Configuration();
+  private static final FileSystem fs;
+  private static final char filler = 'x';
+  private static final Log LOG = LogFactory
+      .getLog(TestAggregatedLogFormat.class);
+
+  static {
+    try {
+      fs = FileSystem.get(conf);
+    } catch (IOException e) {
+      throw new RuntimeException(e);
+    }
+  }
+
+  @Before
+  @After
+  public void cleanupTestDir() throws Exception {
+    Path workDirPath = new Path(testWorkDir.getAbsolutePath());
+    LOG.info("Cleaning test directory [" + workDirPath + "]");
+    fs.delete(workDirPath, true);
+  }
+
+  //Verify the output generated by readAContainerLogs(DataInputStream, Writer)
+  @Test
+  public void testReadAcontainerLogs1() throws Exception {
+    Configuration conf = new Configuration();
+    File workDir = new File(testWorkDir, "testReadAcontainerLogs1");
+    Path remoteAppLogFile =
+        new Path(workDir.getAbsolutePath(), "aggregatedLogFile");
+    Path srcFileRoot = new Path(workDir.getAbsolutePath(), "srcFiles");
+    ContainerId testContainerId = BuilderUtils.newContainerId(1, 1, 1, 1);
+    Path t =
+        new Path(srcFileRoot, testContainerId.getApplicationAttemptId()
+            .getApplicationId().toString());
+    Path srcFilePath = new Path(t, testContainerId.toString());
+
+    int numChars = 80000;
+
+    writeSrcFile(srcFilePath, "stdout", numChars);
+
+    UserGroupInformation ugi = UserGroupInformation.getCurrentUser();
+    LogWriter logWriter = new LogWriter(conf, remoteAppLogFile, ugi);
+
+    LogKey logKey = new LogKey(testContainerId);
+    LogValue logValue =
+        new LogValue(Collections.singletonList(srcFileRoot.toString()),
+            testContainerId);
+
+    logWriter.append(logKey, logValue);
+    logWriter.closeWriter();
+
+    LogReader logReader = new LogReader(conf, remoteAppLogFile);
+    LogKey rLogKey = new LogKey();
+    DataInputStream dis = logReader.next(rLogKey);
+    Writer writer = new StringWriter();
+    LogReader.readAcontainerLogs(dis, writer);
+    
+    String s = writer.toString();
+    int expectedLength =
+        "\n\nLogType:stdout".length() + ("\nLogLength:" + numChars).length()
+            + "\nLog Contents:\n".length() + numChars;
+    Assert.assertTrue("LogType not matched", s.contains("LogType:stdout"));
+    Assert.assertTrue("LogLength not matched", s.contains("LogLength:" + numChars));
+    Assert.assertTrue("Log Contents not matched", s.contains("Log Contents"));
+    
+    StringBuilder sb = new StringBuilder();
+    for (int i = 0 ; i < numChars ; i++) {
+      sb.append(filler);
+    }
+    String expectedContent = sb.toString();
+    Assert.assertTrue("Log content incorrect", s.contains(expectedContent));
+    
+    Assert.assertEquals(expectedLength, s.length());
+  }
+  
+  private void writeSrcFile(Path srcFilePath, String fileName, long length)
+      throws IOException {
+    File dir = new File(srcFilePath.toString());
+    if (!dir.exists()) {
+      if (!dir.mkdirs()) {
+        throw new IOException("Unable to create directory : " + dir);
+      }
+    }
+    File outputFile = new File(new File(srcFilePath.toString()), fileName);
+    FileOutputStream os = new FileOutputStream(outputFile);
+    OutputStreamWriter osw = new OutputStreamWriter(os, "UTF8");
+    int ch = filler;
+    for (int i = 0; i < length; i++) {
+      osw.write(ch);
+    }
+    osw.close();
+  }
+}