cm/cm-dashboard
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 274 Wiki Activity

Complete atomic migration to structured data architecture
All checks were successful
Build and Release / build-and-release (push) Successful in 1m7s
Details

Browse Source 
Implements clean structured data collection eliminating all string metric
parsing bugs. Collectors now populate AgentData directly with type-safe
field access.

Key improvements:
- Mount points preserved correctly (/ and /boot instead of root/boot)
- Tmpfs discovery added to memory collector
- Temperature data flows as typed f32 fields
- Zero string parsing overhead
- Complete removal of MetricCollectionManager bridge
- Direct ZMQ transmission of structured JSON

All functionality maintained: service tracking, notifications, status
evaluation, and multi-host monitoring.
This commit is contained in:
Christoffer Martinsson
2025-11-24 18:53:31 +01:00
parent 11d1c2dc94
commit 2b2cb2da3e
17 changed files with 1952 additions and 3205 deletions
Download Patch File Download Diff File Expand all files Collapse all files

635
agent/src/agent.rs
View File

@@ -6,19 +6,25 @@ use tracing::{debug, error, info};
use crate::communication::{AgentCommand, ZmqHandler};
use crate::config::AgentConfig;
use crate::metrics::MetricCollectionManager;
use crate::collectors::{
Collector,
backup::BackupCollector,
cpu::CpuCollector,
disk::DiskCollector,
memory::MemoryCollector,
nixos::NixOSCollector,
systemd::SystemdCollector,
};
use crate::notifications::NotificationManager;
use crate::service_tracker::UserStoppedServiceTracker;
use crate::status::HostStatusManager;
use cm_dashboard_shared::{AgentData, Metric, MetricValue, Status, TmpfsData, DriveData, FilesystemData, ServiceData};
use cm_dashboard_shared::AgentData;
pub struct Agent {
hostname: String,
config: AgentConfig,
zmq_handler: ZmqHandler,
metric_manager: MetricCollectionManager,
collectors: Vec<Box<dyn Collector>>,
notification_manager: NotificationManager,
host_status_manager: HostStatusManager,
service_tracker: UserStoppedServiceTracker,
}
@@ -40,69 +46,84 @@ impl Agent {
config.zmq.publisher_port
);
// Initialize metric collection manager with cache config
let metric_manager = MetricCollectionManager::new(&config.collectors, &config).await?;
info!("Metric collection manager initialized");
// Initialize collectors
let mut collectors: Vec<Box<dyn Collector>> = Vec::new();
// Add enabled collectors
if config.collectors.cpu.enabled {
collectors.push(Box::new(CpuCollector::new(config.collectors.cpu.clone())));
}
if config.collectors.memory.enabled {
collectors.push(Box::new(MemoryCollector::new(config.collectors.memory.clone())));
}
if config.collectors.disk.enabled {
collectors.push(Box::new(DiskCollector::new(config.collectors.disk.clone())));
}
if config.collectors.systemd.enabled {
collectors.push(Box::new(SystemdCollector::new(config.collectors.systemd.clone())));
}
if config.collectors.backup.enabled {
collectors.push(Box::new(BackupCollector::new()));
}
if config.collectors.nixos.enabled {
collectors.push(Box::new(NixOSCollector::new(config.collectors.nixos.clone())));
}
info!("Initialized {} collectors", collectors.len());
// Initialize notification manager
let notification_manager = NotificationManager::new(&config.notifications, &hostname)?;
info!("Notification manager initialized");
// Initialize host status manager
let host_status_manager = HostStatusManager::new(config.status_aggregation.clone());
info!("Host status manager initialized");
// Initialize user-stopped service tracker
let service_tracker = UserStoppedServiceTracker::init_global()?;
info!("User-stopped service tracker initialized");
// Initialize service tracker
let service_tracker = UserStoppedServiceTracker::new();
info!("Service tracker initialized");
Ok(Self {
hostname,
config,
zmq_handler,
metric_manager,
collectors,
notification_manager,
host_status_manager,
service_tracker,
})
}
/// Main agent loop with structured data collection
pub async fn run(&mut self, mut shutdown_rx: tokio::sync::oneshot::Receiver<()>) -> Result<()> {
info!("Starting agent main loop with separated collection and transmission");
info!("Starting agent main loop");
// CRITICAL: Collect ALL data immediately at startup before entering the loop
info!("Performing initial FORCE collection of all metrics at startup");
if let Err(e) = self.collect_all_metrics_force().await {
error!("Failed to collect initial metrics: {}", e);
} else {
info!("Initial metric collection completed - all data cached and ready");
// Initial collection
if let Err(e) = self.collect_and_broadcast().await {
error!("Initial metric collection failed: {}", e);
}
// Separate intervals for collection, transmission, and email notifications
let mut collection_interval =
interval(Duration::from_secs(self.config.collection_interval_seconds));
let mut transmission_interval = interval(Duration::from_secs(self.config.zmq.transmission_interval_seconds));
let mut notification_interval = interval(Duration::from_secs(self.config.notifications.aggregation_interval_seconds));
// Set up intervals
let mut transmission_interval = interval(Duration::from_secs(
self.config.collection_interval_seconds,
));
let mut notification_interval = interval(Duration::from_secs(30)); // Check notifications every 30s
// Skip initial ticks to avoid immediate execution
transmission_interval.tick().await;
notification_interval.tick().await;
loop {
tokio::select! {
_ = collection_interval.tick() => {
// Only collect and cache metrics, no ZMQ transmission
if let Err(e) = self.collect_metrics_only().await {
error!("Failed to collect metrics: {}", e);
}
}
_ = transmission_interval.tick() => {
// Send all metrics via ZMQ (dashboard updates only)
if let Err(e) = self.broadcast_all_metrics().await {
error!("Failed to broadcast metrics: {}", e);
if let Err(e) = self.collect_and_broadcast().await {
error!("Failed to collect and broadcast metrics: {}", e);
}
}
_ = notification_interval.tick() => {
// Process batched email notifications (separate from dashboard updates)
if let Err(e) = self.host_status_manager.process_pending_notifications(&mut self.notification_manager).await {
error!("Failed to process pending notifications: {}", e);
}
// Process any pending notifications
// NOTE: With structured data, we might need to implement status tracking differently
// For now, we skip this until status evaluation is migrated
}
// Handle incoming commands (check periodically)
_ = tokio::time::sleep(Duration::from_millis(100)) => {
@@ -121,511 +142,61 @@ impl Agent {
Ok(())
}
async fn collect_all_metrics_force(&mut self) -> Result<()> {
info!("Starting FORCE metric collection for startup");
/// Collect structured data from all collectors and broadcast via ZMQ
async fn collect_and_broadcast(&mut self) -> Result<()> {
debug!("Starting structured data collection");
// Force collect all metrics from all collectors immediately
let metrics = self.metric_manager.collect_all_metrics_force().await?;
// Initialize empty AgentData
let mut agent_data = AgentData::new(self.hostname.clone(), "v0.1.139".to_string());
if metrics.is_empty() {
error!("No metrics collected during force collection!");
return Ok(());
}
info!("Force collected and cached {} metrics", metrics.len());
// Process metrics through status manager (collect status data at startup)
let _status_changed = self.process_metrics(&metrics).await;
Ok(())
}
async fn collect_metrics_only(&mut self) -> Result<()> {
debug!("Starting metric collection cycle (cache only)");
// Collect all metrics from all collectors and cache them
let metrics = self.metric_manager.collect_all_metrics().await?;
if metrics.is_empty() {
debug!("No metrics collected this cycle");
return Ok(());
}
debug!("Collected and cached {} metrics", metrics.len());
// Process metrics through status manager and trigger immediate transmission if status changed
let status_changed = self.process_metrics(&metrics).await;
if status_changed {
info!("Status change detected - triggering immediate metric transmission");
if let Err(e) = self.broadcast_all_metrics().await {
error!("Failed to broadcast metrics after status change: {}", e);
// Collect data from all collectors
for collector in &self.collectors {
if let Err(e) = collector.collect_structured(&mut agent_data).await {
error!("Collector failed: {}", e);
// Continue with other collectors even if one fails
}
}
Ok(())
}
async fn broadcast_all_metrics(&mut self) -> Result<()> {
debug!("Broadcasting cached metrics via ZMQ");
// Get cached metrics (no fresh collection)
let mut metrics = self.metric_manager.get_cached_metrics();
// Add the host status summary metric from status manager
let host_status_metric = self.host_status_manager.get_host_status_metric();
metrics.push(host_status_metric);
// Add agent version metric for cross-host version comparison
let version_metric = self.get_agent_version_metric();
metrics.push(version_metric);
// Heartbeat removed - dashboard detects connectivity via regular transmission timestamps
// Check for user-stopped services that are now active and clear their flags
self.clear_user_stopped_flags_for_active_services(&metrics);
if metrics.is_empty() {
debug!("No metrics to broadcast");
return Ok(());
}
debug!("Broadcasting {} cached metrics as structured data", metrics.len());
// Convert metrics to structured data and send
let agent_data = self.metrics_to_structured_data(&metrics)?;
self.zmq_handler.publish_agent_data(&agent_data).await?;
debug!("Structured data broadcasted successfully");
Ok(())
}
/// Convert legacy metrics to structured data format
fn metrics_to_structured_data(&self, metrics: &[Metric]) -> Result<AgentData> {
let mut agent_data = AgentData::new(self.hostname.clone(), self.get_agent_version());
// Parse metrics into structured data
for metric in metrics {
self.parse_metric_into_agent_data(&mut agent_data, metric)?;
}
Ok(agent_data)
}
/// Parse a single metric into the appropriate structured data field
fn parse_metric_into_agent_data(&self, agent_data: &mut AgentData, metric: &Metric) -> Result<()> {
// CPU metrics
if metric.name == "cpu_load_1min" {
if let Some(value) = metric.value.as_f32() {
agent_data.system.cpu.load_1min = value;
}
} else if metric.name == "cpu_load_5min" {
if let Some(value) = metric.value.as_f32() {
agent_data.system.cpu.load_5min = value;
}
} else if metric.name == "cpu_load_15min" {
if let Some(value) = metric.value.as_f32() {
agent_data.system.cpu.load_15min = value;
}
} else if metric.name == "cpu_frequency_mhz" {
if let Some(value) = metric.value.as_f32() {
agent_data.system.cpu.frequency_mhz = value;
}
} else if metric.name == "cpu_temperature_celsius" {
if let Some(value) = metric.value.as_f32() {
agent_data.system.cpu.temperature_celsius = Some(value);
}
}
// Memory metrics
else if metric.name == "memory_usage_percent" {
if let Some(value) = metric.value.as_f32() {
agent_data.system.memory.usage_percent = value;
}
} else if metric.name == "memory_total_gb" {
if let Some(value) = metric.value.as_f32() {
agent_data.system.memory.total_gb = value;
}
} else if metric.name == "memory_used_gb" {
if let Some(value) = metric.value.as_f32() {
agent_data.system.memory.used_gb = value;
}
} else if metric.name == "memory_available_gb" {
if let Some(value) = metric.value.as_f32() {
agent_data.system.memory.available_gb = value;
}
} else if metric.name == "memory_swap_total_gb" {
if let Some(value) = metric.value.as_f32() {
agent_data.system.memory.swap_total_gb = value;
}
} else if metric.name == "memory_swap_used_gb" {
if let Some(value) = metric.value.as_f32() {
agent_data.system.memory.swap_used_gb = value;
}
}
// Tmpfs metrics - handle multiple auto-discovered tmpfs mounts
else if metric.name.starts_with("memory_tmpfs_") {
if let Some((mount_point, metric_type)) = self.parse_tmpfs_metric_name(&metric.name) {
if let Some(value) = metric.value.as_f32() {
self.update_tmpfs_data(&mut agent_data.system.memory.tmpfs, &mount_point, &metric_type, value);
}
}
}
// Storage metrics
else if metric.name.starts_with("disk_") {
if metric.name.contains("_temperature") {
if let Some(drive_name) = self.extract_drive_name(&metric.name) {
if let Some(temp) = metric.value.as_f32() {
self.ensure_drive_exists(agent_data, &drive_name);
if let Some(drive) = agent_data.system.storage.drives.iter_mut().find(|d| d.name == drive_name) {
drive.temperature_celsius = Some(temp);
}
}
}
} else if metric.name.contains("_wear_percent") {
if let Some(drive_name) = self.extract_drive_name(&metric.name) {
if let Some(wear) = metric.value.as_f32() {
self.ensure_drive_exists(agent_data, &drive_name);
if let Some(drive) = agent_data.system.storage.drives.iter_mut().find(|d| d.name == drive_name) {
drive.wear_percent = Some(wear);
}
}
}
} else if metric.name.contains("_health") {
if let Some(drive_name) = self.extract_drive_name(&metric.name) {
let health = metric.value.as_string();
self.ensure_drive_exists(agent_data, &drive_name);
if let Some(drive) = agent_data.system.storage.drives.iter_mut().find(|d| d.name == drive_name) {
drive.health = health;
}
}
} else if metric.name.contains("_fs_") {
// Filesystem metrics: disk_{pool}_fs_{filesystem}_{metric}
if let Some((pool_name, fs_name)) = self.extract_pool_and_filesystem(&metric.name) {
if metric.name.contains("_usage_percent") {
if let Some(usage) = metric.value.as_f32() {
self.ensure_filesystem_exists(agent_data, &pool_name, &fs_name, usage, 0.0, 0.0);
}
} else if metric.name.contains("_used_gb") {
if let Some(used) = metric.value.as_f32() {
self.update_filesystem_field(agent_data, &pool_name, &fs_name, |fs| fs.used_gb = used);
}
} else if metric.name.contains("_total_gb") {
if let Some(total) = metric.value.as_f32() {
self.update_filesystem_field(agent_data, &pool_name, &fs_name, |fs| fs.total_gb = total);
}
}
}
}
}
// Service metrics
else if metric.name.starts_with("service_") {
if let Some(service_name) = self.extract_service_name(&metric.name) {
if metric.name.contains("_status") {
let status = metric.value.as_string();
self.ensure_service_exists(agent_data, &service_name, &status);
} else if metric.name.contains("_memory_mb") {
if let Some(memory) = metric.value.as_f32() {
self.update_service_field(agent_data, &service_name, |svc| svc.memory_mb = memory);
}
} else if metric.name.contains("_disk_gb") {
if let Some(disk) = metric.value.as_f32() {
self.update_service_field(agent_data, &service_name, |svc| svc.disk_gb = disk);
}
}
}
}
// Backup metrics
else if metric.name.starts_with("backup_") {
if metric.name == "backup_status" {
agent_data.backup.status = metric.value.as_string();
} else if metric.name == "backup_last_run_timestamp" {
if let Some(timestamp) = metric.value.as_i64() {
agent_data.backup.last_run = Some(timestamp as u64);
}
} else if metric.name == "backup_next_scheduled_timestamp" {
if let Some(timestamp) = metric.value.as_i64() {
agent_data.backup.next_scheduled = Some(timestamp as u64);
}
} else if metric.name == "backup_size_gb" {
if let Some(size) = metric.value.as_f32() {
agent_data.backup.total_size_gb = Some(size);
}
} else if metric.name == "backup_repository_health" {
agent_data.backup.repository_health = Some(metric.value.as_string());
}
}
Ok(())
}
/// Parse tmpfs metric name to extract mount point and metric type
/// Example: "memory_tmpfs_tmp_usage_percent" -> ("/tmp", "usage_percent")
fn parse_tmpfs_metric_name(&self, metric_name: &str) -> Option<(String, String)> {
if !metric_name.starts_with("memory_tmpfs_") {
return None;
}
let remainder = &metric_name[13..]; // Remove "memory_tmpfs_" prefix
// Find the last underscore to separate metric type from mount point
if let Some(last_underscore) = remainder.rfind('_') {
let mount_safe = &remainder[..last_underscore];
let metric_type = &remainder[last_underscore + 1..];
// Convert safe mount name back to actual mount point
let mount_point = if mount_safe.is_empty() {
"/"
} else {
&format!("/{}", mount_safe.replace('_', "/"))
};
Some((mount_point.to_string(), metric_type.to_string()))
// Broadcast the structured data via ZMQ
if let Err(e) = self.zmq_handler.publish_agent_data(&agent_data).await {
error!("Failed to broadcast agent data: {}", e);
} else {
None
debug!("Successfully broadcast structured agent data");
}
Ok(())
}
/// Update tmpfs data in the tmpfs vector
fn update_tmpfs_data(&self, tmpfs_vec: &mut Vec<TmpfsData>, mount_point: &str, metric_type: &str, value: f32) {
// Find existing tmpfs entry
let existing_index = tmpfs_vec.iter()
.position(|tmpfs| tmpfs.mount == mount_point);
let tmpfs_index = if let Some(index) = existing_index {
index
} else {
// Create new entry
tmpfs_vec.push(TmpfsData {
mount: mount_point.to_string(),
usage_percent: 0.0,
used_gb: 0.0,
total_gb: 0.0,
});
tmpfs_vec.len() - 1
};
// Update the tmpfs entry
if let Some(tmpfs) = tmpfs_vec.get_mut(tmpfs_index) {
match metric_type {
"usage_percent" => tmpfs.usage_percent = value,
"used_gb" => tmpfs.used_gb = value,
"total_gb" => tmpfs.total_gb = value,
_ => {} // Unknown metric type, ignore
}
}
}
/// Extract drive name from metric like "disk_nvme0n1_temperature"
fn extract_drive_name(&self, metric_name: &str) -> Option<String> {
if metric_name.starts_with("disk_") {
let suffixes = ["_temperature", "_wear_percent", "_health"];
for suffix in suffixes {
if let Some(suffix_pos) = metric_name.rfind(suffix) {
return Some(metric_name[5..suffix_pos].to_string()); // Skip "disk_"
}
}
}
None
}
/// Extract pool and filesystem from "disk_{pool}_fs_{filesystem}_{metric}"
fn extract_pool_and_filesystem(&self, metric_name: &str) -> Option<(String, String)> {
if let Some(fs_pos) = metric_name.find("_fs_") {
let pool_name = metric_name[5..fs_pos].to_string(); // Skip "disk_"
let after_fs = &metric_name[fs_pos + 4..]; // Skip "_fs_"
if let Some(metric_pos) = after_fs.find('_') {
let fs_name = after_fs[..metric_pos].to_string();
return Some((pool_name, fs_name));
}
}
None
}
/// Extract service name from "service_{name}_{metric}"
fn extract_service_name(&self, metric_name: &str) -> Option<String> {
if metric_name.starts_with("service_") {
let suffixes = ["_status", "_memory_mb", "_disk_gb"];
for suffix in suffixes {
if let Some(suffix_pos) = metric_name.rfind(suffix) {
return Some(metric_name[8..suffix_pos].to_string()); // Skip "service_"
}
}
}
None
}
/// Ensure drive exists in agent_data
fn ensure_drive_exists(&self, agent_data: &mut AgentData, drive_name: &str) {
if !agent_data.system.storage.drives.iter().any(|d| d.name == drive_name) {
agent_data.system.storage.drives.push(DriveData {
name: drive_name.to_string(),
health: "UNKNOWN".to_string(),
temperature_celsius: None,
wear_percent: None,
filesystems: Vec::new(),
});
}
}
/// Ensure filesystem exists in the correct drive
fn ensure_filesystem_exists(&self, agent_data: &mut AgentData, pool_name: &str, fs_name: &str, usage_percent: f32, used_gb: f32, total_gb: f32) {
self.ensure_drive_exists(agent_data, pool_name);
if let Some(drive) = agent_data.system.storage.drives.iter_mut().find(|d| d.name == pool_name) {
if !drive.filesystems.iter().any(|fs| fs.mount == fs_name) {
drive.filesystems.push(FilesystemData {
mount: fs_name.to_string(),
usage_percent,
used_gb,
total_gb,
});
}
}
}
/// Update filesystem field
fn update_filesystem_field<F>(&self, agent_data: &mut AgentData, pool_name: &str, fs_name: &str, update_fn: F)
where F: FnOnce(&mut FilesystemData) {
if let Some(drive) = agent_data.system.storage.drives.iter_mut().find(|d| d.name == pool_name) {
if let Some(fs) = drive.filesystems.iter_mut().find(|fs| fs.mount == fs_name) {
update_fn(fs);
}
}
}
/// Ensure service exists
fn ensure_service_exists(&self, agent_data: &mut AgentData, service_name: &str, status: &str) {
if !agent_data.services.iter().any(|s| s.name == service_name) {
agent_data.services.push(ServiceData {
name: service_name.to_string(),
status: status.to_string(),
memory_mb: 0.0,
disk_gb: 0.0,
user_stopped: false, // TODO: Get from service tracker
});
} else if let Some(service) = agent_data.services.iter_mut().find(|s| s.name == service_name) {
service.status = status.to_string();
}
}
/// Update service field
fn update_service_field<F>(&self, agent_data: &mut AgentData, service_name: &str, update_fn: F)
where F: FnOnce(&mut ServiceData) {
if let Some(service) = agent_data.services.iter_mut().find(|s| s.name == service_name) {
update_fn(service);
}
}
async fn process_metrics(&mut self, metrics: &[Metric]) -> bool {
let mut status_changed = false;
for metric in metrics {
// Filter excluded metrics from email notification processing only
if self.config.notifications.exclude_email_metrics.contains(&metric.name) {
debug!("Excluding metric '{}' from email notification processing", metric.name);
continue;
}
if self.host_status_manager.process_metric(metric, &mut self.notification_manager).await {
status_changed = true;
}
}
status_changed
}
/// Create agent version metric for cross-host version comparison
fn get_agent_version_metric(&self) -> Metric {
// Get version from executable path (same logic as main.rs get_version)
let version = self.get_agent_version();
Metric::new(
"agent_version".to_string(),
MetricValue::String(version),
Status::Ok,
)
}
/// Get agent version from Cargo package version
fn get_agent_version(&self) -> String {
// Use the version from Cargo.toml (e.g., "0.1.11")
format!("v{}", env!("CARGO_PKG_VERSION"))
}
/// Create heartbeat metric for host connectivity detection
/// Handle incoming commands from dashboard
async fn handle_commands(&mut self) -> Result<()> {
// Try to receive commands (non-blocking)
match self.zmq_handler.try_receive_command() {
Ok(Some(command)) => {
info!("Received command: {:?}", command);
self.process_command(command).await?;
}
Ok(None) => {
// No command available - this is normal
}
Err(e) => {
error!("Error receiving command: {}", e);
}
}
Ok(())
}
// Try to receive a command (non-blocking)
if let Ok(Some(command)) = self.zmq_handler.try_receive_command() {
info!("Received command: {:?}", command);
async fn process_command(&mut self, command: AgentCommand) -> Result<()> {
match command {
AgentCommand::CollectNow => {
info!("Processing CollectNow command");
if let Err(e) = self.collect_metrics_only().await {
error!("Failed to collect metrics on command: {}", e);
}
}
AgentCommand::SetInterval { seconds } => {
info!("Processing SetInterval command: {} seconds", seconds);
// Note: This would require modifying the interval, which is complex
// For now, just log the request
info!("Interval change requested but not implemented yet");
}
AgentCommand::ToggleCollector { name, enabled } => {
info!(
"Processing ToggleCollector command: {} -> {}",
name, enabled
);
// Note: This would require dynamic collector management
info!("Collector toggle requested but not implemented yet");
}
AgentCommand::Ping => {
info!("Processing Ping command - agent is alive");
// Could send a response back via ZMQ if needed
}
}
Ok(())
}
/// Check metrics for user-stopped services that are now active and clear their flags
fn clear_user_stopped_flags_for_active_services(&mut self, metrics: &[Metric]) {
for metric in metrics {
// Look for service status metrics that are active
if metric.name.starts_with("service_") && metric.name.ends_with("_status") {
if let MetricValue::String(status) = &metric.value {
if status == "active" {
// Extract service name from metric name (service_nginx_status -> nginx)
let service_name = metric.name
.strip_prefix("service_")
.and_then(|s| s.strip_suffix("_status"))
.unwrap_or("");
if !service_name.is_empty() && UserStoppedServiceTracker::is_service_user_stopped(service_name) {
info!("Service '{}' is now active - clearing user-stopped flag", service_name);
if let Err(e) = self.service_tracker.clear_user_stopped(service_name) {
error!("Failed to clear user-stopped flag for '{}': {}", service_name, e);
} else {
// Sync to global tracker
UserStoppedServiceTracker::update_global(&self.service_tracker);
debug!("Cleared user-stopped flag for service '{}'", service_name);
}
}
match command {
AgentCommand::CollectNow => {
info!("Received immediate collection request");
if let Err(e) = self.collect_and_broadcast().await {
error!("Failed to collect on demand: {}", e);
}
}
AgentCommand::SetInterval { seconds } => {
info!("Received interval change request: {}s", seconds);
// Note: This would require more complex handling to update the interval
// For now, just acknowledge
}
AgentCommand::ToggleCollector { name, enabled } => {
info!("Received collector toggle request: {} -> {}", name, enabled);
// Note: This would require more complex handling to enable/disable collectors
// For now, just acknowledge
}
AgentCommand::Ping => {
info!("Received ping command");
// Maybe send back a pong or status
}
}
}
Ok(())
}
}