Numa support in DCE

Fix checkstyle error

Remove container reference

Adding test for the numa changes for DCE

Remove unused variable creation

releasing numa resources when container closes and reassign during resassign time

Addressing to comments

Fixing failed test case

Fixing failed test case

Fix extra spaces

Add test for reaquiredContainer

Fix test error

correcting the test cases properly

correcting the test cases properly

correcting the test cases properly

correcting the test cases properly

correcting the test cases properly

Putting extra logs

Author: Samrat002

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/DefaultContainerExecutor.java

@@ -20,6 +20,7 @@
 
 import static org.apache.hadoop.fs.CreateFlag.CREATE;
 import static org.apache.hadoop.fs.CreateFlag.OVERWRITE;
+import static org.apache.hadoop.yarn.conf.YarnConfiguration.numaAwarenessEnabled;
 
 import org.apache.hadoop.classification.VisibleForTesting;
 import java.io.DataOutputStream;
@@ -28,12 +29,13 @@
 import java.io.IOException;
 import java.io.PrintStream;
 import java.net.InetSocketAddress;
-import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.ArrayList;
 import java.util.EnumSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+
 import org.apache.commons.lang3.RandomUtils;
 import org.apache.hadoop.classification.InterfaceAudience.Private;
 import org.apache.hadoop.fs.FileContext;
@@ -51,14 +53,19 @@
 import org.apache.hadoop.yarn.api.records.Resource;
 import org.apache.hadoop.yarn.conf.YarnConfiguration;
 import org.apache.hadoop.yarn.exceptions.ConfigurationException;
+import org.apache.hadoop.yarn.exceptions.YarnException;
 import org.apache.hadoop.yarn.factory.providers.RecordFactoryProvider;
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.container.Container;
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.container.ContainerDiagnosticsUpdateEvent;
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.launcher.ContainerLaunch;
+import org.apache.hadoop.yarn.server.nodemanager.containermanager.linux.resources.ResourceHandlerException;
+import org.apache.hadoop.yarn.server.nodemanager.containermanager.linux.resources.numa.NumaResourceAllocation;
+import org.apache.hadoop.yarn.server.nodemanager.containermanager.linux.resources.numa.NumaResourceAllocator;
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.localizer.ContainerLocalizer;
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.runtime.ContainerExecutionException;
 import org.apache.hadoop.yarn.server.nodemanager.executor.ContainerExecContext;
 import org.apache.hadoop.yarn.server.nodemanager.executor.ContainerLivenessContext;
+import org.apache.hadoop.yarn.server.nodemanager.executor.ContainerReacquisitionContext;
 import org.apache.hadoop.yarn.server.nodemanager.executor.ContainerReapContext;
 import org.apache.hadoop.yarn.server.nodemanager.executor.ContainerSignalContext;
 import org.apache.hadoop.yarn.server.nodemanager.executor.ContainerStartContext;
@@ -86,6 +93,8 @@ public class DefaultContainerExecutor extends ContainerExecutor {
 
   private String logDirPermissions = null;
 
+  private NumaResourceAllocator numaResourceAllocator;
+
   /**
    * Default constructor for use in testing.
    */
@@ -137,7 +146,17 @@ protected void setScriptExecutable(Path script, String owner)
 
   @Override
   public void init(Context nmContext) throws IOException {
-    // nothing to do or verify here
+    if(numaAwarenessEnabled(getConf())) {
+      numaResourceAllocator = new NumaResourceAllocator(nmContext);
+      try {
+        numaResourceAllocator.init(this.getConf());
+        LOG.info("NUMA resources allocation is enabled in DefaultContainer Executor," +
+                " Successfully initialized NUMA resources allocator.");
+      } catch (YarnException e) {
+        LOG.warn("Improper NUMA configuration provided.", e);
+        throw new IOException("Failed to initialize configured numa subsystem!");
+      }
+    }
   }
 
   @Override
@@ -300,11 +319,28 @@ public int launchContainer(ContainerStartContext ctx)
       setScriptExecutable(launchDst, user);
       setScriptExecutable(sb.getWrapperScriptPath(), user);
 
+      // adding numa commands based on configuration
+      String[] numaCommands = new String[]{};
+
+      if (numaResourceAllocator != null) {
+        try {
+          NumaResourceAllocation numaResourceAllocation =
+                  numaResourceAllocator.allocateNumaNodes(container);
+          if (numaResourceAllocation != null) {
+            numaCommands = getNumaCommands(numaResourceAllocation);
+          }
+        } catch (ResourceHandlerException e) {
+          LOG.error("NumaResource Allocation failed!", e);
+          throw new IOException("NumaResource Allocation Error!", e);
+        }
+      }
+
       shExec = buildCommandExecutor(sb.getWrapperScriptPath().toString(),
-          containerIdStr, user, pidFile, container.getResource(),
-          new File(containerWorkDir.toUri().getPath()),
-          container.getLaunchContext().getEnvironment());
-      
+              containerIdStr, user, pidFile, container.getResource(),
+              new File(containerWorkDir.toUri().getPath()),
+              container.getLaunchContext().getEnvironment(),
+              numaCommands);
+
       if (isContainerActive(containerId)) {
         shExec.execute();
       } else {
@@ -350,6 +386,7 @@ public int launchContainer(ContainerStartContext ctx)
       return exitCode;
     } finally {
       if (shExec != null) shExec.close();
+      postComplete(containerId);
     }
     return 0;
   }
