YARN-11687. Update CGroupsResourceCalculator to track usages using cgroupv2

- fix by teszt

Author: K0K0V0K

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/util/ResourceCalculatorProcessTree.java

@@ -167,7 +167,6 @@ public float getCpuUsagePercent() {
    */
   public static ResourceCalculatorProcessTree getResourceCalculatorProcessTree(
     String pid, Class<? extends ResourceCalculatorProcessTree> clazz, Configuration conf) {
-
     if (clazz != null) {
       try {
         Constructor <? extends ResourceCalculatorProcessTree> c = clazz.getConstructor(String.class);

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

@@ -559,6 +559,11 @@ public String getCGroupMountPath() {
     return this.cGroupsMountConfig.getMountPath();
   }
 
+  @Override
+  public String getCGroupV2MountPath() {
+    return this.cGroupsMountConfig.getV2MountPath();
+  }
+
   @Override
   public String toString() {
     return CGroupsHandlerImpl.class.getName() + "{" +

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

@@ -23,10 +23,11 @@
 import java.nio.charset.StandardCharsets;
 import java.nio.file.Path;
 import java.nio.file.Paths;
+import java.util.Arrays;
+import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
-import java.util.stream.Collectors;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -58,7 +59,7 @@ public abstract class AbstractCGroupsResourceCalculator extends ResourceCalculat
   private final String rssMemoryKey;
   private final String virtualMemoryKey;
 
-  public AbstractCGroupsResourceCalculator(
+  protected AbstractCGroupsResourceCalculator(
       String pid,
       List<String> totalJiffiesKeys,
       String rssMemoryKey,
@@ -79,22 +80,19 @@ public void initialize() throws YarnException {
 
   @Override
   public long getCumulativeCpuTime() {
-    //https://docs.kernel.org/admin-guide/cgroup-v2.html#cpu-interface-files
     long totalJiffies = getTotalJiffies();
-    return jiffyLengthMs < 0 || totalJiffies == UNAVAILABLE
+    return jiffyLengthMs == UNAVAILABLE || totalJiffies == UNAVAILABLE
         ? UNAVAILABLE
         : getTotalJiffies() * jiffyLengthMs;
   }
 
   @Override
   public long getRssMemorySize(int olderThanAge) {
-    //https://docs.kernel.org/admin-guide/cgroup-v2.html#memory-interface-files
     return 1 < olderThanAge ? UNAVAILABLE : getStat(rssMemoryKey);
   }
 
   @Override
   public long getVirtualMemorySize(int olderThanAge) {
-    //https://docs.kernel.org/admin-guide/cgroup-v2.html#memory-interface-files
     return 1 < olderThanAge ? UNAVAILABLE : getStat(virtualMemoryKey);
   }
 
@@ -117,70 +115,65 @@ public float getCpuUsagePercent() {
 
   @Override
   public void updateProcessTree() {
-
-    List<Path> filesToRead;
-    try {
-      filesToRead = getFilesToRead();
-    } catch (IOException e) {
-      LOG.error("Failed to get cgroup files to update for pid:" + pid, e);
-      return;
-    }
-
-    for (Path statFile : filesToRead) {
+    for (Path statFile : getFilesToRead()) {
       try {
-        String[] lines = fileToLines(statFile);
-        if (lines != null && 1 == lines.length) {
-          addSingleLineToStat(statFile, lines[0]);
-        } else if (lines != null && 1 < lines.length) {
+        List<String> lines = fileToLines(statFile);
+        if (1 == lines.size()) {
+          addSingleLineToStat(statFile, lines.get(0));
+        } else if (1 < lines.size()) {
           addMultiLineToStat(statFile, lines);
         }
       } catch (IOException e) {
-        LOG.warn("Failed to read cgroup files for pid: " + pid, e);
+        LOG.debug(String.format("Failed to read cgroup file %s for pid %s", statFile, pid), e);
       }
     }
-
+    LOG.debug("After updateProcessTree the {} pid has stats {}", pid, stats);
     cpuTimeTracker.updateElapsedJiffies(BigInteger.valueOf(getTotalJiffies()), clock.getTime());
   }
 
   private void addSingleLineToStat(Path file, String line) {
     stats.put(file.getFileName().toString(), line.trim());
   }
 
-  private void addMultiLineToStat(Path file, String[] lines) {
+  private void addMultiLineToStat(Path file, List<String> lines) {
     for (String line : lines) {
       String[] parts = line.split(" ");
-      stats.put(file.getFileName() + "#" + parts[0], parts[1]);
+      if (1 < parts.length) {
+        stats.put(file.getFileName() + "#" + parts[0], parts[1]);
+      }
     }
   }
 
   private long getTotalJiffies() {
-    List<Long> jiffies = totalJiffiesKeys.stream()
-        .map(this::getStat).collect(Collectors.toList());
-    return jiffies.stream().allMatch(stats -> stats == UNAVAILABLE)
-        ? UNAVAILABLE
-        : jiffies.stream().filter(stat -> UNAVAILABLE < stat).reduce(0L, Long::sum);
+    Long reduce = totalJiffiesKeys.stream()
+        .map(this::getStat)
+        .filter(stats -> stats != UNAVAILABLE)
+        .reduce(0L, Long::sum);
+    return reduce == 0 ? UNAVAILABLE : reduce;
   }
 
-  protected abstract List<Path> getFilesToRead() throws IOException;
-
   private long getStat(String key) {
     return Long.parseLong(stats.getOrDefault(key, String.valueOf(UNAVAILABLE)));
   }
 
-  protected String[] readLinesFromCGroupFileFromProcDir() throws IOException {
+  protected abstract List<Path> getFilesToRead();
+
+  protected List<String> readLinesFromCGroupFileFromProcDir() throws IOException {
     // https://docs.kernel.org/admin-guide/cgroup-v2.html#processes
-    return fileToString(Paths.get(procFs, pid, "cgroup"))
-        .split(System.lineSeparator());
+    Path cgroup = Paths.get(procFs, pid, "cgroup");
+    List<String> result = Arrays.asList(fileToString(cgroup).split(System.lineSeparator()));
+    LOG.debug("The {} pid has the following lines in the procfs cgroup file {}", pid, result);
+    return result;
   }
 
   protected String fileToString(Path path) throws IOException {
     return FileUtils.readFileToString(path.toFile(), StandardCharsets.UTF_8).trim();
   }
 
-  protected String[] fileToLines(Path path) throws IOException {
+  protected List<String> fileToLines(Path path) throws IOException {
     return path.toFile().exists()
-        ? FileUtils.readFileToString(path.toFile(), StandardCharsets.UTF_8).trim().split(System.lineSeparator())
-        : null;
+        ? Arrays.asList(FileUtils.readFileToString(path.toFile(), StandardCharsets.UTF_8).trim().split(System.lineSeparator()))
+        : Collections.emptyList();
   }
 
   @VisibleForTesting
@@ -203,10 +196,6 @@ void setProcFs(String procFs) {
     this.procFs = procFs;
   }
 
-  public CpuTimeTracker getCpuTimeTracker() {
-    return cpuTimeTracker;
-  }
-
   public CGroupsHandler getcGroupsHandler() {
     return cGroupsHandler;
   }

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

@@ -239,4 +239,10 @@ String getCGroupParam(CGroupController controller, String cGroupId,
    * @return parameter value as read from the parameter file
    */
   String getCGroupMountPath();
+
+  /**
+   * Returns CGroupV2 Mount Path.
+   * @return parameter value as read from the parameter file
+   */
+  String getCGroupV2MountPath();
 }

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

@@ -21,16 +21,22 @@
 import java.io.File;
 import java.io.IOException;
 import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
-import java.util.stream.Collectors;
-import java.util.stream.Stream;
 
-import org.apache.hadoop.yarn.exceptions.YarnException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
- * A cgroups file-system based Resource calculator without the process tree
- * features.
+ * A CGroupV1 file-system based Resource calculator without the process tree features.
+ *
+ * Warning!!!
+ * ResourceCalculatorProcessTree can be used with mapreduce.job.process-tree.class property.
+ * However, those instances runs in the mapreduce task, and can not access to the
+ * ResourceHandlerModule, what is only initialised in the NodeManager process not in the container.
+ * So this implementation will not work with the mapreduce.job.process-tree.class property.
  *
  * Limitation: CGroups does not have the ability to measure virtual memory usage.
  * This includes memory reserved but not used.
@@ -40,20 +46,50 @@
  * resource calculator or CombinedResourceCalculator.
  */
 public class CGroupsResourceCalculator extends AbstractCGroupsResourceCalculator {
-  static final String CPU_STAT = "cpuacct.stat";
-  static final String MEM_STAT = "memory.usage_in_bytes";
-  static final String MEMSW_STAT = "memory.memsw.usage_in_bytes";
+  private static final Logger LOG = LoggerFactory.getLogger(CGroupsResourceCalculator.class);
 
-  private File cpuStat;
-  private File memStat;
-  private File memswStat;
+  /**
+   * <a href="https://docs.kernel.org/admin-guide/cgroup-v1/cpuacct.html">Documentation</a>
+   *
+   * ...
+   * cpuacct.stat file lists a few statistics which further divide the CPU time obtained by the cgroup into user and system times.
+   * Currently the following statistics are supported:
+   *  - user: Time spent by tasks of the cgroup in user mode.
+   *  - system: Time spent by tasks of the cgroup in kernel mode.
+   * user and system are in USER_HZ unit.
+   *  ...
+   *
+   * <a href="https://litux.nl/mirror/kerneldevelopment/0672327201/ch10lev1sec3.html">Documentation</a>
+   *
+   * ...
+   * In kernels earlier than 2.6, changing the value of HZ resulted in user-space anomalies.
+   * This happened because values were exported to user-space in units of ticks-per-second.
+   * As these interfaces became permanent, applications grew to rely on a specific value of HZ.
+   * Consequently, changing HZ would scale various exported values by some constantwithout user-space knowing!
+   * Uptime would read 20 hours when it was in fact two!
+   *
+   * To prevent such problems, the kernel needs to scale all exported jiffies values.
+   * It does this by defining USER_HZ, which is the HZ value that user-space expects. On x86,
+   * because HZ was historically 100, USER_HZ is 100. The macro jiffies_to_clock_t()
+   * is then used to scale a tick count in terms of HZ to a tick count in terms of USER_HZ.
+   * The macro used depends on whether USER_HZ and HZ are integer multiples of themselves.
+   * ...
+   *
+   */
+  private static final String CPU_STAT = "cpuacct.stat";
 
   /**
-   * Create resource calculator for all Yarn containers.
+   * <a href="https://docs.kernel.org/admin-guide/cgroup-v1/memory.html#usage-in-bytes">Documentation</a>
+   *
+   * ...
+   * For efficiency, as other kernel components, memory cgroup uses some optimization to avoid unnecessary cacheline false sharing.
+   * usage_in_bytes is affected by the method and doesn’t show ‘exact’ value of memory (and swap) usage,
+   * it’s a fuzz value for efficient access. (Of course, when necessary, it’s synchronized.)
+   *  ...
+   *
    */
-  public CGroupsResourceCalculator() {
-    this(null);
-  }
+  private static final String MEM_STAT = "memory.usage_in_bytes";
+  private static final String MEMSW_STAT = "memory.memsw.usage_in_bytes";
 
   /**
    * Create resource calculator for the container that has the specified pid.
@@ -69,33 +105,41 @@ public CGroupsResourceCalculator(String pid) {
   }
 
   @Override
-  public void initialize() throws YarnException {
-    super.initialize();
+  protected List<Path> getFilesToRead() {
+    List<Path> result = new ArrayList<>();
+
     try {
-      setCGroupFilePaths();
+      String cpuRelative = getCGroupRelativePath(CGroupsHandler.CGroupController.CPUACCT);
+      if (cpuRelative != null) {
+        File cpuDir = new File(cGroupsHandler.getControllerPath(CGroupsHandler.CGroupController.CPUACCT), cpuRelative);
+        result.add(Paths.get(cpuDir.getAbsolutePath(), CPU_STAT));
+      }
     } catch (IOException e) {
-      throw new YarnException("Could not init cgroup files for pid: " + pid, e);
+      LOG.debug("Exception while looking for CPUACCT controller for pid: " + pid, e);
     }
-  }
 
-  @Override
-  protected List<Path> getFilesToRead() throws IOException {
-    return Stream.of(
-        cpuStat,
-        memStat,
-        memswStat
-    ).map(File::toPath).collect(Collectors.toList());
+    try {
+      String memoryRelative = getCGroupRelativePath(CGroupsHandler.CGroupController.MEMORY);
+      if (memoryRelative != null) {
+        File memDir = new File(cGroupsHandler.getControllerPath(CGroupsHandler.CGroupController.MEMORY), memoryRelative);
+        result.add(Paths.get(memDir.getAbsolutePath(), MEM_STAT));
+        result.add(Paths.get(memDir.getAbsolutePath(), MEMSW_STAT));
+      }
+    } catch (IOException e) {
+      LOG.debug("Exception while looking for MEMORY controller for pid: " + pid, e);
+    }
+
+    return result;
   }
 
   private String getCGroupRelativePath(CGroupsHandler.CGroupController controller)
-      throws YarnException, IOException {
+      throws IOException {
     return pid == null
         ? cGroupsHandler.getRelativePathForCGroup("")
         : getCGroupRelativePathForPid(controller);
   }
 
-  private String getCGroupRelativePathForPid(CGroupsHandler.CGroupController controller)
-      throws YarnException, IOException {
+  private String getCGroupRelativePathForPid(CGroupsHandler.CGroupController controller) throws IOException {
     for (String line : readLinesFromCGroupFileFromProcDir()) {
       // example line: 6:cpuacct,cpu:/yarn/container_1
       String[] parts = line.split(":");
@@ -105,19 +149,7 @@ private String getCGroupRelativePathForPid(CGroupsHandler.CGroupController contr
         return cGroupsHandler.getRelativePathForCGroup(cgroup);
       }
     }
-    throw new YarnException(String.format(
-        "Controller %s not found for %s pid", controller, pid));
-  }
-
-  void setCGroupFilePaths() throws YarnException, IOException {
-    File cpuDir = new File(
-        cGroupsHandler.getControllerPath(CGroupsHandler.CGroupController.CPUACCT),
-        getCGroupRelativePath(CGroupsHandler.CGroupController.CPUACCT));
-    File memDir = new File(
-        cGroupsHandler.getControllerPath(CGroupsHandler.CGroupController.MEMORY),
-        getCGroupRelativePath(CGroupsHandler.CGroupController.MEMORY));
-    cpuStat = new File(cpuDir, CPU_STAT);
-    memStat = new File(memDir, MEM_STAT);
-    memswStat = new File(memDir, MEMSW_STAT);
+    LOG.debug("No {} controller found for pid {}", controller, pid);
+    return null;
   }
 }