diff --git a/mindspore_serving/ccsrc/worker/distributed_worker/agent_startup.cc b/mindspore_serving/ccsrc/worker/distributed_worker/agent_startup.cc index 9130c48..b0c00b3 100644 --- a/mindspore_serving/ccsrc/worker/distributed_worker/agent_startup.cc +++ b/mindspore_serving/ccsrc/worker/distributed_worker/agent_startup.cc @@ -16,6 +16,7 @@ #include "worker/distributed_worker/agent_startup.h" #include #include "worker/distributed_worker/notify_distributed/notify_worker.h" +#include "common/grpc_server.h" namespace mindspore { namespace serving { @@ -26,7 +27,7 @@ WorkerAgentStartUp &WorkerAgentStartUp::Instance() { } Status WorkerAgentStartUp::GetAgentsConfigsFromWorker(const std::string &worker_ip, uint32_t worker_port) { - return FAILED; + return GrpcNotifyDistributeWorker::GetAgentsConfigsFromWorker(worker_ip, worker_port, &config_); } Status WorkerAgentStartUp::GetDistributedServableConfig(DistributedServableConfig *config) { diff --git a/mindspore_serving/ccsrc/worker/distributed_worker/distributed_process/distributed_process.cc b/mindspore_serving/ccsrc/worker/distributed_worker/distributed_process/distributed_process.cc index 7758393..224c0fe 100644 --- a/mindspore_serving/ccsrc/worker/distributed_worker/distributed_process/distributed_process.cc +++ b/mindspore_serving/ccsrc/worker/distributed_worker/distributed_process/distributed_process.cc @@ -59,5 +59,51 @@ grpc::Status MSDistributedImpl::AgentFailed(grpc::ServerContext *context, const } return grpc::Status::OK; } + +grpc::Status MSDistributedImpl::AgentConfigAcquire(grpc::ServerContext *context, + const proto::AgentConfigAcquireRequest *request, + proto::AgentConfigAcquireReply *reply) { + Status status(FAILED); + DistributedServableConfig agent_config; + status = servable_->GetDistributedServableConfig(&agent_config); + if (status != SUCCESS) { + MSI_LOG(ERROR) << "Get distributed servable config failed"; + return grpc::Status::CANCELLED; + } + if (agent_config.rank_list.empty()) { + MSI_LOG(ERROR) << "Get distributed servable config failed, config_ not init"; + return grpc::Status::CANCELLED; + } + + MSI_LOG(INFO) << "Begin to set DistributedServableConfig info in reply message"; + // set reply message:AgentConfigAcquireReply, parameter:rank_table_content + reply->set_rank_table_content(agent_config.rank_table_content); + // set reply message:AgentConfigAcquireReply, parameter:rank_list + auto &agent_rank_list = agent_config.rank_list; + for (auto &agent_rank : agent_rank_list) { + auto rank_list = reply->add_rank_list(); + rank_list->set_ip(agent_rank.ip); + rank_list->set_device_id(agent_rank.device_id); + } + // set reply message:AgentConfigAcquireReply, parameter:common_meta + auto reply_common_meta = reply->mutable_common_meta(); + reply_common_meta->set_servable_name(agent_config.common_meta.servable_name); + reply_common_meta->set_with_batch_dim(agent_config.common_meta.with_batch_dim); + auto &without_batch_dim_inputs_list = agent_config.common_meta.without_batch_dim_inputs; + for (auto &without_batch_dim_input : without_batch_dim_inputs_list) { + reply_common_meta->add_without_batch_dim_inputs(without_batch_dim_input); + } + reply_common_meta->set_inputs_count(agent_config.common_meta.inputs_count); + reply_common_meta->set_outputs_count(agent_config.common_meta.outputs_count); + + // set reply message:AgentConfigAcquireReply, parameter:distributed_meta + auto reply_distributed_meta = reply->mutable_distributed_meta(); + reply_distributed_meta->set_rank_size(agent_config.distributed_meta.rank_size); + reply_distributed_meta->set_stage_size(agent_config.distributed_meta.stage_size); + MSI_LOG(INFO) << "Success to set DistributedServableConfig info in reply message"; + + return grpc::Status::OK; +} + } // namespace serving } // namespace mindspore diff --git a/mindspore_serving/ccsrc/worker/distributed_worker/distributed_process/distributed_process.h b/mindspore_serving/ccsrc/worker/distributed_worker/distributed_process/distributed_process.h index 147e7c5..615ea25 100644 --- a/mindspore_serving/ccsrc/worker/distributed_worker/distributed_process/distributed_process.h +++ b/mindspore_serving/ccsrc/worker/distributed_worker/distributed_process/distributed_process.h @@ -43,6 +43,8 @@ class MSDistributedImpl final : public MSWorkerImpl { proto::AgentExitReply *reply) override; grpc::Status AgentFailed(grpc::ServerContext *context, const proto::AgentFailedRequest *request, proto::AgentFailedReply *reply) override; + grpc::Status AgentConfigAcquire(grpc::ServerContext *context, const proto::AgentConfigAcquireRequest *request, + proto::AgentConfigAcquireReply *reply) override; private: std::shared_ptr servable_; diff --git a/mindspore_serving/ccsrc/worker/distributed_worker/distributed_process/distributed_server.h b/mindspore_serving/ccsrc/worker/distributed_worker/distributed_process/distributed_server.h index ca6b967..9affd89 100644 --- a/mindspore_serving/ccsrc/worker/distributed_worker/distributed_process/distributed_server.h +++ b/mindspore_serving/ccsrc/worker/distributed_worker/distributed_process/distributed_server.h @@ -153,6 +153,38 @@ class WorkerAgentFailedContext : public DistributedServiceContext { proto::AgentFailedReply response_; }; +class WorkerAgentConfigAcquireContext : public DistributedServiceContext { + public: + WorkerAgentConfigAcquireContext(MSDistributedImpl *service_impl, proto::MSWorker::AsyncService *async_service, + grpc::ServerCompletionQueue *cq) + : DistributedServiceContext(service_impl, async_service, cq), responder_(&ctx_) {} + + ~WorkerAgentConfigAcquireContext() = default; + static Status EnqueueRequest(MSDistributedImpl *service_impl, proto::MSWorker::AsyncService *async_service, + grpc::ServerCompletionQueue *cq) { + auto call = new WorkerAgentConfigAcquireContext(service_impl, async_service, cq); + call->StartEnqueueRequest(); + return SUCCESS; + } + + void StartEnqueueRequest() override { + state_ = STATE::PROCESS; + async_service_->RequestAgentConfigAcquire(&ctx_, &request_, &responder_, cq_, cq_, this); + } + + void HandleRequest() override { + EnqueueRequest(dist_service_impl_, async_service_, cq_); + state_ = STATE::FINISH; + grpc::Status status = dist_service_impl_->AgentConfigAcquire(&ctx_, &request_, &response_); + responder_.Finish(response_, status, this); + } + + private: + grpc::ServerAsyncResponseWriter responder_; + proto::AgentConfigAcquireRequest request_; + proto::AgentConfigAcquireReply response_; +}; + class DistributedWorkerGrpcServer : public WorkerGrpcServer { public: DistributedWorkerGrpcServer(const std::string &host, int32_t port, MSDistributedImpl *service_impl) @@ -165,6 +197,7 @@ class DistributedWorkerGrpcServer : public WorkerGrpcServer { WorkerAgentRegisterContext::EnqueueRequest(distributed_service_impl_, &svc_, cq_.get()); WorkerAgentExitContext::EnqueueRequest(distributed_service_impl_, &svc_, cq_.get()); WorkerAgentFailedContext::EnqueueRequest(distributed_service_impl_, &svc_, cq_.get()); + WorkerAgentConfigAcquireContext::EnqueueRequest(distributed_service_impl_, &svc_, cq_.get()); return SUCCESS; } diff --git a/mindspore_serving/ccsrc/worker/distributed_worker/distributed_servable.cc b/mindspore_serving/ccsrc/worker/distributed_worker/distributed_servable.cc index 13091b3..030b7fc 100644 --- a/mindspore_serving/ccsrc/worker/distributed_worker/distributed_servable.cc +++ b/mindspore_serving/ccsrc/worker/distributed_worker/distributed_servable.cc @@ -157,7 +157,211 @@ Status DistributedServable::StartServable(const std::string &servable_directory, return SUCCESS; } -Status DistributedServable::InitConfigOnStartup(const std::string &rank_table_json_file) { return FAILED; } +std::string RealPath(const char *path) { + // Return absolute path when path is accessible + std::string res; + char resolved_path[PATH_MAX] = {0}; + if (realpath(path, resolved_path) != nullptr) { + res = resolved_path; + } + + return res; +} + +Status DistributedServable::InitConfigOnStartup(const std::string &rank_table_json_file) { + std::string rank_table_json_abs_path = RealPath(rank_table_json_file.c_str()); + if (rank_table_json_abs_path.empty()) { + return INFER_STATUS_LOG_ERROR(INVALID_INPUTS) << "failed to get realpath of：" << rank_table_json_file.c_str(); + } + + MSI_LOG(INFO) << "Begin to parser rank table json file: " << rank_table_json_file.c_str(); + json rank_table_json; + std::ifstream json_file(rank_table_json_abs_path); + if (!json_file.is_open()) { + return INFER_STATUS_LOG_ERROR(INVALID_INPUTS) << "failed to open rank table file：" << rank_table_json_file.c_str(); + } + + json_file >> rank_table_json; + if (!rank_table_json.is_object()) { + return INFER_STATUS_LOG_ERROR(INVALID_INPUTS) << rank_table_json_file.c_str() << " is not json object"; + } + + if (rank_table_json.find("group_list") != rank_table_json.end()) { + return ParserRankTableWithGroupList(rank_table_json_file, rank_table_json); + } else { + return ParserRankTableWithServerList(rank_table_json_file, rank_table_json); + } +} + +json DistributedServable::ParserArrayInJson(const json &json_array, const std::string &str) { + json temp_array; + auto iter = json_array.find(str); + if (iter == json_array.end()) { + MSI_LOG_ERROR << "Check rank table file failed" << str << "in file is not find"; + return temp_array; + } + if (!iter->is_array()) { + MSI_LOG_ERROR << "Check rank table file failed" << str << "in file is not array"; + return temp_array; + } + temp_array = json_array.at(str); + return temp_array; +} + +json DistributedServable::ParserStringInJson(const json &json_str, const std::string &str) { + json temp_str; + auto iter = json_str.find(str); + if (iter == json_str.end()) { + MSI_LOG_ERROR << "Check rank table file failed" << str << "in file is not find"; + return temp_str; + } + if (!iter->is_string()) { + MSI_LOG_ERROR << "Check rank table file failed" << str << "in file is not string"; + return temp_str; + } + temp_str = json_str.at(str); + return temp_str; +} + +Status DistributedServable::ParserRankTableWithGroupList(const std::string &rank_table_json_file, + const json &rank_table_json) { + MSI_LOG_INFO << "Begin to parser rank table with group list"; + auto server_list = ParserArrayInJson(rank_table_json, "group_list"); + if (server_list.empty()) { + return INFER_STATUS_LOG_ERROR(INVALID_INPUTS) << "group_list attr is empty in" << rank_table_json_file.c_str(); + } + + size_t rank_id = 0; + for (auto &server : server_list) { + auto instance_list = ParserArrayInJson(server, "instance_list"); + if (instance_list.empty()) { + return INFER_STATUS_LOG_ERROR(INVALID_INPUTS) << "instance_list attr is empty in" << rank_table_json_file.c_str(); + } + + for (auto &instance : instance_list) { + auto str_server_id = ParserStringInJson(instance, "server_id"); + if (str_server_id.empty()) { + return INFER_STATUS_LOG_ERROR(INVALID_INPUTS) << "server_id attr is empty in" << rank_table_json_file.c_str(); + } + + OneRankConfig one_rank_config; + one_rank_config.ip = str_server_id; + + auto devices = ParserArrayInJson(instance, "devices"); + if (devices.empty()) { + return INFER_STATUS_LOG_ERROR(INVALID_INPUTS) << "devices attr is empty in" << rank_table_json_file.c_str(); + } + + auto str_device_id = ParserStringInJson(devices.at(0), "device_id"); + if (str_device_id.empty()) { + return INFER_STATUS_LOG_ERROR(INVALID_INPUTS) << "device_id attr is empty in" << rank_table_json_file.c_str(); + } + auto temp_str_device_id = str_device_id.get(); + uint32_t temp_device_id; + auto status = ConvertStr2Int(rank_table_json_file, temp_str_device_id, "device_id", &temp_device_id); + if (status != SUCCESS) { + MSI_LOG_ERROR << "Convert device_id from string to int failed"; + return status; + } + + auto str_rank_id = ParserStringInJson(instance, "rank_id"); + if (str_rank_id.empty()) { + return INFER_STATUS_LOG_ERROR(INVALID_INPUTS) << "rank_id attr is empty in" << rank_table_json_file.c_str(); + } + auto temp_str_rank_id = str_rank_id.get(); + uint32_t temp_rank_id; + status = ConvertStr2Int(rank_table_json_file, temp_str_rank_id, "rank_id", &temp_rank_id); + if (status != SUCCESS) { + MSI_LOG_ERROR << "Convert rank_id from string to int failed"; + return status; + } + + if (rank_id != temp_rank_id) { + return INFER_STATUS_LOG_ERROR(INVALID_INPUTS) + << "device size not match rank_id in" << rank_table_json_file.c_str(); + } + rank_id++; + one_rank_config.device_id = temp_device_id; + config_.rank_list.push_back(one_rank_config); + } + } + MSI_LOG(INFO) << "Success parser rank table json file with group list and save to DistributedServableConfig"; + + return SUCCESS; +} +Status DistributedServable::ConvertStr2Int(const std::string &rank_table_json_file, const std::string ¶_str, + const std::string ¶_key, uint32_t *para_int) const { + try { + *para_int = std::stoi(para_str); + } catch (std::invalid_argument &) { + return INFER_STATUS_LOG_ERROR(INVALID_INPUTS) + << para_key << "attr is invalid argument in" << rank_table_json_file.c_str(); + } catch (std::out_of_range &) { + return INFER_STATUS_LOG_ERROR(INVALID_INPUTS) + << para_key << "attr is out of range in" << rank_table_json_file.c_str(); + } catch (...) { + return INFER_STATUS_LOG_ERROR(INVALID_INPUTS) + << "convert" << para_key << "attr exception occurred in" << rank_table_json_file.c_str(); + } + return SUCCESS; +} + +Status DistributedServable::ParserRankTableWithServerList(const std::string &rank_table_json_file, + const json &rank_table_json) { + MSI_LOG_INFO << "Begin to parser rank table with server list"; + auto server_list = ParserArrayInJson(rank_table_json, "server_list"); + if (server_list.empty()) { + return INFER_STATUS_LOG_ERROR(INVALID_INPUTS) << "server_list attr is empty in" << rank_table_json_file.c_str(); + } + + size_t rank_id = 0; + for (auto &server : server_list) { + auto server_id = ParserStringInJson(server, "server_id"); + if (server_id.empty()) { + return INFER_STATUS_LOG_ERROR(INVALID_INPUTS) << "server_id attr is empty in" << rank_table_json_file.c_str(); + } + + auto device_list = ParserArrayInJson(server, "device"); + if (device_list.empty()) { + return INFER_STATUS_LOG_ERROR(INVALID_INPUTS) << "device attr is empty in" << rank_table_json_file.c_str(); + } + + for (auto &device : device_list) { + OneRankConfig one_rank_config; + one_rank_config.ip = server_id; + auto str_device_id = ParserStringInJson(device, "device_id"); + if (str_device_id.empty()) { + return INFER_STATUS_LOG_ERROR(INVALID_INPUTS) << "device_id attr is empty in" << rank_table_json_file.c_str(); + } + auto temp_str_device_id = str_device_id.get(); + uint32_t temp_device_id; + auto status = ConvertStr2Int(rank_table_json_file, temp_str_device_id, "device_id", &temp_device_id); + if (status != SUCCESS) { + MSI_LOG_ERROR << "Convert device_id from string to int failed"; + return status; + } + + auto str_rank_id = ParserStringInJson(device, "rank_id"); + if (str_rank_id.empty()) { + return INFER_STATUS_LOG_ERROR(INVALID_INPUTS) << "rank_id attr is empty in" << rank_table_json_file.c_str(); + } + auto temp_str_rank_id = str_rank_id.get(); + uint32_t temp_rank_id; + status = ConvertStr2Int(rank_table_json_file, temp_str_rank_id, "rank_id", &temp_rank_id); + if (status != SUCCESS) { + MSI_LOG_ERROR << "Convert rank_id from string to int failed"; + return status; + } + + rank_id++; + one_rank_config.device_id = temp_device_id; + config_.rank_list.push_back(one_rank_config); + } + } + MSI_LOG(INFO) << "Success parser rank table json file with server list and save to DistributedServableConfig"; + + return SUCCESS; +} Status DistributedServable::WaitAgentsReady(uint64_t wait_agents_time_in_seconds) { MSI_LOG_INFO << "Begin waiting ready of all agents"; diff --git a/mindspore_serving/ccsrc/worker/distributed_worker/distributed_servable.h b/mindspore_serving/ccsrc/worker/distributed_worker/distributed_servable.h index c8519d8..8bb4600 100644 --- a/mindspore_serving/ccsrc/worker/distributed_worker/distributed_servable.h +++ b/mindspore_serving/ccsrc/worker/distributed_worker/distributed_servable.h @@ -17,14 +17,17 @@ #ifndef MINDSPORE_SERVING_WORKER_DISTRIBUTED_SERVABLE_H #define MINDSPORE_SERVING_WORKER_DISTRIBUTED_SERVABLE_H +#include #include #include #include #include +#include #include "worker/sevable_base.h" #include "worker/distributed_worker/common.h" #include "worker/distributed_worker/notify_agent/base_notify_agent.h" +using nlohmann::json; namespace mindspore { namespace serving { @@ -85,6 +88,15 @@ class MS_API DistributedServable : public ServableBase { Status CheckRankConfig(); void SetWaitAgentsPromise(bool flag); // agent stubs + Status ParserRankTableWithGroupList(const std::string &rank_table_json_file, const json &rank_table_json); + + Status ParserRankTableWithServerList(const std::string &rank_table_json_file, const json &rank_table_json); + + json ParserArrayInJson(const json &json_array, const std::string &str); + + json ParserStringInJson(const json &json_str, const std::string &str); + Status ConvertStr2Int(const std::string &rank_table_json_file, const std::string ¶_str, + const std::string ¶_key, uint32_t *para_int) const; }; } // namespace serving diff --git a/mindspore_serving/ccsrc/worker/distributed_worker/notify_distributed/notify_worker.cc b/mindspore_serving/ccsrc/worker/distributed_worker/notify_distributed/notify_worker.cc index ad1d522..710343c 100644 --- a/mindspore_serving/ccsrc/worker/distributed_worker/notify_distributed/notify_worker.cc +++ b/mindspore_serving/ccsrc/worker/distributed_worker/notify_distributed/notify_worker.cc @@ -104,5 +104,60 @@ Status GrpcNotifyDistributeWorker::NotifyFailed(const std::string &worker_ip, ui return INFER_STATUS_LOG_ERROR(SYSTEM_ERROR) << "Failed to notify failure of agent"; } +Status GrpcNotifyDistributeWorker::GetAgentsConfigsFromWorker(const std::string &worker_ip, uint32_t worker_port, + DistributedServableConfig *config) { + const int32_t REGISTER_TIME_OUT = 60; + const int32_t REGISTER_INTERVAL = 1; + auto loop = REGISTER_TIME_OUT; + while (loop-- && !ExitSignalHandle::Instance().HasStopped()) { + auto address = worker_ip + ":" + std::to_string(worker_port); + auto channel = GrpcServer::CreateChannel(address); + auto stub = proto::MSWorker::NewStub(channel); + + grpc::ClientContext context; + proto::AgentConfigAcquireRequest request; + proto::AgentConfigAcquireReply reply; + std::chrono::system_clock::time_point deadline = + std::chrono::system_clock::now() + std::chrono::seconds(REGISTER_INTERVAL); + context.set_deadline(deadline); + grpc::Status status = stub->AgentConfigAcquire(&context, request, &reply); + if (status.ok()) { + MSI_LOG(INFO) << "Success to get Agents configs from Worker, and begin to parser"; + // parser reply message:AgentConfigAcquireReply, parameter:rank_table_content + config->rank_table_content = reply.rank_table_content(); + // parser reply message:AgentConfigAcquireReply, parameter:rank_list + for (auto &temp_rank : reply.rank_list()) { + OneRankConfig ome_rank_config; + ome_rank_config.ip = temp_rank.ip(); + ome_rank_config.device_id = temp_rank.device_id(); + config->rank_list.push_back(ome_rank_config); + } + // parser reply message:AgentConfigAcquireReply, parameter:common_meta + auto &temp_common_meta = reply.common_meta(); + config->common_meta.servable_name = temp_common_meta.servable_name(); + config->common_meta.with_batch_dim = temp_common_meta.with_batch_dim(); + for (auto &temp_without_batch_dim_inputs : temp_common_meta.without_batch_dim_inputs()) { + config->common_meta.without_batch_dim_inputs.push_back(temp_without_batch_dim_inputs); + } + config->common_meta.inputs_count = temp_common_meta.inputs_count(); + config->common_meta.outputs_count = temp_common_meta.outputs_count(); + + // parser reply message:AgentConfigAcquireReply, parameter:distributed_meta + auto &temp_distributed_meta = reply.distributed_meta(); + config->distributed_meta.rank_size = temp_distributed_meta.rank_size(); + config->distributed_meta.stage_size = temp_distributed_meta.stage_size(); + MSI_LOG(INFO) << "Success to parser reply message and save to DistributedServableConfig"; + + return SUCCESS; + } + MSI_LOG_INFO << "Grpc message: " << status.error_code() << ", " << status.error_message(); + std::this_thread::sleep_for(std::chrono::milliseconds(REGISTER_INTERVAL * 1000)); + } + if (ExitSignalHandle::Instance().HasStopped()) { + return INFER_STATUS_LOG_WARNING(FAILED) << "Agent exit, stop get Agents configs from Worker"; + } + return INFER_STATUS_LOG_ERROR(SYSTEM_ERROR) << "Failed to get Agents configs from Worker"; +} + } // namespace serving } // namespace mindspore diff --git a/mindspore_serving/ccsrc/worker/distributed_worker/notify_distributed/notify_worker.h b/mindspore_serving/ccsrc/worker/distributed_worker/notify_distributed/notify_worker.h index da509ff..cf11d55 100644 --- a/mindspore_serving/ccsrc/worker/distributed_worker/notify_distributed/notify_worker.h +++ b/mindspore_serving/ccsrc/worker/distributed_worker/notify_distributed/notify_worker.h @@ -37,6 +37,8 @@ class MS_API GrpcNotifyDistributeWorker { Status Unregister(); // from start up, not agent static Status NotifyFailed(const std::string &worker_ip, uint32_t worker_port); + static Status GetAgentsConfigsFromWorker(const std::string &worker_ip, uint32_t worker_port, + DistributedServableConfig *config); private: std::string distributed_worker_ip_; diff --git a/mindspore_serving/proto/ms_distributed.proto b/mindspore_serving/proto/ms_distributed.proto index 27fa6c4..21db3ca 100644 --- a/mindspore_serving/proto/ms_distributed.proto +++ b/mindspore_serving/proto/ms_distributed.proto @@ -27,6 +27,19 @@ message AgentSpec { repeated Tensor outputs = 4; } +message CommonServableMeta { + string servable_name = 1; + bool with_batch_dim = 2; + repeated int64 without_batch_dim_inputs = 3; + int64 inputs_count = 4; + int64 outputs_count = 5; +} + +message DistributedServableMeta { + int64 rank_size = 1; + int64 stage_size = 2; +} + message AgentRegisterRequest { repeated AgentSpec agent_spec = 1; string address = 2; @@ -51,3 +64,17 @@ message AgentFailedRequest { message AgentFailedReply { ErrorMsg error_msg = 1; } + +message AgentConfigAcquireRequest { +} + +message AgentConfigAcquireReply { + message OneRankConfig { + string ip = 1; + int64 device_id = 2; + } + string rank_table_content = 1; + repeated OneRankConfig rank_list = 2; + CommonServableMeta common_meta = 3; + DistributedServableMeta distributed_meta = 4; +} \ No newline at end of file diff --git a/mindspore_serving/proto/ms_worker.proto b/mindspore_serving/proto/ms_worker.proto index 7b2dbe0..b63fd8e 100644 --- a/mindspore_serving/proto/ms_worker.proto +++ b/mindspore_serving/proto/ms_worker.proto @@ -30,4 +30,5 @@ service MSWorker { rpc AgentExit(AgentExitRequest) returns (AgentExitReply) {} rpc AgentRegister(AgentRegisterRequest) returns (AgentRegisterReply) {} rpc AgentFailed(AgentFailedRequest) returns (AgentFailedReply) {} + rpc AgentConfigAcquire(AgentConfigAcquireRequest) returns (AgentConfigAcquireReply) {} }