diff --git a/src/ray/gcs/gcs_server/gcs_node_manager.cc b/src/ray/gcs/gcs_server/gcs_node_manager.cc index 77d12595c..2dd1cd557 100644 --- a/src/ray/gcs/gcs_server/gcs_node_manager.cc +++ b/src/ray/gcs/gcs_server/gcs_node_manager.cc @@ -172,12 +172,14 @@ void GcsNodeManager::HandleRegisterNode(const rpc::RegisterNodeRequest &request, rpc::RegisterNodeReply *reply, rpc::SendReplyCallback send_reply_callback) { ClientID node_id = ClientID::FromBinary(request.node_info().node_id()); - RAY_LOG(INFO) << "Registering node info, node id = " << node_id; + RAY_LOG(INFO) << "Registering node info, node id = " << node_id + << ", address = " << request.node_info().node_manager_address(); AddNode(std::make_shared(request.node_info())); auto on_done = [this, node_id, request, reply, send_reply_callback](const Status &status) { RAY_CHECK_OK(status); - RAY_LOG(INFO) << "Finished registering node info, node id = " << node_id; + RAY_LOG(INFO) << "Finished registering node info, node id = " << node_id + << ", address = " << request.node_info().node_manager_address(); RAY_CHECK_OK(gcs_pub_sub_->Publish(NODE_CHANNEL, node_id.Hex(), request.node_info().SerializeAsString(), nullptr)); GCS_RPC_SEND_REPLY(send_reply_callback, reply, status); @@ -214,7 +216,6 @@ void GcsNodeManager::HandleUnregisterNode(const rpc::UnregisterNodeRequest &requ void GcsNodeManager::HandleGetAllNodeInfo(const rpc::GetAllNodeInfoRequest &request, rpc::GetAllNodeInfoReply *reply, rpc::SendReplyCallback send_reply_callback) { - RAY_LOG(DEBUG) << "Getting all nodes info."; for (const auto &entry : alive_nodes_) { reply->add_node_info_list()->CopyFrom(*entry.second); } @@ -222,7 +223,6 @@ void GcsNodeManager::HandleGetAllNodeInfo(const rpc::GetAllNodeInfoRequest &requ reply->add_node_info_list()->CopyFrom(*entry.second); } GCS_RPC_SEND_REPLY(send_reply_callback, reply, Status::OK()); - RAY_LOG(DEBUG) << "Finished getting all node info."; } void GcsNodeManager::HandleReportHeartbeat(const rpc::ReportHeartbeatRequest &request, @@ -245,7 +245,6 @@ void GcsNodeManager::HandleGetResources(const rpc::GetResourcesRequest &request, rpc::GetResourcesReply *reply, rpc::SendReplyCallback send_reply_callback) { ClientID node_id = ClientID::FromBinary(request.node_id()); - RAY_LOG(DEBUG) << "Getting node resources, node id = " << node_id; auto iter = cluster_resources_.find(node_id); if (iter != cluster_resources_.end()) { for (auto &resource : iter->second.items()) { @@ -253,7 +252,6 @@ void GcsNodeManager::HandleGetResources(const rpc::GetResourcesRequest &request, } } GCS_RPC_SEND_REPLY(send_reply_callback, reply, Status::OK()); - RAY_LOG(DEBUG) << "Finished getting node resources, node id = " << node_id; } void GcsNodeManager::HandleUpdateResources(const rpc::UpdateResourcesRequest &request, diff --git a/src/ray/gcs/gcs_server/gcs_worker_manager.cc b/src/ray/gcs/gcs_server/gcs_worker_manager.cc index 35ca0769d..3aa2898ef 100644 --- a/src/ray/gcs/gcs_server/gcs_worker_manager.cc +++ b/src/ray/gcs/gcs_server/gcs_worker_manager.cc @@ -21,44 +21,56 @@ void GcsWorkerManager::HandleReportWorkerFailure( const rpc::ReportWorkerFailureRequest &request, rpc::ReportWorkerFailureReply *reply, rpc::SendReplyCallback send_reply_callback) { const rpc::Address worker_address = request.worker_failure().worker_address(); - RAY_LOG(DEBUG) << "Reporting worker failure, " << worker_address.DebugString(); + const auto worker_id = WorkerID::FromBinary(worker_address.worker_id()); + const auto node_id = ClientID::FromBinary(worker_address.raylet_id()); + std::stringstream log_stream; + log_stream << "Reporting worker failure, worker id = " << worker_id + << ", node id = " << node_id + << ", address = " << worker_address.ip_address(); + if (request.worker_failure().intentional_disconnect()) { + RAY_LOG(INFO) << log_stream.str(); + } else { + RAY_LOG(WARNING) << log_stream.str(); + } auto worker_failure_data = std::make_shared(); worker_failure_data->CopyFrom(request.worker_failure()); worker_failure_data->set_is_alive(false); - const auto worker_id = WorkerID::FromBinary(worker_address.worker_id()); // Before handle ReportWorkerFailureRequest, you should check if the worker is exists. - auto on_get_done = - [this, worker_address, worker_id, worker_failure_data, reply, send_reply_callback]( - const Status &status, const boost::optional &result) { - if (result) { - auto on_put_done = [this, worker_address, worker_id, worker_failure_data, reply, - send_reply_callback](const Status &status) { - if (!status.ok()) { - RAY_LOG(ERROR) << "Failed to report worker failure, " - << worker_address.DebugString(); - } else { - RAY_CHECK_OK(gcs_pub_sub_->Publish(WORKER_CHANNEL, worker_id.Binary(), - worker_failure_data->SerializeAsString(), - nullptr)); - } - GCS_RPC_SEND_REPLY(send_reply_callback, reply, status); - }; - - // The worker exists in worker table, you can update the info of this worker. - Status report_status = gcs_table_storage_->WorkerTable().Put( - worker_id, *worker_failure_data, on_put_done); - if (!report_status.ok()) { - on_put_done(report_status); - } + auto on_get_done = [this, worker_address, worker_id, node_id, worker_failure_data, + reply, send_reply_callback]( + const Status &status, + const boost::optional &result) { + if (result) { + auto on_put_done = [this, worker_address, worker_id, node_id, worker_failure_data, + reply, send_reply_callback](const Status &status) { + if (!status.ok()) { + RAY_LOG(ERROR) << "Failed to report worker failure, worker id = " << worker_id + << ", node id = " << node_id + << ", address = " << worker_address.ip_address(); } else { - // The worker doesn't exists in worker table. - RAY_LOG(WARNING) << "Failed to report worker failure, the worker doesn't " - "exist, " - << worker_address.DebugString(); - GCS_RPC_SEND_REPLY(send_reply_callback, reply, status); + RAY_CHECK_OK(gcs_pub_sub_->Publish(WORKER_CHANNEL, worker_id.Binary(), + worker_failure_data->SerializeAsString(), + nullptr)); } + GCS_RPC_SEND_REPLY(send_reply_callback, reply, status); }; + + // The worker exists in worker table, you can update the info of this worker. + Status report_status = gcs_table_storage_->WorkerTable().Put( + worker_id, *worker_failure_data, on_put_done); + if (!report_status.ok()) { + on_put_done(report_status); + } + } else { + // The worker doesn't exists in worker table. + RAY_LOG(WARNING) << "Failed to report worker failure, the worker doesn't " + "exist, worker id = " + << worker_id << ", node id = " << node_id + << ", address = " << worker_address.ip_address(); + GCS_RPC_SEND_REPLY(send_reply_callback, reply, status); + } + }; Status status = gcs_table_storage_->WorkerTable().Get(worker_id, on_get_done); if (!status.ok()) { on_get_done(status, boost::none); diff --git a/src/ray/raylet/node_manager.cc b/src/ray/raylet/node_manager.cc index 42e95b8f4..7914b04b1 100644 --- a/src/ray/raylet/node_manager.cc +++ b/src/ray/raylet/node_manager.cc @@ -1382,9 +1382,9 @@ void NodeManager::ProcessDisconnectClientMessage( local_queues_.RemoveDriverTaskId(TaskID::ComputeDriverTaskId(driver_id)); worker_pool_.DisconnectDriver(worker); - RAY_LOG(DEBUG) << "Driver (pid=" << worker->GetProcess().GetId() - << ") is disconnected. " - << "job_id: " << worker->GetAssignedJobId(); + RAY_LOG(INFO) << "Driver (pid=" << worker->GetProcess().GetId() + << ") is disconnected. " + << "job_id: " << worker->GetAssignedJobId(); } client->Close();