diff --git a/src/ray/gcs/accessor.h b/src/ray/gcs/accessor.h index 5c9cef3e7..83dc3de3c 100644 --- a/src/ray/gcs/accessor.h +++ b/src/ray/gcs/accessor.h @@ -657,13 +657,14 @@ class WorkerInfoAccessor { virtual ~WorkerInfoAccessor() = default; /// Subscribe to all unexpected failure of workers from GCS asynchronously. - /// Note that this does not include workers that failed due to node failure. + /// Note that this does not include workers that failed due to node failure + /// and only fileds in WorkerDeltaData would be published. /// /// \param subscribe Callback that will be called each time when a worker failed. /// \param done Callback that will be called when subscription is complete. /// \return Status virtual Status AsyncSubscribeToWorkerFailures( - const ItemCallback &subscribe, + const ItemCallback &subscribe, const StatusCallback &done) = 0; /// Report a worker failure to GCS asynchronously. diff --git a/src/ray/gcs/gcs_client/service_based_accessor.cc b/src/ray/gcs/gcs_client/service_based_accessor.cc index 50379f58a..f9380b78e 100644 --- a/src/ray/gcs/gcs_client/service_based_accessor.cc +++ b/src/ray/gcs/gcs_client/service_based_accessor.cc @@ -1311,11 +1311,11 @@ ServiceBasedWorkerInfoAccessor::ServiceBasedWorkerInfoAccessor( : client_impl_(client_impl) {} Status ServiceBasedWorkerInfoAccessor::AsyncSubscribeToWorkerFailures( - const ItemCallback &subscribe, const StatusCallback &done) { + const ItemCallback &subscribe, const StatusCallback &done) { RAY_CHECK(subscribe != nullptr); subscribe_operation_ = [this, subscribe](const StatusCallback &done) { auto on_subscribe = [subscribe](const std::string &id, const std::string &data) { - rpc::WorkerTableData worker_failure_data; + rpc::WorkerDeltaData worker_failure_data; worker_failure_data.ParseFromString(data); subscribe(worker_failure_data); }; diff --git a/src/ray/gcs/gcs_client/service_based_accessor.h b/src/ray/gcs/gcs_client/service_based_accessor.h index eca23dd11..b498e0acf 100644 --- a/src/ray/gcs/gcs_client/service_based_accessor.h +++ b/src/ray/gcs/gcs_client/service_based_accessor.h @@ -407,7 +407,7 @@ class ServiceBasedWorkerInfoAccessor : public WorkerInfoAccessor { virtual ~ServiceBasedWorkerInfoAccessor() = default; Status AsyncSubscribeToWorkerFailures( - const ItemCallback &subscribe, + const ItemCallback &subscribe, const StatusCallback &done) override; Status AsyncReportWorkerFailure(const std::shared_ptr &data_ptr, diff --git a/src/ray/gcs/gcs_client/test/service_based_gcs_client_test.cc b/src/ray/gcs/gcs_client/test/service_based_gcs_client_test.cc index 64093880f..3b0f731bb 100644 --- a/src/ray/gcs/gcs_client/test/service_based_gcs_client_test.cc +++ b/src/ray/gcs/gcs_client/test/service_based_gcs_client_test.cc @@ -495,7 +495,7 @@ class ServiceBasedGcsClientTest : public ::testing::Test { } bool SubscribeToWorkerFailures( - const gcs::ItemCallback &subscribe) { + const gcs::ItemCallback &subscribe) { std::promise promise; RAY_CHECK_OK(gcs_client_->Workers().AsyncSubscribeToWorkerFailures( subscribe, [&promise](Status status) { promise.set_value(status.ok()); })); @@ -922,7 +922,7 @@ TEST_F(ServiceBasedGcsClientTest, TestStats) { TEST_F(ServiceBasedGcsClientTest, TestWorkerInfo) { // Subscribe to all unexpected failure of workers from GCS. std::atomic worker_failure_count(0); - auto on_subscribe = [&worker_failure_count](const rpc::WorkerTableData &result) { + auto on_subscribe = [&worker_failure_count](const rpc::WorkerDeltaData &result) { ++worker_failure_count; }; ASSERT_TRUE(SubscribeToWorkerFailures(on_subscribe)); @@ -1168,7 +1168,7 @@ TEST_F(ServiceBasedGcsClientTest, TestTaskTableResubscribe) { TEST_F(ServiceBasedGcsClientTest, TestWorkerTableResubscribe) { // Subscribe to all unexpected failure of workers from GCS. std::atomic worker_failure_count(0); - auto on_subscribe = [&worker_failure_count](const rpc::WorkerTableData &result) { + auto on_subscribe = [&worker_failure_count](const rpc::WorkerDeltaData &result) { ++worker_failure_count; }; ASSERT_TRUE(SubscribeToWorkerFailures(on_subscribe)); diff --git a/src/ray/gcs/gcs_server/gcs_worker_manager.cc b/src/ray/gcs/gcs_server/gcs_worker_manager.cc index 894f56535..b67b89dbf 100644 --- a/src/ray/gcs/gcs_server/gcs_worker_manager.cc +++ b/src/ray/gcs/gcs_server/gcs_worker_manager.cc @@ -52,8 +52,15 @@ void GcsWorkerManager::HandleReportWorkerFailure( << ", address = " << worker_address.ip_address(); } else { stats::UnintentionalWorkerFailures.Record(1); + // Only publish worker_id and raylet_id in address as they are the only fields used + // by sub clients. + auto worker_failure_delta = std::make_shared(); + worker_failure_delta->set_worker_id( + worker_failure_data->worker_address().worker_id()); + worker_failure_delta->set_raylet_id( + worker_failure_data->worker_address().raylet_id()); RAY_CHECK_OK(gcs_pub_sub_->Publish(WORKER_CHANNEL, worker_id.Hex(), - worker_failure_data->SerializeAsString(), + worker_failure_delta->SerializeAsString(), nullptr)); } GCS_RPC_SEND_REPLY(send_reply_callback, reply, status); diff --git a/src/ray/protobuf/gcs.proto b/src/ray/protobuf/gcs.proto index ab7611d4e..d0793c35c 100644 --- a/src/ray/protobuf/gcs.proto +++ b/src/ray/protobuf/gcs.proto @@ -393,6 +393,12 @@ message WorkerTableData { map worker_info = 6; } +// Fields to publish when worker fails. +message WorkerDeltaData { + bytes raylet_id = 1; + bytes worker_id = 2; +} + message ResourceMap { map items = 1; } diff --git a/src/ray/raylet/node_manager.cc b/src/ray/raylet/node_manager.cc index d3e531c48..7d4f97d0e 100644 --- a/src/ray/raylet/node_manager.cc +++ b/src/ray/raylet/node_manager.cc @@ -299,8 +299,8 @@ ray::Status NodeManager::RegisterGcs() { // node failure. These workers can be identified by comparing the raylet_id // in their rpc::Address to the ID of a failed raylet. const auto &worker_failure_handler = - [this](const rpc::WorkerTableData &worker_failure_data) { - HandleUnexpectedWorkerFailure(worker_failure_data.worker_address()); + [this](const rpc::WorkerDeltaData &worker_failure_data) { + HandleUnexpectedWorkerFailure(worker_failure_data); }; RAY_CHECK_OK(gcs_client_->Workers().AsyncSubscribeToWorkerFailures( worker_failure_handler, /*done_callback=*/nullptr)); @@ -716,14 +716,14 @@ void NodeManager::NodeRemoved(const NodeID &node_id) { // Clean up workers that were owned by processes that were on the failed // node. - rpc::Address address; - address.set_raylet_id(node_id.Binary()); - HandleUnexpectedWorkerFailure(address); + rpc::WorkerDeltaData data; + data.set_raylet_id(node_id.Binary()); + HandleUnexpectedWorkerFailure(data); } -void NodeManager::HandleUnexpectedWorkerFailure(const rpc::Address &address) { - const WorkerID worker_id = WorkerID::FromBinary(address.worker_id()); - const NodeID node_id = NodeID::FromBinary(address.raylet_id()); +void NodeManager::HandleUnexpectedWorkerFailure(const rpc::WorkerDeltaData &data) { + const WorkerID worker_id = WorkerID::FromBinary(data.worker_id()); + const NodeID node_id = NodeID::FromBinary(data.raylet_id()); if (!worker_id.IsNil()) { RAY_LOG(DEBUG) << "Worker " << worker_id << " failed"; failed_workers_cache_.insert(worker_id); diff --git a/src/ray/raylet/node_manager.h b/src/ray/raylet/node_manager.h index d7b7e6df1..d626e5246 100644 --- a/src/ray/raylet/node_manager.h +++ b/src/ray/raylet/node_manager.h @@ -172,8 +172,8 @@ class NodeManager : public rpc::NodeManagerServiceHandler, /// Handle an unexpected failure notification from GCS pubsub. /// - /// \param worker_address The address of the worker that died. - void HandleUnexpectedWorkerFailure(const rpc::Address &worker_address); + /// \param data The data of the worker that died. + void HandleUnexpectedWorkerFailure(const rpc::WorkerDeltaData &data); /// Handler for the addition of a new node. ///