refactor/nanocld: utils to be great again (#806)

bin/nanocld/src/models/system.rs

@@ -42,6 +42,12 @@ pub struct EventManager {
   pub emitter: SystemEventEmitter,
 }
 
+impl Default for EventManager {
+  fn default() -> Self {
+    Self::new()
+  }
+}
+
 impl EventManager {
   pub fn new() -> Self {
     let (sx, rx) = mpsc::unbounded();

bin/nanocld/src/repositories/cargo.rs

@@ -2,17 +2,24 @@ use std::sync::Arc;
 
 use diesel::prelude::*;
 
-use nanocl_error::io::{IoError, IoResult};
+use futures_util::{stream::FuturesUnordered, StreamExt};
+use nanocl_error::{
+  io::{IoError, IoResult},
+  http::HttpResult,
+};
 
 use nanocl_stubs::{
   generic::{GenericFilter, GenericClause},
-  cargo::Cargo,
+  cargo::{Cargo, CargoDeleteQuery, CargoInspect},
   cargo_spec::{CargoSpecPartial, CargoSpec},
 };
 
 use crate::{
   gen_multiple, gen_where4string, utils,
-  models::{Pool, CargoDb, SpecDb, CargoUpdateDb},
+  objects::generic::*,
+  models::{
+    Pool, CargoDb, SpecDb, CargoUpdateDb, SystemState, NamespaceDb, ProcessDb,
+  },
   schema::cargoes,
 };
 
@@ -173,4 +180,49 @@ impl CargoDb {
     .await?;
     Ok(count)
   }
+
+  /// Return detailed information about the cargo for the given key
+  pub async fn inspect_by_pk(
+    key: &str,
+    state: &SystemState,
+  ) -> HttpResult<CargoInspect> {
+    let cargo = CargoDb::transform_read_by_pk(key, &state.pool).await?;
+    let processes = ProcessDb::read_by_kind_key(key, &state.pool).await?;
+    let (_, _, _, running_instances) = utils::process::count_status(&processes);
+    Ok(CargoInspect {
+      created_at: cargo.created_at,
+      namespace_name: cargo.namespace_name,
+      instance_total: processes.len(),
+      instance_running: running_instances,
+      spec: cargo.spec,
+      instances: processes,
+    })
+  }
+
+  /// This remove all cargo in the given namespace and all their instances (containers)
+  /// from the system (database and docker).
+  pub async fn delete_by_namespace(
+    namespace: &str,
+    state: &SystemState,
+  ) -> HttpResult<()> {
+    let namespace = NamespaceDb::read_by_pk(namespace, &state.pool).await?;
+    let cargoes =
+      CargoDb::read_by_namespace(&namespace.name, &state.pool).await?;
+    cargoes
+      .into_iter()
+      .map(|cargo| async move {
+        CargoDb::del_obj_by_pk(
+          &cargo.spec.cargo_key,
+          &CargoDeleteQuery::default(),
+          state,
+        )
+        .await
+      })
+      .collect::<FuturesUnordered<_>>()
+      .collect::<Vec<HttpResult<_>>>()
+      .await
+      .into_iter()
+      .collect::<HttpResult<Vec<_>>>()?;
+    Ok(())
+  }
 }

bin/nanocld/src/repositories/vm.rs

@@ -1,17 +1,22 @@
 use diesel::prelude::*;
 
-use nanocl_error::io::{IoError, IoResult};
+use nanocl_error::{
+  io::{IoError, IoResult},
+  http::HttpResult,
+};
 
 use nanocl_stubs::{
   generic::{GenericFilter, GenericClause},
-  vm::Vm,
+  vm::{Vm, VmInspect, VmSummary},
   vm_spec::{VmSpecPartial, VmSpec},
 };
 
 use crate::{
   gen_multiple, gen_where4string, utils,
   schema::vms,
-  models::{Pool, VmDb, VmUpdateDb, SpecDb},
+  models::{
+    Pool, VmDb, VmUpdateDb, SpecDb, SystemState, ProcessDb, NamespaceDb,
+  },
 };
 
 use super::generic::*;
@@ -141,4 +146,50 @@ impl VmDb {
       .r#where("namespace_name", GenericClause::Eq(name.to_owned()));
     VmDb::transform_read_by(&filter, pool).await
   }
+
+  /// Get detailed information about a VM by his key
+  pub async fn inspect_by_pk(
+    vm_key: &str,
+    state: &SystemState,
+  ) -> HttpResult<VmInspect> {
+    let vm = VmDb::transform_read_by_pk(vm_key, &state.pool).await?;
+    let processes =
+      ProcessDb::read_by_kind_key(&vm.spec.vm_key, &state.pool).await?;
+    let (_, _, _, running_instances) = utils::process::count_status(&processes);
+    Ok(VmInspect {
+      created_at: vm.created_at,
+      namespace_name: vm.namespace_name,
+      spec: vm.spec,
+      instance_total: processes.len(),
+      instance_running: running_instances,
+      instances: processes,
+    })
+  }
+
+  /// List VMs by namespace
+  pub async fn list_by_namespace(
+    nsp: &str,
+    pool: &Pool,
+  ) -> HttpResult<Vec<VmSummary>> {
+    let namespace = NamespaceDb::read_by_pk(nsp, pool).await?;
+    let vmes = VmDb::read_by_namespace(&namespace.name, pool).await?;
+    let mut vm_summaries = Vec::new();
+    for vm in vmes {
+      let spec = SpecDb::read_by_pk(&vm.spec.key, pool)
+        .await?
+        .try_to_vm_spec()?;
+      let processes =
+        ProcessDb::read_by_kind_key(&vm.spec.vm_key, pool).await?;
+      let (_, _, _, running_instances) =
+        utils::process::count_status(&processes);
+      vm_summaries.push(VmSummary {
+        created_at: vm.created_at,
+        namespace_name: vm.namespace_name,
+        instance_total: processes.len(),
+        instance_running: running_instances,
+        spec: spec.clone(),
+      });
+    }
+    Ok(vm_summaries)
+  }
 }

bin/nanocld/src/services/cargo.rs

@@ -61,7 +61,7 @@ pub async fn inspect_cargo(
 ) -> HttpResult<web::HttpResponse> {
   let namespace = utils::key::resolve_nsp(&qs.namespace);
   let key = utils::key::gen_key(&namespace, &path.1);
-  let cargo = utils::cargo::inspect_by_key(&key, &state).await?;
+  let cargo = CargoDb::inspect_by_pk(&key, &state).await?;
   Ok(web::HttpResponse::Ok().json(&cargo))
 }
 

bin/nanocld/src/services/vm.rs

@@ -47,7 +47,7 @@ pub async fn list_vm(
   qs: web::types::Query<GenericNspQuery>,
 ) -> HttpResult<web::HttpResponse> {
   let namespace = utils::key::resolve_nsp(&qs.namespace);
-  let vms = utils::vm::list_by_namespace(&namespace, &state.pool).await?;
+  let vms = VmDb::list_by_namespace(&namespace, &state.pool).await?;
   Ok(web::HttpResponse::Ok().json(&vms))
 }
 
@@ -73,7 +73,7 @@ pub async fn inspect_vm(
   let name = path.1.to_owned();
   let namespace = utils::key::resolve_nsp(&qs.namespace);
   let key = utils::key::gen_key(&namespace, &name);
-  let vm = utils::vm::inspect_by_key(&key, &state).await?;
+  let vm = VmDb::inspect_by_pk(&key, &state).await?;
   Ok(web::HttpResponse::Ok().json(&vm))
 }
 

bin/nanocld/src/utils/cargo.rs

@@ -13,15 +13,11 @@ use bollard_next::{
 use nanocl_stubs::{
   process::Process,
   generic::{GenericListNspQuery, GenericClause, GenericFilter},
-  cargo::{
-    Cargo, CargoSummary, CargoInspect, CargoKillOptions, CargoStats,
-    CargoStatsQuery, CargoDeleteQuery,
-  },
+  cargo::{Cargo, CargoSummary, CargoKillOptions, CargoStats, CargoStatsQuery},
 };
 
 use crate::{
   utils,
-  objects::generic::*,
   repositories::generic::*,
   models::{SystemState, CargoDb, ProcessDb, NamespaceDb, SecretDb, SpecDb},
 };
@@ -252,7 +248,7 @@ pub async fn delete_instances(
 
 /// Restart cargo instances (containers) by key
 pub async fn restart(key: &str, state: &SystemState) -> HttpResult<()> {
-  let cargo = utils::cargo::inspect_by_key(key, state).await?;
+  let cargo = CargoDb::transform_read_by_pk(key, &state.pool).await?;
   let processes =
     ProcessDb::read_by_kind_key(&cargo.spec.cargo_key, &state.pool).await?;
   processes
@@ -265,10 +261,10 @@ pub async fn restart(key: &str, state: &SystemState) -> HttpResult<()> {
         .map_err(HttpError::from)
     })
     .collect::<FuturesUnordered<_>>()
-    .collect::<Vec<Result<(), HttpError>>>()
+    .collect::<Vec<HttpResult<()>>>()
     .await
     .into_iter()
-    .collect::<Result<Vec<_>, _>>()?;
+    .collect::<HttpResult<Vec<_>>>()?;
   Ok(())
 }
 
@@ -314,51 +310,6 @@ pub async fn list(
   Ok(cargo_summaries)
 }
 
-/// Return detailed information about the cargo for the given key
-pub async fn inspect_by_key(
-  key: &str,
-  state: &SystemState,
-) -> HttpResult<CargoInspect> {
-  let cargo = CargoDb::transform_read_by_pk(key, &state.pool).await?;
-  let processes = ProcessDb::read_by_kind_key(key, &state.pool).await?;
-  let (_, _, _, running_instances) = utils::process::count_status(&processes);
-  Ok(CargoInspect {
-    created_at: cargo.created_at,
-    namespace_name: cargo.namespace_name,
-    instance_total: processes.len(),
-    instance_running: running_instances,
-    spec: cargo.spec,
-    instances: processes,
-  })
-}
-
-/// This remove all cargo in the given namespace and all their instances (containers)
-/// from the system (database and docker).
-pub async fn delete_by_namespace(
-  namespace: &str,
-  state: &SystemState,
-) -> HttpResult<()> {
-  let namespace = NamespaceDb::read_by_pk(namespace, &state.pool).await?;
-  let cargoes =
-    CargoDb::read_by_namespace(&namespace.name, &state.pool).await?;
-  cargoes
-    .into_iter()
-    .map(|cargo| async move {
-      CargoDb::del_obj_by_pk(
-        &cargo.spec.cargo_key,
-        &CargoDeleteQuery::default(),
-        state,
-      )
-      .await
-    })
-    .collect::<FuturesUnordered<_>>()
-    .collect::<Vec<Result<_, HttpError>>>()
-    .await
-    .into_iter()
-    .collect::<Result<Vec<_>, HttpError>>()?;
-  Ok(())
-}
-
 /// Send a signal to a cargo instance the cargo name can be used if the cargo has only one instance
 /// The signal is send to one instance only
 pub async fn kill_by_key(

bin/nanocld/src/utils/namespace.rs

@@ -15,7 +15,6 @@ use nanocl_stubs::{
 };
 
 use crate::{
-  utils,
   repositories::generic::*,
   models::{Pool, SystemState, CargoDb, NamespaceDb},
 };
@@ -56,7 +55,7 @@ pub async fn create(
 
 /// Delete a namespace by name and remove all associated cargo and vm.
 pub async fn delete_by_name(name: &str, state: &SystemState) -> HttpResult<()> {
-  utils::cargo::delete_by_namespace(name, state).await?;
+  CargoDb::delete_by_namespace(name, state).await?;
   NamespaceDb::del_by_pk(name, &state.pool).await?;
   if let Err(err) = state.docker_api.remove_network(name).await {
     log::error!("Unable to remove network {} got error: {}", name, err);
@@ -124,8 +123,7 @@ pub async fn inspect_by_name(
   let models = CargoDb::read_by_namespace(&namespace.name, &state.pool).await?;
   let mut cargoes = Vec::new();
   for cargo in models {
-    let cargo =
-      utils::cargo::inspect_by_key(&cargo.spec.cargo_key, state).await?;
+    let cargo = CargoDb::inspect_by_pk(&cargo.spec.cargo_key, state).await?;
     cargoes.push(cargo);
   }
   let network = state

bin/nanocld/src/utils/vm.rs

@@ -4,60 +4,13 @@ use bollard_next::service::{HostConfig, DeviceMapping};
 
 use nanocl_error::http::HttpResult;
 
-use nanocl_stubs::vm::{Vm, VmSummary, VmInspect};
+use nanocl_stubs::vm::Vm;
 
 use crate::{
   utils,
-  repositories::generic::*,
-  models::{
-    Pool, VmImageDb, SystemState, ProcessDb, NamespaceDb, VmDb, SpecDb,
-  },
+  models::{VmImageDb, SystemState},
 };
 
-/// Get detailed information about a VM by his key
-pub async fn inspect_by_key(
-  vm_key: &str,
-  state: &SystemState,
-) -> HttpResult<VmInspect> {
-  let vm = VmDb::transform_read_by_pk(vm_key, &state.pool).await?;
-  let processes =
-    ProcessDb::read_by_kind_key(&vm.spec.vm_key, &state.pool).await?;
-  let (_, _, _, running_instances) = utils::process::count_status(&processes);
-  Ok(VmInspect {
-    created_at: vm.created_at,
-    namespace_name: vm.namespace_name,
-    spec: vm.spec,
-    instance_total: processes.len(),
-    instance_running: running_instances,
-    instances: processes,
-  })
-}
-
-/// List VMs by namespace
-pub async fn list_by_namespace(
-  nsp: &str,
-  pool: &Pool,
-) -> HttpResult<Vec<VmSummary>> {
-  let namespace = NamespaceDb::read_by_pk(nsp, pool).await?;
-  let vmes = VmDb::read_by_namespace(&namespace.name, pool).await?;
-  let mut vm_summaries = Vec::new();
-  for vm in vmes {
-    let spec = SpecDb::read_by_pk(&vm.spec.key, pool)
-      .await?
-      .try_to_vm_spec()?;
-    let processes = ProcessDb::read_by_kind_key(&vm.spec.vm_key, pool).await?;
-    let (_, _, _, running_instances) = utils::process::count_status(&processes);
-    vm_summaries.push(VmSummary {
-      created_at: vm.created_at,
-      namespace_name: vm.namespace_name,
-      instance_total: processes.len(),
-      instance_running: running_instances,
-      spec: spec.clone(),
-    });
-  }
-  Ok(vm_summaries)
-}
-
 /// Create a VM instance from a VM image
 pub async fn create_instance(
   vm: &Vm,