YARN-11687. WIP

Author: K0K0V0K

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

@@ -0,0 +1,168 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.yarn.server.nodemanager.containermanager.linux.resources;
+
+import org.apache.commons.io.FileUtils;
+import org.apache.commons.lang3.StringUtils;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.apache.hadoop.util.CpuTimeTracker;
+import org.apache.hadoop.util.Shell;
+import org.apache.hadoop.util.SysInfoLinux;
+import org.apache.hadoop.yarn.exceptions.YarnException;
+import org.apache.hadoop.yarn.util.Clock;
+import org.apache.hadoop.yarn.util.ResourceCalculatorProcessTree;
+import org.apache.hadoop.yarn.util.SystemClock;
+
+import java.io.File;
+import java.io.IOException;
+import java.math.BigInteger;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+/**
+ * A CGroupV2 file-system based Resource calculator without the process tree features.
+ *
+ * The feature only works if cluster runs in pure V2 version, because when we read the
+ * /proc/{pid}/cgroup file currently we can not handle multiple lines.
+ *
+ * Another limitation is virtual memory measurement. CGroups does not have the
+ * ability to measure virtual memory usage. This includes memory reserved but
+ * not used. CGroups measures used memory as sa sum of
+ * physical memory and swap usage. This will be returned in the virtual
+ * memory counters.
+ * If the real virtual memory is required please use the legacy procfs based
+ * resource calculator or CombinedResourceCalculator.
+ */
+public class CGroupsV2ResourceCalculator extends ResourceCalculatorProcessTree {
+  private static final Logger LOG = LoggerFactory.getLogger(CGroupsV2ResourceCalculator.class);
+  private final Map<String, String> stats = new ConcurrentHashMap<>();
+  private final Clock clock = SystemClock.getInstance();
+  private final String pid;
+
+  private long jiffyLengthMs = SysInfoLinux.JIFFY_LENGTH_IN_MILLIS;
+  private CpuTimeTracker cpuTimeTracker;
+
+  /**
+   * Create resource calculator for the container that has the specified pid.
+   * @param pid A pid from the cgroup or null for all containers
+   */
+  public CGroupsV2ResourceCalculator(String pid) {
+    super(pid);
+    this.pid = pid;
+  }
+
+  @Override
+  public void initialize() throws YarnException {
+    jiffyLengthMs = SysInfoLinux.JIFFY_LENGTH_IN_MILLIS;
+    cpuTimeTracker = new CpuTimeTracker(jiffyLengthMs);
+  }
+
+  @Override
+  public float getCpuUsagePercent() {
+    return cpuTimeTracker.getCpuTrackerUsagePercent();
+  }
+
+  @Override
+  public long getCumulativeCpuTime() {
+    // https://docs.kernel.org/admin-guide/cgroup-v2.html#cpu-interface-files
+    return jiffyLengthMs < 0
+        ? UNAVAILABLE
+        : getStat("cpu.stat#usage_usec") * jiffyLengthMs;
+  }
+
+  @Override
+  public long getRssMemorySize(int olderThanAge) {
+    // https://docs.kernel.org/admin-guide/cgroup-v2.html#memory-interface-files
+    return 1 < olderThanAge
+        ? UNAVAILABLE
+        : getStat("memory.stat#anon");
+  }
+
+  @Override
+  public long getVirtualMemorySize(int olderThanAge) {
+    // https://docs.kernel.org/admin-guide/cgroup-v2.html#memory-interface-files
+    return 1 < olderThanAge
+        ? UNAVAILABLE
+        : getStat("memory.stat#vmalloc");
+  }
+
+  @Override
+  public String getProcessTreeDump() {
+    // We do not have a process tree in cgroups return just the pid for tracking
+    return pid;
+  }
+
+  @Override
+  public boolean checkPidPgrpidForMatch() {
+    // We do not have a process tree in cgroups returning default ok
+    return true;
+  }
+
+  @Override
+  public void updateProcessTree() {
+    try (Stream<Path> cGroupFiles = Files.list(getCGroupPath())){
+      List<Path> statFiles = cGroupFiles
+          .filter(path -> path.endsWith(".stat"))
+          .collect(Collectors.toList());
+      for (Path statFile : statFiles) {
+        String[] lines = fileToString(statFile.toString()).split(System.lineSeparator());
+        for (String line: lines) {
+          String[] parts = line.split(" ");
+          stats.put(statFile.getFileName() + "#" + parts[0], parts[1]);
+        }
+      }
+      cpuTimeTracker.updateElapsedJiffies(
+          BigInteger.valueOf(getStat("cpu.stat#usage_usec")), clock.getTime());
+      LOG.debug("The {} process has the following stat properties updated: {}", pid, stats);
+    } catch (IOException e) {
+      LOG.warn("Failed to read CGroupV2 stats for process: " + pid, e);
+    }
+  }
+
+  private Path getCGroupPath() throws IOException {
+    String relativePath =  pid == null
+        ? ResourceHandlerModule.getCGroupsHandler().getRelativePathForCGroup("")
+        : getCGroupRelativePathForPid();
+    return Paths.get(ResourceHandlerModule.getCGroupsHandler().getCGroupMountPath(), relativePath);
+  }
+
+  private String getCGroupRelativePathForPid() throws IOException {
+    // https://docs.kernel.org/admin-guide/cgroup-v2.html#processes
+    String pidCGroupsFile = fileToString(Paths.get("/proc", pid, "cgroup").toString());
+    String controllerPath = StringUtils.substringAfterLast(pidCGroupsFile, ":");
+    return ResourceHandlerModule.getCGroupsHandler().getRelativePathForCGroup(controllerPath);
+  }
+
+  private long getStat(String key) {
+    return Long.parseLong(stats.getOrDefault(key, "0"));
+  }
+
+  private String fileToString(String path) throws IOException {
+    return FileUtils.readFileToString(new File(path), StandardCharsets.UTF_8);
+  }
+}