hadoop-mapreduce-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From a..@apache.org
Subject svn commit: r1196458 [13/19] - in /hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project: ./ assembly/ bin/ conf/ dev-support/ hadoop-mapreduce-client/ hadoop-mapreduce-client/hadoop-mapreduce-client-app/ hadoop-mapreduce-client/hadoop-mapreduce-cl...
Date Wed, 02 Nov 2011 05:35:03 GMT
Modified: hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/logaggregation/AggregatedLogFormat.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/logaggregation/AggregatedLogFormat.java?rev=1196458&r1=1196457&r2=1196458&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/logaggregation/AggregatedLogFormat.java (original)
+++ hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/logaggregation/AggregatedLogFormat.java Wed Nov  2 05:34:31 2011
@@ -25,10 +25,16 @@ import java.io.DataOutputStream;
 import java.io.EOFException;
 import java.io.File;
 import java.io.FileInputStream;
+import java.io.InputStreamReader;
 import java.io.IOException;
+import java.io.Writer;
 import java.security.PrivilegedExceptionAction;
 import java.util.EnumSet;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Map.Entry;
 
+import org.apache.commons.io.input.BoundedInputStream;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.conf.Configuration;
@@ -41,6 +47,8 @@ import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.io.Writable;
 import org.apache.hadoop.io.file.tfile.TFile;
 import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hadoop.yarn.YarnException;
+import org.apache.hadoop.yarn.api.records.ApplicationAccessType;
 import org.apache.hadoop.yarn.api.records.ContainerId;
 import org.apache.hadoop.yarn.conf.YarnConfiguration;
 import org.apache.hadoop.yarn.util.ConverterUtils;
