Merge pull request #364 from code0-tech/#363-sagittarius-grpc-timeout-config

made sagittarius rpc timeout configurable

Author: raphael-goetz

docs/installation.mdx

@@ -90,9 +90,10 @@ Aquila configuration is split into shared variables and mode-specific variables.
 | `GRPC_PORT`           | Port for the Aquila gRPC server.                                                                             | `8081`                          |
 | `WITH_HEALTH_SERVICE` | Enables the gRPC health service when set to `true`.                                                          | `false`                         |
 | `SERVICE_CONFIG_PATH` | Path to the JSON service-configuration file used for runtime/action authorization and default action configs. | `./service.configuration.json` |
-| `RUNTIME_STATUS_NOT_RESPONDING_AFTER_SECS` | Seconds a runtime serivce will be marked as `not_responding` when no hearbeat was recieved. | `15s` |
-| `RUNTIME_STATUS_STOPPED_AFTER_NOT_RESPONDING_SECS` | Seconds a runtime serivce will be marked as `stopped` when no hearbeat was recieved. | `30s` |
-| `RUNTIME_STATUS_MONITOR_INTERVAL_SECS` | Interval in which the heartbeat of the services will be flagged/checked. | `3s` |
+| `RUNTIME_STATUS_NOT_RESPONDING_AFTER_SECS` | Seconds before a runtime service is marked as `not_responding` when no heartbeat was received. | `90s` |
+| `RUNTIME_STATUS_STOPPED_AFTER_NOT_RESPONDING_SECS` | Additional seconds before a non-responding runtime service is marked as `stopped`. | `180s` |
+| `RUNTIME_STATUS_MONITOR_INTERVAL_SECS` | Interval in which runtime service heartbeats are checked. | `30s` |
+| `SAGITTARIUS_UNARY_RPC_TIMEOUT_SECS` | Timeout in seconds for unary RPC calls from Aquila to Sagittarius. | `5s` |
 
 ### Static Mode
 

src/configuration/config.rs

@@ -50,6 +50,9 @@ pub struct Config {
 
     /// Interval in seconds for the runtime status timeout monitor loop.
     pub runtime_status_monitor_interval_secs: u64,
+
+    /// Timeout in seconds for unary RPC calls from Aquila to Sagittarius.
+    pub sagittarius_unary_rpc_timeout_secs: u64,
 }
 
 /// Implementation for all relevant `Aquila` startup configurations
@@ -97,6 +100,10 @@ impl Config {
                 "RUNTIME_STATUS_MONITOR_INTERVAL_SECS",
                 30_u64,
             ),
+            sagittarius_unary_rpc_timeout_secs: env_with_default(
+                "SAGITTARIUS_UNARY_RPC_TIMEOUT_SECS",
+                5_u64,
+            ),
         }
     }
 

src/sagittarius/mod.rs

@@ -1,9 +1,5 @@
-use std::time::Duration;
-
 pub mod flow_service_client_impl;
 pub mod module_service_client_impl;
 pub mod retry;
 pub mod runtime_status_service_client_impl;
 pub mod test_execution_client_impl;
-
-pub(crate) const SAGITTARIUS_UNARY_RPC_TIMEOUT: Duration = Duration::from_secs(5);

src/sagittarius/module_service_client_impl.rs

@@ -1,18 +1,23 @@
 use crate::authorization::authorization::get_authorization_metadata;
-use crate::sagittarius::SAGITTARIUS_UNARY_RPC_TIMEOUT;
+use std::time::Duration;
 use tonic::transport::Channel;
 use tonic::{Extensions, Request};
 
 pub struct SagittariusModuleServiceClient {
     client: tucana::sagittarius::module_service_client::ModuleServiceClient<Channel>,
     token: String,
+    unary_rpc_timeout: Duration,
 }
 
 impl SagittariusModuleServiceClient {
-    pub fn new(channel: Channel, token: String) -> Self {
+    pub fn new(channel: Channel, token: String, unary_rpc_timeout: Duration) -> Self {
         let client = tucana::sagittarius::module_service_client::ModuleServiceClient::new(channel);
 
-        Self { client, token }
+        Self {
+            client,
+            token,
+            unary_rpc_timeout,
+        }
     }
 
     pub async fn update_modules(
@@ -32,7 +37,7 @@ impl SagittariusModuleServiceClient {
                 modules: modules_update_request.modules,
             },
         );
-        request.set_timeout(SAGITTARIUS_UNARY_RPC_TIMEOUT);
+        request.set_timeout(self.unary_rpc_timeout);
 
         match self.client.update(request).await {
             Ok(response) => {

src/sagittarius/runtime_status_service_client_impl.rs

@@ -1,17 +1,22 @@
 use crate::authorization::authorization::get_authorization_metadata;
-use crate::sagittarius::SAGITTARIUS_UNARY_RPC_TIMEOUT;
+use std::time::Duration;
 use tonic::{Extensions, Request, transport::Channel};
 use tucana::sagittarius::runtime_status_service_client::RuntimeStatusServiceClient;
 
 pub struct SagittariusRuntimeStatusServiceClient {
     client: RuntimeStatusServiceClient<Channel>,
     token: String,
+    unary_rpc_timeout: Duration,
 }
 
 impl SagittariusRuntimeStatusServiceClient {
-    pub fn new(channel: Channel, token: String) -> Self {
+    pub fn new(channel: Channel, token: String, unary_rpc_timeout: Duration) -> Self {
         let client = RuntimeStatusServiceClient::new(channel);
-        Self { client, token }
+        Self {
+            client,
+            token,
+            unary_rpc_timeout,
+        }
     }
 
     pub async fn update_runtime_status(
@@ -26,7 +31,7 @@ impl SagittariusRuntimeStatusServiceClient {
                 status: runtime_status_request.status,
             },
         );
-        request.set_timeout(SAGITTARIUS_UNARY_RPC_TIMEOUT);
+        request.set_timeout(self.unary_rpc_timeout);
 
         let response = match self.client.update(request).await {
             Ok(response) => {

src/server/dynamic_server.rs

@@ -40,6 +40,7 @@ pub struct AquilaDynamicServer {
     runtime_status_not_responding_after_secs: u64,
     runtime_status_stopped_after_not_responding_secs: u64,
     runtime_status_monitor_interval_secs: u64,
+    sagittarius_unary_rpc_timeout: Duration,
 }
 
 impl AquilaDynamicServer {
@@ -83,20 +84,27 @@ impl AquilaDynamicServer {
             runtime_status_monitor_interval_secs: config
                 .runtime_status_monitor_interval_secs
                 .clone(),
+            sagittarius_unary_rpc_timeout: Duration::from_secs(
+                config.sagittarius_unary_rpc_timeout_secs,
+            ),
         }
     }
 
     pub async fn start(&self) -> Result<(), tonic::transport::Error> {
         let module_service = Arc::new(Mutex::new(SagittariusModuleServiceClient::new(
             self.channel.clone(),
             self.token.clone(),
+            self.sagittarius_unary_rpc_timeout,
         )));
 
         info!("ModuleService started");
 
-        let runtime_status_service = Arc::new(Mutex::new(
-            SagittariusRuntimeStatusServiceClient::new(self.channel.clone(), self.token.clone()),
-        ));
+        let runtime_status_service =
+            Arc::new(Mutex::new(SagittariusRuntimeStatusServiceClient::new(
+                self.channel.clone(),
+                self.token.clone(),
+                self.sagittarius_unary_rpc_timeout,
+            )));
 
         info!("RuntimeStatusService started");
 
@@ -108,7 +116,7 @@ impl AquilaDynamicServer {
             module_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()),