[core] Clean up NODE_DIED task error message (#58638)

Minor follow ups from: #58539

Example error message:
```
Task failed because the node it was running on is dead or unavailable. Node IP: 127.0.0.1, node ID: e55b8ca03ebf3f7418f51533d8d55abeaab75fa9b29e2e6282b47646. This can happen if the node was preempted, had a hardware failure, or its raylet crashed unexpectedly. To see node death information, use `ray list nodes --filter node_id=e55b8ca03ebf3f7418f51533d8d55abeaab75fa9b29e2e6282b47646`, check the Ray dashboard cluster page, search the node ID in the GCS logs, or use `ray logs raylet.out -ip 127.0.0.1`.
```

---------

Signed-off-by: Edward Oakes <ed.nmi.oakes@gmail.com>

Author: edoakes

python/ray/tests/test_ray_event_export_task_events.py

@@ -859,14 +859,14 @@ def validate_task_killed(events: json):
                 expected_task_id_error_info_dict = {
                     (normal_task_id, 0): {
                         "error_type": "NODE_DIED",
-                        "error_message": "Task failed due to the node (where this task was running)  was dead or unavailable",
+                        "error_message": "Task failed because the node it was running on is dead or unavailable",
                     }
                 }
             else:
                 expected_task_id_error_info_dict = {
                     (normal_task_id, 0): {
                         "errorType": "NODE_DIED",
-                        "errorMessage": "Task failed due to the node (where this task was running)  was dead or unavailable",
+                        "errorMessage": "Task failed because the node it was running on is dead or unavailable",
                     }
                 }
             check_task_lifecycle_event_states_and_error_info(

python/ray/tests/test_task_events.py

@@ -191,8 +191,7 @@ def sleep():
         verify_failed_task,
         name="node-killed",
         error_type="NODE_DIED",
-        error_message="Task failed due to the node (where this task was running) "
-        " was dead or unavailable",
+        error_message="Task failed because the node it was running on is dead or unavailable",
     )
 
 
@@ -226,40 +225,6 @@ def task():
     )
 
 
-# TODO(rickyx): Make this work.
-# def test_failed_task_removed_placement_group(shutdown_only, monkeypatch):
-#     ray.init(num_cpus=2, _system_config=_SYSTEM_CONFIG)
-#     from ray.util.placement_group import placement_group, remove_placement_group
-#     from ray.util.scheduling_strategies import PlacementGroupSchedulingStrategy
-#
-#     pg = placement_group([{"CPU": 2}])
-#     ray.get(pg.ready())
-#
-#     @ray.remote(num_cpus=2)
-#     def sleep():
-#         time.sleep(999)
-#
-#     with monkeypatch.context() as m:
-#         m.setenv(
-#             "RAY_testing_asio_delay_us",
-#             "NodeManagerService.grpc_server.RequestWorkerLease=3000000:3000000",
-#         )
-#
-#         sleep.options(
-#             scheduling_strategy=PlacementGroupSchedulingStrategy(placement_group=pg),
-#             name="task-pg-removed",
-#             max_retries=0,
-#         ).remote()
-#
-#     remove_placement_group(pg)
-#
-#     wait_for_condition(
-#         verify_failed_task,
-#         name="task-pg-removed",
-#         error_type="TASK_PLACEMENT_GROUP_REMOVED",
-#     )
-
-
 def test_failed_task_runtime_env_setup(shutdown_only):
     import conda
 

src/ray/core_worker/task_submission/normal_task_submitter.cc

@@ -613,6 +613,7 @@ bool NormalTaskSubmitter::HandleGetWorkerFailureCause(
   rpc::ErrorType task_error_type = rpc::ErrorType::WORKER_DIED;
   std::unique_ptr<rpc::RayErrorInfo> error_info;
   bool fail_immediately = false;
+  NodeID node_id = NodeID::FromBinary(addr.node_id());
   if (get_worker_failure_cause_reply_status.ok()) {
     RAY_LOG(WARNING) << "Worker failure cause for task " << task_id << ": "
                      << ray::gcs::RayErrorInfoToString(
@@ -630,22 +631,21 @@ bool NormalTaskSubmitter::HandleGetWorkerFailureCause(
     RAY_LOG(WARNING) << "Failed to fetch worker failure cause with status "
                      << get_worker_failure_cause_reply_status.ToString()
                      << " worker id: " << WorkerID::FromBinary(addr.worker_id())
-                     << " node id: " << NodeID::FromBinary(addr.node_id())
-                     << " ip: " << addr.ip_address();
+                     << " node id: " << node_id << " ip: " << addr.ip_address();
     task_error_type = rpc::ErrorType::NODE_DIED;
-    std::stringstream buffer;
-    buffer << "Task failed due to the node (where this task was running) "
-           << " was dead or unavailable.\n\nThe node IP: " << addr.ip_address()
-           << ", node ID: " << NodeID::FromBinary(addr.node_id()) << "\n\n"
-           << "This can happen if the instance where the node was running failed, "
-           << "the node was preempted, or raylet crashed unexpectedly "
-           << "(e.g., due to OOM) etc.\n\n"
-           << "To see node death information, use `ray list nodes --filter \"node_id="
-           << NodeID::FromBinary(addr.node_id()) << "\"`, "
-           << "or check Ray dashboard cluster page, or search the node ID in GCS log, "
-           << "or use `ray logs raylet.out -ip " << addr.ip_address() << "`";
+
+    std::string error_message = absl::StrFormat(
+        "Task failed because the node it was running on is dead or unavailable. Node IP: "
+        "%s, node ID: %s. This can happen if the node was preempted, had a hardware "
+        "failure, or its raylet crashed unexpectedly. To see node death information, use "
+        "`ray list nodes --filter node_id=%s`, check the Ray dashboard cluster page, "
+        "search the node ID in the GCS logs, or use `ray logs raylet.out -ip %s`.",
+        addr.ip_address(),
+        node_id.Hex(),
+        node_id.Hex(),
+        addr.ip_address());
     error_info = std::make_unique<rpc::RayErrorInfo>();
-    error_info->set_error_message(buffer.str());
+    error_info->set_error_message(error_message);
     error_info->set_error_type(rpc::ErrorType::NODE_DIED);
   }
   return task_manager_.FailOrRetryPendingTask(task_id,