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

Code before:
          "%s OOM kill suggestions: %s",
          lease_spec.LeaseId().Hex(),
          lease_spec.GetTaskName(),
          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;

      // 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);

      /// since we print the process memory in the message. Destroy should be called
      /// as soon as possible to free up memory.
      DestroyWorker(worker_to_kill,
                    rpc::WorkerExitType::NODE_OUT_OF_MEMORY,
                    worker_exit_message,
                    true /* force */);

      // Record worker killing metrics.
      if (worker_to_kill->GetWorkerType() == rpc::WorkerType::DRIVER) {
        // TODO(sang): Add the job entrypoint to the name.
        memory_manager_worker_eviction_total_count_.Record(
            1, {{"Type", "MemoryManager.DriverEviction.Total"}, {"Name", ""}});
      } else if (worker_to_kill->GetActorId().IsNil()) {
        const RayLease &ray_lease = worker_to_kill->GetGrantedLease();
        memory_manager_worker_eviction_total_count_.Record(
            1,
            {{"Type", "MemoryManager.TaskEviction.Total"},
             {"Name", ray_lease.GetLeaseSpecification().GetTaskName()}});
      } else {
        const RayLease &ray_lease = worker_to_kill->GetGrantedLease();
        memory_manager_worker_eviction_total_count_.Record(
            1,
            {{"Type", "MemoryManager.ActorEviction.Total"},
             {"Name", ray_lease.GetLeaseSpecification().GetTaskName()}});
      }
    }
  };
}

std::string NodeManager::CreateOomKillMessageDetails(
    const std::shared_ptr<WorkerInterface> &worker,
    const NodeID &node_id,
    const MemorySnapshot &system_memory,
    float usage_threshold) const {
  float usage_fraction =

Review comment (bug):
All of the metric emission will need another take in a separate effort to make debugging memory related issues easier. Issue to track: https://github.com/ray-project/ray/issues/60853

Code after:
          std::shared_ptr<WorkerInterface> worker_to_kill =
              worker_to_kill_and_should_retry.first;
          bool should_retry = worker_to_kill_and_should_retry.second;
          if (worker_to_kill == nullptr) {
            RAY_LOG_EVERY_MS(WARNING, 5000) << "Worker killer did not select a worker to "
                                               "kill even though memory usage is high.";
          } else {
            // 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);

            worker_being_killed_ = worker_to_kill;

            std::string oom_kill_details =
                CreateOomKillMessageDetails(worker_to_kill,
                                            self_node_id_,
                                            system_memory,
                                            process_memory_snapshot,
                                            computed_threshold_fraction);
            std::string oom_kill_suggestions =
                CreateOomKillMessageSuggestions(worker_to_kill, should_retry);

            RAY_LOG(INFO) << absl::StrFormat(
                "Killing worker with task %s, kill details: %s, suggestions: %s",
                worker_to_kill->GetGrantedLease().GetLeaseSpecification().DebugString(),
                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;

            // 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);

            /// since we print the process memory in the message. Destroy should be
            /// called as soon as possible to free up memory.