feat(worker/monitor): add metrics collector

Cargo.toml

@@ -18,12 +18,17 @@ chrono = { version = "0.4", default-features = false, features = [
     "std",
     "serde",
 ] }
+eyre = "0.6.12"
+reqwest = { version = "0.12.1" , features = ["json"] }
 serde = { version = "1", features = ["derive"] }
 serde_json = "1"
+sysinfo = "0.30.7"
+tokio = { version = "1.36.0", features = ["full"] }
 uuid = { version = "1", features = ["serde", "v4"] }
 
 [workspace.lints.clippy]
 all = "warn"
 pedantic = "warn"
 wildcard_imports = "allow"
 module_name_repetitions = "allow"
+cast_precision_loss = "allow"

proto/src/common/node.rs

@@ -22,7 +22,8 @@ pub enum NodeKind {
 
 #[derive(Debug, Serialize, Deserialize)]
 pub struct Metrics {
-    pub cpu_usage: f64,
+    /// Collection of CPU usages.
+    pub cpu_usage: Vec<f64>,
     /// The total memory, in MiB.
     pub mem_total_mib: f64,
     /// The used memory, in MiB.

worker/Cargo.toml

@@ -7,3 +7,10 @@ edition.workspace = true
 workspace = true
 
 [dependencies]
+eyre.workspace = true
+proto.workspace = true
+reqwest.workspace = true
+serde.workspace = true
+serde_json.workspace = true
+sysinfo.workspace = true
+tokio.workspace = true

worker/src/main.rs

@@ -1,3 +1,24 @@
-fn main() {
-    println!("Hello, world!");
+mod monitor;
+use std::{thread::sleep, time::Duration};
+
+use eyre::Result;
+use monitor::{MetricsCollector, MetricsSender};
+
+/// `clap` crate report interval placeholder
+/// The value 500 is for quick visualization purposes.
+const REPORT_INTERVAL_IN_MILLIS: u64 = 500;
+
+/// `clap` crate url placeholder
+const AGT_MGR_REQUEST_URL: &str = "http://localhost:8080/http/agt_mgr";
+
+#[tokio::main]
+async fn main() -> Result<()> {
+    let mut monitor = MetricsSender::new(AGT_MGR_REQUEST_URL)?;
+    let mut metrics_report: MetricsCollector = MetricsCollector::new();
+
+    loop {
+        let metrics = metrics_report.get_metrics();
+        monitor.send_request(&metrics).await?;
+        sleep(Duration::from_millis(REPORT_INTERVAL_IN_MILLIS));
+    }
 }

worker/src/monitor/metrics_collector.rs

@@ -0,0 +1,67 @@
+use std::{thread::sleep, time::Duration};
+
+use proto::common::node::Metrics;
+use sysinfo::System;
+
+const CPU_DELAY_IN_MILLIS: u64 = 500;
+
+/// Struct containing system's information such as CPU and Memory.
+pub struct MetricsCollector {
+    system: System,
+}
+impl MetricsCollector {
+    /// Instantiates a new `MetricsCollector`.
+    ///
+    /// Responsable for collecting metrics about the System, such as
+    /// CPU Percentage Usage, Free Memory Space.
+    pub fn new() -> Self {
+        Self {
+            system: System::new_all(),
+        }
+    }
+
+    /// Get the `cpu_usage` and `free_memory` metrics from the current system.
+    ///
+    /// When calling this method, will sleep the thread for
+    /// `CPU_DELAY_IN_MILLIS`.
+    pub fn get_metrics(&mut self) -> Metrics {
+        Metrics {
+            cpu_usage: self.get_cpu_usage(),
+            mem_total_mib: self.get_total_memory(),
+            mem_used_mib: self.get_used_memory(),
+        }
+    }
+
+    fn get_cpu_usage(&mut self) -> Vec<f64> {
+        self.system.refresh_cpu();
+        sleep(Duration::from_millis(CPU_DELAY_IN_MILLIS));
+        self.system.refresh_cpu();
+
+        let all_cpus_usages: Vec<f64> = self
+            .system
+            .cpus()
+            .iter()
+            .map(|cpu| f64::from(cpu.cpu_usage()))
+            .collect();
+
+        all_cpus_usages
+    }
+
+    fn get_total_memory(&mut self) -> f64 {
+        (self.get_total_memory_as_byte() / 1024 / 1024) as f64
+    }
+
+    fn get_total_memory_as_byte(&mut self) -> u64 {
+        self.system.refresh_memory();
+
+        self.system.total_memory()
+    }
+
+    fn get_used_memory(&mut self) -> f64 {
+        (self.get_used_memory_as_byte() / 1024 / 1024) as f64
+    }
+
+    fn get_used_memory_as_byte(&mut self) -> u64 {
+        self.system.used_memory()
+    }
+}

worker/src/monitor/metrics_sender.rs

@@ -0,0 +1,34 @@
+use eyre::Result;
+use proto::common::node::Metrics;
+use reqwest::{Client, Url};
+
+/// A connection pool that sends the collected `Metrics` to the `http`component.
+pub struct MetricsSender {
+    client: Client,
+    base_url: String,
+}
+
+impl MetricsSender {
+    /// Instantiates a new `MetricsSender`.
+    ///
+    ///  Sends requests containing a JSON with the collected metrics
+    ///   to the `http` component from `ctl` node.
+    pub fn new(url: &str) -> Result<Self> {
+        let client = Client::builder().build()?;
+
+        let base_url = Url::parse(url)?.to_string();
+
+        Ok(Self { client, base_url })
+    }
+
+    /// Sends a POST request containing `Metrics` to the `http` component.
+    pub async fn send_request(&mut self, metrics: &Metrics) -> Result<()> {
+        self.client
+            .post(&self.base_url)
+            .json(metrics)
+            .send()
+            .await?;
+
+        Ok(())
+    }
+}

worker/src/monitor/mod.rs

@@ -0,0 +1,5 @@
+mod metrics_collector;
+mod metrics_sender;
+
+pub use metrics_collector::*;
+pub use metrics_sender::*;