diff --git a/src/sagittarius/mod.rs b/src/sagittarius/mod.rs index cba9755..93eaf66 100644 --- a/src/sagittarius/mod.rs +++ b/src/sagittarius/mod.rs @@ -4,7 +4,6 @@ pub mod flow_service_client_impl; pub mod module_service_client_impl; pub mod retry; pub mod runtime_status_service_client_impl; -pub mod runtime_usage_client_impl; pub mod test_execution_client_impl; pub(crate) const SAGITTARIUS_UNARY_RPC_TIMEOUT: Duration = Duration::from_secs(5); diff --git a/src/sagittarius/runtime_usage_client_impl.rs b/src/sagittarius/runtime_usage_client_impl.rs deleted file mode 100644 index bafc1f5..0000000 --- a/src/sagittarius/runtime_usage_client_impl.rs +++ /dev/null @@ -1,70 +0,0 @@ -use std::sync::Arc; - -use tokio::sync::Mutex; -use tonic::{Extensions, Request, transport::Channel}; -use tucana::sagittarius::runtime_usage_service_client::RuntimeUsageServiceClient as SagittariusRuntimeUsageServiceClient; - -use crate::authorization::authorization::get_authorization_metadata; -use crate::sagittarius::SAGITTARIUS_UNARY_RPC_TIMEOUT; - -pub struct SagittariusRuntimeUsageClient { - client: SagittariusRuntimeUsageServiceClient, - token: String, -} - -impl SagittariusRuntimeUsageClient { - pub fn new(channel: Channel, token: String) -> Self { - let client = SagittariusRuntimeUsageServiceClient::new(channel); - Self { client, token } - } - - pub fn new_arc(channel: Channel, token: String) -> Arc> { - Arc::new(Mutex::new(Self::new(channel, token))) - } - - pub async fn update_runtime_usage( - &mut self, - runtime_usage_request: tucana::aquila::RuntimeUsageRequest, - ) -> tucana::aquila::RuntimeUsageResponse { - let sample_count = runtime_usage_request.runtime_usage.len(); - log::debug!( - "Forwarding runtime usage samples to Sagittarius sample_count={}", - sample_count - ); - - let mut request = Request::from_parts( - get_authorization_metadata(&self.token), - Extensions::new(), - tucana::sagittarius::RuntimeUsageRequest { - runtime_usage: runtime_usage_request.runtime_usage, - }, - ); - request.set_timeout(SAGITTARIUS_UNARY_RPC_TIMEOUT); - - let response = match self.client.update(request).await { - Ok(response) => { - log::info!("Successfully transferred Runtime Usages.",); - response.into_inner() - } - Err(err) => { - log::error!("Failed to update Runtime Usage: {:?}", err); - return tucana::aquila::RuntimeUsageResponse { success: false }; - } - }; - - match response.success { - true => log::info!( - "Sagittarius successfully updated runtime usage sample_count={}", - sample_count - ), - false => log::warn!( - "Sagittarius did not update runtime usage sample_count={}", - sample_count - ), - }; - - tucana::aquila::RuntimeUsageResponse { - success: response.success, - } - } -} diff --git a/src/server/dynamic_server.rs b/src/server/dynamic_server.rs index 9b66ce3..a319c0e 100644 --- a/src/server/dynamic_server.rs +++ b/src/server/dynamic_server.rs @@ -3,7 +3,6 @@ use crate::{ sagittarius::{ module_service_client_impl::SagittariusModuleServiceClient, runtime_status_service_client_impl::SagittariusRuntimeStatusServiceClient, - runtime_usage_client_impl::SagittariusRuntimeUsageClient, test_execution_client_impl::SagittariusExecutionResponseSender, }, server::{ @@ -11,7 +10,6 @@ use crate::{ create_readiness_interceptor, module_service_server_impl::AquilaModuleServiceServer, runtime_execution_service_server_impl::AquilaExecutionServiceServer, runtime_status_service_server_impl::AquilaRuntimeStatusServiceServer, - runtime_usage_service_server_impl::AquilaRuntimeUsageServiceServer, }, }; use async_nats::jetstream::kv::Store; @@ -23,7 +21,6 @@ use tucana::aquila::{ action_transfer_service_server::ActionTransferServiceServer, execution_service_server::ExecutionServiceServer, module_service_server::ModuleServiceServer, runtime_status_service_server::RuntimeStatusServiceServer, - runtime_usage_service_server::RuntimeUsageServiceServer, }; pub struct AquilaDynamicServer { @@ -97,13 +94,6 @@ impl AquilaDynamicServer { info!("ModuleService started"); - let runtime_usage_service = Arc::new(Mutex::new(SagittariusRuntimeUsageClient::new( - self.channel.clone(), - self.token.clone(), - ))); - - info!("RuntimeUsageService started"); - let runtime_status_service = Arc::new(Mutex::new( SagittariusRuntimeStatusServiceClient::new(self.channel.clone(), self.token.clone()), )); @@ -118,11 +108,7 @@ impl AquilaDynamicServer { module_service.clone(), self.service_configuration.clone(), ); - let runtime_usage_server = AquilaRuntimeUsageServiceServer::new( - runtime_usage_service.clone(), - self.service_configuration.clone(), - ); - let runtime_status_server = AquilaRuntimeStatusServiceServer::new( + let runtime_status_server = AquilaRuntimeStatusServiceServer::new( runtime_status_service.clone(), self.service_configuration.clone(), Duration::from_secs(self.runtime_status_not_responding_after_secs.clone()), @@ -163,10 +149,6 @@ impl AquilaDynamicServer { module_server, intercept.clone(), )) - .add_service(RuntimeUsageServiceServer::with_interceptor( - runtime_usage_server, - intercept.clone(), - )) .add_service(RuntimeStatusServiceServer::with_interceptor( runtime_status_server, intercept.clone(), @@ -187,10 +169,6 @@ impl AquilaDynamicServer { module_server, intercept.clone(), )) - .add_service(RuntimeUsageServiceServer::with_interceptor( - runtime_usage_server, - intercept.clone(), - )) .add_service(RuntimeStatusServiceServer::with_interceptor( runtime_status_server, intercept.clone(), diff --git a/src/server/mod.rs b/src/server/mod.rs index 4c7e544..e98eba2 100644 --- a/src/server/mod.rs +++ b/src/server/mod.rs @@ -3,7 +3,6 @@ mod interceptor; mod module_service_server_impl; mod runtime_execution_service_server_impl; mod runtime_status_service_server_impl; -mod runtime_usage_service_server_impl; pub mod dynamic_server; pub mod static_server; diff --git a/src/server/runtime_usage_service_server_impl.rs b/src/server/runtime_usage_service_server_impl.rs deleted file mode 100644 index 740ba23..0000000 --- a/src/server/runtime_usage_service_server_impl.rs +++ /dev/null @@ -1,63 +0,0 @@ -use std::sync::Arc; - -use tokio::sync::Mutex; -use tonic::Status; -use tucana::aquila::runtime_usage_service_server::RuntimeUsageService; - -use crate::{ - authorization::authorization::extract_token, configuration::service::ServiceConfiguration, - sagittarius::runtime_usage_client_impl::SagittariusRuntimeUsageClient, -}; - -pub struct AquilaRuntimeUsageServiceServer { - client: Arc>, - service_configuration: ServiceConfiguration, -} - -impl AquilaRuntimeUsageServiceServer { - pub fn new( - client: Arc>, - service_configuration: ServiceConfiguration, - ) -> Self { - Self { - client, - service_configuration, - } - } -} - -#[tonic::async_trait] -impl RuntimeUsageService for AquilaRuntimeUsageServiceServer { - async fn update( - &self, - request: tonic::Request, - ) -> Result, tonic::Status> { - let token = match extract_token(&request) { - Ok(t) => t.to_string(), - Err(status) => { - log::warn!("Rejected runtime usage update reason=missing_or_invalid_token"); - return Err(status); - } - }; - - if !self.service_configuration.has_service(&token, &String::from("taurus")) { - log::warn!("Rejected runtime usage update reason=token_not_registered"); - return Err(Status::unauthenticated("token is not valid")); - } - - let runtime_usage_request = request.into_inner(); - - log::debug!("Received RuntimeUsageRequest"); - - let mut client = self.client.lock().await; - let response = client.update_runtime_usage(runtime_usage_request).await; - log::debug!( - "Completed runtime usage update success={}", - response.success - ); - - Ok(tonic::Response::new(tucana::aquila::RuntimeUsageResponse { - success: response.success, - })) - } -}