CDPD-75272 YARN-11709. NodeManager should be marked unhealthy on localizer config issues (apache#7043) (apache#159)

(cherry picked from commit d1311e5)

Co-authored-by: Benjamin Teke <[email protected]>
(cherry picked from commit 468b7fcda0d8c3e8dd1c28cb913d51dd730c83ff)

Author: zmolnar-inf

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

@@ -164,9 +164,10 @@ public Path localizeClasspathJar(Path jarPath, Path target, String owner)
    *            for starting a localizer.
    * @throws IOException for most application init failures
    * @throws InterruptedException if application init thread is halted by NM
+   * @throws ConfigurationException if config error was found
    */
   public abstract void startLocalizer(LocalizerStartContext ctx)
-    throws IOException, InterruptedException;
+      throws IOException, InterruptedException, ConfigurationException;
 
   /**
    * Prepare the container prior to the launch environment being written.

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

@@ -344,7 +344,7 @@ public void init(Context context) throws IOException {
 
   @Override
   public void startLocalizer(LocalizerStartContext ctx)
-      throws IOException, InterruptedException {
+      throws IOException, InterruptedException, ConfigurationException {
     Path nmPrivateContainerTokensPath = ctx.getNmPrivateContainerTokens();
     InetSocketAddress nmAddr = ctx.getNmAddr();
     String user = ctx.getUser();
@@ -395,9 +395,9 @@ public void startLocalizer(LocalizerStartContext ctx)
     localizerArgs = replaceWithContainerLogDir(localizerArgs, containerLogDir);
 
     initializeContainerOp.appendArgs(localizerArgs);
+    Configuration conf = super.getConf();
 
     try {
-      Configuration conf = super.getConf();
       PrivilegedOperationExecutor privilegedOperationExecutor =
           getPrivilegedOperationExecutor();
 
@@ -409,6 +409,25 @@ public void startLocalizer(LocalizerStartContext ctx)
       LOG.warn("Exit code from container " + locId + " startLocalizer is : "
           + exitCode, e);
 
+      if (exitCode ==
+          ExitCode.INVALID_CONTAINER_EXEC_PERMISSIONS.getExitCode() ||
+          exitCode == ExitCode.INVALID_CONFIG_FILE.getExitCode()) {
+        throw new ConfigurationException("Application " + appId + " initialization failed" +
+            " (exitCode=" + exitCode + ") with an unrecoverable config error. " +
+            "Output: " + e.getOutput(), e);
+      }
+
+      // Check if the failure was due to a missing container-executor binary
+      Throwable cause = e.getCause() != null ? e.getCause() : e;
+      if (cause instanceof IOException) {
+        IOException io = (IOException) cause;
+        if (io.getMessage().contains("No such file or directory")) {
+          throw new ConfigurationException("Application " + appId + " initialization failed" +
+              "(exitCode=" + exitCode + "). Container executor not found at "
+              + getContainerExecutorExecutablePath(conf), e);
+        }
+      }
+
       throw new IOException("Application " + appId + " initialization failed" +
           " (exitCode=" + exitCode + ") with output: " + e.getOutput(), e);
     }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/localizer/ResourceLocalizationService.java

@@ -20,6 +20,7 @@
 import static org.apache.hadoop.fs.CreateFlag.CREATE;
 import static org.apache.hadoop.fs.CreateFlag.OVERWRITE;
 
+import org.apache.hadoop.yarn.exceptions.ConfigurationException;
 import org.apache.hadoop.yarn.server.nodemanager.recovery.RecoveryIterator;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -1260,10 +1261,13 @@ public void run() {
           throw new IOException("All disks failed. "
               + dirsHandler.getDisksHealthReport(false));
         }
-      // TODO handle ExitCodeException separately?
-      } catch (FSError fe) {
-        exception = fe;
-      } catch (Exception e) {
+        // TODO handle ExitCodeException separately?
+      } catch (ConfigurationException e) {
+        exception = e;
+        LOG.error("Failed to launch localizer for " + localizerId + ", due to configuration error. " +
+            "Marking the node unhealthy.", e);
+        nmContext.getNodeStatusUpdater().reportException(e);
+      } catch (Exception | FSError e) {
         exception = e;
       } finally {
         if (exception != null) {

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/TestLinuxContainerExecutorWithMocks.java

@@ -335,7 +335,7 @@ public void testStartLocalizer() throws IOException {
       Assert.assertEquals(result.get(22), "localhost");
       Assert.assertEquals(result.get(23), "8040");
 
-    } catch (InterruptedException e) {
+    } catch (ConfigurationException | InterruptedException e) {
       LOG.error("Error:"+e.getMessage(),e);
       Assert.fail();
     }
@@ -642,6 +642,61 @@ protected PrivilegedOperationExecutor getPrivilegedOperationExecutor() {
           e.getMessage().contains("exitCode"));
     }
 
+    final int[] exitCodesToThrow = {
+        LinuxContainerExecutor.ExitCode.INVALID_CONTAINER_EXEC_PERMISSIONS.getExitCode(),
+        LinuxContainerExecutor.ExitCode.INVALID_CONFIG_FILE.getExitCode(),
+    };
+
+    for (int i = 0; i < exitCodesToThrow.length; i++) {
+      int exitCode = exitCodesToThrow[i];
+      doThrow(new PrivilegedOperationException("invalid config", exitCode, null, null))
+          .when(spyPrivilegedExecutor).executePrivilegedOperation(
+              any(), any(PrivilegedOperation.class),
+              any(), any(), anyBoolean(), anyBoolean());
+
+      try {
+        lce.startLocalizer(new LocalizerStartContext.Builder()
+            .setNmPrivateContainerTokens(nmPrivateCTokensPath)
+            .setNmAddr(address)
+            .setUser(appSubmitter)
+            .setAppId(appId.toString())
+            .setLocId("12345")
+            .setDirsHandler(dirService)
+            .build());
+        Assert.fail("startLocalizer should have thrown a ConfigurationException");
+      } catch (ConfigurationException e) {
+        assertTrue("Unexpected exception " + e,
+            e.getMessage().contains("exitCode=" + exitCode));
+      }
+    }
+
+    doThrow(new PrivilegedOperationException("IO error",
+        new IOException("No such file or directory")))
+        .when(spyPrivilegedExecutor).executePrivilegedOperation(
+            any(), any(PrivilegedOperation.class),
+            any(), any(), anyBoolean(), anyBoolean());
+
+    try {
+      lce.startLocalizer(new LocalizerStartContext.Builder()
+          .setNmPrivateContainerTokens(nmPrivateCTokensPath)
+          .setNmAddr(address)
+          .setUser(appSubmitter)
+          .setAppId(appId.toString())
+          .setLocId("12345")
+          .setDirsHandler(dirService)
+          .build());
+      Assert.fail("startLocalizer should have thrown a ConfigurationException");
+    } catch (ConfigurationException e) {
+      assertTrue("Unexpected exception " + e,
+          e.getMessage().contains("Container executor not found"));
+    }
+
+
+    doThrow(new PrivilegedOperationException("interrupted"))
+        .when(spyPrivilegedExecutor).executePrivilegedOperation(
+            any(), any(PrivilegedOperation.class),
+            any(), any(), anyBoolean(), anyBoolean());
+
     lce.activateContainer(cid, new Path(workDir, "pid.txt"));
     lce.launchContainer(new ContainerStartContext.Builder()
         .setContainer(container)