Repository navigation
Feat raylet sole gcs - #59798
Feat raylet sole gcs#59798Tomlord1122 wants to merge 9 commits into
Conversation
In this commit, we remove the legacy dashboard agent KV writes, ensuring Raylet is the sole writer to GCS. Signed-off-by: Tom Liu (LIU HSIU-CHI) <aa123593465@gmail.com>
Replace internal KV reads with GcsNodeInfo queries. Signed-off-by: Tom Liu (LIU HSIU-CHI) <aa123593465@gmail.com>
Replace internal KV reads with GcsNodeInfo queries across dashboard modules Signed-off-by: Tom Liu (LIU HSIU-CHI) <aa123593465@gmail.com>
…odeInfo Update test utilities and test files to rely on GcsNodeInfo for agent port information instead of internal KV store Signed-off-by: Tom Liu (LIU HSIU-CHI) <aa123593465@gmail.com>
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request refactors how the dashboard agent's port information is managed, making Raylet the sole writer to GCS. This is a solid architectural improvement that enhances consistency. The changes span core logic and test files, and they are generally well-aligned with the stated goal. My review has identified a potential correctness issue in reporter_head.py where dead nodes are not filtered, which could lead to incorrect behavior. I've also included suggestions to improve code quality by addressing code duplication in tests, fixing a type hint mismatch, and optimizing a test helper for better performance.
| for node_id, node_info in node_info_dict.items(): | ||
| if node_info.node_manager_address == ip: | ||
| http_port = node_info.dashboard_agent_listen_port | ||
| grpc_port = node_info.metrics_agent_port | ||
| if grpc_port <= 0: | ||
| # Agent not started or not available | ||
| return None | ||
| return node_id, ip, http_port, grpc_port |
There was a problem hiding this comment.
The current implementation of _get_stub_address_by_ip fetches all nodes, including dead ones. This could lead to returning information for a dead node if a new node reuses its IP address. You should filter for ALIVE nodes to ensure correctness.
To fix this, you'll also need to add from ray.core.generated import gcs_pb2 at the top of the file.
| for node_id, node_info in node_info_dict.items(): | |
| if node_info.node_manager_address == ip: | |
| http_port = node_info.dashboard_agent_listen_port | |
| grpc_port = node_info.metrics_agent_port | |
| if grpc_port <= 0: | |
| # Agent not started or not available | |
| return None | |
| return node_id, ip, http_port, grpc_port | |
| for node_id, node_info in node_info_dict.items(): | |
| if node_info.state != gcs_pb2.GcsNodeInfo.GcsNodeState.ALIVE: | |
| continue | |
| if node_info.node_manager_address == ip: | |
| http_port = node_info.dashboard_agent_listen_port | |
| grpc_port = node_info.metrics_agent_port | |
| if grpc_port <= 0: | |
| # Agent not started or not available | |
| return None | |
| return node_id, ip, http_port, grpc_port |
| def wait_for_dashboard_agent_available(cluster): | ||
| from ray import NodeID | ||
| from ray.core.generated import gcs_pb2 | ||
|
|
||
| gcs_client = GcsClient(address=cluster.address) | ||
| node_id = NodeID.from_hex(cluster.head_node.node_id) | ||
|
|
||
| def get_dashboard_agent_address(): | ||
| return gcs_client.internal_kv_get( | ||
| f"{dashboard_consts.DASHBOARD_AGENT_ADDR_NODE_ID_PREFIX}{cluster.head_node.node_id}".encode(), | ||
| namespace=ray_constants.KV_NAMESPACE_DASHBOARD, | ||
| def is_dashboard_agent_ready(): | ||
| node_info_dict = gcs_client.get_all_node_info( | ||
| timeout=dashboard_consts.GCS_RPC_TIMEOUT_SECONDS, | ||
| state_filter=gcs_pb2.GcsNodeInfo.GcsNodeState.ALIVE, | ||
| ) | ||
| if node_id not in node_info_dict: | ||
| return False | ||
| node_info = node_info_dict[node_id] | ||
| # Dashboard agent is ready when metrics_agent_port is set (grpc port) | ||
| return node_info.metrics_agent_port > 0 | ||
|
|
||
| wait_for_condition(lambda: get_dashboard_agent_address() is not None) | ||
| wait_for_condition(is_dashboard_agent_ready) |
There was a problem hiding this comment.
This wait_for_dashboard_agent_available function is a duplicate of the one in ray._private.test_utils. To improve maintainability and avoid code duplication, please remove this function and import it from ray._private.test_utils at the top of the file.
| ip = node_info.node_manager_address | ||
| http_port = node_info.dashboard_agent_listen_port | ||
| grpc_port = node_info.metrics_agent_port | ||
| return [ip, http_port, grpc_port] |
There was a problem hiding this comment.
The function _fetch_agent_info is type-hinted to return a Tuple[str, int, int], but it currently returns a list. While this works at runtime, it's better to adhere to the type hint for correctness and clarity. Please change the return value to a tuple.
| return [ip, http_port, grpc_port] | |
| return (ip, http_port, grpc_port) |
| def get_agent_grpc_port(node_id_hex: str) -> int: | ||
| node_id = NodeID.from_hex(node_id_hex) | ||
| node_info_dict = gcs_client.get_all_node_info( | ||
| timeout=dashboard_consts.GCS_RPC_TIMEOUT_SECONDS, | ||
| state_filter=gcs_pb2.GcsNodeInfo.GcsNodeState.ALIVE, | ||
| ) | ||
| if node_id not in node_info_dict: | ||
| return -1 | ||
| return node_info_dict[node_id].metrics_agent_port | ||
|
|
||
| wait_for_condition( | ||
| lambda: all( | ||
| get_dashboard_agent_address(node_id) is not None for node_id in node_ids | ||
| ) | ||
| lambda: all(get_agent_grpc_port(node_id) > 0 for node_id in node_ids) | ||
| ) | ||
| grpc_ports = [ | ||
| json.loads(get_dashboard_agent_address(node_id))[2] for node_id in node_ids | ||
| ] | ||
| grpc_ports = [get_agent_grpc_port(node_id) for node_id in node_ids] |
There was a problem hiding this comment.
This implementation is inefficient as it calls get_all_node_info in a loop, once for each node ID within wait_for_condition and then again when constructing grpc_ports. This can be optimized by fetching all node info once and then processing it.
| def get_agent_grpc_port(node_id_hex: str) -> int: | |
| node_id = NodeID.from_hex(node_id_hex) | |
| node_info_dict = gcs_client.get_all_node_info( | |
| timeout=dashboard_consts.GCS_RPC_TIMEOUT_SECONDS, | |
| state_filter=gcs_pb2.GcsNodeInfo.GcsNodeState.ALIVE, | |
| ) | |
| if node_id not in node_info_dict: | |
| return -1 | |
| return node_info_dict[node_id].metrics_agent_port | |
| wait_for_condition( | |
| lambda: all( | |
| get_dashboard_agent_address(node_id) is not None for node_id in node_ids | |
| ) | |
| lambda: all(get_agent_grpc_port(node_id) > 0 for node_id in node_ids) | |
| ) | |
| grpc_ports = [ | |
| json.loads(get_dashboard_agent_address(node_id))[2] for node_id in node_ids | |
| ] | |
| grpc_ports = [get_agent_grpc_port(node_id) for node_id in node_ids] | |
| def get_ports(): | |
| node_info_dict = gcs_client.get_all_node_info( | |
| timeout=dashboard_consts.GCS_RPC_TIMEOUT_SECONDS, | |
| state_filter=gcs_pb2.GcsNodeInfo.GcsNodeState.ALIVE, | |
| ) | |
| ports = [] | |
| for node_id_hex in node_ids: | |
| node_id = NodeID.from_hex(node_id_hex) | |
| if node_id not in node_info_dict or node_info_dict[node_id].metrics_agent_port <= 0: | |
| return None | |
| ports.append(node_info_dict[node_id].metrics_agent_port) | |
| return ports | |
| wait_for_condition(lambda: get_ports() is not None) | |
| grpc_ports = get_ports() |
Signed-off-by: Tom Liu (LIU HSIU-CHI) <aa123593465@gmail.com>
12f6435 to
dbd7693
Compare
Signed-off-by: Tom Liu (LIU HSIU-CHI) <aa123593465@gmail.com>
Signed-off-by: Tom Liu (LIU HSIU-CHI) <aa123593465@gmail.com>
Signed-off-by: Tom Liu (LIU HSIU-CHI) <aa123593465@gmail.com>
f4c06e4 to
b784f2c
Compare
Description
This PR makes Raylet the sole writer to GCS by removing the legacy logic where the dashboard agent writes its port information directly to GCS's internal KV store.
Background
Previously, the dashboard agent wrote port information to two internal KV keys:
DASHBOARD_AGENT_ADDR_NODE_ID_PREFIX:{node_id}→[ip, http_port, grpc_port]DASHBOARD_AGENT_ADDR_IP_PREFIX:{ip}→[node_id, http_port, grpc_port]After #59613, Raylet already:
persist_port()files written by the dashboard agentGcsNodeInfoduring node registrationThis PR removes the redundant KV writes and updates all consumers to read from
GcsNodeInfoinstead.Changes
5ec0657: Remove dashboard agent direct writes to GCS
async_internal_kv_put()calls fromdashboard/agent.pypersist_port()calls (Raylet depends on these files)9b32247: Update StateDataSourceClient to read from GcsNodeInfo
get_log_service_stub(): UsesGetAllNodeInfoRequestwith node selectorip_to_node_id(): Queries alive nodes, filters by IP addressa7c394f: Update dashboard modules to read from GcsNodeInfo
ReporterHead: Updated_get_stub_address_by_node_id()and_get_stub_address_by_ip()JobHead: Updated_fetch_agent_info()to useasync_get_all_node_info()NodeHead: Removed KV cleanup on node death (GcsNodeInfo is auto-managed by GCS)f74a05b: Update tests to verify GcsNodeInfo integration
metrics_agent_port > 0instead of KV keysArchitecture After This Change
Related issues
Fixes #59666
Additional information
Files Modified
dashboard/agent.py,util/state/state_manager.py,dashboard/modules/reporter/reporter_head.py,dashboard/modules/job/job_head.py,dashboard/modules/node/node_head.py_private/test_utils.py,tests/test_state_api.py,tests/test_ray_event_export_task_events.py,dashboard/tests/test_dashboard.py,dashboard/modules/job/tests/test_sdk.py,dashboard/modules/aggregator/tests/test_aggregator_agent.py,dashboard/modules/aggregator/tests/test_ray_actor_events.pyBackward Compatibility
DASHBOARD_AGENT_ADDR_NODE_ID_PREFIXandDASHBOARD_AGENT_ADDR_IP_PREFIXstill exist but are unused