refactor(spider-core)!: Unify external resource group credentials (fixes #468).

components/spider-client/src/client.rs

@@ -9,6 +9,7 @@ use spider_core::types::id::JobId;
 use spider_core::types::id::ResourceGroupId;
 use spider_core::types::io::TaskInput;
 use spider_core::types::io::TaskOutput;
+use spider_core::types::resource_group::ExternalResourceGroupCredentials;
 use spider_utils::grpc::retry::RetryConfig;
 use tonic::transport::Endpoint;
 
@@ -181,12 +182,9 @@ impl SpiderClient {
     /// * [`ClientError::Server`] for any other server-reported error.
     pub async fn add_resource_group(
         &self,
-        external_resource_group_id: String,
-        password: Vec<u8>,
+        credentials: ExternalResourceGroupCredentials,
     ) -> Result<ResourceGroupId, ClientError> {
-        self.resource_group
-            .add_resource_group(external_resource_group_id, password)
-            .await
+        self.resource_group.add_resource_group(credentials).await
     }
 
     /// Verifies a resource group's password.
@@ -308,6 +306,7 @@ fn assert_client_futures_send(
     resource_group_id: ResourceGroupId,
     job_id: JobId,
     task_graph: &TaskGraph,
+    credentials: ExternalResourceGroupCredentials,
 ) {
     const fn assert_send<FutureType: Send>(_: &FutureType) {}
     assert_send(&client.submit_job(resource_group_id, task_graph, Vec::new()));
@@ -316,6 +315,6 @@ fn assert_client_futures_send(
     assert_send(&client.get_job_state(job_id));
     assert_send(&client.get_job_outputs(job_id));
     assert_send(&client.get_job_error(job_id));
-    assert_send(&client.add_resource_group(String::new(), Vec::new()));
+    assert_send(&client.add_resource_group(credentials));
     assert_send(&client.verify_resource_group(resource_group_id, Vec::new()));
 }

components/spider-client/src/grpc/resource_group.rs

@@ -3,6 +3,7 @@
 use std::num::NonZeroUsize;
 
 use spider_core::types::id::ResourceGroupId;
+use spider_core::types::resource_group::ExternalResourceGroupCredentials;
 use spider_proto_rust::storage::ResourceGroupManagementServiceClient;
 use spider_proto_rust::storage::{self};
 use spider_utils::grpc::client::ConnectionPool;
@@ -66,15 +67,13 @@ impl ResourceGroupManagementClient {
     /// * Forwards [`ResourceGroupManagementServiceClient::add_resource_group`]'s status on failure.
     pub async fn add_resource_group(
         &self,
-        external_resource_group_id: String,
-        password: Vec<u8>,
+        credentials: ExternalResourceGroupCredentials,
     ) -> Result<ResourceGroupId, ClientError> {
         let pool = self.connection_pool.clone();
         let response = call_with_retry(self.retry_config, move || {
             let mut client = pool.get_client();
             let request = storage::AddResourceGroupRequest {
-                external_resource_group_id: external_resource_group_id.clone(),
-                password: password.clone(),
+                credentials: Some(credentials.clone().into()),
             };
             async move { client.add_resource_group(request).await }
         })

components/spider-core/src/types/mod.rs

@@ -1,3 +1,4 @@
 pub mod id;
 pub mod io;
+pub mod resource_group;
 pub mod scheduler;

components/spider-core/src/types/resource_group.rs

@@ -0,0 +1,11 @@
+//! Resource group types shared across Spider components.
+
+/// Credentials identifying and authenticating an external resource group.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct ExternalResourceGroupCredentials {
+    /// The external resource group ID.
+    pub external_resource_group_id: String,
+
+    /// The resource group password.
+    pub password: Vec<u8>,
+}

components/spider-proto-rust/src/generated/storage.rs

@@ -134,10 +134,8 @@ pub struct ReportTaskFailureRequest {
 }
 #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
 pub struct AddResourceGroupRequest {
-    #[prost(string, tag = "1")]
-    pub external_resource_group_id: ::prost::alloc::string::String,
-    #[prost(bytes = "vec", tag = "2")]
-    pub password: ::prost::alloc::vec::Vec<u8>,
+    #[prost(message, optional, tag = "1")]
+    pub credentials: ::core::option::Option<ExternalResourceGroupCredentials>,
 }
 #[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
 pub struct ResourceGroupIdResponse {

components/spider-proto-rust/src/lib.rs

@@ -5,6 +5,7 @@ pub mod error;
 pub mod id;
 pub mod io;
 pub mod job;
+pub mod resource_group;
 pub mod scheduler_registration;
 pub mod unpack;
 

components/spider-proto-rust/src/resource_group.rs

@@ -0,0 +1,48 @@
+//! Conversions between protobuf and Spider core resource group types.
+
+use spider_core::types::resource_group::ExternalResourceGroupCredentials;
+
+use crate::storage;
+
+impl From<ExternalResourceGroupCredentials> for storage::ExternalResourceGroupCredentials {
+    fn from(credentials: ExternalResourceGroupCredentials) -> Self {
+        Self {
+            external_resource_group_id: credentials.external_resource_group_id,
+            password: credentials.password,
+        }
+    }
+}
+
+impl From<storage::ExternalResourceGroupCredentials> for ExternalResourceGroupCredentials {
+    fn from(credentials: storage::ExternalResourceGroupCredentials) -> Self {
+        Self {
+            external_resource_group_id: credentials.external_resource_group_id,
+            password: credentials.password,
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use prost::Message;
+    use spider_core::types::resource_group::ExternalResourceGroupCredentials;
+
+    use crate::storage;
+
+    #[test]
+    fn test_external_resource_group_credentials_protocol_round_trip() {
+        let credentials = ExternalResourceGroupCredentials {
+            external_resource_group_id: "external-resource-group".to_owned(),
+            password: vec![0, 1, 2, 255],
+        };
+
+        let encoded =
+            storage::ExternalResourceGroupCredentials::from(credentials.clone()).encode_to_vec();
+        let decoded = ExternalResourceGroupCredentials::from(
+            storage::ExternalResourceGroupCredentials::decode(encoded.as_slice())
+                .expect("external resource group credentials should decode"),
+        );
+
+        assert_eq!(decoded, credentials);
+    }
+}

components/spider-proto-rust/src/unpack/storage.rs

@@ -9,6 +9,7 @@ use spider_core::types::id::ResourceGroupId;
 use spider_core::types::id::SessionId;
 use spider_core::types::id::TaskId;
 use spider_core::types::id::TaskInstanceId;
+use spider_core::types::resource_group::ExternalResourceGroupCredentials;
 use spider_utils::config::Host;
 use tonic::Code;
 
@@ -142,15 +143,14 @@ impl RequestUnpack for ReportTaskFailureRequest {
     }
 }
 
-/// Unpacks [`AddResourceGroupRequest`] into a tuple containing:
-///
-/// * The external resource group ID.
-/// * The password.
+/// Unpacks [`AddResourceGroupRequest`] into external resource group credentials.
 impl RequestUnpack for AddResourceGroupRequest {
-    type Unpacked = (String, Vec<u8>);
+    type Unpacked = ExternalResourceGroupCredentials;
 
     fn unpack(self) -> Result<Self::Unpacked, UnpackError> {
-        Ok((self.external_resource_group_id, self.password))
+        self.credentials
+            .map(Into::into)
+            .ok_or_else(|| invalid_argument("resource group credentials are missing".to_owned()))
     }
 }
 
@@ -171,16 +171,14 @@ impl RequestUnpack for VerifyResourceGroupRequest {
 /// * The execution manager's IP address.
 /// * The external resource group credentials, if present.
 impl RequestUnpack for RegisterExecutionManagerRequest {
-    type Unpacked = (IpAddr, Option<(String, Vec<u8>)>);
+    type Unpacked = (IpAddr, Option<ExternalResourceGroupCredentials>);
 
     fn unpack(self) -> Result<Self::Unpacked, UnpackError> {
         let ip_address = self
             .ip_address
             .parse::<IpAddr>()
             .map_err(|error| invalid_argument(format!("invalid IP address: {error}")))?;
-        let resource_group_credentials = self
-            .resource_group_credentials
-            .map(|credentials| (credentials.external_resource_group_id, credentials.password));
+        let resource_group_credentials = self.resource_group_credentials.map(Into::into);
         Ok((ip_address, resource_group_credentials))
     }
 }

components/spider-proto/storage/storage.proto

@@ -139,8 +139,7 @@ message ReportTaskFailureRequest {
 }
 
 message AddResourceGroupRequest {
-  string external_resource_group_id = 1;
-  bytes password = 2;
+  ExternalResourceGroupCredentials credentials = 1;
 }
 
 message ResourceGroupIdResponse {

components/spider-storage/src/db/mariadb.rs

@@ -11,6 +11,7 @@ use spider_core::types::id::SchedulerId;
 use spider_core::types::id::SessionId;
 use spider_core::types::io::SerializedTaskOutputs;
 use spider_core::types::io::TaskOutput;
+use spider_core::types::resource_group::ExternalResourceGroupCredentials;
 use spider_core::types::scheduler::RegisteredScheduler;
 use spider_derive::MySqlEnum;
 use spider_utils::config::Host;
@@ -23,7 +24,6 @@ use crate::db::DbError;
 use crate::db::DbStorage;
 use crate::db::ExecutionManagerLivenessManagement;
 use crate::db::ExternalJobOrchestration;
-use crate::db::ExternalResourceGroupCredentials;
 use crate::db::InternalJobOrchestration;
 use crate::db::RecoverableJobContext;
 use crate::db::ResourceGroupManagement;

components/spider-storage/src/db/mod.rs

@@ -7,7 +7,6 @@ pub use mariadb::MariaDbStorageConnector;
 pub use protocol::DbStorage;
 pub use protocol::ExecutionManagerLivenessManagement;
 pub use protocol::ExternalJobOrchestration;
-pub use protocol::ExternalResourceGroupCredentials;
 pub use protocol::InternalJobOrchestration;
 pub use protocol::RecoverableJobContext;
 pub use protocol::ResourceGroupManagement;

components/spider-storage/src/db/protocol.rs

@@ -8,6 +8,7 @@ use spider_core::types::id::ResourceGroupId;
 use spider_core::types::id::SchedulerId;
 use spider_core::types::id::SessionId;
 use spider_core::types::io::TaskOutput;
+use spider_core::types::resource_group::ExternalResourceGroupCredentials;
 use spider_core::types::scheduler::RegisteredScheduler;
 
 use crate::db::error::DbError;
@@ -24,15 +25,6 @@ pub struct RecoverableJobContext {
     pub outputs: Option<Vec<TaskOutput>>,
 }
 
-/// Credentials identifying and authenticating an external resource group.
-pub struct ExternalResourceGroupCredentials {
-    /// The external resource group ID.
-    pub external_resource_group_id: String,
-
-    /// The resource group password.
-    pub password: Vec<u8>,
-}
-
 /// The database storage interface. A database storage must implement the following traits:
 ///
 /// * [`ExternalJobOrchestration`]

components/spider-storage/src/grpc.rs

@@ -22,7 +22,6 @@ use tonic::Status;
 use crate::cache::error::CacheError;
 use crate::db::DbError;
 use crate::db::DbStorage;
-use crate::db::ExternalResourceGroupCredentials;
 use crate::inbound_queue::InboundQueueEntry;
 use crate::inbound_queue::InboundQueueSender;
 use crate::state::ServiceState;
@@ -770,14 +769,14 @@ impl<
         &self,
         request: Request<storage::AddResourceGroupRequest>,
     ) -> Result<Response<storage::ResourceGroupIdResponse>, Status> {
-        let (external_id, password) = request.into_inner().unpack()?;
-        tracing::info!(external_id = % external_id, "Add resource group request received.");
+        let credentials = request.into_inner().unpack()?;
+        tracing::info!(
+            external_id = % credentials.external_resource_group_id,
+            "Add resource group request received."
+        );
         let rg_id = self
             .inner
-            .add_resource_group(ExternalResourceGroupCredentials {
-                external_resource_group_id: external_id,
-                password,
-            })
+            .add_resource_group(credentials)
             .await
             .map_err(|error| {
                 self.resource_group_management_service_error_handler(error, "add_resource_group")
@@ -822,15 +821,7 @@ impl<
         tracing::info!(% ip_address, "Execution manager registration request received.");
         let (em_id, resource_group_id) = self
             .inner
-            .register_execution_manager(
-                ip_address,
-                resource_group_credentials.map(|(external_resource_group_id, password)| {
-                    ExternalResourceGroupCredentials {
-                        external_resource_group_id,
-                        password,
-                    }
-                }),
-            )
+            .register_execution_manager(ip_address, resource_group_credentials)
             .await
             .map_err(|error| {
                 self.execution_manager_liveness_service_error_handler(

components/spider-storage/src/state/service.rs

@@ -14,6 +14,7 @@ use spider_core::types::id::TaskInstanceId;
 use spider_core::types::io::ExecutionContext;
 use spider_core::types::io::TaskOutput;
 use spider_core::types::io::TaskOutputsSerializer;
+use spider_core::types::resource_group::ExternalResourceGroupCredentials;
 use spider_core::types::scheduler::RegisteredScheduler;
 use spider_tdl::error::TdlError;
 use spider_utils::config::Host;
@@ -23,7 +24,6 @@ use crate::cache::error::CacheError;
 use crate::cache::error::InternalError;
 use crate::cache::job::SharedJobControlBlock;
 use crate::db::DbStorage;
-use crate::db::ExternalResourceGroupCredentials;
 use crate::inbound_queue::CleanupTaskMarker;
 use crate::inbound_queue::CommitTaskMarker;
 use crate::inbound_queue::InboundQueueEntry;

components/spider-storage/src/state/test_utils.rs

@@ -12,6 +12,7 @@ use spider_core::types::id::SchedulerId;
 use spider_core::types::id::SessionId;
 use spider_core::types::id::TaskInstanceId;
 use spider_core::types::io::TaskOutput;
+use spider_core::types::resource_group::ExternalResourceGroupCredentials;
 use spider_core::types::scheduler::RegisteredScheduler;
 
 use crate::cache::error::InternalError;
@@ -21,7 +22,6 @@ use crate::db::DbError;
 use crate::db::DbStorage;
 use crate::db::ExecutionManagerLivenessManagement;
 use crate::db::ExternalJobOrchestration;
-use crate::db::ExternalResourceGroupCredentials;
 use crate::db::InternalJobOrchestration;
 use crate::db::RecoverableJobContext;
 use crate::db::ResourceGroupManagement;

components/spider-storage/src/task_instance_pool.rs

@@ -560,11 +560,11 @@ mod tests {
     use spider_core::types::id::JobId;
     use spider_core::types::id::ResourceGroupId;
     use spider_core::types::io::TaskInput;
+    use spider_core::types::resource_group::ExternalResourceGroupCredentials;
     use tokio::sync::Mutex;
 
     use super::*;
     use crate::db::DbError;
-    use crate::db::ExternalResourceGroupCredentials;
     use crate::job_submission::create_validated_submission;
 
     const DEFAULT_CHANNEL_SIZE: usize = 128;

components/spider-storage/tests/mariadb_infra.rs

@@ -1,7 +1,7 @@
 use spider_core::types::id::ResourceGroupId;
+use spider_core::types::resource_group::ExternalResourceGroupCredentials;
 use spider_storage::DatabaseConfig;
 use spider_storage::DatabaseCredentials;
-use spider_storage::db::ExternalResourceGroupCredentials;
 use spider_storage::db::MariaDbStorageConnector;
 use spider_storage::db::ResourceGroupManagement;
 

components/spider-storage/tests/mariadb_test.rs

@@ -8,10 +8,10 @@ use spider_core::types::id::JobId;
 use spider_core::types::id::ResourceGroupId;
 use spider_core::types::id::SchedulerId;
 use spider_core::types::io::TaskInput;
+use spider_core::types::resource_group::ExternalResourceGroupCredentials;
 use spider_storage::db::DbError;
 use spider_storage::db::ExecutionManagerLivenessManagement;
 use spider_storage::db::ExternalJobOrchestration;
-use spider_storage::db::ExternalResourceGroupCredentials;
 use spider_storage::db::InternalJobOrchestration;
 use spider_storage::db::MariaDbStorageConnector;
 use spider_storage::db::ResourceGroupManagement;

components/spider-storage/tests/runtime_recovery_test.rs

@@ -8,10 +8,10 @@ use spider_core::types::id::JobId;
 use spider_core::types::id::TaskInstanceId;
 use spider_core::types::io::TaskOutput;
 use spider_core::types::io::TaskOutputsSerializer;
+use spider_core::types::resource_group::ExternalResourceGroupCredentials;
 use spider_storage::cache::error::CacheError;
 use spider_storage::cache::error::StaleStateError;
 use spider_storage::db::ExternalJobOrchestration;
-use spider_storage::db::ExternalResourceGroupCredentials;
 use spider_storage::inbound_queue::CleanupTaskMarker;
 use spider_storage::inbound_queue::CommitTaskMarker;
 use spider_storage::inbound_queue::InboundQueueConfig;