[core][dashboard] Return the correct error message when trying to kill non-existent actors (#51355)

```sh
# Launch a Ray cluster with Ray dashboard
ray start --head --include-dashboard=True

# Use `/api/actors/kill` to kill an non-existent actor
curl -X GET -w "%{http_code}" "http://localhost:8265/api/actors/kill?actor_id=abc&force_kill=false"

# [response of the curl command]
# The error code is 200, and the msg is wrong.
{"result": true, "msg": "Requested actor with id abc to terminate. It will exit once running tasks complete", "data": {}}200%
```

This PR corrects the error message, and the fix of status code is a
follow up.

With this PR, the status code changes from 200 to 500. It should be 404.
This will be fixed in a follow up PR.

<img width="1317" alt="image"
src="https://github.com/user-attachments/assets/eaef961b-4229-4c92-9168-f2eb69434d85"
/>


Signed-off-by: Kai-Hsun Chen <[email protected]>

Author: kevin85421

Diff for: python/ray/dashboard/modules/snapshot/snapshot_head.py

@@ -89,21 +89,31 @@ async def kill_actor_gcs(self, req) -> aiohttp.web.Response:
                 success=False, message="actor_id is required."
             )
 
-        await self.gcs_aio_client.kill_actor(
+        status_code = await self.gcs_aio_client.kill_actor(
             ActorID.from_hex(actor_id),
             force_kill,
             no_restart,
             timeout=SNAPSHOT_API_TIMEOUT_SECONDS,
         )
 
-        message = (
-            f"Force killed actor with id {actor_id}"
-            if force_kill
-            else f"Requested actor with id {actor_id} to terminate. "
-            + "It will exit once running tasks complete"
-        )
+        success = status_code == 200
+        if status_code == 404:
+            message = f"Actor with id {actor_id} not found."
+        elif status_code == 500:
+            message = f"Failed to kill actor with id {actor_id}."
+        elif status_code == 200:
+            message = (
+                f"Force killed actor with id {actor_id}"
+                if force_kill
+                else f"Requested actor with id {actor_id} to terminate. "
+                + "It will exit once running tasks complete"
+            )
+        else:
+            message = f"Unknown status code: {status_code}. Please open a bug report in the Ray repository."
 
-        return dashboard_optional_utils.rest_response(success=True, message=message)
+        # TODO(kevin85421): The utility function needs to be refactored to handle
+        # different status codes. Currently, it only returns 200 and 500.
+        return dashboard_optional_utils.rest_response(success=success, message=message)
 
     @routes.get("/api/component_activities")
     async def get_component_activities(self, req) -> aiohttp.web.Response:

Diff for: python/ray/dashboard/modules/snapshot/tests/test_actors.py

@@ -35,17 +35,18 @@ def _actor_killed_loop(worker_pid: str, timeout_secs=3) -> bool:
     return dead
 
 
-def _kill_actor_using_dashboard_gcs(webui_url: str, actor_id: str, force_kill=False):
+def _kill_actor_using_dashboard_gcs(
+    webui_url: str, actor_id: str, expected_status_code: int, force_kill=False
+):
     resp = requests.get(
         webui_url + KILL_ACTOR_ENDPOINT,
         params={
             "actor_id": actor_id,
             "force_kill": force_kill,
         },
     )
-    resp.raise_for_status()
+    assert resp.status_code == expected_status_code
     resp_json = resp.json()
-    assert resp_json["result"] is True, "msg" in resp_json
     return resp_json
 
 
@@ -71,8 +72,19 @@ def loop(self):
     worker_pid = ray.get(a.f.remote())  # noqa
     actor_id = a._ray_actor_id.hex()
 
+    OK = 200
+    INTERNAL_ERROR = 500
+
+    # Kill an non-existent actor
+    # TODO(kevin85421): It should return 404 instead of 500.
+    resp = _kill_actor_using_dashboard_gcs(
+        webui_url, "non-existent-actor-id", INTERNAL_ERROR
+    )
+    assert "not found" in resp["msg"]
+
     # Kill the actor
-    _kill_actor_using_dashboard_gcs(webui_url, actor_id, force_kill=False)
+    resp = _kill_actor_using_dashboard_gcs(webui_url, actor_id, OK, force_kill=False)
+    assert "It will exit once running tasks complete" in resp["msg"]
     assert _actor_killed_loop(worker_pid)
 
     # Create an actor and have it loop
@@ -82,11 +94,13 @@ def loop(self):
     a.loop.remote()
 
     # Try to kill the actor, it should not die since a task is running
-    _kill_actor_using_dashboard_gcs(webui_url, actor_id, force_kill=False)
+    resp = _kill_actor_using_dashboard_gcs(webui_url, actor_id, OK, force_kill=False)
+    assert "It will exit once running tasks complete" in resp["msg"]
     assert not _actor_killed_loop(worker_pid, timeout_secs=1)
 
     # Force kill the actor
-    _kill_actor_using_dashboard_gcs(webui_url, actor_id, force_kill=True)
+    resp = _kill_actor_using_dashboard_gcs(webui_url, actor_id, OK, force_kill=True)
+    assert "Force killed actor with id" in resp["msg"]
     assert _actor_killed_loop(worker_pid)
 
 

Diff for: python/ray/includes/gcs_client.pxi

@@ -421,14 +421,12 @@ cdef class InnerGcsClient:
             CActorID c_actor_id = actor_id.native()
 
         with nogil:
-            check_status_timeout_as_rpc_error(
-                self.inner.get().Actors().AsyncKillActor(
-                    c_actor_id,
-                    force_kill,
-                    no_restart,
-                    StatusPyCallback(convert_status, assign_and_decrement_fut,  fut),
-                    timeout_ms
-                )
+            self.inner.get().Actors().AsyncKillActor(
+                c_actor_id,
+                force_kill,
+                no_restart,
+                StatusPyCallback(convert_status, assign_and_decrement_fut,  fut),
+                timeout_ms
             )
         return asyncio.wrap_future(fut)
     #############################################################
@@ -739,12 +737,24 @@ cdef convert_get_cluster_status_reply(
         return None, e
 
 cdef convert_status(CRayStatus status) with gil:
-    # -> None
+    # This function is currently only used by `async_kill_actor` to
+    # convert RayStatus to an HTTP status code.
+    #
+    # Returns:
+    #   Tuple[int, Optional[Exception]]:
+    #     - int: HTTP status code.
+    #       (1) 200: Success
+    #       (2) 404: Actor not found
+    #       (3) 500: Other errors
+    #     - Optional[Exception]: Exception raised by RayStatus
     try:
+        if status.IsNotFound():
+            return 404, None
         check_status_timeout_as_rpc_error(status)
-        return None, None
+        return 200, None
     except Exception as e:
-        return None, e
+        return 500, e
+
 cdef convert_optional_str_none_for_not_found(
         CRayStatus status, optional[c_string] c_str) with gil:
     # If status is NotFound, return None.

Diff for: src/ray/gcs/gcs_server/gcs_actor_manager.cc

@@ -689,18 +689,28 @@ void GcsActorManager::HandleKillActorViaGcs(rpc::KillActorViaGcsRequest request,
                                             rpc::KillActorViaGcsReply *reply,
                                             rpc::SendReplyCallback send_reply_callback) {
   const auto &actor_id = ActorID::FromBinary(request.actor_id());
-  bool force_kill = request.force_kill();
-  bool no_restart = request.no_restart();
-  if (no_restart) {
-    DestroyActor(actor_id, GenKilledByApplicationCause(GetActor(actor_id)));
+  auto it = registered_actors_.find(actor_id);
+  if (it != registered_actors_.end()) {
+    bool force_kill = request.force_kill();
+    bool no_restart = request.no_restart();
+    if (no_restart) {
+      DestroyActor(actor_id, GenKilledByApplicationCause(GetActor(actor_id)));
+    } else {
+      KillActor(actor_id, force_kill);
+    }
+
+    GCS_RPC_SEND_REPLY(send_reply_callback, reply, Status::OK());
+    RAY_LOG(DEBUG).WithField(actor_id.JobId()).WithField(actor_id)
+        << "Finished killing actor, force_kill = " << force_kill
+        << ", no_restart = " << no_restart;
   } else {
-    KillActor(actor_id, force_kill);
+    GCS_RPC_SEND_REPLY(send_reply_callback,
+                       reply,
+                       Status::NotFound(absl::StrFormat(
+                           "Could not find actor with ID %s.", actor_id.Hex())));
+    RAY_LOG(DEBUG).WithField(actor_id.JobId()).WithField(actor_id)
+        << "Could not find actor with ID " << actor_id.Hex();
   }
-
-  GCS_RPC_SEND_REPLY(send_reply_callback, reply, Status::OK());
-  RAY_LOG(DEBUG).WithField(actor_id.JobId()).WithField(actor_id)
-      << "Finished killing actor, force_kill = " << force_kill
-      << ", no_restart = " << no_restart;
   ++counts_[CountType::KILL_ACTOR_REQUEST];
 }
 

Diff for: src/ray/gcs/gcs_server/gcs_actor_manager.h

@@ -673,7 +673,7 @@ class GcsActorManager : public rpc::ActorInfoHandler {
   /// messages come from a Driver/Worker caused by some network problems.
   absl::flat_hash_map<ActorID, std::vector<CreateActorCallback>>
       actor_to_create_callbacks_;
-  /// All registered actors (unresoved and pending actors are also included).
+  /// All registered actors (unresolved and pending actors are also included).
   /// TODO(swang): Use unique_ptr instead of shared_ptr.
   absl::flat_hash_map<ActorID, std::shared_ptr<GcsActor>> registered_actors_;
   /// All destroyed actors.