hadoop-mapreduce-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From vino...@apache.org
Subject svn commit: r1101502 [2/2] - in /hadoop/mapreduce/branches/MR-279: ./ yarn/yarn-server/yarn-server-nodemanager/src/main/c/container-executor/impl/ yarn/yarn-server/yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/ yarn/ya...
Date Tue, 10 May 2011 15:18:59 GMT
Modified: hadoop/mapreduce/branches/MR-279/yarn/yarn-server/yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/localizer/ResourceLocalizationService.java
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/branches/MR-279/yarn/yarn-server/yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/localizer/ResourceLocalizationService.java?rev=1101502&r1=1101501&r2=1101502&view=diff
==============================================================================
--- hadoop/mapreduce/branches/MR-279/yarn/yarn-server/yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/localizer/ResourceLocalizationService.java (original)
+++ hadoop/mapreduce/branches/MR-279/yarn/yarn-server/yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/localizer/ResourceLocalizationService.java Tue May 10 15:18:58 2011
@@ -54,6 +54,7 @@ import org.apache.commons.logging.LogFac
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.CommonConfigurationKeys;
 import org.apache.hadoop.fs.FileContext;
+import org.apache.hadoop.fs.LocalDirAllocator;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.permission.FsPermission;
 import org.apache.hadoop.net.NetUtils;
@@ -67,6 +68,7 @@ import org.apache.hadoop.yarn.factory.pr
 import org.apache.hadoop.yarn.ipc.YarnRPC;
 import org.apache.hadoop.yarn.server.nodemanager.ContainerExecutor;
 import org.apache.hadoop.yarn.server.nodemanager.DeletionService;
+import org.apache.hadoop.yarn.server.nodemanager.NMConfig;
 import org.apache.hadoop.yarn.server.nodemanager.api.LocalizationProtocol;
 import org.apache.hadoop.yarn.server.nodemanager.api.protocolrecords.LocalResourceStatus;
 import org.apache.hadoop.yarn.server.nodemanager.api.protocolrecords.LocalizerAction;
@@ -102,16 +104,16 @@ public class ResourceLocalizationService
   public static final FsPermission NM_PRIVATE_PERM = new FsPermission((short) 0700);
 
   private Server server;
-  private InetSocketAddress locAddr;
+  private InetSocketAddress localizationServerAddress;
   private List<Path> logDirs;
   private List<Path> localDirs;
   private List<Path> sysDirs;
   private final ContainerExecutor exec;
   protected final Dispatcher dispatcher;
   private final DeletionService delService;
-  private LocalizerTracker localizers;
-  private final RecordFactory recordFactory =
-    RecordFactoryProvider.getRecordFactory(null);
+  private LocalizerTracker localizerTracker;
+  private RecordFactory recordFactory;
+  private final LocalDirAllocator localDirsSelector;
 
   //private final LocalResourcesTracker publicRsrc;
   private final ConcurrentMap<String,LocalResourcesTracker> privateRsrc =
@@ -125,6 +127,7 @@ public class ResourceLocalizationService
     this.exec = exec;
     this.dispatcher = dispatcher;
     this.delService = delService;
+    this.localDirsSelector = new LocalDirAllocator(NMConfig.NM_LOCAL_DIR);
   }
 
   FileContext getLocalFileContext(Configuration conf) {
@@ -137,6 +140,7 @@ public class ResourceLocalizationService
 
   @Override
   public void init(Configuration conf) {
+    this.recordFactory = RecordFactoryProvider.getRecordFactory(conf);
     try {
       // TODO queue deletions here, rather than NM init?
       FileContext lfs = getLocalFileContext(conf);
@@ -172,16 +176,16 @@ public class ResourceLocalizationService
     localDirs = Collections.unmodifiableList(localDirs);
     logDirs = Collections.unmodifiableList(logDirs);
     sysDirs = Collections.unmodifiableList(sysDirs);
-    locAddr = NetUtils.createSocketAddr(
+    localizationServerAddress = NetUtils.createSocketAddr(
       conf.get(NM_LOCALIZER_BIND_ADDRESS, DEFAULT_NM_LOCALIZER_BIND_ADDRESS));
-    localizers = new LocalizerTracker();
-    dispatcher.register(LocalizerEventType.class, localizers);
+    localizerTracker = new LocalizerTracker();
+    dispatcher.register(LocalizerEventType.class, localizerTracker);
     super.init(conf);
   }
 
   @Override
   public LocalizerHeartbeatResponse heartbeat(LocalizerStatus status) {
-    return localizers.processHeartbeat(status);
+    return localizerTracker.processHeartbeat(status);
   }
 
   @Override
@@ -202,8 +206,8 @@ public class ResourceLocalizationService
           LocalizerSecurityInfo.class, SecurityInfo.class);
       secretManager = new LocalizerTokenSecretManager();
     }
-    return rpc.getServer(
-        LocalizationProtocol.class, this, locAddr, conf, secretManager);
+    return rpc.getServer(LocalizationProtocol.class, this,
+        localizationServerAddress, conf, secretManager);
   }
 
   @Override
@@ -211,8 +215,8 @@ public class ResourceLocalizationService
     if (server != null) {
       server.close();
     }
-    if (localizers != null) {
-      localizers.stop();
+    if (localizerTracker != null) {
+      localizerTracker.stop();
     }
     super.stop();
   }
