[Ray Client] Transfer dashboard_url over gRPC instead of ray.remote

python/ray/client_builder.py

@@ -186,8 +186,9 @@ def connect(self) -> ClientContext:
             ray_init_kwargs=self._remote_init_kwargs,
             metadata=self._metadata,
         )
-        get_dashboard_url = ray.remote(ray._private.worker.get_dashboard_url)
-        dashboard_url = ray.get(get_dashboard_url.options(num_cpus=0).remote())
+
+        dashboard_url = ray.util.client.ray._get_dashboard_url()
+
         cxt = ClientContext(
             dashboard_url=dashboard_url,
             python_version=client_info_dict["python_version"],

python/ray/tests/test_client_reconnect.py

@@ -165,7 +165,17 @@ def ListNamedActors(
         return self._call_inner_function(request, context, "ListNamedActors")
 
     def ClusterInfo(self, request, context=None) -> ray_client_pb2.ClusterInfoResponse:
-        return self._call_inner_function(request, context, "ClusterInfo")
+        # Cluster info is currently used for health checks and isn't retried, so
+        # don't inject errors.
+        # TODO(ckw): update ClusterInfo so that retries are only skipped for PING
+        try:
+            return self.stub.ClusterInfo(
+                request, metadata=context.invocation_metadata()
+            )
+        except grpc.RpcError as e:
+            context.set_code(e.code())
+            context.set_details(e.details())
+            raise
 
     def Terminate(self, req, context=None):
         return self._call_inner_function(req, context, "Terminate")

python/ray/util/client/__init__.py

@@ -19,7 +19,7 @@
 
 # This version string is incremented to indicate breaking changes in the
 # protocol that require upgrading the client version.
-CURRENT_PROTOCOL_VERSION = "2022-10-05"
+CURRENT_PROTOCOL_VERSION = "2022-12-06"
 
 
 class _ClientContext:

python/ray/util/client/api.py

@@ -397,3 +397,10 @@ def _register_callback(
         self, ref: "ClientObjectRef", callback: Callable[["DataResponse"], None]
     ) -> None:
         self.worker.register_callback(ref, callback)
+
+    def _get_dashboard_url(self) -> str:
+        import ray.core.generated.ray_client_pb2 as ray_client_pb2
+
+        return self.worker.get_cluster_info(
+            ray_client_pb2.ClusterInfoType.DASHBOARD_URL
+        ).get("dashboard_url", "")

python/ray/util/client/server/server.py

@@ -286,6 +286,8 @@ def _return_debug_cluster_info(self, request, context=None) -> str:
             data = ray.timeline()
         elif request.type == ray_client_pb2.ClusterInfoType.PING:
             data = {}
+        elif request.type == ray_client_pb2.ClusterInfoType.DASHBOARD_URL:
+            data = {"dashboard_url": ray._private.worker.get_dashboard_url()}
         else:
             raise TypeError("Unsupported cluster info type")
         return json.dumps(data)

src/ray/protobuf/ray_client.proto

@@ -189,6 +189,7 @@ message ClusterInfoType {
     RUNTIME_CONTEXT = 4;
     TIMELINE = 5;
     PING = 6;
+    DASHBOARD_URL = 7;
   }
 }