@@ -372,16 +409,19 @@ public int relaunchContainer(ContainerStartContext ctx)
    * as the current working directory for the command. If null,
    * the current working directory is not modified.
    * @param environment the container environment
+   * @param numaCommands list of prefix numa commands
    * @return the new {@link ShellCommandExecutor}
    * @see ShellCommandExecutor
    */
-  protected CommandExecutor buildCommandExecutor(String wrapperScriptPath, 
-      String containerIdStr, String user, Path pidFile, Resource resource,
-      File workDir, Map<String, String> environment) {
-    
+  protected CommandExecutor buildCommandExecutor(String wrapperScriptPath,
+                            String containerIdStr, String user, Path pidFile, Resource resource,
+                            File workDir, Map<String, String> environment, String[] numaCommands) {
+
     String[] command = getRunCommand(wrapperScriptPath,
         containerIdStr, user, pidFile, this.getConf(), resource);
 
+    command = concatStringCommands(command, numaCommands);
+
     LOG.info("launchContainer: {}", Arrays.toString(command));
     return new ShellCommandExecutor(
         command,
@@ -1040,4 +1080,92 @@ public void updateYarnSysFS(Context ctx, String user,
       String appId, String spec) throws IOException {
     throw new ServiceStateException("Implementation unavailable");
   }
+
+  @Override
+  public int reacquireContainer(ContainerReacquisitionContext ctx)
+          throws IOException, InterruptedException {
+    try {
+      if (numaResourceAllocator != null) {
+        numaResourceAllocator.recoverNumaResource(ctx.getContainerId());
+      }
+      return super.reacquireContainer(ctx);
+    } finally {
+      postComplete(ctx.getContainerId());
+    }
+  }
+
+  /**
+   * clean up and release of resources.
+   *
+   * @param containerId containerId of running container
+   */
+  public void postComplete(final ContainerId containerId) {
+    if (numaResourceAllocator != null) {
+      try {
+        numaResourceAllocator.releaseNumaResource(containerId);
+      } catch (ResourceHandlerException e) {
+        LOG.warn("NumaResource release failed for " +
+                "containerId: {}. Exception: ", containerId, e);
+      }
+    }
+  }
+
+  /**
+   * @param resourceAllocation NonNull NumaResourceAllocation object reference
+   * @return Array of numa specific commands
+   */
+  String[] getNumaCommands(NumaResourceAllocation resourceAllocation) {
+    String[] numaCommand = new String[3];
+    numaCommand[0] = this.getConf().get(YarnConfiguration.NM_NUMA_AWARENESS_NUMACTL_CMD,
+            YarnConfiguration.DEFAULT_NM_NUMA_AWARENESS_NUMACTL_CMD);
+    numaCommand[1] = "--interleave=" + String.join(",", resourceAllocation.getMemNodes());
+    numaCommand[2] = "--cpunodebind=" + String.join(",", resourceAllocation.getCpuNodes());
+    return numaCommand;
+
+  }
+
+  /**
+   * @param firstStringArray  Array of String
+   * @param secondStringArray Array of String
+   * @return combined array of string where first elements are from firstStringArray
+   * and later are the elements from secondStringArray
+   */
+  String[] concatStringCommands(String[] firstStringArray, String[] secondStringArray) {
+
+    if(firstStringArray == null && secondStringArray == null) {
+      return null;
+    }
+
+    int len = 0;
+
+    if(firstStringArray != null){
+      len = len + firstStringArray.length;
+    }
+
+    if(secondStringArray != null){
+      len = len + secondStringArray.length;
+    }
+
+    if (len == 0) {
+      return new String[]{};
+    }
+
+    String[] ret = new String[len];
+    int idx = 0;
+    for (int i=0; firstStringArray !=null && i < firstStringArray.length; i++) {
+      ret[idx] = firstStringArray[i];
+      idx++;
+    }
+    for (int i=0; secondStringArray !=null && i < secondStringArray.length; i++) {
+      ret[idx] = secondStringArray[i];
+      idx++;
+    }
+    return ret;
+  }
+
+  @VisibleForTesting
+  public void setNumaResourceAllocator(NumaResourceAllocator numaResourceAllocator) {
+    this.numaResourceAllocator = numaResourceAllocator;
+  }
+
 }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/WindowsSecureContainerExecutor.java

@@ -718,10 +718,10 @@ public void startLocalizer(LocalizerStartContext ctx) throws IOException,
   @Override
   protected CommandExecutor buildCommandExecutor(String wrapperScriptPath,
       String containerIdStr, String userName, Path pidFile, Resource resource,
-      File wordDir, Map<String, String> environment) {
+      File wordDir, Map<String, String> environment, String[] numaCommands) {
      return new WintuilsProcessStubExecutor(
          wordDir.toString(),
-         containerIdStr, userName, pidFile.toString(), 
+         containerIdStr, userName, pidFile.toString(),
          "cmd /c " + wrapperScriptPath);
    }
 

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/linux/resources/numa/NumaResourceAllocator.java

@@ -143,7 +143,7 @@ public void init(Configuration conf) throws YarnException {
   }
 
   @VisibleForTesting
-  String executeNGetCmdOutput(Configuration conf) throws YarnException {
+  public String executeNGetCmdOutput(Configuration conf) throws YarnException {
     String numaCtlCmd = conf.get(
         YarnConfiguration.NM_NUMA_AWARENESS_NUMACTL_CMD,
         YarnConfiguration.DEFAULT_NM_NUMA_AWARENESS_NUMACTL_CMD);