[Core] Fix OOM kill msg wrong threshold w/ resource isolation (ray-project#62948)

## Description
The OOM kill message re-computed the memory threshold at kill time via
`GetMemoryThreshold()`, which under `--enable-resource-isolation` could
read a different cgroup `memory.max` value than at init time (e.g. "max"
instead of digits), silently falling back to 0.95. Fix: have each
monitor pass the threshold it actually fired on through the callback
instead of letting NodeManager guess.

Now the OOM msg should be something like this

```
  Memory on the node (IP: 10.0.0.1, ID: abc123) was 7.20GB / 8.00GB (0.900000);
  OOM kill reason: Memory usage 7728742400B exceeded threshold of 6871947674B (85.9% of 8589934592B total);
  Object store memory usage: [...];
  Ray killed 1 worker(s) based on the killing policy: [...];
```

## Related issues
> Link related issues: "Fixes ray-project#1234", "Closes ray-project#1234", or "Related to
ray-project#1234".

## Additional information
> Optional: Add implementation details, API changes, usage examples,
screenshots, etc.

---------

Signed-off-by: You-Cheng Lin <mses010108@gmail.com>
Signed-off-by: You-Cheng Lin <106612301+owenowenisme@users.noreply.github.com>
Signed-off-by: phattruong <23120318@student.hcmus.edu.vn>

Author: owenowenisme

src/ray/common/event_memory_monitor.cc

@@ -192,7 +192,7 @@ void EventMemoryMonitor::MonitoringThreadMain() {
 
       if (high_modified && IsEnabled()) {
         Disable();
-        kill_workers_callback_();
+        kill_workers_callback_("user cgroup memory upper bound was met or exceeded");
       }
     } else {
       RAY_LOG(ERROR) << absl::StrFormat(

src/ray/common/memory_monitor_interface.h

@@ -20,6 +20,7 @@
 #include <functional>
 #include <optional>
 #include <ostream>
+#include <string>
 
 #include "absl/container/flat_hash_map.h"
 #include "ray/util/compat.h"
@@ -70,8 +71,10 @@ using ProcessesMemorySnapshot = absl::flat_hash_map<pid_t, int64_t>;
 
 /**
  * @brief Callback to trigger worker oom killing when under memory pressure.
+ * @param trigger_reason A human-readable description of why the monitor triggered
+ *        the kill (e.g. threshold exceeded, cgroup event, PSI pressure).
  */
-using KillWorkersCallback = std::function<void()>;
+using KillWorkersCallback = std::function<void(std::string trigger_reason)>;
 
 /**
  * @brief implementations of this interface monitors the memory usage of the node

src/ray/common/pressure_memory_monitor.cc

@@ -161,7 +161,7 @@ void PressureMemoryMonitor::MonitoringThreadMain() {
     if (fds[0].revents & POLLPRI) {
       if (IsEnabled()) {
         Disable();
-        kill_workers_callback_();
+        kill_workers_callback_("user cgroup memory pressure was detected");
       }
     } else if (fds[0].revents & POLLERR) {
       RAY_LOG(ERROR) << "Got POLLERR while monitoring memory pressure. "

src/ray/common/tests/event_memory_monitor_test.cc

@@ -66,7 +66,7 @@ class EventMemoryMonitorTest : public ::testing::Test {
 TEST_F(EventMemoryMonitorTest, TestNonexistentCgroupPathFailsGracefully) {
   std::string nonexistent_path = "/nonexistent/cgroup/path";
   StatusSetOr<std::unique_ptr<EventMemoryMonitor>, StatusT::IOError> result =
-      EventMemoryMonitor::Create(std::move(nonexistent_path), []() {});
+      EventMemoryMonitor::Create(std::move(nonexistent_path), [](std::string) {});
 
   ASSERT_TRUE(result.has_error())
       << "Failed to catch invalid cgroup path when creating EventMemoryMonitor";
@@ -78,7 +78,7 @@ TEST_F(EventMemoryMonitorTest, TestMissingMemoryEventsFileFailsGracefully) {
   RAY_CHECK(empty_dir_or.ok()) << empty_dir_or.status().ToString();
   std::unique_ptr<TempDirectory> empty_dir = std::move(empty_dir_or.value());
   StatusSetOr<std::unique_ptr<EventMemoryMonitor>, StatusT::IOError> result =
-      EventMemoryMonitor::Create(empty_dir->GetPath(), []() {});
+      EventMemoryMonitor::Create(empty_dir->GetPath(), [](std::string) {});
 
   ASSERT_TRUE(result.has_error())
       << "Failed to catch invalid cgroup configuration when creating EventMemoryMonitor";
@@ -87,7 +87,7 @@ TEST_F(EventMemoryMonitorTest, TestMissingMemoryEventsFileFailsGracefully) {
 
 TEST_F(EventMemoryMonitorTest, TestSuccessfulCreationWithValidPath) {
   StatusSetOr<std::unique_ptr<EventMemoryMonitor>, StatusT::IOError> result =
-      EventMemoryMonitor::Create(mock_cgroup_dir_->GetPath(), []() {});
+      EventMemoryMonitor::Create(mock_cgroup_dir_->GetPath(), [](std::string) {});
   ASSERT_TRUE(result.has_value())
       << "Failed to create EventMemoryMonitor: " << result.message();
   std::unique_ptr<EventMemoryMonitor> monitor = std::move(result.value());
@@ -98,7 +98,9 @@ TEST_F(EventMemoryMonitorTest, TestCallbackCalledWhenHighEventChanges) {
   WriteMemoryEventsFile(events_file_->GetPath(), 0, 0);
 
   auto callback_latch = std::make_shared<boost::latch>(1);
-  KillWorkersCallback callback = [callback_latch]() { callback_latch->count_down(); };
+  KillWorkersCallback callback = [callback_latch](std::string) {
+    callback_latch->count_down();
+  };
 
   StatusSetOr<std::unique_ptr<EventMemoryMonitor>, StatusT::IOError> result =
       EventMemoryMonitor::Create(std::move(mock_cgroup_dir_->GetPath()),
@@ -116,7 +118,9 @@ TEST_F(EventMemoryMonitorTest, TestNoCallbackWhenValuesUnchanged) {
   WriteMemoryEventsFile(events_file_->GetPath(), 0, 0);
 
   auto callback_latch = std::make_shared<boost::latch>(1);
-  KillWorkersCallback callback = [callback_latch]() { callback_latch->count_down(); };
+  KillWorkersCallback callback = [callback_latch](std::string) {
+    callback_latch->count_down();
+  };
 
   StatusSetOr<std::unique_ptr<EventMemoryMonitor>, StatusT::IOError> result =
       EventMemoryMonitor::Create(mock_cgroup_dir_->GetPath(), callback);
@@ -133,7 +137,9 @@ TEST_F(EventMemoryMonitorTest, TestNoCallbackWhenIrrelevantEventChanges) {
   WriteMemoryEventsFile(events_file_->GetPath(), 0, 0);
 
   auto callback_latch = std::make_shared<boost::latch>(1);
-  KillWorkersCallback callback = [callback_latch]() { callback_latch->count_down(); };
+  KillWorkersCallback callback = [callback_latch](std::string) {
+    callback_latch->count_down();
+  };
 
   StatusSetOr<std::unique_ptr<EventMemoryMonitor>, StatusT::IOError> result =
       EventMemoryMonitor::Create(mock_cgroup_dir_->GetPath(), callback);
@@ -153,7 +159,7 @@ TEST_F(EventMemoryMonitorTest, TestMultipleCallbacksOnMultipleChanges) {
   auto latch2 = std::make_shared<boost::latch>(1);
   auto latch3 = std::make_shared<boost::latch>(1);
   std::atomic<int> callback_count{0};
-  KillWorkersCallback callback = [&callback_count, latch1, latch2, latch3]() {
+  KillWorkersCallback callback = [&callback_count, latch1, latch2, latch3](std::string) {
     int count = ++callback_count;
     if (count == 1) {
       latch1->count_down();

src/ray/common/tests/memory_monitor_factory_test.cc

@@ -78,7 +78,7 @@ TEST_F(MemoryMonitorFactoryTest,
   FakeCgroupManager cgroup_manager(kUserMemoryMaxBytes, kUserMemoryHighBytes);
 
   std::vector<std::unique_ptr<MemoryMonitorInterface>> monitors =
-      MemoryMonitorFactory::Create([]() {},
+      MemoryMonitorFactory::Create([](std::string) {},
                                    /*resource_isolation_enabled=*/false,
                                    cgroup_manager);
 
@@ -93,7 +93,7 @@ TEST_F(MemoryMonitorFactoryTest,
   TempFile pressure_file(cgroup_manager.GetPath() + "/memory.pressure");
 
   std::vector<std::unique_ptr<MemoryMonitorInterface>> monitors =
-      MemoryMonitorFactory::Create([]() {},
+      MemoryMonitorFactory::Create([](std::string) {},
                                    /*resource_isolation_enabled=*/true,
                                    cgroup_manager);
 

src/ray/common/tests/pressure_memory_monitor_test.cc

@@ -99,7 +99,8 @@ TEST_F(PressureMemoryMonitorTest, TestInvalidStallDurationReturnsFalse) {
 TEST_F(PressureMemoryMonitorTest, TestNonexistentCgroupPathFailsGracefully) {
   MemoryPsi psi = {.mode = "some", .stall_proportion = 0.5f, .stall_duration_s = 2};
   std::string nonexistent_path = "/nonexistent/cgroup/path";
-  auto result = PressureMemoryMonitor::Create(psi, std::move(nonexistent_path), []() {});
+  auto result =
+      PressureMemoryMonitor::Create(psi, std::move(nonexistent_path), [](std::string) {});
 
   ASSERT_TRUE(result.has_error())
       << "Failed to catch invalid cgroup path when creating PressureMemoryMonitor";
@@ -109,7 +110,8 @@ TEST_F(PressureMemoryMonitorTest, TestNonexistentCgroupPathFailsGracefully) {
 TEST_F(PressureMemoryMonitorTest, TestMonitorCreationWritesTriggerStringToFile) {
   MemoryPsi psi = {.mode = "some", .stall_proportion = 0.5f, .stall_duration_s = 2};
 
-  auto result = PressureMemoryMonitor::Create(psi, mock_cgroup_dir_->GetPath(), []() {});
+  auto result =
+      PressureMemoryMonitor::Create(psi, mock_cgroup_dir_->GetPath(), [](std::string) {});
   ASSERT_TRUE(result.has_value())
       << "Failed to create PressureMemoryMonitor: " << result.message();
 
@@ -162,7 +164,9 @@ TEST_F(PressureMemoryMonitorTest,
   close(listener);
 
   std::shared_ptr<boost::latch> has_called_once = std::make_shared<boost::latch>(1);
-  auto kill_workers_callback = [has_called_once]() { has_called_once->count_down(); };
+  auto kill_workers_callback = [has_called_once](std::string) {
+    has_called_once->count_down();
+  };
   std::unique_ptr<PressureMemoryMonitor> monitor =
       std::make_unique<PressureMemoryMonitor>(
           mock_cgroup_dir_->GetPath(), listener_fd, kill_workers_callback);

src/ray/common/tests/threshold_memory_monitor_test.cc

@@ -75,7 +75,7 @@ TEST_F(ThresholdMemoryMonitorTest, TestMonitorTriggerCanDetectMemoryUsage) {
   MakeThresholdMemoryMonitor(
       0 /*memory_usage_threshold_bytes*/,
       1 /*refresh_interval_ms*/,
-      [has_checked_once]() { has_checked_once->count_down(); },
+      [has_checked_once](std::string) { has_checked_once->count_down(); },
       "" /*root_cgroup_path*/);
   has_checked_once->wait();
 }
@@ -105,7 +105,7 @@ TEST_F(ThresholdMemoryMonitorTest,
   MakeThresholdMemoryMonitor(
       memory_usage_threshold_bytes,  // (70%)
       1 /*refresh_interval_ms*/,
-      [has_checked_once]() { has_checked_once->count_down(); },
+      [has_checked_once](std::string) { has_checked_once->count_down(); },
       cgroup_dir /*root_cgroup_path*/);
 
   has_checked_once->wait();
@@ -137,7 +137,7 @@ TEST_F(ThresholdMemoryMonitorTest,
   MakeThresholdMemoryMonitor(
       memory_usage_threshold_bytes,  // (70%)
       1 /*refresh_interval_ms*/,
-      [callback_triggered]() { callback_triggered->store(true); },
+      [callback_triggered](std::string) { callback_triggered->store(true); },
       cgroup_dir /*root_cgroup_path*/);
 
   std::this_thread::sleep_for(std::chrono::seconds(5));
@@ -172,7 +172,7 @@ TEST_F(ThresholdMemoryMonitorTest,
   MakeResourceIsolatedThresholdMemoryMonitor(
       threshold_bytes,
       1 /*refresh_interval_ms*/,
-      [has_checked_once]() { has_checked_once->count_down(); },
+      [has_checked_once](std::string) { has_checked_once->count_down(); },
       "" /*root_cgroup_path*/,
       user_cgroup_dir,
       system_cgroup_dir);
@@ -213,7 +213,7 @@ TEST_F(
   MakeResourceIsolatedThresholdMemoryMonitor(
       threshold_bytes,
       1 /*refresh_interval_ms*/,
-      [has_checked_once]() { has_checked_once->count_down(); },
+      [has_checked_once](std::string) { has_checked_once->count_down(); },
       "" /*root_cgroup_path*/,
       user_cgroup_dir,
       system_cgroup_dir);
@@ -253,7 +253,7 @@ TEST_F(ThresholdMemoryMonitorTest,
   MakeResourceIsolatedThresholdMemoryMonitor(
       threshold_bytes,
       1 /*refresh_interval_ms*/,
-      [callback_triggered]() { callback_triggered->store(true); },
+      [callback_triggered](std::string) { callback_triggered->store(true); },
       "" /*root_cgroup_path*/,
       user_cgroup_dir,
       system_cgroup_dir);

src/ray/common/threshold_memory_monitor.cc

@@ -62,16 +62,26 @@ ThresholdMemoryMonitor::ThresholdMemoryMonitor(KillWorkersCallback kill_workers_
 
   runner_->RunFnPeriodically(
       [this] {
-        bool is_usage_above_threshold = false;
+        std::optional<MemoryUsageSnapshot> exceeded_snapshot;
         if (resource_isolation_enabled_) {
-          is_usage_above_threshold = IsResourceIsolationThresholdExceeded();
+          exceeded_snapshot = IsResourceIsolationThresholdExceeded();
         } else {
-          is_usage_above_threshold = IsHostMemoryThresholdExceeded();
+          exceeded_snapshot = IsHostMemoryThresholdExceeded();
         }
 
-        if (is_usage_above_threshold && IsEnabled()) {
+        if (exceeded_snapshot.has_value() && IsEnabled()) {
+          const MemoryUsageSnapshot &cur_memory_snapshot = exceeded_snapshot.value();
           Disable();
-          kill_workers_callback_();
+          std::string trigger_reason = absl::StrFormat(
+              "Memory usage %dB exceeded threshold of %dB (%.1f%% of %dB total)",
+              cur_memory_snapshot.used_bytes,
+              memory_usage_threshold_bytes_,
+              (cur_memory_snapshot.total_bytes > 0
+                   ? static_cast<float>(memory_usage_threshold_bytes_) /
+                         static_cast<float>(cur_memory_snapshot.total_bytes) * 100
+                   : 0.0f),
+              cur_memory_snapshot.total_bytes);
+          kill_workers_callback_(trigger_reason);
         }
       },
       monitor_interval_ms,
@@ -94,7 +104,8 @@ bool ThresholdMemoryMonitor::IsEnabled() const {
   return !worker_killing_in_progress_.load();
 }
 
-bool ThresholdMemoryMonitor::IsHostMemoryThresholdExceeded() {
+std::optional<MemoryUsageSnapshot>
+ThresholdMemoryMonitor::IsHostMemoryThresholdExceeded() {
   MemoryUsageSnapshot cur_memory_snapshot =
       MemoryMonitorUtils::TakeSystemMemoryUsageSnapshot(root_cgroup_path_);
   int64_t used_memory_bytes = cur_memory_snapshot.used_bytes;
@@ -104,7 +115,7 @@ bool ThresholdMemoryMonitor::IsHostMemoryThresholdExceeded() {
     RAY_LOG_EVERY_MS(WARNING, MemoryMonitorInterface::kLogIntervalMs)
         << "Unable to capture node memory. Monitor will not be able "
         << "to detect memory usage above threshold.";
-    return false;
+    return std::nullopt;
   }
   bool is_usage_above_threshold = used_memory_bytes > memory_usage_threshold_bytes_;
   if (is_usage_above_threshold) {
@@ -117,10 +128,13 @@ bool ThresholdMemoryMonitor::IsHostMemoryThresholdExceeded() {
         static_cast<float>(memory_usage_threshold_bytes_) /
             static_cast<float>(total_memory_bytes));
   }
-  return is_usage_above_threshold;
+  return is_usage_above_threshold
+             ? std::optional<MemoryUsageSnapshot>(cur_memory_snapshot)
+             : std::nullopt;
 }
 
-bool ThresholdMemoryMonitor::IsResourceIsolationThresholdExceeded() {
+std::optional<MemoryUsageSnapshot>
+ThresholdMemoryMonitor::IsResourceIsolationThresholdExceeded() {
   StatusSetOr<MemoryUsageSnapshot, StatusT::NotFound> user_slice_memory_snapshot_or =
       MemoryMonitorUtils::TakeUserSliceMemoryUsageSnapshot(user_cgroup_path_,
                                                            system_cgroup_path_);
@@ -130,7 +144,7 @@ bool ThresholdMemoryMonitor::IsResourceIsolationThresholdExceeded() {
         "The threshold memory monitor will not be able to provide resource isolation "
         "protection.",
         user_slice_memory_snapshot_or.message());
-    return false;
+    return std::nullopt;
   }
   MemoryUsageSnapshot user_slice_memory_snapshot = user_slice_memory_snapshot_or.value();
   bool is_usage_above_threshold =
@@ -145,7 +159,9 @@ bool ThresholdMemoryMonitor::IsResourceIsolationThresholdExceeded() {
         static_cast<float>(memory_usage_threshold_bytes_) /
             static_cast<float>(user_slice_memory_snapshot.total_bytes));
   }
-  return is_usage_above_threshold;
+  return is_usage_above_threshold
+             ? std::optional<MemoryUsageSnapshot>(user_slice_memory_snapshot)
+             : std::nullopt;
 }
 
 }  // namespace ray

src/ray/common/threshold_memory_monitor.h

@@ -16,6 +16,7 @@
 
 #include <cstdint>
 #include <memory>
+#include <optional>
 #include <thread>
 
 #include "ray/asio/periodical_runner.h"
@@ -83,16 +84,18 @@ class ThresholdMemoryMonitor : public MemoryMonitorInterface {
    *
    * @return True if the memory usage is above the threshold.
    */
-  bool IsHostMemoryThresholdExceeded();
+  /// Returns the memory snapshot if the host memory usage exceeds the threshold,
+  /// or std::nullopt otherwise.
+  std::optional<MemoryUsageSnapshot> IsHostMemoryThresholdExceeded();
 
   /**
    * @brief Checks if the memory usage across all user slice processes,
    *        including their object store usage, exceeds their allowed
    *        threshold under resource isolation mode on this node.
    *
-   * @return True if the user process memory usage is above the threshold.
+   * @return The memory snapshot if above threshold, std::nullopt otherwise.
    */
-  bool IsResourceIsolationThresholdExceeded();
+  std::optional<MemoryUsageSnapshot> IsResourceIsolationThresholdExceeded();
 
   /// Callback function that executes at each monitoring interval,
   /// on a dedicated thread managed by this class.

src/ray/raylet/node_manager.cc

@@ -3068,12 +3068,12 @@ void NodeManager::ReleaseKillWorkerInProgress() {
 
 // Picks the workers and kills the process if the memory usage is above the threshold.
 KillWorkersCallback NodeManager::CreateKillWorkersCallback() {
-  return [this]() {
+  return [this](std::string trigger_reason) {
     if (!MarkKillWorkerInProgress()) {
       return;
     }
     io_service_.post(
-        [this]() {
+        [this, trigger_reason = std::move(trigger_reason)]() {
           std::vector<std::shared_ptr<WorkerInterface>> workers =
               worker_pool_.GetAllRegisteredWorkers(/* filter_dead_workers */ true,
                                                    /* filter_io_workers */ true);
@@ -3116,25 +3116,13 @@ KillWorkersCallback NodeManager::CreateKillWorkersCallback() {
             return;
           }
 
-          // Compute the memory usage threshold
-          int64_t total_memory_bytes = memory_usage_snapshot.total_bytes;
-          int64_t computed_threshold_bytes = MemoryMonitorUtils::GetMemoryThreshold(
-              total_memory_bytes,
-              RayConfig::instance().memory_usage_threshold(),
-              RayConfig::instance().min_memory_free_bytes(),
-              initial_config_.enable_resource_isolation,
-              *cgroup_manager_);
-          float computed_threshold_fraction =
-              static_cast<float>(computed_threshold_bytes) /
-              static_cast<float>(total_memory_bytes);
-
           std::string oom_kill_details = CreateOomKillMessageDetails(
               workers_to_kill_and_should_retry,
               self_node_id_,
               memory_usage_snapshot,
               store_client_->GetMemoryUsage().value_or("Not available"),
               process_memory_snapshot,
-              computed_threshold_fraction);
+              trigger_reason);
           std::string oom_kill_suggestions =
               CreateOomKillMessageSuggestions(workers_to_kill_and_should_retry);
 
@@ -3204,7 +3192,7 @@ std::string NodeManager::CreateOomKillMessageDetails(
     const MemoryUsageSnapshot &memory_usage_snapshot,
     const std::string &object_store_memory_usage,
     const ProcessesMemorySnapshot &process_memory_snapshot,
-    float usage_threshold) const {
+    const std::string &trigger_reason) const {
   if (workers_to_kill.empty()) {
     return "";
   }
@@ -3261,8 +3249,8 @@ std::string NodeManager::CreateOomKillMessageDetails(
   }
 
   return absl::StrFormat(
-      "Memory on the node (IP: %s, ID: %s) was %sGB / %sGB (%f), "
-      "which exceeds the memory usage threshold of %f; "
+      "Memory on the node (IP: %s, ID: %s) was %sGB / %sGB (%f); "
+      "OOM kill reason: %s; "
       "Object store memory usage: [%s]; "
       "Ray killed %d worker(s) based on the killing policy: "
       "[%s]; "
@@ -3274,7 +3262,7 @@ std::string NodeManager::CreateOomKillMessageDetails(
       used_bytes_gb,
       total_bytes_gb,
       usage_fraction,
-      usage_threshold,
+      trigger_reason,
       absl::StrReplaceAll(object_store_memory_usage, {{"\n", "; "}}),
       workers_to_kill.size(),
       absl::StrJoin(worker_details, "; "),