feat: added update for runtime status service to runner

Author: raphael-goetz

crates/base/src/client/mod.rs

@@ -1,6 +1,7 @@
 use std::time::{SystemTime, UNIX_EPOCH};
 
-use tonic::transport::Channel;
+use tokio::time::sleep;
+use tonic::transport::{Channel, Endpoint};
 use tucana::{
     aquila::{
         RuntimeStatusUpdateRequest, runtime_status_service_client::RuntimeStatusServiceClient,
@@ -9,15 +10,72 @@ use tucana::{
     shared::{AdapterConfiguration, AdapterRuntimeStatus, RuntimeFeature},
 };
 
-struct DracoRuntimeStatusService {
+pub struct DracoRuntimeStatusService {
     channel: Channel,
     identifier: String,
     features: Vec<RuntimeFeature>,
     configs: Vec<AdapterConfiguration>,
 }
 
+const MAX_BACKOFF: u64 = 2000 * 60;
+const MAX_RETRIES: i8 = 10;
+
+// Will create a channel and retry if its not possible
+pub async fn create_channel_with_retry(channel_name: &str, url: String) -> Channel {
+    let mut backoff = 100;
+    let mut retries = 0;
+
+    loop {
+        let channel = match Endpoint::from_shared(url.clone()) {
+            Ok(c) => {
+                log::debug!("Creating a new endpoint for the: {} Service", channel_name);
+                c.connect_timeout(std::time::Duration::from_secs(2))
+                    .timeout(std::time::Duration::from_secs(10))
+            }
+            Err(err) => {
+                panic!(
+                    "Cannot create Endpoint for Service: `{}`. Reason: {:?}",
+                    channel_name, err
+                );
+            }
+        };
+
+        match channel.connect().await {
+            Ok(ch) => {
+                return ch;
+            }
+            Err(err) => {
+                log::warn!(
+                    "Retry connect to `{}` using url: `{}` failed: {:?}, retrying in {}ms",
+                    channel_name,
+                    url,
+                    err,
+                    backoff
+                );
+                sleep(std::time::Duration::from_millis(backoff)).await;
+
+                backoff = (backoff * 2).min(MAX_BACKOFF);
+                retries += 1;
+
+                if retries >= MAX_RETRIES {
+                    panic!("Reached max retries to url {}", url)
+                }
+            }
+        }
+    }
+}
 impl DracoRuntimeStatusService {
-    fn new(
+    pub async fn from_url(
+        aquila_url: String,
+        identifier: String,
+        features: Vec<RuntimeFeature>,
+        configs: Vec<AdapterConfiguration>,
+    ) -> Self {
+        let channel = create_channel_with_retry("Aquila", aquila_url).await;
+        Self::new(channel, identifier, features, configs)
+    }
+
+    pub fn new(
         channel: Channel,
         identifier: String,
         features: Vec<RuntimeFeature>,
@@ -35,7 +93,7 @@ impl DracoRuntimeStatusService {
         self.features.push(feat);
     }
 
-    async fn update_runtime_status_by_status(
+    pub async fn update_runtime_status_by_status(
         &self,
         status: tucana::shared::adapter_runtime_status::Status,
     ) {

crates/base/src/runner.rs

@@ -1,4 +1,5 @@
 use crate::{
+    client::DracoRuntimeStatusService,
     config::AdapterConfig,
     store::AdapterStore,
     traits::{LoadConfig, Server as AdapterServer},
@@ -8,6 +9,7 @@ use std::sync::Arc;
 use tokio::signal;
 use tonic::transport::Server;
 use tonic_health::pb::health_server::HealthServer;
+use tucana::shared::{AdapterConfiguration, RuntimeFeature};
 
 /// Context passed to adapter server implementations containing all shared resources
 pub struct ServerContext<C: LoadConfig> {
@@ -23,6 +25,11 @@ pub struct ServerRunner<C: LoadConfig> {
 }
 
 impl<C: LoadConfig> ServerRunner<C> {
+    
+    pub fn get_server_config(&self) -> Arc<C> {
+        self.context.server_config.clone()
+    }
+
     pub async fn new<S: AdapterServer<C>>(server: S) -> anyhow::Result<Self> {
         env_logger::Builder::from_default_env()
             .filter_level(log::LevelFilter::Debug)
@@ -50,11 +57,34 @@ impl<C: LoadConfig> ServerRunner<C> {
         })
     }
 
-    pub async fn serve(self) -> anyhow::Result<()> {
+    pub async fn serve(
+        self,
+        runtime_feature: Vec<RuntimeFeature>,
+        runtime_config: Vec<AdapterConfiguration>,
+    ) -> anyhow::Result<()> {
         let config = self.context.adapter_config.clone();
+        let mut runtime_status_service: Option<DracoRuntimeStatusService> = None;
+
         log::info!("Starting Draco Variant: {}", config.draco_variant);
 
         if !config.is_static() {
+            runtime_status_service = Some(
+                DracoRuntimeStatusService::from_url(
+                    config.aquila_url.clone(),
+                    config.draco_variant.clone(),
+                    runtime_feature,
+                    runtime_config,
+                )
+                .await,
+            );
+
+            if let Some(ser) = &runtime_status_service {
+                ser.update_runtime_status_by_status(
+                    tucana::shared::adapter_runtime_status::Status::NotReady,
+                )
+                .await;
+            };
+
             let definition_service = FlowUpdateService::from_url(
                 config.aquila_url.clone(),
                 config.definition_path.as_str(),
@@ -95,6 +125,13 @@ impl<C: LoadConfig> ServerRunner<C> {
         } = self;
         // Init the adapter server (e.g. create underlying HTTP server)
         server.init(&context).await?;
+
+        if let Some(ser) = &runtime_status_service {
+            ser.update_runtime_status_by_status(
+                tucana::shared::adapter_runtime_status::Status::Running,
+            )
+            .await;
+        };
         log::info!("Draco successfully initialized.");
 
         #[cfg(unix)]
@@ -158,6 +195,12 @@ impl<C: LoadConfig> ServerRunner<C> {
                 }
             }
         }
+        if let Some(ser) = &runtime_status_service {
+            ser.update_runtime_status_by_status(
+                tucana::shared::adapter_runtime_status::Status::Stopped,
+            )
+            .await;
+        };
 
         log::info!("Draco shutdown complete");
         Ok(())