Repository: ray-project/ray
File: src/ray/raylet/node_manager.cc (C++)

Code before:

          // Compute the memory usage threshold
          int64_t total_memory_bytes = system_memory.total_bytes;
          int64_t computed_threshold_bytes = MemoryMonitorUtils::GetMemoryThreshold(
              total_memory_bytes,
              RayConfig::instance().memory_usage_threshold(),
              RayConfig::instance().min_memory_free_bytes());
          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_,
                                          system_memory,
                                          process_memory_snapshot,
                                          computed_threshold_fraction);
          std::string oom_kill_suggestions =
              CreateOomKillMessageSuggestions(workers_to_kill_and_should_retry);

          RAY_LOG(INFO) << absl::StrFormat(
              "Killing %d worker(s), kill details: %s, suggestions: %s",
              workers_to_kill_and_should_retry.size(),
              oom_kill_details,
              oom_kill_suggestions);

          std::string worker_exit_message = absl::StrFormat(
              "Task was killed due to the node running low on memory. %s, %s",
              oom_kill_details,
              oom_kill_suggestions);

          // Report the event to the dashboard.
          RAY_EVENT_EVERY_MS(ERROR, "Out of Memory", 10 * 1000) << worker_exit_message;

          for (const auto &[worker_to_kill, should_retry] :
               workers_to_kill_and_should_retry) {
            // Mark the worker as failure and raise an exception from a caller.
            rpc::RayErrorInfo worker_failure_reason;
            worker_failure_reason.set_error_message(worker_exit_message);
            worker_failure_reason.set_error_type(rpc::ErrorType::OUT_OF_MEMORY);
            SetWorkerFailureReason(
                worker_to_kill->GetGrantedLeaseId(), worker_failure_reason, should_retry);

            DestroyWorker(worker_to_kill,
                          rpc::WorkerExitType::NODE_OUT_OF_MEMORY,
                          worker_exit_message,
                          true /* force */);

            if (worker_to_kill->GetWorkerType() == rpc::WorkerType::DRIVER) {
              // TODO(sang): Add the job entrypoint to the name.
              memory_manager_worker_eviction_total_count_.Record(

Review comment (suggestion):
In a follow up, we should make sure this message is helpful to operators/users. We should also add it to the docs if it's not already there. 

Code after:

          // Compute the memory usage threshold
          int64_t total_memory_bytes = system_memory.total_bytes;
          int64_t computed_threshold_bytes = MemoryMonitorUtils::GetMemoryThreshold(
              total_memory_bytes,
              RayConfig::instance().memory_usage_threshold(),
              RayConfig::instance().min_memory_free_bytes());
          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_,
              system_memory,
              store_client_->GetMemoryUsage().value_or("Not available"),
              process_memory_snapshot,
              computed_threshold_fraction);
          std::string oom_kill_suggestions =
              CreateOomKillMessageSuggestions(workers_to_kill_and_should_retry);

          RAY_LOG(INFO) << absl::StrFormat(
              "Killing %d worker(s), kill details: %s, suggestions: %s",
              workers_to_kill_and_should_retry.size(),
              oom_kill_details,
              oom_kill_suggestions);

          std::string worker_exit_message = absl::StrFormat(
              "%d worker(s) were killed due to the node running low on memory. %s, %s",
              workers_to_kill_and_should_retry.size(),
              oom_kill_details,
              oom_kill_suggestions);

          // Report the event to the dashboard.
          RAY_EVENT_EVERY_MS(ERROR, "Out of Memory", 10 * 1000) << worker_exit_message;

          for (const auto &[worker_to_kill, should_retry] :
               workers_to_kill_and_should_retry) {
            // Mark the worker as failure and raise an exception from a caller.
            rpc::RayErrorInfo worker_failure_reason;
            worker_failure_reason.set_error_message(worker_exit_message);
            worker_failure_reason.set_error_type(rpc::ErrorType::OUT_OF_MEMORY);
            SetWorkerFailureReason(
                worker_to_kill->GetGrantedLeaseId(), worker_failure_reason, should_retry);

            DestroyWorker(worker_to_kill,
                          rpc::WorkerExitType::NODE_OUT_OF_MEMORY,
                          worker_exit_message,
                          true /* force */);

            if (worker_to_kill->GetWorkerType() == rpc::WorkerType::DRIVER) {