Remove grpcio from minimal dependency (ray-project#38243)

Removes grpcio from ray dependency while adds it to ray[default] dependency.

Removed use of grpc client: gcs_node_info_stub.DrainNode in autoscaler. Moved it into raylet.pyx.
Removed use of grpc client: gcs_node_resources_stub.GetAllResourceUsage in autoscaler. Moved it into raylet.pyx.
Removed use of grpc status code. Moved it into raylet.pyx.
Removed Ray Client tests in test_runtime_env_ray_minimal.py

---------

Signed-off-by: Ruiyang Wang <rywang014@gmail.com>
Signed-off-by: e428265 <arvind.chandramouli@lmco.com>

dashboard/agent.py

@@ -16,7 +16,6 @@
 import ray.dashboard.consts as dashboard_consts
 import ray.dashboard.utils as dashboard_utils
 from ray.dashboard.consts import _PARENT_DEATH_THREASHOLD
-from ray._private.gcs_pubsub import GcsAioPublisher
 from ray._raylet import GcsClient
 from ray._private.gcs_utils import GcsAioClient
 from ray._private.ray_logging import (
@@ -31,12 +30,6 @@
 # Import psutil after ray so the packaged version is used.
 import psutil
 
-try:
-    from grpc import aio as aiogrpc
-except ImportError:
-    from grpc.experimental import aio as aiogrpc
-
-
 # Publishes at most this number of lines of Raylet logs, when the Raylet dies
 # unexpectedly.
 _RAYLET_LOG_MAX_PUBLISH_LINES = 20
@@ -52,19 +45,6 @@
 
 logger = logging.getLogger(__name__)
 
-# We would want to suppress deprecating warnings from aiogrpc library
-# with the usage of asyncio.get_event_loop() in python version >=3.10
-# This could be removed once https://github.com/grpc/grpc/issues/32526
-# is released, and we used higher versions of grpcio that that.
-if sys.version_info.major >= 3 and sys.version_info.minor >= 10:
-    import warnings
-
-    with warnings.catch_warnings():
-        warnings.simplefilter("ignore", category=DeprecationWarning)
-        aiogrpc.init_grpc_aio()
-else:
-    aiogrpc.init_grpc_aio()
-
 
 class DashboardAgent:
     def __init__(
@@ -117,13 +97,44 @@ def __init__(
             assert self.ppid > 0
             logger.info("Parent pid is %s", self.ppid)
 
-        # Setup raylet channel
-        options = ray_constants.GLOBAL_GRPC_OPTIONS
-        self.aiogrpc_raylet_channel = ray._private.utils.init_grpc_channel(
-            f"{self.ip}:{self.node_manager_port}", options, asynchronous=True
-        )
+        # grpc server is None in mininal.
+        self.server = None
+        # http_server is None in minimal.
+        self.http_server = None
+
+        # Used by the agent and sub-modules.
+        # TODO(architkulkarni): Remove gcs_client once the agent exclusively uses
+        # gcs_aio_client and not gcs_client.
+        self.gcs_client = GcsClient(address=self.gcs_address)
+        _initialize_internal_kv(self.gcs_client)
+        assert _internal_kv_initialized()
+        self.gcs_aio_client = GcsAioClient(address=self.gcs_address)
+
+        if not self.minimal:
+            self._init_non_minimal()
+
+    def _init_non_minimal(self):
+        from ray._private.gcs_pubsub import GcsAioPublisher
+        self.aio_publisher = GcsAioPublisher(address=self.gcs_address)
+
+        try:
+            from grpc import aio as aiogrpc
+        except ImportError:
+            from grpc.experimental import aio as aiogrpc
+
+        # We would want to suppress deprecating warnings from aiogrpc library
+        # with the usage of asyncio.get_event_loop() in python version >=3.10
+        # This could be removed once https://github.com/grpc/grpc/issues/32526
+        # is released, and we used higher versions of grpcio that that.
+        if sys.version_info.major >= 3 and sys.version_info.minor >= 10:
+            import warnings
+
+            with warnings.catch_warnings():
+                warnings.simplefilter("ignore", category=DeprecationWarning)
+                aiogrpc.init_grpc_aio()
+        else:
+            aiogrpc.init_grpc_aio()
 
-        # Setup grpc server
         self.server = aiogrpc.server(options=(("grpc.so_reuseport", 0),))
         grpc_ip = "127.0.0.1" if self.ip == "127.0.0.1" else "0.0.0.0"
         try:
@@ -143,19 +154,6 @@ def __init__(
         else:
             logger.info("Dashboard agent grpc address: %s:%s", grpc_ip, self.grpc_port)
 
-        # If the agent is started as non-minimal version, http server should
-        # be configured to communicate with the dashboard in a head node.
-        self.http_server = None
-
-        # Used by the agent and sub-modules.
-        # TODO(architkulkarni): Remove gcs_client once the agent exclusively uses
-        # gcs_aio_client and not gcs_client.
-        self.gcs_client = GcsClient(address=self.gcs_address)
-        _initialize_internal_kv(self.gcs_client)
-        assert _internal_kv_initialized()
-        self.gcs_aio_client = GcsAioClient(address=self.gcs_address)
-        self.publisher = GcsAioPublisher(address=self.gcs_address)
-
     async def _configure_http_server(self, modules):
         from ray.dashboard.http_server_agent import HttpServerAgent
 
@@ -180,9 +178,16 @@ def _load_modules(self):
 
     @property
     def http_session(self):
-        assert self.http_server, "Accessing unsupported API in a minimal ray."
+        assert self.http_server, \
+            "Accessing unsupported API (HttpServerAgent) in a minimal ray."
         return self.http_server.http_session
 
+    @property
+    def publisher(self):
+        assert self.aio_publisher, \
+            "Accessing unsupported API (GcsAioPublisher) in a minimal ray."
+        return self.aio_publisher
+
     async def run(self):
         async def _check_parent():
             """Check if raylet is dead and fate-share if it is."""
@@ -311,9 +316,10 @@ async def _check_parent():
         # TODO: Use async version if performance is an issue
         # -1 should indicate that http server is not started.
         http_port = -1 if not self.http_server else self.http_server.http_port
+        grpc_port = -1 if not self.server else self.grpc_port
         await self.gcs_aio_client.internal_kv_put(
             f"{dashboard_consts.DASHBOARD_AGENT_PORT_PREFIX}{self.node_id}".encode(),
-            json.dumps([http_port, self.grpc_port]).encode(),
+            json.dumps([http_port, grpc_port]).encode(),
             True,
             namespace=ray_constants.KV_NAMESPACE_DASHBOARD,
         )

dashboard/optional_deps.py

@@ -16,3 +16,4 @@
 from aiohttp.typedefs import PathLike  # noqa: F401
 from aiohttp.web import RouteDef  # noqa: F401
 import pydantic  # noqa: F401
+import grpc  # noqa: F401

dashboard/utils.py

@@ -50,7 +50,7 @@ async def run(self, server):
         """
         Run the module in an asyncio loop. An agent module can provide
         servicers to the server.
-        :param server: Asyncio GRPC server.
+        :param server: Asyncio GRPC server, or None if ray is minimal.
         """
 
     @staticmethod
@@ -79,7 +79,7 @@ async def run(self, server):
         """
         Run the module in an asyncio loop. A head module can provide
         servicers to the server.
-        :param server: Asyncio GRPC server.
+        :param server: Asyncio GRPC server, or None if ray is minimal.
         """
 
     @staticmethod

doc/source/cluster/running-applications/job-submission/ray-client.rst

@@ -3,6 +3,9 @@
 Ray Client
 ==========
 
+.. warning::
+   Ray Client requires pip package `ray[client]`. If you installed the minimal Ray (e.g. `pip install ray`), please reinstall by executing `pip install ray[client]`.
+
 **What is the Ray Client?**
 
 The Ray Client is an API that connects a Python script to a **remote** Ray cluster. Effectively, it allows you to leverage a remote Ray cluster just like you would with Ray running on your local machine.

python/ray/_private/gcs_utils.py

@@ -1,13 +1,6 @@
-import enum
 import logging
-import inspect
-import os
-import asyncio
-from functools import wraps
 from typing import Optional
 
-import grpc
-
 from ray._private import ray_constants
 
 import ray._private.gcs_aio_client
@@ -98,60 +91,6 @@ def create_gcs_channel(address: str, aio=False):
     return init_grpc_channel(address, options=_GRPC_OPTIONS, asynchronous=aio)
 
 
-# This global variable is used for testing only
-_called_freq = {}
-
-
-def _auto_reconnect(f):
-    # This is for testing to count the frequence
-    # of gcs call
-    if inspect.iscoroutinefunction(f):
-
-        @wraps(f)
-        async def wrapper(self, *args, **kwargs):
-            if "TEST_RAY_COLLECT_KV_FREQUENCY" in os.environ:
-                global _called_freq
-                name = f.__name__
-                if name not in _called_freq:
-                    _called_freq[name] = 0
-                _called_freq[name] += 1
-
-            remaining_retry = self._nums_reconnect_retry
-            while True:
-                try:
-                    return await f(self, *args, **kwargs)
-                except grpc.RpcError as e:
-                    if e.code() in (
-                        grpc.StatusCode.UNAVAILABLE,
-                        grpc.StatusCode.UNKNOWN,
-                    ):
-                        if remaining_retry <= 0:
-                            logger.error(
-                                "Failed to connect to GCS. Please check"
-                                " `gcs_server.out` for more details."
-                            )
-                            raise
-                        logger.debug(
-                            "Failed to send request to gcs, reconnecting. " f"Error {e}"
-                        )
-                        try:
-                            self._connect()
-                        except Exception:
-                            logger.error(f"Connecting to gcs failed. Error {e}")
-                        await asyncio.sleep(1)
-                        remaining_retry -= 1
-                        continue
-                    raise
-
-        return wrapper
-    else:
-
-        raise NotImplementedError(
-            "This code moved to Cython, see "
-            "https://github.com/ray-project/ray/pull/33769"
-        )
-
-
 class GcsChannel:
     def __init__(self, gcs_address: Optional[str] = None, aio: bool = False):
         self._gcs_address = gcs_address
@@ -171,13 +110,6 @@ def channel(self):
         return self._channel
 
 
-class GcsCode(enum.IntEnum):
-    # corresponding to ray/src/ray/common/status.h
-    OK = 0
-    NotFound = 17
-    GrpcUnavailable = 26
-
-
 # re-export
 GcsAioClient = ray._private.gcs_aio_client.GcsAioClient
 

python/ray/_private/parameter.py

@@ -6,8 +6,10 @@
 import ray._private.ray_constants as ray_constants
 from ray._private.utils import (
     validate_node_labels,
+    check_ray_client_dependencies_installed,
 )
 
+
 logger = logging.getLogger(__name__)
 
 
@@ -397,8 +399,12 @@ def _check_usage(self):
                     raise ValueError(
                         "max_worker_port must be higher than min_worker_port."
                     )
-
         if self.ray_client_server_port is not None:
+            if not check_ray_client_dependencies_installed():
+                raise ValueError(
+                    "Ray Client requires pip package `ray[client]`. "
+                    "If you installed the minimal Ray (e.g. `pip install ray`), "
+                    "please reinstall by executing `pip install ray[client]`.")
             if (
                 self.ray_client_server_port < 1024
                 or self.ray_client_server_port > 65535

python/ray/_private/test_utils.py

@@ -24,15 +24,13 @@
 import requests
 from ray._raylet import Config
 
-import grpc
 import numpy as np
 import psutil  # We must import psutil after ray because we bundle it with ray.
 from ray._private import (
     ray_constants,
 )
 from ray._private.worker import RayContext
 import yaml
-from grpc._channel import _InactiveRpcError
 
 import ray
 import ray._private.gcs_utils as gcs_utils
@@ -45,9 +43,7 @@
 from ray.core.generated import (
     gcs_pb2,
     node_manager_pb2,
-    node_manager_pb2_grpc,
     gcs_service_pb2,
-    gcs_service_pb2_grpc,
 )
 from ray.util.queue import Empty, Queue, _QueueActor
 from ray.util.scheduling_strategies import NodeAffinitySchedulingStrategy
@@ -1474,6 +1470,9 @@ async def get_total_killed_nodes(self):
             return self.killed_nodes
 
         def _kill_raylet(self, ip, port, graceful=False):
+            import grpc
+            from grpc._channel import _InactiveRpcError
+            from ray.core.generated import node_manager_pb2_grpc
             raylet_address = f"{ip}:{port}"
             channel = grpc.insecure_channel(raylet_address)
             stub = node_manager_pb2_grpc.NodeManagerServiceStub(channel)
@@ -1694,6 +1693,8 @@ def wandb_setup_api_key_hook():
 
 # Get node stats from node manager.
 def get_node_stats(raylet, num_retry=5, timeout=2):
+    import grpc
+    from ray.core.generated import node_manager_pb2_grpc
     raylet_address = f'{raylet["NodeManagerAddress"]}:{raylet["NodeManagerPort"]}'
     channel = ray._private.utils.init_grpc_channel(raylet_address)
     stub = node_manager_pb2_grpc.NodeManagerServiceStub(channel)
@@ -1711,6 +1712,7 @@ def get_node_stats(raylet, num_retry=5, timeout=2):
 
 # Gets resource usage assuming gcs is local.
 def get_resource_usage(gcs_address, timeout=10):
+    from ray.core.generated import gcs_service_pb2_grpc
     if not gcs_address:
         gcs_address = ray.worker._global_node.gcs_address
 
@@ -1739,6 +1741,9 @@ def get_load_metrics_report(webui_url):
 
 # Send a RPC to the raylet to have it self-destruct its process.
 def kill_raylet(raylet, graceful=False):
+    import grpc
+    from grpc._channel import _InactiveRpcError
+    from ray.core.generated import node_manager_pb2_grpc
     raylet_address = f'{raylet["NodeManagerAddress"]}:{raylet["NodeManagerPort"]}'
     channel = grpc.insecure_channel(raylet_address)
     stub = node_manager_pb2_grpc.NodeManagerServiceStub(channel)

python/ray/_private/tls_utils.py

@@ -2,8 +2,6 @@
 import os
 import socket
 
-import grpc
-
 
 def generate_self_signed_tls_certs():
     """Create self-signed key/cert pair for testing.
@@ -68,6 +66,7 @@ def generate_self_signed_tls_certs():
 
 
 def add_port_to_grpc_server(server, address):
+    import grpc
     if os.environ.get("RAY_USE_TLS", "0").lower() in ("1", "true"):
         server_cert_chain, private_key, ca_cert = load_certs_from_env()
         credentials = grpc.ssl_server_credentials(