@@ -48,32 +56,50 @@ import org.apache.hadoop.yarn.util.Conve
 public class AggregatedLogFormat {
 
   static final Log LOG = LogFactory.getLog(AggregatedLogFormat.class);
-
+  private static final LogKey APPLICATION_ACL_KEY = new LogKey("APPLICATION_ACL");
+  private static final LogKey APPLICATION_OWNER_KEY = new LogKey("APPLICATION_OWNER");
+  private static final LogKey VERSION_KEY = new LogKey("VERSION");
+  private static final Map<String, LogKey> RESERVED_KEYS;
+  //Maybe write out the retention policy.
+  //Maybe write out a list of containerLogs skipped by the retention policy.
+  private static final int VERSION = 1;
+
+  static {
+    RESERVED_KEYS = new HashMap<String, AggregatedLogFormat.LogKey>();
+    RESERVED_KEYS.put(APPLICATION_ACL_KEY.toString(), APPLICATION_ACL_KEY);
+    RESERVED_KEYS.put(APPLICATION_OWNER_KEY.toString(), APPLICATION_OWNER_KEY);
+    RESERVED_KEYS.put(VERSION_KEY.toString(), VERSION_KEY);
+  }
+  
   public static class LogKey implements Writable {
 
-    private String containerId;
+    private String keyString;
 
     public LogKey() {
 
     }
 
     public LogKey(ContainerId containerId) {
-      this.containerId = ConverterUtils.toString(containerId);
+      this.keyString = containerId.toString();
     }
 
+    public LogKey(String keyString) {
+      this.keyString = keyString;
+    }
+    
     @Override
     public void write(DataOutput out) throws IOException {
-      out.writeUTF(this.containerId);
+      out.writeUTF(this.keyString);
     }
 
     @Override
     public void readFields(DataInput in) throws IOException {
-      this.containerId = in.readUTF();
+      this.keyString = in.readUTF();
     }
 
     @Override
     public String toString() {
-      return this.containerId;
+      return this.keyString;
     }
   }
 
@@ -81,6 +107,8 @@ public class AggregatedLogFormat {
 
     private final String[] rootLogDirs;
     private final ContainerId containerId;
+    // TODO Maybe add a version string here. Instead of changing the version of
+    // the entire k-v format
 
     public LogValue(String[] rootLogDirs, ContainerId containerId) {
       this.rootLogDirs = rootLogDirs;
@@ -141,7 +169,8 @@ public class AggregatedLogFormat {
               public FSDataOutputStream run() throws Exception {
                 return FileContext.getFileContext(conf).create(
                     remoteAppLogFile,
-                    EnumSet.of(CreateFlag.CREATE), new Options.CreateOpts[] {});
+                    EnumSet.of(CreateFlag.CREATE, CreateFlag.OVERWRITE),
+                    new Options.CreateOpts[] {});
               }
             });
       } catch (InterruptedException e) {
@@ -154,6 +183,40 @@ public class AggregatedLogFormat {
           new TFile.Writer(this.fsDataOStream, 256 * 1024, conf.get(
               YarnConfiguration.NM_LOG_AGG_COMPRESSION_TYPE,
               YarnConfiguration.DEFAULT_NM_LOG_AGG_COMPRESSION_TYPE), null, conf);
+      //Write the version string
+      writeVersion();
+    }
+
+    private void writeVersion() throws IOException {
+      DataOutputStream out = this.writer.prepareAppendKey(-1);
+      VERSION_KEY.write(out);
+      out.close();
+      out = this.writer.prepareAppendValue(-1);
+      out.writeInt(VERSION);
+      out.close();
+      this.fsDataOStream.hflush();
+    }
+
+    public void writeApplicationOwner(String user) throws IOException {
+      DataOutputStream out = this.writer.prepareAppendKey(-1);
+      APPLICATION_OWNER_KEY.write(out);
+      out.close();
+      out = this.writer.prepareAppendValue(-1);
+      out.writeUTF(user);
+      out.close();
+    }
+
+    public void writeApplicationACLs(Map<ApplicationAccessType, String> appAcls)
+        throws IOException {
+      DataOutputStream out = this.writer.prepareAppendKey(-1);
+      APPLICATION_ACL_KEY.write(out);
+      out.close();
+      out = this.writer.prepareAppendValue(-1);
+      for (Entry<ApplicationAccessType, String> entry : appAcls.entrySet()) {
+        out.writeUTF(entry.getKey().toString());
+        out.writeUTF(entry.getValue());
+      }
+      out.close();
     }
 
     public void append(LogKey logKey, LogValue logValue) throws IOException {
@@ -184,12 +247,13 @@ public class AggregatedLogFormat {
 
     private final FSDataInputStream fsDataIStream;
     private final TFile.Reader.Scanner scanner;
+    private final TFile.Reader reader;
 
     public LogReader(Configuration conf, Path remoteAppLogFile)
         throws IOException {
       FileContext fileContext = FileContext.getFileContext(conf);
       this.fsDataIStream = fileContext.open(remoteAppLogFile);
-      TFile.Reader reader =
+      reader =
           new TFile.Reader(this.fsDataIStream, fileContext.getFileStatus(
               remoteAppLogFile).getLen(), conf);
       this.scanner = reader.createScanner();
@@ -198,6 +262,69 @@ public class AggregatedLogFormat {
     private boolean atBeginning = true;
 
     /**
+     * Returns the owner of the application.
+     * 
+     * @return the application owner.
+     * @throws IOException
+     */
+    public String getApplicationOwner() throws IOException {
+      TFile.Reader.Scanner ownerScanner = reader.createScanner();
+      LogKey key = new LogKey();
+      while (!ownerScanner.atEnd()) {
+        TFile.Reader.Scanner.Entry entry = ownerScanner.entry();
+        key.readFields(entry.getKeyStream());
+        if (key.toString().equals(APPLICATION_OWNER_KEY.toString())) {
+          DataInputStream valueStream = entry.getValueStream();
+          return valueStream.readUTF();
+        }
+        ownerScanner.advance();
+      }
+      return null;
+    }
+
+    /**
+     * Returns ACLs for the application. An empty map is returned if no ACLs are
+     * found.
+     * 
+     * @return a map of the Application ACLs.
+     * @throws IOException
+     */
+    public Map<ApplicationAccessType, String> getApplicationAcls()
+        throws IOException {
+      // TODO Seek directly to the key once a comparator is specified.
+      TFile.Reader.Scanner aclScanner = reader.createScanner();
+      LogKey key = new LogKey();
+      Map<ApplicationAccessType, String> acls =
+          new HashMap<ApplicationAccessType, String>();
+      while (!aclScanner.atEnd()) {
+        TFile.Reader.Scanner.Entry entry = aclScanner.entry();
+        key.readFields(entry.getKeyStream());
+        if (key.toString().equals(APPLICATION_ACL_KEY.toString())) {
+          DataInputStream valueStream = entry.getValueStream();
+          while (true) {
+            String appAccessOp = null;
+            String aclString = null;
+            try {
+              appAccessOp = valueStream.readUTF();
+            } catch (EOFException e) {
+              // Valid end of stream.
+              break;
+            }
+            try {
+              aclString = valueStream.readUTF();
+            } catch (EOFException e) {
+              throw new YarnException("Error reading ACLs", e);
+            }
+            acls.put(ApplicationAccessType.valueOf(appAccessOp), aclString);
+          }
+
+        }
+        aclScanner.advance();
+      }
+      return acls;
+    }
+    
+    /**
      * Read the next key and return the value-stream.
      * 
      * @param key
@@ -215,10 +342,99 @@ public class AggregatedLogFormat {
       }
       TFile.Reader.Scanner.Entry entry = this.scanner.entry();
       key.readFields(entry.getKeyStream());
+      // Skip META keys
+      if (RESERVED_KEYS.containsKey(key.toString())) {
+        return next(key);
+      }
       DataInputStream valueStream = entry.getValueStream();
       return valueStream;
     }
 
+    
+    //TODO  Change Log format and interfaces to be containerId specific.
+    // Avoid returning completeValueStreams.
+//    public List<String> getTypesForContainer(DataInputStream valueStream){}
+//    
+//    /**
+//     * @param valueStream
+//     *          The Log stream for the container.
+//     * @param fileType
+//     *          the log type required.
+//     * @return An InputStreamReader for the required log type or null if the
+//     *         type is not found.
+//     * @throws IOException
+//     */
+//    public InputStreamReader getLogStreamForType(DataInputStream valueStream,
+//        String fileType) throws IOException {
+//      valueStream.reset();
+//      try {
+//        while (true) {
+//          String ft = valueStream.readUTF();
+//          String fileLengthStr = valueStream.readUTF();
+//          long fileLength = Long.parseLong(fileLengthStr);
+//          if (ft.equals(fileType)) {
+//            BoundedInputStream bis =
+//                new BoundedInputStream(valueStream, fileLength);
+//            return new InputStreamReader(bis);
+//          } else {
+//            long totalSkipped = 0;
+//            long currSkipped = 0;
+//            while (currSkipped != -1 && totalSkipped < fileLength) {
+//              currSkipped = valueStream.skip(fileLength - totalSkipped);
+//              totalSkipped += currSkipped;
+//            }
+//            // TODO Verify skip behaviour.
+//            if (currSkipped == -1) {
+//              return null;
+//            }
+//          }
+//        }
+//      } catch (EOFException e) {
+//        return null;
+//      }
+//    }
+
+    /**
+     * Writes all logs for a single container to the provided writer.
+     * @param valueStream
+     * @param writer
+     * @throws IOException
+     */
+    public static void readAcontainerLogs(DataInputStream valueStream,
+        Writer writer) throws IOException {
+      int bufferSize = 65536;
+      char[] cbuf = new char[bufferSize];
+      String fileType;
+      String fileLengthStr;
+      long fileLength;
+
+      while (true) {
+        try {
+          fileType = valueStream.readUTF();
+        } catch (EOFException e) {
+          // EndOfFile
+          return;
+        }
+        fileLengthStr = valueStream.readUTF();
+        fileLength = Long.parseLong(fileLengthStr);
+        writer.write("\n\nLogType:");
+        writer.write(fileType);
+        writer.write("\nLogLength:");
+        writer.write(fileLengthStr);
+        writer.write("\nLog Contents:\n");
+        // ByteLevel
+        BoundedInputStream bis =
+            new BoundedInputStream(valueStream, fileLength);
+        InputStreamReader reader = new InputStreamReader(bis);
+        int currentRead = 0;
+        int totalRead = 0;
+        while ((currentRead = reader.read(cbuf, 0, bufferSize)) != -1) {
+          writer.write(cbuf);
+          totalRead += currentRead;
+        }
+      }
+    }
+
     /**
      * Keep calling this till you get a {@link EOFException} for getting logs of
      * all types for a single container.

Modified: hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/logaggregation/AppLogAggregatorImpl.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/logaggregation/AppLogAggregatorImpl.java?rev=1196458&r1=1196457&r2=1196458&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/logaggregation/AppLogAggregatorImpl.java (original)
+++ hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/logaggregation/AppLogAggregatorImpl.java Wed Nov  2 05:34:31 2011
@@ -18,8 +18,9 @@
 
 package org.apache.hadoop.yarn.server.nodemanager.containermanager.logaggregation;
 
-import java.io.File;
 import java.io.IOException;
+import java.security.PrivilegedExceptionAction;
+import java.util.Map;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -27,11 +28,16 @@ import java.util.concurrent.atomic.Atomi
 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.ApplicationId;
+import org.apache.hadoop.yarn.api.records.ApplicationAccessType;
 import org.apache.hadoop.yarn.api.records.ContainerId;
+import org.apache.hadoop.yarn.event.Dispatcher;
 import org.apache.hadoop.yarn.server.nodemanager.DeletionService;
+import org.apache.hadoop.yarn.server.nodemanager.containermanager.application.ApplicationEvent;
+import org.apache.hadoop.yarn.server.nodemanager.containermanager.application.ApplicationEventType;
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.logaggregation.AggregatedLogFormat.LogKey;
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.logaggregation.AggregatedLogFormat.LogValue;
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.logaggregation.AggregatedLogFormat.LogWriter;
@@ -42,7 +48,10 @@ public class AppLogAggregatorImpl implem
   private static final Log LOG = LogFactory
       .getLog(AppLogAggregatorImpl.class);
   private static final int THREAD_SLEEP_TIME = 1000;
+  private static final String TMP_FILE_SUFFIX = ".tmp";
 
+  private final Dispatcher dispatcher;
+  private final ApplicationId appId;
   private final String applicationId;
   private boolean logAggregationDisabled = false;
   private final Configuration conf;
@@ -50,26 +59,34 @@ public class AppLogAggregatorImpl implem
   private final UserGroupInformation userUgi;
   private final String[] rootLogDirs;
   private final Path remoteNodeLogFileForApp;
+  private final Path remoteNodeTmpLogFileForApp;
   private final ContainerLogsRetentionPolicy retentionPolicy;
 
   private final BlockingQueue<ContainerId> pendingContainers;
   private final AtomicBoolean appFinishing = new AtomicBoolean();
   private final AtomicBoolean appAggregationFinished = new AtomicBoolean();
+  private final Map<ApplicationAccessType, String> appAcls;
 
   private LogWriter writer = null;
 
-  public AppLogAggregatorImpl(DeletionService deletionService,
-      Configuration conf, ApplicationId appId, UserGroupInformation userUgi,
-      String[] localRootLogDirs, Path remoteNodeLogFileForApp,
-      ContainerLogsRetentionPolicy retentionPolicy) {
+  public AppLogAggregatorImpl(Dispatcher dispatcher,
+      DeletionService deletionService, Configuration conf, ApplicationId appId,
+      UserGroupInformation userUgi, String[] localRootLogDirs,
+      Path remoteNodeLogFileForApp,
+      ContainerLogsRetentionPolicy retentionPolicy,
+      Map<ApplicationAccessType, String> appAcls) {
+    this.dispatcher = dispatcher;
     this.conf = conf;
     this.delService = deletionService;
+    this.appId = appId;
     this.applicationId = ConverterUtils.toString(appId);
     this.userUgi = userUgi;
     this.rootLogDirs = localRootLogDirs;
     this.remoteNodeLogFileForApp = remoteNodeLogFileForApp;
+    this.remoteNodeTmpLogFileForApp = getRemoteNodeTmpLogFileForApp();
     this.retentionPolicy = retentionPolicy;
     this.pendingContainers = new LinkedBlockingQueue<ContainerId>();
+    this.appAcls = appAcls;
   }
 
   private void uploadLogsForContainer(ContainerId containerId) {
@@ -80,11 +97,15 @@ public class AppLogAggregatorImpl implem
 
     // Lazy creation of the writer
     if (this.writer == null) {
-      LOG.info("Starting aggregate log-file for app " + this.applicationId);
+      LOG.info("Starting aggregate log-file for app " + this.applicationId
+          + " at " + this.remoteNodeTmpLogFileForApp);
       try {
         this.writer =
-            new LogWriter(this.conf, this.remoteNodeLogFileForApp,
+            new LogWriter(this.conf, this.remoteNodeTmpLogFileForApp,
                 this.userUgi);
+        //Write ACLs once when and if the writer is created.
+        this.writer.writeApplicationACLs(appAcls);
+        this.writer.writeApplicationOwner(this.userUgi.getShortUserName());
       } catch (IOException e) {
         LOG.error("Cannot create writer for app " + this.applicationId
             + ". Disabling log-aggregation for this app.", e);
@@ -105,8 +126,8 @@ public class AppLogAggregatorImpl implem
   }
 
   @Override
-  public void run() {
-
+  @SuppressWarnings("unchecked")
+  public void run() {    
     ContainerId containerId;
 
     while (!this.appFinishing.get()) {
@@ -141,10 +162,33 @@ public class AppLogAggregatorImpl implem
       this.writer.closeWriter();
       LOG.info("Finished aggregate log-file for app " + this.applicationId);
     }
-
+    try {
+      userUgi.doAs(new PrivilegedExceptionAction<Object>() {
+        @Override
+        public Object run() throws Exception {
+          FileSystem remoteFS = FileSystem.get(conf);
+          remoteFS.rename(remoteNodeTmpLogFileForApp, remoteNodeLogFileForApp);
+          return null;
+        }
+      });
+    } catch (Exception e) {
+      LOG.error("Failed to move temporary log file to final location: ["
+          + remoteNodeTmpLogFileForApp + "] to [" + remoteNodeLogFileForApp
+          + "]", e);
+    }
+    
+    this.dispatcher.getEventHandler().handle(
+        new ApplicationEvent(this.appId,
+            ApplicationEventType.APPLICATION_LOG_HANDLING_FINISHED));
+        
     this.appAggregationFinished.set(true);
   }
 
+  private Path getRemoteNodeTmpLogFileForApp() {
+    return new Path(remoteNodeLogFileForApp.getParent(),
+        (remoteNodeLogFileForApp.getName() + TMP_FILE_SUFFIX));
+  }
+
   private boolean shouldUploadLogs(ContainerId containerId,
       boolean wasContainerSuccessful) {
 

Modified: hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/logaggregation/LogAggregationService.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/logaggregation/LogAggregationService.java?rev=1196458&r1=1196457&r2=1196458&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/logaggregation/LogAggregationService.java (original)
+++ hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/logaggregation/LogAggregationService.java Wed Nov  2 05:34:31 2011
@@ -18,9 +18,9 @@
 
 package org.apache.hadoop.yarn.server.nodemanager.containermanager.logaggregation;
 
-import java.net.InetAddress;
-import java.net.UnknownHostException;
+import java.io.IOException;
 import java.security.PrivilegedExceptionAction;
+import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.ExecutorService;
@@ -31,54 +31,92 @@ import org.apache.commons.logging.LogFac
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.permission.FsPermission;
 import org.apache.hadoop.security.Credentials;
 import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hadoop.security.token.Token;
 import org.apache.hadoop.security.token.TokenIdentifier;
 import org.apache.hadoop.yarn.YarnException;
 import org.apache.hadoop.yarn.api.records.ApplicationId;
+import org.apache.hadoop.yarn.api.records.ApplicationAccessType;
 import org.apache.hadoop.yarn.api.records.ContainerId;
+import org.apache.hadoop.yarn.api.records.NodeId;
 import org.apache.hadoop.yarn.conf.YarnConfiguration;
-import org.apache.hadoop.yarn.event.EventHandler;
+import org.apache.hadoop.yarn.event.Dispatcher;
 import org.apache.hadoop.yarn.server.nodemanager.Context;
 import org.apache.hadoop.yarn.server.nodemanager.DeletionService;
-import org.apache.hadoop.yarn.server.nodemanager.containermanager.logaggregation.event.LogAggregatorEvent;
+import org.apache.hadoop.yarn.server.nodemanager.containermanager.loghandler.LogHandler;
+import org.apache.hadoop.yarn.server.nodemanager.containermanager.loghandler.event.LogHandlerAppFinishedEvent;
+import org.apache.hadoop.yarn.server.nodemanager.containermanager.loghandler.event.LogHandlerAppStartedEvent;
+import org.apache.hadoop.yarn.server.nodemanager.containermanager.loghandler.event.LogHandlerContainerFinishedEvent;
+import org.apache.hadoop.yarn.server.nodemanager.containermanager.loghandler.event.LogHandlerEvent;
 import org.apache.hadoop.yarn.service.AbstractService;
-import org.apache.hadoop.yarn.util.ConverterUtils;
+
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
 
 public class LogAggregationService extends AbstractService implements
-    EventHandler<LogAggregatorEvent> {
+    LogHandler {
 
   private static final Log LOG = LogFactory
       .getLog(LogAggregationService.class);
 
+  /*
+   * Expected deployment TLD will be 1777, owner=<NMOwner>, group=<NMGroup -
+   * Group to which NMOwner belongs> App dirs will be created as 750,
+   * owner=<AppOwner>, group=<NMGroup>: so that the owner and <NMOwner> can
+   * access / modify the files.
+   * <NMGroup> should obviously be a limited access group.
+   */
+  /**
+   * Permissions for the top level directory under which app directories will be
+   * created.
+   */
+  private static final FsPermission TLDIR_PERMISSIONS = FsPermission
+      .createImmutable((short) 01777);
+  /**
+   * Permissions for the Application directory.
+   */
+  private static final FsPermission APP_DIR_PERMISSIONS = FsPermission
+      .createImmutable((short) 0750);
+
   private final Context context;
   private final DeletionService deletionService;
+  private final Dispatcher dispatcher;
 
   private String[] localRootLogDirs;
   Path remoteRootLogDir;
-  private String nodeFile;
+  String remoteRootLogDirSuffix;
+  private NodeId nodeId;
 
   private final ConcurrentMap<ApplicationId, AppLogAggregator> appLogAggregators;
 
   private final ExecutorService threadPool;
 
-  public LogAggregationService(Context context,
+  public LogAggregationService(Dispatcher dispatcher, Context context,
       DeletionService deletionService) {
     super(LogAggregationService.class.getName());
+    this.dispatcher = dispatcher;
     this.context = context;
     this.deletionService = deletionService;
     this.appLogAggregators =
         new ConcurrentHashMap<ApplicationId, AppLogAggregator>();
-    this.threadPool = Executors.newCachedThreadPool();
+    this.threadPool = Executors.newCachedThreadPool(
+        new ThreadFactoryBuilder()
+          .setNameFormat("LogAggregationService #%d")
+          .build());
   }
 
   public synchronized void init(Configuration conf) {
     this.localRootLogDirs =
-        conf.getStrings(YarnConfiguration.NM_LOG_DIRS, YarnConfiguration.DEFAULT_NM_LOG_DIRS);
+        conf.getStrings(YarnConfiguration.NM_LOG_DIRS,
+            YarnConfiguration.DEFAULT_NM_LOG_DIRS);
     this.remoteRootLogDir =
         new Path(conf.get(YarnConfiguration.NM_REMOTE_APP_LOG_DIR,
             YarnConfiguration.DEFAULT_NM_REMOTE_APP_LOG_DIR));
+    this.remoteRootLogDirSuffix =
+        conf.get(YarnConfiguration.NM_REMOTE_APP_LOG_DIR_SUFFIX,
+            YarnConfiguration.DEFAULT_NM_REMOTE_APP_LOG_DIR_SUFFIX);
+
     super.init(conf);
   }
 
@@ -86,37 +124,231 @@ public class LogAggregationService exten
   public synchronized void start() {
     // NodeId is only available during start, the following cannot be moved
     // anywhere else.
-    this.nodeFile = this.context.getNodeId().toString();
+    this.nodeId = this.context.getNodeId();
+    verifyAndCreateRemoteLogDir(getConfig());
     super.start();
   }
+  
+  @Override
+  public synchronized void stop() {
+    LOG.info(this.getName() + " waiting for pending aggregation during exit");
+    for (AppLogAggregator appLogAggregator : this.appLogAggregators.values()) {
+      appLogAggregator.join();
+    }
+    super.stop();
+  }
+  
+  /**
+   * Constructs the full filename for an application's log file per node.
+   * @param remoteRootLogDir
+   * @param appId
+   * @param user
+   * @param nodeId
+   * @param suffix
+   * @return the remote log file.
+   */
+  public static Path getRemoteNodeLogFileForApp(Path remoteRootLogDir,
+      ApplicationId appId, String user, NodeId nodeId, String suffix) {
+    return new Path(getRemoteAppLogDir(remoteRootLogDir, appId, user, suffix),
+        getNodeString(nodeId));
+  }
+
+  /**
+   * Gets the remote app log dir.
+   * @param remoteRootLogDir
+   * @param appId
+   * @param user
+   * @param suffix
+   * @return the remote application specific log dir.
+   */
+  public static Path getRemoteAppLogDir(Path remoteRootLogDir,
+      ApplicationId appId, String user, String suffix) {
+    return new Path(getRemoteLogSuffixedDir(remoteRootLogDir, user, suffix),
+        appId.toString());
+  }
+
+  /**
+   * Gets the remote suffixed log dir for the user.
+   * @param remoteRootLogDir
+   * @param user
+   * @param suffix
+   * @return the remote suffixed log dir.
+   */
+  private static Path getRemoteLogSuffixedDir(Path remoteRootLogDir,
+      String user, String suffix) {
+    if (suffix == null || suffix.isEmpty()) {
+      return getRemoteLogUserDir(remoteRootLogDir, user);
+    }
+    // TODO Maybe support suffix to be more than a single file.
+    return new Path(getRemoteLogUserDir(remoteRootLogDir, user), suffix);
+  }
+
+  // TODO Add a utility method to list available log files. Ignore the
+  // temporary ones.
+  
+  /**
+   * Gets the remote log user dir.
+   * @param remoteRootLogDir
+   * @param user
+   * @return the remote per user log dir.
+   */
+  private static Path getRemoteLogUserDir(Path remoteRootLogDir, String user) {
+    return new Path(remoteRootLogDir, user);
+  }
+
+  /**
+   * Returns the suffix component of the log dir.
+   * @param conf
+   * @return the suffix which will be appended to the user log dir.
+   */
+  public static String getRemoteNodeLogDirSuffix(Configuration conf) {
+    return conf.get(YarnConfiguration.NM_REMOTE_APP_LOG_DIR_SUFFIX,
+        YarnConfiguration.DEFAULT_NM_REMOTE_APP_LOG_DIR_SUFFIX);
+  }
+
+  
+  /**
+   * Converts a nodeId to a form used in the app log file name.
+   * @param nodeId
+   * @return the node string to be used to construct the file name.
+   */
+  private static String getNodeString(NodeId nodeId) {
+    return nodeId.toString().replace(":", "_");
+  }
+
+  
+
+
+  
+  private void verifyAndCreateRemoteLogDir(Configuration conf) {
+    // Checking the existance of the TLD
+    FileSystem remoteFS = null;
+    try {
+      remoteFS = FileSystem.get(conf);
+    } catch (IOException e) {
+      throw new YarnException("Unable to get Remote FileSystem isntance", e);
+    }
+    boolean remoteExists = false;
+    try {
+      remoteExists = remoteFS.exists(this.remoteRootLogDir);
+    } catch (IOException e) {
+      throw new YarnException("Failed to check for existance of remoteLogDir ["
+          + this.remoteRootLogDir + "]");
+    }
+    if (remoteExists) {
+      try {
+        FsPermission perms =
+            remoteFS.getFileStatus(this.remoteRootLogDir).getPermission();
+        if (!perms.equals(TLDIR_PERMISSIONS)) {
+          LOG.warn("Remote Root Log Dir [" + this.remoteRootLogDir
+              + "] already exist, but with incorrect permissions. "
+              + "Expected: [" + TLDIR_PERMISSIONS + "], Found: [" + perms
+              + "]." + " The cluster may have problems with multiple users.");
+        }
+      } catch (IOException e) {
+        throw new YarnException(
+            "Failed while attempting to check permissions for dir ["
+                + this.remoteRootLogDir + "]");
+      }
+    } else {
+      LOG.warn("Remote Root Log Dir [" + this.remoteRootLogDir
+          + "] does not exist. Attempting to create it.");
+      try {
+        Path qualified =
+            this.remoteRootLogDir.makeQualified(remoteFS.getUri(),
+                remoteFS.getWorkingDirectory());
+        remoteFS.mkdirs(qualified, new FsPermission(TLDIR_PERMISSIONS));
+        remoteFS.setPermission(qualified, new FsPermission(TLDIR_PERMISSIONS));
+      } catch (IOException e) {
+        throw new YarnException("Failed to create remoteLogDir ["
+            + this.remoteRootLogDir + "]", e);
+      }
+    }
 
-  Path getRemoteNodeLogFileForApp(ApplicationId appId) {
-    return getRemoteNodeLogFileForApp(this.remoteRootLogDir, appId,
-        this.nodeFile);
   }
 
-  static Path getRemoteNodeLogFileForApp(Path remoteRootLogDir,
-      ApplicationId appId, String nodeFile) {
-    return new Path(getRemoteAppLogDir(remoteRootLogDir, appId),
-        nodeFile);
+  Path getRemoteNodeLogFileForApp(ApplicationId appId, String user) {
+    return LogAggregationService.getRemoteNodeLogFileForApp(
+        this.remoteRootLogDir, appId, user, this.nodeId,
+        this.remoteRootLogDirSuffix);
   }
 
-  static Path getRemoteAppLogDir(Path remoteRootLogDir,
-      ApplicationId appId) {
-    return new Path(remoteRootLogDir, ConverterUtils.toString(appId));
+  private void createDir(FileSystem fs, Path path, FsPermission fsPerm)
+      throws IOException {
+    fs.mkdirs(path, new FsPermission(fsPerm));
+    fs.setPermission(path, new FsPermission(fsPerm));
   }
 
-  @Override
-  public synchronized void stop() {
-    LOG.info(this.getName() + " waiting for pending aggregation during exit");
-    for (AppLogAggregator appLogAggregator : this.appLogAggregators.values()) {
-      appLogAggregator.join();
+  private void createAppDir(final String user, final ApplicationId appId,
+      UserGroupInformation userUgi) {
+    try {
+      userUgi.doAs(new PrivilegedExceptionAction<Object>() {
+        @Override
+        public Object run() throws Exception {
+          // TODO: Reuse FS for user?
+          FileSystem remoteFS = null;
+          Path userDir = null;
+          Path suffixDir = null;
+          Path appDir = null;
+          try {
+            remoteFS = FileSystem.get(getConfig());
+          } catch (IOException e) {
+            LOG.error("Failed to get remote FileSystem while processing app "
+                + appId, e);
+            throw e;
+          }
+          try {
+            userDir =
+                getRemoteLogUserDir(
+                    LogAggregationService.this.remoteRootLogDir, user);
+            userDir =
+                userDir.makeQualified(remoteFS.getUri(),
+                    remoteFS.getWorkingDirectory());
+            createDir(remoteFS, userDir, APP_DIR_PERMISSIONS);
+          } catch (IOException e) {
+            LOG.error("Failed to create user dir [" + userDir
+                + "] while processing app " + appId);
+            throw e;
+          }
+          try {
+            suffixDir =
+                getRemoteLogSuffixedDir(
+                    LogAggregationService.this.remoteRootLogDir, user,
+                    LogAggregationService.this.remoteRootLogDirSuffix);
+            suffixDir =
+                suffixDir.makeQualified(remoteFS.getUri(),
+                    remoteFS.getWorkingDirectory());
+            createDir(remoteFS, suffixDir, APP_DIR_PERMISSIONS);
+          } catch (IOException e) {
+            LOG.error("Failed to create suffixed user dir [" + suffixDir
+                + "] while processing app " + appId);
+            throw e;
+          }
+          try {
+            appDir =
+                getRemoteAppLogDir(LogAggregationService.this.remoteRootLogDir,
+                    appId, user,
+                    LogAggregationService.this.remoteRootLogDirSuffix);
+            appDir =
+                appDir.makeQualified(remoteFS.getUri(),
+                    remoteFS.getWorkingDirectory());
+            createDir(remoteFS, appDir, APP_DIR_PERMISSIONS);
+          } catch (IOException e) {
+            LOG.error("Failed to  create application log dir [" + appDir
+                + "] while processing app " + appId);
+            throw e;
+          }
+          return null;
+        }
+      });
+    } catch (Exception e) {
+      throw new YarnException(e);
     }
-    super.stop();
   }
 
   private void initApp(final ApplicationId appId, String user,
-      Credentials credentials, ContainerLogsRetentionPolicy logRetentionPolicy) {
+      Credentials credentials, ContainerLogsRetentionPolicy logRetentionPolicy,
+      Map<ApplicationAccessType, String> appAcls) {
 
     // Get user's FileSystem credentials
     UserGroupInformation userUgi =
@@ -128,41 +360,27 @@ public class LogAggregationService exten
       }
     }
 
+    // Create the app dir
+    createAppDir(user, appId, userUgi);
+
     // New application
     AppLogAggregator appLogAggregator =
-        new AppLogAggregatorImpl(this.deletionService, getConfig(), appId,
-            userUgi, this.localRootLogDirs,
-            getRemoteNodeLogFileForApp(appId), logRetentionPolicy);
+        new AppLogAggregatorImpl(this.dispatcher, this.deletionService, getConfig(), appId,
+            userUgi, this.localRootLogDirs, 
+            getRemoteNodeLogFileForApp(appId, user), logRetentionPolicy, appAcls);
     if (this.appLogAggregators.putIfAbsent(appId, appLogAggregator) != null) {
       throw new YarnException("Duplicate initApp for " + appId);
     }
 
-    // Create the app dir
-    try {
-      userUgi.doAs(new PrivilegedExceptionAction<Object>() {
-        @Override
-        public Object run() throws Exception {
-          // TODO: Reuse FS for user?
-          FileSystem remoteFS = FileSystem.get(getConfig());
-          remoteFS.mkdirs(getRemoteAppLogDir(
-              LogAggregationService.this.remoteRootLogDir, appId)
-              .makeQualified(remoteFS.getUri(),
-                  remoteFS.getWorkingDirectory()));
-          return null;
-        }
-      });
-    } catch (Exception e) {
-      throw new YarnException(e);
-    }
 
-    // Get the user configuration for the list of containers that need log
+    // TODO Get the user configuration for the list of containers that need log
     // aggregation.
 
     // Schedule the aggregator.
     this.threadPool.execute(appLogAggregator);
   }
 
-  private void stopContainer(ContainerId containerId, String exitCode) {
+  private void stopContainer(ContainerId containerId, int exitCode) {
 
     // A container is complete. Put this containers' logs up for aggregation if
     // this containers' logs are needed.
@@ -174,7 +392,7 @@ public class LogAggregationService exten
     }
     this.appLogAggregators.get(
         containerId.getApplicationAttemptId().getApplicationId())
-        .startContainerLogAggregation(containerId, exitCode.equals("0"));
+        .startContainerLogAggregation(containerId, exitCode == 0);
   }
 
   private void stopApp(ApplicationId appId) {
@@ -190,28 +408,30 @@ public class LogAggregationService exten
   }
 
   @Override
-  public void handle(LogAggregatorEvent event) {
-//    switch (event.getType()) {
-//    case APPLICATION_STARTED:
-//      LogAggregatorAppStartedEvent appStartEvent =
-//          (LogAggregatorAppStartedEvent) event;
-//      initApp(appStartEvent.getApplicationId(), appStartEvent.getUser(),
-//          appStartEvent.getCredentials(),
-//          appStartEvent.getLogRetentionPolicy());
-//      break;
-//    case CONTAINER_FINISHED:
-//      LogAggregatorContainerFinishedEvent containerFinishEvent =
-//          (LogAggregatorContainerFinishedEvent) event;
-//      stopContainer(containerFinishEvent.getContainerId(),
-//          containerFinishEvent.getExitCode());
-//      break;
-//    case APPLICATION_FINISHED:
-//      LogAggregatorAppFinishedEvent appFinishedEvent =
-//          (LogAggregatorAppFinishedEvent) event;
-//      stopApp(appFinishedEvent.getApplicationId());
-//      break;
-//    default:
-//      ; // Ignore
-//    }
+  public void handle(LogHandlerEvent event) {
+    switch (event.getType()) {
+      case APPLICATION_STARTED:
+        LogHandlerAppStartedEvent appStartEvent =
+            (LogHandlerAppStartedEvent) event;
+        initApp(appStartEvent.getApplicationId(), appStartEvent.getUser(),
+            appStartEvent.getCredentials(),
+            appStartEvent.getLogRetentionPolicy(),
+            appStartEvent.getApplicationAcls());
+        break;
+      case CONTAINER_FINISHED:
+        LogHandlerContainerFinishedEvent containerFinishEvent =
+            (LogHandlerContainerFinishedEvent) event;
+        stopContainer(containerFinishEvent.getContainerId(),
+            containerFinishEvent.getExitCode());
+        break;
+      case APPLICATION_FINISHED:
+        LogHandlerAppFinishedEvent appFinishedEvent =
+            (LogHandlerAppFinishedEvent) event;
+        stopApp(appFinishedEvent.getApplicationId());
+        break;
+      default:
+        ; // Ignore
+    }
+
   }
 }

Modified: hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/logaggregation/LogDumper.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/logaggregation/LogDumper.java?rev=1196458&r1=1196457&r2=1196458&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/logaggregation/LogDumper.java (original)
+++ hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/logaggregation/LogDumper.java Wed Nov  2 05:34:31 2011
@@ -35,6 +35,7 @@ import org.apache.hadoop.fs.FileContext;
 import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.RemoteIterator;
+import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hadoop.util.Tool;
 import org.apache.hadoop.yarn.api.records.ApplicationId;
 import org.apache.hadoop.yarn.conf.YarnConfiguration;
@@ -49,6 +50,7 @@ public class LogDumper extends Configure
   private static final String CONTAINER_ID_OPTION = "containerId";
   private static final String APPLICATION_ID_OPTION = "applicationId";
   private static final String NODE_ADDRESS_OPTION = "nodeAddress";
+  private static final String APP_OWNER_OPTION = "appOwner";
 
   @Override
   public int run(String[] args) throws Exception {
@@ -57,6 +59,7 @@ public class LogDumper extends Configure
     opts.addOption(APPLICATION_ID_OPTION, true, "ApplicationId");
     opts.addOption(CONTAINER_ID_OPTION, true, "ContainerId");
     opts.addOption(NODE_ADDRESS_OPTION, true, "NodeAddress");
+    opts.addOption(APP_OWNER_OPTION, true, "AppOwner");
 
     if (args.length < 1) {
       HelpFormatter formatter = new HelpFormatter();
@@ -68,11 +71,13 @@ public class LogDumper extends Configure
     String appIdStr = null;
     String containerIdStr = null;
     String nodeAddress = null;
+    String appOwner = null;
     try {
       CommandLine commandLine = parser.parse(opts, args, true);
       appIdStr = commandLine.getOptionValue(APPLICATION_ID_OPTION);
       containerIdStr = commandLine.getOptionValue(CONTAINER_ID_OPTION);
       nodeAddress = commandLine.getOptionValue(NODE_ADDRESS_OPTION);
+      appOwner = commandLine.getOptionValue(APP_OWNER_OPTION);
     } catch (ParseException e) {
       System.out.println("options parsing failed: " + e.getMessage());
 
@@ -95,8 +100,11 @@ public class LogDumper extends Configure
 
     DataOutputStream out = new DataOutputStream(System.out);
 
+    if (appOwner == null || appOwner.isEmpty()) {
+      appOwner = UserGroupInformation.getCurrentUser().getShortUserName();
+    }
     if (containerIdStr == null && nodeAddress == null) {
-      dumpAllContainersLogs(appId, out);
+      dumpAllContainersLogs(appId, appOwner, out);
     } else if ((containerIdStr == null && nodeAddress != null)
         || (containerIdStr != null && nodeAddress == null)) {
       System.out.println("ContainerId or NodeAddress cannot be null!");
@@ -110,13 +118,33 @@ public class LogDumper extends Configure
       AggregatedLogFormat.LogReader reader =
           new AggregatedLogFormat.LogReader(getConf(),
               LogAggregationService.getRemoteNodeLogFileForApp(
-                  remoteRootLogDir, appId, nodeAddress));
+                  remoteRootLogDir,
+                  appId,
+                  appOwner,
+                  ConverterUtils.toNodeId(nodeAddress),
+                  getConf().get(YarnConfiguration.NM_REMOTE_APP_LOG_DIR_SUFFIX,
+                      YarnConfiguration.DEFAULT_NM_REMOTE_APP_LOG_DIR_SUFFIX)));
       return dumpAContainerLogs(containerIdStr, reader, out);
     }
 
     return 0;
   }
 
+  public void dumpAContainersLogs(String appId, String containerId,
+      String nodeId, String jobOwner) throws IOException {
+    Path remoteRootLogDir =
+        new Path(getConf().get(YarnConfiguration.NM_REMOTE_APP_LOG_DIR,
+            YarnConfiguration.DEFAULT_NM_REMOTE_APP_LOG_DIR));
+    String suffix = LogAggregationService.getRemoteNodeLogDirSuffix(getConf());
+    AggregatedLogFormat.LogReader reader =
+        new AggregatedLogFormat.LogReader(getConf(),
+            LogAggregationService.getRemoteNodeLogFileForApp(remoteRootLogDir,
+                ConverterUtils.toApplicationId(appId), jobOwner,
+                ConverterUtils.toNodeId(nodeId), suffix));
+    DataOutputStream out = new DataOutputStream(System.out);
+    dumpAContainerLogs(containerId, reader, out);
+  }
+
   private int dumpAContainerLogs(String containerIdStr,
       AggregatedLogFormat.LogReader reader, DataOutputStream out)
       throws IOException {
@@ -146,22 +174,26 @@ public class LogDumper extends Configure
     return 0;
   }
 
-  private void
-      dumpAllContainersLogs(ApplicationId appId, DataOutputStream out)
-          throws IOException {
+  private void dumpAllContainersLogs(ApplicationId appId, String appOwner,
+      DataOutputStream out) throws IOException {
     Path remoteRootLogDir =
         new Path(getConf().get(YarnConfiguration.NM_REMOTE_APP_LOG_DIR,
             YarnConfiguration.DEFAULT_NM_REMOTE_APP_LOG_DIR));
+    String user = appOwner;
+    String logDirSuffix =
+        getConf().get(YarnConfiguration.NM_REMOTE_APP_LOG_DIR,
+            YarnConfiguration.DEFAULT_NM_REMOTE_APP_LOG_DIR_SUFFIX);
+    //TODO Change this to get a list of files from the LAS.
     Path remoteAppLogDir =
-        LogAggregationService.getRemoteAppLogDir(remoteRootLogDir, appId);
+        LogAggregationService.getRemoteAppLogDir(remoteRootLogDir, appId, user,
+            logDirSuffix);
     RemoteIterator<FileStatus> nodeFiles =
         FileContext.getFileContext().listStatus(remoteAppLogDir);
     while (nodeFiles.hasNext()) {
       FileStatus thisNodeFile = nodeFiles.next();
       AggregatedLogFormat.LogReader reader =
           new AggregatedLogFormat.LogReader(getConf(),
-              LogAggregationService.getRemoteNodeLogFileForApp(
-                  remoteRootLogDir, appId, thisNodeFile.getPath().getName()));
+              new Path(remoteAppLogDir, thisNodeFile.getPath().getName()));
       try {
 
         DataInputStream valueStream;

Modified: hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/monitor/ContainersMonitorImpl.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/monitor/ContainersMonitorImpl.java?rev=1196458&r1=1196457&r2=1196458&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/monitor/ContainersMonitorImpl.java (original)
+++ hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/monitor/ContainersMonitorImpl.java Wed Nov  2 05:34:31 2011
@@ -28,6 +28,7 @@ import java.util.Map.Entry;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.util.StringUtils;
 import org.apache.hadoop.yarn.api.records.ContainerId;
 import org.apache.hadoop.yarn.conf.YarnConfiguration;
 import org.apache.hadoop.yarn.event.AsyncDispatcher;
@@ -39,6 +40,8 @@ import org.apache.hadoop.yarn.service.Ab
 import org.apache.hadoop.yarn.util.ProcfsBasedProcessTree;
 import org.apache.hadoop.yarn.util.ResourceCalculatorPlugin;
 
+import com.google.inject.internal.Preconditions;
+
 public class ContainersMonitorImpl extends AbstractService implements
     ContainersMonitor {
 
@@ -67,11 +70,6 @@ public class ContainersMonitorImpl exten
    */
   public static final long DISABLED_MEMORY_LIMIT = -1L;
 
-  private static final String MEMORY_USAGE_STRING =
-      "Memory usage of ProcessTree %s for container-id %s : Virtual %d bytes, "
-          +
-          "limit : %d bytes; Physical %d bytes, limit %d bytes";
-
   public ContainersMonitorImpl(ContainerExecutor exec,
       AsyncDispatcher dispatcher, Context context) {
     super("containers-monitor");
@@ -110,33 +108,33 @@ public class ContainersMonitorImpl exten
       }
     }
 
-    // ///////// Virtual memory configuration //////
-    this.maxVmemAllottedForContainers =
-        conf.getLong(YarnConfiguration.NM_VMEM_GB, YarnConfiguration.DEFAULT_NM_VMEM_GB);
-    this.maxVmemAllottedForContainers =
-        this.maxVmemAllottedForContainers * 1024 * 1024 * 1024L; //Normalize
-
-    if (this.maxVmemAllottedForContainers > totalPhysicalMemoryOnNM) {
-      LOG.info("totalMemoryAllottedForContainers > totalPhysicalMemoryOnNM."
-          + " Thrashing might happen.");
-    }
-
     // ///////// Physical memory configuration //////
-    long reservedPmemOnNM =
-        conf.getLong(YarnConfiguration.NM_RESERVED_MEMORY_MB, DISABLED_MEMORY_LIMIT);
-    reservedPmemOnNM =
-        reservedPmemOnNM == DISABLED_MEMORY_LIMIT
-            ? DISABLED_MEMORY_LIMIT
-            : reservedPmemOnNM * 1024 * 1024; // normalize to bytes
-
-    if (reservedPmemOnNM == DISABLED_MEMORY_LIMIT
-        || totalPhysicalMemoryOnNM == DISABLED_MEMORY_LIMIT) {
-      this.maxPmemAllottedForContainers = DISABLED_MEMORY_LIMIT;
-    } else {
-      this.maxPmemAllottedForContainers =
-          totalPhysicalMemoryOnNM - reservedPmemOnNM;
+    this.maxPmemAllottedForContainers =
+        conf.getLong(YarnConfiguration.NM_PMEM_MB, YarnConfiguration.DEFAULT_NM_PMEM_MB);
+    this.maxPmemAllottedForContainers =
+        this.maxPmemAllottedForContainers * 1024 * 1024L; //Normalize to bytes
+
+    if (totalPhysicalMemoryOnNM != DISABLED_MEMORY_LIMIT &&
+        this.maxPmemAllottedForContainers >
+        totalPhysicalMemoryOnNM * 0.80f) {
+      LOG.warn("NodeManager configured with " +
+          StringUtils.humanReadableInt(maxPmemAllottedForContainers) +
+          " physical memory allocated to containers, which is more than " +
+          "80% of the total physical memory available (" +
+          StringUtils.humanReadableInt(totalPhysicalMemoryOnNM) +
+          "). Thrashing might happen.");
     }
 
+    // ///////// Virtual memory configuration //////
+    float vmemRatio = conf.getFloat(
+        YarnConfiguration.NM_VMEM_PMEM_RATIO,
+        YarnConfiguration.DEFAULT_NM_VMEM_PMEM_RATIO);
+    Preconditions.checkArgument(vmemRatio > 0.99f,
+        YarnConfiguration.NM_VMEM_PMEM_RATIO +
+        " should be at least 1.0");
+    this.maxVmemAllottedForContainers =
+      (long)(vmemRatio * maxPmemAllottedForContainers);
+
     super.init(conf);
   }
 
@@ -309,7 +307,7 @@ public class ContainersMonitorImpl exten
 
   private class MonitoringThread extends Thread {
     public MonitoringThread() {
-
+      super("Container Monitor");
     }
 
     @Override
@@ -399,9 +397,10 @@ public class ContainersMonitorImpl exten
             long curRssMemUsageOfAgedProcesses = pTree.getCumulativeRssmem(1);
             long vmemLimit = ptInfo.getVmemLimit();
             long pmemLimit = ptInfo.getPmemLimit();
-            LOG.info(String.format(MEMORY_USAGE_STRING, pId,
-                containerId.toString(), currentVmemUsage, vmemLimit,
-                currentPmemUsage, pmemLimit));
+            LOG.info(String.format(
+                "Memory usage of ProcessTree %s for container-id %s: ",
+                     pId, containerId.toString()) +
+                formatUsageString(currentVmemUsage, vmemLimit, currentPmemUsage, pmemLimit));
 
             boolean isMemoryOverLimit = false;
             String msg = "";
@@ -411,18 +410,10 @@ public class ContainersMonitorImpl exten
               // Container (the root process) is still alive and overflowing
               // memory.
               // Dump the process-tree and then clean it up.
-              msg =
-                  "Container [pid="
-                      + pId
-                      + ",containerID="
-                      + containerId
-                      + "] is running beyond memory-limits. Current usage : "
-                      + currentVmemUsage
-                      + "bytes. Limit : "
-                      + vmemLimit
-                      + "bytes. Killing container. "
-                      + "\nDump of the process-tree for " + containerId
-                      + " : \n" + pTree.getProcessTreeDump();
+              msg = formatErrorMessage("virtual",
+                  currentVmemUsage, vmemLimit,
+                  currentPmemUsage, pmemLimit,
+                  pId, containerId, pTree);
               isMemoryOverLimit = true;
             } else if (isPhysicalMemoryCheckEnabled()
                 && isProcessTreeOverLimit(containerId.toString(),
@@ -431,18 +422,10 @@ public class ContainersMonitorImpl exten
               // Container (the root process) is still alive and overflowing
               // memory.
               // Dump the process-tree and then clean it up.
-              msg =
-                  "Container [pid="
-                      + pId
-                      + ",tipID="
-                      + containerId
-                      + "] is running beyond physical memory-limits."
-                      + " Current usage : "
-                      + currentPmemUsage
-                      + "bytes. Limit : "
-                      + pmemLimit
-                      + "bytes. Killing container. \nDump of the process-tree for "
-                      + containerId + " : \n" + pTree.getProcessTreeDump();
+              msg = formatErrorMessage("physical",
+                  currentVmemUsage, vmemLimit,
+                  currentPmemUsage, pmemLimit,
+                  pId, containerId, pTree);
               isMemoryOverLimit = true;
             }
 
@@ -484,6 +467,31 @@ public class ContainersMonitorImpl exten
         }
       }
     }
+
+    private String formatErrorMessage(String memTypeExceeded,
+        long currentVmemUsage, long vmemLimit,
+        long currentPmemUsage, long pmemLimit,
+        String pId, ContainerId containerId, ProcfsBasedProcessTree pTree) {
+      return
+        String.format("Container [pid=%s,containerID=%s] is running beyond %s memory limits. ",
+            pId, containerId, memTypeExceeded) +
+        "Current usage: " +
+        formatUsageString(currentVmemUsage, vmemLimit,
+                          currentPmemUsage, pmemLimit) +
+        ". Killing container.\n" +
+        "Dump of the process-tree for " + containerId + " :\n" +
+        pTree.getProcessTreeDump();
+    }
+
+    private String formatUsageString(long currentVmemUsage, long vmemLimit,
+        long currentPmemUsage, long pmemLimit) {
+      return String.format("%sb of %sb physical memory used; " +
+          "%sb of %sb virtual memory used",
+          StringUtils.humanReadableInt(currentPmemUsage),
+          StringUtils.humanReadableInt(pmemLimit),
+          StringUtils.humanReadableInt(currentVmemUsage),
+          StringUtils.humanReadableInt(vmemLimit));
+    }
   }
 
   @Override

Modified: hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/webapp/AllContainersPage.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/webapp/AllContainersPage.java?rev=1196458&r1=1196457&r2=1196458&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/webapp/AllContainersPage.java (original)
+++ hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/webapp/AllContainersPage.java Wed Nov  2 05:34:31 2011
@@ -91,7 +91,8 @@ public class AllContainersPage extends N
             ._()
             .td()._(container.getContainerState())._()
             .td()
-                .a(url("containerlogs", containerIdStr), "logs")._()
+                .a(url("containerlogs", containerIdStr, container.getUser()),
+                    "logs")._()
           ._();
       }
       tableBody._()._()._();

Modified: hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/webapp/ContainerLogsPage.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/webapp/ContainerLogsPage.java?rev=1196458&r1=1196457&r2=1196458&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/webapp/ContainerLogsPage.java (original)
+++ hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/webapp/ContainerLogsPage.java Wed Nov  2 05:34:31 2011
@@ -18,9 +18,17 @@
 
 package org.apache.hadoop.yarn.server.nodemanager.webapp;
 
+import static org.apache.hadoop.yarn.server.nodemanager.webapp.NMWebParams.CONTAINER_ID;
+import static org.apache.hadoop.yarn.util.StringHelper.join;
+import static org.apache.hadoop.yarn.webapp.view.JQueryUI.ACCORDION;
+import static org.apache.hadoop.yarn.webapp.view.JQueryUI.ACCORDION_ID;
+import static org.apache.hadoop.yarn.webapp.view.JQueryUI.THEMESWITCHER_ID;
+import static org.apache.hadoop.yarn.webapp.view.JQueryUI.initID;
+
 import java.io.File;
 import java.io.FileReader;
 import java.io.IOException;
+import java.io.InputStreamReader;
 import java.util.ArrayList;
 import java.util.EnumSet;
 import java.util.List;
@@ -28,155 +36,266 @@ import java.util.List;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.LocalDirAllocator;
 import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hadoop.util.StringUtils;
+import org.apache.hadoop.yarn.api.records.ApplicationAccessType;
+import org.apache.hadoop.yarn.api.records.ApplicationId;
 import org.apache.hadoop.yarn.api.records.ContainerId;
 import org.apache.hadoop.yarn.conf.YarnConfiguration;
 import org.apache.hadoop.yarn.server.nodemanager.Context;
+import org.apache.hadoop.yarn.server.nodemanager.containermanager.application.Application;
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.container.Container;
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.container.ContainerState;
+import org.apache.hadoop.yarn.server.nodemanager.containermanager.launcher.ContainerLaunch;
+import org.apache.hadoop.yarn.server.security.ApplicationACLsManager;
 import org.apache.hadoop.yarn.util.ConverterUtils;
 import org.apache.hadoop.yarn.webapp.SubView;
-import org.apache.hadoop.yarn.webapp.hamlet.Hamlet;
-import org.apache.hadoop.yarn.webapp.hamlet.Hamlet.DIV;
 import org.apache.hadoop.yarn.webapp.view.HtmlBlock;
 
 import com.google.inject.Inject;
 
 public class ContainerLogsPage extends NMView {
+  
+  public static final String REDIRECT_URL = "redirect.url";
+  
+  @Override protected void preHead(Page.HTML<_> html) {
+    String redirectUrl = $(REDIRECT_URL);
+    if (redirectUrl == null || redirectUrl.isEmpty()) {
+      set(TITLE, join("Logs for ", $(CONTAINER_ID)));
+      html.meta_http("refresh", "10");
+    } else {
+      if (redirectUrl.equals("false")) {
+        set(TITLE, join("Failed redirect for ", $(CONTAINER_ID)));
+        //Error getting redirect url. Fall through.
+      } else {
+        set(TITLE, join("Redirecting to log server for ", $(CONTAINER_ID)));
+        html.meta_http("refresh", "1; url=" + redirectUrl);
+      }
+    }
+    
+    set(ACCORDION_ID, "nav");
+    set(initID(ACCORDION, "nav"), "{autoHeight:false, active:0}");
+    set(THEMESWITCHER_ID, "themeswitcher");
+  }
+
   @Override
   protected Class<? extends SubView> content() {
     return ContainersLogsBlock.class;
   }
 
   public static class ContainersLogsBlock extends HtmlBlock implements
-      NMWebParams {
-
+      NMWebParams {    
     private final Configuration conf;
     private final LocalDirAllocator logsSelector;
     private final Context nmContext;
+    private final ApplicationACLsManager aclsManager;
 
     @Inject
-    public ContainersLogsBlock(Configuration conf, Context context) {
+    public ContainersLogsBlock(Configuration conf, Context context,
+        ApplicationACLsManager aclsManager) {
       this.conf = conf;
       this.logsSelector = new LocalDirAllocator(YarnConfiguration.NM_LOG_DIRS);
       this.nmContext = context;
+      this.aclsManager = aclsManager;
     }
 
     @Override
     protected void render(Block html) {
-      DIV<Hamlet> div = html.div("#content");
 
+      String redirectUrl = $(REDIRECT_URL);
+      if (redirectUrl !=null && redirectUrl.equals("false")) {
+        html.h1("Failed while trying to construct the redirect url to the log" +
+        		" server. Log Server url may not be configured");
+        //Intentional fallthrough.
+      }
+      
       ContainerId containerId;
       try {
         containerId = ConverterUtils.toContainerId($(CONTAINER_ID));
-      } catch (IOException e) {
-        div.h1("Invalid containerId " + $(CONTAINER_ID))._();
+      } catch (IllegalArgumentException e) {
+        html.h1("Invalid containerId " + $(CONTAINER_ID));
         return;
       }
 
+      ApplicationId applicationId = containerId.getApplicationAttemptId()
+          .getApplicationId();
+      Application application = this.nmContext.getApplications().get(
+          applicationId);
       Container container = this.nmContext.getContainers().get(containerId);
 
-      if (container == null) {
-        div.h1(
-            "Unknown container. Container is either not yet running or "
+      if (application == null) {
+        html.h1(
+            "Unknown container. Container either has not started or "
                 + "has already completed or "
-                + "doesn't belong to this node at all.")._();
-      } else if (EnumSet.of(ContainerState.NEW, ContainerState.LOCALIZING,
-          ContainerState.LOCALIZING).contains(container.getContainerState())) {
-        div.h1("Container is not yet running. Current state is "
-                + container.getContainerState())
-              ._();
-      } else if (EnumSet.of(ContainerState.RUNNING,
+                + "doesn't belong to this node at all.");
+        return;
+      }
+      if (container == null) {
+        // Container may have alerady completed, but logs not aggregated yet.
+        printLogs(html, containerId, applicationId, application);
+        return;
+      }
+
+      if (EnumSet.of(ContainerState.NEW, ContainerState.LOCALIZING,
+          ContainerState.LOCALIZED).contains(container.getContainerState())) {
+        html.h1("Container is not yet running. Current state is "
+                + container.getContainerState());
+        return;
+      }
+
+      if (container.getContainerState() == ContainerState.LOCALIZATION_FAILED) {
+        html.h1("Container wasn't started. Localization failed.");
+        return;
+      }
+
+      if (EnumSet.of(ContainerState.RUNNING,
           ContainerState.EXITED_WITH_FAILURE,
           ContainerState.EXITED_WITH_SUCCESS).contains(
           container.getContainerState())) {
+        printLogs(html, containerId, applicationId, application);
+        return;
+      }
+      if (EnumSet.of(ContainerState.KILLING,
+          ContainerState.CONTAINER_CLEANEDUP_AFTER_KILL,
+          ContainerState.CONTAINER_RESOURCES_CLEANINGUP).contains(
+          container.getContainerState())) {
+        //Container may have generated some logs before being killed.
+        printLogs(html, containerId, applicationId, application);
+        return;
+      }
+      if (container.getContainerState().equals(ContainerState.DONE)) {
+        // Prev state unknown. Logs may be available.
+        printLogs(html, containerId, applicationId, application);
+        return;
+      } else {
+        html.h1("Container is no longer running...");
+        return;
+      }
+    }
 
-        if (!$(CONTAINER_LOG_TYPE).isEmpty()) {
-          File logFile = null;
-          try {
-            logFile =
-                new File(this.logsSelector
-                    .getLocalPathToRead(
-                        ConverterUtils.toString(
-                            containerId.getApplicationAttemptId().getApplicationId())
-                            + Path.SEPARATOR + $(CONTAINER_ID)
-                            + Path.SEPARATOR
-                            + $(CONTAINER_LOG_TYPE), this.conf).toUri()
-                    .getPath());
-          } catch (Exception e) {
-            div.h1("Cannot find this log on the local disk.")._();
-          }
-          div.h1(logFile == null ? "Unknown LogFile" : logFile.getName());
-          long start =
-              $("start").isEmpty() ? -4 * 1024 : Long.parseLong($("start"));
-          start = start < 0 ? logFile.length() + start : start;
-          start = start < 0 ? 0 : start;
-          long end =
-              $("end").isEmpty() ? logFile.length() : Long
-                  .parseLong($("end"));
-          end = end < 0 ? logFile.length() + end : end;
-          end = end < 0 ? logFile.length() : end;
-          if (start > end) {
-            writer().write("Invalid start and end values!");
-          } else {
+    private void printLogs(Block html, ContainerId containerId,
+        ApplicationId applicationId, Application application) {
+      // Check for the authorization.
+      String remoteUser = request().getRemoteUser();
+      UserGroupInformation callerUGI = null;
+
+      if (remoteUser != null) {
+        callerUGI = UserGroupInformation.createRemoteUser(remoteUser);
+      }
+      if (callerUGI != null
+          && !this.aclsManager.checkAccess(callerUGI,
+              ApplicationAccessType.VIEW_APP, application.getUser(),
+              applicationId)) {
+        html.h1(
+            "User [" + remoteUser
+                + "] is not authorized to view the logs for application "
+                + applicationId);
+        return;
+      }
+
+      if (!$(CONTAINER_LOG_TYPE).isEmpty()) {
+        File logFile = null;
+        try {
+          logFile =
+              new File(this.logsSelector
+                  .getLocalPathToRead(
+                      ContainerLaunch.getRelativeContainerLogDir(
+                          applicationId.toString(), containerId.toString())
+                          + Path.SEPARATOR + $(CONTAINER_LOG_TYPE), this.conf)
+                  .toUri().getPath());
+        } catch (Exception e) {
+          html.h1("Cannot find this log on the local disk.");
+          return;
+        }
+        long start =
+            $("start").isEmpty() ? -4 * 1024 : Long.parseLong($("start"));
+        start = start < 0 ? logFile.length() + start : start;
+        start = start < 0 ? 0 : start;
+        long end =
+            $("end").isEmpty() ? logFile.length() : Long.parseLong($("end"));
+        end = end < 0 ? logFile.length() + end : end;
+        end = end < 0 ? logFile.length() : end;
+        if (start > end) {
+          html.h1("Invalid start and end values. Start: [" + start + "]"
+              + ", end[" + end + "]");
+          return;
+        } else {
+          InputStreamReader reader = null;
           try {
             long toRead = end - start;
             if (toRead < logFile.length()) {
-                div._("Showing " + toRead + " bytes. Click ")
-                    .a(url("containerlogs", $(CONTAINER_ID),
-                        logFile.getName()), "here")
-                    ._(" for full log").br()._();
+              html.p()._("Showing " + toRead + " bytes. Click ")
+                  .a(url("containerlogs", $(CONTAINER_ID), $(APP_OWNER), 
+                      logFile.getName(), "?start=0"), "here").
+                      _(" for full log")._();
             }
             // TODO: Use secure IO Utils to avoid symlink attacks.
-            //TODO Fix findBugs close warning along with IOUtils change
-            FileReader reader = new FileReader(logFile);
-            char[] cbuf = new char[65536];
-            reader.skip(start);
+            // TODO Fix findBugs close warning along with IOUtils change
+            reader = new FileReader(logFile);
+            int bufferSize = 65536;
+            char[] cbuf = new char[bufferSize];
+
+            long skipped = 0;
+            long totalSkipped = 0;
+            while (totalSkipped < start) {
+              skipped = reader.skip(start - totalSkipped);
+              totalSkipped += skipped;
+            }
+
             int len = 0;
-            int totalRead = 0;
+            int currentToRead = toRead > bufferSize ? bufferSize : (int) toRead;
             writer().write("<pre>");
-            while ((len = reader.read(cbuf, 0, (int) toRead)) > 0
-                && totalRead < (end - start)) {
+
+            while ((len = reader.read(cbuf, 0, currentToRead)) > 0
+                && toRead > 0) {
               writer().write(cbuf, 0, len); // TODO: HTMl Quoting?
-              totalRead += len;
-              toRead = toRead - totalRead;
+              toRead = toRead - len;
+              currentToRead = toRead > bufferSize ? bufferSize : (int) toRead;
             }
+
             reader.close();
             writer().write("</pre>");
+
           } catch (IOException e) {
-              writer().write(
-                  "Exception reading log-file "
-                      + StringUtils.stringifyException(e));
-          } 
-        }
-          div._();
-        } else {
-          // Just print out the log-types
-          List<File> containerLogsDirs =
-              getContainerLogDirs(this.conf, containerId);
-          for (File containerLogsDir : containerLogsDirs) {
-            for (File logFile : containerLogsDir.listFiles()) {
-              div
-                  .p()
-                  .a(
-                      url("containerlogs", $(CONTAINER_ID),
-                          logFile.getName(), "?start=-4076"),
-                      logFile.getName() + " : Total file length is "
-                          + logFile.length() + " bytes.")
-                  ._();
+            html.h1("Exception reading log-file. Log file was likely aggregated. "
+                + StringUtils.stringifyException(e));
+          } finally {
+            if (reader != null) {
+              try {
+                reader.close();
+              } catch (IOException e) {
+                // Ignore
+              }
             }
           }
-          div._();
         }
       } else {
-        div.h1("Container is no longer running..")._();
+        // Just print out the log-types
+        List<File> containerLogsDirs =
+            getContainerLogDirs(this.conf, containerId);
+        boolean foundLogFile = false;
+        for (File containerLogsDir : containerLogsDirs) {
+          for (File logFile : containerLogsDir.listFiles()) {
+            foundLogFile = true;
+            html.p()
+                .a(url("containerlogs", $(CONTAINER_ID), $(APP_OWNER), 
+                    logFile.getName(), "?start=-4096"),
+                    logFile.getName() + " : Total file length is "
+                        + logFile.length() + " bytes.")._();
+          }
+        }
+        if (!foundLogFile) {
+          html.h1("No logs available for container " + containerId.toString());
+          return;
+        }
       }
+      return;
     }
 
     static List<File>
         getContainerLogDirs(Configuration conf, ContainerId containerId) {
-      String[] logDirs =
-          conf.getStrings(YarnConfiguration.NM_LOG_DIRS, YarnConfiguration.DEFAULT_NM_LOG_DIRS);
+      String[] logDirs = conf.getStrings(YarnConfiguration.NM_LOG_DIRS,
+          YarnConfiguration.DEFAULT_NM_LOG_DIRS);
       List<File> containerLogDirs = new ArrayList<File>(logDirs.length);
       for (String logDir : logDirs) {
         String appIdStr = 
@@ -188,6 +307,5 @@ public class ContainerLogsPage extends N
       }
       return containerLogDirs;
     }
-    
   }
 }

Modified: hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/webapp/ContainerPage.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/webapp/ContainerPage.java?rev=1196458&r1=1196457&r2=1196458&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/webapp/ContainerPage.java (original)
+++ hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/webapp/ContainerPage.java Wed Nov  2 05:34:31 2011
@@ -22,8 +22,6 @@ import static org.apache.hadoop.yarn.uti
 import static org.apache.hadoop.yarn.webapp.view.JQueryUI.ACCORDION;
 import static org.apache.hadoop.yarn.webapp.view.JQueryUI.initID;
 
-import java.io.IOException;
-
 import org.apache.hadoop.yarn.api.records.ContainerId;
 import org.apache.hadoop.yarn.api.records.ContainerStatus;
 import org.apache.hadoop.yarn.conf.YarnConfiguration;
@@ -66,7 +64,7 @@ public class ContainerPage extends NMVie
       ContainerId containerID;
       try {
         containerID = ConverterUtils.toContainerId($(CONTAINER_ID));
-      } catch (IOException e) {
+      } catch (IllegalArgumentException e) {
         html.p()._("Invalid containerId " + $(CONTAINER_ID))._();
         return;
       }
@@ -91,7 +89,8 @@ public class ContainerPage extends NMVie
         ._("User", container.getUser())
         ._("TotalMemoryNeeded",
             container.getLaunchContext().getResource().getMemory())
-        ._("logs", ujoin("containerlogs", $(CONTAINER_ID)), "Link to logs");
+        ._("logs", ujoin("containerlogs", $(CONTAINER_ID), container.getUser()),
+            "Link to logs");
       html._(InfoBlock.class);
     }
   }

Modified: hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/webapp/NMController.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/webapp/NMController.java?rev=1196458&r1=1196457&r2=1196458&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/webapp/NMController.java (original)
+++ hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/webapp/NMController.java Wed Nov  2 05:34:31 2011
@@ -21,15 +21,27 @@ package org.apache.hadoop.yarn.server.no
 import static org.apache.hadoop.yarn.util.StringHelper.join;
 
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.yarn.api.records.ApplicationId;
+import org.apache.hadoop.yarn.api.records.ContainerId;
+import org.apache.hadoop.yarn.conf.YarnConfiguration;
+import org.apache.hadoop.yarn.server.nodemanager.Context;
+import org.apache.hadoop.yarn.server.nodemanager.containermanager.application.Application;
+import org.apache.hadoop.yarn.util.ConverterUtils;
 import org.apache.hadoop.yarn.webapp.Controller;
 
 import com.google.inject.Inject;
 
 public class NMController extends Controller implements NMWebParams {
 
+  private Context nmContext;
+  private Configuration nmConf;
+  
   @Inject
-  public NMController(Configuration nmConf, RequestContext requestContext) {
+  public NMController(Configuration nmConf, RequestContext requestContext,
+      Context nmContext) {
     super(requestContext);
+    this.nmContext = nmContext;
+    this.nmConf = nmConf;
   }
 
   @Override
@@ -63,6 +75,31 @@ public class NMController extends Contro
   }
 
   public void logs() {
+    String containerIdStr = $(CONTAINER_ID);
+    ContainerId containerId = null;
+    try {
+      containerId = ConverterUtils.toContainerId(containerIdStr);
+    } catch (IllegalArgumentException e) {
+      render(ContainerLogsPage.class);
+      return;
+    }
+    ApplicationId appId =
+        containerId.getApplicationAttemptId().getApplicationId();
+    Application app = nmContext.getApplications().get(appId);
+    if (app == null
+        && nmConf.getBoolean(YarnConfiguration.NM_LOG_AGGREGATION_ENABLED,
+            YarnConfiguration.DEFAULT_NM_LOG_AGGREGATION_ENABLED)) {
+      String logServerUrl = nmConf.get(YarnConfiguration.YARN_LOG_SERVER_URL);
+      String redirectUrl = null;
+      if (logServerUrl == null || logServerUrl.isEmpty()) {
+        redirectUrl = "false";
+      } else {
+        redirectUrl =
+            url(logServerUrl, nmContext.getNodeId().toString(), containerIdStr,
+                containerIdStr, $(APP_OWNER));
+      }
+      set(ContainerLogsPage.REDIRECT_URL, redirectUrl);
+    }
     render(ContainerLogsPage.class);
   }
 }

Modified: hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/webapp/NMWebParams.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/webapp/NMWebParams.java?rev=1196458&r1=1196457&r2=1196458&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/webapp/NMWebParams.java (original)
+++ hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/webapp/NMWebParams.java Wed Nov  2 05:34:31 2011
@@ -23,4 +23,6 @@ public interface NMWebParams {
   String APPLICATION_ID = "nm.appId";
   String CONTAINER_ID = "nm.containerId";
   String CONTAINER_LOG_TYPE= "nm.containerLogType";
+  String ENTITY_STRING = "nm.entityString";
+  String APP_OWNER = "nm.appOwner";
 }

Modified: hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/webapp/WebServer.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/webapp/WebServer.java?rev=1196458&r1=1196457&r2=1196458&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/webapp/WebServer.java (original)
+++ hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/webapp/WebServer.java Wed Nov  2 05:34:31 2011
@@ -27,6 +27,7 @@ import org.apache.hadoop.yarn.YarnExcept
 import org.apache.hadoop.yarn.conf.YarnConfiguration;
 import org.apache.hadoop.yarn.server.nodemanager.Context;
 import org.apache.hadoop.yarn.server.nodemanager.ResourceView;
+import org.apache.hadoop.yarn.server.security.ApplicationACLsManager;
 import org.apache.hadoop.yarn.service.AbstractService;
 import org.apache.hadoop.yarn.webapp.WebApp;
 import org.apache.hadoop.yarn.webapp.WebApps;
@@ -36,13 +37,14 @@ public class WebServer extends AbstractS
   private static final Log LOG = LogFactory.getLog(WebServer.class);
 
   private final Context nmContext;
-  private final ResourceView resourceView;
+  private final NMWebApp nmWebApp;
   private WebApp webApp;
 
-  public WebServer(Context nmContext, ResourceView resourceView) {
+  public WebServer(Context nmContext, ResourceView resourceView,
+      ApplicationACLsManager aclsManager) {
     super(WebServer.class.getName());
     this.nmContext = nmContext;
-    this.resourceView = resourceView;
+    this.nmWebApp = new NMWebApp(resourceView, aclsManager);
   }
 
   @Override
@@ -56,10 +58,8 @@ public class WebServer extends AbstractS
         YarnConfiguration.DEFAULT_NM_WEBAPP_ADDRESS);
     LOG.info("Instantiating NMWebApp at " + bindAddress);
     try {
-      this.webApp =
-          WebApps.$for("node", Context.class, this.nmContext)
-              .at(bindAddress).with(getConfig())
-              .start(new NMWebApp(this.resourceView));
+      this.webApp = WebApps.$for("node", Context.class, this.nmContext).at(
+          bindAddress).with(getConfig()).start(this.nmWebApp);
     } catch (Exception e) {
       String msg = "NMWebapps failed to start.";
       LOG.error(msg, e);
@@ -79,14 +79,18 @@ public class WebServer extends AbstractS
   public static class NMWebApp extends WebApp implements NMWebParams {
 
     private final ResourceView resourceView;
+    private final ApplicationACLsManager aclsManager;
 
-    public NMWebApp(ResourceView resourceView) {
+    public NMWebApp(ResourceView resourceView,
+        ApplicationACLsManager aclsManager) {
       this.resourceView = resourceView;
+      this.aclsManager = aclsManager;
     }
 
     @Override
     public void setup() {
       bind(ResourceView.class).toInstance(this.resourceView);
+      bind(ApplicationACLsManager.class).toInstance(this.aclsManager);
       route("/", NMController.class, "info");
       route("/node", NMController.class, "node");
       route("/allApplications", NMController.class, "allApplications");
@@ -95,7 +99,8 @@ public class WebServer extends AbstractS
           "application");
       route(pajoin("/container", CONTAINER_ID), NMController.class,
           "container");
-      route(pajoin("/containerlogs", CONTAINER_ID, CONTAINER_LOG_TYPE),
+      route(
+          pajoin("/containerlogs", CONTAINER_ID, APP_OWNER, CONTAINER_LOG_TYPE),
           NMController.class, "logs");
     }
 

Propchange: hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/native/container-executor/
------------------------------------------------------------------------------
--- svn:ignore (added)
+++ svn:ignore Wed Nov  2 05:34:31 2011
@@ -0,0 +1,11 @@
+configure
+Makefile.in
+config.log
+config.status
+depcomp
+compile
+missing
+Makefile
+aclocal.m4
+container-executor
+install-sh

Propchange: hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/native/container-executor/impl/
------------------------------------------------------------------------------
--- svn:ignore (added)
+++ svn:ignore Wed Nov  2 05:34:31 2011
@@ -0,0 +1,2 @@
+.dirstamp
+.deps

Modified: hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/resources/container-log4j.properties
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/resources/container-log4j.properties?rev=1196458&r1=1196457&r2=1196458&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/resources/container-log4j.properties (original)
+++ hadoop/common/branches/HDFS-1623/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/resources/container-log4j.properties Wed Nov  2 05:34:31 2011
@@ -20,7 +20,7 @@ log4j.appender.CLA.containerLogDir=${yar
 log4j.appender.CLA.totalLogFileSize=${yarn.app.mapreduce.container.log.filesize}
 
 log4j.appender.CLA.layout=org.apache.log4j.PatternLayout
-log4j.appender.CLA.layout.ConversionPattern=%d{ISO8601} %p %c: %m%n
+log4j.appender.CLA.layout.ConversionPattern=%d{ISO8601} %p [%t] %c: %m%n
 
 #
 # Event Counter Appender



Mime
View raw message