[Core] Bind gRPC servers to the node address family

python/ray/_common/tests/test_network_utils.py

@@ -2,7 +2,7 @@
 
 import pytest
 
-from ray._common.network_utils import is_localhost
+from ray._common.network_utils import get_all_interfaces_ip, is_localhost
 
 
 def test_is_localhost():
@@ -13,5 +13,11 @@ def test_is_localhost():
     assert not is_localhost("2001:db8::1")
 
 
+def test_get_all_interfaces_ip_for_host():
+    assert get_all_interfaces_ip() in ("0.0.0.0", "::")
+    assert get_all_interfaces_ip("192.0.2.1") == "0.0.0.0"
+    assert get_all_interfaces_ip("2001:db8::1") == "::"
+
+
 if __name__ == "__main__":
     sys.exit(pytest.main(["-v", __file__]))

python/ray/dashboard/agent.py

@@ -36,6 +36,16 @@
 logger = logging.getLogger(__name__)
 
 
+def _build_grpc_address(node_ip_address: str, grpc_port: int) -> str:
+    if node_ip_address == "localhost":
+        grpc_ip = get_localhost_ip()
+    elif is_localhost(node_ip_address):
+        grpc_ip = node_ip_address
+    else:
+        grpc_ip = get_all_interfaces_ip(node_ip_address)
+    return build_address(grpc_ip, grpc_port)
+
+
 class DashboardAgent:
     def __init__(
         self,
@@ -161,11 +171,8 @@ def _init_non_minimal(self):
             ),  # noqa
         )
 
