Files
watcheragent/WatcherAgent/src/api.rs
donpat1to d7a58e00da
All checks were successful
Rust Cross-Platform Build / Detect Rust Project (push) Successful in 5s
Rust Cross-Platform Build / Set Tag Name (push) Successful in 5s
Rust Cross-Platform Build / Run Tests (push) Successful in 1m13s
Rust Cross-Platform Build / Build (x86_64-unknown-linux-gnu) (push) Successful in 3m20s
Rust Cross-Platform Build / Build (x86_64-pc-windows-gnu) (push) Successful in 4m7s
Rust Cross-Platform Build / Build and Push Docker Image (push) Successful in 2m22s
Rust Cross-Platform Build / Workflow Summary (push) Successful in 1s
Rust Cross-Platform Build / Create Tag (push) Successful in 5s
added listing all running containers
2025-09-29 14:54:03 +02:00

241 lines
8.5 KiB
Rust

use std::time::Duration;
use crate::hardware::HardwareInfo;
use crate::models::{HeartbeatDto, IdResponse, MetricDto, RegistrationDto, ServerMessage, Acknowledgment};
use crate::docker::serverclientcomm::handle_server_message;
use anyhow::Result;
use reqwest::{Client, StatusCode};
use std::error::Error;
use tokio::time::sleep;
use bollard::Docker;
pub async fn register_with_server(
base_url: &str,
) -> Result<(i32, String), Box<dyn Error + Send + Sync>> {
// First get local IP
let ip = local_ip_address::local_ip()?.to_string();
println!("Local IP address detected: {}", ip);
// Get server ID from backend (this will retry until successful)
let (server_id, registered_ip) = get_server_id_by_ip(base_url, &ip).await?;
// Create HTTP client for registration
let client = Client::builder()
.danger_accept_invalid_certs(true)
.build()?;
// Collect hardware info
let hardware = HardwareInfo::collect().await?;
// Prepare registration data
let registration = RegistrationDto {
id: server_id,
ip_address: registered_ip.clone(),
cpu_type: hardware.cpu.name.clone().unwrap_or_default(),
cpu_cores: (hardware.cpu.cores).unwrap_or_default(),
gpu_type: hardware.gpu.name.clone().unwrap_or_default(),
ram_size: hardware.memory.total.unwrap_or_default(),
};
// Try to register (will retry on failure)
loop {
println!("Attempting to register with server...");
let url = format!("{}/monitoring/register-agent-by-id", base_url);
match client.post(&url).json(&registration).send().await {
Ok(resp) if resp.status().is_success() => {
println!("✅ Successfully registered with server.");
return Ok((server_id, registered_ip));
}
Ok(resp) => {
let status = resp.status();
let text = resp.text().await.unwrap_or_default();
println!(
"⚠️ Registration failed ({}): {} (will retry in 10 seconds)",
status, text
);
}
Err(err) => {
println!("⚠️ Registration error: {} (will retry in 10 seconds)", err);
}
}
sleep(Duration::from_secs(10)).await;
}
}
async fn get_server_id_by_ip(
base_url: &str,
ip: &str,
) -> Result<(i32, String), Box<dyn Error + Send + Sync>> {
let client = Client::builder()
.danger_accept_invalid_certs(true)
.build()?;
let url = format!("{}/monitoring/server-id-by-ip?ipAddress={}", base_url, ip);
loop {
println!("Attempting to fetch server ID for IP {}...", ip);
match client.get(&url).send().await {
Ok(resp) if resp.status().is_success() => {
let text = resp.text().await?;
println!("Raw response: {}", text); // Debug output
let id_resp: IdResponse = serde_json::from_str(&text).map_err(|e| {
println!("Failed to parse response: {}", e);
e
})?;
println!(
"✅ Received ID {} for IP {}",
id_resp.id, id_resp.ip_address
);
return Ok((id_resp.id, id_resp.ip_address));
}
Ok(resp) if resp.status() == StatusCode::NOT_FOUND => {
println!(
"❌ Server with IP {} not found in database (will retry in 10 seconds)",
ip
);
sleep(Duration::from_secs(10)).await;
}
Ok(resp) => {
println!(
"⚠️ Server responded with status: {} - {}",
resp.status(),
resp.text().await?
);
sleep(Duration::from_secs(10)).await;
}
Err(err) => {
println!("⚠️ Request failed: {} (will retry in 10 seconds)", err);
sleep(Duration::from_secs(10)).await;
}
}
}
}
pub async fn heartbeat_loop(base_url: &str, ip: &str) -> Result<(), Box<dyn Error + Send + Sync>> {
let client = Client::builder()
.danger_accept_invalid_certs(true)
.build()?;
let url = format!("{}/heartbeat/receive", base_url);
loop {
let payload = HeartbeatDto {
ip_address: ip.to_string(),
};
match client.post(&url).json(&payload).send().await {
Ok(res) if res.status().is_success() => {
println!("✅ Heartbeat sent successfully.");
}
Ok(res) => eprintln!("Server responded with status: {}", res.status()),
Err(e) => eprintln!("Heartbeat error: {}", e),
}
sleep(Duration::from_secs(20)).await;
}
}
pub async fn send_metrics(
base_url: &str,
metrics: &MetricDto,
) -> Result<(), Box<dyn Error + Send + Sync>> {
let client = Client::new();
let url = format!("{}/monitoring/metric", base_url);
println!("Metrics: {:?}", metrics);
match client.post(&url).json(&metrics).send().await {
Ok(res) => println!(
"✅ Sent metrics for server {} | Status: {}",
metrics.server_id,
res.status()
),
Err(err) => eprintln!("❌ Failed to send metrics: {}", err),
}
Ok(())
}
pub async fn listening_to_server(docker: &Docker, base_url: &str) -> Result<(), Box<dyn Error + Send + Sync>> {
let url = format!("{}/api/message", base_url);
let client = reqwest::Client::new();
loop {
// Get message from server
let resp = client.get(&url).send().await;
match resp {
Ok(response) => {
if response.status().is_success() {
match response.json::<ServerMessage>().await {
Ok(msg) => {
// Acknowledge receipt immediately
if let Err(e) = send_acknowledgment(&client, base_url, &msg.message_id, "received", "Message received successfully").await {
eprintln!("Failed to send receipt acknowledgment: {}", e);
}
// Handle the message
let result = handle_server_message(docker, msg.clone()).await;
// Send execution result acknowledgment
let (status, details) = match result {
Ok(_) => ("success", "Message executed successfully".to_string()),
Err(e) => ("error", format!("Execution failed: {}", e)),
};
if let Err(e) = send_acknowledgment(&client, base_url, &msg.message_id, status, &details).await {
eprintln!("Failed to send execution acknowledgment: {}", e);
}
}
Err(e) => {
eprintln!("Failed to parse message: {}", e);
}
}
} else if response.status() == reqwest::StatusCode::NO_CONTENT {
// No new messages, continue polling
println!("No new messages from server");
} else {
eprintln!("Server returned error status: {}", response.status());
}
}
Err(e) => {
eprintln!("Failed to reach server: {}", e);
}
}
// Poll every 5 seconds (or use WebSocket for real-time)
sleep(Duration::from_secs(5)).await;
}
}
async fn send_acknowledgment(
client: &reqwest::Client,
base_url: &str,
message_id: &str,
status: &str,
details: &str,
) -> Result<(), Box<dyn Error + Send + Sync>> {
let ack_url = format!("{}/api/acknowledge", base_url);
let acknowledgment = Acknowledgment {
message_id: message_id.to_string(),
status: status.to_string(),
details: details.to_string(),
};
let response = client
.post(&ack_url)
.json(&acknowledgment)
.send()
.await?;
if response.status().is_success() {
println!("Acknowledgment sent successfully for message {}", message_id);
} else {
eprintln!("Server returned error for acknowledgment: {}", response.status());
}
Ok(())
}