Add the implementation of the telemetry source manager Add the telemetry source manager, and two implementation for I2C sensors with sysfs direct polling, and D-Bus sensors with polling/matching over D-Bus. Please see the README.md in the telemetry_source_manager for the design, and the API document inside telemetry_source_manager.rs. The sensor_config.rs is for extract sensor configuration from EntityManager's configuration files. Google-Bug-Id: 366492368 Change-Id: I956558239876757487659694aeb86dbe32c82987 Signed-off-by: Yongbing Chen <yongbingchen@google.com>
diff --git a/.markdownlint.yaml b/.markdownlint.yaml new file mode 100644 index 0000000..581a705 --- /dev/null +++ b/.markdownlint.yaml
@@ -0,0 +1,9 @@ +default: true +MD033: + allowed_elements: + - br # This allows use <br> to break lines +MD013: + line_length: 80 + code_blocks: false # This disables line length checking in code blocks + tables: false # Optional: Also common to disable for tables +MD051: false # Disable link fragment checking
diff --git a/streaming_telemetry/src/telemetry_source_manager/README.md b/streaming_telemetry/src/telemetry_source_manager/README.md new file mode 100644 index 0000000..2b7ac2e --- /dev/null +++ b/streaming_telemetry/src/telemetry_source_manager/README.md
@@ -0,0 +1,189 @@ +# Serverless Telemetry Source Manager + +## Introduction + +The Serverless Telemetry Source Manager aims to simplify telemetry management by +eliminating the need for separate systemd services, as used in approaches like +the OpenBMC [D-Bus sensor daemon](https://github.com/openbmc/dbus-sensors.git). +This solution removes the daemon layer and inter-process communication (IPC), +resulting in a more straightforward implementation. + +## Northbound API + +The telemetry source manager exposes three main APIs: + +1. **`add_source()`**: Adds (or removes) a telemetry source to the manager, + which will then poll the source for value updates. +2. **`query_sources()`**: Queries telemetry sources by secondary key (one of the + source properties, such as the FRU it belongs to) or by primary key: source + name. +3. **`subscribe_telemetries()`**: Allows subscribers to receive telemetry + events, by periodical or when value changes. + +## Southbound API + +Clients must provide a method to poll the telemetry value when adding a source +to the manager, as defined by the asynchronous trait `AsyncReadValue`. For I2C +sources, this can be a simple wrapper to read the I2C device via the sysfs +interface. This interface abstraction facilitates integration with third-party +vendors. + +## Use case example + +The `src/main.rs` file demonstrates a complete use case from a client's +perspective.<br> Typically, one client handles the manager creation, but there +is currently no limit on how many clients can query and subscribe +simultaneously. + +1. Adding telemetry sources<br> Telemetry sources can be added to the manager, + and the polling task will run in the background automatically when + subscription happens:<br> + + ```rust + // Mock implementation of AsyncReadValue + struct MockAsyncReadValue { // Implementation details... } + + #[async_trait::async_trait] + impl AsyncReadValue<f64> for MockAsyncReadValue { + async fn read_values( + &self, + sampling_interval: Duration, + reporting_interval: Duration, + ) -> Result<Vec<(f64, SystemTime)>, std::io::Error> { + // Implementation details... + } + + async fn get_value(&self) -> Result<(f64, SystemTime), std::io::Error> { + // Implementation details... + } + } + + // The telemetry source manager can be cloned and shared by other threads/tasks freely + let telemetry_source_manager = TelemetrySourceManager::new(); + + let mock_reader = Arc::new(MockAsyncReadValue::new()); + + // Add telemetry sources to the manager + // A source can have an arbitrary number of properties as key/value pairs + // The property key names are not restricted, though they should follow the Redfish data scheme + let source1 = TelemetrySource::new( + "Source1".to_string(), + Some(vec![ + ("FRU".to_string(), "CPU".to_string()), + ("ReadingType".to_string(), "Temperature".to_string()), + ]), + mock_reader.clone(), + Duration::from_secs(1), + Duration::from_millis(100), + Duration::from_secs(5), + ); + telemetry_source_manager.add_source::<f64>(source1).await.unwrap(); + + // Add more sources as needed... + + ``` + +2. Query telemetry sources<br> + + 2.1 By primary key: + + ```rust + // Query sources by primary keys: empty input means all sources + if let Ok(sources) = telemetry_source_manager.query_sources_by_names(&[]) { + println!("All sources in manager:"); + for source in sources { + println!("{}", source); + } + } + + // Query source by primary keys: one source + if let Ok(sources) = telemetry_source_manager.query_sources_by_names(&["Source1"]) { + assert_eq!(sources.len(), 1); + } + ``` + + 2.2 By secondary keys: + + ```rust + // Query all available secondary keys in the manager + let keys = telemetry_source_manager.query_all_secondary_keys(); + println!("Secondary keys: {:?}", keys); + + // Query sources by secondary key + if let Ok(sources) = telemetry_source_manager.query_sources_by_property(("FRU", "CPU")) { + for source in sources { + println!("Source with FRU CPU: {}", source); + } + } + ``` + +3. Subscribe to telemetry events<br> + + 3.1 Query sources by reading type, then subscribe to telemetry events from + all of them + + ```rust + if let Ok(temperature_sources) = + telemetry_source_manager.query_sources_by_property(("ReadingType", "Temperature")) + { + for source in temperature_sources { + println!("Subscribe to temperature source: {}", source); + if let Ok(mut subscription) = + telemetry_source_manager.subscribe_telemetries::<f64>(&source, SubscriptionType::OnChange) + { + tokio::spawn(async move { + while let Some(telemetries) = subscription.telemetry_receiver.recv().await { + for telemetry in telemetries { + println!("Received {:?}", telemetry); + } + } + }); + } + } + } + ``` + + 3.2 Subscribe to Periodical telemetry events, which will increase the polling + rate of this source + + ```rust + if let Ok(subscription) = telemetry_source_manager.subscribe_telemetries::<f64>( + "Source2", + SubscriptionType::Periodical(Duration::from_millis(1), Duration::from_millis(10)), + ) { + let telemetry_source_manager_clone = telemetry_source_manager.clone(); + if let Ok(sources) = telemetry_source_manager_clone.query_sources_by_names(&["Source2"]) { + println!("Source2 before subscription dropped: {}", sources[0]); + } + + tokio::spawn(async move { + // Subscription handling logic... + }); + } + ``` + +## Sample output + +```sh +All sources in manager: +Source2 +Source3 +Source1 +Secondary keys: ["ReadingType", "FRU"] +Source with FRU CPU: Source1 +Source with FRU CPU: Source3 +Subscribe to temperature source: Source1 +Subscribe to temperature source: Source2 +Source2 before subscription dropped: Source2 +Received TypedTelemetry { source_name: "Source3", telemetry_type: "OnChange", value: 66.81906494537024, timestamp: SystemTime { tv_sec: 616376, tv_nsec: 949746575 } } +Received TypedTelemetry { source_name: "Source2", telemetry_type: "Periodical", value: 21.584232735733732, timestamp: SystemTime { tv_sec: 616376, tv_nsec: 949765075 } } +Received TypedTelemetry { source_name: "Source1", telemetry_type: "Periodical", value: 69.1666621987969, timestamp: SystemTime { tv_sec: 616376, tv_nsec: 949819185 } } +... +Subscription to Source2 dropped after 15 seconds +Source2 after subscription dropped: Source2 +... +``` + +This README provides an overview of the Serverless Telemetry Source Manager, its +APIs, and examples of how to use it. For more detailed information, please refer +to the source code and comments within the `telemetry_source_manager.rs` file.
diff --git a/streaming_telemetry/src/telemetry_source_manager/dbus_sensors/dbus_sensors.rs b/streaming_telemetry/src/telemetry_source_manager/dbus_sensors/dbus_sensors.rs new file mode 100644 index 0000000..cae9d91 --- /dev/null +++ b/streaming_telemetry/src/telemetry_source_manager/dbus_sensors/dbus_sensors.rs
@@ -0,0 +1,240 @@ +//! This module provides functionality for creating and managing D-Bus sensors for the telemetry_source_manager. + +use crate::telemetry_source_manager::dbus_sensors::ObjectMapper::ObjectMapperProxy; +use crate::telemetry_source_manager::dbus_sensors::Value::ValueProxy; +use crate::telemetry_source_manager::sensor_configs::{FruInfo, Sensor as SensorConfig}; +use crate::telemetry_source_manager::telemetry_source_manager::{AsyncReadValue, TelemetrySource}; +use anyhow::Result; +use regex::Regex; +use std::collections::HashMap; +use std::sync::Arc; +use std::time::{Duration, SystemTime}; +use tokio::time::sleep; +use zbus::names::BusName; +use zbus::zvariant::ObjectPath; +use zbus::Connection; + +/// Fetches the service name and interfaces for a given sensor path. +/// +/// Equivalent to below sample busctl command +///~# busctl call xyz.openbmc_project.ObjectMapper /xyz/openbmc_project/object_mapper xyz.openbmc_project.ObjectMapper GetObject sas "/xyz/openbmc_project/sensors/voltage/P1_SEQ_VDD_CPU0" 0 +/// a{sas} 2 "xyz.openbmc_project.ObjectMapper" 3 "org.freedesktop.DBus.Introspectable" "org.freedesktop.DBus.Peer" "org.freedesktop.DBus.Properties" "xyz.openbmc_project.PSUSensor" 5 "xyz.openbmc_project.Association.Definitions" "xyz.openbmc_project.Sensor.Threshold.Critical" "xyz.openbmc_project.Sensor.Value" "xyz.openbmc_project.State.Decorator.Availability" "xyz.openbmc_project.State.Decorator.OperationalStatus" +/// +/// # Arguments +/// +/// * `connection` - The D-Bus connection. +/// * `sensor_path` - The D-Bus path of the sensor. +/// +/// # Returns +/// +/// A tuple containing the service name and a vector of interface names. +async fn fetch_sensor_service_name( + connection: &Connection, + sensor_path: &str, +) -> Result<(String, Vec<String>), Box<dyn std::error::Error>> { + let proxy = ObjectMapperProxy::builder(connection) + .destination("xyz.openbmc_project.ObjectMapper")? + .path("/xyz/openbmc_project/object_mapper")? + .build() + .await?; + + let result = proxy.get_object(sensor_path, &Vec::<&str>::new()).await?; + + // Search for a service that provides the "xyz.openbmc_project.Sensor.Value" interface + let (service_name, interfaces) = result + .into_iter() + .find(|(_, interfaces)| interfaces.contains(&"xyz.openbmc_project.Sensor.Value".into())) + .unwrap_or_default(); // This will default to (String::new(), Vec::new()) if not found + + Ok((service_name, interfaces)) +} + +/// Maps a PLDM sensor name to its corresponding D-Bus path and service. +/// +/// +/// Example: map sensor name ${sensor_type}_${sensor_name} to D-Bus path as +/// "/xyz/openbmc_project/sensors/${sensor_type}/${sensor_name} +/// +/// # Arguments +/// +/// * `connection` - The D-Bus connection. +/// * `sensor_name` - The name of the PLDM sensor. +/// +/// # Returns +/// +/// A tuple containing the service information and the D-Bus path for the sensor. +async fn map_pldm_sensor_to_dbus( + connection: &Connection, + sensor_name: &str, +) -> Result<((String, Vec<String>), String), Box<dyn std::error::Error>> { + let fan_pwm_re = Regex::new(r"^fan\d+_pwm$")?; + let fan_tach_re = Regex::new(r"^fan\d+_tach$")?; + + // Determine sensor type based on the last suffix + let sensor_type = if sensor_name.contains("_CURR") { + "current" + } else if sensor_name.contains("_PWR") || sensor_name.contains("_POWER") { + "power" + } else if sensor_name.contains("_TEMP") { + "temperature" + } else if fan_pwm_re.is_match(sensor_name) { + "fan_pwm" + } else if fan_tach_re.is_match(sensor_name) { + "fan_tach" + } else { + "unknown" + }; + + let sensor_path = format!( + "/xyz/openbmc_project/sensors/{}/{}", + sensor_type, sensor_name + ); + + let sensor_service = fetch_sensor_service_name(connection, &sensor_path).await?; + + Ok((sensor_service, sensor_path)) +} + +pub struct DbusSensorReader { + #[allow(dead_code)] + name: String, + #[allow(dead_code)] + connection: Connection, + proxy: ValueProxy<'static>, +} + +impl DbusSensorReader { + /// Creates a new DbusSensorReader instance. + /// + /// # Arguments + /// + /// * `sensor_name` - The sensor name. + /// + /// # Returns + /// + /// A Result containing the new DbusSensorReader instance or an error. + pub async fn new(sensor_name: String) -> Result<Self, Box<dyn std::error::Error>> { + let connection = Connection::system().await?; + let (sensor_services, sensor_path) = + map_pldm_sensor_to_dbus(&connection, &sensor_name).await?; + let (service_name, _interfaces) = sensor_services; + let sensor_service = BusName::try_from(service_name)?; + let sensor_path = ObjectPath::try_from(sensor_path)?; + let proxy = ValueProxy::builder(&connection) + .destination(&sensor_service)? + .path(sensor_path)? + .build() + .await?; + + Ok(Self { + name: sensor_name, + connection, + proxy, + }) + } + + async fn read_single_value(&self) -> Result<f64, Box<dyn std::error::Error>> { + Ok(self.proxy.value().await?) + } +} + +#[async_trait::async_trait] +impl AsyncReadValue<f64> for DbusSensorReader { + /// Reads sensor values asynchronously. + /// + /// # Arguments + /// + /// * `sampling_interval` - The interval at which to sample the sensor. + /// * `reporting_interval` - The interval at which to report sensor values. + /// + /// # Returns + /// + /// A Result containing a vector of sensor values and timestamps, or an error. + async fn read_values( + &self, + sampling_interval: Duration, + reporting_interval: Duration, + ) -> std::io::Result<Vec<(f64, SystemTime)>> { + let mut values = Vec::new(); + let start_time = SystemTime::now(); + loop { + let now = SystemTime::now(); + let value = self + .read_single_value() + .await + .map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e.to_string()))?; + values.push((value, now)); + if now.duration_since(start_time).unwrap() >= reporting_interval { + break; + } + sleep(sampling_interval).await; + } + Ok(values) + } + + /// Read current value and return it with timestamp. + /// + /// # Returns + /// + /// A value with its corresponding timestamp. + async fn get_value(&self) -> std::io::Result<(f64, SystemTime)> { + let value = self + .read_single_value() + .await + .map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e.to_string()))?; + Ok((value, SystemTime::now())) + } +} + +/// Creates a D-Bus sensor from the provided configuration. +/// +/// # Arguments +/// +/// * `sensor_config` - The sensor configuration. +/// * `all_sensor_frus` - A HashMap containing FRU (Field Replaceable Unit) information for sensors. +/// * `sampling_interval` - The interval at which to sample the sensor (in milliseconds). +/// * `reporting_interval` - The interval at which to report sensor values (in milliseconds). +/// +/// # Returns +/// +/// A Result containing the created Sensor instance or an error. +pub async fn create_dbus_sensor( + sensor_config: &SensorConfig, + all_sensor_frus: &HashMap<String, FruInfo>, + sampling_interval: u64, + reporting_interval: u64, +) -> Result<Box<TelemetrySource<f64>>, Box<dyn std::error::Error>> { + let reader = Arc::new(DbusSensorReader::new(sensor_config.Name.clone()).await?); + + let mut secondary_keys = Vec::new(); + + // Add sensor's ReadingType as secondary key + secondary_keys.push(( + "ReadingType".to_string(), + format!("{:?}", sensor_config.reading_type.1), + )); + + // Add sensor's type as secondary key + secondary_keys.push(("Type".to_string(), "PLDM".to_string())); + + // Add sensor's FruInfo if available + if let Some(fru_info) = all_sensor_frus.get(&sensor_config.Name) { + if let Some(location_code) = &fru_info.location_code { + secondary_keys.push(("LocationCode".to_string(), location_code.clone())); + } + if let Some(service_label) = &fru_info.service_label { + secondary_keys.push(("ServiceLabel".to_string(), service_label.clone())); + } + } + + let sensor = TelemetrySource::new( + sensor_config.Name.clone(), + Some(secondary_keys), + reader, + Duration::from_millis(sampling_interval), + Duration::from_millis(sampling_interval), + Duration::from_millis(reporting_interval), + ); + + Ok(sensor) +}
diff --git a/streaming_telemetry/src/telemetry_source_manager/i2c_sensors.rs b/streaming_telemetry/src/telemetry_source_manager/i2c_sensors.rs new file mode 100644 index 0000000..80b9c50 --- /dev/null +++ b/streaming_telemetry/src/telemetry_source_manager/i2c_sensors.rs
@@ -0,0 +1,253 @@ +//! This module provides functionality for creating and managing I2C sensors for the telemetry_source_manager. +#![allow(clippy::type_complexity)] + +use crate::telemetry_source_manager::sensor_configs::{ + find_sensor_file, FruInfo, PSUProperty, Sensor as SensorConfig, +}; +use crate::telemetry_source_manager::telemetry_source_manager::{AsyncReadValue, TelemetrySource}; +use anyhow::Result; +use std::collections::HashMap; +use std::str::FromStr; +use std::sync::Arc; +use std::thread; +use std::time::{Duration, Instant, SystemTime}; +use tokio::sync::mpsc; +use tokio::sync::Mutex; +use tokio_uring::fs::File as UringFile; + +const BUFFER_SIZE: usize = 128; + +/// Represents an I2C sensor reader that implements AsyncReadValue for the telemetry_source_manager. +pub struct I2cSensorReader { + // This mutex is only for making this struct to have Send + Sync, as required by trait AsyncReadValue + // There should be no real contender for this lock in data path, unless the get_value and read_values + // are called at same time. + value_receiver: Arc<Mutex<mpsc::Receiver<Vec<(f64, SystemTime)>>>>, + #[allow(dead_code)] + psu_property: PSUProperty, + sampling_interval: Arc<std::sync::atomic::AtomicU64>, + reporting_interval: Arc<std::sync::atomic::AtomicU64>, +} + +impl I2cSensorReader { + /// Creates a new I2cSensorReader instance. + /// + /// # Arguments + /// + /// * `sysfs_path` - The sysfs path for the sensor. + /// * `psu_property` - The PSU property associated with the sensor. + /// * `sampling_interval` - The interval at which to sample the sensor. + /// * `reporting_interval` - The interval at which to report sensor values. + /// + /// # Returns + /// + /// A Result containing the new I2cSensorReader instance or an error. + + pub fn new( + sysfs_path: String, + psu_property: PSUProperty, + sampling_interval: Duration, + reporting_interval: Duration, + ) -> Result<Self> { + let (tx, rx) = mpsc::channel(16); + let value_receiver = Arc::new(Mutex::new(rx)); + let sampling_interval = Arc::new(std::sync::atomic::AtomicU64::new( + sampling_interval.as_millis() as u64, + )); + let reporting_interval = Arc::new(std::sync::atomic::AtomicU64::new( + reporting_interval.as_millis() as u64, + )); + + let sampling_interval_clone = sampling_interval.clone(); + let reporting_interval_clone = reporting_interval.clone(); + + thread::spawn(move || { + tokio_uring::start(async move { + let file = match UringFile::open(&sysfs_path).await { + Ok(f) => f, + Err(e) => { + eprintln!("Error opening file {}: {:?}", sysfs_path, e); + return; + } + }; + + let mut buf = vec![0u8; BUFFER_SIZE]; + let mut batch_values = Vec::new(); + let mut last_report_time = SystemTime::now(); + let factor = 10.0_f64.powi(psu_property.scale); + + loop { + let loop_start = Instant::now(); + + let (res, new_buf) = file.read_at(buf, 0).await; + buf = new_buf; // Reuse the buffer + + match res { + Ok(bytes_read) => { + if bytes_read == 0 { + continue; + } + + let s = std::str::from_utf8(&buf[..bytes_read]).unwrap_or_default(); + let raw_value = f64::from_str(s.trim()).unwrap_or_default(); + let final_value = (raw_value / factor) + psu_property.offset as f64; + + let now = SystemTime::now(); + batch_values.push((final_value, now)); + + if now + .duration_since(last_report_time) + .unwrap_or(Duration::ZERO) + >= Duration::from_millis( + reporting_interval_clone + .load(std::sync::atomic::Ordering::Relaxed), + ) + { + if tx.send(batch_values.clone()).await.is_err() { + // The receiver has been dropped, exit the thread + break; + } + batch_values.clear(); + last_report_time = now; + } + } + Err(e) => { + eprintln!("Error reading file {}: {:?}", sysfs_path, e); + } + } + + let elapsed = loop_start.elapsed(); + let intended_interval = Duration::from_millis( + sampling_interval_clone.load(std::sync::atomic::Ordering::Relaxed), + ); + + if elapsed < intended_interval { + let sleep_duration = intended_interval - elapsed; + tokio::time::sleep(sleep_duration).await; + } + } + }); + }); + + Ok(Self { + value_receiver, + psu_property, + sampling_interval, + reporting_interval, + }) + } +} + +#[async_trait::async_trait] +impl AsyncReadValue<f64> for I2cSensorReader { + /// Reads sensor values asynchronously. + /// + /// # Arguments + /// + /// * `sampling_interval` - The interval at which to sample the sensor. + /// * `reporting_interval` - The interval at which to report sensor values. + /// + /// # Returns + /// + /// A Result containing a vector of sensor values and timestamps, or an error. + async fn read_values( + &self, + sampling_interval: Duration, + reporting_interval: Duration, + ) -> Result<Vec<(f64, SystemTime)>, std::io::Error> { + self.sampling_interval.store( + sampling_interval.as_millis() as u64, + std::sync::atomic::Ordering::Relaxed, + ); + self.reporting_interval.store( + reporting_interval.as_millis() as u64, + std::sync::atomic::Ordering::Relaxed, + ); + let mut value_receiver = self.value_receiver.lock().await; + match value_receiver.recv().await { + Some(values) => Ok(values), + None => Err(std::io::Error::new( + std::io::ErrorKind::Other, + "io_uring thread has terminated", + )), + } + } + + /// Read current value and return it with timestamp. + /// + /// # Returns + /// + /// A value with its corresponding timestamp. + async fn get_value(&self) -> Result<(f64, SystemTime), std::io::Error> { + let mut value_receiver = self.value_receiver.lock().await; + match value_receiver.recv().await { + Some(values) => values.last().cloned().ok_or_else(|| { + std::io::Error::new(std::io::ErrorKind::Other, "No values received") + }), + None => Err(std::io::Error::new( + std::io::ErrorKind::Other, + "io_uring thread has terminated", + )), + } + } +} + +/// Creates an I2C sensor for telemetry_source_manager from the provided configuration. +/// +/// # Arguments +/// +/// * `sensor_config` - The sensor configuration. +/// * `all_sensor_frus` - A HashMap containing FRU (Field Replaceable Unit) information for sensors. +/// * `sampling_interval` - The interval at which to sample the sensor (in milliseconds). +/// * `reporting_interval` - The interval at which to report sensor values (in milliseconds). +/// +/// # Returns +/// +/// A Result containing the created Sensor instance or an error. +pub async fn create_i2c_sensor( + sensor_config: &SensorConfig, + all_sensor_frus: &HashMap<String, FruInfo>, + sampling_interval: u64, + reporting_interval: u64, +) -> Result<Box<TelemetrySource<f64>>, Box<dyn std::error::Error>> { + let sysfs_path = find_sensor_file(sensor_config)?; + let psu_property = sensor_config.reading_type.0.clone(); + let reader = Arc::new(I2cSensorReader::new( + sysfs_path, + psu_property, + Duration::from_millis(sampling_interval), + Duration::from_millis(reporting_interval), + )?); + + let mut secondary_keys = Vec::new(); + + // Add sensor's ReadingType as secondary key + secondary_keys.push(( + "ReadingType".to_string(), + sensor_config.reading_type.1.to_string(), + )); + + // Add sensor's type as secondary key + secondary_keys.push(("Type".to_string(), "I2C".to_string())); + + // Add sensor's FruInfo if available + if let Some(fru_info) = all_sensor_frus.get(&sensor_config.Name) { + if let Some(location_code) = &fru_info.location_code { + secondary_keys.push(("LocationCode".to_string(), location_code.clone())); + } + if let Some(service_label) = &fru_info.service_label { + secondary_keys.push(("ServiceLabel".to_string(), service_label.clone())); + } + } + + let sensor = TelemetrySource::new( + sensor_config.Name.clone(), + Some(secondary_keys), + reader, + Duration::from_millis(sampling_interval), + Duration::from_millis(1), + Duration::from_millis(reporting_interval), + ); + + Ok(sensor) +}
diff --git a/streaming_telemetry/src/telemetry_source_manager/mod.rs b/streaming_telemetry/src/telemetry_source_manager/mod.rs new file mode 100644 index 0000000..f44501e --- /dev/null +++ b/streaming_telemetry/src/telemetry_source_manager/mod.rs
@@ -0,0 +1,6 @@ +#![allow(clippy::module_inception)] +pub mod dbus_sensors; +pub mod i2c_sensors; +pub mod sensor_configs; +pub mod telemetry_source_manager; +pub mod telemetry_source_manager_api;
diff --git a/streaming_telemetry/src/telemetry_source_manager/sensor_configs.rs b/streaming_telemetry/src/telemetry_source_manager/sensor_configs.rs new file mode 100644 index 0000000..a14962a --- /dev/null +++ b/streaming_telemetry/src/telemetry_source_manager/sensor_configs.rs
@@ -0,0 +1,894 @@ +//! This module provides functionality for extracting sensor configurations from OpenBMC Entity +//! Manager's configuration files + +use lazy_static::lazy_static; +use log::debug; +use serde::{Deserialize, Serialize}; +use serde_json::Value; +use std::collections::HashMap; +use std::fs; +use std::io; +use std::path::{Path, PathBuf}; + +/// Represents the type of reading a sensor provides. +#[derive(Debug, Clone, Copy, Default, Serialize, Deserialize)] +pub enum ReadingType { + Voltage, + Current, + Power, + Temperature, + Fan, + #[default] + Unknown, +} + +impl std::fmt::Display for ReadingType { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + ReadingType::Voltage => write!(f, "Voltage"), + ReadingType::Current => write!(f, "Current"), + ReadingType::Power => write!(f, "Power"), + ReadingType::Temperature => write!(f, "Temperature"), + ReadingType::Fan => write!(f, "Fan"), + ReadingType::Unknown => write!(f, "Unknown"), + } + } +} + +/// Represents properties of a Power Supply Unit (PSU) sensor. +#[derive(Debug, Clone, Default, Deserialize, Serialize)] +pub struct PSUProperty { + pub name: String, + pub max: i32, + pub min: i32, + pub scale: i32, + pub offset: i32, +} + +/// Represents a sensor configuration. +#[allow(non_snake_case)] +#[derive(Debug, Clone, Default, Deserialize, Serialize)] +pub struct Sensor { + #[serde(default)] + pub Name: String, + #[serde(rename = "Type")] + pub sensor_type: String, + #[serde(default)] + pub Bus: Option<String>, + #[serde(default)] + pub Address: Option<String>, + #[serde(default)] + /// Use Labels[0] to select from this for the threshold for this sensor + pub Thresholds: Option<Vec<Threshold>>, + #[serde(default)] + pub Index: Option<u32>, + #[serde(default)] + pub Labels: Option<Vec<String>>, + #[serde(default)] + pub extra: HashMap<String, Value>, + #[serde(default)] + pub reading_type: (PSUProperty, ReadingType), +} + +/// Represents a threshold configuration for a sensor. +#[allow(non_snake_case)] +#[derive(Debug, Clone, Default, Deserialize, Serialize)] +pub struct Threshold { + pub Direction: String, + pub Name: String, + pub Severity: u32, + pub Value: f64, + #[serde(default)] + pub Index: Option<u32>, + #[serde(default)] + pub Label: Option<String>, +} + +#[rustfmt::skip] +lazy_static! { + static ref LABEL_MATCH: HashMap<&'static str, (PSUProperty, ReadingType)> = { + let mut m = HashMap::new(); + m.insert("pin", (PSUProperty { name: "Input Power".to_string(), max: 3000, min: 0, scale: 6, offset: 0 }, ReadingType::Power)); + m.insert("pin1", (PSUProperty { name: "Input Power".to_string(), max: 3000, min: 0, scale: 6, offset: 0 }, ReadingType::Power)); + m.insert("pin2", (PSUProperty { name: "Input Power".to_string(), max: 3000, min: 0, scale: 6, offset: 0 }, ReadingType::Power)); + m.insert("pout1", (PSUProperty { name: "Output Power".to_string(), max: 3000, min: 0, scale: 6, offset: 0 }, ReadingType::Power)); + m.insert("pout2", (PSUProperty { name: "Output Power".to_string(), max: 3000, min: 0, scale: 6, offset: 0 }, ReadingType::Power)); + m.insert("pout3", (PSUProperty { name: "Output Power".to_string(), max: 3000, min: 0, scale: 6, offset: 0 }, ReadingType::Power)); + m.insert("power1", (PSUProperty { name: "Output Power".to_string(), max: 3000, min: 0, scale: 6, offset: 0 }, ReadingType::Power)); + m.insert("power2", (PSUProperty { name: "Output Power".to_string(), max: 3000, min: 0, scale: 6, offset: 0 }, ReadingType::Power)); + m.insert("power3", (PSUProperty { name: "Output Power".to_string(), max: 3000, min: 0, scale: 6, offset: 0 }, ReadingType::Power)); + m.insert("power4", (PSUProperty { name: "Output Power".to_string(), max: 3000, min: 0, scale: 6, offset: 0 }, ReadingType::Power)); + m.insert("maxpin", (PSUProperty { name: "Max Input Power".to_string(), max: 3000, min: 0, scale: 6, offset: 0 }, ReadingType::Power)); + m.insert("vin", (PSUProperty { name: "Input Voltage".to_string(), max: 300, min: 0, scale: 3, offset: 0 }, ReadingType::Voltage)); + m.insert("vin1", (PSUProperty { name: "Input Voltage".to_string(), max: 300, min: 0, scale: 3, offset: 0 }, ReadingType::Voltage)); + m.insert("vin2", (PSUProperty { name: "Input Voltage".to_string(), max: 300, min: 0, scale: 3, offset: 0 }, ReadingType::Voltage)); + m.insert("maxvin", (PSUProperty { name: "Max Input Voltage".to_string(), max: 300, min: 0, scale: 3, offset: 0 }, ReadingType::Voltage)); + m.insert("in_voltage0", (PSUProperty { name: "Output Voltage".to_string(), max: 255, min: 0, scale: 3, offset: 0 }, ReadingType::Voltage)); + m.insert("in_voltage1", (PSUProperty { name: "Output Voltage".to_string(), max: 255, min: 0, scale: 3, offset: 0 }, ReadingType::Voltage)); + m.insert("in_voltage2", (PSUProperty { name: "Output Voltage".to_string(), max: 255, min: 0, scale: 3, offset: 0 }, ReadingType::Voltage)); + m.insert("in_voltage3", (PSUProperty { name: "Output Voltage".to_string(), max: 255, min: 0, scale: 3, offset: 0 }, ReadingType::Voltage)); + m.insert("vout1", (PSUProperty { name: "Output Voltage".to_string(), max: 255, min: 0, scale: 3, offset: 0 }, ReadingType::Voltage)); + m.insert("vout2", (PSUProperty { name: "Output Voltage".to_string(), max: 255, min: 0, scale: 3, offset: 0 }, ReadingType::Voltage)); + m.insert("vout3", (PSUProperty { name: "Output Voltage".to_string(), max: 255, min: 0, scale: 3, offset: 0 }, ReadingType::Voltage)); + m.insert("vout4", (PSUProperty { name: "Output Voltage".to_string(), max: 255, min: 0, scale: 3, offset: 0 }, ReadingType::Voltage)); + m.insert("vout5", (PSUProperty { name: "Output Voltage".to_string(), max: 255, min: 0, scale: 3, offset: 0 }, ReadingType::Voltage)); + m.insert("vout6", (PSUProperty { name: "Output Voltage".to_string(), max: 255, min: 0, scale: 3, offset: 0 }, ReadingType::Voltage)); + m.insert("vout7", (PSUProperty { name: "Output Voltage".to_string(), max: 255, min: 0, scale: 3, offset: 0 }, ReadingType::Voltage)); + m.insert("vout8", (PSUProperty { name: "Output Voltage".to_string(), max: 255, min: 0, scale: 3, offset: 0 }, ReadingType::Voltage)); + m.insert("vout9", (PSUProperty { name: "Output Voltage".to_string(), max: 255, min: 0, scale: 3, offset: 0 }, ReadingType::Voltage)); + m.insert("vout10", (PSUProperty { name: "Output Voltage".to_string(), max: 255, min: 0, scale: 3, offset: 0 }, ReadingType::Voltage)); + m.insert("vout11", (PSUProperty { name: "Output Voltage".to_string(), max: 255, min: 0, scale: 3, offset: 0 }, ReadingType::Voltage)); + m.insert("vout12", (PSUProperty { name: "Output Voltage".to_string(), max: 255, min: 0, scale: 3, offset: 0 }, ReadingType::Voltage)); + m.insert("vout13", (PSUProperty { name: "Output Voltage".to_string(), max: 255, min: 0, scale: 3, offset: 0 }, ReadingType::Voltage)); + m.insert("vout14", (PSUProperty { name: "Output Voltage".to_string(), max: 255, min: 0, scale: 3, offset: 0 }, ReadingType::Voltage)); + m.insert("vout15", (PSUProperty { name: "Output Voltage".to_string(), max: 255, min: 0, scale: 3, offset: 0 }, ReadingType::Voltage)); + m.insert("vout16", (PSUProperty { name: "Output Voltage".to_string(), max: 255, min: 0, scale: 3, offset: 0 }, ReadingType::Voltage)); + m.insert("vout17", (PSUProperty { name: "Output Voltage".to_string(), max: 255, min: 0, scale: 3, offset: 0 }, ReadingType::Voltage)); + m.insert("vout18", (PSUProperty { name: "Output Voltage".to_string(), max: 255, min: 0, scale: 3, offset: 0 }, ReadingType::Voltage)); + m.insert("vout19", (PSUProperty { name: "Output Voltage".to_string(), max: 255, min: 0, scale: 3, offset: 0 }, ReadingType::Voltage)); + m.insert("vout20", (PSUProperty { name: "Output Voltage".to_string(), max: 255, min: 0, scale: 3, offset: 0 }, ReadingType::Voltage)); + m.insert("vout21", (PSUProperty { name: "Output Voltage".to_string(), max: 255, min: 0, scale: 3, offset: 0 }, ReadingType::Voltage)); + m.insert("vout22", (PSUProperty { name: "Output Voltage".to_string(), max: 255, min: 0, scale: 3, offset: 0 }, ReadingType::Voltage)); + m.insert("vout23", (PSUProperty { name: "Output Voltage".to_string(), max: 255, min: 0, scale: 3, offset: 0 }, ReadingType::Voltage)); + m.insert("vout24", (PSUProperty { name: "Output Voltage".to_string(), max: 255, min: 0, scale: 3, offset: 0 }, ReadingType::Voltage)); + m.insert("vout25", (PSUProperty { name: "Output Voltage".to_string(), max: 255, min: 0, scale: 3, offset: 0 }, ReadingType::Voltage)); + m.insert("vout26", (PSUProperty { name: "Output Voltage".to_string(), max: 255, min: 0, scale: 3, offset: 0 }, ReadingType::Voltage)); + m.insert("vout27", (PSUProperty { name: "Output Voltage".to_string(), max: 255, min: 0, scale: 3, offset: 0 }, ReadingType::Voltage)); + m.insert("vout28", (PSUProperty { name: "Output Voltage".to_string(), max: 255, min: 0, scale: 3, offset: 0 }, ReadingType::Voltage)); + m.insert("vout29", (PSUProperty { name: "Output Voltage".to_string(), max: 255, min: 0, scale: 3, offset: 0 }, ReadingType::Voltage)); + m.insert("vout30", (PSUProperty { name: "Output Voltage".to_string(), max: 255, min: 0, scale: 3, offset: 0 }, ReadingType::Voltage)); + m.insert("vout31", (PSUProperty { name: "Output Voltage".to_string(), max: 255, min: 0, scale: 3, offset: 0 }, ReadingType::Voltage)); + m.insert("vout32", (PSUProperty { name: "Output Voltage".to_string(), max: 255, min: 0, scale: 3, offset: 0 }, ReadingType::Voltage)); + m.insert("vmon", (PSUProperty { name: "Auxiliary Input Voltage".to_string(), max: 255, min: 0, scale: 3, offset: 0 }, ReadingType::Voltage)); + m.insert("in0", (PSUProperty { name: "Output Voltage".to_string(), max: 255, min: 0, scale: 3, offset: 0 }, ReadingType::Voltage)); + m.insert("in1", (PSUProperty { name: "Output Voltage".to_string(), max: 255, min: 0, scale: 3, offset: 0 }, ReadingType::Voltage)); + m.insert("in2", (PSUProperty { name: "Output Voltage".to_string(), max: 255, min: 0, scale: 3, offset: 0 }, ReadingType::Voltage)); + m.insert("in3", (PSUProperty { name: "Output Voltage".to_string(), max: 255, min: 0, scale: 3, offset: 0 }, ReadingType::Voltage)); + m.insert("in4", (PSUProperty { name: "Output Voltage".to_string(), max: 255, min: 0, scale: 3, offset: 0 }, ReadingType::Voltage)); + m.insert("in5", (PSUProperty { name: "Output Voltage".to_string(), max: 255, min: 0, scale: 3, offset: 0 }, ReadingType::Voltage)); + m.insert("in6", (PSUProperty { name: "Output Voltage".to_string(), max: 255, min: 0, scale: 3, offset: 0 }, ReadingType::Voltage)); + m.insert("in7", (PSUProperty { name: "Output Voltage".to_string(), max: 255, min: 0, scale: 3, offset: 0 }, ReadingType::Voltage)); + m.insert("iin", (PSUProperty { name: "Input Current".to_string(), max: 20, min: 0, scale: 3, offset: 0 }, ReadingType::Current)); + m.insert("iin1", (PSUProperty { name: "Input Current".to_string(), max: 20, min: 0, scale: 3, offset: 0 }, ReadingType::Current)); + m.insert("iin2", (PSUProperty { name: "Input Current".to_string(), max: 20, min: 0, scale: 3, offset: 0 }, ReadingType::Current)); + m.insert("iout1", (PSUProperty { name: "Output Current".to_string(), max: 255, min: 0, scale: 3, offset: 0 }, ReadingType::Current)); + m.insert("iout2", (PSUProperty { name: "Output Current".to_string(), max: 255, min: 0, scale: 3, offset: 0 }, ReadingType::Current)); + m.insert("iout3", (PSUProperty { name: "Output Current".to_string(), max: 255, min: 0, scale: 3, offset: 0 }, ReadingType::Current)); + m.insert("iout4", (PSUProperty { name: "Output Current".to_string(), max: 255, min: 0, scale: 3, offset: 0 }, ReadingType::Current)); + m.insert("iout5", (PSUProperty { name: "Output Current".to_string(), max: 255, min: 0, scale: 3, offset: 0 }, ReadingType::Current)); + m.insert("iout6", (PSUProperty { name: "Output Current".to_string(), max: 255, min: 0, scale: 3, offset: 0 }, ReadingType::Current)); + m.insert("iout7", (PSUProperty { name: "Output Current".to_string(), max: 255, min: 0, scale: 3, offset: 0 }, ReadingType::Current)); + m.insert("iout8", (PSUProperty { name: "Output Current".to_string(), max: 255, min: 0, scale: 3, offset: 0 }, ReadingType::Current)); + m.insert("iout9", (PSUProperty { name: "Output Current".to_string(), max: 255, min: 0, scale: 3, offset: 0 }, ReadingType::Current)); + m.insert("iout10", (PSUProperty { name: "Output Current".to_string(), max: 255, min: 0, scale: 3, offset: 0 }, ReadingType::Current)); + m.insert("iout11", (PSUProperty { name: "Output Current".to_string(), max: 255, min: 0, scale: 3, offset: 0 }, ReadingType::Current)); + m.insert("iout12", (PSUProperty { name: "Output Current".to_string(), max: 255, min: 0, scale: 3, offset: 0 }, ReadingType::Current)); + m.insert("iout13", (PSUProperty { name: "Output Current".to_string(), max: 255, min: 0, scale: 3, offset: 0 }, ReadingType::Current)); + m.insert("iout14", (PSUProperty { name: "Output Current".to_string(), max: 255, min: 0, scale: 3, offset: 0 }, ReadingType::Current)); + m.insert("curr1", (PSUProperty { name: "Output Current".to_string(), max: 255, min: 0, scale: 3, offset: 0 }, ReadingType::Current)); + m.insert("curr2", (PSUProperty { name: "Output Current".to_string(), max: 255, min: 0, scale: 3, offset: 0 }, ReadingType::Current)); + m.insert("curr3", (PSUProperty { name: "Output Current".to_string(), max: 255, min: 0, scale: 3, offset: 0 }, ReadingType::Current)); + m.insert("curr4", (PSUProperty { name: "Output Current".to_string(), max: 255, min: 0, scale: 3, offset: 0 }, ReadingType::Current)); + m.insert("maxiout1", (PSUProperty { name: "Max Output Current".to_string(), max: 255, min: 0, scale: 3, offset: 0 }, ReadingType::Current)); + m.insert("temp1", (PSUProperty { name: "Temperature".to_string(), max: 127, min: -128, scale: 3, offset: 0 }, ReadingType::Temperature)); + m.insert("temp2", (PSUProperty { name: "Temperature".to_string(), max: 127, min: -128, scale: 3, offset: 0 }, ReadingType::Temperature)); + m.insert("temp3", (PSUProperty { name: "Temperature".to_string(), max: 127, min: -128, scale: 3, offset: 0 }, ReadingType::Temperature)); + m.insert("temp4", (PSUProperty { name: "Temperature".to_string(), max: 127, min: -128, scale: 3, offset: 0 }, ReadingType::Temperature)); + m.insert("temp5", (PSUProperty { name: "Temperature".to_string(), max: 127, min: -128, scale: 3, offset: 0 }, ReadingType::Temperature)); + m.insert("temp6", (PSUProperty { name: "Temperature".to_string(), max: 127, min: -128, scale: 3, offset: 0 }, ReadingType::Temperature)); + m.insert("maxtemp1", (PSUProperty { name: "Max Temperature".to_string(), max: 127, min: -128, scale: 3, offset: 0 }, ReadingType::Temperature)); + m.insert("fan1", (PSUProperty { name: "Fan Speed 1".to_string(), max: 30000, min: 0, scale: 0, offset: 0 }, ReadingType::Fan)); + m.insert("fan2", (PSUProperty { name: "Fan Speed 2".to_string(), max: 30000, min: 0, scale: 0, offset: 0 }, ReadingType::Fan)); + m.insert("fan3", (PSUProperty { name: "Fan Speed 3".to_string(), max: 30000, min: 0, scale: 0, offset: 0 }, ReadingType::Fan)); + m.insert("fan4", (PSUProperty { name: "Fan Speed 4".to_string(), max: 30000, min: 0, scale: 0, offset: 0 }, ReadingType::Fan)); + m + }; +} + +/// Infers the reading type of a sensor based on its configuration. +/// +/// # Arguments +/// +/// * `sensor` - The sensor configuration to infer the reading type from. +/// +/// # Returns +/// +/// A tuple containing the PSUProperty and ReadingType for the sensor. +fn infer_reading_type(sensor: &Sensor) -> (PSUProperty, ReadingType) { + // First, try to match using the Labels field + if let Some(labels) = &sensor.Labels { + if let Some(first_label) = labels.first() { + if let Some((psu_property, reading_type)) = LABEL_MATCH.get(first_label.as_str()) { + return (psu_property.clone(), *reading_type); + } + + // If no exact match, try to match prefixes + for (label, (psu_property, reading_type)) in LABEL_MATCH.iter() { + if first_label.starts_with(label) { + return (psu_property.clone(), *reading_type); + } + } + } + } + + // If Labels matching fails, fall back to matching based on the sensor Name + let name_lower = sensor.Name.to_lowercase(); + + // Fallback logic + if name_lower.contains("temp") || name_lower.ends_with('t') { + if let Some((psu_property, reading_type)) = LABEL_MATCH.get("temp1") { + return (psu_property.clone(), *reading_type); + } + } else if name_lower.contains("curr") + || name_lower.contains("iout") + || name_lower.ends_with('i') + { + if let Some((psu_property, reading_type)) = LABEL_MATCH.get("curr1") { + return (psu_property.clone(), *reading_type); + } + } else if name_lower.contains("power") + || name_lower.contains("pout") + || name_lower.contains("pin") + { + if let Some((psu_property, reading_type)) = LABEL_MATCH.get("power1") { + return (psu_property.clone(), *reading_type); + } + } else if name_lower.contains("volt") + || name_lower.contains("vin") + || name_lower.contains("vout") + || name_lower.ends_with('v') + { + if let Some((psu_property, reading_type)) = LABEL_MATCH.get("vin") { + return (psu_property.clone(), *reading_type); + } + } else if name_lower.contains("fan") || name_lower.contains("tach") { + if let Some((psu_property, reading_type)) = LABEL_MATCH.get("fan1") { + return (psu_property.clone(), *reading_type); + } + } + + (PSUProperty::default(), ReadingType::Unknown) +} + +/// Extracts sensor configurations from the 'Exposes' field of a configuration object. +/// +/// # Arguments +/// +/// * `exposes` - A slice of JSON Value objects containing sensor configurations. +/// +/// # Returns +/// +/// A vector of Sensor structs extracted from the configuration. +#[rustfmt::skip] +fn extract_sensors_from_exposes(exposes: &[Value]) -> Vec<Sensor> { + let mut sensors = Vec::new(); + + for expose in exposes { + debug!("Captured expose: {:?}", expose); + // Handle cases where there are multiple sensors defined within a single expose object, each corresponding to a label, like "P0_ADM1266". + if let Some(labels) = expose.get("Labels").and_then(|v| v.as_array()) { + for label in labels { + debug!("Captured expose by label: {:?}", label); + if let Some(label_name) = label.as_str() { + let name_key = format!("{}_Name", label_name); + let sensor_name = expose + .get(&name_key) + .and_then(|v| v.as_str()) + .unwrap_or(label_name) + .to_string(); + + let mut sensor = Sensor { + Name: sensor_name.clone(), + sensor_type: expose + .get("Type") + .and_then(|v| v.as_str()) + .unwrap_or("") + .to_string(), + Bus: expose.get("Bus").and_then(|v| match v { + Value::String(s) => Some(s.clone()), + Value::Number(n) => n.as_u64().map(|n| n.to_string()), + _ => None, + }), + Address: expose + .get("Address") + .and_then(|v| v.as_str().map(String::from)), + Thresholds: expose + .get("Thresholds") + .and_then(|v| serde_json::from_value(v.clone()).ok()), + Index: None, + Labels: Some(vec![label_name.to_string()]), + extra: HashMap::new(), + reading_type: Default::default(), + }; + + sensor.reading_type = infer_reading_type(&sensor); + debug!("Add captured expose by labels: {:?}", sensor); + sensors.push(sensor); + } + } + } + // Try to deserialize the entire expose object into a Sensor struct, like "fan2_tach". + else if let Ok(mut sensor) = serde_json::from_value::<Sensor>(expose.clone()) { + debug!("Captured expose by Sensor type: {:?}", expose); + // Filter for valid sensor types, some entry in config file are not sensors, like + // MAX31790, use this to filter them out + let valid_hwmon_types = [ + "ADC", "MAX31725", "MAX31730", "TMP75", "TMP421", "TMP441", "TMP112", "EMC1412", + "EMC1413", "ADM1032", "ADT7470", "G781", "MAX6581", "MAX6654", "LM75", "LM95245", + "ADT7460", "ADT7490", "W83795G", "SI7020", "NCT7802", "pmbus", "ADM1266", "ADM1272", + "ADM1275", "ADM1278", "ADM1293", "ADM1294", "IR35221", "IR38062", "IR38164", "IR38263", + "ISL68137", "ISL68220", "ISL69225", "ISL69227", "ISL69228", "ISL69234", "ISL69236", + "ISL69239", "ISL69243", "ISL69247", "ISL69254", "ISL69259", "ISL69260", "LM25066", + "RAA228000", "RAA228004", "RAA228006", "RAA228228", "RAA229001", "RAA229004", + "TPS53679", "TPS53688", "TPS53681", "TPS53622", "TPS53819", "TPS53915", "TPS53916", + "TPS53647", "TPS53667", "TPS53676", "TPS53661", "TPS53641", "TPS53631", "TPS53625", + "TPS53M30", "UCD90120", "UCD90124", "UCD90160", "UCD90320", "UCD9090", "UCD90910", + "XDPE11280", "XDPE12284", "XDPE132G5C", "MAX20730", "MAX20734", "MAX20743", + "MAX34451", "MAX34460", "MAX34461", "MAX77836", "HwmonTempSensor", "VirtualSensor", + "ADCSensor", "FanSensor", "NVMe", "TMP431", "I2CFan" + ]; + + let valid_pldm_types = ["PLDM"]; + + let valid_other_types = [ + "ExternalSensor", "Fan", "System", "DownstreamPort", + "AgoraV3AD Upstream Port", "SuperBigGulp Upstream Port", + "MyMachine CPLD Upstream Port", "MyMachine 12 Upstream Port", + "MyMachine 13 Upstream Port", "MyMachine 8 Upstream Port", + "MyMachine 11 Upstream Port" + ]; + + if valid_hwmon_types.contains(&sensor.sensor_type.as_str()) { + // Handle I2CFan type specifically + if sensor.sensor_type == "I2CFan" { + if let Some(connector) = expose.get("Connector") { + if let Some(pwm_name) = connector.get("PwmName").and_then(|v| v.as_str()) { + let mut pwm_sensor = sensor.clone(); + pwm_sensor.Name = pwm_name.to_string(); + pwm_sensor.sensor_type = "PWMSensor".to_string(); + pwm_sensor.reading_type = infer_reading_type(&pwm_sensor); + debug!("Add PWM sensor: {:?}", pwm_sensor); + sensors.push(pwm_sensor); + } + } + } + sensor.reading_type = infer_reading_type(&sensor); + debug!("Add captured expose {:?} by Sensor type, sensor: {:?}", expose, sensor); + } else if valid_pldm_types.contains(&sensor.sensor_type.as_str()) { + sensor.sensor_type = "PLDM".to_string(); + sensor.reading_type = infer_reading_type(&sensor); + } else if valid_other_types.contains(&sensor.sensor_type.as_str()) { + // FIXME: hardcode sensor NAme for PWMSensor from config + // Add captured expose by Sensor type: Object {"HotPluggable": Bool(false), "LocationType": String("Connector"), "Name": String("Fan3"), "PWMSensor": String("fan3_pwm"), "ServiceLabel": String("FAN3"), "TachSensor": String("fan3_tach"), "Type": String("Fan")} + // extract_sensors, config /usr/share/entity-manager/configurations/heizo.json, all sensors from config: Sensor { Name: "Fan0", sensor_type: "Fan", Bus: None, Address: None, Thresholds: None, Index: None, Labels: None, extra: {}, reading_type: (PSUProperty { name: "Fan Speed 1", max: 30000, min: 0, scale: 0, offset: 0 }, Fan) } + if sensor.sensor_type == "Fan" { + let pwm_sensor = expose.get("PWMSensor").and_then(|p| p.as_str()).unwrap_or("").to_owned(); + sensor.Name = pwm_sensor; + sensor.sensor_type = "PWMSensor".to_string(); + } + } else { + debug!("Skip captured expose: {:?}", expose); + continue; // Skip sensors with unrecognized types + } + + sensor.reading_type = infer_reading_type(&sensor); + debug!("Add captured expose {:?} by Sensor type, sensor: {:?}", expose, sensor); + sensors.push(sensor); + } + // A catch-all for sensors that don't fit into the first two categories. + else if let Some(sensor_type) = expose.get("Type").and_then(|t| t.as_str()) { + debug!("Captured expose by Type: {:?}", expose); + + let labels = expose.get("Labels").and_then(|v| v.as_array()).map(|arr| { + arr.iter() + .filter_map(|v| v.as_str()) + .map(String::from) + .collect::<Vec<String>>() + }); + + let mut label_map = HashMap::new(); + if let Some(labels) = &labels { + for label in labels { + let name_key = format!("{}_Name", label); + if let Some(sensor_name) = expose.get(&name_key).and_then(|v| v.as_str()) { + label_map.insert(label.clone(), sensor_name.to_string()); + } + } + } + + let bus = expose.get("Bus").and_then(|b| match b { + Value::Number(n) => Some(n.to_string()), + Value::String(s) => Some(s.clone()), + _ => None, + }); + let address = expose + .get("Address") + .and_then(|a| a.as_str().map(String::from)); + + // Create individual sensors for each label + for (label, sensor_name) in label_map { + let mut thresholds = Vec::new(); + if let Some(thresh_array) = expose.get("Thresholds").and_then(|t| t.as_array()) { + for thresh in thresh_array { + if let Ok(t) = serde_json::from_value::<Threshold>(thresh.clone()) { + if t.Label.as_ref().map(|l| l == &label).unwrap_or(false) { + thresholds.push(t); + } + } + } + } + + let mut sensor = Sensor { + reading_type: Default::default(), + Name: sensor_name.clone(), + sensor_type: sensor_type.to_string(), + Bus: bus.clone(), + Address: address.clone(), + Thresholds: if thresholds.is_empty() { + None + } else { + Some(thresholds) + }, + Index: None, + Labels: Some(vec![label]), + extra: HashMap::new(), + }; + sensor.reading_type = infer_reading_type(&sensor); + + debug!("Add captured expose by Type and label: {:?}", sensor); + sensors.push(sensor); + } + } + } + + sensors +} + +/// Recursively extracts sensor configurations from a JSON Value. +/// +/// # Arguments +/// +/// * `value` - A JSON Value that may contain sensor configurations. +/// +/// # Returns +/// +/// A vector of Sensor structs extracted from the Value. +fn extract_sensors_from_value(value: &Value) -> Vec<Sensor> { + match value { + Value::Array(arr) => arr.iter().flat_map(extract_sensors_from_value).collect(), + Value::Object(_) => extract_sensors_from_exposes(&[value.clone()]), + _ => Vec::new(), + } +} + +/// Extracts sensor configurations from a JSON configuration file. +/// +/// # Arguments +/// +/// * `config_file` - The path to the JSON configuration file. +/// +/// # Returns +/// +/// A Result containing a vector of Sensor structs or an error. +pub async fn extract_sensors(config_file: &str) -> Result<Vec<Sensor>, Box<dyn std::error::Error>> { + let file_content = tokio::fs::read_to_string(config_file).await?; + let config: Value = serde_json::from_str(&file_content)?; + + let sensors = match config { + Value::Array(configs) => { + let mut all_sensors = Vec::new(); + for c in configs { + if let Some(exposes) = c.get("Exposes") { + all_sensors.extend(extract_sensors_from_value(exposes)); + } + } + all_sensors + } + Value::Object(config) => { + if let Some(exposes) = config.get("Exposes") { + extract_sensors_from_value(exposes) + } else { + Vec::new() + } + } + _ => Vec::new(), + }; + + Ok(sensors) +} + +/// Finds the sysfs file path for a given sensor. +/// +/// # Arguments +/// +/// * `sensor` - The Sensor configuration to find the file for. +/// +/// # Returns +/// +/// A Result containing the sysfs file path as a String or an error. +pub fn find_sensor_file(sensor: &Sensor) -> io::Result<String> { + debug!("find_sensor_file for sensor {:?}", sensor); + if let (Some(bus), Some(address)) = (&sensor.Bus, &sensor.Address) { + // Strip "0x" prefix if present + let address_stripped = address.trim_start_matches("0x"); + + let base_path = Path::new("/sys/class/hwmon"); + let result = search_hwmon(base_path, bus, address_stripped, sensor); + if result.is_ok() { + debug!( + "find_sensor_file in hwmon for sensor {:?}, found result {:?}", + sensor, result + ); + return result; + } + + // For /sys/bus/i2c/devices + let base_path = Path::new("/sys/bus/i2c/devices"); + let result = search_i2c(base_path, bus, address_stripped, sensor); + if result.is_ok() { + debug!( + "find_sensor_file in i2c for sensor {:?}, found result {:?}", + sensor, result + ); + return result; + } + } + + Err(io::Error::new( + io::ErrorKind::NotFound, + "Sensor file not found", + )) +} + +/// Searches for a sensor file in the hwmon directory. +/// +/// # Arguments +/// +/// * `base_path` - The base path to start the search from. +/// * `bus` - The bus identifier. +/// * `address_stripped` - The stripped address of the sensor. +/// * `sensor` - The Sensor configuration to search for. +/// +/// # Returns +/// +/// A Result containing the sensor file path as a String or an error. +fn search_hwmon( + base_path: &Path, + bus: &str, + address_stripped: &str, + sensor: &Sensor, +) -> io::Result<String> { + let entries = fs::read_dir(base_path)?; + + for entry in entries { + let entry = entry?; + let symlink_path = entry.path(); + let real_path = fs::read_link(&symlink_path)?; + let full_real_path = base_path.join(&real_path); + + if let Some(path_str) = full_real_path.to_str() { + if path_str.contains(bus) && path_str.contains(address_stripped) { + let sensor_file = find_specific_sensor_file(&full_real_path, sensor)?; + return Ok(sensor_file.to_string_lossy().into_owned()); + } + } + } + Err(io::Error::new( + io::ErrorKind::NotFound, + "Sensor file not found in hwmon", + )) +} + +/// Searches for a sensor file in the i2c devices directory. +/// +/// # Arguments +/// +/// * `base_path` - The base path to start the search from. +/// * `bus` - The bus identifier. +/// * `address` - The address of the sensor. +/// * `sensor` - The Sensor configuration to search for. +/// +/// # Returns +/// +/// A Result containing the sensor file path as a String or an error. +fn search_i2c(base_path: &Path, bus: &str, address: &str, sensor: &Sensor) -> io::Result<String> { + let i2c_path = base_path.join(format!("i2c-{}", bus)); + let device_path = i2c_path.join(format!("{}-00{}", bus, address)); + + debug!( + "search_i2c: hwmon_dir {:?} for sensor {:?}", + device_path.join("hwmon"), + sensor.Name + ); + let hwmon_dir = fs::read_dir(device_path.join("hwmon"))? + .filter_map(Result::ok) + .next() + .ok_or_else(|| io::Error::new(io::ErrorKind::NotFound, "Hwmon directory not found"))?; + + let sensor_file = find_specific_sensor_file(&hwmon_dir.path(), sensor)?; + Ok(sensor_file.to_string_lossy().into_owned()) +} + +/// Finds the specific sensor file within a hwmon directory. +/// +/// # Arguments +/// +/// * `hwmon_path` - The path to the hwmon directory. +/// * `sensor` - The Sensor configuration to search for. +/// +/// # Returns +/// +/// A Result containing the PathBuf of the sensor file or an error. +fn find_specific_sensor_file(hwmon_path: &Path, sensor: &Sensor) -> io::Result<PathBuf> { + let hwmon_dir = fs::read_dir(hwmon_path)?; + let file_pattern = get_file_pattern(sensor); + debug!( + "Matching file_pattern {:?} under hwmon_dir {:?} for sensor name {}", + file_pattern, hwmon_dir, sensor.Name + ); + + for file in hwmon_dir { + let file = file?; + if file.file_name().to_string_lossy().contains(&file_pattern) { + debug!( + "Matching file {:?} under hwmon_path: {:?} for sensor name {}", + file, hwmon_path, sensor.Name + ); + return Ok(file.path()); + } + } + + Err(io::Error::new( + io::ErrorKind::NotFound, + "Specific sensor file not found", + )) +} + +/// Gets the file pattern for a normal (non-pmbus) sensor. +/// +/// # Arguments +/// +/// * `sensor` - The Sensor configuration to get the pattern for. +/// +/// # Returns +/// +/// A String containing the file pattern for the sensor. +fn get_file_pattern_normal(sensor: &Sensor) -> String { + let mut index = 1; // Default index if not specified + + if let Some(sensor_index) = sensor.Index { + index = sensor_index; + } else if let Some(labels) = &sensor.Labels { + // Assert that there's only one label + // SAFEFY: this should only be called at server initialization, so it can unwrap + assert!( + labels.len() == 1, + "Expected exactly one label, found {}", + labels.len() + ); + + // Extract the number suffix from the label + if let Some(label) = labels.first() { + if let Ok(number_str) = label + .chars() + .skip_while(|c| !c.is_numeric()) + .collect::<String>() + .parse::<u32>() + { + index = number_str; + } + } + } + + debug!("sensor {} has type {}", sensor.Name, sensor.sensor_type); + match sensor.reading_type.1 { + ReadingType::Voltage => format!("in{}_input", index), + ReadingType::Current => format!("curr{}_input", index), + ReadingType::Power => format!("power{}_input", index), + ReadingType::Temperature => format!("temp{}_input", index), + ReadingType::Fan => format!("fan{}_input", index + 1), + _ => format!("in{}_input", index), // Default to voltage for unknown types + } +} + +/// Gets the file pattern for a sensor. +/// +/// # Arguments +/// +/// * `sensor` - The Sensor configuration to get the pattern for. +/// +/// # Returns +/// +/// A String containing the file pattern for the sensor. +fn get_file_pattern(sensor: &Sensor) -> String { + match sensor.sensor_type.as_str() { + // pmbus has special mapping rule, not worthy to include them in the normal + // handling logic + "pmbus" => { + if let Some(labels) = &sensor.Labels { + if let Some(label) = labels.first() { + match label.as_str() { + "vout1" => "in2_input".to_string(), + "iout1" => "curr2_input".to_string(), + "vin" => "in1_input".to_string(), + "pin" | "pout1" => "power2_input".to_string(), + "temp1" => "temp1_input".to_string(), + _ => "in1_input".to_string(), + } + } else { + "in1_input".to_string() + } + } else { + "in1_input".to_string() + } + } + _ => + // logic for normal sensor types + { + get_file_pattern_normal(sensor) + } + } +} + +/// Represents Field Replaceable Unit (FRU) information for a sensor. +#[derive(Debug, Clone)] +pub struct FruInfo { + /// FRU name for this device + pub service_label: Option<String>, + /// FRU name of its parent device, if this is an embedded device + pub location_code: Option<String>, + /// The Concatenation of all Parent FRU's ServiceLabels until root + #[allow(dead_code)] + pub part_location_context: Option<String>, +} + +/// Extracts FRU information for sensors from a configuration file. +/// +/// TODO: not fully verified yet +/// +/// # Arguments +/// +/// * `config_file` - The path to the configuration file. +/// +/// # Returns +/// +/// A Result containing a HashMap of sensor names to FruInfo or an error. +pub fn extract_sensor_frus( + config_file: &str, +) -> Result<HashMap<String, FruInfo>, Box<dyn std::error::Error>> { + let file_content = std::fs::read_to_string(config_file)?; + let config: Value = serde_json::from_str(&file_content)?; + + let mut sensor_frus = HashMap::new(); + let mut sensor_to_fan_map = HashMap::new(); + + if let Value::Array(configs) = config { + for c in configs { + process_component(&c, &mut sensor_frus, &mut sensor_to_fan_map)?; + } + } + + // Update sensor FRU info with fan service labels + let fan_service_labels: HashMap<_, _> = sensor_frus + .iter() + .filter_map(|(name, info)| { + info.service_label + .as_ref() + .map(|label| (name.clone(), label.clone())) + }) + .collect(); + + for (sensor_name, fan_name) in sensor_to_fan_map { + if let Some(fru_info) = sensor_frus.get_mut(&sensor_name) { + if let Some(service_label) = fan_service_labels.get(&fan_name) { + fru_info.service_label = Some(service_label.clone()); + } + } + } + + Ok(sensor_frus) +} + +/// Processes a component from the configuration and extracts sensor FRU information. +/// +/// TODO: not fully verified yet +/// +/// # Arguments +/// +/// * `component` - The JSON Value representing the component. +/// * `sensor_frus` - A mutable reference to the HashMap storing sensor FRU information. +/// * `sensor_to_fan_map` - A mutable reference to the HashMap mapping sensors to fans. +/// +/// # Returns +/// +/// A Result indicating success or an error. +fn process_component( + component: &Value, + sensor_frus: &mut HashMap<String, FruInfo>, + sensor_to_fan_map: &mut HashMap<String, String>, +) -> Result<(), Box<dyn std::error::Error>> { + if let Some(location_code) = component + .get("xyz.openbmc_project.Inventory.Decorator.LocationCode") + .and_then(|l| l.get("LocationCode")) + .and_then(|l| l.as_str()) + { + let fru_info = FruInfo { + location_code: Some(location_code.to_string()), + service_label: None, + part_location_context: None, + }; + + if let Some(exposes) = component.get("Exposes").and_then(|e| e.as_array()) { + for expose in exposes { + if let Some(expose_type) = expose.get("Type").and_then(|t| t.as_str()) { + match expose_type { + "Fan" => { + let fan_name = + expose.get("Name").and_then(|n| n.as_str()).unwrap_or(""); + let service_label = expose.get("ServiceLabel").and_then(|s| s.as_str()); + let tach_sensor = expose.get("TachSensor").and_then(|t| t.as_str()); + let pwm_sensor = expose.get("PWMSensor").and_then(|p| p.as_str()); + + let mut fan_fru_info = fru_info.clone(); + fan_fru_info.service_label = service_label.map(String::from); + + sensor_frus.insert(fan_name.to_string(), fan_fru_info); + + if let Some(tach) = tach_sensor { + sensor_to_fan_map.insert(tach.to_string(), fan_name.to_string()); + } + if let Some(pwm) = pwm_sensor { + sensor_to_fan_map.insert(pwm.to_string(), fan_name.to_string()); + } + } + _ => { + extract_sensors_from_expose(expose, sensor_frus, &fru_info); + } + } + } + } + } + } + + Ok(()) +} + +/// Extracts sensor information from an 'Expose' object in the configuration. +/// +/// TODO: not fully verified yet +/// +/// # Arguments +/// +/// * `expose` - The JSON Value representing the 'Expose' object. +/// * `sensor_frus` - A mutable reference to the HashMap storing sensor FRU information. +/// * `fru_info` - The FruInfo for the current component. +fn extract_sensors_from_expose( + expose: &Value, + sensor_frus: &mut HashMap<String, FruInfo>, + fru_info: &FruInfo, +) { + if let Some(sensor_type) = expose.get("Type").and_then(|t| t.as_str()) { + match sensor_type { + "ADM1266" | "pmbus" | "ADM1272" => { + if let Some(labels) = expose.get("Labels").and_then(|l| l.as_array()) { + for label in labels { + if let Some(label_str) = label.as_str() { + let name_key = format!("{}_Name", label_str); + if let Some(sensor_name) = + expose.get(&name_key).and_then(|n| n.as_str()) + { + sensor_frus.insert(sensor_name.to_string(), fru_info.clone()); + } + } + } + } + } + "PLDM" | "MAX31725" | "TMP431" | "I2CFan" | "ADC" | "ExternalSensor" => { + if let Some(sensor_name) = expose.get("Name").and_then(|n| n.as_str()) { + sensor_frus.insert(sensor_name.to_string(), fru_info.clone()); + } + } + _ => {} + } + } +}
diff --git a/streaming_telemetry/src/telemetry_source_manager/telemetry_source_manager.rs b/streaming_telemetry/src/telemetry_source_manager/telemetry_source_manager.rs new file mode 100644 index 0000000..7a54330 --- /dev/null +++ b/streaming_telemetry/src/telemetry_source_manager/telemetry_source_manager.rs
@@ -0,0 +1,1609 @@ +#![doc = include_str!("README.md")] + +use anyhow::Result; +use dashmap::DashMap; +use std::any::Any; +use std::collections::HashMap; +use std::fmt; +use std::fmt::Debug; +use std::marker::PhantomData; +use std::sync::Mutex; +use std::sync::{ + atomic::{AtomicBool, AtomicUsize, Ordering}, + Arc, +}; +use std::time::{Duration, SystemTime}; +use tokio::sync::mpsc::{Receiver, Sender}; +use tokio::sync::{mpsc, Notify, RwLock}; + +/// A trait for asynchronous value reading. +#[async_trait::async_trait] +pub trait AsyncReadValue<T: Clone + Send + 'static>: Send + Sync { + /// Read values and return them with timestamps. + /// + /// # Arguments + /// + /// * `sampling_interval` - The interval between each sample. + /// * `reporting_interval` - The interval for reporting batched values. + /// + /// # Returns + /// + /// A vector of values with their corresponding timestamps. + async fn read_values( + &self, + sampling_interval: Duration, + reporting_interval: Duration, + ) -> Result<Vec<(T, SystemTime)>, std::io::Error>; + + /// Read current value and return it with timestamp. + /// + /// # Returns + /// + /// A value with its corresponding timestamp. + #[allow(dead_code)] + async fn get_value(&self) -> Result<(T, SystemTime), std::io::Error>; +} + +/// Telemetry Source Data Model. +pub struct TelemetrySource<T: Clone + Send + 'static> { + /// Primary key. + name: String, + /// Secondary keys, like Some(vec![("FRU", "CPU"), ("ReadingType", "Temperature")]). + secondary_keys: Option<Vec<(String, String)>>, + /// Client provided reader to get telemetry source value update. + value_reader: Arc<dyn AsyncReadValue<T>>, + /// Default update interval for polling telemetry source value. + default_update_interval: Duration, + /// Lowest limit of update interval for telemetry source HW. + minimum_update_interval: Duration, + /// Flag to stop background telemetry source HW polling task. + stop_flag: Arc<AtomicBool>, + /// Current update interval for polling telemetry source value. + current_update_interval: AtomicUsize, // Store in milliseconds + /// Current reporting interval for batch reporting. + current_reporting_interval: AtomicUsize, // Store in milliseconds + /// Flag to wake up the polling task. + wakeup_flag: Arc<Notify>, + /// Flag to indicate if the telemetry source is currently polling. + is_polling: Arc<std::sync::RwLock<bool>>, +} + +impl<T: Clone + Send + 'static> TelemetrySource<T> { + /// Creates a new `TelemetrySource`. + /// + /// # Arguments + /// + /// * `name` - A string that holds the primary key for the telemetry source. + /// * `secondary_keys` - An optional vector of key-value pairs representing additional keys. + /// * `value_reader` - An `Arc` containing a trait object that implements `AsyncReadValue`. + /// * `default_update_interval` - The default duration between value updates. + /// * `minimum_update_interval` - The minimum allowable duration between value updates. + /// * `reporting_interval` - The interval of batch reporting of values. + /// + /// # Returns + /// + /// A new `Box<TelemetrySource>` instance. + pub fn new( + name: String, + secondary_keys: Option<Vec<(String, String)>>, + value_reader: Arc<dyn AsyncReadValue<T>>, + default_update_interval: Duration, + minimum_update_interval: Duration, + reporting_interval: Duration, + ) -> Box<Self> { + Box::new(Self { + name, + secondary_keys, + value_reader, + default_update_interval, + minimum_update_interval, + stop_flag: Arc::new(AtomicBool::new(false)), + current_update_interval: AtomicUsize::new(default_update_interval.as_millis() as usize), + current_reporting_interval: AtomicUsize::new(reporting_interval.as_millis() as usize), + wakeup_flag: Arc::new(Notify::new()), + is_polling: Arc::new(std::sync::RwLock::new(false)), + }) + } +} + +impl<T: Clone + Send + 'static> fmt::Debug for TelemetrySource<T> { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("TelemetrySource") + .field("name", &self.name) + .field("secondary_keys", &self.secondary_keys) + .field("default_update_interval", &self.default_update_interval) + .field("minimum_update_interval", &self.minimum_update_interval) + .field("current_update_interval", &self.get_current_interval()) + .field( + "current_reporting_interval", + &self.get_current_reporting_interval(), + ) + .finish() + } +} + +/// Define the TelemetrySourceTrait for type erase purpose, allowing different types +/// of telemetry sources to be stored and managed uniformly in the TelemetrySourceManager +pub trait TelemetrySourceTrait: Send + Sync { + /// Gets the name of the telemetry source. + fn get_name(&self) -> &str; + + /// Gets the secondary keys of the telemetry source. + fn get_secondary_keys(&self) -> Option<&Vec<(String, String)>>; + + /// Sets the stop flag for the telemetry source. + #[allow(dead_code)] + fn set_stop_flag(&self, value: bool); + + /// Gets the current polling interval. + fn get_current_interval(&self) -> Duration; + + /// Sets the current polling interval. + fn set_current_interval(&self, new_interval: Duration); + + /// Gets the current reporting interval. + fn get_current_reporting_interval(&self) -> Duration; + + /// Sets the current reporting interval. + fn set_current_reporting_interval(&self, new_interval: Duration); + + /// Gets the default update interval. + fn get_default_update_interval(&self) -> Duration; + + /// Gets the wakeup flag for the telemetry source. + fn get_wakeup_flag(&self) -> Arc<Notify>; + + /// Gets the is_polling flag for the telemetry source. + fn get_is_polling(&self) -> Arc<std::sync::RwLock<bool>>; + + /// Converts the trait object to Any. + fn as_any(&self) -> &dyn Any; +} + +impl<T: Clone + Send + 'static> TelemetrySourceTrait for TelemetrySource<T> { + fn get_name(&self) -> &str { + &self.name + } + + fn get_secondary_keys(&self) -> Option<&Vec<(String, String)>> { + self.secondary_keys.as_ref() + } + + fn set_current_interval(&self, new_interval: Duration) { + let new_interval_ms = std::cmp::max( + new_interval.as_millis(), + self.minimum_update_interval.as_millis(), + ); + let new_interval_ms = + std::cmp::min(new_interval_ms, self.default_update_interval.as_millis()) as usize; + self.current_update_interval + .store(new_interval_ms, Ordering::SeqCst); + if self.current_reporting_interval.load(Ordering::SeqCst) < new_interval_ms { + self.set_current_reporting_interval(Duration::from_millis(new_interval_ms as u64)); + } + } + + fn set_current_reporting_interval(&self, new_interval: Duration) { + let new_interval_ms = std::cmp::max( + new_interval.as_millis() as usize, + self.current_update_interval.load(Ordering::SeqCst), + ); + self.current_reporting_interval + .store(new_interval_ms, Ordering::SeqCst); + } + + fn get_current_interval(&self) -> Duration { + Duration::from_millis(self.current_update_interval.load(Ordering::SeqCst) as u64) + } + + fn get_current_reporting_interval(&self) -> Duration { + Duration::from_millis(self.current_reporting_interval.load(Ordering::SeqCst) as u64) + } + + fn get_default_update_interval(&self) -> Duration { + self.default_update_interval + } + + fn set_stop_flag(&self, value: bool) { + self.stop_flag.store(value, Ordering::SeqCst); + } + + fn get_wakeup_flag(&self) -> Arc<Notify> { + self.wakeup_flag.clone() + } + + fn get_is_polling(&self) -> Arc<std::sync::RwLock<bool>> { + self.is_polling.clone() + } + + fn as_any(&self) -> &dyn Any { + self + } +} + +/// Telemetry struct representing a value change or periodic update. +#[derive(Clone, Debug, PartialEq)] +pub struct TypedTelemetry<T: Clone + Send + 'static> { + /// Telemetry source name associated with the telemetry. + pub source_name: String, + /// Telemetry type as defined in Redfish spec for the Telemetry. + pub telemetry_type: String, + /// Value associated with the telemetry. + pub value: T, + /// Sampling timestamp. + pub timestamp: SystemTime, +} + +/// This trait allows for type-agnostic telemetry handling, enabling the system to work +/// with different types of telemetries without knowing their specific types. +pub trait Telemetry: Send + Debug { + /// Gets the source name of the telemetry. + #[allow(dead_code)] + fn source_name(&self) -> &str; + + /// Gets the telemetry type. + #[allow(dead_code)] + fn telemetry_type(&self) -> &str; + + /// Gets the timestamp of the telemetry. + #[allow(dead_code)] + fn timestamp(&self) -> SystemTime; + + /// Converts the trait object to Any. + fn as_any(&self) -> &dyn Any; +} + +impl<T: Clone + Send + Debug + 'static> Telemetry for TypedTelemetry<T> { + fn source_name(&self) -> &str { + &self.source_name + } + + fn telemetry_type(&self) -> &str { + &self.telemetry_type + } + + fn timestamp(&self) -> SystemTime { + self.timestamp + } + + fn as_any(&self) -> &dyn Any { + self + } +} + +pub type SampleInterval = Duration; +pub type ReportInterval = Duration; + +/// Subscription Type Enum. +#[derive(Clone, Debug, PartialEq)] +pub enum SubscriptionType { + /// Whenever value changed, send an Telemetry. + OnChange, + /// Sampling interval, when it's shorter than the telemetry source's update_interval, the Telemetry + /// will be sent at update_interval (but never shorter than the minimum_update_interval). + /// Reporting interval, which should be no shorter than the sampling interval + Periodical(SampleInterval, ReportInterval), +} + +/// Type alias for telemetry source subscription ID. +type SubscriptionId = usize; +/// Type alias for telemetry sender. +type TelemetrySender = Sender<Vec<Box<dyn Telemetry + Send>>>; + +/// Represents a subscription to a telemetry source. +/// +/// This trait defines the interface for managing subscriptions to telemetry data. +/// Implementations of this trait should be able to handle different types of +/// telemetry data and subscription models. +pub trait Subscription: Send + Sync { + /// Returns a reference to the telemetry sender associated with this subscription. + /// + /// # Returns + /// + /// A reference to the `TelemetrySender` used to send telemetry data. + fn get_sender(&self) -> &TelemetrySender; + + /// Returns the type of this subscription. + /// + /// # Returns + /// + /// A reference to the `SubscriptionType` indicating whether this is an OnChange + /// or Periodical subscription. + fn get_subscription_type(&self) -> &SubscriptionType; + + /// Retrieves the timestamp of the last telemetry update. + /// + /// # Returns + /// + /// An `Option<SystemTime>` representing the timestamp of the last telemetry update, + /// or `None` if no update has occurred yet. + #[allow(dead_code)] + fn get_last_telemetry_timestamp(&self) -> Option<SystemTime>; + + /// Sets the timestamp of the last telemetry update. + /// + /// # Arguments + /// + /// * `timestamp` - A `SystemTime` representing the new last update timestamp. + #[allow(dead_code)] + fn set_last_telemetry_timestamp(&self, timestamp: SystemTime); + + /// Retrieves the current value of the telemetry data. + /// + /// # Returns + /// + /// A `Box<dyn Any>` containing the current telemetry value. + #[allow(dead_code)] + fn get_current_value(&self) -> Box<dyn Any>; + + /// Sets the current value of the telemetry data. + /// + /// # Arguments + /// + /// * `value` - A `Box<dyn Any>` containing the new telemetry value. + #[allow(dead_code)] + fn set_current_value(&self, value: Box<dyn Any>); + + /// Converts this trait object to `&dyn Any`. + /// + /// This method is used for downcasting. + /// + /// # Returns + /// + /// A reference to `dyn Any`. + #[allow(dead_code)] + fn as_any(&self) -> &dyn Any; + + /// Converts this mutable trait object to `&mut dyn Any`. + /// + /// This method is used for mutable downcasting. + /// + /// # Returns + /// + /// A mutable reference to `dyn Any`. + fn as_any_mut(&mut self) -> &mut dyn Any; +} + +/// A typed implementation of the `Subscription` trait. +/// +/// This struct provides a concrete implementation of `Subscription` for a specific +/// telemetry data type `T`. +/// +/// # Type Parameters +/// +/// * `T`: The type of telemetry data this subscription handles. It must be `Clone`, +/// `Send`, and have a `'static` lifetime. +pub struct TypedSubscription<T: Clone + Send + 'static> { + /// The sender used to transmit telemetry data. + sender: TelemetrySender, + + /// The type of this subscription (OnChange or Periodical). + subscription_type: SubscriptionType, + + /// The timestamp of the last telemetry update, wrapped in a mutex for thread-safe access. + last_telemetry_timestamp: Mutex<Option<SystemTime>>, + + /// The current value of the telemetry data, wrapped in a mutex for thread-safe access. + current_value: Mutex<Option<T>>, +} + +impl<T: Clone + Send + 'static> Subscription for TypedSubscription<T> { + fn get_sender(&self) -> &TelemetrySender { + &self.sender + } + + fn get_subscription_type(&self) -> &SubscriptionType { + &self.subscription_type + } + + fn get_last_telemetry_timestamp(&self) -> Option<SystemTime> { + *self.last_telemetry_timestamp.lock().unwrap() + } + + fn set_last_telemetry_timestamp(&self, timestamp: SystemTime) { + *self.last_telemetry_timestamp.lock().unwrap() = Some(timestamp); + } + + fn get_current_value(&self) -> Box<dyn Any> { + Box::new(self.current_value.lock().unwrap().clone()) + } + + fn set_current_value(&self, value: Box<dyn Any>) { + if let Ok(typed_value) = value.downcast::<T>() { + *self.current_value.lock().unwrap() = Some(*typed_value); + } + } + + fn as_any(&self) -> &dyn Any { + self + } + + fn as_any_mut(&mut self) -> &mut dyn Any { + self + } +} + +pub struct TelemetrySourceManager { + /// Object store with primary key. + sources: DashMap<String, Arc<Box<dyn TelemetrySourceTrait>>>, + /// Secondary key maps, as <key_name, HashMap<key_value, Vec<primary_keys>>. + secondary_map: DashMap<String, HashMap<String, Vec<String>>>, + /// Subscription map as <telemetry source name, Subscriptions>. + telemetry_subscriptions: DashMap<String, HashMap<SubscriptionId, Box<dyn Subscription>>>, + /// Atomic variable used to generate next subscription_id. + next_subscription_id: AtomicUsize, + /// The lock is only needed for add/remove telemetry sources. + lock: RwLock<()>, +} + +impl TelemetrySourceManager { + /// Creates a new `TelemetrySourceManager`. + /// + /// # Returns + /// + /// A new `Arc<TelemetrySourceManager>` instance. + pub fn new() -> Arc<Self> { + Arc::new(Self { + sources: DashMap::new(), + secondary_map: DashMap::new(), + telemetry_subscriptions: DashMap::new(), + next_subscription_id: AtomicUsize::new(0), + lock: RwLock::new(()), + }) + } + + /// Adds an telemetry source to the manager. + /// + /// # Arguments + /// + /// * `source` - The telemetry source to be added. + /// + /// # Returns + /// + /// A Result indicating success or failure. + pub async fn add_source<T: Clone + Send + PartialEq + Debug + 'static>( + self: &Arc<Self>, + source: Box<dyn TelemetrySourceTrait>, + ) -> Result<()> { + let _write_guard = self.lock.write().await; + let source_name = source.get_name().to_string(); + if let Some(secondary_keys) = source.get_secondary_keys() { + for (key_name, key_value) in secondary_keys { + self.secondary_map + .entry(key_name.clone()) + .or_default() + .entry(key_value.clone()) + .or_default() + .push(source_name.clone()); + } + } + let source = Arc::new(source); + self.sources.insert(source_name.clone(), source.clone()); + let manager = self.clone(); + + tokio::spawn(async move { + loop { + if source + .as_any() + .downcast_ref::<TelemetrySource<T>>() + .expect("Failed to downcast source") + .stop_flag + .load(Ordering::SeqCst) + { + break; + } + + source.get_wakeup_flag().notified().await; + + if !*source.get_is_polling().read().unwrap() { + continue; + } + + loop { + if source + .as_any() + .downcast_ref::<TelemetrySource<T>>() + .expect("Failed to downcast source") + .stop_flag + .load(Ordering::SeqCst) + { + return; + } + + let update_interval = source.get_current_interval(); + let reporting_interval = source.get_current_reporting_interval(); + + if !*source.get_is_polling().read().unwrap() { + break; + } + + match source + .as_any() + .downcast_ref::<TelemetrySource<T>>() + .expect("Failed to downcast source") + .value_reader + .read_values(update_interval, reporting_interval) + .await + { + Ok(values) => { + if !values.is_empty() { + if let Err(e) = manager.send_telemetries(&source_name, values).await + { + eprintln!("Error sending telemetries: {}", e); + } + } + } + Err(e) => eprintln!("Error reading values: {}", e), + } + } + } + }); + + Ok(()) + } + + /// Removes an telemetry source from the manager. + /// + /// CAUTION! All reference got from those query interfaces to this source must be dropped before making this call, + /// otherwise this call will deadlock. + /// + /// # Arguments + /// + /// * `source_name` - The name of the telemetry source to remove. + /// + /// # Returns + /// + /// A Result indicating success or failure. + #[allow(dead_code)] + pub async fn remove_source(self: &Arc<Self>, source_name: &str) -> Result<()> { + if let Some((_, source)) = self.sources.remove(source_name) { + source.set_stop_flag(true); + *source.get_is_polling().write().unwrap() = false; + + if let Some(secondary_keys) = source.get_secondary_keys() { + for (key_name, key_value) in secondary_keys { + if let Some(mut key_map) = self.secondary_map.get_mut(key_name) { + key_map.retain(|k, v| { + if k == key_value { + v.retain(|name| name != source_name); + !v.is_empty() + } else { + true + } + }); + } + } + } + self.telemetry_subscriptions.remove(source_name); + Ok(()) + } else { + Err(anyhow::anyhow!("Source not found: {}", source_name)) + } + } + + /// Returns a list of all secondary keys. + /// + /// # Returns + /// + /// A vector of strings representing all secondary keys. + #[allow(dead_code)] + pub fn query_all_secondary_keys(&self) -> Vec<String> { + self.secondary_map + .iter() + .map(|entry| entry.key().clone()) + .collect() + } + + /// Queries telemetry sources by primary keys. + /// + /// # Arguments + /// + /// * `primary_keys` - A slice of primary key strings. + /// + /// # Returns + /// + /// A Result containing a vector of telemetry sources matching the primary keys, + /// or an error if no sources are found. + pub fn query_sources_by_names(&self, primary_keys: &[&str]) -> Result<Vec<String>> { + let result: Vec<_> = if primary_keys.is_empty() { + self.sources + .iter() + .map(|entry| entry.key().clone()) + .collect() + } else { + primary_keys + .iter() + .filter_map(|&key| { + if self.sources.contains_key(key) { + Some(key.to_string()) + } else { + None + } + }) + .collect() + }; + if result.is_empty() { + Err(anyhow::anyhow!("No sources found")) + } else { + Ok(result) + } + } + + /// Queries telemetry sources by a secondary key property. + /// + /// # Arguments + /// + /// * `key` - A tuple containing the secondary key name and value. + /// + /// # Returns + /// + /// A Result containing a vector of telemetry sources matching the secondary key property, + /// or an error if the secondary key is not found or no sources match. + #[allow(dead_code)] + pub fn query_sources_by_property(&self, key: (&str, &str)) -> Result<Vec<String>> { + self.secondary_map + .get(key.0) + .and_then(|map| map.get(key.1).cloned()) + .ok_or_else(|| anyhow::anyhow!("Secondary key not found: {}", key.0)) + .and_then(|sources| { + if sources.is_empty() { + Err(anyhow::anyhow!("No sources found for the given property")) + } else { + Ok(sources) + } + }) + } + + /// Subscribes to telemetry source telemetries. + /// + /// CAUTION! The returned TelemetrySourceSubscription need to be moved to a separate task to + /// run, otherwise drop it may cause deadlock. + /// + /// # Arguments + /// + /// * `source_name` - The name of the telemetry source to subscribe to. + /// * `subscription_type` - The type of subscription (OnChange, Periodical). + /// + /// # Returns + /// + /// A Result containing an TelemetrySourceSubscription, or an error if the source is not found. + pub fn subscribe_telemetries<T: Clone + Send + PartialEq + 'static>( + self: &Arc<Self>, + source_name: &str, + subscription_type: SubscriptionType, + ) -> Result<TelemetrySourceSubscription<T>> { + if self.sources.contains_key(source_name) { + let (telemetry_sender, telemetry_receiver) = mpsc::channel(100); + let subscription_id = self.next_subscription_id.fetch_add(1, Ordering::SeqCst); + + let subscription: TypedSubscription<T> = TypedSubscription { + sender: telemetry_sender.clone(), + subscription_type: subscription_type.clone(), + last_telemetry_timestamp: Mutex::new(None), + current_value: Mutex::new(None), + }; + + self.telemetry_subscriptions + .entry(source_name.to_string()) + .or_default() + .insert(subscription_id, Box::new(subscription)); + + if self.telemetry_subscriptions.get(source_name).unwrap().len() == 1 { + if let Some(source) = self.sources.get(source_name) { + *source.get_is_polling().write().unwrap() = true; + source.get_wakeup_flag().notify_one(); + } + } + + let subscriptions = self.telemetry_subscriptions.get(source_name).unwrap(); + self.update_polling_interval(source_name, &subscriptions); + + Ok(TelemetrySourceSubscription { + source_name: source_name.to_string(), + subscription_id, + telemetry_receiver, + subscription_type, + telemetry_source_manager: self.clone(), + _phantom: PhantomData, + }) + } else { + Err(anyhow::anyhow!("Source not found: {}", source_name)) + } + } + + /// Unsubscribes from telemetry source telemetries. + /// + /// # Arguments + /// + /// * `source_name` - The name of the telemetry source. + /// * `subscription_id` - The subscription ID to unsubscribe. + /// + /// # Returns + /// + /// A Result indicating success or failure. + pub fn unsubscribe_telemetries(&self, source_name: &str, subscription_id: usize) -> Result<()> { + if let Some(mut subscriptions) = self.telemetry_subscriptions.get_mut(source_name) { + subscriptions.remove(&subscription_id); + if subscriptions.is_empty() { + if let Some(source) = self.sources.get(source_name) { + *source.get_is_polling().write().unwrap() = false; + source.set_current_interval(source.get_default_update_interval()); + } + } else { + self.update_polling_interval(source_name, &subscriptions); + } + Ok(()) + } else { + Err(anyhow::anyhow!("Source not found: {}", source_name)) + } + } + + /// Updates the polling interval for an telemetry source based on its subscriptions. + /// + /// # Arguments + /// + /// * `source_name` - The name of the telemetry source. + /// * `subscriptions` - The current subscriptions for the telemetry source. + fn update_polling_interval( + &self, + source_name: &str, + subscriptions: &HashMap<SubscriptionId, Box<dyn Subscription>>, + ) { + if let Some(source) = self.sources.get(source_name) { + let min_sampling_interval = subscriptions + .values() + .filter_map(|sub| match sub.get_subscription_type() { + SubscriptionType::Periodical(sampling_interval, _) => Some(*sampling_interval), + _ => None, + }) + .min() + .unwrap_or_else(|| source.get_default_update_interval()); + + source.set_current_interval(min_sampling_interval); + + let min_reporting_interval = subscriptions + .values() + .filter_map(|sub| match sub.get_subscription_type() { + SubscriptionType::Periodical(_, reporting_interval) => { + Some(*reporting_interval) + } + _ => None, + }) + .min() + .unwrap_or_else(|| source.get_current_reporting_interval()); + + source.set_current_reporting_interval(min_reporting_interval); + } + } + + /// Sends telemetry source telemetries to subscribers. + /// + /// # Arguments + /// + /// * `source_name` - The name of the telemetry source. + /// * `values` - The telemetry source telemetries to send. + /// + /// # Returns + /// + /// A Result indicating success or failure. + async fn send_telemetries<T: Clone + Send + Debug + PartialEq + 'static>( + &self, + source_name: &str, + values: Vec<(T, SystemTime)>, + ) -> Result<()> { + if let Some(mut subscriptions) = self.telemetry_subscriptions.get_mut(source_name) { + for (_, subscription) in subscriptions.iter_mut() { + let typed_subscription = subscription + .as_any_mut() + .downcast_mut::<TypedSubscription<T>>() + .unwrap(); + let telemetry_sender = typed_subscription.get_sender().clone(); + let subscription_type = typed_subscription.get_subscription_type().clone(); + let mut telemetries = Vec::new(); + + match subscription_type { + SubscriptionType::OnChange => { + for (value, timestamp) in &values { + let mut current_value = + typed_subscription.current_value.lock().unwrap(); + if current_value.as_ref() != Some(value) { + let telemetry = TypedTelemetry { + source_name: source_name.to_string(), + telemetry_type: "ValueChanged".to_string(), + value: value.clone(), + timestamp: *timestamp, + }; + telemetries.push(Box::new(telemetry) as Box<dyn Telemetry + Send>); + *current_value = Some(value.clone()); + } + } + } + SubscriptionType::Periodical(_, reporting_interval) => { + let mut should_send = false; + for (value, timestamp) in &values { + let mut last_timestamp = + typed_subscription.last_telemetry_timestamp.lock().unwrap(); + if let Some(last) = *last_timestamp { + match timestamp.duration_since(last) { + Ok(duration) if duration >= reporting_interval => { + should_send = true; + *last_timestamp = Some(*timestamp); + } + Ok(_) => { + // Continue accumulating samples + } + Err(_) => { + eprintln!("System time went backwards"); + *last_timestamp = Some(*timestamp); + should_send = true; + } + } + } else { + *last_timestamp = Some(*timestamp); + should_send = true; + } + + let telemetry = TypedTelemetry { + source_name: source_name.to_string(), + telemetry_type: "Periodical".to_string(), + value: value.clone(), + timestamp: *timestamp, + }; + telemetries.push(Box::new(telemetry) as Box<dyn Telemetry + Send>); + } + + if !should_send { + continue; + } + } + } + + if !telemetries.is_empty() { + telemetry_sender + .send(telemetries) + .await + .map_err(|e| anyhow::anyhow!("Failed to send telemetry: {}", e))?; + } + } + } + Ok(()) + } + + /// Gets the current value of an telemetry source. + /// + /// # Arguments + /// + /// * `source_name` - The name of the telemetry source. + /// + /// # Returns + /// + /// A Result containing the current value and timestamp, or an error if the source is not found. + #[allow(dead_code)] + pub async fn get_value<T: Clone + Send + 'static>( + &self, + source_name: &str, + ) -> Result<(T, SystemTime)> { + if let Some(source) = self.sources.get(source_name) { + if let Some(telemetry_source) = source.as_any().downcast_ref::<TelemetrySource<T>>() { + telemetry_source + .value_reader + .get_value() + .await + .map_err(|e| anyhow::anyhow!("Failed to get value: {}", e)) + } else { + Err(anyhow::anyhow!("Failed to downcast source")) + } + } else { + Err(anyhow::anyhow!("Source not found: {}", source_name)) + } + } +} + +/// Telemetry Source Subscription. +pub struct TelemetrySourceSubscription<T: Clone + Send + PartialEq + 'static> { + source_name: String, + subscription_id: SubscriptionId, + /// MPSC receiver for subscriber to receive Telemetries. + pub telemetry_receiver: Receiver<Vec<Box<dyn Telemetry + Send>>>, + #[allow(dead_code)] + subscription_type: SubscriptionType, + telemetry_source_manager: Arc<TelemetrySourceManager>, + _phantom: PhantomData<T>, +} + +impl<T: Clone + Send + PartialEq + 'static> Drop for TelemetrySourceSubscription<T> { + fn drop(&mut self) { + if let Err(e) = self + .telemetry_source_manager + .unsubscribe_telemetries(&self.source_name, self.subscription_id) + { + eprintln!("Error unsubscribing telemetries: {}", e); + } + } +} + +// Implement Debug for TelemetrySourceSubscription +impl<T: Clone + Send + PartialEq + 'static> fmt::Debug for TelemetrySourceSubscription<T> { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("TelemetrySourceSubscription") + .field("source_name", &self.source_name) + .field("subscription_id", &self.subscription_id) + .field("subscription_type", &self.subscription_type) + .finish() + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::atomic::Ordering; + + use rand::Rng; + use tokio::time; + + use tokio::sync::Mutex; + + struct MockAsyncReadValue { + last_read_time: std::sync::Mutex<SystemTime>, + current_value: std::sync::Mutex<f64>, + } + + impl MockAsyncReadValue { + fn new() -> Self { + Self { + last_read_time: std::sync::Mutex::new(SystemTime::now()), + current_value: std::sync::Mutex::new(0.0), + } + } + } + + #[async_trait::async_trait] + impl AsyncReadValue<f64> for MockAsyncReadValue { + async fn read_values( + &self, + sampling_interval: Duration, + reporting_interval: Duration, + ) -> Result<Vec<(f64, SystemTime)>, std::io::Error> { + // Sleep interval is for mock purpose only + time::sleep(reporting_interval).await; + + let now = SystemTime::now(); + let mut values = Vec::new(); + + let mut last_time = { + let mut last_time = self.last_read_time.lock().unwrap(); + let new_last_time = std::cmp::max(*last_time, now - reporting_interval); + *last_time = new_last_time; + new_last_time + }; + + while last_time < now { + values.push((rand::thread_rng().gen::<f64>() * 100.0, last_time)); + last_time = last_time.checked_add(sampling_interval).unwrap_or(now); + } + + { + let mut last_time_lock = self.last_read_time.lock().unwrap(); + *last_time_lock = last_time; + } + + Ok(values) + } + + async fn get_value(&self) -> Result<(f64, SystemTime), std::io::Error> { + let now = SystemTime::now(); + let value = { + let mut value = self.current_value.lock().unwrap(); + *value = rand::thread_rng().gen::<f64>() * 100.0; + *value + }; + Ok((value, now)) + } + } + + #[tokio::test] + async fn test_telemetry_source_manager() { + let telemetry_source_manager = TelemetrySourceManager::new(); + let start_time = SystemTime::now(); + let mock_reader = Arc::new(MockAsyncReadValue::new()); + + // Add telemetry sources + let source1 = TelemetrySource::new( + "Source1".to_string(), + Some(vec![ + ("FRU".to_string(), "CPU".to_string()), + ("ReadingType".to_string(), "Temperature".to_string()), + ]), + mock_reader.clone(), + Duration::from_millis(10), + Duration::from_millis(1), + Duration::from_millis(50), + ); + let _ = telemetry_source_manager.add_source::<f64>(source1).await; + + let mock_reader_2 = Arc::new(MockAsyncReadValue::new()); + let source2 = TelemetrySource::new( + "Source2".to_string(), + Some(vec![ + ("FRU".to_string(), "GPU".to_string()), + ("ReadingType".to_string(), "Temperature".to_string()), + ]), + mock_reader_2.clone(), + Duration::from_millis(20), + Duration::from_millis(2), + Duration::from_millis(100), + ); + let _ = telemetry_source_manager.add_source::<f64>(source2).await; + + // Query all telemetry sources + let all_sources = telemetry_source_manager + .query_sources_by_names(&[]) + .unwrap(); + assert_eq!( + all_sources.len(), + 2, + "There should be 2 telemetry sources in the manager" + ); + + // Query by name + let source1 = telemetry_source_manager + .query_sources_by_names(&["Source1"]) + .unwrap(); + assert_eq!(source1.len(), 1); + assert_eq!(source1[0], "Source1"); + + // Query all secondary keys + let keys = telemetry_source_manager.query_all_secondary_keys(); + assert!(keys.contains(&"FRU".to_string())); + assert!(keys.contains(&"ReadingType".to_string())); + + // Query by secondary key + let cpu_sources = telemetry_source_manager + .query_sources_by_property(("FRU", "CPU")) + .unwrap(); + assert_eq!( + cpu_sources.len(), + 1, + "There should be 1 telemetry source with FRU 'CPU'" + ); + + // Verify that polling is not started + let source1 = telemetry_source_manager.sources.get("Source1").unwrap(); + assert!( + !*source1.get_is_polling().read().unwrap(), + "Polling should not start before subscription" + ); + + // Subscribe to telemetries from a source + let mut handles: Vec<tokio::task::JoinHandle<()>> = Vec::new(); + if let Ok(mut subscription) = telemetry_source_manager.subscribe_telemetries::<f64>( + "Source1", + SubscriptionType::Periodical(Duration::from_millis(10), Duration::from_millis(100)), + ) { + assert!( + *source1.get_is_polling().read().unwrap(), + "Polling should start after subscription" + ); + + let handle = tokio::spawn(async move { + let mut telemetry_count = 0; + while telemetry_count < 5 { + if let Some(telemetries) = subscription.telemetry_receiver.recv().await { + telemetry_count += 1; + for telemetry in telemetries.into_iter() { + let typed_telemetry = telemetry + .as_any() + .downcast_ref::<TypedTelemetry<f64>>() + .unwrap(); + assert!( + typed_telemetry.timestamp <= SystemTime::now(), + "Telemetry timestamp is in the future" + ); + assert!( + typed_telemetry.timestamp >= start_time, + "Telemetry timestamp is before start time" + ); + } + } + } + drop(subscription); + }); + handles.push(handle); + } + + // Wait for all telemetry handlers to complete + for handle in handles { + handle.await.unwrap(); + } + + assert!( + !*source1.get_is_polling().read().unwrap(), + "Polling should stop after all subscriptions dropped" + ); + + // Drop this first so you can call remove_source on it + drop(source1); + + // Remove an telemetry source + let _ = telemetry_source_manager.remove_source("Source1").await; + let remaining_sources = telemetry_source_manager + .query_sources_by_names(&[]) + .unwrap(); + assert_eq!( + remaining_sources.len(), + 1, + "There should be 1 telemetry source remaining" + ); + assert_eq!(remaining_sources[0], "Source2"); + + // Test get_value method + let value_result = telemetry_source_manager.get_value::<f64>("Source2").await; + assert!(value_result.is_ok(), "get_value should succeed"); + let (value, timestamp) = value_result.unwrap(); + assert!( + value >= 0.0 && value <= 100.0, + "Value should be between 0 and 100" + ); + assert!( + timestamp <= SystemTime::now(), + "Timestamp should not be in the future" + ); + + // Test get_value for non-existent source + let error_result = telemetry_source_manager + .get_value::<f64>("NonExistentSource") + .await; + assert!( + error_result.is_err(), + "get_value should fail for non-existent source" + ); + + println!("Done!"); + } + + struct MockAsyncReadValueF64 { + last_read_time: Mutex<SystemTime>, + current_value: Mutex<f64>, + } + + impl MockAsyncReadValueF64 { + fn new() -> Self { + Self { + last_read_time: Mutex::new(SystemTime::now()), + current_value: Mutex::new(0.0), + } + } + } + + #[async_trait::async_trait] + impl AsyncReadValue<f64> for MockAsyncReadValueF64 { + async fn read_values( + &self, + sampling_interval: Duration, + reporting_interval: Duration, + ) -> Result<Vec<(f64, SystemTime)>, std::io::Error> { + time::sleep(reporting_interval).await; + + let now = SystemTime::now(); + let mut values = Vec::new(); + + let mut last_time = { + let mut last_time = self.last_read_time.lock().await; + let new_last_time = std::cmp::max(*last_time, now - reporting_interval); + *last_time = new_last_time; + new_last_time + }; + + while last_time < now { + let value = rand::thread_rng().gen::<f64>() * 100.0; + values.push((value, last_time)); + last_time = last_time.checked_add(sampling_interval).unwrap_or(now); + } + + { + let mut last_time_lock = self.last_read_time.lock().await; + *last_time_lock = last_time; + } + + Ok(values) + } + + async fn get_value(&self) -> Result<(f64, SystemTime), std::io::Error> { + let now = SystemTime::now(); + let value = *self.current_value.lock().await; + Ok((value, now)) + } + } + + /// Why not make MockAsyncReadValue a generic struct of 'T'? + /// because this will make those downcast_ref fail -- in the TelemetrySourceManager, it does not + /// have the knowledge of type 'T' for each TelemetrySource. + struct MockAsyncReadValueBytes { + last_read_time: Mutex<SystemTime>, + current_value: Mutex<Arc<[u8]>>, + } + + impl MockAsyncReadValueBytes { + fn new() -> Self { + Self { + last_read_time: Mutex::new(SystemTime::now()), + current_value: Mutex::new(Arc::new([0u8; 10])), + } + } + } + + #[async_trait::async_trait] + impl AsyncReadValue<Arc<[u8]>> for MockAsyncReadValueBytes { + async fn read_values( + &self, + sampling_interval: Duration, + reporting_interval: Duration, + ) -> Result<Vec<(Arc<[u8]>, SystemTime)>, std::io::Error> { + time::sleep(reporting_interval).await; + + let now = SystemTime::now(); + let mut values = Vec::new(); + + let mut last_time = { + let mut last_time = self.last_read_time.lock().await; + let new_last_time = std::cmp::max(*last_time, now - reporting_interval); + *last_time = new_last_time; + new_last_time + }; + + while last_time < now { + let value: Arc<[u8]> = Arc::new(rand::thread_rng().gen::<[u8; 10]>()); + values.push((value, last_time)); + last_time = last_time.checked_add(sampling_interval).unwrap_or(now); + } + + { + let mut last_time_lock = self.last_read_time.lock().await; + *last_time_lock = last_time; + } + + Ok(values) + } + + async fn get_value(&self) -> Result<(Arc<[u8]>, SystemTime), std::io::Error> { + let now = SystemTime::now(); + let value = self.current_value.lock().await.clone(); + Ok((value, now)) + } + } + + #[tokio::test] + async fn test_telemetry_source_manager_with_different_types() { + let telemetry_source_manager = TelemetrySourceManager::new(); + + // Test with f64 + let mock_reader_f64 = Arc::new(MockAsyncReadValueF64::new()); + let source_f64 = TelemetrySource::new( + "SourceF64".to_string(), + None, + mock_reader_f64.clone(), + Duration::from_millis(10), + Duration::from_millis(1), + Duration::from_millis(50), + ); + let _ = telemetry_source_manager.add_source::<f64>(source_f64).await; + + // Test with Arc<[u8]> + let mock_reader_bytes = Arc::new(MockAsyncReadValueBytes::new()); + let source_bytes = TelemetrySource::new( + "SourceBytes".to_string(), + None, + mock_reader_bytes.clone(), + Duration::from_millis(20), + Duration::from_millis(2), + Duration::from_millis(100), + ); + let _ = telemetry_source_manager + .add_source::<Arc<[u8]>>(source_bytes) + .await; + + // Test get_value for f64 + let result_f64 = telemetry_source_manager.get_value::<f64>("SourceF64").await; + assert!(result_f64.is_ok(), "get_value should succeed for f64"); + let (value_f64, timestamp_f64) = result_f64.unwrap(); + assert!( + value_f64 >= 0.0 && value_f64 <= 100.0, + "f64 value should be between 0 and 100" + ); + assert!( + timestamp_f64 <= SystemTime::now(), + "f64 timestamp should not be in the future" + ); + + // Test get_value for Arc<[u8]> + let result_bytes = telemetry_source_manager + .get_value::<Arc<[u8]>>("SourceBytes") + .await; + assert!( + result_bytes.is_ok(), + "get_value should succeed for Arc<[u8]>" + ); + let (value_bytes, timestamp_bytes) = result_bytes.unwrap(); + assert_eq!(value_bytes.len(), 10, "Byte array should have length 10"); + assert!( + timestamp_bytes <= SystemTime::now(), + "Bytes timestamp should not be in the future" + ); + + // Test subscribing to telemetries for both types + let mut subscription_f64 = telemetry_source_manager + .subscribe_telemetries::<f64>("SourceF64", SubscriptionType::OnChange) + .unwrap(); + + let mut subscription_bytes = telemetry_source_manager + .subscribe_telemetries::<Arc<[u8]>>("SourceBytes", SubscriptionType::OnChange) + .unwrap(); + + // Receive and verify telemetries for f64 + if let Some(telemetries) = tokio::time::timeout( + Duration::from_secs(6), + subscription_f64.telemetry_receiver.recv(), + ) + .await + .unwrap() + { + for telemetry in telemetries { + let typed_telemetry = telemetry + .as_any() + .downcast_ref::<TypedTelemetry<f64>>() + .unwrap(); + assert!( + typed_telemetry.value >= 0.0 && typed_telemetry.value <= 100.0, + "f64 telemetry value should be between 0 and 100" + ); + } + } + + // Receive and verify telemetries for Arc<[u8]> + if let Some(telemetries) = tokio::time::timeout( + Duration::from_secs(6), + subscription_bytes.telemetry_receiver.recv(), + ) + .await + .unwrap() + { + for telemetry in telemetries { + let typed_telemetry = telemetry + .as_any() + .downcast_ref::<TypedTelemetry<Arc<[u8]>>>() + .unwrap(); + assert_eq!( + typed_telemetry.value.len(), + 10, + "Byte array in telemetry should have length 10" + ); + } + } + + // Test get_value for non-existent source + let error_result = telemetry_source_manager + .get_value::<f64>("NonExistentSource") + .await; + assert!( + error_result.is_err(), + "get_value should fail for non-existent source" + ); + } + + struct MockAsyncReadSubParam { + sampling_interval: Arc<AtomicUsize>, + reporting_interval: Arc<AtomicUsize>, + } + + impl MockAsyncReadSubParam { + fn new() -> Self { + Self { + sampling_interval: Arc::new(AtomicUsize::new(0)), + reporting_interval: Arc::new(AtomicUsize::new(0)), + } + } + } + + #[async_trait::async_trait] + impl AsyncReadValue<f64> for MockAsyncReadSubParam { + async fn read_values( + &self, + sampling_interval: Duration, + reporting_interval: Duration, + ) -> Result<Vec<(f64, SystemTime)>, std::io::Error> { + self.sampling_interval + .store(sampling_interval.as_millis() as usize, Ordering::SeqCst); + self.reporting_interval + .store(reporting_interval.as_millis() as usize, Ordering::SeqCst); + + let start = tokio::time::Instant::now(); + let mut values = Vec::new(); + + while start.elapsed() < reporting_interval { + values.push((0.0, SystemTime::now())); + tokio::time::sleep(sampling_interval).await; + } + + Ok(values) + } + + async fn get_value(&self) -> Result<(f64, SystemTime), std::io::Error> { + Ok((0.0, SystemTime::now())) + } + } + + #[tokio::test] + async fn test_single_subscription_intervals() { + let telemetry_source_manager = TelemetrySourceManager::new(); + let mock_reader = Arc::new(MockAsyncReadSubParam::new()); + + let source = TelemetrySource::new( + "TestSource".to_string(), + None, + mock_reader.clone(), + Duration::from_secs(1), + Duration::from_millis(10), + Duration::from_secs(10), + ); + telemetry_source_manager + .add_source::<f64>(source) + .await + .unwrap(); + + let sampling_interval = Duration::from_millis(20); + let reporting_interval = Duration::from_millis(100); + let mut subscription = telemetry_source_manager + .subscribe_telemetries::<f64>( + "TestSource", + SubscriptionType::Periodical(sampling_interval, reporting_interval), + ) + .unwrap(); + + // Wait for the first batch of telemetry to be received + let timeout_duration = Duration::from_millis(150); + match tokio::time::timeout(timeout_duration, subscription.telemetry_receiver.recv()).await { + Ok(Some(telemetries)) => { + // Check the intervals + assert_eq!( + mock_reader.sampling_interval.load(Ordering::SeqCst), + sampling_interval.as_millis() as usize, + "Sampling interval should match the subscription" + ); + assert_eq!( + mock_reader.reporting_interval.load(Ordering::SeqCst), + reporting_interval.as_millis() as usize, + "Reporting interval should match the subscription" + ); + + // Check batch reporting + let expected_count = reporting_interval.as_millis() / sampling_interval.as_millis(); + let acceptable_range = (expected_count - 1)..=(expected_count + 1); + assert!( + acceptable_range.contains(&(telemetries.len() as u128)), + "Number of telemetries ({}) should be approximately equal to reporting_interval/sampling_interval ({})", + telemetries.len(), + expected_count + ); + } + Ok(None) => panic!("Channel closed unexpectedly"), + Err(_) => panic!("Timeout waiting for telemetry"), + } + } + + #[tokio::test] + async fn test_multiple_subscriptions_and_unsubscribe() { + let telemetry_source_manager = TelemetrySourceManager::new(); + let mock_reader = Arc::new(MockAsyncReadSubParam::new()); + + let source = TelemetrySource::new( + "TestSource".to_string(), + None, + mock_reader.clone(), + Duration::from_secs(1), + Duration::from_millis(10), + Duration::from_secs(10), + ); + telemetry_source_manager + .add_source::<f64>(source) + .await + .unwrap(); + + let sampling_interval1 = Duration::from_millis(20); + let reporting_interval1 = Duration::from_millis(80); + let mut subscription1 = telemetry_source_manager + .subscribe_telemetries::<f64>( + "TestSource", + SubscriptionType::Periodical(sampling_interval1, reporting_interval1), + ) + .unwrap(); + + let sampling_interval2 = Duration::from_millis(30); + let reporting_interval2 = Duration::from_millis(60); + let mut subscription2 = telemetry_source_manager + .subscribe_telemetries::<f64>( + "TestSource", + SubscriptionType::Periodical(sampling_interval2, reporting_interval2), + ) + .unwrap(); + + // Wait for the first telemetry from either subscription + let timeout_duration = Duration::from_millis(100); + match tokio::time::timeout(timeout_duration, async { + tokio::select! { + _ = subscription1.telemetry_receiver.recv() => {}, + _ = subscription2.telemetry_receiver.recv() => {}, + } + }) + .await + { + Ok(_) => { + assert_eq!( + mock_reader.sampling_interval.load(Ordering::SeqCst), + sampling_interval1.as_millis() as usize, + "Sampling interval should be the minimum of the two subscriptions" + ); + assert_eq!( + mock_reader.reporting_interval.load(Ordering::SeqCst), + reporting_interval2.as_millis() as usize, + "Reporting interval should be the minimum of the two subscriptions" + ); + } + Err(_) => panic!("Timeout waiting for telemetry"), + } + // Unsubscribe the first subscription + drop(subscription1); + + // Wait a bit to ensure the unsubscribe takes effect + tokio::time::sleep(Duration::from_millis(100)).await; + + // Wait for the next telemetry from the remaining subscription + match tokio::time::timeout(timeout_duration, subscription2.telemetry_receiver.recv()).await + { + Ok(Some(_)) => { + assert_eq!( + mock_reader.sampling_interval.load(Ordering::SeqCst), + sampling_interval2.as_millis() as usize, + "Sampling interval should now match the remaining subscription" + ); + assert_eq!( + mock_reader.reporting_interval.load(Ordering::SeqCst), + reporting_interval2.as_millis() as usize, + "Reporting interval should now match the remaining subscription" + ); + } + Ok(None) => panic!("Channel closed unexpectedly"), + Err(_) => panic!("Timeout waiting for telemetry after unsubscribe"), + } + + // Unsubscribe the second subscription + drop(subscription2); + + // Wait a bit to ensure the unsubscribe takes effect + tokio::time::sleep(Duration::from_millis(100)).await; + + // Trigger a new read by creating a temporary subscription + let mut temp_subscription = telemetry_source_manager + .subscribe_telemetries::<f64>( + "TestSource", + SubscriptionType::Periodical(Duration::from_millis(1), Duration::from_millis(10)), + ) + .unwrap(); + + match tokio::time::timeout( + timeout_duration, + temp_subscription.telemetry_receiver.recv(), + ) + .await + { + Ok(Some(_)) => { + assert_eq!( + mock_reader.sampling_interval.load(Ordering::SeqCst), + Duration::from_millis(10).as_millis() as usize, + "Sampling interval should not lower than minimum_update_interval" + ); + } + Ok(None) => panic!("Channel closed unexpectedly"), + Err(_) => panic!("Timeout waiting for telemetry after all unsubscribes"), + } + } +}