[core] (cgroups 10/n) Adding support in CgroupManager and CgroupDriver to move processes into system cgroup (#56446)

This PR stacks on #56352 .

For more details about the resource isolation project see
#54703.

This PR the following functions to move a process into the system
cgroup:
* CgroupManagerInterface::AddProcessToSystemCgroup
* CgroupDriverInterface::AddProcessToCgroup

I've also added integration tests for SysFsCgroupDriver and unit tests
for CgroupManager.

Let me explain how these APIs will be used. In the next PR, the raylet
will
* be passed a list of pids of system processes that are started before
the raylet starts and need to be moved into the system cgroup (e.g.
gcs_server)
* call CgroupManagerInterface::AddProcessToSystemCgroup for each of
these pids to move them into the system cgroup.

---------

Signed-off-by: Ibrahim Rabbani <[email protected]>
Co-authored-by: Edward Oakes <[email protected]>
Signed-off-by: elliot-barn <[email protected]>

Author: 2 people

src/ray/common/cgroup2/cgroup_driver_interface.h

@@ -206,5 +206,26 @@ class CgroupDriverInterface {
     */
   virtual StatusOr<std::unordered_set<std::string>> GetEnabledControllers(
       const std::string &cgroup) = 0;
+
+  /**
+    Adds the process to the specified cgroup.
+
+    To move the pid, the process must have read, write, and execute permissions for the
+      1) the cgroup the pid is currently in i.e. the source cgroup.
+      2) the destination cgroup.
+      3) the lowest common ancestor of the source and destination cgroups.
+
+    @param cgroup to move the process into.
+    @param pid of the process that will be moved.
+
+    @return Status::OK if the process was moved successfully into the cgroup.
+    @return Status::NotFound if the cgroup does not exist.
+    @return Status::PermissionDenied if process doesn't have read, write, and execute
+    permissions for the cgroup.
+    @return Status::InvalidArgument if the pid is invalid, does not exist, or any other
+    error.
+   */
+  virtual Status AddProcessToCgroup(const std::string &cgroup,
+                                    const std::string &pid) = 0;
 };
 }  // namespace ray

src/ray/common/cgroup2/cgroup_manager.cc

@@ -26,6 +26,7 @@
 #include "ray/common/cgroup2/cgroup_driver_interface.h"
 #include "ray/common/cgroup2/scoped_cgroup_operation.h"
 #include "ray/common/status_or.h"
+#include "ray/util/logging.h"
 
 namespace ray {
 
@@ -291,4 +292,22 @@ Status CgroupManager::Initialize(int64_t system_reserved_cpu_weight,
 
   return Status::OK();
 }
+
+Status CgroupManager::AddProcessToSystemCgroup(const std::string &pid) {
+  Status s = cgroup_driver_->AddProcessToCgroup(system_leaf_cgroup_, pid);
+  // TODO(#54703): Add link to OSS documentation once available.
+  RAY_CHECK(!s.IsNotFound()) << "Failed to move process " << pid << " into system cgroup "
+                             << system_leaf_cgroup_
+                             << "because the cgroup was not found. "
+                                "If resource isolation is enabled, Ray's cgroup "
+                                "hierarchy must not be modified "
+                                "while Ray is running.";
+  RAY_CHECK(!s.IsPermissionDenied())
+      << "Failed to move process " << pid << " into system cgroup " << system_leaf_cgroup_
+      << " because Ray does not have read, write, and execute "
+         "permissions for the cgroup. If resource isolation is enabled, Ray's cgroup "
+         "hierarchy must not be modified while Ray is running.";
+
+  return s;
+}
 }  // namespace ray

src/ray/common/cgroup2/cgroup_manager.h

