From 523e2e4d41d8c04380a6ed48ae2d3035c18204dd Mon Sep 17 00:00:00 2001 From: 400Ping Date: Thu, 3 Sep 2026 00:09:58 +0800 Subject: [PATCH 1/3] [Core] Bind gRPC servers to the node address family Signed-off-by: 400Ping --- .../ray/_common/tests/test_network_utils.py | 8 ++- python/ray/dashboard/agent.py | 14 ++-- .../dashboard/tests/test_dashboard_agent.py | 11 ++++ python/ray/includes/network_util.pxd | 1 + python/ray/includes/network_util.pxi | 18 +++-- src/ray/core_worker/core_worker_process.cc | 6 +- src/ray/core_worker/tests/core_worker_test.cc | 4 +- src/ray/gcs/gcs_server.cc | 2 +- .../tests/gcs_health_check_manager_test.cc | 2 +- src/ray/object_manager/object_manager.cc | 2 +- .../tests/python_gcs_subscriber_auth_test.cc | 2 +- src/ray/raylet/node_manager.cc | 5 +- .../tests/grpc_auth_token_tests.cc | 3 +- src/ray/rpc/grpc_server.cc | 41 +++++++++++- src/ray/rpc/grpc_server.h | 39 +++++------ src/ray/rpc/tests/BUILD.bazel | 1 + src/ray/rpc/tests/grpc_server_client_test.cc | 66 +++++++++++++++++-- src/ray/util/network_util.cc | 12 ++-- src/ray/util/network_util.h | 5 ++ src/ray/util/tests/network_util_test.cc | 5 ++ 20 files changed, 185 insertions(+), 62 deletions(-) create mode 100644 python/ray/dashboard/tests/test_dashboard_agent.py diff --git a/python/ray/_common/tests/test_network_utils.py b/python/ray/_common/tests/test_network_utils.py index 8aac0e1be420..15f85aa850a5 100644 --- a/python/ray/_common/tests/test_network_utils.py +++ b/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__])) diff --git a/python/ray/dashboard/agent.py b/python/ray/dashboard/agent.py index 9955bc7bc221..6256ff2866a3 100644 --- a/python/ray/dashboard/agent.py +++ b/python/ray/dashboard/agent.py @@ -36,6 +36,15 @@ logger = logging.getLogger(__name__) +def _build_grpc_address(node_ip_address: str, grpc_port: int) -> str: + grpc_ip = ( + get_localhost_ip() + if is_localhost(node_ip_address) + else get_all_interfaces_ip(node_ip_address) + ) + return build_address(grpc_ip, grpc_port) + + class DashboardAgent: def __init__( self, @@ -161,11 +170,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( diff --git a/python/ray/dashboard/tests/test_dashboard_agent.py b/python/ray/dashboard/tests/test_dashboard_agent.py new file mode 100644 index 000000000000..0da98ef15610 --- /dev/null +++ b/python/ray/dashboard/tests/test_dashboard_agent.py @@ -0,0 +1,11 @@ +from unittest.mock import Mock + +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") diff --git a/python/ray/includes/network_util.pxd b/python/ray/includes/network_util.pxd index 0a1b4cfa1f3c..8e65cca7c08f 100644 --- a/python/ray/includes/network_util.pxd +++ b/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) diff --git a/python/ray/includes/network_util.pxi b/python/ray/includes/network_util.pxi index 87bfd742b7a7..cde6dd4d9e8a 100644 --- a/python/ray/includes/network_util.pxi +++ b/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') diff --git a/src/ray/core_worker/core_worker_process.cc b/src/ray/core_worker/core_worker_process.cc index 7feacdc2675d..6dbcd43ac51b 100644 --- a/src/ray/core_worker/core_worker_process.cc +++ b/src/ray/core_worker/core_worker_process.cc @@ -277,10 +277,8 @@ std::shared_ptr CoreWorkerProcessImpl::CreateCoreWorker( std::make_shared(std::move(raylet_address), *client_call_manager_, /*raylet_unavailable_timeout_callback=*/[] {}); - auto core_worker_server = - std::make_unique(WorkerTypeString(options.worker_type), - assigned_port, - IsLocalhost(options.node_ip_address)); + auto core_worker_server = std::make_unique( + 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( diff --git a/src/ray/core_worker/tests/core_worker_test.cc b/src/ray/core_worker/tests/core_worker_test.cc index bdf297bf3a39..abc7d371a71a 100644 --- a/src/ray/core_worker/tests/core_worker_test.cc +++ b/src/ray/core_worker/tests/core_worker_test.cc @@ -124,8 +124,8 @@ class CoreWorkerTest : public ::testing::Test { auto service_handler = std::make_unique(); auto worker_context = std::make_unique( WorkerType::WORKER, WorkerID::FromRandom(), JobID::FromInt(1)); - auto core_worker_server = - std::make_unique(WorkerTypeString(options.worker_type), 0, true); + auto core_worker_server = std::make_unique( + WorkerTypeString(options.worker_type), 0, "127.0.0.1"); core_worker_server->RegisterService( std::make_unique( io_service_, *service_handler, /*max_active_rpcs_per_handler_=*/-1), diff --git a/src/ray/gcs/gcs_server.cc b/src/ray/gcs/gcs_server.cc index bf6fb79f27e1..ff33f78de8c5 100644 --- a/src/ray/gcs/gcs_server.cc +++ b/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, diff --git a/src/ray/gcs/tests/gcs_health_check_manager_test.cc b/src/ray/gcs/tests/gcs_health_check_manager_test.cc index a0695f4e8c13..b0a42f9a3c7d 100644 --- a/src/ray/gcs/tests/gcs_health_check_manager_test.cc +++ b/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(node_id.Hex(), port, true); + auto server = std::make_shared(node_id.Hex(), port, "127.0.0.1"); auto channel = grpc::CreateChannel(BuildAddress("localhost", port), grpc::InsecureChannelCredentials()); diff --git a/src/ray/object_manager/object_manager.cc b/src/ray/object_manager/object_manager.cc index 7febee8ce590..ec4393a22658 100644 --- a/src/ray/object_manager/object_manager.cc +++ b/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, diff --git a/src/ray/pubsub/tests/python_gcs_subscriber_auth_test.cc b/src/ray/pubsub/tests/python_gcs_subscriber_auth_test.cc index 5ce3c2787560..aa488e7025fc 100644 --- a/src/ray/pubsub/tests/python_gcs_subscriber_auth_test.cc +++ b/src/ray/pubsub/tests/python_gcs_subscriber_auth_test.cc @@ -119,7 +119,7 @@ class PythonGcsSubscriberAuthTest : public ::testing::Test { server_ = std::make_unique("test-gcs-server", 0, // Random port - true, + "127.0.0.1", 1, 7200000, auth_token); diff --git a/src/ray/raylet/node_manager.cc b/src/ray/raylet/node_manager.cc index ff9e598c9c95..7f89e92447b5 100644 --- a/src/ray/raylet/node_manager.cc +++ b/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), diff --git a/src/ray/rpc/authentication/tests/grpc_auth_token_tests.cc b/src/ray/rpc/authentication/tests/grpc_auth_token_tests.cc index 0b8af4dff5df..96d14046c810 100644 --- a/src/ray/rpc/authentication/tests/grpc_auth_token_tests.cc +++ b/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(""); } - 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(handler_io_service_, test_service_handler_), false); diff --git a/src/ray/rpc/grpc_server.cc b/src/ray/rpc/grpc_server.cc index 8e0deb5f0da1..ef860c940487 100644 --- a/src/ray/rpc/grpc_server.cc +++ b/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 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 // diff --git a/src/ray/rpc/grpc_server.h b/src/ray/rpc/grpc_server.h index 36388ba249f2..77a0c6c1590f 100644 --- a/src/ray/rpc/grpc_server.h +++ b/src/ray/rpc/grpc_server.h @@ -25,11 +25,19 @@ #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 { + +/// Internal helper that returns the address on which a gRPC server should listen for a +/// Ray node address. This is not a public networking API. +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 +106,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 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 auth_token = nullptr); /// Destruct this gRPC server. ~GrpcServer() { Shutdown(); } @@ -178,9 +174,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_; diff --git a/src/ray/rpc/tests/BUILD.bazel b/src/ray/rpc/tests/BUILD.bazel index 8218bf450e28..e18df0c643b6 100644 --- a/src/ray/rpc/tests/BUILD.bazel +++ b/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", ], ) diff --git a/src/ray/rpc/tests/grpc_server_client_test.cc b/src/ray/rpc/tests/grpc_server_client_test.cc index 6a371f34eb06..e7affbfd10ac 100644 --- a/src/ray/rpc/tests/grpc_server_client_test.cc +++ b/src/ray/rpc/tests/grpc_server_client_test.cc @@ -14,12 +14,14 @@ #include #include +#include #include #include "gtest/gtest.h" #include "ray/rpc/grpc_client.h" #include "ray/rpc/grpc_server.h" #include "ray/rpc/tests/grpc_test_common.h" +#include "ray/util/network_util.h" #include "src/ray/protobuf/test_service.grpc.pb.h" namespace ray { @@ -27,7 +29,18 @@ namespace rpc { class TestGrpcServerClientFixture : public ::testing::Test { public: - void SetUp() { + virtual std::string NodeAddress() const { return "127.0.0.1"; } + + virtual std::string ClientAddress() const { return "127.0.0.1"; } + + virtual bool RequiresIpv6() const { return false; } + + void SetUp() override { + if (RequiresIpv6() && + !CheckPortFree(boost::asio::ip::tcp::v6().family(), /*port=*/0)) { + GTEST_SKIP() << "IPv6 sockets are not available in this test environment."; + } + // Prepare and start test server. handler_thread_ = std::make_unique([this]() { /// The asio work to keep handler_io_service_ alive. @@ -35,7 +48,7 @@ class TestGrpcServerClientFixture : public ::testing::Test { handler_io_service_work_(handler_io_service_.get_executor()); handler_io_service_.run(); }); - grpc_server_.reset(new GrpcServer("test", 0, true)); + grpc_server_.reset(new GrpcServer("test", 0, NodeAddress())); grpc_server_->RegisterService( std::make_unique(handler_io_service_, test_service_handler_), false); @@ -56,27 +69,29 @@ class TestGrpcServerClientFixture : public ::testing::Test { client_call_manager_.reset( new ClientCallManager(client_io_service_, false, /*local_address=*/"")); grpc_client_.reset(new GrpcClient( - "127.0.0.1", grpc_server_->GetPort(), *client_call_manager_)); + ClientAddress(), grpc_server_->GetPort(), *client_call_manager_)); } void ShutdownClient() { grpc_client_.reset(); client_call_manager_.reset(); client_io_service_.stop(); - if (client_thread_->joinable()) { + if (client_thread_ && client_thread_->joinable()) { client_thread_->join(); } } void ShutdownServer() { - grpc_server_->Shutdown(); + if (grpc_server_) { + grpc_server_->Shutdown(); + } handler_io_service_.stop(); - if (handler_thread_->joinable()) { + if (handler_thread_ && handler_thread_->joinable()) { handler_thread_->join(); } } - void TearDown() { + void TearDown() override { // Cleanup stuffs. ShutdownClient(); ShutdownServer(); @@ -102,6 +117,15 @@ class TestGrpcServerClientFixture : public ::testing::Test { std::unique_ptr> grpc_client_; }; +class TestGrpcServerClientIpv6Fixture : public TestGrpcServerClientFixture { + public: + std::string NodeAddress() const override { return "2001:db8::1"; } + + std::string ClientAddress() const override { return "::1"; } + + bool RequiresIpv6() const override { return true; } +}; + TEST_F(TestGrpcServerClientFixture, TestBasic) { // Send request PingRequest request; @@ -116,6 +140,34 @@ TEST_F(TestGrpcServerClientFixture, TestBasic) { } } +TEST_F(TestGrpcServerClientIpv6Fixture, TestIpv6WildcardBind) { + struct CallState { + std::atomic done{false}; + std::atomic success{false}; + }; + + PingRequest request; + auto state = std::make_shared(); + Ping(request, [state](const Status &status, const PingReply &reply) { + RAY_LOG(INFO) << "replied, status=" << status; + state->success = status.ok(); + state->done = true; + }); + for (int attempt = 0; attempt < 100 && !state->done; attempt++) { + std::this_thread::sleep_for(std::chrono::milliseconds(100)); + } + ASSERT_TRUE(state->done); + EXPECT_TRUE(state->success); +} + +TEST(GrpcServerTest, TestBindAddressMatchesNodeAddressFamily) { + EXPECT_EQ(internal::GetGrpcServerBindAddress("localhost"), GetLocalhostIP()); + EXPECT_EQ(internal::GetGrpcServerBindAddress("127.0.0.1"), "127.0.0.1"); + EXPECT_EQ(internal::GetGrpcServerBindAddress("::1"), "::1"); + EXPECT_EQ(internal::GetGrpcServerBindAddress("192.0.2.1"), "0.0.0.0"); + EXPECT_EQ(internal::GetGrpcServerBindAddress("2001:db8::1"), "::"); +} + TEST_F(TestGrpcServerClientFixture, TestBackpressure) { // Send a request which won't be replied to. PingRequest request; diff --git a/src/ray/util/network_util.cc b/src/ray/util/network_util.cc index 881886c9ffd1..bf374c41ded7 100644 --- a/src/ray/util/network_util.cc +++ b/src/ray/util/network_util.cc @@ -278,16 +278,14 @@ std::string GetLocalhostIP() { } std::string GetAllInterfacesIP() { - static const std::string all_interfaces_ip = []() { - std::string localhost = GetLocalhostIP(); - if (localhost == "::1" || localhost.find(':') != std::string::npos) { - return std::string("::"); - } - return std::string("0.0.0.0"); - }(); + static const std::string all_interfaces_ip = GetAllInterfacesIP(GetLocalhostIP()); return all_interfaces_ip; } +std::string GetAllInterfacesIP(const std::string &ip_address) { + return IsIPv6(ip_address) ? "::" : "0.0.0.0"; +} + std::string GetNodeIpAddressFromPerspective(const std::optional &address) { std::vector> test_addresses; if (address.has_value()) { diff --git a/src/ray/util/network_util.h b/src/ray/util/network_util.h index a428dcc07798..569988a00dfe 100644 --- a/src/ray/util/network_util.h +++ b/src/ray/util/network_util.h @@ -74,6 +74,11 @@ std::string GetLocalhostIP(); /// \return "0.0.0.0" for IPv4 or "::" for IPv6 std::string GetAllInterfacesIP(); +/// Get the IP address to bind to all network interfaces for an address family. +/// \param ip_address The literal IP address used to select the address family. +/// \return "0.0.0.0" for an IPv4 address or "::" for an IPv6 address. +std::string GetAllInterfacesIP(const std::string &ip_address); + /// Check whether the given port is available for the specified address family. /// Notice, the check could be non-authentic if there're concurrent port assignments. /// \param family The address family to check (AF_INET for IPv4, AF_INET6 for IPv6). diff --git a/src/ray/util/tests/network_util_test.cc b/src/ray/util/tests/network_util_test.cc index 5904bb537d1e..67f7839abebd 100644 --- a/src/ray/util/tests/network_util_test.cc +++ b/src/ray/util/tests/network_util_test.cc @@ -129,4 +129,9 @@ TEST(NetworkUtilTest, TestIsIPv6) { EXPECT_FALSE(IsIPv6("::1::2")); } +TEST(NetworkUtilTest, TestGetAllInterfacesIPForHost) { + EXPECT_EQ(GetAllInterfacesIP("192.0.2.1"), "0.0.0.0"); + EXPECT_EQ(GetAllInterfacesIP("2001:db8::1"), "::"); +} + } // namespace ray From ac759e252d37de24b610216c01ae3ef66bf23254 Mon Sep 17 00:00:00 2001 From: 400Ping Date: Thu, 3 Sep 2026 08:08:53 +0800 Subject: [PATCH 2/3] [Fix] Address CI error Signed-off-by: 400Ping --- .../dashboard/tests/test_dashboard_agent.py | 7 ++++ src/ray/ray_syncer/tests/BUILD.bazel | 1 + src/ray/ray_syncer/tests/ray_syncer_test.cc | 1 + src/ray/rpc/tests/grpc_server_client_test.cc | 34 +++++++++++++++++-- 4 files changed, 40 insertions(+), 3 deletions(-) diff --git a/python/ray/dashboard/tests/test_dashboard_agent.py b/python/ray/dashboard/tests/test_dashboard_agent.py index 0da98ef15610..de88e915c7a6 100644 --- a/python/ray/dashboard/tests/test_dashboard_agent.py +++ b/python/ray/dashboard/tests/test_dashboard_agent.py @@ -1,5 +1,8 @@ +import sys from unittest.mock import Mock +import pytest + from ray.dashboard import agent @@ -9,3 +12,7 @@ def test_build_grpc_address_uses_ipv6_node_address(monkeypatch): assert agent._build_grpc_address("2001:db8::1", 12345) == "[::]:12345" get_all_interfaces_ip.assert_called_once_with("2001:db8::1") + + +if __name__ == "__main__": + sys.exit(pytest.main(["-v", __file__])) diff --git a/src/ray/ray_syncer/tests/BUILD.bazel b/src/ray/ray_syncer/tests/BUILD.bazel index d8446a558ff7..4dd8138e6b87 100644 --- a/src/ray/ray_syncer/tests/BUILD.bazel +++ b/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", diff --git a/src/ray/ray_syncer/tests/ray_syncer_test.cc b/src/ray/ray_syncer/tests/ray_syncer_test.cc index b6b170aeb0bc..3f44e9ffbfae 100644 --- a/src/ray/ray_syncer/tests/ray_syncer_test.cc +++ b/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" diff --git a/src/ray/rpc/tests/grpc_server_client_test.cc b/src/ray/rpc/tests/grpc_server_client_test.cc index e7affbfd10ac..ed2bae4b1e6e 100644 --- a/src/ray/rpc/tests/grpc_server_client_test.cc +++ b/src/ray/rpc/tests/grpc_server_client_test.cc @@ -12,6 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. +#include #include #include #include @@ -26,6 +27,34 @@ namespace ray { namespace rpc { +namespace { + +bool IsIpv6LoopbackAvailable() { + boost::asio::io_context io_context; + boost::asio::ip::tcp::acceptor acceptor(io_context); + boost::system::error_code error; + acceptor.open(boost::asio::ip::tcp::v6(), error); + if (error) { + return false; + } + acceptor.bind({boost::asio::ip::address_v6::loopback(), /*port=*/0}, error); + if (error) { + return false; + } + acceptor.listen(boost::asio::socket_base::max_listen_connections, error); + if (error) { + return false; + } + auto endpoint = acceptor.local_endpoint(error); + if (error) { + return false; + } + boost::asio::ip::tcp::socket socket(io_context); + socket.connect(endpoint, error); + return !error; +} + +} // namespace class TestGrpcServerClientFixture : public ::testing::Test { public: @@ -36,9 +65,8 @@ class TestGrpcServerClientFixture : public ::testing::Test { virtual bool RequiresIpv6() const { return false; } void SetUp() override { - if (RequiresIpv6() && - !CheckPortFree(boost::asio::ip::tcp::v6().family(), /*port=*/0)) { - GTEST_SKIP() << "IPv6 sockets are not available in this test environment."; + if (RequiresIpv6() && !IsIpv6LoopbackAvailable()) { + GTEST_SKIP() << "IPv6 loopback is not available in this test environment."; } // Prepare and start test server. From 8bdc32f151cbd05af56a57e56093b85b1a44e83e Mon Sep 17 00:00:00 2001 From: 400Ping Date: Fri, 4 Sep 2026 14:52:29 +0800 Subject: [PATCH 3/3] [Fix] Address cursor review Signed-off-by: 400Ping --- python/ray/dashboard/agent.py | 11 +++++----- .../dashboard/tests/test_dashboard_agent.py | 21 +++++++++++++++++++ src/ray/rpc/grpc_server.h | 11 ++++++++-- 3 files changed, 36 insertions(+), 7 deletions(-) diff --git a/python/ray/dashboard/agent.py b/python/ray/dashboard/agent.py index 6256ff2866a3..41263b579a49 100644 --- a/python/ray/dashboard/agent.py +++ b/python/ray/dashboard/agent.py @@ -37,11 +37,12 @@ def _build_grpc_address(node_ip_address: str, grpc_port: int) -> str: - grpc_ip = ( - get_localhost_ip() - if is_localhost(node_ip_address) - else get_all_interfaces_ip(node_ip_address) - ) + 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) diff --git a/python/ray/dashboard/tests/test_dashboard_agent.py b/python/ray/dashboard/tests/test_dashboard_agent.py index de88e915c7a6..9ee621ae41b3 100644 --- a/python/ray/dashboard/tests/test_dashboard_agent.py +++ b/python/ray/dashboard/tests/test_dashboard_agent.py @@ -14,5 +14,26 @@ def test_build_grpc_address_uses_ipv6_node_address(monkeypatch): 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__])) diff --git a/src/ray/rpc/grpc_server.h b/src/ray/rpc/grpc_server.h index 77a0c6c1590f..2ace47f9e9e1 100644 --- a/src/ray/rpc/grpc_server.h +++ b/src/ray/rpc/grpc_server.h @@ -32,8 +32,15 @@ namespace rpc { namespace internal { -/// Internal helper that returns the address on which a gRPC server should listen for a -/// Ray node address. This is not a public networking API. +/** + * @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