-        grpc_ip = (
-            get_localhost_ip() if is_localhost(self.ip) else get_all_interfaces_ip()
-        )
         self.grpc_port = add_port_to_grpc_server(
-            self.server, build_address(grpc_ip, self.grpc_port)
+            self.server, _build_grpc_address(self.ip, self.grpc_port)
         )
 
         persist_port(

python/ray/dashboard/tests/test_dashboard_agent.py

@@ -0,0 +1,39 @@
+import sys
+from unittest.mock import Mock
+
+import pytest
+
+from ray.dashboard import agent
+
+
+def test_build_grpc_address_uses_ipv6_node_address(monkeypatch):
+    get_all_interfaces_ip = Mock(return_value="::")
+    monkeypatch.setattr(agent, "get_all_interfaces_ip", get_all_interfaces_ip)
+
+    assert agent._build_grpc_address("2001:db8::1", 12345) == "[::]:12345"
+    get_all_interfaces_ip.assert_called_once_with("2001:db8::1")
+
+
+@pytest.mark.parametrize(
+    ("node_ip_address", "expected_address"),
+    [
+        ("127.0.0.1", "127.0.0.1:12345"),
+        ("::1", "[::1]:12345"),
+    ],
+)
+def test_build_grpc_address_preserves_literal_loopback(
+    node_ip_address, expected_address
+):
+    assert agent._build_grpc_address(node_ip_address, 12345) == expected_address
+
+
+def test_build_grpc_address_resolves_localhost(monkeypatch):
+    get_localhost_ip = Mock(return_value="::1")
+    monkeypatch.setattr(agent, "get_localhost_ip", get_localhost_ip)
+
+    assert agent._build_grpc_address("localhost", 12345) == "[::1]:12345"
+    get_localhost_ip.assert_called_once_with()
+
+
+if __name__ == "__main__":
+    sys.exit(pytest.main(["-v", __file__]))

python/ray/includes/network_util.pxd

@@ -12,4 +12,5 @@ cdef extern from "ray/util/network_util.h" namespace "ray":
     bool IsIPv6(const string &host)
     string GetLocalhostIP()
     string GetAllInterfacesIP()
+    string GetAllInterfacesIP(const string &ip_address)
     bool IsLocalhost(const string &address)

python/ray/includes/network_util.pxi

@@ -100,13 +100,23 @@ def get_localhost_ip() -> str:
     return result.decode('utf-8')
 
 
-def get_all_interfaces_ip() -> str:
+def get_all_interfaces_ip(ip_address: Optional[str] = None) -> str:
     """Get the IP address to bind to all network interfaces.
 
-    Returns "0.0.0.0" for IPv4 or "::" for IPv6, depending on the system's
-    localhost resolution.
+    Args:
+        ip_address: If provided, use this literal IP address to select the address
+            family. If omitted, use the system's localhost resolution.
+
+    Returns:
+        "0.0.0.0" for IPv4 or "::" for IPv6.
     """
-    cdef string result = GetAllInterfacesIP()
+    cdef string result
+    cdef string ip_address_c
+    if ip_address is None:
+        result = GetAllInterfacesIP()
+    else:
+        ip_address_c = ip_address.encode('utf-8')
+        result = GetAllInterfacesIP(ip_address_c)
     return result.decode('utf-8')
 
 

src/ray/core_worker/core_worker_process.cc

@@ -277,10 +277,8 @@ std::shared_ptr<CoreWorker> CoreWorkerProcessImpl::CreateCoreWorker(
       std::make_shared<rpc::RayletClient>(std::move(raylet_address),
                                           *client_call_manager_,
                                           /*raylet_unavailable_timeout_callback=*/[] {});
-  auto core_worker_server =
-      std::make_unique<rpc::GrpcServer>(WorkerTypeString(options.worker_type),
-                                        assigned_port,
-                                        IsLocalhost(options.node_ip_address));
+  auto core_worker_server = std::make_unique<rpc::GrpcServer>(
+      WorkerTypeString(options.worker_type), assigned_port, options.node_ip_address);
   // Start RPC server after all the task receivers are properly initialized and we have
   // our assigned port from the raylet.
   core_worker_server->RegisterService(

src/ray/core_worker/tests/core_worker_test.cc

@@ -124,8 +124,8 @@ class CoreWorkerTest : public ::testing::Test {
     auto service_handler = std::make_unique<CoreWorkerServiceHandlerProxy>();
     auto worker_context = std::make_unique<WorkerContext>(
         WorkerType::WORKER, WorkerID::FromRandom(), JobID::FromInt(1));
-    auto core_worker_server =
-        std::make_unique<rpc::GrpcServer>(WorkerTypeString(options.worker_type), 0, true);
+    auto core_worker_server = std::make_unique<rpc::GrpcServer>(
+        WorkerTypeString(options.worker_type), 0, "127.0.0.1");
     core_worker_server->RegisterService(
         std::make_unique<rpc::CoreWorkerGrpcService>(
             io_service_, *service_handler, /*max_active_rpcs_per_handler_=*/-1),

src/ray/gcs/gcs_server.cc

@@ -74,7 +74,7 @@ GcsServer::GcsServer(const ray::gcs::GcsServerConfig &config,
       storage_type_(GetStorageType()),
       rpc_server_(config.grpc_server_name,
                   config.grpc_server_port,
-                  IsLocalhost(config.node_ip_address),
+                  config.node_ip_address,
                   config.grpc_server_thread_num,
                   /*keepalive_time_ms=*/RayConfig::instance().grpc_keepalive_time_ms()),
       client_call_manager_(main_service,

src/ray/gcs/tests/gcs_health_check_manager_test.cc

@@ -92,7 +92,7 @@ class GcsHealthCheckManagerTest : public ::testing::Test {
     auto node_id = NodeID::FromRandom();
     auto port = GetFreePort();
     RAY_LOG(INFO) << "Get port " << port;
-    auto server = std::make_shared<rpc::GrpcServer>(node_id.Hex(), port, true);
+    auto server = std::make_shared<rpc::GrpcServer>(node_id.Hex(), port, "127.0.0.1");
 
     auto channel = grpc::CreateChannel(BuildAddress("localhost", port),
                                        grpc::InsecureChannelCredentials());

src/ray/object_manager/object_manager.cc

@@ -90,7 +90,7 @@ ObjectManager::ObjectManager(
       rpc_service_(rpc_service),
       object_manager_server_("ObjectManager",
                              config_.object_manager_port,
-                             IsLocalhost(config_.object_manager_address),
+                             config_.object_manager_address,
                              config_.rpc_service_threads_number),
       client_call_manager_(main_service,
                            /*record_stats=*/true,

src/ray/pubsub/tests/python_gcs_subscriber_auth_test.cc

@@ -119,7 +119,7 @@ class PythonGcsSubscriberAuthTest : public ::testing::Test {
 
     server_ = std::make_unique<rpc::GrpcServer>("test-gcs-server",
                                                 0,  // Random port
-                                                true,
+                                                "127.0.0.1",
                                                 1,
                                                 7200000,
                                                 auth_token);

src/ray/ray_syncer/tests/BUILD.bazel

@@ -16,6 +16,7 @@ ray_cc_test(
         "//src/ray/ray_syncer",
         "//src/ray/rpc:grpc_server",
         "//src/ray/rpc/authentication:authentication_token",
+        "//src/ray/rpc/authentication:authentication_token_loader",
         "//src/ray/util:env",
         "//src/ray/util:network_util",
         "//src/ray/util:path_utils",

src/ray/ray_syncer/tests/ray_syncer_test.cc

@@ -40,6 +40,7 @@
 #include "ray/ray_syncer/ray_syncer_client.h"
 #include "ray/ray_syncer/ray_syncer_server.h"
 #include "ray/rpc/authentication/authentication_token.h"
+#include "ray/rpc/authentication/authentication_token_loader.h"
 #include "ray/rpc/grpc_server.h"
 #include "ray/util/env.h"
 #include "ray/util/network_util.h"

src/ray/raylet/node_manager.cc

@@ -259,9 +259,8 @@ NodeManager::NodeManager(
                                                std::chrono::milliseconds(delay_ms)));
                     }),
       runtime_env_agent_port_(config.runtime_env_agent_port),
-      node_manager_server_("NodeManager",
-                           config.node_manager_port,
-                           IsLocalhost(config.node_manager_address)),
+      node_manager_server_(
+          "NodeManager", config.node_manager_port, config.node_manager_address),
       local_object_manager_(local_object_manager),
       leased_workers_(leased_workers),
       local_gc_interval_ns_(RayConfig::instance().local_gc_interval_s() * 1e9),

src/ray/rpc/authentication/tests/grpc_auth_token_tests.cc

@@ -73,7 +73,8 @@ class TestGrpcServerClientTokenAuthFixture : public ::testing::Test {
       // Explicitly set empty token (no auth required)
       server_auth_token = std::make_shared<AuthenticationToken>("");
     }
-    grpc_server_.reset(new GrpcServer("test", 0, true, 1, 7200000, server_auth_token));
+    grpc_server_.reset(
+        new GrpcServer("test", 0, "127.0.0.1", 1, 7200000, server_auth_token));
     grpc_server_->RegisterService(
         std::make_unique<TestGrpcService>(handler_io_service_, test_service_handler_),
         false);

src/ray/rpc/grpc_server.cc

@@ -34,6 +34,41 @@
 namespace ray {
 namespace rpc {
 
+namespace internal {
+
+std::string GetGrpcServerBindAddress(const std::string &node_ip_address) {
+  if (node_ip_address == "localhost") {
+    return GetLocalhostIP();
+  }
+  if (IsLocalhost(node_ip_address)) {
+    return node_ip_address;
+  }
+  return GetAllInterfacesIP(node_ip_address);
+}
+
+}  // namespace internal
+
+GrpcServer::GrpcServer(std::string name,
+                       const uint32_t port,
+                       std::string node_ip_address,
+                       int num_threads,
+                       int64_t keepalive_time_ms,
+                       std::shared_ptr<const AuthenticationToken> auth_token)
+    : name_(std::move(name)),
+      port_(port),
+      bind_address_(internal::GetGrpcServerBindAddress(node_ip_address)),
+      is_shutdown_(true),
+      num_threads_(num_threads),
+      keepalive_time_ms_(keepalive_time_ms) {
+  // Initialize auth token: use provided value or load from AuthenticationTokenLoader.
+  if (auth_token) {
+    auth_token_ = std::move(auth_token);
+  } else {
+    auth_token_ = AuthenticationTokenLoader::instance().GetToken();
+  }
+  Init();
+}
+
 void GrpcServer::Init() {
   RAY_CHECK(num_threads_ > 0) << "Num of threads in gRPC must be greater than 0";
   cqs_.resize(num_threads_);
@@ -64,8 +99,7 @@ void GrpcServer::Shutdown() {
 
 void GrpcServer::Run() {
   uint32_t specified_port = port_;
-  std::string server_address = BuildAddress(
-      (listen_to_localhost_only_ ? GetLocalhostIP() : GetAllInterfacesIP()), port_);
+  std::string server_address = BuildAddress(bind_address_, port_);
   grpc::ServerBuilder builder;
   // Disable the SO_REUSEPORT option. We don't need it in ray. If the option is enabled
   // (default behavior in grpc), we may see multiple workers listen on the same port and
@@ -141,7 +175,8 @@ void GrpcServer::Run() {
       << "Try running sudo lsof -i :" << specified_port
       << " to check if there are other processes listening to the port.";
   RAY_CHECK(port_ > 0);
-  RAY_LOG(INFO) << name_ << " server started, listening on port " << port_ << ".";
+  RAY_LOG(INFO) << name_ << " server started, listening on "
+                << BuildAddress(bind_address_, port_) << ".";
 
   // Create calls for all the server call factories
   //

src/ray/rpc/grpc_server.h

@@ -25,11 +25,26 @@
 
 #include "ray/asio/instrumented_io_context.h"
 #include "ray/rpc/authentication/authentication_token.h"
-#include "ray/rpc/authentication/authentication_token_loader.h"
 #include "ray/rpc/server_call.h"
 
 namespace ray {
 namespace rpc {
+
+namespace internal {
+
+/**
+ * @brief Return the address on which a gRPC server should listen for a Ray node.
+ *
+ * This is an internal helper, not a public networking API.
+ *
+ * @param node_ip_address The IP address advertised by the Ray node.
+ * @return The literal loopback address, or the all-interfaces address for the same
+ * address family.
+ */
+std::string GetGrpcServerBindAddress(const std::string &node_ip_address);
+
+}  // namespace internal
+
 /// \param MAX_ACTIVE_RPCS Maximum number of RPCs to handle at the same time. -1 means no
 /// limit.
 #define _RPC_SERVICE_HANDLER(                                                      \
@@ -98,31 +113,19 @@ class GrpcServer {
   /// \param[in] name Name of this server, used for logging and debugging purpose.
   /// \param[in] port The port to bind this server to. If it's 0, a random available port
   ///  will be chosen.
-  /// \param[in] listen_to_localhost_only If true, binds only on localhost, not other
-  /// interfaces. \param[in] num_threads Number of gRPC completion queue threads to use.
+  /// \param[in] node_ip_address The address advertised by this Ray node. The server binds
+  /// to localhost for a loopback address, or to the all-interfaces address for the same
+  /// address family otherwise.
+  /// \param[in] num_threads Number of gRPC completion queue threads to use.
   /// \param[in] keepalive_time_ms Connection keepalive time (ms).
   /// \param[in] auth_token Authentication token that clients must present when making
   /// RPCs to the server. If nullptr, no authentication token is required.
   GrpcServer(std::string name,
              const uint32_t port,
-             bool listen_to_localhost_only,
+             std::string node_ip_address,
              int num_threads = 1,
              int64_t keepalive_time_ms = 7200000, /*2 hours, grpc default*/
-             std::shared_ptr<const AuthenticationToken> auth_token = nullptr)
-      : name_(std::move(name)),
-        port_(port),
-        listen_to_localhost_only_(listen_to_localhost_only),
-        is_shutdown_(true),
-        num_threads_(num_threads),
-        keepalive_time_ms_(keepalive_time_ms) {
-    // Initialize auth token: use provided value or load from AuthenticationTokenLoader
-    if (auth_token) {
-      auth_token_ = auth_token;
-    } else {
-      auth_token_ = AuthenticationTokenLoader::instance().GetToken();
-    }
-    Init();
-  }
+             std::shared_ptr<const AuthenticationToken> auth_token = nullptr);
 
   /// Destruct this gRPC server.
   ~GrpcServer() { Shutdown(); }
@@ -178,9 +181,8 @@ class GrpcServer {
   /// Port of this server.
   int port_;
 
-  /// Listen to localhost (127.0.0.1) only if it's true, otherwise listen to all network
-  /// interfaces (0.0.0.0)
-  const bool listen_to_localhost_only_;
+  /// The IP address on which this server listens.
+  const std::string bind_address_;
 
   /// Token representing ID of this cluster.
   ClusterID cluster_id_;

src/ray/rpc/tests/BUILD.bazel

@@ -34,6 +34,7 @@ ray_cc_test(
         "//src/ray/protobuf:test_service_cc_grpc",
         "//src/ray/rpc:grpc_client",
         "//src/ray/rpc:grpc_server",
+        "//src/ray/util:network_util",
     ],
 )