donpat1to/watcheragent
1
0
Fork 0
Code Issues 8 Pull Requests Actions Packages Projects 1 Releases 1 Wiki Activity

changed for endpoints

Browse Source
This commit is contained in:
Patrick Mahnke-Hartmann
2025-07-29 14:05:34 +02:00
parent f8d2a5c483
commit e663839bee
1 changed files with 76 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files

131
WatcherAgent/src/main.rs
View File

@@ -6,26 +6,7 @@ use std::{error::Error, fs, process::Command, time::Duration};
use sysinfo::{CpuExt, DiskExt, System, SystemExt}; use sysinfo::{CpuExt, DiskExt, System, SystemExt};
use tokio::time::{interval, sleep}; use tokio::time::{interval, sleep};
// Shared data structures // Data structures matching the C# DTOs
#[derive(Serialize, Debug)]
struct Metric {
timestamp: String,
server_id: i32,
cpu_load: f32,
cpu_temp: f32,
gpu_load: f32,
gpu_temp: f32,
gpu_vram_size: u32,
gpu_vram_usage: u32,
ram_load: f32,
ram_size: u32,
disk_size: u32,
disk_usage: f32,
disk_temp: f32,
net_in: u64,
net_out: u64,
}
#[derive(Serialize, Debug)] #[derive(Serialize, Debug)]
struct RegistrationDto { struct RegistrationDto {
#[serde(rename = "id")] #[serde(rename = "id")]
@@ -42,6 +23,40 @@ struct RegistrationDto {
ram_size: f64, ram_size: f64,
} }
#[derive(Serialize, Debug)]
struct MetricDto {
#[serde(rename = "serverId")]
server_id: i32,
#[serde(rename = "ipAddress")]
ip_address: String,
#[serde(rename = "cpu_Load")]
cpu_load: f64,
#[serde(rename = "cpu_Temp")]
cpu_temp: f64,
#[serde(rename = "gpu_Load")]
gpu_load: f64,
#[serde(rename = "gpu_Temp")]
gpu_temp: f64,
#[serde(rename = "gpu_Vram_Size")]
gpu_vram_size: f64,
#[serde(rename = "gpu_Vram_Usage")]
gpu_vram_usage: f64,
#[serde(rename = "ram_Load")]
ram_load: f64,
#[serde(rename = "ram_Size")]
ram_size: f64,
#[serde(rename = "disk_Size")]
disk_size: f64,
#[serde(rename = "disk_Usage")]
disk_usage: f64,
#[serde(rename = "disk_Temp")]
disk_temp: f64,
#[serde(rename = "net_In")]
net_in: f64,
#[serde(rename = "net_Out")]
net_out: f64,
}
#[derive(Deserialize)] #[derive(Deserialize)]
struct IdResponse { struct IdResponse {
id: i32, id: i32,
@@ -54,7 +69,6 @@ struct HeartbeatPayload {
ip_address: String, ip_address: String,
} }
// Hardware info collection
struct HardwareInfo { struct HardwareInfo {
cpu_type: String, cpu_type: String,
cpu_cores: i32, cpu_cores: i32,
@@ -133,7 +147,6 @@ impl HardwareInfo {
} }
} }
// Registration module
async fn register_with_server(base_url: &str) -> Result<(i32, String), Box<dyn Error>> { async fn register_with_server(base_url: &str) -> Result<(i32, String), Box<dyn Error>> {
let client = Client::builder() let client = Client::builder()
.danger_accept_invalid_certs(true) .danger_accept_invalid_certs(true)
@@ -182,7 +195,6 @@ async fn register_with_server(base_url: &str) -> Result<(i32, String), Box<dyn E
Ok((id, ip_address)) Ok((id, ip_address))
} }
// Heartbeat module
async fn heartbeat_loop(base_url: &str, ip: &str) -> Result<(), Box<dyn Error>> { async fn heartbeat_loop(base_url: &str, ip: &str) -> Result<(), Box<dyn Error>> {
let client = Client::builder() let client = Client::builder()
.danger_accept_invalid_certs(true) .danger_accept_invalid_certs(true)
@@ -206,25 +218,26 @@ async fn heartbeat_loop(base_url: &str, ip: &str) -> Result<(), Box<dyn Error>>
} }
} }
// Metrics module
struct MetricsCollector { struct MetricsCollector {
sys: System, sys: System,
nvml: Option<Nvml>, nvml: Option<Nvml>,
server_id: i32, server_id: i32,
ip_address: String,
} }
impl MetricsCollector { impl MetricsCollector {
fn new(server_id: i32) -> Self { fn new(server_id: i32, ip_address: String) -> Self {
Self { Self {
sys: System::new(), sys: System::new(),
nvml: Nvml::init().ok(), nvml: Nvml::init().ok(),
server_id, server_id,
ip_address,
} }
} }
async fn collect_and_send_loop(&mut self, base_url: &str) -> Result<(), Box<dyn Error>> { async fn collect_and_send_loop(&mut self, base_url: &str) -> Result<(), Box<dyn Error>> {
let client = Client::new(); let client = Client::new();
let url = format!("{}/metric/receive", base_url); let url = format!("{}/monitoring/metric", base_url);
let mut interval = interval(Duration::from_secs(30)); let mut interval = interval(Duration::from_secs(30));
loop { loop {
@@ -233,9 +246,8 @@ impl MetricsCollector {
match client.post(&url).json(&metric).send().await { match client.post(&url).json(&metric).send().await {
Ok(res) => println!( Ok(res) => println!(
"✅ Sent metrics: {} @ {} | Status: {}", "✅ Sent metrics for server {} | Status: {}",
metric.server_id, metric.server_id,
metric.timestamp,
res.status() res.status()
), ),
Err(err) => eprintln!("❌ Failed to send metrics: {}", err), Err(err) => eprintln!("❌ Failed to send metrics: {}", err),
@@ -243,17 +255,18 @@ impl MetricsCollector {
} }
} }
fn collect_metrics(&mut self) -> Metric { fn collect_metrics(&mut self) -> MetricDto {
self.sys.refresh_all(); self.sys.refresh_all();
// CPU // CPU
let cpu_load = self.sys.global_cpu_info().cpu_usage(); let cpu_load = self.sys.global_cpu_info().cpu_usage() as f64;
let cpu_temp = get_cpu_temp().unwrap_or(0.0); let cpu_temp = get_cpu_temp().unwrap_or(0.0) as f64;
// RAM // RAM
let total_memory = self.sys.total_memory(); let total_memory = self.sys.total_memory();
let used_memory = self.sys.used_memory(); let used_memory = self.sys.used_memory();
let ram_load = (used_memory as f32 / total_memory as f32) * 100.0; let ram_load = (used_memory as f64 / total_memory as f64) * 100.0;
let ram_size = (total_memory as f64) / 1024.0 / 1024.0;
// Disk // Disk
let disk = self.sys.disks().first(); let disk = self.sys.disks().first();
@@ -261,11 +274,11 @@ impl MetricsCollector {
let total = d.total_space(); let total = d.total_space();
let available = d.available_space(); let available = d.available_space();
( (
total / 1024 / 1024, (total as f64) / 1024.0 / 1024.0 / 1024.0, // Convert to GB
(total - available) as f32 / total as f32 * 100.0, (total - available) as f64 / total as f64 * 100.0,
) )
} else { } else {
(0, 0.0) (0.0, 0.0)
}; };
// GPU (NVIDIA) // GPU (NVIDIA)
@@ -273,46 +286,56 @@ impl MetricsCollector {
if let Ok(device) = nvml.device_by_index(0) { if let Ok(device) = nvml.device_by_index(0) {
let temp = device let temp = device
.temperature(nvml_wrapper::enum_wrappers::device::TemperatureSensor::Gpu) .temperature(nvml_wrapper::enum_wrappers::device::TemperatureSensor::Gpu)
.unwrap_or(0) as f32; .unwrap_or(0) as f64;
let load = device let load = device
.utilization_rates() .utilization_rates()
.map(|u| u.gpu as f32) .map(|u| u.gpu as f64)
.unwrap_or(0.0); .unwrap_or(0.0);
let mem = device.memory_info().ok(); let mem = device.memory_info().ok();
let used = mem.clone().map(|m| m.used / 1024 / 1024).unwrap_or(0); let used = mem
let total = mem.map(|m| m.total / 1024 / 1024).unwrap_or(0); .clone()
(temp, load, used as u32, total as u32) .map(|m| (m.used as f64) / 1024.0 / 1024.0 / 1024.0)
.unwrap_or(0.0); // GB
let total = mem
.map(|m| (m.total as f64) / 1024.0 / 1024.0 / 1024.0)
.unwrap_or(0.0); // GB
(temp, load, used, total)
} else { } else {
(0.0, 0.0, 0, 0) (0.0, 0.0, 0.0, 0.0)
} }
} else { } else {
(0.0, 0.0, 0, 0) (0.0, 0.0, 0.0, 0.0)
}; };
// Network // Network (convert bytes to bits)
let (net_in, net_out) = get_network_traffic().unwrap_or((0, 0)); let (net_in, net_out) = get_network_traffic().unwrap_or((0, 0));
let net_in_bits = (net_in as f64) * 8.0;
let net_out_bits = (net_out as f64) * 8.0;
Metric { MetricDto {
timestamp: Utc::now().to_rfc3339(),
server_id: self.server_id, server_id: self.server_id,
ip_address: self.ip_address.clone(),
cpu_load, cpu_load,
cpu_temp, cpu_temp,
gpu_load, gpu_load,
gpu_temp, gpu_temp,
gpu_vram_size: vram_total, gpu_vram_size: vram_total,
gpu_vram_usage: vram_used, gpu_vram_usage: if vram_total > 0.0 {
(vram_used / vram_total) * 100.0
} else {
0.0
},
ram_load, ram_load,
ram_size: (total_memory / 1024) as u32, ram_size,
disk_size: disk_size as u32, disk_size,
disk_usage: disk_used, disk_usage: disk_used,
disk_temp: 0.0, disk_temp: 0.0, // not supported
net_in, net_in: net_in_bits,
net_out, net_out: net_out_bits,
} }
} }
} }
// Helper functions
fn get_cpu_temp() -> Option<f32> { fn get_cpu_temp() -> Option<f32> {
let output = Command::new("sensors").output().ok()?; let output = Command::new("sensors").output().ok()?;
let stdout = String::from_utf8_lossy(&output.stdout); let stdout = String::from_utf8_lossy(&output.stdout);
@@ -349,7 +372,6 @@ fn get_network_traffic() -> Option<(u64, u64)> {
Some((rx_total, tx_total)) Some((rx_total, tx_total))
} }
// Main agent
#[tokio::main] #[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> { async fn main() -> Result<(), Box<dyn Error>> {
let server_base_url = "http://localhost:5258"; let server_base_url = "http://localhost:5258";
@@ -370,12 +392,11 @@ async fn main() -> Result<(), Box<dyn Error>> {
// Start metrics collection // Start metrics collection
println!("Starting metrics collection..."); println!("Starting metrics collection...");
let mut metrics_collector = MetricsCollector::new(server_id); let mut metrics_collector = MetricsCollector::new(server_id, ip_address);
metrics_collector metrics_collector
.collect_and_send_loop(server_base_url) .collect_and_send_loop(server_base_url)
.await?; .await?;
// This line is theoretically unreachable because both loops are infinite
heartbeat_handle.await?; heartbeat_handle.await?;
Ok(()) Ok(())
} }