Feat ping with node usage (#298)

* refactor: move gateway http logic to separate part

* feat: collect node metrics to gateway services

Author: giangndm

Cargo.lock

@@ -63,7 +63,7 @@ We are actively refactoring entire media server and network stack with [sans-io-
 | MoQ                 | Media-over-Quic                                                                   | ❌     |
 | Monitoring          | Dashboard for monitoring                                                          | ❌     |
 | Recording           | Record stream                                                                     | ❌     |
-| Gateway             | External gateway [RFC-0003](https://github.com/8xFF/rfcs/pull/3)                  | 🚧     |
+| Gateway             | External gateway [RFC-0003](https://github.com/8xFF/rfcs/pull/3)                  | 🚀     |
 | Connector           | External event handling                                                           | ❌     |
 
 Status:

README.md

@@ -28,12 +28,14 @@ convert-enum = { workspace = true }
 num_enum = { workspace = true }
 derive_more = { workspace = true }
 rcgen = { version = "0.13", optional = true }
-maxminddb = "0.24.0"
+maxminddb = { version = "0.24.0", optional = true }
+sysinfo = { version = "0.30.12", optional = true }
 
 [features]
 default = ["gateway", "media", "connector", "cert_utils"]
-gateway = ["media-server-gateway", "quinn_vnet"]
-media = ["media-server-runner", "quinn_vnet"]
+gateway = ["media-server-gateway", "quinn_vnet", "node_metrics", "maxminddb"]
+media = ["media-server-runner", "quinn_vnet", "node_metrics"]
 connector = ["quinn_vnet"]
 cert_utils = ["rcgen", "rustls"]
 quinn_vnet = ["rustls", "quinn"]
+node_metrics = ["sysinfo"]

bin/Cargo.toml

@@ -3,4 +3,7 @@
 pub enum MediaServerError {
     GatewayRpcError = 0x00020001,
     InvalidConnId = 0x00020002,
+    NodePoolEmpty = 0x00020003,
+    MediaResError = 0x00020004,
+    NotImplemented = 0x00020005,
 }

bin/src/errors.rs

@@ -4,6 +4,8 @@ use atm0s_sdn::{NodeAddr, NodeId};
 
 mod errors;
 mod http;
+#[cfg(feature = "node_metrics")]
+mod node_metrics;
 #[cfg(feature = "quinn_vnet")]
 mod quinn;
 pub mod server;

bin/src/lib.rs

@@ -0,0 +1,67 @@
+use std::{
+    sync::mpsc::{channel, Receiver},
+    time::Duration,
+};
+
+use media_server_gateway::NodeMetrics;
+use sans_io_runtime::ErrorDebugger2;
+use sysinfo::{Disks, System};
+
+const REFRESH_INTERVAL_SECONDS: u64 = 2;
+
+pub struct NodeMetricsCollector {
+    rx: Receiver<NodeMetrics>,
+}
+
+impl Default for NodeMetricsCollector {
+    fn default() -> Self {
+        let (tx, rx) = channel();
+        let mut sys = System::new_all();
+        let mut disks = Disks::new();
+
+        disks.refresh_list();
+        sys.refresh_all();
+        sys.refresh_cpu();
+
+        std::thread::spawn(move || {
+            loop {
+                disks.refresh();
+                sys.refresh_all();
+                sys.refresh_cpu();
+
+                let mut sum = 0.0;
+                for cpu in sys.cpus() {
+                    sum += cpu.cpu_usage();
+                }
+
+                let mut disk_used = 0;
+                let mut disk_sum = 0;
+                for disk in disks.iter() {
+                    disk_sum += disk.total_space();
+                    disk_used += disk.total_space() - disk.available_space();
+                }
+
+                tx.send(NodeMetrics {
+                    cpu: (sum as usize / sys.cpus().len()) as u8,
+                    memory: (100 * sys.used_memory() / sys.total_memory()) as u8,
+                    disk: (100 * disk_used / disk_sum) as u8,
+                })
+                .print_err2("Collect node metrics error");
+
+                // Sleeping to let time for the system to run for long
+                // enough to have useful information.
+                std::thread::sleep(Duration::from_secs(REFRESH_INTERVAL_SECONDS));
+            }
+        });
+
+        Self { rx }
+    }
+}
+
+impl NodeMetricsCollector {
+    /// Only return data in each interval, if not return None.
+    /// Node that this method must node blocking thread
+    pub fn pop_measure(&mut self) -> Option<NodeMetrics> {
+        self.rx.try_recv().ok()
+    }
+}