@@ -69,6 +69,28 @@ class CgroupManager : public CgroupManagerInterface {
   CgroupManager(CgroupManager &&);
   CgroupManager &operator=(CgroupManager &&);
 
+  /**
+    Moves the process into the system leaf cgroup (@see
+    CgroupManagerInterface::kLeafCgroupName).
+
+    To move the pid, the process must have read, write, and execute permissions for the
+      1) the cgroup the pid is currently in i.e. the source cgroup.
+      2) the system leaf cgroup i.e. the destination cgroup.
+      3) the lowest common ancestor of the source and destination cgroups.
+
+    TODO(#54703): There currently is not a good way to signal to the caller that
+    the method can cause a FATAL error. Revisit this once we've settled on a pattern.
+
+    NOTE: If the process does not have adequate cgroup permissions or the system leaf
+    cgroup does not exist, this will fail a RAY_CHECK.
+
+    @param pid of the process to move into the system leaf cgroup.
+
+    @return Status::OK if pid moved successfully.
+    @return Status::NotFound if the system cgroup does not exist.
+  */
+  Status AddProcessToSystemCgroup(const std::string &pid) override;
+
   /**
     Performs cleanup in reverse order from the Initialize function:
       1. remove resource constraints to the system and application cgroups.

src/ray/common/cgroup2/cgroup_manager_interface.h

@@ -45,7 +45,27 @@ class CgroupManagerInterface {
   // TODO(#54703): These will be implemented in a later PR to move processes
   // into a cgroup.
   // virtual Status AddProcessToApplicationCgroup(int) = 0;
-  // virtual Status AddProcessToSystemCgroup(int) = 0;
+
+  /**
+    Moves the process into the system leaf cgroup (@see kLeafCgroupName).
+
+    To move the pid, the process must have read, write, and execute permissions for the
+      1) the cgroup the pid is currently in i.e. the source cgroup.
+      2) the system leaf cgroup i.e. the destination cgroup.
+      3) the lowest common ancestor of the source and destination cgroups.
+
+    TODO(#54703): There currently is not a good way to signal to the caller that
+    the method can cause a FATAL error. Revisit this once we've settled on a pattern.
+
+    NOTE: If the process does not have adequate cgroup permissions or the system leaf
+    cgroup does not exist, this will fail a RAY_CHECK.
+
+    @param pid of the process to move into the system leaf cgroup.
+
+    @return Status::OK if pid moved successfully.
+    @return Status::NotFound if the system cgroup does not exist.
+  */
+  virtual Status AddProcessToSystemCgroup(const std::string &pid) = 0;
 
   /**
     Cleans up the cgroup hierarchy, disables all controllers and removes all

src/ray/common/cgroup2/fake_cgroup_driver.h

@@ -126,6 +126,7 @@ class FakeCgroupDriver : public CgroupDriverInterface {
   Status add_constraint_s_ = Status::OK();
   Status available_controllers_s_ = Status::OK();
   Status enabled_controllers_s_ = Status::OK();
+  Status add_process_to_cgroup_s_ = Status::OK();
 
   // These have no side-effects.
   Status CheckCgroupv2Enabled() override { return check_cgroup_enabled_s_; }
@@ -222,6 +223,10 @@ class FakeCgroupDriver : public CgroupDriverInterface {
     }
     return (*cgroups_)[cgroup].enabled_controllers_;
   }
+
+  Status AddProcessToCgroup(const std::string &cgroup, const std::string &pid) override {
+    return add_process_to_cgroup_s_;
+  }
 };
 
 }  // namespace ray

src/ray/common/cgroup2/integration_tests/sysfs_cgroup_driver_integration_test.cc

@@ -622,4 +622,72 @@ TEST_F(SysFsCgroupDriverIntegrationTest, AddResourceConstraintSucceeds) {
   Status s = driver.AddConstraint(cgroup->GetPath(), "cpu", "cpu.weight", "500");
   ASSERT_TRUE(s.ok()) << s.ToString();
 }
+
+TEST_F(SysFsCgroupDriverIntegrationTest, AddProcessToCgroupFailsIfCgroupDoesNotExist) {
+  auto cgroup_or_status = TempCgroupDirectory::Create(test_cgroup_path_, S_IRWXU);
+  ASSERT_TRUE(cgroup_or_status.ok()) << cgroup_or_status.ToString();
+  auto cgroup = std::move(cgroup_or_status.value());
+  std::string non_existent_path =
+      cgroup->GetPath() + std::filesystem::path::preferred_separator + "nope";
+  SysFsCgroupDriver driver;
+  Status s = driver.AddProcessToCgroup(non_existent_path, "123");
+  ASSERT_TRUE(s.IsNotFound()) << s.ToString();
+}
+
+TEST_F(SysFsCgroupDriverIntegrationTest,
+       AddProcessToCgroupFailsIfCNotReadWriteExecPermissionsForCgroup) {
+  auto cgroup_or_status = TempCgroupDirectory::Create(test_cgroup_path_, S_IREAD);
+  ASSERT_TRUE(cgroup_or_status.ok()) << cgroup_or_status.ToString();
+  auto cgroup = std::move(cgroup_or_status.value());
+  SysFsCgroupDriver driver;
+  Status s = driver.AddProcessToCgroup(cgroup->GetPath(), "123");
+  ASSERT_TRUE(s.IsPermissionDenied()) << s.ToString();
+}
+
+TEST_F(SysFsCgroupDriverIntegrationTest, AddProcessToCgroupFailsIfProcessDoesNotExist) {
+  auto cgroup_or_status = TempCgroupDirectory::Create(test_cgroup_path_, S_IRWXU);
+  ASSERT_TRUE(cgroup_or_status.ok()) << cgroup_or_status.ToString();
+  auto cgroup = std::move(cgroup_or_status.value());
+  SysFsCgroupDriver driver;
+  Status s = driver.AddProcessToCgroup(cgroup->GetPath(), "123");
+  ASSERT_TRUE(s.IsInvalidArgument()) << s.ToString();
+}
+
+TEST_F(SysFsCgroupDriverIntegrationTest,
+       AddProcessToCgroupSucceedsIfProcessExistsAndCorrectPermissions) {
+  auto cgroup_or_status = TempCgroupDirectory::Create(test_cgroup_path_, S_IRWXU);
+  ASSERT_TRUE(cgroup_or_status.ok()) << cgroup_or_status.ToString();
+  auto cgroup = std::move(cgroup_or_status.value());
+  auto child_cgroup_or_status = TempCgroupDirectory::Create(cgroup->GetPath(), S_IRWXU);
+  ASSERT_TRUE(child_cgroup_or_status.ok()) << child_cgroup_or_status.ToString();
+  auto child_cgroup = std::move(child_cgroup_or_status.value());
+  StatusOr<std::pair<pid_t, int>> child_process_s =
+      StartChildProcessInCgroup(cgroup->GetPath());
+  ASSERT_TRUE(child_process_s.ok()) << child_process_s.ToString();
+  auto [child_pid, child_pidfd] = child_process_s.value();
+  SysFsCgroupDriver driver;
+  Status s =
+      driver.AddProcessToCgroup(child_cgroup->GetPath(), std::to_string(child_pid));
+  ASSERT_TRUE(s.ok()) << s.ToString();
+  // Assert that the child's pid is actually in the new file.
+  std::string child_cgroup_procs_file_path = child_cgroup->GetPath() +
+                                             std::filesystem::path::preferred_separator +
+                                             "cgroup.procs";
+  std::ifstream child_cgroup_procs_file(child_cgroup_procs_file_path);
+  ASSERT_TRUE(child_cgroup_procs_file.is_open())
+      << "Could not open file " << child_cgroup_procs_file_path << ".";
+  std::unordered_set<int> child_cgroup_pids;
+  int pid = -1;
+  while (child_cgroup_procs_file >> pid) {
+    ASSERT_FALSE(child_cgroup_procs_file.fail())
+        << "Unable to read pid from file " << child_cgroup_procs_file_path;
+    child_cgroup_pids.emplace(pid);
+  }
+  EXPECT_EQ(child_cgroup_pids.size(), 1);
+  EXPECT_TRUE(child_cgroup_pids.find(child_pid) != child_cgroup_pids.end());
+  Status terminate_s =
+      TerminateChildProcessAndWaitForTimeout(child_pid, child_pidfd, 5000);
+  ASSERT_TRUE(terminate_s.ok()) << terminate_s.ToString();
+}
+
 }  // namespace ray

src/ray/common/cgroup2/sysfs_cgroup_driver.cc

@@ -423,4 +423,35 @@ StatusOr<std::unordered_set<std::string>> SysFsCgroupDriver::ReadControllerFile(
   return StatusOr<std::unordered_set<std::string>>(controllers);
 }
 
+Status SysFsCgroupDriver::AddProcessToCgroup(const std::string &cgroup,
+                                             const std::string &process) {
+  RAY_RETURN_NOT_OK(CheckCgroup(cgroup));
+  std::filesystem::path cgroup_procs_file_path =
+      cgroup / std::filesystem::path(kCgroupProcsFilename);
+
+  int fd = open(cgroup_procs_file_path.c_str(), O_RDWR);
+
+  if (fd == -1) {
+    return Status::InvalidArgument(absl::StrFormat(
+        "Failed to write pid %s to cgroup.procs for cgroup %s with error %s",
+        process,
+        cgroup,
+        strerror(errno)));
+  }
+
+  ssize_t bytes_written = write(fd, process.c_str(), process.size());
+
+  if (bytes_written != static_cast<ssize_t>(process.size())) {
+    close(fd);
+    return Status::InvalidArgument(absl::StrFormat(
+        "Failed to write pid %s to cgroup.procs for cgroup %s with error %s",
+        process,
+        cgroup,
+        strerror(errno)));
+  }
+
+  close(fd);
+  return Status::OK();
+}
+
 }  // namespace ray

src/ray/common/cgroup2/sysfs_cgroup_driver.h

@@ -255,6 +255,29 @@ class SysFsCgroupDriver : public CgroupDriverInterface {
                        const std::string &constraint,
                        const std::string &constraint_value) override;
 
+  /**
+    Attempts to write pid to the cgroup.procs file of the specified cgroup.
+
+    To write a pid to a cgroup.procs file, the process must have read, write, and execute
+    to the source, destination, and lowest-common ancestor of source and destination
+    cgroups.
+
+    For more details, see the documentation:
+    - @see https://docs.kernel.org/admin-guide/cgroup-v2.html#delegation-containment
+    - @see https://docs.kernel.org/admin-guide/cgroup-v2.html#core-interface-files
+
+    @param cgroup to move the process into.
+    @param pid pid of the process that will be moved.
+
+    @return Status::OK if the process was moved successfully into the cgroup.
+    @return Status::NotFound if the cgroup does not exist.
+    @return Status::PermissionDenied if current user doesn't have read, write, and execute
+    permissions for the cgroup.
+    @return Status::InvalidArgument if the pid is invalid, does not exist, or any other
+    error.
+   */
+  Status AddProcessToCgroup(const std::string &cgroup, const std::string &pid) override;
+
  private:
   /**
     @param controller_file_path the absolute path of the controller file to read which is

src/ray/common/cgroup2/tests/cgroup_manager_test.cc

@@ -283,4 +283,85 @@ TEST(CgroupManagerTest, CreateSucceedsWithCleanupInOrder) {
   ASSERT_EQ((*deleted_cgroups)[4].second, node_cgroup_path);
 }
 
+TEST(CgroupManagerTest, AddProcessToSystemCgroupFailsIfInvalidProcess) {
+  std::shared_ptr<std::unordered_map<std::string, FakeCgroup>> cgroups =
+      std::make_shared<std::unordered_map<std::string, FakeCgroup>>();
+  cgroups->emplace("/sys/fs/cgroup",
+                   FakeCgroup{"/sys/fs/cgroup", {5}, {}, {"cpu", "memory"}, {}});
+  FakeCgroup base_cgroup{"/sys/fs/cgroup"};
+
+  std::unique_ptr<FakeCgroupDriver> driver = FakeCgroupDriver::Create(cgroups);
+  driver->add_process_to_cgroup_s_ = Status::InvalidArgument("");
+
+  auto cgroup_manager_s = CgroupManager::Create(
+      "/sys/fs/cgroup", "node_id_123", 100, 1000000, std::move(driver));
+  ASSERT_TRUE(cgroup_manager_s.ok()) << cgroup_manager_s.ToString();
+
+  std::unique_ptr<CgroupManager> cgroup_manager = std::move(cgroup_manager_s.value());
+  Status s = cgroup_manager->AddProcessToSystemCgroup("-1");
+  ASSERT_TRUE(s.IsInvalidArgument()) << s.ToString();
+}
+
+TEST(CgroupManagerTest, AddProcessToSystemCgroupIsFatalIfSystemCgroupDoesNotExist) {
+  std::shared_ptr<std::unordered_map<std::string, FakeCgroup>> cgroups =
+      std::make_shared<std::unordered_map<std::string, FakeCgroup>>();
+  cgroups->emplace("/sys/fs/cgroup",
+                   FakeCgroup{"/sys/fs/cgroup", {5}, {}, {"cpu", "memory"}, {}});
+  FakeCgroup base_cgroup{"/sys/fs/cgroup"};
+
+  std::unique_ptr<FakeCgroupDriver> driver = FakeCgroupDriver::Create(cgroups);
+  driver->add_process_to_cgroup_s_ = Status::NotFound("");
+
+  auto cgroup_manager_s = CgroupManager::Create(
+      "/sys/fs/cgroup", "node_id_123", 100, 1000000, std::move(driver));
+  ASSERT_TRUE(cgroup_manager_s.ok()) << cgroup_manager_s.ToString();
+
+  std::unique_ptr<CgroupManager> cgroup_manager = std::move(cgroup_manager_s.value());
+
+  EXPECT_DEATH((void)cgroup_manager->AddProcessToSystemCgroup("-1"),
+               "Failed to move.*not found");
+}
+
+TEST(CgroupManagerTest,
+     AddProcessToSystemCgroupIsFatalIfProcessDoesNotHavePermissionsForSystemCgroup) {
+  std::shared_ptr<std::unordered_map<std::string, FakeCgroup>> cgroups =
+      std::make_shared<std::unordered_map<std::string, FakeCgroup>>();
+  cgroups->emplace("/sys/fs/cgroup",
+                   FakeCgroup{"/sys/fs/cgroup", {5}, {}, {"cpu", "memory"}, {}});
+  FakeCgroup base_cgroup{"/sys/fs/cgroup"};
+
+  std::unique_ptr<FakeCgroupDriver> driver = FakeCgroupDriver::Create(cgroups);
+  driver->add_process_to_cgroup_s_ = Status::PermissionDenied("");
+
+  auto cgroup_manager_s = CgroupManager::Create(
+      "/sys/fs/cgroup", "node_id_123", 100, 1000000, std::move(driver));
+  ASSERT_TRUE(cgroup_manager_s.ok()) << cgroup_manager_s.ToString();
+
+  std::unique_ptr<CgroupManager> cgroup_manager = std::move(cgroup_manager_s.value());
+
+  EXPECT_DEATH((void)cgroup_manager->AddProcessToSystemCgroup("-1"),
+               "Failed to move.*permissions");
+}
+
+TEST(
+    CgroupManagerTest,
+    AddProcessToSystemCgroupSucceedsIfSystemCgroupExistsWithCorrectPermissionsAndValidProcess) {
+  std::shared_ptr<std::unordered_map<std::string, FakeCgroup>> cgroups =
+      std::make_shared<std::unordered_map<std::string, FakeCgroup>>();
+  cgroups->emplace("/sys/fs/cgroup",
+                   FakeCgroup{"/sys/fs/cgroup", {5}, {}, {"cpu", "memory"}, {}});
+  FakeCgroup base_cgroup{"/sys/fs/cgroup"};
+
+  std::unique_ptr<FakeCgroupDriver> driver = FakeCgroupDriver::Create(cgroups);
+
+  auto cgroup_manager_s = CgroupManager::Create(
+      "/sys/fs/cgroup", "node_id_123", 100, 1000000, std::move(driver));
+  ASSERT_TRUE(cgroup_manager_s.ok()) << cgroup_manager_s.ToString();
+
+  std::unique_ptr<CgroupManager> cgroup_manager = std::move(cgroup_manager_s.value());
+
+  Status s = cgroup_manager->AddProcessToSystemCgroup("5");
+  ASSERT_TRUE(s.ok()) << s.ToString();
+}
+
 }  // namespace ray