Nats: db proposal

.gitignore

@@ -16,6 +16,7 @@ target/
 # nats
 rust/*/tmp
 rust/*/jwt
+rust/*/*/test_leaf_server/*
 rust/*/*/test_leaf_server.conf
 rust/*/*/leaf_server.conf
 rust/*/*/resolver.conf

rust/clients/host_agent/src/gen_leaf_server.rs

@@ -3,7 +3,7 @@ use util_libs::nats_server::{self, JetStreamConfig, LeafNodeRemote, LeafServer,
 const LEAF_SERVE_NAME: &str = "test_leaf_server";
 const LEAF_SERVER_CONFIG_PATH: &str = "test_leaf_server.conf";
 
-pub async fn run(_user_creds_path: &str) {
+pub async fn run(user_creds_path: &str) {
     let leaf_server_remote_conn_url = nats_server::get_hub_server_url();
     let leaf_client_conn_domain = "127.0.0.1";
     let leaf_client_conn_port = 4111;
@@ -27,7 +27,7 @@ pub async fn run(_user_creds_path: &str) {
     let leaf_node_remotes = vec![LeafNodeRemote {
         // sys account user (automated)
         url: leaf_server_remote_conn_url.to_string(),
-        credentials_path: None, // Some(user_creds_path.to_string()),
+        credentials_path: Some(user_creds_path.to_string()),
     }];
 
     // Create a new Leaf Server instance

rust/clients/host_agent/src/main.rs

@@ -1,19 +1,16 @@
 /*
- This client is associated with the:
-- WORKLOAD account
-- hpos user
+This client is associated with the:
+  - WORKLOAD account
+  - hpos user
 
-// This client is responsible for:
-  - subscribing to workload streams
-    - installing new workloads
-    - removing workloads
-    - send workload status upon request
+This client is responsible for subscribing the host agent to workload stream endpoints:
+  - installing new workloads
+  - removing workloads
   - sending active periodic workload reports
+  - sending workload status upon request
 */
 
-// mod auth;
-// mod utils;
-mod workloads;
+mod workload_manager;
 use anyhow::Result;
 use clap::Parser;
 use dotenv::dotenv;
@@ -57,12 +54,10 @@ async fn main() -> Result<(), AgentCliError> {
 }
 
 async fn daemonize() -> Result<(), async_nats::Error> {
-    // let user_creds_path = auth::initializer::run().await?;
-    let user_creds_path = "placeholder_creds_that_will_not_be_read".to_string();
-    gen_leaf_server::run(&user_creds_path).await;
-
-    let user_creds_path = nats_js_client::get_nats_client_creds("HOLO", "HPOS", "hpos");
-    workloads::manager::run(&user_creds_path).await?;
-
+    // let (host_pubkey, host_creds_path) = auth::initializer::run().await?;
+    let host_creds_path = nats_js_client::get_nats_client_creds("HOLO", "HPOS", "hpos");
+    let host_pubkey = "host_id_placeholder>";
+    gen_leaf_server::run(&host_creds_path).await;
+    workload_manager::run(host_pubkey, &host_creds_path).await?;
     Ok(())
 }

rust/clients/host_agent/src/workloads/manager.rs → rust/clients/host_agent/src/workload_manager.rs

@@ -7,32 +7,33 @@
   - subscribing to workload streams
     - installing new workloads
     - removing workloads
-    - send workload status upon request
-  - sending active periodic workload reports
+    - sending workload status upon request
+    - sending active periodic workload reports
 */
 
-use super::endpoints;
 use anyhow::{anyhow, Result};
-// use mongodb::{options::ClientOptions, Client as MongoDBClient};
-use std::time::Duration;
+use mongodb::{options::ClientOptions, Client as MongoDBClient};
+use std::{sync::Arc, time::Duration};
 use util_libs::{
-    // db::mongodb::get_mongodb_url,
+    db::mongodb::get_mongodb_url,
     js_stream_service::JsServiceParamsPartial,
-    nats_js_client::{self, EndpointType, JsClient},
+    nats_js_client::{self, EndpointType,  },
 };
 use workload::{
     WorkloadApi, WORKLOAD_SRV_DESC, WORKLOAD_SRV_NAME, WORKLOAD_SRV_SUBJ, WORKLOAD_SRV_VERSION,
 };
+use async_nats::Message;
 
 const HOST_AGENT_CLIENT_NAME: &str = "Host Agent";
-const HOST_AGENT_CLIENT_INBOX_PREFIX: &str = "_host_inbox";
+const HOST_AGENT_INBOX_PREFIX: &str = "_host_inbox";
 
-// TODO: Use _user_creds_path for auth once we add in the more resilient auth pattern.
-pub async fn run(user_creds_path: &str) -> Result<(), async_nats::Error> {
+// TODO: Use _host_creds_path for auth once we add in the more resilient auth pattern.
+pub async fn run(host_pubkey: &str, host_creds_path: &str) -> Result<(), async_nats::Error> {
     log::info!("HPOS Agent Client: Connecting to server...");
-    // ==================== NATS Setup ====================
-    log::info!("user_creds_path : {}", user_creds_path);
+    log::info!("host_creds_path : {}", host_creds_path);
+    log::info!("host_pubkey : {}", host_pubkey);
 
+    // ==================== NATS Setup ====================
     // Connect to Nats server
     let nats_url = nats_js_client::get_nats_url();
     log::info!("nats_url : {}", nats_url);
@@ -49,34 +50,33 @@ pub async fn run(user_creds_path: &str) -> Result<(), async_nats::Error> {
 
     // Spin up Nats Client and loaded in the Js Stream Service
     let host_workload_client =
-        nats_js_client::DefaultJsClient::new(nats_js_client::NewDefaultJsClientParams {
+        nats_js_client::JsClient::new(nats_js_client::NewJsClientParams {
             nats_url,
             name: HOST_AGENT_CLIENT_NAME.to_string(),
             inbox_prefix: format!(
                 "{}_{}",
-                HOST_AGENT_CLIENT_INBOX_PREFIX, "<host_id_placeholder>"
+                HOST_AGENT_INBOX_PREFIX, host_pubkey
             ),
             service_params: vec![workload_stream_service_params],
-            credentials_path: None, // Some(user_creds_path.to_string()),
+            credentials_path: Some(host_creds_path.to_string()),
             opts: vec![nats_js_client::with_event_listeners(event_listeners)],
             ping_interval: Some(Duration::from_secs(10)),
             request_timeout: Some(Duration::from_secs(5)),
         })
         .await?;
 
     // ==================== DB Setup ====================
-
     // Create a new MongoDB Client and connect it to the cluster
-    // let mongo_uri = get_mongodb_url();
-    // let client_options = ClientOptions::parse(mongo_uri).await?;
-    // let client = MongoDBClient::with_options(client_options)?;
+    let mongo_uri = get_mongodb_url();
+    let client_options = ClientOptions::parse(mongo_uri).await?;
+    let client = MongoDBClient::with_options(client_options)?;
 
     // Generate the Workload API with access to db
-    let workload_api = WorkloadApi::new().await?;
+    let workload_api = WorkloadApi::new(&client).await?;
 
     // ==================== API ENDPOINTS ====================
     // Register Workload Streams for Host Agent to consume
-    // (subjects should be published by orchestrator or nats-db-connector)
+    // NB: Subjects are published by orchestrator or nats-db-connector
     let workload_service = host_workload_client
         .get_js_service(WORKLOAD_SRV_NAME.to_string())
         .await
@@ -85,44 +85,48 @@ pub async fn run(user_creds_path: &str) -> Result<(), async_nats::Error> {
         ))?;
 
     workload_service
-        .add_local_consumer(
+        .add_local_consumer::<workload::types::ApiResult>(
             "start_workload",
             "start",
-            EndpointType::Async(endpoints::start_workload(&workload_api).await),
+            EndpointType::Async(workload_api.call(|api: WorkloadApi, msg: Arc<Message>| {
+                async move {
+                    api.start_workload(msg).await
+                }
+            })),
             None,
         )
         .await?;
 
     workload_service
-        .add_local_consumer(
-            "signal_status_update",
-            "signal_status_update",
-            EndpointType::Async(endpoints::signal_status_update(&workload_api).await),
+        .add_local_consumer::<workload::types::ApiResult>(
+            "send_workload_status",
+            "send_status",
+            EndpointType::Async(workload_api.call(|api: WorkloadApi, msg: Arc<Message>| {
+                async move {
+                    api.send_workload_status(msg).await
+                }
+            })),
             None,
         )
         .await?;
 
     workload_service
-        .add_local_consumer(
-            "remove_workload",
-            "remove",
-            EndpointType::Async(endpoints::remove_workload(&workload_api).await),
+        .add_local_consumer::<workload::types::ApiResult>(
+            "uninstall_workload",
+            "uninstall",
+            EndpointType::Async(workload_api.call(|api: WorkloadApi, msg: Arc<Message>| {
+                async move {
+                    api.uninstall_workload(msg).await
+                }
+            })),
             None,
         )
         .await?;
 
     // Only exit program when explicitly requested
     tokio::signal::ctrl_c().await?;
 
-    // TODO: bring this back with actual implementation
-    // log::warn!("CTRL+C detected. Please press CTRL+C again within 5 seconds to confirm exit...");
-    // tokio::select! {
-    //     _ = tokio::time::sleep(tokio::time::Duration::from_secs(5)) => { log::warn!("Resuming service.") },
-    //     _ = tokio::signal::ctrl_c() => log::error!("Shutting down."),
-    // }
-
     // Close client and drain internal buffer before exiting to make sure all messages are sent
     host_workload_client.close().await?;
-
     Ok(())
 }

rust/services/workload/Cargo.toml

@@ -14,6 +14,8 @@ env_logger = { workspace = true }
 log = { workspace = true }
 dotenv = { workspace = true }
 thiserror = { workspace = true }
+semver = "1.0.24"
+rand = "0.8.5"
 mongodb = "3.1"
 bson = { version = "2.6.1", features = ["chrono-0_4"] }
 url = { version = "2", features = ["serde"] }