feat(node-agent): implement node lifecycle

Add method to connect to cluster and stream node status. Retries  in case of the connection closing.

Signed-off-by: Martin Moreira de Jesus <martin.moreiradj@icloud.com>
Signed-off-by: Noé Tarbouriech <noe.tarbouriech@etu.umontpellier.fr>
Co-Authored-By: Noé Tarbouriech <noe.tarbouriech@etu.umontpellier.fr>

Author: Martin Moreira de Jesus

Cargo.lock

@@ -19,6 +19,10 @@ tracing-subscriber = "0.3.17"
 tracing-log = "0.1.3"
 uuid = { version = "1.4.1", features = ["v4", "macro-diagnostics", "fast-rng"] }
 tower-http = { version = "0.4.3", features = ["trace"] }
+clap = { version = "4.4.2", features = ["env", "derive"] }
+clap-verbosity-flag = "2.0.1"
+async-stream = "0.3.5"
+sysinfo = "0.29.9"
 
 [build-dependencies]
 tonic-build = "0.9.2"

node-agent/Cargo.toml

@@ -0,0 +1,27 @@
+//! Command-line arguments.
+use clap::Parser;
+use clap_verbosity_flag::{InfoLevel, Verbosity};
+
+/// Scheduler service for the Orka container orchestration system.
+#[derive(Parser, Debug)]
+#[command(author, version, about, long_about = None)]
+pub struct CliArguments {
+    #[arg(long, default_value_t = 3, env)]
+    pub lifecycle_retries: i32,
+
+    /// The port to start the node agent on.
+    #[arg(long, default_value_t = 50052, env)]
+    pub node_agent_port: u16,
+
+    /// The address of the scheduler to connect the node agent to.
+    #[arg(long, default_value = "[::]", env)]
+    pub scheduler_address: String,
+
+    /// The port of the scheduler to connect the node agent to.
+    #[arg(long, default_value_t = 50051, env)]
+    pub scheduler_port: u16,
+
+    /// Verbosity level.
+    #[command(flatten)]
+    pub verbose: Verbosity<InfoLevel>,
+}

node-agent/src/args.rs

@@ -1,9 +1,77 @@
+mod args;
 mod workload_manager;
 
-use orka_proto::node_agent::Workload;
+use clap::Parser;
+use orka_proto::{node_agent::Workload, scheduler_agent::{lifecycle_service_client::LifecycleServiceClient, ConnectionRequest, status_update_service_client::StatusUpdateServiceClient}};
+use tracing::{info, error};
+use uuid::Uuid;
 use workload_manager::container::client::ContainerClient;
+use anyhow::Result;
+use tracing_log::AsTrace;
+
+use crate::{workload_manager::container::metrics::metrics::any_to_resource, args::CliArguments};
+use crate::workload_manager::node::metrics::stream_node_status;
+
+async fn execute_node_lifecycle(
+    node_id: Uuid,
+    node_agent_port: u16,
+    scheduler_connection_string: String,
+) -> Result<()> {
+    info!(
+        "Connecting to scheduler on {}",
+        scheduler_connection_string
+    );
+
+    loop {
+        tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
 
-use crate::workload_manager::container::metrics::metrics::any_to_resource;
+        let mut lifecycle_client =
+            match LifecycleServiceClient::connect(scheduler_connection_string.clone()).await {
+                Ok(client) => Ok(client),
+                Err(e) => {
+                    error!("Failed to connect to scheduler: {:?}", e);
+                    Err(e)
+                }
+            }?;
+
+        match lifecycle_client
+            .join_cluster(ConnectionRequest {
+                id: node_id.to_string(),
+                port: node_agent_port as u32,
+            })
+            .await
+        {
+            Ok(_) => Ok(()),
+            Err(e) => {
+                error!("Failed to join cluster: {:?}", e);
+                Err(e)
+            }
+        }?;
+
+        info!("Joined cluster");
+
+        let mut client =
+            match StatusUpdateServiceClient::connect(scheduler_connection_string.clone()).await {
+                Ok(client) => Ok(client),
+                Err(e) => {
+                    error!("Failed to connect to scheduler: {:?}", e);
+                    Err(e)
+                }
+            }?;
+
+        match stream_node_status(node_id, &mut client, 15).await {
+            Ok(_) => {
+                info!("Node status stream ended, retrying");
+            }
+            Err(error) => {
+                error!(
+                    "Node status stream failed: {:?}, reconnecting to scheduler",
+                    error
+                );
+            }
+        };
+    }
+}
 
 const CID: &str = "nginx";
 
@@ -51,4 +119,42 @@ async fn main() {
         Ok(_) => panic!("Workload should not exist"),
         Err(_) => println!("Workload does not exist"),
     };
+
+    let args = CliArguments::parse();
+
+    tracing_subscriber::fmt()
+        .with_max_level(args.verbose.log_level_filter().as_trace())
+        .init();
+
+    info!(
+        app_name = env!("CARGO_PKG_NAME"),
+        app_version = env!("CARGO_PKG_VERSION"),
+        "Starting",
+    );
+
+    info!("Arguments: {:?}", args);
+
+    let scheduler_connection_string =
+        format!("http://{}:{}", args.scheduler_address, args.scheduler_port);
+
+    let node_id = Uuid::new_v4();
+
+    let retries = args.lifecycle_retries;
+
+    for _ in 0..retries {
+        match execute_node_lifecycle(
+            node_id,
+            args.node_agent_port,
+            scheduler_connection_string.clone(),
+        )
+        .await
+        {
+            Ok(_) => {
+                // will never be reached
+            }
+            Err(e) => {
+                error!("Failed to execute node lifecycle: {:?}", e);
+            }
+        }
+    }
 }

node-agent/src/main.rs

@@ -1 +1,2 @@
 pub mod container;
+pub mod node;