@@ -233,7 +237,7 @@ public class ResourceLocalizationService
       if (null != appRsrc.putIfAbsent(ConverterUtils.toString(app.getAppId()),
           new LocalResourcesTrackerImpl(dispatcher))) {
         LOG.warn("Initializing application " + app + " already present");
-        assert false;
+        assert false; // TODO: FIXME assert doesn't help
       }
       // 1) Signal container init
       dispatcher.getEventHandler().handle(new ApplicationInitedEvent(
@@ -261,6 +265,8 @@ public class ResourceLocalizationService
           appRsrc.get(ConverterUtils.toString(c.getContainerID().getAppId()));
         break;
       }
+      // We get separate events one each for all resources of one visibility. So
+      // all the resources in this event are of the same visibility.
       for (LocalResourceRequest req : rsrcReqs.getRequestedResources()) {
         tracker.handle(new ResourceRequestEvent(req, vis, ctxt));
       }
@@ -270,24 +276,25 @@ public class ResourceLocalizationService
         ((ContainerLocalizationEvent)event).getContainer();
 
       // Delete the container directories
-      userName = container.getUser();;
+      userName = container.getUser();
       String containerIDStr = container.toString();
       appIDStr =
         ConverterUtils.toString(container.getContainerID().getAppId());
       for (Path localDir : localDirs) {
+
+        // Delete the user-owned container-dir
         Path usersdir = new Path(localDir, ContainerLocalizer.USERCACHE);
-        Path userdir =
-            new Path(usersdir, userName);
+        Path userdir = new Path(usersdir, userName);
         Path allAppsdir = new Path(userdir, ContainerLocalizer.APPCACHE);
         Path appDir = new Path(allAppsdir, appIDStr);
-        Path containerDir =
-            new Path(appDir, containerIDStr);
-        delService.delete(userName, containerDir, null);
+        Path containerDir = new Path(appDir, containerIDStr);
+        delService.delete(userName, containerDir, new Path[] {});
 
+        // Delete the nmPrivate container-dir
         Path sysDir = new Path(localDir, NM_PRIVATE_DIR);
         Path appSysDir = new Path(sysDir, appIDStr);
         Path containerSysDir = new Path(appSysDir, containerIDStr);
-        delService.delete(null, containerSysDir, null);
+        delService.delete(null, containerSysDir,  new Path[] {});
       }
 
       dispatcher.getEventHandler().handle(new ContainerEvent(
@@ -298,24 +305,28 @@ public class ResourceLocalizationService
 
       Application application =
           ((ApplicationLocalizationEvent) event).getApplication();
-      if (null == appRsrc.remove(application)) {
+      LocalResourcesTracker appLocalRsrcsTracker = appRsrc.remove(application);
+      if (null == appLocalRsrcsTracker) {
         LOG.warn("Removing uninitialized application " + application);
       }
+      // TODO: What to do with appLocalRsrcsTracker?
 
       // Delete the application directories
       userName = application.getUser();
       appIDStr = application.toString();
       for (Path localDir : localDirs) {
+
+        // Delete the user-owned app-dir
         Path usersdir = new Path(localDir, ContainerLocalizer.USERCACHE);
-        Path userdir =
-            new Path(usersdir, userName);
+        Path userdir = new Path(usersdir, userName);
         Path allAppsdir = new Path(userdir, ContainerLocalizer.APPCACHE);
         Path appDir = new Path(allAppsdir, appIDStr);
-        delService.delete(userName, appDir, null);
+        delService.delete(userName, appDir, new Path[] {});
 
+        // Delete the nmPrivate app-dir
         Path sysDir = new Path(localDir, NM_PRIVATE_DIR);
         Path appSysDir = new Path(sysDir, appIDStr);
-        delService.delete(null, appSysDir, null);
+        delService.delete(null, appSysDir, new Path[] {});
       }
 
       // TODO: decrement reference counts of all resources associated with this
@@ -328,109 +339,114 @@ public class ResourceLocalizationService
     }
   }
 
+  /**
+   * Sub-component handling the spawning of {@link ContainerLocalizer}s
+   * 
+   */
   class LocalizerTracker implements EventHandler<LocalizerEvent> {
 
-    private final Map<String,LocalizerRunner> trackers;
+    private final Map<String,LocalizerRunner> localizerRunners;
 
     LocalizerTracker() {
       this(new HashMap<String,LocalizerRunner>());
     }
 
     LocalizerTracker(Map<String,LocalizerRunner> trackers) {
-      this.trackers = trackers;
+      this.localizerRunners = trackers;
     }
 
     public LocalizerHeartbeatResponse processHeartbeat(LocalizerStatus status) {
       String locId = status.getLocalizerId();
-      synchronized (trackers) {
-        LocalizerRunner localizer = trackers.get(locId);
-        if (null == localizer) {
+      synchronized (localizerRunners) {
+        LocalizerRunner localizerRunner = localizerRunners.get(locId);
+        if (null == localizerRunner) {
           // TODO process resources anyway
+          LOG.info("Unknown localizer with localizerId " + locId
+              + " is sending heartbeat. Ordering it to DIE");
           LocalizerHeartbeatResponse response =
             recordFactory.newRecordInstance(LocalizerHeartbeatResponse.class);
           response.setLocalizerAction(LocalizerAction.DIE);
           return response;
         }
-        return localizer.update(status.getResources());
+        return localizerRunner.update(status.getResources());
       }
     }
 
     public void stop() {
-      for (LocalizerRunner localizer : trackers.values()) {
+      for (LocalizerRunner localizer : localizerRunners.values()) {
         localizer.interrupt();
       }
     }
 
     @Override
     public void handle(LocalizerEvent event) {
-      synchronized (trackers) {
-        String locId = event.getLocalizerId();
-        LocalizerRunner localizer = trackers.get(locId);
+      synchronized (localizerRunners) {
+        String localizerId = event.getLocalizerId();
+        LocalizerRunner localizerRunner = localizerRunners.get(localizerId);
         switch(event.getType()) {
           case REQUEST_RESOURCE_LOCALIZATION:
             // 0) find running localizer or start new thread
             LocalizerResourceRequestEvent req =
               (LocalizerResourceRequestEvent)event;
-            if (null == localizer) {
-              LOG.info("Created localizer for " + req.getLocalizerId());
-              // TODO: ROUND_ROBIN below.
-              localizer = new LocalizerRunner(req.getContext(),
-                  sysDirs.get(0), req.getLocalizerId(), logDirs.get(0));
-              trackers.put(locId, localizer);
-              localizer.start();
+            if (null == localizerRunner) {
+              LOG.info("Created localizerRunner for " + req.getLocalizerId());
+            localizerRunner =
+                new LocalizerRunner(req.getContext(), req.getLocalizerId());
+              localizerRunners.put(localizerId, localizerRunner);
+              localizerRunner.start();
             }
             // 1) propagate event
-            localizer.addResource(req);
+            localizerRunner.addResource(req);
             break;
           case ABORT_LOCALIZATION:
+            // TODO: Who calls this?
             // 0) find running localizer, interrupt and remove
-            if (null == localizer) {
+            if (null == localizerRunner) {
               return; // ignore; already gone
             }
-            trackers.remove(locId);
-            localizer.interrupt();
+            localizerRunners.remove(localizerId);
+            localizerRunner.interrupt();
             break;
         }
       }
     }
   }
 
+  /**
+   * Runs the {@link ContainerLocalizer} itself in a separate process with
+   * access to user's credentials. One {@link LocalizerRunner} per localizerId.
+   * 
+   */
   class LocalizerRunner extends Thread {
 
     final LocalizerContext context;
     final String localizerId;
-    final Path nmPrivate;
-    final Path rootLogDir;
     final Map<LocalResourceRequest,LocalizerResourceRequestEvent> scheduled;
     final List<LocalizerResourceRequestEvent> pending;
 
     private final RecordFactory recordFactory =
       RecordFactoryProvider.getRecordFactory(null);
 
-    LocalizerRunner(LocalizerContext context, Path nmPrivate,
-        String localizerId, Path logDir) {
-      this(context, nmPrivate, localizerId, logDir,
-          new ArrayList<LocalizerResourceRequestEvent>(),
-          new HashMap<LocalResourceRequest,LocalizerResourceRequestEvent>());
-    }
-
-    LocalizerRunner(LocalizerContext context, Path nmPrivate,
-        String localizerId, Path logDir,
-        List<LocalizerResourceRequestEvent> pending,
-        Map<LocalResourceRequest,LocalizerResourceRequestEvent> scheduled) {
-      this.nmPrivate = nmPrivate;
+    LocalizerRunner(LocalizerContext context, String localizerId) {
       this.context = context;
       this.localizerId = localizerId;
-      this.rootLogDir = logDir;
-      this.pending = pending;
-      this.scheduled = scheduled;
+      this.pending = new ArrayList<LocalizerResourceRequestEvent>();
+      this.scheduled =
+          new HashMap<LocalResourceRequest, LocalizerResourceRequestEvent>();
     }
 
     public void addResource(LocalizerResourceRequestEvent request) {
+      // TDOO: Synchronization
       pending.add(request);
     }
 
-    LocalResource findNextResource() {
+    /**
+     * Find next resource to be given to a spawned localizer.
+     * 
+     * @return
+     */
+    private LocalResource findNextResource() {
+      // TODO: Synchronization
       for (Iterator<LocalizerResourceRequestEvent> i = pending.iterator();
            i.hasNext();) {
         LocalizerResourceRequestEvent evt = i.next();
@@ -457,16 +473,19 @@ public class ResourceLocalizationService
 
     // TODO this sucks. Fix it later
     LocalizerHeartbeatResponse update(
-        List<LocalResourceStatus> stats) {
+        List<LocalResourceStatus> remoteResourceStatuses) {
       LocalizerHeartbeatResponse response =
         recordFactory.newRecordInstance(LocalizerHeartbeatResponse.class);
 
-      if (stats.isEmpty()) {
+      // The localizer has just spawned. Start giving it resources for
+      // remote-fetching.
+      if (remoteResourceStatuses.isEmpty()) {
         LocalResource next = findNextResource();
         if (next != null) {
           response.setLocalizerAction(LocalizerAction.LIVE);
           response.addResource(next);
         } else if (pending.isEmpty()) {
+          // TODO: Synchronization
           response.setLocalizerAction(LocalizerAction.DIE);
         } else {
           response.setLocalizerAction(LocalizerAction.LIVE);
@@ -474,7 +493,7 @@ public class ResourceLocalizationService
         return response;
       }
 
-      for (LocalResourceStatus stat : stats) {
+      for (LocalResourceStatus stat : remoteResourceStatuses) {
         LocalResource rsrc = stat.getResource();
         LocalResourceRequest req = null;
         try {
@@ -498,6 +517,7 @@ public class ResourceLocalizationService
                     stat.getLocalSize()));
             } catch (URISyntaxException e) { }
             if (pending.isEmpty()) {
+              // TODO: Synchronization
               response.setLocalizerAction(LocalizerAction.DIE);
               break;
             }
@@ -534,20 +554,28 @@ public class ResourceLocalizationService
     @SuppressWarnings("unchecked") // dispatcher not typed
     public void run() {
       try {
+        // Use LocalDirAllocator to get nmPrivateDir
+        Path nmPrivateCTokensPath =
+            localDirsSelector.getLocalPathForWrite(
+                NM_PRIVATE_DIR
+                    + Path.SEPARATOR
+                    + String.format(ContainerLocalizer.TOKEN_FILE_FMT,
+                        localizerId), getConfig());
         // 0) init queue, etc.
         // 1) write credentials to private dir
         DataOutputStream tokenOut = null;
         try {
           Credentials credentials = context.getCredentials();
-          Path cTokens = new Path(nmPrivate, String.format(
-                ContainerLocalizer.TOKEN_FILE_FMT, localizerId));
           FileContext lfs = getLocalFileContext(getConfig());
-          tokenOut = lfs.create(cTokens, EnumSet.of(CREATE, OVERWRITE));
+          tokenOut =
+              lfs.create(nmPrivateCTokensPath, EnumSet.of(CREATE, OVERWRITE));
           LOG.info("Writing credentials to the nmPrivate file "
-              + cTokens.toString() + ". Credentials list: ");
-          for (Token<? extends TokenIdentifier> tk :
-              credentials.getAllTokens()) {
-            LOG.info(tk.getService() + " : " + tk.encodeToUrlString());
+              + nmPrivateCTokensPath.toString() + ". Credentials list: ");
+          if (LOG.isDebugEnabled()) {
+            for (Token<? extends TokenIdentifier> tk : credentials
+                .getAllTokens()) {
+              LOG.debug(tk.getService() + " : " + tk.encodeToUrlString());
+            }
           }
           credentials.writeTokenStorageToStream(tokenOut);
         } finally {
@@ -556,9 +584,10 @@ public class ResourceLocalizationService
           }
         }
         // 2) exec initApplication and wait
-        exec.startLocalizer(nmPrivate, locAddr, context.getUser(),
+        exec.startLocalizer(nmPrivateCTokensPath, localizationServerAddress,
+            context.getUser(),
             ConverterUtils.toString(context.getContainerId().getAppId()),
-            localizerId, rootLogDir, localDirs);
+            localizerId, localDirs);
       } catch (Exception e) {
         // 3) on error, report failure to Container and signal ABORT
         // 3.1) notify resource of failed localization

Modified: hadoop/mapreduce/branches/MR-279/yarn/yarn-server/yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/localizer/event/ContainerLocalizationRequestEvent.java
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/branches/MR-279/yarn/yarn-server/yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/localizer/event/ContainerLocalizationRequestEvent.java?rev=1101502&r1=1101501&r2=1101502&view=diff
==============================================================================
--- hadoop/mapreduce/branches/MR-279/yarn/yarn-server/yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/localizer/event/ContainerLocalizationRequestEvent.java (original)
+++ hadoop/mapreduce/branches/MR-279/yarn/yarn-server/yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/localizer/event/ContainerLocalizationRequestEvent.java Tue May 10 15:18:58 2011
@@ -23,13 +23,20 @@ import org.apache.hadoop.yarn.api.record
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.container.Container;
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.localizer.LocalResourceRequest;
 
-public class ContainerLocalizationRequestEvent extends ContainerLocalizationEvent {
+public class ContainerLocalizationRequestEvent extends
+    ContainerLocalizationEvent {
 
   private final LocalResourceVisibility vis;
   private final Collection<LocalResourceRequest> reqs;
 
-  public ContainerLocalizationRequestEvent(Container c, Collection<LocalResourceRequest> reqs,
-      LocalResourceVisibility vis) {
+  /**
+   * Event requesting the localization of the reqs all with visibility vis
+   * @param c
+   * @param reqs
+   * @param vis
+   */
+  public ContainerLocalizationRequestEvent(Container c,
+      Collection<LocalResourceRequest> reqs, LocalResourceVisibility vis) {
     super(LocalizationEventType.INIT_CONTAINER_RESOURCES, c);
     this.vis = vis;
     this.reqs = reqs;

Modified: hadoop/mapreduce/branches/MR-279/yarn/yarn-server/yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/localizer/event/ResourceEvent.java
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/branches/MR-279/yarn/yarn-server/yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/localizer/event/ResourceEvent.java?rev=1101502&r1=1101501&r2=1101502&view=diff
==============================================================================
--- hadoop/mapreduce/branches/MR-279/yarn/yarn-server/yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/localizer/event/ResourceEvent.java (original)
+++ hadoop/mapreduce/branches/MR-279/yarn/yarn-server/yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/localizer/event/ResourceEvent.java Tue May 10 15:18:58 2011
@@ -29,7 +29,7 @@ public class ResourceEvent extends Abstr
     this.rsrc = rsrc;
   }
 
-  public LocalResourceRequest getLocalResource() {
+  public LocalResourceRequest getLocalResourceRequest() {
     return rsrc;
   }
 

Modified: hadoop/mapreduce/branches/MR-279/yarn/yarn-server/yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/logaggregation/AggregatedLogFormat.java
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/branches/MR-279/yarn/yarn-server/yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/logaggregation/AggregatedLogFormat.java?rev=1101502&r1=1101501&r2=1101502&view=diff
==============================================================================
--- hadoop/mapreduce/branches/MR-279/yarn/yarn-server/yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/logaggregation/AggregatedLogFormat.java (original)
+++ hadoop/mapreduce/branches/MR-279/yarn/yarn-server/yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/logaggregation/AggregatedLogFormat.java Tue May 10 15:18:58 2011
@@ -78,31 +78,41 @@ public class AggregatedLogFormat {
 
   public static class LogValue {
 
-    private final File containerLogDir;
+    private final String[] rootLogDirs;
+    private final ContainerId containerId;
 
-    public LogValue(File containerLogDir) {
-      this.containerLogDir = containerLogDir;
+    public LogValue(String[] rootLogDirs, ContainerId containerId) {
+      this.rootLogDirs = rootLogDirs;
+      this.containerId = containerId;
     }
 
     public void write(DataOutputStream out) throws IOException {
-      if (!this.containerLogDir.isDirectory()) {
-        return; // ContainerDir may have been deleted by the user.
-      }
+      for (String rootLogDir : this.rootLogDirs) {
+        File appLogDir =
+            new File(rootLogDir, ConverterUtils.toString(this.containerId
+                .getAppId()));
+        File containerLogDir =
+            new File(appLogDir, ConverterUtils.toString(this.containerId));
+
+        if (!containerLogDir.isDirectory()) {
+          continue; // ContainerDir may have been deleted by the user.
+        }
 
-      for (File logFile : this.containerLogDir.listFiles()) {
+        for (File logFile : containerLogDir.listFiles()) {
 
-        // Write the logFile Type
-        out.writeUTF(logFile.getName());
+          // Write the logFile Type
+          out.writeUTF(logFile.getName());
 
-        // Write the log length as UTF so that it is printable
-        out.writeUTF(String.valueOf(logFile.length()));
+          // Write the log length as UTF so that it is printable
+          out.writeUTF(String.valueOf(logFile.length()));
 
-        // Write the log itself
-        FileInputStream in = new FileInputStream(logFile);
-        byte[] buf = new byte[65535];
-        int len = 0;
-        while ((len = in.read(buf)) != -1) {
-          out.write(buf, 0, len);
+          // Write the log itself
+          FileInputStream in = new FileInputStream(logFile);
+          byte[] buf = new byte[65535];
+          int len = 0;
+          while ((len = in.read(buf)) != -1) {
+            out.write(buf, 0, len);
+          }
         }
       }
     }

Modified: hadoop/mapreduce/branches/MR-279/yarn/yarn-server/yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/logaggregation/AppLogAggregatorImpl.java
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/branches/MR-279/yarn/yarn-server/yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/logaggregation/AppLogAggregatorImpl.java?rev=1101502&r1=1101501&r2=1101502&view=diff
==============================================================================
--- hadoop/mapreduce/branches/MR-279/yarn/yarn-server/yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/logaggregation/AppLogAggregatorImpl.java (original)
+++ hadoop/mapreduce/branches/MR-279/yarn/yarn-server/yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/logaggregation/AppLogAggregatorImpl.java Tue May 10 15:18:58 2011
@@ -20,8 +20,6 @@ package org.apache.hadoop.yarn.server.no
 
 import java.io.File;
 import java.io.IOException;
-import java.util.ArrayList;
-import java.util.List;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -45,12 +43,12 @@ public class AppLogAggregatorImpl implem
       .getLog(AppLogAggregatorImpl.class);
   private static final int THREAD_SLEEP_TIME = 1000;
 
-  private final ApplicationId applicationId;
+  private final String applicationId;
   private boolean logAggregationDisabled = false;
   private final Configuration conf;
   private final DeletionService delService;
   private final UserGroupInformation userUgi;
-  private final File localAppLogDir;
+  private final String[] rootLogDirs;
   private final Path remoteNodeLogFileForApp;
   private final ContainerLogsRetentionPolicy retentionPolicy;
 
@@ -62,22 +60,18 @@ public class AppLogAggregatorImpl implem
 
   public AppLogAggregatorImpl(DeletionService deletionService,
       Configuration conf, ApplicationId appId, UserGroupInformation userUgi,
-      File localAppLogDir, Path remoteNodeLogFileForApp,
+      String[] localRootLogDirs, Path remoteNodeLogFileForApp,
       ContainerLogsRetentionPolicy retentionPolicy) {
     this.conf = conf;
     this.delService = deletionService;
-    this.applicationId = appId;
+    this.applicationId = ConverterUtils.toString(appId);
     this.userUgi = userUgi;
-    this.localAppLogDir = localAppLogDir;
+    this.rootLogDirs = localRootLogDirs;
     this.remoteNodeLogFileForApp = remoteNodeLogFileForApp;
     this.retentionPolicy = retentionPolicy;
     this.pendingContainers = new LinkedBlockingQueue<ContainerId>();
   }
 
-  private File getLocalContainerLogDir(ContainerId containerId) {
-    return new File(this.localAppLogDir, ConverterUtils.toString(containerId));
-  }
-
   private void uploadLogsForContainer(ContainerId containerId) {
 
     if (this.logAggregationDisabled) {
@@ -99,11 +93,9 @@ public class AppLogAggregatorImpl implem
       }
     }
 
-    File containerLogDir = getLocalContainerLogDir(containerId);
-    LOG.info("Uploading logs for container " + containerId + " from "
-        + containerLogDir);
+    LOG.info("Uploading logs for container " + containerId);
     LogKey logKey = new LogKey(containerId);
-    LogValue logValue = new LogValue(containerLogDir);
+    LogValue logValue = new LogValue(this.rootLogDirs, containerId);
     try {
       this.writer.append(logKey, logValue);
     } catch (IOException e) {
@@ -135,9 +127,12 @@ public class AppLogAggregatorImpl implem
       uploadLogsForContainer(containerId);
     }
 
-    // Remove the local app-log-dir
-    this.delService.delete(this.userUgi.getShortUserName(), new Path(
-        this.localAppLogDir.getAbsolutePath()), new Path[] {});
+    // Remove the local app-log-dirs
+    for (String rootLogDir : this.rootLogDirs) {
+      File localAppLogDir = new File(rootLogDir, this.applicationId);
+      this.delService.delete(this.userUgi.getShortUserName(), new Path(
+          localAppLogDir.getAbsolutePath()), new Path[] {});
+    }
 
     if (this.writer != null) {
       this.writer.closeWriter();

Modified: hadoop/mapreduce/branches/MR-279/yarn/yarn-server/yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/logaggregation/LogAggregationService.java
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/branches/MR-279/yarn/yarn-server/yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/logaggregation/LogAggregationService.java?rev=1101502&r1=1101501&r2=1101502&view=diff
==============================================================================
--- hadoop/mapreduce/branches/MR-279/yarn/yarn-server/yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/logaggregation/LogAggregationService.java (original)
+++ hadoop/mapreduce/branches/MR-279/yarn/yarn-server/yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/logaggregation/LogAggregationService.java Tue May 10 15:18:58 2011
@@ -21,9 +21,6 @@ package org.apache.hadoop.yarn.server.no
 import static org.apache.hadoop.yarn.server.nodemanager.NMConfig.DEFAULT_NM_BIND_ADDRESS;
 import static org.apache.hadoop.yarn.server.nodemanager.NMConfig.NM_BIND_ADDRESS;
 
-import java.io.File;
-import java.io.IOException;
-import java.nio.ByteBuffer;
 import java.security.PrivilegedExceptionAction;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
@@ -33,10 +30,8 @@ import java.util.concurrent.Executors;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.FileContext;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.io.DataInputByteBuffer;
 import org.apache.hadoop.security.Credentials;
 import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hadoop.security.token.Token;
@@ -62,7 +57,7 @@ public class LogAggregationService exten
 
   private final DeletionService deletionService;
 
-  private File localRootLogDir;
+  private String[] localRootLogDirs;
   Path remoteRootLogDir;
   private String nodeFile;
 
@@ -86,8 +81,8 @@ public class LogAggregationService exten
   }
 
   public synchronized void init(Configuration conf) {
-    this.localRootLogDir =
-        new File(conf.get(NMConfig.NM_LOG_DIR, NMConfig.DEFAULT_NM_LOG_DIR));
+    this.localRootLogDirs =
+        conf.getStrings(NMConfig.NM_LOG_DIR, NMConfig.DEFAULT_NM_LOG_DIR);
     this.remoteRootLogDir =
         new Path(conf.get(NMConfig.REMOTE_USER_LOG_DIR,
             NMConfig.DEFAULT_REMOTE_APP_LOG_DIR));
@@ -119,10 +114,6 @@ public class LogAggregationService exten
     return new Path(remoteRootLogDir, ConverterUtils.toString(appId));
   }
 
-  File getLocalAppLogDir(ApplicationId appId) {
-    return new File(this.localRootLogDir, ConverterUtils.toString(appId));
-  }
-
   @Override
   public synchronized void stop() {
     LOG.info(this.getName() + " waiting for pending aggregation during exit");
@@ -148,7 +139,7 @@ public class LogAggregationService exten
     // New application
     AppLogAggregator appLogAggregator =
         new AppLogAggregatorImpl(this.deletionService, getConfig(), appId,
-            userUgi, getLocalAppLogDir(appId),
+            userUgi, this.localRootLogDirs,
             getRemoteNodeLogFileForApp(appId), logRetentionPolicy);
     if (this.appLogAggregators.putIfAbsent(appId, appLogAggregator) != null) {
       throw new YarnException("Duplicate initApp for " + appId);

Modified: hadoop/mapreduce/branches/MR-279/yarn/yarn-server/yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/webapp/ContainerLogsPage.java
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/branches/MR-279/yarn/yarn-server/yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/webapp/ContainerLogsPage.java?rev=1101502&r1=1101501&r2=1101502&view=diff
==============================================================================
--- hadoop/mapreduce/branches/MR-279/yarn/yarn-server/yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/webapp/ContainerLogsPage.java (original)
+++ hadoop/mapreduce/branches/MR-279/yarn/yarn-server/yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/webapp/ContainerLogsPage.java Tue May 10 15:18:58 2011
@@ -24,14 +24,19 @@ import static org.apache.hadoop.yarn.ser
 import java.io.File;
 import java.io.FileReader;
 import java.io.IOException;
+import java.util.ArrayList;
 import java.util.EnumSet;
+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.util.StringUtils;
 import org.apache.hadoop.yarn.api.records.ContainerId;
 import org.apache.hadoop.yarn.factories.RecordFactory;
 import org.apache.hadoop.yarn.factory.providers.RecordFactoryProvider;
 import org.apache.hadoop.yarn.server.nodemanager.Context;
+import org.apache.hadoop.yarn.server.nodemanager.NMConfig;
 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.util.ConverterUtils;
@@ -52,12 +57,14 @@ public class ContainerLogsPage extends N
       NMWebParams {
 
     private final Configuration conf;
+    private final LocalDirAllocator logsSelector;
     private final Context nmContext;
     private final RecordFactory recordFactory;
 
     @Inject
     public ContainersLogsBlock(Configuration conf, Context context) {
       this.conf = conf;
+      this.logsSelector = new LocalDirAllocator(NMConfig.NM_LOG_DIR);
       this.nmContext = context;
       this.recordFactory = RecordFactoryProvider.getRecordFactory(conf);
     }
@@ -81,10 +88,19 @@ public class ContainerLogsPage extends N
           container.getContainerState())) {
 
         if (!$(CONTAINER_LOG_TYPE).isEmpty()) {
-          // TODO: Get the following from logs' owning component.
-          File containerLogsDir =
-              getContainerLogDir(this.conf, containerId);
-          File logFile = new File(containerLogsDir, $(CONTAINER_LOG_TYPE));
+          File logFile = null;
+          try {
+            logFile =
+                new File(this.logsSelector
+                    .getLocalPathToRead(
+                        ConverterUtils.toString(containerId.getAppId())
+                            + 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.getName());
           long start =
               $("start").isEmpty() ? -4 * 1024 : Long.parseLong($("start"));
@@ -130,19 +146,20 @@ public class ContainerLogsPage extends N
           div._();
         } else {
           // Just print out the log-types
-          File containerLogsDir =
-              getContainerLogDir(this.conf, containerId);
-          // TODO: No nested dir structure. Fix MR userlogs.
-          for (File logFile : containerLogsDir.listFiles()) {
-            div
-              .p()
-                .a(
-                    url($(NM_HTTP_URL), "yarn", "containerlogs",
-                        $(CONTAINER_ID),
-                        logFile.getName(), "?start=-4076"),
-                    logFile.getName() + " : Total file length is " 
-                    + logFile.length() + " bytes.")
-              ._();
+          List<File> containerLogsDirs =
+              getContainerLogDirs(this.conf, containerId);
+          for (File containerLogsDir : containerLogsDirs) {
+            for (File logFile : containerLogsDir.listFiles()) {
+              div
+                  .p()
+                  .a(
+                      url($(NM_HTTP_URL), "yarn", "containerlogs",
+                          $(CONTAINER_ID),
+                          logFile.getName(), "?start=-4076"),
+                      logFile.getName() + " : Total file length is "
+                          + logFile.length() + " bytes.")
+                  ._();
+            }
           }
           div._();
         }
@@ -151,16 +168,18 @@ public class ContainerLogsPage extends N
       }
     }
 
-    static File
-        getContainerLogDir(Configuration conf, ContainerId containerId) {
+    static List<File>
+        getContainerLogDirs(Configuration conf, ContainerId containerId) {
       String[] logDirs =
           conf.getStrings(NM_LOG_DIR, DEFAULT_NM_LOG_DIR);
-      File logDir = new File(logDirs[0]); // TODO: In case of ROUND_ROBIN
-      String appIdStr = ConverterUtils.toString(containerId.getAppId());
-      File appLogDir = new File(logDir, appIdStr);
-      String containerIdStr = ConverterUtils.toString(containerId);
-      File containerLogDir = new File(appLogDir, containerIdStr);
-      return containerLogDir;
+      List<File> containerLogDirs = new ArrayList<File>(logDirs.length);
+      for (String logDir : logDirs) {
+        String appIdStr = ConverterUtils.toString(containerId.getAppId());
+        File appLogDir = new File(logDir, appIdStr);
+        String containerIdStr = ConverterUtils.toString(containerId);
+        containerLogDirs.add(new File(appLogDir, containerIdStr));
+      }
+      return containerLogDirs;
     }
     
   }

Modified: hadoop/mapreduce/branches/MR-279/yarn/yarn-server/yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/TestDeletionService.java
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/branches/MR-279/yarn/yarn-server/yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/TestDeletionService.java?rev=1101502&r1=1101501&r2=1101502&view=diff
==============================================================================
--- hadoop/mapreduce/branches/MR-279/yarn/yarn-server/yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/TestDeletionService.java (original)
+++ hadoop/mapreduce/branches/MR-279/yarn/yarn-server/yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/TestDeletionService.java Tue May 10 15:18:58 2011
@@ -96,9 +96,11 @@ public class TestDeletionService {
     System.out.println("SEED: " + seed);
     List<Path> dirs = buildDirs(r, base, 20);
     createDirs(new Path("."), dirs);
-    DeletionService del =
-      new DeletionService(new FakeDefaultContainerExecutor());
-    del.init(new Configuration());
+    FakeDefaultContainerExecutor exec = new FakeDefaultContainerExecutor();
+    Configuration conf = new Configuration();
+    exec.setConf(conf);
+    DeletionService del = new DeletionService(exec);
+    del.init(conf);
     del.start();
     try {
       for (Path p : dirs) {

Modified: hadoop/mapreduce/branches/MR-279/yarn/yarn-server/yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/TestEventFlow.java
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/branches/MR-279/yarn/yarn-server/yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/TestEventFlow.java?rev=1101502&r1=1101501&r2=1101502&view=diff
==============================================================================
--- hadoop/mapreduce/branches/MR-279/yarn/yarn-server/yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/TestEventFlow.java (original)
+++ hadoop/mapreduce/branches/MR-279/yarn/yarn-server/yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/TestEventFlow.java Tue May 10 15:18:58 2011
@@ -77,6 +77,7 @@ public class TestEventFlow {
     conf.set(NMConfig.REMOTE_USER_LOG_DIR, remoteLogDir.getAbsolutePath());
 
     ContainerExecutor exec = new DefaultContainerExecutor();
+    exec.setConf(conf);
     DeletionService del = new DeletionService(exec);
     Dispatcher dispatcher = new AsyncDispatcher();
     NodeHealthCheckerService healthChecker = null;

Modified: hadoop/mapreduce/branches/MR-279/yarn/yarn-server/yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/BaseContainerManagerTest.java
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/branches/MR-279/yarn/yarn-server/yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/BaseContainerManagerTest.java?rev=1101502&r1=1101501&r2=1101502&view=diff
==============================================================================
--- hadoop/mapreduce/branches/MR-279/yarn/yarn-server/yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/BaseContainerManagerTest.java (original)
+++ hadoop/mapreduce/branches/MR-279/yarn/yarn-server/yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/BaseContainerManagerTest.java Tue May 10 15:18:58 2011
@@ -90,7 +90,7 @@ public abstract class BaseContainerManag
 
   protected Configuration conf = new YarnConfiguration();
   protected Context context = new NMContext();
-  protected ContainerExecutor exec = new DefaultContainerExecutor();
+  protected ContainerExecutor exec;
   protected DeletionService delSrvc;
   protected String user = "nobody";
 
@@ -111,7 +111,9 @@ public abstract class BaseContainerManag
   protected ContainerManagerImpl containerManager = null;
 
   protected ContainerExecutor createContainerExecutor() {
-    return new DefaultContainerExecutor();
+    DefaultContainerExecutor exec = new DefaultContainerExecutor();
+    exec.setConf(conf);
+    return exec;
   }
 
   @Before

Modified: hadoop/mapreduce/branches/MR-279/yarn/yarn-server/yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/localizer/TestContainerLocalizer.java
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/branches/MR-279/yarn/yarn-server/yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/localizer/TestContainerLocalizer.java?rev=1101502&r1=1101501&r2=1101502&view=diff
==============================================================================
--- hadoop/mapreduce/branches/MR-279/yarn/yarn-server/yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/localizer/TestContainerLocalizer.java (original)
+++ hadoop/mapreduce/branches/MR-279/yarn/yarn-server/yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/localizer/TestContainerLocalizer.java Tue May 10 15:18:58 2011
@@ -90,15 +90,13 @@ public class TestContainerLocalizer {
     final String appId = "app_RM_0";
     final String cId = "container_0";
     final InetSocketAddress nmAddr = new InetSocketAddress("foobar", 4344);
-    final Path logDir = lfs.makeQualified(new Path(basedir, "logs"));
     final List<Path> localDirs = new ArrayList<Path>();
     for (int i = 0; i < 4; ++i) {
       localDirs.add(lfs.makeQualified(new Path(basedir, i + "")));
     }
     RecordFactory mockRF = getMockLocalizerRecordFactory();
     ContainerLocalizer concreteLoc = new ContainerLocalizer(lfs, user,
-        appId, cId, logDir, localDirs,
-        new HashMap<LocalResource,Future<Path>>(), mockRF);
+        appId, cId, localDirs, mockRF);
     ContainerLocalizer localizer = spy(concreteLoc);
 
     // return credential stream instead of opening local file
@@ -166,29 +164,25 @@ public class TestContainerLocalizer {
     // run localization
     assertEquals(0, localizer.runLocalization(nmAddr));
 
-    // verify created cache, application dirs
+    // verify created cache
     for (Path p : localDirs) {
       Path base = new Path(new Path(p, ContainerLocalizer.USERCACHE), user);
       Path privcache = new Path(base, ContainerLocalizer.FILECACHE);
       // $x/usercache/$user/filecache
-      verify(spylfs).mkdir(eq(privcache), isA(FsPermission.class), eq(true));
+      verify(spylfs).mkdir(eq(privcache), isA(FsPermission.class), eq(false));
       Path appDir =
         new Path(base, new Path(ContainerLocalizer.APPCACHE, appId));
       // $x/usercache/$user/appcache/$appId/filecache
       Path appcache = new Path(appDir, ContainerLocalizer.FILECACHE);
-      verify(spylfs).mkdir(eq(appcache), isA(FsPermission.class), eq(true));
+      verify(spylfs).mkdir(eq(appcache), isA(FsPermission.class), eq(false));
       // $x/usercache/$user/appcache/$appId/output
       Path appOutput = new Path(appDir, ContainerLocalizer.OUTPUTDIR);
-      verify(spylfs).mkdir(eq(appOutput), isA(FsPermission.class), eq(true));
+      verify(spylfs).mkdir(eq(appOutput), isA(FsPermission.class), eq(false));
     }
 
     // verify tokens read at expected location
     verify(spylfs).open(tokenPath);
 
-    // verify log dir creation
-    verify(spylfs).mkdir(eq(new Path(logDir, appId)),
-        isA(FsPermission.class), anyBoolean());
-
     // verify downloaded resources reported to NM
     verify(nmProxy).heartbeat(argThat(new HBMatches(rsrcA)));
     verify(nmProxy).heartbeat(argThat(new HBMatches(rsrcB)));

Modified: hadoop/mapreduce/branches/MR-279/yarn/yarn-server/yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/logaggregation/TestLogAggregationService.java
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/branches/MR-279/yarn/yarn-server/yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/logaggregation/TestLogAggregationService.java?rev=1101502&r1=1101501&r2=1101502&view=diff
==============================================================================
--- hadoop/mapreduce/branches/MR-279/yarn/yarn-server/yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/logaggregation/TestLogAggregationService.java (original)
+++ hadoop/mapreduce/branches/MR-279/yarn/yarn-server/yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/logaggregation/TestLogAggregationService.java Tue May 10 15:18:58 2011
@@ -99,7 +99,8 @@ public class TestLogAggregationService e
     ApplicationId application1 = BuilderUtils.newApplicationId(1234, 1);
 
     // AppLogDir should be created
-    File app1LogDir = logAggregationService.getLocalAppLogDir(application1);
+    File app1LogDir =
+        new File(localLogDir, ConverterUtils.toString(application1));
     app1LogDir.mkdir();
     logAggregationService
         .handle(new LogAggregatorAppStartedEvent(
@@ -141,7 +142,8 @@ public class TestLogAggregationService e
     ApplicationId application1 = BuilderUtils.newApplicationId(1234, 1);
 
     // AppLogDir should be created
-    File app1LogDir = logAggregationService.getLocalAppLogDir(application1);
+    File app1LogDir =
+      new File(localLogDir, ConverterUtils.toString(application1));
     app1LogDir.mkdir();
     logAggregationService
         .handle(new LogAggregatorAppStartedEvent(
@@ -173,7 +175,8 @@ public class TestLogAggregationService e
     ApplicationId application1 = BuilderUtils.newApplicationId(1234, 1);
 
     // AppLogDir should be created
-    File app1LogDir = logAggregationService.getLocalAppLogDir(application1);
+    File app1LogDir =
+      new File(localLogDir, ConverterUtils.toString(application1));
     app1LogDir.mkdir();
     logAggregationService
         .handle(new LogAggregatorAppStartedEvent(
@@ -189,7 +192,8 @@ public class TestLogAggregationService e
 
     ApplicationId application2 = BuilderUtils.newApplicationId(1234, 2);
 
-    File app2LogDir = logAggregationService.getLocalAppLogDir(application2);
+    File app2LogDir =
+      new File(localLogDir, ConverterUtils.toString(application2));
     app2LogDir.mkdir();
     logAggregationService.handle(new LogAggregatorAppStartedEvent(
         application2, this.user, null,
@@ -209,7 +213,8 @@ public class TestLogAggregationService e
 
     ApplicationId application3 = BuilderUtils.newApplicationId(1234, 3);
 
-    File app3LogDir = logAggregationService.getLocalAppLogDir(application3);
+    File app3LogDir =
+      new File(localLogDir, ConverterUtils.toString(application3));
     app3LogDir.mkdir();
     logAggregationService.handle(new LogAggregatorAppStartedEvent(
         application3, this.user, null,

Modified: hadoop/mapreduce/branches/MR-279/yarn/yarn-server/yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/webapp/TestNMWebServer.java
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/branches/MR-279/yarn/yarn-server/yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/webapp/TestNMWebServer.java?rev=1101502&r1=1101501&r2=1101502&view=diff
==============================================================================
--- hadoop/mapreduce/branches/MR-279/yarn/yarn-server/yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/webapp/TestNMWebServer.java (original)
+++ hadoop/mapreduce/branches/MR-279/yarn/yarn-server/yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/webapp/TestNMWebServer.java Tue May 10 15:18:58 2011
@@ -122,8 +122,8 @@ public class TestNMWebServer {
         throws IOException {
     // ContainerLogDir should be created
     File containerLogDir =
-        ContainerLogsPage.ContainersLogsBlock.getContainerLogDir(conf,
-            containerId);
+        ContainerLogsPage.ContainersLogsBlock.getContainerLogDirs(conf,
+            containerId).get(0);
     containerLogDir.mkdirs();
     for (String fileType : new String[] { "stdout", "stderr", "syslog" }) {
       Writer writer = new FileWriter(new File(containerLogDir, fileType));



Mime
View raw message