diff --git a/components/spider-execution-manager/src/client/grpc/liveness.rs b/components/spider-execution-manager/src/client/grpc/liveness.rs index 013475657..612d6255b 100644 --- a/components/spider-execution-manager/src/client/grpc/liveness.rs +++ b/components/spider-execution-manager/src/client/grpc/liveness.rs @@ -6,13 +6,14 @@ use async_trait::async_trait; use spider_core::types::id::{ExecutionManagerId, SessionId}; use spider_proto_rust::storage::{ self, - execution_manager_liveness_error, execution_manager_liveness_service_client::ExecutionManagerLivenessServiceClient, - register_execution_manager_response, - update_execution_manager_heartbeat_response, }; use spider_utils::grpc::client::ConnectionPool; -use tonic::transport::{Channel, Endpoint}; +use tonic::{ + Code, + Status, + transport::{Channel, Endpoint}, +}; use crate::client::liveness::{LivenessClient, LivenessResponseError, RegistrationResponse}; @@ -59,7 +60,7 @@ impl LivenessClient for GrpcLivenessClient { .get_client() .register_execution_manager(request) .await - .map_err(to_transport_error)? + .map_err(|status| status_to_error(&status))? .into_inner(); register_response_to_result(response) @@ -77,28 +78,27 @@ impl LivenessClient for GrpcLivenessClient { .get_client() .update_execution_manager_heartbeat(request) .await - .map_err(to_transport_error)? + .map_err(|status| status_to_error(&status))? .into_inner(); - heartbeat_response_to_result(response) + Ok(heartbeat_response_to_result(response)) } } -impl From for LivenessResponseError { - fn from(error: storage::ExecutionManagerLivenessError) -> Self { - match execution_manager_liveness_error::ErrCode::try_from(error.err_code) { - Ok(execution_manager_liveness_error::ErrCode::MarkedDead) => Self::MarkedDead, - Ok(execution_manager_liveness_error::ErrCode::InvalidInput) => { - Self::IllegalId(error.message) - } - Ok( - execution_manager_liveness_error::ErrCode::Server - | execution_manager_liveness_error::ErrCode::Unspecified, - ) => Self::Transport(error.message), - Err(error) => Self::Transport(format!( - "unknown execution manager liveness error kind: {error}" - )), - } +/// Maps an execution-manager-liveness gRPC [`Status`] to a [`LivenessResponseError`]. +/// +/// # Returns +/// +/// The [`LivenessResponseError`] for `status`'s code: +/// +/// * [`LivenessResponseError::MarkedDead`] for `FAILED_PRECONDITION`. +/// * [`LivenessResponseError::IllegalId`] for `INVALID_ARGUMENT`. +/// * [`LivenessResponseError::Transport`] for any other code. +fn status_to_error(status: &Status) -> LivenessResponseError { + match status.code() { + Code::FailedPrecondition => LivenessResponseError::MarkedDead, + Code::InvalidArgument => LivenessResponseError::IllegalId(status.message().to_owned()), + _ => LivenessResponseError::Transport(status.message().to_owned()), } } @@ -109,38 +109,24 @@ impl From for LivenessResponseError { fn register_response_to_result( response: storage::RegisterExecutionManagerResponse, ) -> Result { - match response.result { - Some(register_execution_manager_response::Result::Registration(registration)) => { - Ok(RegistrationResponse { - em_id: ExecutionManagerId::from(registration.execution_manager_id), - session_id: registration.session_id, - }) - } - Some(register_execution_manager_response::Result::Error(error)) => Err(error.into()), - None => Err(LivenessResponseError::Transport( - "register execution manager response missing result".to_owned(), - )), - } + let registration = response.registration.ok_or_else(|| { + LivenessResponseError::Transport( + "register execution manager response missing registration".to_owned(), + ) + })?; + Ok(RegistrationResponse { + em_id: ExecutionManagerId::from(registration.execution_manager_id), + session_id: registration.session_id, + }) } /// # Returns /// -/// [`storage::UpdateExecutionManagerHeartbeatResponse`] converted into -/// [`Result`]. -fn heartbeat_response_to_result( +/// The [`SessionId`] carried by `response`. +const fn heartbeat_response_to_result( response: storage::UpdateExecutionManagerHeartbeatResponse, -) -> Result { - match response.result { - Some(update_execution_manager_heartbeat_response::Result::SessionId(session_id)) => { - Ok(session_id) - } - Some(update_execution_manager_heartbeat_response::Result::Error(error)) => { - Err(error.into()) - } - None => Err(LivenessResponseError::Transport( - "update execution manager heartbeat response missing result".to_owned(), - )), - } +) -> SessionId { + response.session_id } /// Converts a displayable transport-layer error into [`LivenessResponseError::Transport`]. @@ -165,12 +151,10 @@ mod tests { const EM_ID: ExecutionManagerId = ExecutionManagerId::from(5); let response = storage::RegisterExecutionManagerResponse { - result: Some(register_execution_manager_response::Result::Registration( - storage::ExecutionManagerRegistration { - execution_manager_id: EM_ID.get(), - session_id: SESSION_ID, - }, - )), + registration: Some(storage::ExecutionManagerRegistration { + execution_manager_id: EM_ID.get(), + session_id: SESSION_ID, + }), }; let registration = register_response_to_result(response) @@ -185,36 +169,79 @@ mod tests { ); } + #[test] + fn register_response_to_result_rejects_missing_registration() { + let response = storage::RegisterExecutionManagerResponse { registration: None }; + + assert!(matches!( + register_response_to_result(response), + Err(LivenessResponseError::Transport(_)) + )); + } + + #[test] + fn register_response_to_result_accepts_zero_session_id() { + let response = storage::RegisterExecutionManagerResponse { + registration: Some(storage::ExecutionManagerRegistration { + execution_manager_id: 5, + session_id: 0, + }), + }; + + let registration = register_response_to_result(response) + .expect("registration response conversion should succeed"); + + assert_eq!(registration.session_id, 0); + } + #[test] fn heartbeat_response_to_result_returns_session_id() { const SESSION_ID: SessionId = 9; let response = storage::UpdateExecutionManagerHeartbeatResponse { - result: Some( - update_execution_manager_heartbeat_response::Result::SessionId(SESSION_ID), - ), + session_id: SESSION_ID, }; - let session_id = heartbeat_response_to_result(response) - .expect("heartbeat response conversion should succeed"); + let session_id = heartbeat_response_to_result(response); assert_eq!(session_id, SESSION_ID); } #[test] - fn liveness_storage_error_maps_invalid_input_to_illegal_id() { - const ERROR_MSG: &str = "bad em id"; + fn heartbeat_response_to_result_accepts_zero_session_id() { + let response = storage::UpdateExecutionManagerHeartbeatResponse { session_id: 0 }; - let error = storage::ExecutionManagerLivenessError { - err_code: execution_manager_liveness_error::ErrCode::InvalidInput.into(), - message: ERROR_MSG.to_owned(), - }; + assert_eq!(heartbeat_response_to_result(response), 0); + } - match LivenessResponseError::from(error) { - LivenessResponseError::IllegalId(message) => { - assert_eq!(message, ERROR_MSG); - } - error => panic!("unexpected liveness response error: {error:?}"), + #[test] + fn status_maps_failed_precondition_to_marked_dead() { + let status = tonic::Status::failed_precondition("already dead"); + + assert!(matches!( + status_to_error(&status), + LivenessResponseError::MarkedDead + )); + } + + #[test] + fn status_maps_invalid_argument_to_illegal_id() { + const ERROR_MSG: &str = "bad em id"; + let status = tonic::Status::invalid_argument(ERROR_MSG); + + match status_to_error(&status) { + LivenessResponseError::IllegalId(message) => assert_eq!(message, ERROR_MSG), + error => panic!("unexpected liveness status mapping: {error:?}"), } } + + #[test] + fn status_maps_other_codes_to_transport() { + let status = tonic::Status::internal("boom"); + + assert!(matches!( + status_to_error(&status), + LivenessResponseError::Transport(_) + )); + } } diff --git a/components/spider-proto-rust/src/generated/storage.rs b/components/spider-proto-rust/src/generated/storage.rs index 23831481f..b4d4fcc5e 100644 --- a/components/spider-proto-rust/src/generated/storage.rs +++ b/components/spider-proto-rust/src/generated/storage.rs @@ -42,18 +42,8 @@ pub struct PollReadyTasksRequest { } #[derive(Clone, PartialEq, ::prost::Message)] pub struct PollReadyTasksResponse { - #[prost(oneof = "poll_ready_tasks_response::Result", tags = "1, 2")] - pub result: ::core::option::Option, -} -/// Nested message and enum types in `PollReadyTasksResponse`. -pub mod poll_ready_tasks_response { - #[derive(Clone, PartialEq, ::prost::Oneof)] - pub enum Result { - #[prost(message, tag = "1")] - Tasks(super::ReadyTasks), - #[prost(message, tag = "2")] - Error(super::InboundQueueResponseError), - } + #[prost(message, optional, tag = "1")] + pub tasks: ::core::option::Option, } #[derive(Clone, PartialEq, ::prost::Message)] pub struct ReadyTasks { @@ -148,23 +138,11 @@ pub struct AddResourceGroupRequest { pub external_resource_group_id: ::prost::alloc::string::String, #[prost(bytes = "vec", tag = "2")] pub password: ::prost::alloc::vec::Vec, - #[prost(uint64, tag = "3")] - pub session_id: u64, } -#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] +#[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)] pub struct ResourceGroupIdResponse { - #[prost(oneof = "resource_group_id_response::Result", tags = "1, 2")] - pub result: ::core::option::Option, -} -/// Nested message and enum types in `ResourceGroupIdResponse`. -pub mod resource_group_id_response { - #[derive(Clone, PartialEq, Eq, Hash, ::prost::Oneof)] - pub enum Result { - #[prost(uint64, tag = "1")] - ResourceGroupId(u64), - #[prost(message, tag = "2")] - Error(super::ResourceGroupManagementError), - } + #[prost(uint64, tag = "1")] + pub resource_group_id: u64, } #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] pub struct VerifyResourceGroupRequest { @@ -172,8 +150,6 @@ pub struct VerifyResourceGroupRequest { pub resource_group_id: u64, #[prost(bytes = "vec", tag = "2")] pub password: ::prost::alloc::vec::Vec, - #[prost(uint64, tag = "3")] - pub session_id: u64, } #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] pub struct RegisterExecutionManagerRequest { @@ -192,40 +168,15 @@ pub struct ExecutionManagerRegistration { #[prost(uint64, tag = "2")] pub session_id: u64, } -#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] +#[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)] pub struct RegisterExecutionManagerResponse { - #[prost(oneof = "register_execution_manager_response::Result", tags = "1, 2")] - pub result: ::core::option::Option, -} -/// Nested message and enum types in `RegisterExecutionManagerResponse`. -pub mod register_execution_manager_response { - #[derive(Clone, PartialEq, Eq, Hash, ::prost::Oneof)] - pub enum Result { - #[prost(message, tag = "1")] - Registration(super::ExecutionManagerRegistration), - #[prost(message, tag = "2")] - Error(super::ExecutionManagerLivenessError), - } + #[prost(message, optional, tag = "1")] + pub registration: ::core::option::Option, } -#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] +#[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)] pub struct UpdateExecutionManagerHeartbeatResponse { - #[prost( - oneof = "update_execution_manager_heartbeat_response::Result", - tags = "1, 2" - )] - pub result: ::core::option::Option< - update_execution_manager_heartbeat_response::Result, - >, -} -/// Nested message and enum types in `UpdateExecutionManagerHeartbeatResponse`. -pub mod update_execution_manager_heartbeat_response { - #[derive(Clone, PartialEq, Eq, Hash, ::prost::Oneof)] - pub enum Result { - #[prost(uint64, tag = "1")] - SessionId(u64), - #[prost(message, tag = "2")] - Error(super::ExecutionManagerLivenessError), - } + #[prost(uint64, tag = "1")] + pub session_id: u64, } #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] pub struct RegisterSchedulerRequest { @@ -241,20 +192,10 @@ pub struct SchedulerRegistration { #[prost(uint64, tag = "2")] pub session_id: u64, } -#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] +#[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)] pub struct RegisterSchedulerResponse { - #[prost(oneof = "register_scheduler_response::Result", tags = "1, 2")] - pub result: ::core::option::Option, -} -/// Nested message and enum types in `RegisterSchedulerResponse`. -pub mod register_scheduler_response { - #[derive(Clone, PartialEq, Eq, Hash, ::prost::Oneof)] - pub enum Result { - #[prost(message, tag = "1")] - Registration(super::SchedulerRegistration), - #[prost(message, tag = "2")] - Error(super::SchedulerRegistrationError), - } + #[prost(message, optional, tag = "1")] + pub registration: ::core::option::Option, } #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] pub struct Scheduler { @@ -272,245 +213,14 @@ pub struct SchedulerRegistrations { } #[derive(Clone, PartialEq, ::prost::Message)] pub struct GetSchedulersResponse { - #[prost(oneof = "get_schedulers_response::Result", tags = "1, 2")] - pub result: ::core::option::Option, -} -/// Nested message and enum types in `GetSchedulersResponse`. -pub mod get_schedulers_response { - #[derive(Clone, PartialEq, ::prost::Oneof)] - pub enum Result { - #[prost(message, tag = "1")] - Schedulers(super::SchedulerRegistrations), - #[prost(message, tag = "2")] - Error(super::SchedulerRegistrationError), - } + #[prost(message, optional, tag = "1")] + pub schedulers: ::core::option::Option, } #[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)] pub struct GetSessionResponse { #[prost(uint64, tag = "1")] pub session_id: u64, } -#[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)] -pub struct TaskInstanceOperationResponse {} -#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] -pub struct ResourceGroupOperationResponse { - #[prost(oneof = "resource_group_operation_response::Result", tags = "1, 2")] - pub result: ::core::option::Option, -} -/// Nested message and enum types in `ResourceGroupOperationResponse`. -pub mod resource_group_operation_response { - #[derive(Clone, PartialEq, Eq, Hash, ::prost::Oneof)] - pub enum Result { - #[prost(message, tag = "1")] - Ok(super::super::common::Void), - #[prost(message, tag = "2")] - Error(super::ResourceGroupManagementError), - } -} -#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] -pub struct InboundQueueResponseError { - #[prost(enumeration = "inbound_queue_response_error::ErrCode", tag = "1")] - pub err_code: i32, - #[prost(string, tag = "2")] - pub message: ::prost::alloc::string::String, -} -/// Nested message and enum types in `InboundQueueResponseError`. -pub mod inbound_queue_response_error { - #[derive( - Clone, - Copy, - Debug, - PartialEq, - Eq, - Hash, - PartialOrd, - Ord, - ::prost::Enumeration - )] - #[repr(i32)] - pub enum ErrCode { - Unspecified = 0, - InboundClosed = 1, - Server = 2, - InvalidInput = 3, - } - impl ErrCode { - /// String value of the enum field names used in the ProtoBuf definition. - /// - /// The values are not transformed in any way and thus are considered stable - /// (if the ProtoBuf definition does not change) and safe for programmatic use. - pub fn as_str_name(&self) -> &'static str { - match self { - Self::Unspecified => "ERR_CODE_UNSPECIFIED", - Self::InboundClosed => "INBOUND_CLOSED", - Self::Server => "SERVER", - Self::InvalidInput => "INVALID_INPUT", - } - } - /// Creates an enum from field names used in the ProtoBuf definition. - pub fn from_str_name(value: &str) -> ::core::option::Option { - match value { - "ERR_CODE_UNSPECIFIED" => Some(Self::Unspecified), - "INBOUND_CLOSED" => Some(Self::InboundClosed), - "SERVER" => Some(Self::Server), - "INVALID_INPUT" => Some(Self::InvalidInput), - _ => None, - } - } - } -} -#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] -pub struct ResourceGroupManagementError { - #[prost(enumeration = "resource_group_management_error::ErrCode", tag = "1")] - pub err_code: i32, - #[prost(string, tag = "2")] - pub message: ::prost::alloc::string::String, - #[prost(uint64, tag = "3")] - pub storage_session: u64, -} -/// Nested message and enum types in `ResourceGroupManagementError`. -pub mod resource_group_management_error { - #[derive( - Clone, - Copy, - Debug, - PartialEq, - Eq, - Hash, - PartialOrd, - Ord, - ::prost::Enumeration - )] - #[repr(i32)] - pub enum ErrCode { - Unspecified = 0, - StaleSession = 1, - Server = 2, - InvalidInput = 3, - } - impl ErrCode { - /// String value of the enum field names used in the ProtoBuf definition. - /// - /// The values are not transformed in any way and thus are considered stable - /// (if the ProtoBuf definition does not change) and safe for programmatic use. - pub fn as_str_name(&self) -> &'static str { - match self { - Self::Unspecified => "ERR_CODE_UNSPECIFIED", - Self::StaleSession => "STALE_SESSION", - Self::Server => "SERVER", - Self::InvalidInput => "INVALID_INPUT", - } - } - /// Creates an enum from field names used in the ProtoBuf definition. - pub fn from_str_name(value: &str) -> ::core::option::Option { - match value { - "ERR_CODE_UNSPECIFIED" => Some(Self::Unspecified), - "STALE_SESSION" => Some(Self::StaleSession), - "SERVER" => Some(Self::Server), - "INVALID_INPUT" => Some(Self::InvalidInput), - _ => None, - } - } - } -} -#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] -pub struct ExecutionManagerLivenessError { - #[prost(enumeration = "execution_manager_liveness_error::ErrCode", tag = "1")] - pub err_code: i32, - #[prost(string, tag = "2")] - pub message: ::prost::alloc::string::String, -} -/// Nested message and enum types in `ExecutionManagerLivenessError`. -pub mod execution_manager_liveness_error { - #[derive( - Clone, - Copy, - Debug, - PartialEq, - Eq, - Hash, - PartialOrd, - Ord, - ::prost::Enumeration - )] - #[repr(i32)] - pub enum ErrCode { - Unspecified = 0, - MarkedDead = 1, - InvalidInput = 2, - Server = 3, - } - impl ErrCode { - /// String value of the enum field names used in the ProtoBuf definition. - /// - /// The values are not transformed in any way and thus are considered stable - /// (if the ProtoBuf definition does not change) and safe for programmatic use. - pub fn as_str_name(&self) -> &'static str { - match self { - Self::Unspecified => "ERR_CODE_UNSPECIFIED", - Self::MarkedDead => "MARKED_DEAD", - Self::InvalidInput => "INVALID_INPUT", - Self::Server => "SERVER", - } - } - /// Creates an enum from field names used in the ProtoBuf definition. - pub fn from_str_name(value: &str) -> ::core::option::Option { - match value { - "ERR_CODE_UNSPECIFIED" => Some(Self::Unspecified), - "MARKED_DEAD" => Some(Self::MarkedDead), - "INVALID_INPUT" => Some(Self::InvalidInput), - "SERVER" => Some(Self::Server), - _ => None, - } - } - } -} -#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] -pub struct SchedulerRegistrationError { - #[prost(enumeration = "scheduler_registration_error::ErrCode", tag = "1")] - pub err_code: i32, - #[prost(string, tag = "2")] - pub message: ::prost::alloc::string::String, -} -/// Nested message and enum types in `SchedulerRegistrationError`. -pub mod scheduler_registration_error { - #[derive( - Clone, - Copy, - Debug, - PartialEq, - Eq, - Hash, - PartialOrd, - Ord, - ::prost::Enumeration - )] - #[repr(i32)] - pub enum ErrCode { - Unspecified = 0, - Server = 1, - } - impl ErrCode { - /// String value of the enum field names used in the ProtoBuf definition. - /// - /// The values are not transformed in any way and thus are considered stable - /// (if the ProtoBuf definition does not change) and safe for programmatic use. - pub fn as_str_name(&self) -> &'static str { - match self { - Self::Unspecified => "ERR_CODE_UNSPECIFIED", - Self::Server => "SERVER", - } - } - /// Creates an enum from field names used in the ProtoBuf definition. - pub fn from_str_name(value: &str) -> ::core::option::Option { - match value { - "ERR_CODE_UNSPECIFIED" => Some(Self::Unspecified), - "SERVER" => Some(Self::Server), - _ => None, - } - } - } -} #[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, ::prost::Enumeration)] #[repr(i32)] pub enum JobState { @@ -1389,7 +1099,7 @@ pub mod task_instance_management_service_client { &mut self, request: impl tonic::IntoRequest, ) -> std::result::Result< - tonic::Response, + tonic::Response, tonic::Status, > { self.inner @@ -1418,7 +1128,7 @@ pub mod task_instance_management_service_client { &mut self, request: impl tonic::IntoRequest, ) -> std::result::Result< - tonic::Response, + tonic::Response, tonic::Status, > { self.inner @@ -1469,14 +1179,14 @@ pub mod task_instance_management_service_server { &self, request: tonic::Request, ) -> std::result::Result< - tonic::Response, + tonic::Response, tonic::Status, >; async fn report_task_failure( &self, request: tonic::Request, ) -> std::result::Result< - tonic::Response, + tonic::Response, tonic::Status, >; } @@ -1617,7 +1327,7 @@ pub mod task_instance_management_service_server { T: TaskInstanceManagementService, > tonic::server::UnaryService for ReportTaskSuccessSvc { - type Response = super::TaskInstanceOperationResponse; + type Response = super::super::common::Void; type Future = BoxFuture< tonic::Response, tonic::Status, @@ -1668,7 +1378,7 @@ pub mod task_instance_management_service_server { T: TaskInstanceManagementService, > tonic::server::UnaryService for ReportTaskFailureSvc { - type Response = super::TaskInstanceOperationResponse; + type Response = super::super::common::Void; type Future = BoxFuture< tonic::Response, tonic::Status, @@ -2351,7 +2061,7 @@ pub mod resource_group_management_service_client { &mut self, request: impl tonic::IntoRequest, ) -> std::result::Result< - tonic::Response, + tonic::Response, tonic::Status, > { self.inner @@ -2402,7 +2112,7 @@ pub mod resource_group_management_service_server { &self, request: tonic::Request, ) -> std::result::Result< - tonic::Response, + tonic::Response, tonic::Status, >; } @@ -2543,7 +2253,7 @@ pub mod resource_group_management_service_server { T: ResourceGroupManagementService, > tonic::server::UnaryService for VerifyResourceGroupSvc { - type Response = super::ResourceGroupOperationResponse; + type Response = super::super::common::Void; type Future = BoxFuture< tonic::Response, tonic::Status, diff --git a/components/spider-proto/storage/storage.proto b/components/spider-proto/storage/storage.proto index 9cf35f802..11caf6e3f 100644 --- a/components/spider-proto/storage/storage.proto +++ b/components/spider-proto/storage/storage.proto @@ -15,8 +15,8 @@ service JobOrchestrationService { service TaskInstanceManagementService { rpc RegisterTaskInstance(RegisterTaskInstanceRequest) returns (RegisterTaskInstanceResponse); - rpc ReportTaskSuccess(ReportTaskSuccessRequest) returns (TaskInstanceOperationResponse); - rpc ReportTaskFailure(ReportTaskFailureRequest) returns (TaskInstanceOperationResponse); + rpc ReportTaskSuccess(ReportTaskSuccessRequest) returns (common.Void); + rpc ReportTaskFailure(ReportTaskFailureRequest) returns (common.Void); } service InboundQueueService { @@ -27,7 +27,7 @@ service InboundQueueService { service ResourceGroupManagementService { rpc AddResourceGroup(AddResourceGroupRequest) returns (ResourceGroupIdResponse); - rpc VerifyResourceGroup(VerifyResourceGroupRequest) returns (ResourceGroupOperationResponse); + rpc VerifyResourceGroup(VerifyResourceGroupRequest) returns (common.Void); } service ExecutionManagerLivenessService { @@ -78,10 +78,7 @@ message PollReadyTasksRequest { } message PollReadyTasksResponse { - oneof result { - ReadyTasks tasks = 1; - InboundQueueResponseError error = 2; - } + ReadyTasks tasks = 1; } message ReadyTasks { @@ -144,20 +141,15 @@ message ReportTaskFailureRequest { message AddResourceGroupRequest { string external_resource_group_id = 1; bytes password = 2; - uint64 session_id = 3; } message ResourceGroupIdResponse { - oneof result { - uint64 resource_group_id = 1; - ResourceGroupManagementError error = 2; - } + uint64 resource_group_id = 1; } message VerifyResourceGroupRequest { uint64 resource_group_id = 1; bytes password = 2; - uint64 session_id = 3; } message RegisterExecutionManagerRequest { @@ -174,17 +166,11 @@ message ExecutionManagerRegistration { } message RegisterExecutionManagerResponse { - oneof result { - ExecutionManagerRegistration registration = 1; - ExecutionManagerLivenessError error = 2; - } + ExecutionManagerRegistration registration = 1; } message UpdateExecutionManagerHeartbeatResponse { - oneof result { - uint64 session_id = 1; - ExecutionManagerLivenessError error = 2; - } + uint64 session_id = 1; } message RegisterSchedulerRequest { @@ -198,10 +184,7 @@ message SchedulerRegistration { } message RegisterSchedulerResponse { - oneof result { - SchedulerRegistration registration = 1; - SchedulerRegistrationError error = 2; - } + SchedulerRegistration registration = 1; } message Scheduler { @@ -215,10 +198,7 @@ message SchedulerRegistrations { } message GetSchedulersResponse { - oneof result { - SchedulerRegistrations schedulers = 1; - SchedulerRegistrationError error = 2; - } + SchedulerRegistrations schedulers = 1; } message GetSessionResponse { @@ -235,59 +215,3 @@ enum JobState { FAILED = 6; CANCELLED = 7; } - -message TaskInstanceOperationResponse {} - -message ResourceGroupOperationResponse { - oneof result { - common.Void ok = 1; - ResourceGroupManagementError error = 2; - } -} - -message InboundQueueResponseError { - enum ErrCode { - ERR_CODE_UNSPECIFIED = 0; - INBOUND_CLOSED = 1; - SERVER = 2; - INVALID_INPUT = 3; - } - - ErrCode err_code = 1; - string message = 2; -} - -message ResourceGroupManagementError { - enum ErrCode { - ERR_CODE_UNSPECIFIED = 0; - STALE_SESSION = 1; - SERVER = 2; - INVALID_INPUT = 3; - } - - ErrCode err_code = 1; - string message = 2; - uint64 storage_session = 3; -} - -message ExecutionManagerLivenessError { - enum ErrCode { - ERR_CODE_UNSPECIFIED = 0; - MARKED_DEAD = 1; - INVALID_INPUT = 2; - SERVER = 3; - } - - ErrCode err_code = 1; - string message = 2; -} - -message SchedulerRegistrationError { - enum ErrCode { - ERR_CODE_UNSPECIFIED = 0; - SERVER = 1; - } - - ErrCode err_code = 1; - string message = 2; -} \ No newline at end of file diff --git a/components/spider-scheduler/src/error.rs b/components/spider-scheduler/src/error.rs index df7c52789..a05a85494 100644 --- a/components/spider-scheduler/src/error.rs +++ b/components/spider-scheduler/src/error.rs @@ -5,10 +5,6 @@ use spider_core::types::id::{JobId, SessionId}; /// Errors returned by [`crate::storage_client::SchedulerStorageClient`] operations. #[derive(Debug, thiserror::Error)] pub enum StorageClientError { - /// The inbound queue is closed and can no longer yield ready entries. - #[error("inbound queue is closed")] - InboundClosed, - /// No job with the requested identifier exists. #[error("job not found: {0:?}")] JobNotFound(JobId), diff --git a/components/spider-scheduler/src/storage_client/grpc.rs b/components/spider-scheduler/src/storage_client/grpc.rs index 4e575bca3..2eb9c68e2 100644 --- a/components/spider-scheduler/src/storage_client/grpc.rs +++ b/components/spider-scheduler/src/storage_client/grpc.rs @@ -9,14 +9,13 @@ use spider_core::{ }; use spider_proto_rust::storage::{ self, - inbound_queue_response_error, inbound_queue_service_client::InboundQueueServiceClient, job_orchestration_service_client::JobOrchestrationServiceClient, - poll_ready_tasks_response, }; use spider_utils::grpc::client::ConnectionPool; use tonic::{ Code, + Status, transport::{Channel, Endpoint}, }; @@ -82,7 +81,7 @@ impl SchedulerStorageClient for GrpcSchedulerStorageClient { .get_client() .poll_ready_tasks(request) .await - .map_err(to_transport_error)? + .map_err(|status| inbound_status_to_error(&status))? .into_inner(); poll_ready_tasks_response_to_result(response) } @@ -98,7 +97,7 @@ impl SchedulerStorageClient for GrpcSchedulerStorageClient { .get_client() .poll_ready_commit_tasks(request) .await - .map_err(to_transport_error)? + .map_err(|status| inbound_status_to_error(&status))? .into_inner(); poll_ready_tasks_response_to_result(response) } @@ -114,7 +113,7 @@ impl SchedulerStorageClient for GrpcSchedulerStorageClient { .get_client() .poll_ready_cleanup_tasks(request) .await - .map_err(to_transport_error)? + .map_err(|status| inbound_status_to_error(&status))? .into_inner(); poll_ready_tasks_response_to_result(response) } @@ -137,19 +136,20 @@ impl SchedulerStorageClient for GrpcSchedulerStorageClient { } } -impl From for StorageClientError { - fn from(error: storage::InboundQueueResponseError) -> Self { - match inbound_queue_response_error::ErrCode::try_from(error.err_code) { - Ok(inbound_queue_response_error::ErrCode::InboundClosed) => Self::InboundClosed, - Ok(inbound_queue_response_error::ErrCode::InvalidInput) => { - Self::InvalidInput(error.message) - } - Ok( - inbound_queue_response_error::ErrCode::Server - | inbound_queue_response_error::ErrCode::Unspecified, - ) => Self::Server(error.message), - Err(error) => Self::Transport(format!("unknown scheduler storage error kind: {error}")), - } +/// Maps an inbound-queue gRPC [`Status`] to a [`StorageClientError`]. +/// +/// # Returns +/// +/// The [`StorageClientError`] for `status`'s code: +/// +/// * [`StorageClientError::Transport`] for `UNAVAILABLE` (a lost or unestablished connection). +/// * [`StorageClientError::InvalidInput`] for `INVALID_ARGUMENT`. +/// * [`StorageClientError::Server`] for any other code. +fn inbound_status_to_error(status: &Status) -> StorageClientError { + match status.code() { + Code::Unavailable => to_transport_error(status.message()), + Code::InvalidArgument => to_invalid_input_error(status.message()), + _ => StorageClientError::Server(status.message().to_owned()), } } @@ -179,13 +179,12 @@ fn poll_ready_tasks_request( fn poll_ready_tasks_response_to_result( response: storage::PollReadyTasksResponse, ) -> Result<(SessionId, Vec), StorageClientError> { - match response.result { - Some(poll_ready_tasks_response::Result::Tasks(tasks)) => ready_tasks_to_result(tasks), - Some(poll_ready_tasks_response::Result::Error(error)) => Err(error.into()), - None => Err(StorageClientError::Transport( - "poll ready tasks response missing `result` message".to_owned(), - )), - } + let tasks = response.tasks.ok_or_else(|| { + StorageClientError::Transport( + "poll ready tasks response missing `tasks` message".to_owned(), + ) + })?; + ready_tasks_to_result(tasks) } /// # Returns @@ -269,10 +268,7 @@ fn to_invalid_input_error(error: impl std::fmt::Display) -> StorageClientError { #[cfg(test)] mod tests { use spider_core::types::id::{JobId, ResourceGroupId, TaskId}; - use spider_proto_rust::{ - common, - storage::{self, inbound_queue_response_error, poll_ready_tasks_response}, - }; + use spider_proto_rust::{common, storage}; use super::*; @@ -284,16 +280,14 @@ mod tests { #[test] fn poll_ready_tasks_response_converts_entries() { let response = storage::PollReadyTasksResponse { - result: Some(poll_ready_tasks_response::Result::Tasks( - storage::ReadyTasks { - session_id: SESSION_ID, - tasks: vec![storage::ReadyTask { - resource_group_id: RESOURCE_GROUP_ID, - job_id: JOB_ID, - task_id: Some(common::TaskId::from(TaskId::Index(TASK_INDEX))), - }], - }, - )), + tasks: Some(storage::ReadyTasks { + session_id: SESSION_ID, + tasks: vec![storage::ReadyTask { + resource_group_id: RESOURCE_GROUP_ID, + job_id: JOB_ID, + task_id: Some(common::TaskId::from(TaskId::Index(TASK_INDEX))), + }], + }), }; let (session_id, entries) = poll_ready_tasks_response_to_result(response) @@ -315,16 +309,14 @@ mod tests { const MISSING_TASK_ID_MESSAGE: &str = "missing task ID"; let response = storage::PollReadyTasksResponse { - result: Some(poll_ready_tasks_response::Result::Tasks( - storage::ReadyTasks { - session_id: SESSION_ID, - tasks: vec![storage::ReadyTask { - resource_group_id: RESOURCE_GROUP_ID, - job_id: JOB_ID, - task_id: None, - }], - }, - )), + tasks: Some(storage::ReadyTasks { + session_id: SESSION_ID, + tasks: vec![storage::ReadyTask { + resource_group_id: RESOURCE_GROUP_ID, + job_id: JOB_ID, + task_id: None, + }], + }), }; match poll_ready_tasks_response_to_result(response) { @@ -336,17 +328,44 @@ mod tests { } #[test] - fn inbound_queue_response_error_maps_inbound_closed() { - const ERROR_MESSAGE: &str = "closed"; + fn poll_ready_tasks_response_rejects_missing_tasks() { + let response = storage::PollReadyTasksResponse { tasks: None }; - let error = storage::InboundQueueResponseError { - err_code: inbound_queue_response_error::ErrCode::InboundClosed.into(), - message: ERROR_MESSAGE.to_owned(), - }; + assert!(matches!( + poll_ready_tasks_response_to_result(response), + Err(StorageClientError::Transport(_)) + )); + } + + #[test] + fn inbound_status_maps_unavailable_to_transport() { + const MESSAGE: &str = "inbound queue is closed"; + let status = tonic::Status::unavailable(MESSAGE); + + match inbound_status_to_error(&status) { + StorageClientError::Transport(message) => assert_eq!(message, MESSAGE), + error => panic!("unexpected inbound status mapping: {error:?}"), + } + } + + #[test] + fn inbound_status_maps_invalid_argument_to_invalid_input() { + const MESSAGE: &str = "bad max_items"; + let status = tonic::Status::invalid_argument(MESSAGE); + + match inbound_status_to_error(&status) { + StorageClientError::InvalidInput(message) => assert_eq!(message, MESSAGE), + error => panic!("unexpected inbound status mapping: {error:?}"), + } + } + + #[test] + fn inbound_status_maps_other_codes_to_server() { + let status = tonic::Status::internal("boom"); assert!(matches!( - StorageClientError::from(error), - StorageClientError::InboundClosed + inbound_status_to_error(&status), + StorageClientError::Server(_) )); } } diff --git a/components/spider-scheduler/src/storage_client/mod.rs b/components/spider-scheduler/src/storage_client/mod.rs index 808048e0f..e805dad82 100644 --- a/components/spider-scheduler/src/storage_client/mod.rs +++ b/components/spider-scheduler/src/storage_client/mod.rs @@ -39,8 +39,9 @@ pub trait SchedulerStorageClient: Send + Sync + Clone { /// /// Returns an error if: /// - /// * [`StorageClientError::InboundClosed`] if the regular-task lane is closed and can no longer - /// yield entries. + /// * [`StorageClientError::Server`] if the storage service returns an error. + /// * [`StorageClientError::Transport`] if the storage transport fails or returns malformed + /// data. async fn poll_ready( &self, max_items: usize, @@ -65,8 +66,9 @@ pub trait SchedulerStorageClient: Send + Sync + Clone { /// /// Returns an error if: /// - /// * [`StorageClientError::InboundClosed`] if the commit-task lane is closed and can no longer - /// yield entries. + /// * [`StorageClientError::Server`] if the storage service returns an error. + /// * [`StorageClientError::Transport`] if the storage transport fails or returns malformed + /// data. async fn poll_commit_ready( &self, max_items: usize, @@ -91,8 +93,9 @@ pub trait SchedulerStorageClient: Send + Sync + Clone { /// /// Returns an error if: /// - /// * [`StorageClientError::InboundClosed`] if the cleanup-task lane is closed and can no longer - /// yield entries. + /// * [`StorageClientError::Server`] if the storage service returns an error. + /// * [`StorageClientError::Transport`] if the storage transport fails or returns malformed + /// data. async fn poll_cleanup_ready( &self, max_items: usize, diff --git a/components/spider-storage/src/grpc.rs b/components/spider-storage/src/grpc.rs index d1cc012b4..ea0a145ed 100644 --- a/components/spider-storage/src/grpc.rs +++ b/components/spider-storage/src/grpc.rs @@ -419,7 +419,7 @@ impl< async fn report_task_success( &self, request: Request, - ) -> Result, Status> { + ) -> Result, Status> { let (session_id, job_id, task_id, task_instance_id, serialized_outputs) = request.into_inner().unpack()?; tracing::info!( @@ -457,13 +457,13 @@ impl< self.task_instance_management_service_error_handler(error, "report_task_success") })?; - Ok(Response::new(storage::TaskInstanceOperationResponse {})) + Ok(Response::new(common::Void {})) } async fn report_task_failure( &self, request: Request, - ) -> Result, Status> { + ) -> Result, Status> { let (session_id, job_id, task_id, task_instance_id, error_message) = request.into_inner().unpack()?; tracing::info!( @@ -481,7 +481,7 @@ impl< .map_err(|error| { self.task_instance_management_service_error_handler(error, "report_task_failure") })?; - Ok(Response::new(storage::TaskInstanceOperationResponse {})) + Ok(Response::new(common::Void {})) } } @@ -533,7 +533,7 @@ impl< async fn verify_resource_group( &self, _request: Request, - ) -> Result, Status> { + ) -> Result, Status> { todo!("Not implemented") } }