fix: (controller) restart after failing

controller/lib/src/external_api/instance/controller.rs

@@ -1,12 +1,12 @@
 use std::sync::Arc;
 
-use crate::external_api::generic::model::{APIResponse, APIResponseMetadata};
+use crate::external_api::generic::model::{APIResponse, APIResponseMetadata, Pagination};
 use crate::external_api::interface::ActixAppState;
 use crate::external_api::workload::controller::WorkloadControllerError;
 use crate::external_api::workload::service::WorkloadServiceError;
 
 use super::super::workload::service::WorkloadService;
-use super::model::{Instance, InstanceDTO, InstanceVector, Pagination};
+use super::model::{Instance, InstanceDTO};
 use super::service::{InstanceService, InstanceServiceError};
 use actix_web::http::StatusCode;
 use actix_web::{web, HttpResponse, Responder, Scope};
@@ -89,13 +89,18 @@ impl InstanceController {
         body: web::Json<InstanceDTO>,
         data: web::Data<ActixAppState>,
     ) -> impl Responder {
-        let instance_service =
-            match InstanceService::new(&data.grpc_address, &data.etcd_address).await {
-                Ok(service) => service,
-                Err(err) => {
-                    return InstanceControllerError::InstanceServiceError(err).into();
-                }
-            };
+        let instance_service = match InstanceService::new(
+            &data.grpc_address,
+            &data.etcd_address,
+            data.grpc_client_connection_max_retries,
+        )
+        .await
+        {
+            Ok(service) => service,
+            Err(err) => {
+                return InstanceControllerError::InstanceServiceError(err).into();
+            }
+        };
 
         let mut workload_service = match WorkloadService::new(&data.etcd_address).await {
             Ok(service) => service,
@@ -142,13 +147,18 @@ impl InstanceController {
         params: web::Path<(String, String)>,
         data: web::Data<ActixAppState>,
     ) -> impl Responder {
-        let mut instance_service =
-            match InstanceService::new(&data.grpc_address, &data.etcd_address).await {
-                Ok(service) => service,
-                Err(err) => {
-                    return InstanceControllerError::InstanceServiceError(err).into();
-                }
-            };
+        let mut instance_service = match InstanceService::new(
+            &data.grpc_address,
+            &data.etcd_address,
+            data.grpc_client_connection_max_retries,
+        )
+        .await
+        {
+            Ok(service) => service,
+            Err(err) => {
+                return InstanceControllerError::InstanceServiceError(err).into();
+            }
+        };
 
         let (namespace, name) = params.into_inner();
         match instance_service
@@ -183,13 +193,18 @@ impl InstanceController {
         params: web::Path<(String, String)>,
         data: web::Data<ActixAppState>,
     ) -> impl Responder {
-        let mut instance_service =
-            match InstanceService::new(&data.grpc_address, &data.etcd_address).await {
-                Ok(service) => service,
-                Err(err) => {
-                    return InstanceControllerError::InstanceServiceError(err).into();
-                }
-            };
+        let mut instance_service = match InstanceService::new(
+            &data.grpc_address,
+            &data.etcd_address,
+            data.grpc_client_connection_max_retries,
+        )
+        .await
+        {
+            Ok(service) => service,
+            Err(err) => {
+                return InstanceControllerError::InstanceServiceError(err).into();
+            }
+        };
 
         let (namespace, name) = params.into_inner();
         match instance_service
@@ -216,27 +231,31 @@ impl InstanceController {
         pagination: Option<web::Query<Pagination>>,
         data: web::Data<ActixAppState>,
     ) -> impl Responder {
-        let mut instance_service =
-            match InstanceService::new(&data.grpc_address, &data.etcd_address).await {
-                Ok(service) => service,
-                Err(err) => {
-                    return InstanceControllerError::InstanceServiceError(err).into();
-                }
-            };
+        let mut instance_service = match InstanceService::new(
+            &data.grpc_address,
+            &data.etcd_address,
+            data.grpc_client_connection_max_retries,
+        )
+        .await
+        {
+            Ok(service) => service,
+            Err(err) => {
+                return InstanceControllerError::InstanceServiceError(err).into();
+            }
+        };
 
-        match pagination {
+        let instances = match pagination {
             Some(pagination) => {
-                let instances = instance_service
+                instance_service
                     .get_instances(pagination.limit, pagination.offset, &namespace)
-                    .await;
-
-                InstanceVector::new(instances).to_http()
+                    .await
             }
-            None => {
-                let instances = instance_service.get_instances(0, 0, &namespace).await;
+            None => instance_service.get_instances(0, 0, &namespace).await,
+        };
 
-                InstanceVector::new(instances).to_http()
-            }
-        }
+        HttpResponse::build(StatusCode::OK).json(APIResponse::<Vec<Instance>> {
+            data: instances,
+            metadata: APIResponseMetadata::default(),
+        })
     }
 }

controller/lib/src/external_api/instance/mod.rs

@@ -1,4 +1,3 @@
 pub mod controller;
-mod filter;
 pub mod model;
 pub mod service;

controller/lib/src/external_api/instance/model.rs

@@ -1,11 +1,13 @@
 use std::net::Ipv4Addr;
 
-use actix_web::HttpResponse;
 use proto::controller::{InstanceState, Type};
 use rand::{distributions::Alphanumeric, Rng};
 use serde::{Deserialize, Serialize};
 
-use crate::external_api::generic::model::{APIResponse, APIResponseMetadata};
+#[derive(Deserialize, Serialize)]
+pub struct InstanceDTO {
+    pub workload_name: String,
+}
 
 #[derive(Deserialize, Serialize, Clone, Debug)]
 pub struct Instance {
@@ -23,40 +25,12 @@ pub struct Instance {
     pub namespace: String,
 }
 
-#[derive(Deserialize, Serialize)]
-pub struct InstanceVector {
-    pub instances: Vec<Instance>,
-}
-impl InstanceVector {
-    pub fn new(instances: Vec<Instance>) -> InstanceVector {
-        InstanceVector { instances }
-    }
-    pub fn to_http(self) -> HttpResponse {
-        HttpResponse::Ok().json(APIResponse::<Vec<Instance>> {
-            data: self.instances,
-            metadata: APIResponseMetadata::default(),
-        })
-    }
-}
-
 #[derive(Deserialize, Serialize, Clone, Debug)]
 pub struct Resource {
     pub limit: Option<ResourceSummary>,
     pub usage: Option<ResourceSummary>,
 }
 
-#[derive(Deserialize, Serialize)]
-pub struct InstanceDTO {
-    pub workload_name: String,
-}
-
-#[derive(Deserialize, Serialize)]
-
-pub struct Pagination {
-    pub limit: u32,
-    pub offset: u32,
-}
-
 #[derive(Deserialize, Serialize, Clone, Debug)]
 pub struct ResourceSummary {
     pub cpu: u64,

controller/lib/src/external_api/instance/service.rs

@@ -1,16 +1,17 @@
 use std::net::{Ipv4Addr, SocketAddr};
 use std::sync::Arc;
+use std::time::Duration;
 
-use super::filter::InstanceFilterService;
 use super::model::Instance;
 use crate::etcd::{EtcdClient, EtcdClientError};
+use crate::external_api::generic::filter::FilterService;
 use crate::external_api::workload::model::Workload;
 use crate::grpc_client::interface::{SchedulerClientInterface, SchedulerClientInterfaceError};
 use log::{debug, trace};
 use proto::controller::InstanceState;
 use serde_json;
 use thiserror::Error;
-use tokio::sync::Mutex;
+use tokio::{sync::Mutex, time};
 use tonic::{Request, Status};
 
 #[derive(Debug, Error)]
@@ -32,7 +33,7 @@ pub enum InstanceServiceError {
 pub struct InstanceService {
     grpc_service: SchedulerClientInterface,
     etcd_service: EtcdClient,
-    filter_service: InstanceFilterService,
+    filter_service: FilterService,
 }
 
 // `InstanceService` is a struct that is inspired from Controllers Provider Modules architectures. It is used as a service in the InstanceController. A service can use other services.
@@ -46,15 +47,19 @@ impl InstanceService {
     pub async fn new(
         grpc_address: &str,
         etcd_address: &SocketAddr,
+        grpc_client_connection_max_retries: u32,
     ) -> Result<Self, InstanceServiceError> {
         Ok(InstanceService {
-            grpc_service: SchedulerClientInterface::new(grpc_address.to_string())
-                .await
-                .map_err(InstanceServiceError::SchedulerClientInterfaceError)?,
+            grpc_service: SchedulerClientInterface::new(
+                grpc_address.to_string(),
+                grpc_client_connection_max_retries,
+            )
+            .await
+            .map_err(InstanceServiceError::SchedulerClientInterfaceError)?,
             etcd_service: EtcdClient::new(etcd_address.to_string())
                 .await
                 .map_err(InstanceServiceError::EtcdError)?,
-            filter_service: InstanceFilterService::new(),
+            filter_service: FilterService::new(),
         })
     }
 
@@ -149,6 +154,8 @@ impl InstanceService {
     pub fn schedule_instance(this: Arc<Mutex<Self>>, mut instance: Instance) {
         //Spawn a thread to start the instance
         tokio::spawn(async move {
+            let mut cooldown = 1;
+
             loop {
                 let mut stream = this
                     .clone()
@@ -196,8 +203,9 @@ impl InstanceService {
                 }
 
                 instance.num_restarts += 1;
-
-                debug!("Restarting instance {}", instance.id);
+                debug!("Restarting instance {} in {}s", instance.id, cooldown);
+                time::sleep(Duration::from_secs(cooldown)).await;
+                cooldown *= 2;
 
                 this.clone()
                     .lock()

controller/lib/src/external_api/interface.rs

@@ -24,6 +24,7 @@ pub struct ExternalAPIInterface {}
 pub struct ActixAppState {
     pub etcd_address: SocketAddr,
     pub grpc_address: String,
+    pub grpc_client_connection_max_retries: u32,
 }
 
 impl ExternalAPIInterface {
@@ -32,6 +33,7 @@ impl ExternalAPIInterface {
         num_workers: usize,
         etcd_address: SocketAddr,
         grpc_address: String,
+        grpc_client_connection_max_retries: u32,
     ) -> Result<Self, ExternalAPIInterfaceError> {
         let mut etcd_client = EtcdClient::new(etcd_address.to_string())
             .await
@@ -65,6 +67,7 @@ impl ExternalAPIInterface {
                 .app_data(web::Data::new(ActixAppState {
                     etcd_address,
                     grpc_address: grpc_address.clone(),
+                    grpc_client_connection_max_retries,
                 }))
                 .route("/health", web::get().to(HttpResponse::Ok))
                 .service(workload::controller::WorkloadController {}.services())

controller/lib/src/external_api/namespace/controller.rs

@@ -115,18 +115,19 @@ impl NamespaceController {
             Err(err) => return NamespaceControllerError::NamespaceServiceError(err).into(),
         };
 
-        match pagination {
+        let namespaces = match pagination {
             Some(pagination) => {
-                let namespaces = namespace_service
+                namespace_service
                     .get_all_namespace(pagination.limit, pagination.offset)
-                    .await;
-                namespaces.to_http()
+                    .await
             }
-            None => {
-                let namespaces = namespace_service.get_all_namespace(0, 0).await;
-                namespaces.to_http()
-            }
-        }
+            None => namespace_service.get_all_namespace(0, 0).await,
+        };
+
+        HttpResponse::build(StatusCode::OK).json(APIResponse::<Vec<Namespace>> {
+            metadata: APIResponseMetadata::default(),
+            data: namespaces,
+        })
     }
 
     pub async fn patch_namespace(