Compare commits
5 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 3fdcec8047 | |||
| 1fcaf4a670 | |||
| 885e19f7fd | |||
| a7b69b8ae7 | |||
| 2d290f40b2 |
6
Cargo.lock
generated
6
Cargo.lock
generated
@@ -279,7 +279,7 @@ checksum = "a1d728cc89cf3aee9ff92b05e62b19ee65a02b5702cff7d5a377e32c6ae29d8d"
|
||||
|
||||
[[package]]
|
||||
name = "cm-dashboard"
|
||||
version = "0.1.229"
|
||||
version = "0.1.234"
|
||||
dependencies = [
|
||||
"anyhow",
|
||||
"chrono",
|
||||
@@ -301,7 +301,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "cm-dashboard-agent"
|
||||
version = "0.1.229"
|
||||
version = "0.1.234"
|
||||
dependencies = [
|
||||
"anyhow",
|
||||
"async-trait",
|
||||
@@ -325,7 +325,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "cm-dashboard-shared"
|
||||
version = "0.1.229"
|
||||
version = "0.1.234"
|
||||
dependencies = [
|
||||
"chrono",
|
||||
"serde",
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
[package]
|
||||
name = "cm-dashboard-agent"
|
||||
version = "0.1.229"
|
||||
version = "0.1.234"
|
||||
edition = "2021"
|
||||
|
||||
[dependencies]
|
||||
|
||||
@@ -34,6 +34,7 @@ pub struct Agent {
|
||||
collectors: Vec<TimedCollector>,
|
||||
notification_manager: NotificationManager,
|
||||
previous_status: Option<SystemStatus>,
|
||||
cached_agent_data: AgentData,
|
||||
}
|
||||
|
||||
/// Track system component status for change detection
|
||||
@@ -143,6 +144,9 @@ impl Agent {
|
||||
let notification_manager = NotificationManager::new(&config.notifications, &hostname)?;
|
||||
info!("Notification manager initialized");
|
||||
|
||||
// Initialize cached agent data
|
||||
let cached_agent_data = AgentData::new(hostname.clone(), env!("CARGO_PKG_VERSION").to_string());
|
||||
|
||||
Ok(Self {
|
||||
hostname,
|
||||
config,
|
||||
@@ -150,6 +154,7 @@ impl Agent {
|
||||
collectors,
|
||||
notification_manager,
|
||||
previous_status: None,
|
||||
cached_agent_data,
|
||||
})
|
||||
}
|
||||
|
||||
@@ -199,10 +204,8 @@ impl Agent {
|
||||
async fn collect_and_broadcast(&mut self) -> Result<()> {
|
||||
debug!("Starting structured data collection");
|
||||
|
||||
// Initialize empty AgentData
|
||||
let mut agent_data = AgentData::new(self.hostname.clone(), env!("CARGO_PKG_VERSION").to_string());
|
||||
|
||||
// Collect data from collectors whose intervals have elapsed
|
||||
// Update cached_agent_data with new data
|
||||
let now = Instant::now();
|
||||
for timed_collector in &mut self.collectors {
|
||||
let should_collect = match timed_collector.last_collection {
|
||||
@@ -211,7 +214,7 @@ impl Agent {
|
||||
};
|
||||
|
||||
if should_collect {
|
||||
if let Err(e) = timed_collector.collector.collect_structured(&mut agent_data).await {
|
||||
if let Err(e) = timed_collector.collector.collect_structured(&mut self.cached_agent_data).await {
|
||||
error!("Collector {} failed: {}", timed_collector.name, e);
|
||||
// Update last_collection time even on failure to prevent immediate retries
|
||||
timed_collector.last_collection = Some(now);
|
||||
@@ -226,13 +229,22 @@ impl Agent {
|
||||
}
|
||||
}
|
||||
|
||||
// Update timestamp on cached data
|
||||
self.cached_agent_data.timestamp = std::time::SystemTime::now()
|
||||
.duration_since(std::time::UNIX_EPOCH)
|
||||
.unwrap()
|
||||
.as_secs();
|
||||
|
||||
// Clone for notification check (to avoid borrow issues)
|
||||
let agent_data_snapshot = self.cached_agent_data.clone();
|
||||
|
||||
// Check for status changes and send notifications
|
||||
if let Err(e) = self.check_status_changes_and_notify(&agent_data).await {
|
||||
if let Err(e) = self.check_status_changes_and_notify(&agent_data_snapshot).await {
|
||||
error!("Failed to check status changes: {}", e);
|
||||
}
|
||||
|
||||
// Broadcast the structured data via ZMQ
|
||||
if let Err(e) = self.zmq_handler.publish_agent_data(&agent_data).await {
|
||||
// Broadcast the cached structured data via ZMQ
|
||||
if let Err(e) = self.zmq_handler.publish_agent_data(&agent_data_snapshot).await {
|
||||
error!("Failed to broadcast agent data: {}", e);
|
||||
} else {
|
||||
debug!("Successfully broadcast structured agent data");
|
||||
|
||||
@@ -66,6 +66,10 @@ impl DiskCollector {
|
||||
|
||||
/// Collect all storage data and populate AgentData
|
||||
async fn collect_storage_data(&self, agent_data: &mut AgentData) -> Result<(), CollectorError> {
|
||||
// Clear drives and pools to prevent duplicates when updating cached data
|
||||
agent_data.system.storage.drives.clear();
|
||||
agent_data.system.storage.pools.clear();
|
||||
|
||||
// Step 1: Get mount points and their backing devices
|
||||
let mount_devices = self.get_mount_devices().await?;
|
||||
|
||||
|
||||
@@ -200,13 +200,16 @@ impl Collector for MemoryCollector {
|
||||
debug!("Collecting memory metrics");
|
||||
let start = std::time::Instant::now();
|
||||
|
||||
// Clear tmpfs list to prevent duplicates when updating cached data
|
||||
agent_data.system.memory.tmpfs.clear();
|
||||
|
||||
// Parse memory info from /proc/meminfo
|
||||
let info = self.parse_meminfo().await?;
|
||||
|
||||
// Populate memory data directly
|
||||
self.populate_memory_data(&info, agent_data).await?;
|
||||
|
||||
// Collect tmpfs data
|
||||
// Collect tmpfs data
|
||||
self.populate_tmpfs_data(agent_data).await?;
|
||||
|
||||
let duration = start.elapsed();
|
||||
|
||||
@@ -159,6 +159,19 @@ impl SystemdCollector {
|
||||
}
|
||||
}
|
||||
|
||||
if service_name.contains("openvpn-vpn-connection") && status_info.active_state == "active" {
|
||||
if let Some(external_ip) = self.get_vpn_external_ip() {
|
||||
let metrics = Vec::new();
|
||||
|
||||
sub_services.push(SubServiceData {
|
||||
name: format!("IP: {}", external_ip),
|
||||
service_status: Status::Ok,
|
||||
metrics,
|
||||
service_type: "vpn_route".to_string(),
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
// Create complete service data
|
||||
let service_data = ServiceData {
|
||||
name: service_name.clone(),
|
||||
@@ -836,11 +849,45 @@ impl SystemdCollector {
|
||||
_ => value, // Assume bytes if no unit
|
||||
}
|
||||
}
|
||||
|
||||
/// Get VPN external IP by querying through the vpn namespace
|
||||
fn get_vpn_external_ip(&self) -> Option<String> {
|
||||
let output = Command::new("timeout")
|
||||
.args(&[
|
||||
"5",
|
||||
"sudo",
|
||||
"ip",
|
||||
"netns",
|
||||
"exec",
|
||||
"vpn",
|
||||
"curl",
|
||||
"-s",
|
||||
"--max-time",
|
||||
"4",
|
||||
"https://ifconfig.me"
|
||||
])
|
||||
.output()
|
||||
.ok()?;
|
||||
|
||||
if output.status.success() {
|
||||
let ip = String::from_utf8_lossy(&output.stdout).trim().to_string();
|
||||
if !ip.is_empty() && ip.contains('.') {
|
||||
debug!("VPN external IP: {}", ip);
|
||||
return Some(ip);
|
||||
}
|
||||
}
|
||||
|
||||
debug!("Failed to get VPN external IP");
|
||||
None
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl Collector for SystemdCollector {
|
||||
async fn collect_structured(&self, agent_data: &mut AgentData) -> Result<(), CollectorError> {
|
||||
// Clear services to prevent duplicates when updating cached data
|
||||
agent_data.services.clear();
|
||||
|
||||
// Use cached complete data if available and fresh
|
||||
if let Some(cached_complete_services) = self.get_cached_complete_services() {
|
||||
for service_data in cached_complete_services {
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
[package]
|
||||
name = "cm-dashboard"
|
||||
version = "0.1.229"
|
||||
version = "0.1.234"
|
||||
edition = "2021"
|
||||
|
||||
[dependencies]
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
[package]
|
||||
name = "cm-dashboard-shared"
|
||||
version = "0.1.229"
|
||||
version = "0.1.234"
|
||||
edition = "2021"
|
||||
|
||||
[dependencies]
|
||||
|
||||
Reference in New Issue
Block a user