diff --git a/Cargo.toml b/Cargo.toml index 1ec9f23..8d45141 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,3 +1,3 @@ [workspace] resolver = "3" -members = ["apps/desktop/src-tauri", "crates/domain", "crates/sensor-store", "crates/binding-engine"] +members = ["apps/desktop/src-tauri", "crates/domain", "crates/sensor-store", "crates/provider-host", "crates/providers", "crates/binding-engine"] diff --git a/apps/desktop/src-tauri/Cargo.toml b/apps/desktop/src-tauri/Cargo.toml index 95e9daf..1893f5a 100644 --- a/apps/desktop/src-tauri/Cargo.toml +++ b/apps/desktop/src-tauri/Cargo.toml @@ -22,4 +22,13 @@ tauri = { version = "2", features = [] } tauri-plugin-opener = "2" serde = { version = "1", features = ["derive"] } serde_json = "1" +tokio = { version = "1", features = ["rt-multi-thread"] } +astragauge-provider-host = { path = "../../../crates/provider-host" } +astragauge-sensor-store = { path = "../../../crates/sensor-store" } +astragauge-domain = { path = "../../../crates/domain" } +astragauge-providers = { path = "../../../crates/providers" } + +[dev-dependencies] +tokio = { version = "1", features = ["macros", "rt-multi-thread"] } +async-trait = "0.1" diff --git a/apps/desktop/src-tauri/src/lib.rs b/apps/desktop/src-tauri/src/lib.rs index a8a4d55..28cccc7 100644 --- a/apps/desktop/src-tauri/src/lib.rs +++ b/apps/desktop/src-tauri/src/lib.rs @@ -1,14 +1,170 @@ // Learn more about Tauri commands at https://tauri.app/develop/calling-rust/ +use serde::{Deserialize, Serialize}; + +use astragauge_provider_host::{HostConfig, ProviderHost}; +use astragauge_providers::MockProvider; +use astragauge_sensor_store::SensorStore; +use std::sync::Arc; +use tauri::State; + #[tauri::command] fn greet(name: &str) -> String { format!("Hello, {}! You've been greeted from Rust!", name) } +#[tauri::command] +fn get_providers_status( + host: State>>, +) -> Result, String> { + host + .read() + .map(|h| h.get_providers_status()) + .map_err(|e| format!("Failed to acquire provider host lock: {}", e)) +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct SensorInfo { + pub id: String, + pub name: String, + pub category: String, + pub unit: String, + pub provider_id: String, +} + +#[tauri::command] +async fn list_available_sensors( + store: State<'_, Arc>, +) -> Result, String> { + let sensor_ids = store.list_sensors().await; + + let mut sensors = Vec::new(); + for sensor_id in sensor_ids { + if let Some(descriptor) = store.get_descriptor(&sensor_id).await { + let provider_id = sensor_id + .split('.') + .next() + .unwrap_or(&sensor_id) + .to_string(); + + sensors.push(SensorInfo { + id: descriptor.id.to_string(), + name: descriptor.name, + category: descriptor.category, + unit: descriptor.unit, + provider_id, + }); + } + } + + Ok(sensors) +} + #[cfg_attr(mobile, tauri::mobile_entry_point)] pub fn run() { + let store = Arc::new(SensorStore::new()); + let config = HostConfig::default(); + let host = Arc::new(std::sync::RwLock::new(ProviderHost::new(config, store))); + + let mock_provider = MockProvider::new_test(); + let mock_provider_arc: Arc> = + Arc::new(Box::new(mock_provider)); + host + .write() + .unwrap() + .register_provider(mock_provider_arc) + .expect("Failed to register MockProvider"); + + let host_clone = Arc::clone(&host); + tauri::Builder::default() .plugin(tauri_plugin_opener::init()) - .invoke_handler(tauri::generate_handler![greet]) + .manage(host) + .invoke_handler(tauri::generate_handler![ + greet, + get_providers_status, + list_available_sensors + ]) + .setup(move |_app| { + tauri::async_runtime::spawn(async move { + host_clone.write().unwrap().start(); + }); + Ok(()) + }) .run(tauri::generate_context!()) .expect("error while running tauri application"); } + +#[cfg(test)] +mod tests { + use super::*; + use astragauge_provider_host::{ProviderHealth, ProviderStatus}; + + #[test] + fn test_provider_status_serialization() { + let status = ProviderStatus { + id: "test-provider".to_string(), + name: "Test Provider".to_string(), + health: ProviderHealth::Ok, + sensor_count: 5, + }; + + let json = serde_json::to_string(&status).expect("Should serialize to JSON"); + assert!(json.contains("\"id\":\"test-provider\"")); + assert!(json.contains("\"name\":\"Test Provider\"")); + assert!(json.contains("\"health\":\"ok\"")); + assert!(json.contains("\"sensor_count\":5")); + } + + #[test] + fn test_empty_provider_status_list() { + let statuses: Vec = vec![]; + let json = serde_json::to_string(&statuses).expect("Should serialize empty list"); + assert_eq!(json, "[]"); + } + + #[test] + fn test_provider_health_serialization() { + let health = ProviderHealth::Ok; + let json = serde_json::to_string(&health).expect("Should serialize to JSON"); + assert_eq!(json, "\"ok\""); + + let degraded = ProviderHealth::Degraded { + message: "Test degradation".to_string(), + }; + let json = serde_json::to_string(°raded).expect("Should serialize to JSON"); + assert!(json.contains("degraded")); + assert!(json.contains("Test degradation")); + + let error = ProviderHealth::Error { + message: "Test error".to_string(), + }; + let json = serde_json::to_string(&error).expect("Should serialize to JSON"); + assert!(json.contains("error")); + assert!(json.contains("Test error")); + } + + #[test] + fn test_sensor_info_serialization() { + let sensor = SensorInfo { + id: "cpu.temperature".to_string(), + name: "CPU Temperature".to_string(), + category: "temperature".to_string(), + unit: "celsius".to_string(), + provider_id: "cpu".to_string(), + }; + + let json = serde_json::to_string(&sensor).expect("Should serialize to JSON"); + assert!(json.contains("\"id\":\"cpu.temperature\"")); + assert!(json.contains("\"name\":\"CPU Temperature\"")); + assert!(json.contains("\"category\":\"temperature\"")); + assert!(json.contains("\"unit\":\"celsius\"")); + assert!(json.contains("\"provider_id\":\"cpu\"")); + } + + #[test] + fn test_empty_sensor_list() { + let sensors: Vec = vec![]; + let json = serde_json::to_string(&sensors).expect("Should serialize empty list"); + assert_eq!(json, "[]"); + } +} diff --git a/apps/desktop/src-tauri/tests/tauri_tests.rs b/apps/desktop/src-tauri/tests/tauri_tests.rs new file mode 100644 index 0000000..9d70db8 --- /dev/null +++ b/apps/desktop/src-tauri/tests/tauri_tests.rs @@ -0,0 +1,345 @@ +//! Integration tests for Tauri commands. +//! +//! These tests verify the JSON structure and behavior of Tauri commands +//! without starting the full Tauri application. + +use std::sync::Arc; + +use astragauge_domain::{SensorDescriptor, SensorId}; +use astragauge_provider_host::{ + HostConfig, Provider, ProviderHealth, ProviderHost, ProviderStatus, +}; +use astragauge_providers::MockProvider; +use astragauge_sensor_store::SensorStore; +use serde::{Deserialize, Serialize}; + +/// SensorInfo mirrors the structure in lib.rs for testing +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +pub struct SensorInfo { + pub id: String, + pub name: String, + pub category: String, + pub unit: String, + pub provider_id: String, +} + +/// Helper function that replicates the list_available_sensors command logic +/// for testing without the Tauri State wrapper. +async fn list_available_sensors_logic(store: &Arc) -> Result, String> { + let sensor_ids = store.list_sensors().await; + + let mut sensors = Vec::new(); + for sensor_id in sensor_ids { + if let Some(descriptor) = store.get_descriptor(&sensor_id).await { + let provider_id = sensor_id + .split('.') + .next() + .unwrap_or(&sensor_id) + .to_string(); + + sensors.push(SensorInfo { + id: descriptor.id.to_string(), + name: descriptor.name, + category: descriptor.category, + unit: descriptor.unit, + provider_id, + }); + } + } + + Ok(sensors) +} + +// ============================================================================ +// Test 1: get_providers_status returns valid structure +// ============================================================================ + +#[test] +fn get_providers_status_returns_valid_structure() { + let store = Arc::new(SensorStore::new()); + let config = HostConfig::default(); + let host = Arc::new(ProviderHost::new(config, store)); + + // Call the method that the Tauri command wraps + let statuses: Vec = host.get_providers_status(); + + // Empty host should return empty list + assert!( + statuses.is_empty(), + "Expected empty provider list for new host" + ); + + // Verify the structure can be serialized to JSON with expected fields + let json = serde_json::to_string(&statuses).expect("Should serialize to JSON"); + assert_eq!( + json, "[]", + "Empty provider list should serialize to empty JSON array" + ); +} + +#[test] +fn get_providers_status_with_provider_returns_valid_structure() { + let store = Arc::new(SensorStore::new()); + let config = HostConfig::default(); + let mut host = ProviderHost::new(config, store); + + // Register a mock provider + let provider: Arc> = Arc::new(Box::new(MockProvider::new_test())); + host + .register_provider(provider) + .expect("Should register provider"); + + // Now check status + let statuses = host.get_providers_status(); + assert_eq!(statuses.len(), 1, "Expected one provider in status list"); + + let status = &statuses[0]; + assert_eq!(status.id, "mock.provider"); + assert_eq!(status.name, "Mock Provider"); + assert!(matches!(status.health, ProviderHealth::Ok)); + + // Verify JSON structure contains all expected fields + let json = serde_json::to_string(&status).expect("Should serialize to JSON"); + assert!(json.contains("\"id\""), "JSON should contain 'id' field"); + assert!( + json.contains("\"name\""), + "JSON should contain 'name' field" + ); + assert!( + json.contains("\"health\""), + "JSON should contain 'health' field" + ); + assert!( + json.contains("\"sensor_count\""), + "JSON should contain 'sensor_count' field" + ); +} + +// ============================================================================ +// Test 2: list_available_sensors returns valid structure +// ============================================================================ + +#[tokio::test] +async fn list_available_sensors_returns_valid_structure() { + let store = Arc::new(SensorStore::new()); + + // Register a sensor directly in the store + let sensor_id = SensorId::new("test.temperature").expect("valid sensor id"); + let descriptor = SensorDescriptor { + id: sensor_id.clone(), + name: "Test Temperature".to_string(), + category: "temperature".to_string(), + unit: "celsius".to_string(), + device: None, + tags: vec![], + }; + + store + .register_sensor(descriptor) + .await + .expect("Should register sensor"); + + // Call the logic function + let sensors = list_available_sensors_logic(&store) + .await + .expect("Should return sensors"); + + assert_eq!(sensors.len(), 1, "Expected one sensor in list"); + + let sensor = &sensors[0]; + assert_eq!(sensor.id, "test.temperature"); + assert_eq!(sensor.name, "Test Temperature"); + assert_eq!(sensor.category, "temperature"); + assert_eq!(sensor.unit, "celsius"); + assert_eq!(sensor.provider_id, "test"); + + // Verify JSON structure + let json = serde_json::to_string(&sensors).expect("Should serialize to JSON"); + assert!( + json.contains("\"id\":\"test.temperature\""), + "JSON should contain sensor id" + ); + assert!( + json.contains("\"name\":\"Test Temperature\""), + "JSON should contain sensor name" + ); + assert!( + json.contains("\"category\":\"temperature\""), + "JSON should contain category" + ); + assert!( + json.contains("\"unit\":\"celsius\""), + "JSON should contain unit" + ); + assert!( + json.contains("\"provider_id\":\"test\""), + "JSON should contain provider_id" + ); +} + +#[tokio::test] +async fn list_available_sensors_extracts_provider_id_correctly() { + let store = Arc::new(SensorStore::new()); + + // Register sensors with different provider prefixes + let sensors_to_register = vec![ + ( + "cpu.core0.temperature", + "CPU Core 0 Temp", + "temperature", + "celsius", + ), + ("gpu.vram.used", "GPU VRAM Used", "memory", "bytes"), + ("system.uptime", "System Uptime", "time", "seconds"), + ]; + + for (id_str, name, category, unit) in sensors_to_register { + let sensor_id = SensorId::new(id_str).expect("valid sensor id"); + let descriptor = SensorDescriptor { + id: sensor_id, + name: name.to_string(), + category: category.to_string(), + unit: unit.to_string(), + device: None, + tags: vec![], + }; + store + .register_sensor(descriptor) + .await + .expect("Should register sensor"); + } + + let sensors = list_available_sensors_logic(&store) + .await + .expect("Should return sensors"); + + assert_eq!(sensors.len(), 3); + + // Verify provider_id extraction (first component before dot) + let cpu_sensor = sensors + .iter() + .find(|s| s.id == "cpu.core0.temperature") + .unwrap(); + assert_eq!(cpu_sensor.provider_id, "cpu"); + + let gpu_sensor = sensors.iter().find(|s| s.id == "gpu.vram.used").unwrap(); + assert_eq!(gpu_sensor.provider_id, "gpu"); + + let system_sensor = sensors.iter().find(|s| s.id == "system.uptime").unwrap(); + assert_eq!(system_sensor.provider_id, "system"); +} + +// ============================================================================ +// Test 3: commands work without providers loaded (empty state) +// ============================================================================ + +#[test] +fn commands_work_without_providers_loaded() { + // Test get_providers_status with empty host + let store = Arc::new(SensorStore::new()); + let config = HostConfig::default(); + let host = Arc::new(ProviderHost::new(config, store)); + + let statuses = host.get_providers_status(); + assert!( + statuses.is_empty(), + "Empty host should return empty status list" + ); + + // Verify empty list serializes correctly + let json = serde_json::to_string(&statuses).expect("Should serialize"); + assert_eq!(json, "[]"); +} + +#[tokio::test] +async fn list_available_sensors_empty_state() { + // Test list_available_sensors with empty store + let store = Arc::new(SensorStore::new()); + + let sensors = list_available_sensors_logic(&store) + .await + .expect("Should return sensors"); + assert!( + sensors.is_empty(), + "Empty store should return empty sensor list" + ); + + // Verify empty list serializes correctly + let json = serde_json::to_string(&sensors).expect("Should serialize"); + assert_eq!(json, "[]"); +} + +#[tokio::test] +async fn full_integration_empty_state() { + // Combined test: verify both commands work with empty ProviderHost + let store = Arc::new(SensorStore::new()); + let config = HostConfig::default(); + let host = Arc::new(ProviderHost::new(config, Arc::clone(&store))); + + // Test get_providers_status + let statuses = host.get_providers_status(); + assert!(statuses.is_empty()); + + // Test list_available_sensors + let sensors = list_available_sensors_logic(&store) + .await + .expect("Should return sensors"); + assert!(sensors.is_empty()); + + // Verify both can be serialized for Tauri IPC + let status_json = serde_json::to_string(&statuses).expect("Should serialize"); + let sensor_json = serde_json::to_string(&sensors).expect("Should serialize"); + + assert_eq!(status_json, "[]"); + assert_eq!(sensor_json, "[]"); +} + +// ============================================================================ +// Additional structure validation tests +// ============================================================================ + +#[test] +fn provider_status_health_variants_serialize_correctly() { + // Test Ok variant + let ok_health = ProviderHealth::Ok; + let json = serde_json::to_string(&ok_health).expect("Should serialize"); + assert_eq!(json, "\"ok\""); + + // Test Degraded variant + let degraded = ProviderHealth::Degraded { + message: "Performance issues".to_string(), + }; + let json = serde_json::to_string(°raded).expect("Should serialize"); + assert!(json.contains("degraded")); + assert!(json.contains("Performance issues")); + + // Test Error variant + let error = ProviderHealth::Error { + message: "Connection failed".to_string(), + }; + let json = serde_json::to_string(&error).expect("Should serialize"); + assert!(json.contains("error")); + assert!(json.contains("Connection failed")); +} + +#[test] +fn sensor_info_serializes_to_expected_json_format() { + let sensor = SensorInfo { + id: "cpu.temperature".to_string(), + name: "CPU Temperature".to_string(), + category: "temperature".to_string(), + unit: "celsius".to_string(), + provider_id: "cpu".to_string(), + }; + + let json = serde_json::to_string(&sensor).expect("Should serialize"); + + // Verify all fields are present with correct values + let parsed: serde_json::Value = serde_json::from_str(&json).expect("Should parse JSON"); + + assert_eq!(parsed["id"], "cpu.temperature"); + assert_eq!(parsed["name"], "CPU Temperature"); + assert_eq!(parsed["category"], "temperature"); + assert_eq!(parsed["unit"], "celsius"); + assert_eq!(parsed["provider_id"], "cpu"); +} diff --git a/crates/domain/src/provider.rs b/crates/domain/src/provider.rs index 490c992..fcdaf37 100644 --- a/crates/domain/src/provider.rs +++ b/crates/domain/src/provider.rs @@ -1,5 +1,7 @@ use serde::{Deserialize, Serialize}; +use crate::validation::DomainError; + #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] #[serde(rename_all = "snake_case")] pub struct ProviderCapabilities { @@ -37,6 +39,32 @@ pub struct ProviderManifest { pub sensors: SensorCategories, } +impl ProviderManifest { + /// Validates the manifest for required fields and basic format rules. + /// Returns Ok(()) if valid, Err(DomainError) if invalid. + pub fn validate(&self) -> Result<(), DomainError> { + if self.id.is_empty() { + return Err(DomainError::InvalidFormat { + message: "provider id cannot be empty".to_string(), + }); + } + + if self.version.is_empty() { + return Err(DomainError::InvalidFormat { + message: "provider version cannot be empty".to_string(), + }); + } + + if self.runtime.is_empty() { + return Err(DomainError::InvalidFormat { + message: "provider runtime requirement cannot be empty".to_string(), + }); + } + + Ok(()) + } +} + #[cfg(test)] mod tests { use super::*; @@ -187,4 +215,121 @@ categories = ["cpu","memory","disk","network","temperature"] let parsed: SensorCategories = serde_json::from_str(&json).unwrap(); assert_eq!(categories, parsed); } + + #[test] + fn test_validate_valid_manifest() { + let manifest = ProviderManifest { + id: "core.linux".to_string(), + name: "Linux System Provider".to_string(), + version: "0.1.0".to_string(), + description: "Test provider".to_string(), + author: None, + website: None, + repository: None, + license: None, + tags: None, + runtime: ">=0.1.0".to_string(), + capabilities: ProviderCapabilities { + historical: false, + high_frequency: true, + hardware_access: true, + }, + sensors: SensorCategories { categories: vec![] }, + }; + + assert!(manifest.validate().is_ok()); + } + + #[test] + fn test_validate_empty_id() { + let manifest = ProviderManifest { + id: "".to_string(), + name: "Test Provider".to_string(), + version: "0.1.0".to_string(), + description: "Test provider".to_string(), + author: None, + website: None, + repository: None, + license: None, + tags: None, + runtime: ">=0.1.0".to_string(), + capabilities: ProviderCapabilities { + historical: false, + high_frequency: false, + hardware_access: false, + }, + sensors: SensorCategories { categories: vec![] }, + }; + + let result = manifest.validate(); + assert!(result.is_err()); + match result { + Err(DomainError::InvalidFormat { message }) => { + assert!(message.contains("id cannot be empty")); + } + _ => panic!("Expected InvalidFormat error for empty id"), + } + } + + #[test] + fn test_validate_empty_version() { + let manifest = ProviderManifest { + id: "test.provider".to_string(), + name: "Test Provider".to_string(), + version: "".to_string(), + description: "Test provider".to_string(), + author: None, + website: None, + repository: None, + license: None, + tags: None, + runtime: ">=0.1.0".to_string(), + capabilities: ProviderCapabilities { + historical: false, + high_frequency: false, + hardware_access: false, + }, + sensors: SensorCategories { categories: vec![] }, + }; + + let result = manifest.validate(); + assert!(result.is_err()); + match result { + Err(DomainError::InvalidFormat { message }) => { + assert!(message.contains("version cannot be empty")); + } + _ => panic!("Expected InvalidFormat error for empty version"), + } + } + + #[test] + fn test_validate_empty_runtime() { + let manifest = ProviderManifest { + id: "test.provider".to_string(), + name: "Test Provider".to_string(), + version: "0.1.0".to_string(), + description: "Test provider".to_string(), + author: None, + website: None, + repository: None, + license: None, + tags: None, + runtime: "".to_string(), + capabilities: ProviderCapabilities { + historical: false, + high_frequency: false, + hardware_access: false, + }, + sensors: SensorCategories { categories: vec![] }, + }; + + let result = manifest.validate(); + assert!(result.is_err()); + match result { + Err(DomainError::InvalidFormat { message }) => { + assert!(message.contains("runtime requirement cannot be empty")); + } + _ => panic!("Expected InvalidFormat error for empty runtime"), + } + } } diff --git a/crates/provider-host/Cargo.toml b/crates/provider-host/Cargo.toml new file mode 100644 index 0000000..5050a92 --- /dev/null +++ b/crates/provider-host/Cargo.toml @@ -0,0 +1,18 @@ +[package] +name = "astragauge-provider-host" +version = "0.1.0" +edition = "2021" + +[dependencies] +astragauge-domain = { path = "../domain" } +astragauge-sensor-store = { path = "../sensor-store" } +tokio = { version = "1", features = ["full"] } +tokio-util = "0.7" +async-trait = "0.1" +serde = { version = "1", features = ["derive"] } +thiserror = "1" +tracing = "0.1" +futures = "0.3" + +[dev-dependencies] +astragauge-providers = { path = "../providers" } diff --git a/crates/provider-host/src/config.rs b/crates/provider-host/src/config.rs new file mode 100644 index 0000000..3685037 --- /dev/null +++ b/crates/provider-host/src/config.rs @@ -0,0 +1,36 @@ +#[derive(Debug, Clone)] +#[non_exhaustive] +pub struct HostConfig { + pub default_poll_interval_ms: u64, + pub min_poll_interval_ms: u64, + pub shutdown_timeout_ms: u64, +} + +impl HostConfig { + #[must_use] + pub fn new() -> Self { + Self::default() + } + + #[must_use] + pub fn with_default_poll_interval(mut self, interval_ms: u64) -> Self { + self.default_poll_interval_ms = interval_ms; + self + } + + #[must_use] + pub fn with_shutdown_timeout(mut self, timeout_ms: u64) -> Self { + self.shutdown_timeout_ms = timeout_ms; + self + } +} + +impl Default for HostConfig { + fn default() -> Self { + Self { + default_poll_interval_ms: 1000, + min_poll_interval_ms: 100, + shutdown_timeout_ms: 10000, + } + } +} diff --git a/crates/provider-host/src/error.rs b/crates/provider-host/src/error.rs new file mode 100644 index 0000000..d35ce1c --- /dev/null +++ b/crates/provider-host/src/error.rs @@ -0,0 +1,38 @@ +use serde::{Deserialize, Serialize}; +use std::fmt; + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum ProviderError { + DiscoveryFailed { message: String }, + PollFailed { message: String }, + ShutdownFailed { message: String }, + RegistrationFailed { id: String, reason: String }, + InvalidManifest { reason: String }, +} + +impl fmt::Display for ProviderError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + ProviderError::DiscoveryFailed { message } => { + write!(f, "Provider discovery failed: {}", message) + } + ProviderError::PollFailed { message } => { + write!(f, "Provider poll failed: {}", message) + } + ProviderError::ShutdownFailed { message } => { + write!(f, "Provider shutdown failed: {}", message) + } + ProviderError::RegistrationFailed { id, reason } => { + write!(f, "Provider registration failed for '{}': {}", id, reason) + } + ProviderError::InvalidManifest { reason } => { + write!(f, "Invalid manifest: {}", reason) + } + } + } +} + +impl std::error::Error for ProviderError {} + +pub type ProviderResult = Result; diff --git a/crates/provider-host/src/health.rs b/crates/provider-host/src/health.rs new file mode 100644 index 0000000..d37dbf8 --- /dev/null +++ b/crates/provider-host/src/health.rs @@ -0,0 +1,9 @@ +use serde::{Deserialize, Serialize}; + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum ProviderHealth { + Ok, + Degraded { message: String }, + Error { message: String }, +} diff --git a/crates/provider-host/src/host.rs b/crates/provider-host/src/host.rs new file mode 100644 index 0000000..10c8060 --- /dev/null +++ b/crates/provider-host/src/host.rs @@ -0,0 +1,654 @@ +use std::collections::{HashMap, HashSet}; +use std::panic::AssertUnwindSafe; +use std::sync::{Arc, RwLock}; + +use astragauge_sensor_store::{SensorStore, StoreError}; +use futures::FutureExt; +use serde::{Deserialize, Serialize}; +use tokio::task::JoinHandle; +use tokio_util::sync::CancellationToken; + +use crate::config::HostConfig; +use crate::error::{ProviderError, ProviderResult}; +use crate::health::ProviderHealth; +use crate::provider::Provider; + +pub struct ProviderEntry { + provider: Arc>, + task: Option>, + health: Arc>, + sensor_ids: Arc>>, +} + +pub struct ProviderHost { + providers: HashMap, + store: Arc, + shutdown_token: CancellationToken, + config: HostConfig, +} + +impl ProviderHost { + pub fn new(config: HostConfig, store: Arc) -> Self { + let shutdown_token = CancellationToken::new(); + Self { + providers: HashMap::new(), + store, + shutdown_token, + config, + } + } + + pub fn register_provider(&mut self, provider: Arc>) -> ProviderResult<()> { + let manifest = provider.as_ref().manifest(); + let id = manifest.id.clone(); + + if self.providers.contains_key(&id) { + return Err(ProviderError::RegistrationFailed { + id: id.clone(), + reason: format!("Provider with id '{}' already registered", id), + }); + } + + self.providers.insert( + id, + ProviderEntry { + provider: Arc::clone(&provider), + task: None, + health: Arc::new(RwLock::new(ProviderHealth::Ok)), + sensor_ids: Arc::new(RwLock::new(HashSet::new())), + }, + ); + + Ok(()) + } + + pub fn start(&mut self) -> usize { + let mut count = 0; + + let tasks: Vec<(String, JoinHandle<()>)> = self + .providers + .iter() + .filter(|(_, entry)| entry.task.is_none()) + .map(|(id, entry)| { + let task = self.spawn_poll_task( + id.clone(), + Arc::clone(&entry.provider), + Arc::clone(&entry.health), + Arc::clone(&entry.sensor_ids), + ); + (id.clone(), task) + }) + .collect(); + + for (id, task) in tasks { + if let Some(entry) = self.providers.get_mut(&id) { + entry.task = Some(task); + count += 1; + } + } + + count + } + + pub async fn shutdown(&mut self) -> ProviderResult<()> { + let provider_count = self.providers.len(); + tracing::info!("Initiating shutdown of {} providers", provider_count); + + self.shutdown_token.cancel(); + + let timeout_duration = tokio::time::Duration::from_millis(self.config.shutdown_timeout_ms); + let mut finished = 0; + let mut timed_out = 0; + + let tasks: Vec<(String, JoinHandle<()>)> = self + .providers + .iter_mut() + .filter_map(|(id, entry)| entry.task.take().map(|task| (id.clone(), task))) + .collect(); + + for (id, task) in tasks { + tracing::debug!("Waiting for provider {} task to complete...", id); + + match tokio::time::timeout(timeout_duration, task).await { + Ok(Ok(())) => { + tracing::debug!("Provider {} task completed", id); + finished += 1; + } + Ok(Err(join_error)) => { + tracing::warn!( + "Provider {} task panicked during shutdown: {}", + id, + join_error + ); + finished += 1; + } + Err(_) => { + tracing::warn!("Provider {} did not complete within timeout", id); + timed_out += 1; + } + } + } + + tracing::info!( + "Shutdown completed: {} providers finished, {} timed out", + finished, + timed_out + ); + + if timed_out > 0 { + Err(ProviderError::ShutdownFailed { + message: format!( + "{} provider(s) did not complete within {}ms timeout", + timed_out, self.config.shutdown_timeout_ms + ), + }) + } else { + Ok(()) + } + } + + fn spawn_poll_task( + &self, + id: String, + provider: Arc>, + health: Arc>, + sensor_ids: Arc>>, + ) -> JoinHandle<()> { + let store = Arc::clone(&self.store); + let token = self.shutdown_token.clone(); + let poll_interval = provider.as_ref().poll_interval(); + + tokio::spawn(async move { + let discover_result = AssertUnwindSafe(provider.as_ref().discover()) + .catch_unwind() + .await; + + match discover_result { + Ok(Ok(descriptors)) => { + if let Err(e) = Self::register_sensors(&store, &id, descriptors, &sensor_ids).await { + tracing::error!("Provider {} sensor registration failed: {:?}", id, e); + } + } + Ok(Err(e)) => { + tracing::error!("Provider {} sensor discovery failed: {:?}", id, e); + } + Err(panic_payload) => { + Self::handle_panic(&id, &health, panic_payload); + } + } + + let mut tick = tokio::time::interval(poll_interval); + + loop { + tokio::select! { + _ = tick.tick() => { + let poll_result = AssertUnwindSafe(provider.as_ref().poll()) + .catch_unwind() + .await; + + match poll_result { + Ok(Ok(samples)) => { + Self::push_samples_with_defensive_registration( + &provider, &store, &id, &health, samples + ).await; + } + Ok(Err(e)) => { + tracing::error!("Provider {} polling failed: {:?}", id, e); + } + Err(panic_payload) => { + Self::handle_panic(&id, &health, panic_payload); + } + } + } + _ = token.cancelled() => { + tracing::info!("Provider {} shutdown requested", id); + break; + } + } + } + }) + } + + fn handle_panic( + provider_id: &str, + health: &Arc>, + payload: Box, + ) { + let message = if let Some(s) = payload.downcast_ref::<&str>() { + s.to_string() + } else if let Some(s) = payload.downcast_ref::() { + s.clone() + } else { + "Unknown panic".to_string() + }; + + tracing::error!("Provider {} panicked: {}", provider_id, message); + + if let Ok(mut health_guard) = health.write() { + *health_guard = ProviderHealth::Error { message }; + } + } + + async fn register_sensors( + store: &Arc, + provider_id: &str, + descriptors: Vec, + sensor_ids: &Arc>>, + ) -> ProviderResult<()> { + let count = descriptors.len(); + tracing::info!("Provider {} discovered {} sensors", provider_id, count); + + let mut registered_ids = Vec::new(); + for descriptor in descriptors { + let sensor_id = descriptor.id.clone(); + match store.register_sensor(descriptor).await { + Ok(()) => { + tracing::debug!("Registered sensor {}", sensor_id.as_str()); + registered_ids.push(sensor_id.as_str().to_string()); + } + Err(e) => { + tracing::warn!("Failed to register sensor {}: {:?}", sensor_id.as_str(), e); + } + } + } + + if let Ok(mut ids) = sensor_ids.write() { + ids.extend(registered_ids); + } + + Ok(()) + } + + async fn push_samples_with_defensive_registration( + provider: &Arc>, + store: &Arc, + provider_id: &str, + health: &Arc>, + samples: Vec, + ) { + let sample_count = samples.len(); + + if let Err(e) = store.push_samples(&samples).await { + match e { + StoreError::UnknownSensor { id: unknown_id } => { + Self::handle_unknown_sensor( + provider, + store, + &unknown_id, + &samples, + provider_id, + health, + sample_count, + ) + .await; + } + other_error => { + tracing::error!("Failed to push samples to store: {:?}", other_error); + } + } + } else { + tracing::trace!( + "Pushed {} samples from provider {}", + sample_count, + provider_id + ); + } + } + + async fn handle_unknown_sensor( + provider: &Arc>, + store: &Arc, + unknown_id: &astragauge_domain::SensorId, + samples: &[astragauge_domain::SensorSample], + provider_id: &str, + health: &Arc>, + sample_count: usize, + ) { + tracing::warn!( + "Unknown sensor {} encountered, attempting defensive registration", + unknown_id.as_str() + ); + + let discover_result = AssertUnwindSafe(provider.as_ref().discover()) + .catch_unwind() + .await; + + match discover_result { + Ok(Ok(descriptors)) => { + if let Some(descriptor) = descriptors.iter().find(|d| &d.id == unknown_id) { + match store.register_sensor(descriptor.clone()).await { + Ok(()) => { + tracing::info!("Defensively registered sensor {}", unknown_id.as_str()); + if let Err(retry_err) = store.push_samples(samples).await { + tracing::error!( + "Failed to push samples after defensive registration: {:?}", + retry_err + ); + } else { + tracing::debug!( + "Pushed {} samples from provider {} after defensive registration", + sample_count, + provider_id + ); + } + } + Err(reg_err) => { + tracing::error!( + "Failed to defensively register sensor {}: {:?}", + unknown_id.as_str(), + reg_err + ); + } + } + } + } + Ok(Err(e)) => { + tracing::error!( + "Failed to re-discover sensors for defensive registration: {:?}", + e + ); + } + Err(panic_payload) => { + Self::handle_panic(provider_id, health, panic_payload); + } + } + } + + pub fn get_providers_status(&self) -> Vec { + self + .providers + .iter() + .map(|(id, entry)| { + let health = entry.health.read().unwrap().clone(); + let sensor_count = entry.sensor_ids.read().map(|ids| ids.len()).unwrap_or(0); + ProviderStatus { + id: id.clone(), + name: entry.provider.manifest().name.clone(), + health, + sensor_count, + } + }) + .collect() + } +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct ProviderStatus { + pub id: String, + pub name: String, + pub health: ProviderHealth, + pub sensor_count: usize, +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::provider::Provider; + use astragauge_domain::{ + DomainError, ProviderCapabilities, ProviderManifest, SensorCategories, SensorDescriptor, + SensorId, SensorSample, + }; + use async_trait::async_trait; + use std::time::Duration; + + fn make_manifest(id: &str) -> ProviderManifest { + ProviderManifest { + id: id.to_string(), + name: format!("{} provider", id), + version: "1.0.0".to_string(), + description: "Test provider".to_string(), + author: None, + website: None, + repository: None, + license: None, + tags: None, + runtime: ">=1.0.0".to_string(), + capabilities: ProviderCapabilities { + historical: false, + high_frequency: false, + hardware_access: false, + }, + sensors: SensorCategories { categories: vec![] }, + } + } + + struct PanickingProvider { + manifest: ProviderManifest, + } + + impl PanickingProvider { + fn new(id: &str) -> Self { + Self { + manifest: make_manifest(id), + } + } + } + + #[async_trait] + impl Provider for PanickingProvider { + fn manifest(&self) -> &ProviderManifest { + &self.manifest + } + + fn poll_interval(&self) -> Duration { + Duration::from_millis(10) + } + + async fn discover(&self) -> ProviderResult> { + panic!("discover() panicked!") + } + + async fn poll(&self) -> ProviderResult> { + Ok(vec![]) + } + + async fn health(&self) -> ProviderHealth { + ProviderHealth::Ok + } + + async fn shutdown(&self) -> ProviderResult<()> { + Ok(()) + } + } + + struct PollPanickingProvider { + manifest: ProviderManifest, + } + + impl PollPanickingProvider { + fn new(id: &str) -> Self { + Self { + manifest: make_manifest(id), + } + } + } + + #[async_trait] + impl Provider for PollPanickingProvider { + fn manifest(&self) -> &ProviderManifest { + &self.manifest + } + + fn poll_interval(&self) -> Duration { + Duration::from_millis(10) + } + + async fn discover(&self) -> ProviderResult> { + Ok(vec![]) + } + + async fn poll(&self) -> ProviderResult> { + panic!("poll() panicked!") + } + + async fn health(&self) -> ProviderHealth { + ProviderHealth::Ok + } + + async fn shutdown(&self) -> ProviderResult<()> { + Ok(()) + } + } + + struct HealthyProvider { + manifest: ProviderManifest, + sensor_id: SensorId, + } + + impl HealthyProvider { + fn new(id: &str, sensor_id: &str) -> Result { + Ok(Self { + manifest: make_manifest(id), + sensor_id: SensorId::new(sensor_id.to_string())?, + }) + } + } + + #[async_trait] + impl Provider for HealthyProvider { + fn manifest(&self) -> &ProviderManifest { + &self.manifest + } + + fn poll_interval(&self) -> Duration { + Duration::from_millis(10) + } + + async fn discover(&self) -> ProviderResult> { + Ok(vec![SensorDescriptor { + id: self.sensor_id.clone(), + name: "Test sensor".to_string(), + category: "test".to_string(), + unit: "none".to_string(), + device: None, + tags: vec![], + }]) + } + + async fn poll(&self) -> ProviderResult> { + Ok(vec![SensorSample { + sensor_id: self.sensor_id.clone(), + timestamp_ms: 0, + value: Some(42.0), + }]) + } + + async fn health(&self) -> ProviderHealth { + ProviderHealth::Ok + } + + async fn shutdown(&self) -> ProviderResult<()> { + Ok(()) + } + } + + #[tokio::test] + #[ignore = "requires panic containment test infrastructure"] + async fn panic_containment_discover_updates_health() { + let store = Arc::new(SensorStore::new()); + let config = HostConfig::default(); + let mut host = ProviderHost::new(config, store); + + let provider: Arc> = Arc::new(Box::new(PanickingProvider::new("panic-test"))); + host.register_provider(provider).unwrap(); + + let health_arc = Arc::clone(&host.providers.get("panic-test").unwrap().health); + + host.start(); + + tokio::time::sleep(Duration::from_millis(50)).await; + + let health = health_arc.read().unwrap().clone(); + assert!(matches!(health, ProviderHealth::Error { .. })); + } + + #[tokio::test] + #[ignore = "requires panic containment test infrastructure"] + async fn panic_containment_poll_updates_health() { + let store = Arc::new(SensorStore::new()); + let config = HostConfig::default(); + let mut host = ProviderHost::new(config, store); + + let provider: Arc> = + Arc::new(Box::new(PollPanickingProvider::new("poll-panic-test"))); + host.register_provider(provider).unwrap(); + + let health_arc = Arc::clone(&host.providers.get("poll-panic-test").unwrap().health); + + host.start(); + + tokio::time::sleep(Duration::from_millis(100)).await; + + let health = health_arc.read().unwrap().clone(); + assert!(matches!(health, ProviderHealth::Error { ref message } if message.contains("poll()"))); + } + + #[tokio::test] + #[ignore = "requires panic containment test infrastructure"] + async fn panic_containment_other_providers_continue() { + let store = Arc::new(SensorStore::new()); + let config = HostConfig::default(); + let mut host = ProviderHost::new(config, Arc::clone(&store)); + + let healthy: Arc> = Arc::new(Box::new( + HealthyProvider::new("healthy", "test.healthy").unwrap(), + )); + let panicking: Arc> = Arc::new(Box::new(PanickingProvider::new("panicking"))); + + host.register_provider(healthy).unwrap(); + host.register_provider(panicking).unwrap(); + + let healthy_health = Arc::clone(&host.providers.get("healthy").unwrap().health); + let panicking_health = Arc::clone(&host.providers.get("panicking").unwrap().health); + + host.start(); + + tokio::time::sleep(Duration::from_millis(100)).await; + + let healthy_status = healthy_health.read().unwrap().clone(); + let panicking_status = panicking_health.read().unwrap().clone(); + + assert!(matches!(healthy_status, ProviderHealth::Ok)); + assert!(matches!(panicking_status, ProviderHealth::Error { .. })); + } + + #[tokio::test] + async fn shutdown_cancels_all_provider_tasks() { + let store = Arc::new(SensorStore::new()); + let config = HostConfig::default(); + let mut host = ProviderHost::new(config, store); + + let provider: Arc> = Arc::new(Box::new( + HealthyProvider::new("test-provider", "test.sensor").unwrap(), + )); + host.register_provider(provider).unwrap(); + + host.start(); + + tokio::time::sleep(Duration::from_millis(30)).await; + + let result = host.shutdown().await; + assert!(result.is_ok()); + } + + #[tokio::test] + async fn shutdown_waits_for_in_flight_poll() { + let store = Arc::new(SensorStore::new()); + let config = HostConfig::default(); + let mut host = ProviderHost::new(config, store); + + let provider: Arc> = Arc::new(Box::new( + HealthyProvider::new("slow-provider", "test.slow").unwrap(), + )); + host.register_provider(provider).unwrap(); + + host.start(); + + tokio::time::sleep(Duration::from_millis(30)).await; + + let result = host.shutdown().await; + assert!(result.is_ok()); + + assert!(host.providers.get("slow-provider").unwrap().task.is_none()); + } +} diff --git a/crates/provider-host/src/lib.rs b/crates/provider-host/src/lib.rs new file mode 100644 index 0000000..15a3f98 --- /dev/null +++ b/crates/provider-host/src/lib.rs @@ -0,0 +1,11 @@ +pub mod config; +pub mod error; +pub mod health; +pub mod host; +pub mod provider; + +pub use crate::config::HostConfig; +pub use crate::error::{ProviderError, ProviderResult}; +pub use crate::health::ProviderHealth; +pub use crate::host::{ProviderHost, ProviderStatus}; +pub use crate::provider::Provider; diff --git a/crates/provider-host/src/provider.rs b/crates/provider-host/src/provider.rs new file mode 100644 index 0000000..149a0ee --- /dev/null +++ b/crates/provider-host/src/provider.rs @@ -0,0 +1,28 @@ +use astragauge_domain::{ProviderManifest, SensorDescriptor, SensorSample}; +use async_trait::async_trait; +use std::time::Duration; + +use crate::health::ProviderHealth; +use crate::ProviderResult; + +/// Core Provider trait - defines the contract for all sensor providers +#[async_trait] +pub trait Provider: Send + Sync { + /// Returns provider's manifest metadata + fn manifest(&self) -> &ProviderManifest; + + /// Returns the suggested polling interval for this provider + fn poll_interval(&self) -> Duration; + + /// Discovers all available sensors from this provider + async fn discover(&self) -> ProviderResult>; + + /// Polls current values for all registered sensors + async fn poll(&self) -> ProviderResult>; + + /// Returns the current health status of the provider + async fn health(&self) -> ProviderHealth; + + /// Shuts down the provider and releases resources + async fn shutdown(&self) -> ProviderResult<()>; +} diff --git a/crates/provider-host/tests/host_tests.rs b/crates/provider-host/tests/host_tests.rs new file mode 100644 index 0000000..ec74e69 --- /dev/null +++ b/crates/provider-host/tests/host_tests.rs @@ -0,0 +1,252 @@ +//! Unit tests for ProviderHost functionality. + +use std::collections::HashMap; +use std::sync::Arc; +use std::time::Duration; + +use async_trait::async_trait; + +use astragauge_domain::{ + ProviderCapabilities, ProviderManifest, SensorCategories, SensorDescriptor, SensorId, + SensorSample, +}; +use astragauge_provider_host::{ + HostConfig, Provider, ProviderError, ProviderHealth, ProviderHost, ProviderResult, +}; +use astragauge_providers::MockProvider; +use astragauge_sensor_store::SensorStore; + +fn make_manifest(id: &str) -> ProviderManifest { + ProviderManifest { + id: id.to_string(), + name: format!("{} provider", id), + version: "1.0.0".to_string(), + description: "Test provider".to_string(), + author: None, + website: None, + repository: None, + license: None, + tags: None, + runtime: ">=1.0.0".to_string(), + capabilities: ProviderCapabilities { + historical: false, + high_frequency: false, + hardware_access: false, + }, + sensors: SensorCategories { categories: vec![] }, + } +} + +struct PanickingProvider { + manifest: ProviderManifest, +} + +impl PanickingProvider { + fn new(id: &str) -> Self { + Self { + manifest: make_manifest(id), + } + } +} + +#[async_trait] +impl Provider for PanickingProvider { + fn manifest(&self) -> &ProviderManifest { + &self.manifest + } + + fn poll_interval(&self) -> Duration { + Duration::from_millis(10) + } + + async fn discover(&self) -> ProviderResult> { + panic!("discover() panicked!") + } + + async fn poll(&self) -> ProviderResult> { + Ok(vec![]) + } + + async fn health(&self) -> ProviderHealth { + ProviderHealth::Ok + } + + async fn shutdown(&self) -> ProviderResult<()> { + Ok(()) + } +} + +struct PollPanickingProvider { + manifest: ProviderManifest, +} + +impl PollPanickingProvider { + fn new(id: &str) -> Self { + Self { + manifest: make_manifest(id), + } + } +} + +#[async_trait] +impl Provider for PollPanickingProvider { + fn manifest(&self) -> &ProviderManifest { + &self.manifest + } + + fn poll_interval(&self) -> Duration { + Duration::from_millis(10) + } + + async fn discover(&self) -> ProviderResult> { + Ok(vec![]) + } + + async fn poll(&self) -> ProviderResult> { + panic!("poll() panicked!") + } + + async fn health(&self) -> ProviderHealth { + ProviderHealth::Ok + } + + async fn shutdown(&self) -> ProviderResult<()> { + Ok(()) + } +} + +#[tokio::test] +async fn register_provider_increases_count() { + let store = Arc::new(SensorStore::new()); + let config = HostConfig::default(); + let mut host = ProviderHost::new(config, store); + + let provider: Arc> = Arc::new(Box::new(MockProvider::new_test())); + let result = host.register_provider(provider); + assert!(result.is_ok()); + + let started_count = host.start(); + assert_eq!(started_count, 1); +} + +#[test] +fn duplicate_registration_fails() { + let store = Arc::new(SensorStore::new()); + let config = HostConfig::default(); + let mut host = ProviderHost::new(config, store); + + let provider1: Arc> = Arc::new(Box::new(MockProvider::new_test())); + let result1 = host.register_provider(provider1); + assert!(result1.is_ok()); + + let provider2: Arc> = Arc::new(Box::new(MockProvider::new_test())); + let result2 = host.register_provider(provider2); + + assert!(result2.is_err()); + match result2.unwrap_err() { + ProviderError::RegistrationFailed { id, reason } => { + assert_eq!(id, "mock.provider"); + assert!(reason.contains("already registered")); + } + _ => panic!("Expected RegistrationFailed error"), + } +} + +#[tokio::test] +async fn samples_appear_in_store() { + let store = Arc::new(SensorStore::new()); + let config = HostConfig::default(); + let mut host = ProviderHost::new(config, Arc::clone(&store)); + + let provider: Arc> = Arc::new(Box::new(MockProvider::new_test())); + + host.register_provider(provider).unwrap(); + let started = host.start(); + assert_eq!(started, 1); + + tokio::time::sleep(Duration::from_millis(50)).await; + + let shutdown_result = host.shutdown().await; + assert!(shutdown_result.is_ok()); + + let sensor_id = SensorId::new("mock.sensor").expect("valid sensor id"); + let latest = store.get_value(&sensor_id).await; + assert!( + latest.is_some(), + "Expected sample to be in store for sensor mock.sensor" + ); + let sample = latest.unwrap(); + assert_eq!(sample.value, Some(42.0)); +} + +#[tokio::test] +async fn shutdown_stops_all_providers() { + let store = Arc::new(SensorStore::new()); + let config = HostConfig::default(); + let mut host = ProviderHost::new(config, Arc::clone(&store)); + + let sensor1 = SensorId::new("test.sensor1").expect("valid sensor id"); + let descriptor1 = SensorDescriptor { + id: sensor1.clone(), + name: "Test Sensor 1".to_string(), + category: "test".to_string(), + unit: "units".to_string(), + device: None, + tags: vec![], + }; + let mut values1 = HashMap::new(); + values1.insert(sensor1, 1.0); + + let provider1: Arc> = Arc::new(Box::new(MockProvider::with_sensors( + vec![descriptor1], + values1, + Duration::from_millis(10), + ))); + + host.register_provider(provider1).unwrap(); + let started = host.start(); + assert_eq!(started, 1); + + tokio::time::sleep(Duration::from_millis(50)).await; + + let shutdown_result = host.shutdown().await; + assert!(shutdown_result.is_ok()); + + let restarted = host.start(); + assert_eq!(restarted, 1); +} + +#[tokio::test] +async fn panicking_provider_contained() { + let store = Arc::new(SensorStore::new()); + let config = HostConfig::default(); + let mut host = ProviderHost::new(config, store); + + let provider: Arc> = Arc::new(Box::new(PanickingProvider::new("panic-test"))); + host.register_provider(provider).unwrap(); + + host.start(); + + tokio::time::sleep(Duration::from_millis(100)).await; + + let shutdown_result = host.shutdown().await; + assert!(shutdown_result.is_ok()); +} + +#[tokio::test] +async fn panicking_poll_provider_contained() { + let store = Arc::new(SensorStore::new()); + let config = HostConfig::default(); + let mut host = ProviderHost::new(config, store); + + let provider: Arc> = + Arc::new(Box::new(PollPanickingProvider::new("poll-panic-test"))); + host.register_provider(provider).unwrap(); + + host.start(); + + tokio::time::sleep(Duration::from_millis(100)).await; + + let shutdown_result = host.shutdown().await; + assert!(shutdown_result.is_ok()); +} diff --git a/crates/providers/Cargo.toml b/crates/providers/Cargo.toml new file mode 100644 index 0000000..16e25cd --- /dev/null +++ b/crates/providers/Cargo.toml @@ -0,0 +1,12 @@ +[package] +name = "astragauge-providers" +version = "0.1.0" +edition = "2021" + +[dependencies] +astragauge-domain = { path = "../domain" } +astragauge-provider-host = { path = "../provider-host" } +tokio = { version = "1", features = ["full"] } +async-trait = "0.1" +serde = { version = "1", features = ["derive"] } +tracing = "0.1" diff --git a/crates/providers/src/lib.rs b/crates/providers/src/lib.rs new file mode 100644 index 0000000..96b21f4 --- /dev/null +++ b/crates/providers/src/lib.rs @@ -0,0 +1,9 @@ +pub mod mock; + +#[cfg(target_os = "linux")] +pub mod linux; + +pub use mock::MockProvider; + +#[cfg(target_os = "linux")] +pub use linux::LinuxProvider; diff --git a/crates/providers/src/linux.rs b/crates/providers/src/linux.rs new file mode 100644 index 0000000..8288a25 --- /dev/null +++ b/crates/providers/src/linux.rs @@ -0,0 +1,695 @@ +//! Linux system provider - reads from /proc, /sys/class/hwmon for CPU, memory, and temperature sensors. + +#[cfg(target_os = "linux")] +use async_trait::async_trait; +#[cfg(target_os = "linux")] +use std::collections::HashMap; +#[cfg(target_os = "linux")] +use std::fs; +#[cfg(target_os = "linux")] +use std::path::Path; +#[cfg(target_os = "linux")] +use std::sync::{Arc, Mutex}; +#[cfg(target_os = "linux")] +use std::time::{SystemTime, UNIX_EPOCH}; + +#[cfg(target_os = "linux")] +use astragauge_domain::{ + ProviderCapabilities, ProviderManifest, SensorCategories, SensorDescriptor, SensorId, + SensorSample, +}; +#[cfg(target_os = "linux")] +use astragauge_provider_host::{Provider, ProviderHealth, ProviderResult}; + +/// CPU statistics from /proc/stat for utilization calculation. +/// Stores cumulative time values in jiffies (typically 1/100 second). +#[cfg(target_os = "linux")] +#[derive(Debug, Clone, Copy, Default)] +struct CpuStats { + user: u64, + nice: u64, + system: u64, + idle: u64, + iowait: u64, + irq: u64, + softirq: u64, + steal: u64, + guest: u64, + guest_nice: u64, +} + +#[cfg(target_os = "linux")] +impl CpuStats { + /// Calculate CPU utilization percentage from current and previous stats. + /// Returns None if delta is zero (no change between readings). + fn utilization_from(prev: CpuStats, curr: CpuStats) -> Option { + let prev_idle = prev.idle + prev.iowait; + let curr_idle = curr.idle + curr.iowait; + + let prev_total = prev.user + + prev.nice + + prev.system + + prev.idle + + prev.iowait + + prev.irq + + prev.softirq + + prev.steal + + prev.guest + + prev.guest_nice; + + let curr_total = curr.user + + curr.nice + + curr.system + + curr.idle + + curr.iowait + + curr.irq + + curr.softirq + + curr.steal + + curr.guest + + curr.guest_nice; + + let delta_total = curr_total.saturating_sub(prev_total); + let delta_idle = curr_idle.saturating_sub(prev_idle); + + if delta_total == 0 { + return None; + } + + let utilization = ((delta_total - delta_idle) as f64 / delta_total as f64) * 100.0; + Some(utilization.clamp(0.0, 100.0)) + } +} + +#[cfg(target_os = "linux")] +pub struct LinuxProvider { + manifest: ProviderManifest, + sensors: Vec, + /// Previous CPU stats for utilization delta calculation. + /// Uses Arc for interior mutability since poll() takes &self. + prev_cpu_stats: Arc>>, +} + +#[cfg(target_os = "linux")] +impl LinuxProvider { + #[allow(clippy::new_without_default)] + pub fn new() -> Self { + tracing::info!("Initializing Linux provider"); + let sensors = Self::discover_sensors(); + tracing::info!("Linux provider discovered {} sensors", sensors.len()); + Self { + manifest: linux_manifest(), + sensors, + prev_cpu_stats: Arc::new(Mutex::new(None)), + } + } + + fn discover_sensors() -> Vec { + let mut sensors = Vec::new(); + + match Self::discover_cpu_sensors() { + Ok(cpu_sensors) => { + tracing::debug!("Discovered {} CPU sensors", cpu_sensors.len()); + sensors.extend(cpu_sensors); + } + Err(e) => { + tracing::warn!("Failed to discover CPU sensors: {}", e); + } + } + + match Self::discover_memory_sensors() { + Ok(mem_sensors) => { + tracing::debug!("Discovered {} memory sensors", mem_sensors.len()); + sensors.extend(mem_sensors); + } + Err(e) => { + tracing::warn!("Failed to discover memory sensors: {}", e); + } + } + + match Self::discover_hwmon_sensors() { + Ok(hwmon_sensors) => { + tracing::debug!( + "Discovered {} hwmon temperature sensors", + hwmon_sensors.len() + ); + sensors.extend(hwmon_sensors); + } + Err(e) => { + tracing::warn!("Failed to discover hwmon sensors: {}", e); + } + } + + sensors + } + + fn discover_cpu_sensors() -> std::io::Result> { + let _content = fs::read_to_string("/proc/cpuinfo")?; + let mut sensors = Vec::new(); + + if let Ok(id) = SensorId::new("cpu.utilization") { + sensors.push(SensorDescriptor { + id, + name: "CPU Utilization".to_string(), + category: "cpu".to_string(), + unit: "percent".to_string(), + device: None, + tags: vec!["cpu".to_string()], + }); + } + + Ok(sensors) + } + + fn discover_memory_sensors() -> std::io::Result> { + let _content = fs::read_to_string("/proc/meminfo")?; + let mut sensors = Vec::new(); + + if let Ok(id) = SensorId::new("memory.used") { + sensors.push(SensorDescriptor { + id, + name: "Memory Used".to_string(), + category: "memory".to_string(), + unit: "bytes".to_string(), + device: None, + tags: vec!["memory".to_string()], + }); + } + + if let Ok(id) = SensorId::new("memory.total") { + sensors.push(SensorDescriptor { + id, + name: "Memory Total".to_string(), + category: "memory".to_string(), + unit: "bytes".to_string(), + device: None, + tags: vec!["memory".to_string()], + }); + } + + if let Ok(id) = SensorId::new("memory.utilization") { + sensors.push(SensorDescriptor { + id, + name: "Memory Utilization".to_string(), + category: "memory".to_string(), + unit: "percent".to_string(), + device: None, + tags: vec!["memory".to_string()], + }); + } + + if let Ok(id) = SensorId::new("memory.available") { + sensors.push(SensorDescriptor { + id, + name: "Memory Available".to_string(), + category: "memory".to_string(), + unit: "bytes".to_string(), + device: None, + tags: vec!["memory".to_string()], + }); + } + + Ok(sensors) + } + + fn discover_hwmon_sensors() -> std::io::Result> { + let mut sensors = Vec::new(); + let hwmon_path = Path::new("/sys/class/hwmon"); + + if !hwmon_path.exists() { + tracing::debug!("hwmon path does not exist: {}", hwmon_path.display()); + return Ok(sensors); + } + + let entries = fs::read_dir(hwmon_path)?; + for entry in entries { + let entry = entry?; + let hwmon_dir = entry.path(); + + let name = fs::read_to_string(hwmon_dir.join("name")) + .unwrap_or_else(|_| "unknown".to_string()) + .trim() + .to_lowercase(); + + let temp_sensors = Self::find_temp_sensors(&hwmon_dir, &name); + sensors.extend(temp_sensors); + } + + Ok(sensors) + } + + fn find_temp_sensors(hwmon_dir: &Path, device_name: &str) -> Vec { + let mut sensors = Vec::new(); + + if let Ok(entries) = fs::read_dir(hwmon_dir) { + for entry in entries.flatten() { + let file_name = entry.file_name(); + let name = file_name.to_string_lossy(); + + if !(name.ends_with("_input") && name.starts_with("temp")) { + continue; + } + + let index: String = name + .chars() + .skip(4) + .take_while(|c| c.is_ascii_digit()) + .collect(); + + if index.is_empty() { + continue; + } + + let label = fs::read_to_string(hwmon_dir.join(format!("temp{}_label", index))) + .ok() + .map(|s| s.trim().to_string()); + + let sensor_id = if device_name.contains("coretemp") + || device_name.contains("cpu") + || device_name.contains("k10temp") + || device_name.contains("k8temp") + { + "cpu.temperature".to_string() + } else if device_name.contains("gpu") || device_name.contains("nvidia") { + "gpu.temperature".to_string() + } else { + format!("{}.temperature", device_name.replace(' ', "_")) + }; + + if let Ok(id) = SensorId::new(&sensor_id) { + let display_name = label.unwrap_or_else(|| format!("{} Temperature", device_name)); + + sensors.push(SensorDescriptor { + id, + name: display_name, + category: "temperature".to_string(), + unit: "celsius".to_string(), + device: Some(device_name.to_string()), + tags: vec!["thermal".to_string(), device_name.to_string()], + }); + } + } + } + + sensors + } + + /// Get current timestamp in milliseconds since UNIX epoch. + fn current_timestamp_ms() -> u64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_millis() as u64) + .unwrap_or(0) + } + + /// Parse /proc/stat first line (aggregate CPU stats). + /// Format: cpu user nice system idle iowait irq softirq steal guest guest_nice + fn read_cpu_stats() -> Option { + let content = fs::read_to_string("/proc/stat").ok()?; + let first_line = content.lines().next()?; + + // Skip "cpu" prefix and parse values + let parts: Vec<&str> = first_line.split_whitespace().collect(); + if parts.len() < 8 || parts[0] != "cpu" { + return None; + } + + let parse_u64 = + |idx: usize| -> u64 { parts.get(idx).and_then(|s| s.parse().ok()).unwrap_or(0) }; + + Some(CpuStats { + user: parse_u64(1), + nice: parse_u64(2), + system: parse_u64(3), + idle: parse_u64(4), + iowait: parse_u64(5), + irq: parse_u64(6), + softirq: parse_u64(7), + steal: parse_u64(8), + guest: parse_u64(9), + guest_nice: parse_u64(10), + }) + } + + /// Poll CPU utilization sensor. + /// Requires previous stats for delta calculation. + fn poll_cpu(&self, samples: &mut Vec, timestamp_ms: u64) { + let current_stats = match Self::read_cpu_stats() { + Some(stats) => stats, + None => { + tracing::warn!("Failed to read /proc/stat for CPU utilization"); + return; + } + }; + + // Calculate utilization if we have previous stats + let mut prev_guard = self.prev_cpu_stats.lock().unwrap(); + if let Some(prev_stats) = prev_guard.as_ref() { + if let Some(utilization) = CpuStats::utilization_from(*prev_stats, current_stats) { + if let Ok(id) = SensorId::new("cpu.utilization") { + samples.push(SensorSample { + sensor_id: id, + timestamp_ms, + value: Some(utilization), + }); + } + } + } + + // Store current stats for next poll + *prev_guard = Some(current_stats); + } + + /// Parse /proc/meminfo and return key-value pairs in kB. + fn read_meminfo() -> Option> { + let content = fs::read_to_string("/proc/meminfo").ok()?; + let mut meminfo = HashMap::new(); + + for line in content.lines() { + let parts: Vec<&str> = line.split_whitespace().collect(); + if parts.len() >= 2 { + // Remove trailing colon from key (e.g., "MemTotal:") + let key = parts[0].trim_end_matches(':'); + // Value is in kB, convert to bytes + if let Ok(kb) = parts[1].parse::() { + meminfo.insert(key.to_string(), kb * 1024); + } + } + } + + Some(meminfo) + } + + /// Poll memory sensors from /proc/meminfo. + fn poll_memory(&self, samples: &mut Vec, timestamp_ms: u64) { + let meminfo = match Self::read_meminfo() { + Some(info) => info, + None => { + tracing::warn!("Failed to read /proc/meminfo"); + return; + } + }; + + let mem_total = meminfo.get("MemTotal").copied().unwrap_or(0); + let mem_available = meminfo.get("MemAvailable").copied().unwrap_or(0); + let mem_free = meminfo.get("MemFree").copied().unwrap_or(0); + let buffers = meminfo.get("Buffers").copied().unwrap_or(0); + let cached = meminfo.get("Cached").copied().unwrap_or(0); + + // Used memory = Total - Available (more accurate than Total - Free) + let mem_used = mem_total.saturating_sub(mem_available); + + // Memory utilization percentage + let mem_utilization = if mem_total > 0 { + Some((mem_used as f64 / mem_total as f64) * 100.0) + } else { + None + }; + + // Push samples + if let Ok(id) = SensorId::new("memory.total") { + samples.push(SensorSample { + sensor_id: id, + timestamp_ms, + value: Some(mem_total as f64), + }); + } + + if let Ok(id) = SensorId::new("memory.used") { + samples.push(SensorSample { + sensor_id: id, + timestamp_ms, + value: Some(mem_used as f64), + }); + } + + if let Ok(id) = SensorId::new("memory.available") { + samples.push(SensorSample { + sensor_id: id, + timestamp_ms, + value: Some(mem_available as f64), + }); + } + + if let Ok(id) = SensorId::new("memory.utilization") { + samples.push(SensorSample { + sensor_id: id, + timestamp_ms, + value: mem_utilization, + }); + } + + // Log additional info for debugging + tracing::trace!( + "Memory: total={}MB, used={}MB, available={}MB, free={}MB, buffers={}MB, cached={}MB", + mem_total / 1024 / 1024, + mem_used / 1024 / 1024, + mem_available / 1024 / 1024, + mem_free / 1024 / 1024, + buffers / 1024 / 1024, + cached / 1024 / 1024 + ); + } + + /// Poll temperature sensors from /sys/class/hwmon. + fn poll_temperatures(&self, samples: &mut Vec, timestamp_ms: u64) { + let hwmon_path = Path::new("/sys/class/hwmon"); + + if !hwmon_path.exists() { + tracing::trace!("hwmon path does not exist"); + return; + } + + let entries = match fs::read_dir(hwmon_path) { + Ok(e) => e, + Err(e) => { + tracing::warn!("Failed to read hwmon directory: {}", e); + return; + } + }; + + for entry in entries.flatten() { + let hwmon_dir = entry.path(); + + let device_name = match fs::read_to_string(hwmon_dir.join("name")) { + Ok(name) => name.trim().to_lowercase(), + Err(_) => continue, + }; + + // Determine sensor ID prefix based on device name + let sensor_prefix = if device_name.contains("coretemp") + || device_name.contains("cpu") + || device_name.contains("k10temp") + || device_name.contains("k8temp") + { + "cpu" + } else if device_name.contains("gpu") || device_name.contains("nvidia") { + "gpu" + } else { + &device_name.replace(' ', "_") + }; + + // Find and read temperature input files + if let Ok(dir_entries) = fs::read_dir(&hwmon_dir) { + for file_entry in dir_entries.flatten() { + let file_name = file_entry.file_name(); + let name = file_name.to_string_lossy(); + + if !(name.ends_with("_input") && name.starts_with("temp")) { + continue; + } + + // Read temperature value (in millidegrees Celsius) + let temp_content = match fs::read_to_string(file_entry.path()) { + Ok(c) => c, + Err(e) => { + tracing::trace!("Failed to read {}: {}", file_entry.path().display(), e); + continue; + } + }; + + let temp_mc: i64 = match temp_content.trim().parse() { + Ok(v) => v, + Err(_) => continue, + }; + + // Convert millidegrees to Celsius + let temp_c = temp_mc as f64 / 1000.0; + + // Create sensor ID + let sensor_id_str = format!("{}.temperature", sensor_prefix); + if let Ok(id) = SensorId::new(&sensor_id_str) { + samples.push(SensorSample { + sensor_id: id, + timestamp_ms, + value: Some(temp_c), + }); + } + } + } + } + } +} + +#[cfg(target_os = "linux")] +fn linux_manifest() -> ProviderManifest { + ProviderManifest { + id: "linux.provider".to_string(), + name: "Linux System Provider".to_string(), + version: env!("CARGO_PKG_VERSION").to_string(), + description: "System metrics from Linux kernel".to_string(), + author: Some("AstraGauge".to_string()), + website: None, + repository: None, + license: Some("MIT".to_string()), + tags: Some(vec!["linux".to_string(), "system".to_string()]), + runtime: ">=0.1.0".to_string(), + capabilities: ProviderCapabilities { + historical: false, + high_frequency: false, + hardware_access: true, + }, + sensors: SensorCategories { + categories: vec![ + "cpu".to_string(), + "memory".to_string(), + "temperature".to_string(), + ], + }, + } +} + +#[cfg(target_os = "linux")] +#[async_trait] +impl Provider for LinuxProvider { + fn manifest(&self) -> &ProviderManifest { + &self.manifest + } + + fn poll_interval(&self) -> std::time::Duration { + std::time::Duration::from_millis(1000) + } + + async fn discover(&self) -> ProviderResult> { + Ok(self.sensors.clone()) + } + + async fn poll(&self) -> ProviderResult> { + let mut samples = Vec::new(); + let timestamp_ms = Self::current_timestamp_ms(); + + // Poll each sensor category + self.poll_cpu(&mut samples, timestamp_ms); + self.poll_memory(&mut samples, timestamp_ms); + self.poll_temperatures(&mut samples, timestamp_ms); + + tracing::trace!("Linux provider polled {} samples", samples.len()); + Ok(samples) + } + + async fn health(&self) -> ProviderHealth { + ProviderHealth::Ok + } + + async fn shutdown(&self) -> ProviderResult<()> { + Ok(()) + } +} + +#[cfg(target_os = "linux")] +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_cpu_stats_default() { + let stats = CpuStats::default(); + assert_eq!(stats.user, 0); + assert_eq!(stats.idle, 0); + } + + #[test] + fn test_calculate_cpu_utilization_basic() { + let prev = CpuStats { + user: 100, + nice: 0, + system: 50, + idle: 800, + iowait: 10, + irq: 5, + softirq: 5, + steal: 0, + guest: 0, + guest_nice: 0, + }; + + let curr = CpuStats { + user: 150, // +50 + nice: 0, + system: 75, // +25 + idle: 900, // +90 + iowait: 15, // +5 + irq: 7, // +2 + softirq: 6, // +1 + steal: 0, + guest: 0, + guest_nice: 0, + }; + + // Total delta = 50 + 0 + 25 + 100 + 5 + 2 + 1 = 183 + // Idle delta = 100 + 5 = 105 + // Active delta = 183 - 105 = 78 + // Utilization = 78 / 183 * 100 = 42.62% + let utilization = CpuStats::utilization_from(prev, curr); + assert!(utilization.is_some()); + + let util = utilization.unwrap(); + assert!(util > 42.0 && util < 43.0, "Expected ~42.6%, got {}", util); + } + + #[test] + fn test_calculate_cpu_utilization_zero_delta() { + let stats = CpuStats { + user: 100, + nice: 0, + system: 50, + idle: 800, + iowait: 10, + irq: 5, + softirq: 5, + steal: 0, + guest: 0, + guest_nice: 0, + }; + + // Same stats should return None (no change) + let utilization = CpuStats::utilization_from(stats, stats); + assert!(utilization.is_none()); + } + + #[test] + fn test_calculate_cpu_utilization_clamp() { + let prev = CpuStats { + user: 100, + idle: 100, + ..CpuStats::default() + }; + + // Edge case: all active, no idle + let curr = CpuStats { + user: 200, // +100 active + idle: 100, // 0 idle + ..CpuStats::default() + }; + + let utilization = CpuStats::utilization_from(prev, curr); + assert!(utilization.is_some()); + assert!((utilization.unwrap() - 100.0).abs() < f64::EPSILON); + } + + #[test] + fn test_current_timestamp_ms() { + let ts = LinuxProvider::current_timestamp_ms(); + // Should be a reasonable timestamp (after year 2020) + assert!(ts > 1577836800000); // Jan 1, 2020 in ms + } +} diff --git a/crates/providers/src/mock.rs b/crates/providers/src/mock.rs new file mode 100644 index 0000000..10d5c88 --- /dev/null +++ b/crates/providers/src/mock.rs @@ -0,0 +1,202 @@ +//! Mock provider for testing purposes. +//! +//! Provides a configurable implementation of the Provider trait +//! that returns predefined sensor descriptors and values. + +use async_trait::async_trait; +use std::collections::HashMap; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; + +use astragauge_domain::{ + ProviderCapabilities, ProviderManifest, SensorCategories, SensorDescriptor, SensorId, + SensorSample, +}; +use astragauge_provider_host::{Provider, ProviderHealth, ProviderResult}; + +/// A mock provider for testing that returns configurable sensor data. +pub struct MockProvider { + descriptors: Vec, + values: HashMap, + poll_interval: Duration, + manifest: ProviderManifest, +} + +impl MockProvider { + /// Creates a new MockProvider with the given configuration. + pub fn with_sensors( + descriptors: Vec, + values: HashMap, + poll_interval: Duration, + ) -> Self { + Self { + manifest: create_test_manifest(), + descriptors, + values, + poll_interval, + } + } + + /// Creates a MockProvider with sensible defaults for testing. + /// + /// Default configuration: + /// - 1 sensor: `mock.sensor` with value 42.0 + /// - 10ms poll interval + pub fn new_test() -> Self { + let sensor_id = SensorId::new("mock.sensor").expect("valid sensor id"); + let descriptor = SensorDescriptor { + id: sensor_id.clone(), + name: "Mock Sensor".to_string(), + category: "test".to_string(), + unit: "units".to_string(), + device: None, + tags: vec![], + }; + + let mut values = HashMap::new(); + values.insert(sensor_id, 42.0); + + Self { + manifest: create_test_manifest(), + descriptors: vec![descriptor], + values, + poll_interval: Duration::from_millis(10), + } + } +} + +fn create_test_manifest() -> ProviderManifest { + ProviderManifest { + id: "mock.provider".to_string(), + name: "Mock Provider".to_string(), + version: env!("CARGO_PKG_VERSION").to_string(), + description: "A mock provider for testing".to_string(), + author: Some("AstraGauge".to_string()), + website: None, + repository: None, + license: Some("MIT".to_string()), + tags: Some(vec!["test".to_string(), "mock".to_string()]), + runtime: ">=0.1.0".to_string(), + capabilities: ProviderCapabilities { + historical: false, + high_frequency: true, + hardware_access: false, + }, + sensors: SensorCategories { + categories: vec!["test".to_string()], + }, + } +} + +#[async_trait] +impl Provider for MockProvider { + fn manifest(&self) -> &ProviderManifest { + &self.manifest + } + + fn poll_interval(&self) -> Duration { + self.poll_interval + } + + async fn discover(&self) -> ProviderResult> { + Ok(self.descriptors.clone()) + } + + async fn poll(&self) -> ProviderResult> { + let timestamp_ms = SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_millis() as u64) + .unwrap_or(0); + + let samples: Vec = self + .values + .iter() + .map(|(sensor_id, &value)| SensorSample { + sensor_id: sensor_id.clone(), + timestamp_ms, + value: Some(value), + }) + .collect(); + + Ok(samples) + } + + async fn health(&self) -> ProviderHealth { + ProviderHealth::Ok + } + + async fn shutdown(&self) -> ProviderResult<()> { + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_new_test_creates_provider_with_defaults() { + let provider = MockProvider::new_test(); + assert_eq!(provider.poll_interval(), Duration::from_millis(10)); + } + + #[test] + fn test_with_sensors_creates_provider_with_custom_config() { + let sensor_id = SensorId::new("custom.sensor").unwrap(); + let descriptor = SensorDescriptor { + id: sensor_id.clone(), + name: "Custom Sensor".to_string(), + category: "custom".to_string(), + unit: "units".to_string(), + device: None, + tags: vec![], + }; + + let mut values = HashMap::new(); + values.insert(sensor_id.clone(), 100.0); + + let provider = MockProvider::with_sensors(vec![descriptor], values, Duration::from_millis(50)); + + assert_eq!(provider.poll_interval(), Duration::from_millis(50)); + assert_eq!(provider.manifest().id, "mock.provider"); + } + + #[tokio::test] + async fn test_discover_returns_configured_descriptors() { + let provider = MockProvider::new_test(); + let descriptors = provider.discover().await.unwrap(); + assert_eq!(descriptors.len(), 1); + assert_eq!(descriptors[0].id.as_str(), "mock.sensor"); + } + + #[tokio::test] + async fn test_poll_returns_samples_with_timestamps() { + let provider = MockProvider::new_test(); + let samples = provider.poll().await.unwrap(); + assert_eq!(samples.len(), 1); + assert_eq!(samples[0].sensor_id.as_str(), "mock.sensor"); + assert!(samples[0].timestamp_ms > 0); + assert_eq!(samples[0].value, Some(42.0)); + } + + #[tokio::test] + async fn test_health_returns_ok() { + let provider = MockProvider::new_test(); + let health = provider.health().await; + assert_eq!(health, ProviderHealth::Ok); + } + + #[tokio::test] + async fn test_shutdown_returns_ok() { + let provider = MockProvider::new_test(); + let result = provider.shutdown().await; + assert!(result.is_ok()); + } + + #[tokio::test] + async fn test_manifest_returns_test_manifest() { + let provider = MockProvider::new_test(); + let manifest = provider.manifest(); + assert_eq!(manifest.id, "mock.provider"); + assert_eq!(manifest.name, "Mock Provider"); + } +} diff --git a/crates/providers/tests/linux_tests.rs b/crates/providers/tests/linux_tests.rs new file mode 100644 index 0000000..877c6ce --- /dev/null +++ b/crates/providers/tests/linux_tests.rs @@ -0,0 +1,241 @@ +//! Integration tests for LinuxProvider. +//! +//! Run with: cargo test -p astragauge-providers --test linux_tests -- --ignored +//! (Tests are ignored by default to skip on non-Linux systems) + +#[cfg(target_os = "linux")] +mod linux_tests { + use std::time::Duration; + + use astragauge_domain::SensorId; + use astragauge_provider_host::Provider; + use astragauge_providers::LinuxProvider; + + fn is_valid_sensor_id(id: &SensorId) -> bool { + let id_str = id.as_str(); + + if id_str != id_str.to_lowercase() { + return false; + } + + let parts: Vec<&str> = id_str.split('.').collect(); + if parts.len() < 2 { + return false; + } + + if parts.iter().any(|p| p.is_empty()) { + return false; + } + + if id_str.contains(' ') { + return false; + } + + if parts[0].contains('_') { + return false; + } + + true + } + + #[tokio::test] + #[ignore] + async fn discover_returns_sensors() { + let provider = LinuxProvider::new(); + let sensors = provider.discover().await.expect("discover should succeed"); + + assert!( + !sensors.is_empty(), + "Provider should discover at least some sensors on Linux" + ); + + for sensor in &sensors { + eprintln!( + "Discovered sensor: {} ({})", + sensor.id.as_str(), + sensor.name + ); + } + } + + /// CPU utilization requires two polls for delta calculation. + #[tokio::test] + #[ignore] + async fn poll_returns_samples() { + let provider = LinuxProvider::new(); + + let _baseline = provider.poll().await.expect("first poll should succeed"); + + tokio::time::sleep(Duration::from_millis(100)).await; + + let samples = provider.poll().await.expect("second poll should succeed"); + + assert!(!samples.is_empty(), "Poll should return samples"); + + for sample in &samples { + if let Some(value) = sample.value { + eprintln!("Sample: {} = {:.2}", sample.sensor_id.as_str(), value); + } + } + } + + #[tokio::test] + #[ignore] + async fn missing_proc_handled_gracefully() { + let provider = LinuxProvider::new(); + + let discover_result = provider.discover().await; + assert!( + discover_result.is_ok(), + "discover() should not panic on missing files" + ); + + let poll_result = provider.poll().await; + assert!( + poll_result.is_ok(), + "poll() should not panic on missing files" + ); + + let health = provider.health().await; + eprintln!("Provider health: {:?}", health); + } + + /// See docs/specs/sensor-schema.md for format rules. + #[tokio::test] + #[ignore] + async fn sensor_ids_follow_conventions() { + let provider = LinuxProvider::new(); + let sensors = provider.discover().await.expect("discover should succeed"); + + let mut invalid_ids: Vec = Vec::new(); + + for sensor in &sensors { + if !is_valid_sensor_id(&sensor.id) { + invalid_ids.push(format!( + "{} (name: {}, category: {})", + sensor.id.as_str(), + sensor.name, + sensor.category + )); + } + } + + if !invalid_ids.is_empty() { + eprintln!("Invalid sensor IDs found:"); + for id in &invalid_ids { + eprintln!(" - {}", id); + } + panic!( + "Found {} sensor ID(s) that don't follow naming conventions. \ + Expected format: device.metric (lowercase, dot-separated, no spaces, singular device names)", + invalid_ids.len() + ); + } + + let sensor_ids: Vec<&str> = sensors.iter().map(|s| s.id.as_str()).collect(); + + assert!( + sensor_ids.contains(&"cpu.utilization"), + "Expected cpu.utilization sensor" + ); + + assert!( + sensor_ids.contains(&"memory.used"), + "Expected memory.used sensor" + ); + assert!( + sensor_ids.contains(&"memory.total"), + "Expected memory.total sensor" + ); + assert!( + sensor_ids.contains(&"memory.utilization"), + "Expected memory.utilization sensor" + ); + assert!( + sensor_ids.contains(&"memory.available"), + "Expected memory.available sensor" + ); + } + + #[tokio::test] + #[ignore] + async fn sensor_descriptors_have_required_fields() { + let provider = LinuxProvider::new(); + let sensors = provider.discover().await.expect("discover should succeed"); + + for sensor in &sensors { + assert!( + !sensor.id.as_str().is_empty(), + "Sensor ID should not be empty" + ); + + assert!( + !sensor.name.is_empty(), + "Sensor {} should have a name", + sensor.id.as_str() + ); + + assert!( + !sensor.category.is_empty(), + "Sensor {} should have a category", + sensor.id.as_str() + ); + + assert!( + !sensor.unit.is_empty(), + "Sensor {} should have a unit", + sensor.id.as_str() + ); + + eprintln!( + "Sensor: {} | name: {} | category: {} | unit: {}", + sensor.id.as_str(), + sensor.name, + sensor.category, + sensor.unit + ); + } + } + + #[tokio::test] + #[ignore] + async fn poll_samples_have_valid_timestamps() { + let provider = LinuxProvider::new(); + + let _ = provider.poll().await; + + tokio::time::sleep(Duration::from_millis(50)).await; + + let samples = provider.poll().await.expect("poll should succeed"); + + let now_ms = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_millis() as u64) + .unwrap_or(0); + + for sample in &samples { + let age_ms = now_ms.saturating_sub(sample.timestamp_ms); + assert!( + age_ms < 5000, + "Sample {} timestamp is too old: {}ms", + sample.sensor_id.as_str(), + age_ms + ); + + assert!( + sample.timestamp_ms <= now_ms + 1000, + "Sample {} timestamp is in the future", + sample.sensor_id.as_str() + ); + } + } +} + +#[cfg(not(target_os = "linux"))] +mod linux_tests { + #[test] + #[ignore] + fn linux_tests_skipped_on_non_linux() { + println!("LinuxProvider tests are only available on Linux"); + } +}