Add Redfish chassis resource parsing logic
This change allows caller to get all sensors under all chassis, through
D-Bus call to D-Bus services (from ObjectMapper to individual sensor
services).
The nested parsing nature of this process (from find all chassis to find
all sensors under each chassis) is handled with the composite query
concept, see [this doc](https://developer.salesforce.com/docs/atlas.en-us.api_rest.meta/api_rest/dome_composite_record_manipulation.html) for example.
This is to server the subscribe by wildcard/xpath like syntax with
Redfish resource URIs, if client subscribe by server configuration, this
path will not be executed.
Google-Bug-Id: 366492368
Change-Id: I2fab35da49472d3d33133d7178e3a295d172163e
Signed-off-by: Yongbing Chen <yongbingchen@google.com>
diff --git a/streaming_telemetry/src/app_state.rs b/streaming_telemetry/src/app_state.rs
new file mode 100644
index 0000000..ea2913a
--- /dev/null
+++ b/streaming_telemetry/src/app_state.rs
@@ -0,0 +1,16 @@
+//! This module defines the central application state structure used throughout the application.
+
+use crate::handlers::chassis::ChassisState;
+use crate::telemetry_source_manager::telemetry_source_manager::TelemetrySourceManager;
+use std::sync::Arc;
+
+/// Represents the global state of the application.
+///
+/// This structure holds shared state that can be accessed across different parts of the application.
+/// It is typically wrapped in an `Arc` for thread-safe sharing.
+pub struct AppState {
+ /// The current state of the chassis.
+ pub chassis_state: ChassisState,
+ /// A thread-safe reference to the sensor database.
+ pub telemetry_source_manager: Arc<TelemetrySourceManager>,
+}
diff --git a/streaming_telemetry/src/composite_query.rs b/streaming_telemetry/src/composite_query.rs
new file mode 100644
index 0000000..e4f322e
--- /dev/null
+++ b/streaming_telemetry/src/composite_query.rs
@@ -0,0 +1,69 @@
+//! This module defines structures for handling composite requests and responses in an API system.
+//! It includes types for parsing incoming composite requests, generating responses, and
+//! managing individual sub-requests and sub-responses within a composite operation.
+
+// For using Json name directly
+#![allow(non_snake_case)]
+
+use serde_json::Value;
+
+/// Represents a composite request containing multiple sub-requests.
+#[derive(serde::Deserialize)]
+pub struct CompositeRequest {
+ /// A vector of sub-requests to be processed as part of this composite request.
+ #[allow(dead_code)]
+ pub compositeRequest: Vec<SubRequest>,
+}
+
+/// Represents an individual sub-request within a composite request.
+#[derive(Debug, Default, Clone, PartialEq, serde::Deserialize)]
+pub struct SubRequest {
+ /// The HTTP method for this sub-request (e.g., "GET", "POST").
+ pub method: String,
+ /// The URL for this sub-request.
+ pub url: String,
+ /// A unique identifier for this sub-request within the composite request.
+ pub referenceId: String,
+ /// If true, the entire composite request should be aborted if this sub-request fails.
+ pub abortOnFailure: Option<bool>,
+ /// The body of the sub-request, if any.
+ pub body: Option<Value>,
+}
+
+/// Represents the response to a composite request.
+#[derive(serde::Serialize)]
+pub struct CompositeResponse {
+ /// A vector of sub-responses corresponding to each sub-request in the original composite request.
+ pub compositeResponse: Vec<SubResponse>,
+}
+
+/// Represents the possible types of responses that can be returned from a request.
+#[derive(Debug)]
+pub enum RequestResponse {
+ /// A JSON response.
+ Json(Value),
+}
+
+impl Default for RequestResponse {
+ /// Provides a default error response.
+ fn default() -> Self {
+ RequestResponse::Json(serde_json::json!({
+ "status": "error",
+ "message": "Invalid response",
+ "code": 500
+ }))
+ }
+}
+
+/// Represents the response to an individual sub-request within a composite request.
+#[derive(Default, serde::Serialize)]
+pub struct SubResponse {
+ /// The body of the sub-response.
+ pub body: Option<Value>,
+ /// Any HTTP headers included in the sub-response.
+ pub httpHeaders: Option<Value>,
+ /// The HTTP status code for this sub-response.
+ pub httpStatusCode: u16,
+ /// The reference ID of the corresponding sub-request.
+ pub referenceId: String,
+}
diff --git a/streaming_telemetry/src/dbus_client/dbus_client.rs b/streaming_telemetry/src/dbus_client/dbus_client.rs
new file mode 100644
index 0000000..92968c5
--- /dev/null
+++ b/streaming_telemetry/src/dbus_client/dbus_client.rs
@@ -0,0 +1,402 @@
+//! This module provides functionality for interacting with sensors via D-Bus in an OpenBMC environment.
+//! It includes functions for retrieving sensor information, mapping Redfish sensors to D-Bus paths,
+//! and polling sensor values.
+
+use crate::dbus_client::ObjectMapper::ObjectMapperProxy;
+use std::collections::HashMap;
+use zbus::names::BusName;
+use zbus::zvariant::ObjectPath;
+use zbus::Connection;
+
+use redfish_codegen::models::odata_v4;
+use redfish_codegen::models::{
+ resource::Health,
+ resource::State,
+ resource::Status,
+ sensor::v1_7_0::Sensor as SensorModel,
+ sensor::v1_7_0::{ReadingType, Threshold},
+};
+
+/// Retrieves objects with a specific interface from D-Bus.
+///
+/// Equivalent to below sample busctl command
+/// ~# busctl call xyz.openbmc_project.ObjectMapper /xyz/openbmc_project/object_mapper xyz.openbmc_project.ObjectMapper GetSubTree sias "/xyz/openbmc_project" 0 1 "xyz.openbmc_project.Inventory.Item.Board"
+/// a{sa{sas}} 8 "/xyz/openbmc_project/inventory/system/board/MyBMC" 1 "xyz.openbmc_project.EntityManager" 7 "xyz.openbmc_project.AddObject" "xyz.openbmc_project.Association.Definitions" "xyz.openbmc_project.Inventory.Connector.Slot"
+///
+/// # Arguments
+///
+/// * `services` - A vector of service names to filter by. If empty, all services are considered.
+/// * `subtee_root` - The root of the substree to search for.
+/// * `interface` - The interface name to search for.
+///
+/// # Returns
+///
+/// A `HashMap` where keys are service names and values are vectors of object paths.
+pub async fn get_objects_with_interface(
+ services: Vec<String>,
+ subtee_root: &str,
+ interface: &str,
+) -> Result<HashMap<String, Vec<String>>, Box<dyn std::error::Error>> {
+ let connection = Connection::system().await?;
+ let proxy = ObjectMapperProxy::builder(&connection)
+ .destination("xyz.openbmc_project.ObjectMapper")?
+ .path("/xyz/openbmc_project/object_mapper")?
+ .build()
+ .await?;
+
+ let depth = 0;
+ let sensor_interface = [interface];
+ let subtree = proxy
+ .get_sub_tree(subtee_root, depth, &sensor_interface)
+ .await?;
+
+ let mut objects_by_service: HashMap<String, Vec<String>> = HashMap::new();
+
+ for (path, service_map) in subtree.iter() {
+ for service in service_map.keys() {
+ if services.is_empty() || services.contains(service) {
+ objects_by_service
+ .entry(service.clone())
+ .or_default()
+ .push(path.clone());
+ }
+ }
+ }
+
+ Ok(objects_by_service)
+}
+
+/// Retrieves all sensors associated with a specific chassis.
+///
+/// Equivalent to below sample busctl command
+/// ~# busctl call xyz.openbmc_project.ObjectMapper /xyz/openbmc_project/inventory/system/board/MyBoard/all_sensors org.freedesktop.DBus.Properties Get ss xyz.openbmc_project.Association endpoints
+/// v as 61 "/xyz/openbmc_project/sensors/temperature/Tray_dT2" "/xyz/openbmc_project/sensors/temperature/Tray_dT1" "/xyz/openbmc_project/sensors/temperature/Tray_dT0" "/xyz/openbmc_project/sensors/temperature/fleeting0" "/xyz/openbmc_project/sensors/temperature/fleeting1"
+///
+/// # Arguments
+///
+/// * `chassis` - The D-Bus path of the chassis.
+///
+/// # Returns
+///
+/// A vector of sensor names.
+pub async fn get_sensors_under_a_chassis(
+ chassis: &str,
+) -> Result<Vec<String>, Box<dyn std::error::Error>> {
+ let connection = Connection::system().await?;
+ let proxy = zbus::fdo::PropertiesProxy::builder(&connection)
+ .destination("xyz.openbmc_project.ObjectMapper")?
+ .path(chassis)?
+ .build()
+ .await?;
+
+ let interface = zbus_names::InterfaceName::try_from("xyz.openbmc_project.Association")?;
+ let result = proxy.get(interface, "endpoints").await?;
+
+ let sensors: Vec<String> = result.try_into()?;
+ let sensors = sensors
+ .iter()
+ .map(|path| {
+ let components: Vec<&str> = path.split('/').collect();
+ if components.len() >= 2 {
+ let sensor_type = components[components.len() - 2];
+ let sensor_name = components[components.len() - 1];
+ format!("{}_{}", sensor_type, sensor_name)
+ } else {
+ String::new()
+ }
+ })
+ .collect();
+
+ Ok(sensors)
+}
+
+// ~# 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"
+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 Redfish sensor name to its corresponding D-Bus path and service.
+///
+/// Example: map ${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 Redfish sensor name.
+///
+/// # Returns
+///
+/// A tuple containing the service name, interfaces, and D-Bus path for the sensor.
+async fn map_redfish_sensor_to_dbus(
+ connection: &Connection,
+ sensor_name: &str,
+) -> Result<((String, Vec<String>), String), Box<dyn std::error::Error>> {
+ let parts: Vec<&str> = sensor_name.splitn(2, '_').collect();
+ if parts.len() < 2 {
+ return Err(anyhow::anyhow!("Invalid Redfish sensor name pattern").into());
+ }
+
+ // WARNING! make "fanpwm_fan1_pwm" to "/xyz/openbmc_project/sensors/fan_pwm/fan1_pwm"
+ let (sensor_type, sensor_name) = if parts[0].starts_with("fan") && parts[1].contains('_') {
+ let fan_parts: Vec<&str> = parts[1].splitn(2, '_').collect();
+ (format!("fan_{}", fan_parts[0]), fan_parts[1])
+ } else {
+ (parts[0].to_string(), parts[1])
+ };
+
+ 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))
+}
+
+/// Converts a D-Bus unit to a Redfish reading type.
+///
+/// # Arguments
+///
+/// * `dbus_unit` - The D-Bus unit string.
+///
+/// # Returns
+///
+/// An `Option<ReadingType>` corresponding to the D-Bus unit.
+fn dbus_unit_to_redfish_reading_type(dbus_unit: &str) -> Option<ReadingType> {
+ match dbus_unit {
+ "xyz.openbmc_project.Sensor.Value.Unit.DegreesC" => Some(ReadingType::Temperature),
+ "xyz.openbmc_project.Sensor.Value.Unit.PercentRH" => Some(ReadingType::Humidity),
+ "xyz.openbmc_project.Sensor.Value.Unit.Watts" => Some(ReadingType::Power),
+ "xyz.openbmc_project.Sensor.Value.Unit.Joules" => Some(ReadingType::EnergyJoules),
+ "xyz.openbmc_project.Sensor.Value.Unit.Volts" => Some(ReadingType::Voltage),
+ "xyz.openbmc_project.Sensor.Value.Unit.Amperes" => Some(ReadingType::Current),
+ "xyz.openbmc_project.Sensor.Value.Unit.RPMS" => Some(ReadingType::Rotational),
+ "xyz.openbmc_project.Sensor.Value.Unit.Pascals" => Some(ReadingType::Pressure),
+ "xyz.openbmc_project.Sensor.Value.Unit.Meters" => Some(ReadingType::Altitude),
+ "xyz.openbmc_project.Sensor.Value.Unit.Percent" => Some(ReadingType::Percent),
+ "xyz.openbmc_project.Sensor.Value.Unit.CFM" => Some(ReadingType::AirFlow),
+ _ => None,
+ }
+}
+
+// ~# busctl call xyz.openbmc_project.PSUSensor /xyz/openbmc_project/sensors/voltage/P1_SEQ_VDD_CPU0 org.freedesktop.DBus.Properties GetAll s "xyz.openbmc_project.Sensor.Value"
+// a{sv} 4 "Unit" s "xyz.openbmc_project.Sensor.Value.Unit.Volts" "MaxValue" d 1.02 "MinValue" d 0 "Value" d 0.927
+async fn get_values(
+ proxy: &zbus::fdo::PropertiesProxy<'_>,
+ sensor_model: &mut SensorModel,
+) -> Result<(), Box<dyn std::error::Error>> {
+ let interface = zbus_names::InterfaceName::try_from("xyz.openbmc_project.Sensor.Value")?;
+ let result: HashMap<String, zbus::zvariant::OwnedValue> = proxy.get_all(interface).await?;
+
+ if let Some(value) = result.get("Unit") {
+ let value: zbus::zvariant::Value = zbus::zvariant::Value::from(value);
+ let unit: String = String::try_from(value)?;
+ sensor_model.reading_type = dbus_unit_to_redfish_reading_type(&unit);
+ sensor_model.reading_units = Some(unit);
+ }
+ if let Some(value) = result.get("MaxValue") {
+ let value: zbus::zvariant::Value = zbus::zvariant::Value::from(value);
+ let max_value = f64::try_from(value)?;
+ sensor_model.reading_range_max = Some(max_value);
+ }
+ if let Some(value) = result.get("MinValue") {
+ let value: zbus::zvariant::Value = zbus::zvariant::Value::from(value);
+ let min_value = f64::try_from(value)?;
+ sensor_model.reading_range_min = Some(min_value);
+ }
+ if let Some(value) = result.get("Value") {
+ let value: zbus::zvariant::Value = zbus::zvariant::Value::from(value);
+ let value = f64::try_from(value)?;
+ sensor_model.reading = Some(value);
+ }
+
+ Ok(())
+}
+
+// ~# busctl call xyz.openbmc_project.PSUSensor /xyz/openbmc_project/sensors/voltage/P1_SEQ_VDD_CPU0 org.freedesktop.DBus.Properties GetAll s "xyz.openbmc_project.Association.Definitions"
+// a{sv} 1 "Associations" a(sss) 2 "inventory" "sensors" "/xyz/openbmc_project/inventory/system/board/BoardName" "chassis" "all_sensors" "/xyz/openbmc_project/inventory/system/board/BoardName"
+async fn get_association(
+ proxy: &zbus::fdo::PropertiesProxy<'_>,
+ sensor_model: &mut SensorModel,
+) -> Result<(), Box<dyn std::error::Error>> {
+ let interface =
+ zbus_names::InterfaceName::try_from("xyz.openbmc_project.Association.Definitions")?;
+ let result: HashMap<String, zbus::zvariant::OwnedValue> = proxy.get_all(interface).await?;
+ if let Some(associations_value) = result.get("Associations") {
+ let associations_value: zbus::zvariant::Value =
+ zbus::zvariant::Value::from(associations_value);
+ if let zbus::zvariant::Value::Array(v) = associations_value {
+ let associations: Vec<(String, String, String)> = Vec::try_from(v)?;
+ if let Some((_, _, chassis)) = associations.first() {
+ sensor_model.related_item = Some(vec![odata_v4::IdRef {
+ odata_id: Some(odata_v4::Id(chassis.to_owned())),
+ }]);
+ }
+ }
+ }
+
+ Ok(())
+}
+
+// ~# busctl call xyz.openbmc_project.PSUSensor /xyz/openbmc_project/sensors/voltage/P1_SEQ_VDD_CPU0 org.freedesktop.DBus.Properties GetAll s "xyz.openbmc_project.Sensor.Threshold.Critical"
+// a{sv} 4 "CriticalHigh" d 1.02 "CriticalAlarmHigh" b false "CriticalLow" d 0.68 "CriticalAlarmLow" b false
+async fn get_threshhold(
+ proxy: &zbus::fdo::PropertiesProxy<'_>,
+ sensor_model: &mut SensorModel,
+) -> Result<(), Box<dyn std::error::Error>> {
+ let threshold_interface =
+ zbus_names::InterfaceName::try_from("xyz.openbmc_project.Sensor.Threshold.Critical")?;
+ let threshold_result: HashMap<String, zbus::zvariant::OwnedValue> =
+ proxy.get_all(threshold_interface).await?;
+
+ let mut thresholds = redfish_codegen::models::sensor::v1_7_0::Thresholds::default();
+
+ if let Some(value) = threshold_result.get("CriticalHigh") {
+ let value: zbus::zvariant::Value = zbus::zvariant::Value::from(value);
+ let critical_high = f64::try_from(value)?;
+ thresholds.upper_critical = Some(Threshold {
+ reading: Some(critical_high),
+ ..Default::default()
+ });
+ }
+ if let Some(value) = threshold_result.get("CriticalAlarmHigh") {
+ let value: zbus::zvariant::Value = zbus::zvariant::Value::from(value);
+ let _critical_alarm_high: bool = bool::try_from(value)?;
+ }
+ if let Some(value) = threshold_result.get("CriticalLow") {
+ let value: zbus::zvariant::Value = zbus::zvariant::Value::from(value);
+ let critical_low = f64::try_from(value)?;
+ thresholds.lower_critical = Some(Threshold {
+ reading: Some(critical_low),
+ ..Default::default()
+ });
+ }
+ if let Some(value) = threshold_result.get("CriticalAlarmLow") {
+ let value: zbus::zvariant::Value = zbus::zvariant::Value::from(value);
+ let _critical_alarm_low: bool = bool::try_from(value)?;
+ }
+ sensor_model.thresholds = Some(thresholds);
+ // TODO! did not find reading for LowerCaution and UpperCaution
+
+ Ok(())
+}
+
+// ~# busctl call xyz.openbmc_project.PSUSensor /xyz/openbmc_project/sensors/voltage/P1_SEQ_VDD_CPU0 org.freedesktop.DBus.Properties GetAll s "xyz.openbmc_project.State.Decorator.Availability"
+// a{sv} 1 "Available" b true
+async fn get_availablity(
+ proxy: &zbus::fdo::PropertiesProxy<'_>,
+ sensor_model: &mut SensorModel,
+) -> Result<(), Box<dyn std::error::Error>> {
+ let interface =
+ zbus_names::InterfaceName::try_from("xyz.openbmc_project.State.Decorator.Availability")?;
+ let result: HashMap<String, zbus::zvariant::OwnedValue> = proxy.get_all(interface).await?;
+
+ if let Some(value) = result.get("Available") {
+ let value: zbus::zvariant::Value = zbus::zvariant::Value::from(value);
+ let availability: bool = bool::try_from(value)?;
+ if sensor_model.status.is_none() {
+ sensor_model.status = Some(Status::default());
+ }
+ if let Some(ref mut status) = sensor_model.status {
+ status.state = Some(if availability {
+ State::Enabled
+ } else {
+ State::Disabled
+ });
+ }
+ }
+
+ Ok(())
+}
+
+// ~# busctl call xyz.openbmc_project.PSUSensor /xyz/openbmc_project/sensors/voltage/P1_SEQ_VDD_CPU0 org.freedesktop.DBus.Properties GetAll s "xyz.openbmc_project.State.Decorator.OperationalStatus"
+// a{sv} 1 "Functional" b true
+async fn get_operational(
+ proxy: &zbus::fdo::PropertiesProxy<'_>,
+ sensor_model: &mut SensorModel,
+) -> Result<(), Box<dyn std::error::Error>> {
+ let interface = zbus_names::InterfaceName::try_from(
+ "xyz.openbmc_project.State.Decorator.OperationalStatus",
+ )?;
+ let result: HashMap<String, zbus::zvariant::OwnedValue> = proxy.get_all(interface).await?;
+
+ if let Some(value) = result.get("Functional") {
+ let value: zbus::zvariant::Value = zbus::zvariant::Value::from(value);
+ let operational: bool = bool::try_from(value)?;
+ if sensor_model.status.is_none() {
+ sensor_model.status = Some(Status::default());
+ }
+ if let Some(ref mut status) = sensor_model.status {
+ status.health = Some(if operational {
+ Health::OK
+ } else {
+ Health::Critical
+ });
+ }
+ }
+
+ Ok(())
+}
+
+/// Retrieves sensor information for a single sensor and updates the provided `SensorModel`.
+///
+/// # Arguments
+///
+/// * `sensor_name` - The name of the sensor.
+/// * `sensor_model` - A mutable reference to the `SensorModel` to be updated.
+///
+/// # Returns
+///
+/// `Ok(())` if successful, or an error if the operation fails.
+pub async fn get_one_sensor(
+ sensor_name: &str,
+ sensor_model: &mut SensorModel,
+) -> Result<(), Box<dyn std::error::Error>> {
+ let connection = Connection::system().await?;
+ let (sensor_services, sensor_path) =
+ map_redfish_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 = zbus::fdo::PropertiesProxy::builder(&connection)
+ .destination(&sensor_service)?
+ .path(sensor_path)?
+ .build()
+ .await?;
+
+ get_values(&proxy, sensor_model).await?;
+ if interfaces.contains(&"xyz.openbmc_project.Association.Definitions".to_string()) {
+ get_association(&proxy, sensor_model).await?;
+ }
+ if interfaces.contains(&"xyz.openbmc_project.Sensor.Threshold.Critical".to_string()) {
+ get_threshhold(&proxy, sensor_model).await?;
+ }
+ if interfaces.contains(&"xyz.openbmc_project.State.Decorator.Availability".to_string()) {
+ get_availablity(&proxy, sensor_model).await?;
+ }
+ if interfaces.contains(&"xyz.openbmc_project.State.Decorator.OperationalStatus".to_string()) {
+ get_operational(&proxy, sensor_model).await?;
+ }
+
+ Ok(())
+}
diff --git a/streaming_telemetry/src/handlers/chassis.rs b/streaming_telemetry/src/handlers/chassis.rs
new file mode 100644
index 0000000..0557a48
--- /dev/null
+++ b/streaming_telemetry/src/handlers/chassis.rs
@@ -0,0 +1,628 @@
+//! This module handles chassis-related operations, including chassis collection, individual chassis,
+//! sensors collection, and individual sensors. It also includes functionality for event subscription
+//! and processing.
+
+// This is for using Json name directly
+#![allow(non_snake_case)]
+
+use axum::body::to_bytes;
+use axum::{
+ extract::{Json, Path, Query, State},
+ http::StatusCode,
+ response::IntoResponse,
+};
+use dashmap::DashMap;
+use futures::future::join_all;
+use serde_json::json;
+use serde_json::Value;
+use std::convert::Infallible;
+use std::sync::Arc;
+
+use axum::{extract::OriginalUri, http::Uri};
+
+use redfish_codegen::models::odata_v4;
+use redfish_codegen::models::{
+ chassis::v1_23_0::Actions as ChassisActions, chassis::v1_23_0::Chassis as ChassisModel,
+ chassis::v1_23_0::ChassisType, chassis::v1_23_0::Reset as ChassisReset,
+ chassis_collection::ChassisCollection as ChassisCollectionModel, resource,
+ sensor::v1_7_0::Sensor as SensorModel,
+ sensor_collection::SensorCollection as SensorCollectionModel,
+};
+
+use crate::dbus_client::{get_objects_with_interface, get_one_sensor, get_sensors_under_a_chassis};
+
+use crate::app_state::AppState;
+use crate::composite_query::{RequestResponse, SubRequest};
+
+/// Represents the state of the chassis, including a cache for chassis data.
+#[derive(Debug, Default)]
+pub struct ChassisState {
+ pub cache: DashMap<String, serde_json::Value>,
+}
+
+/// Represents the query parameters for expanding resources.
+#[derive(serde::Deserialize, Debug, Default)]
+struct ExpandQuery {
+ #[serde(rename = "$expand")]
+ expand: Option<String>,
+}
+
+/// Retrieves all chassis from the system.
+///
+/// # Returns
+///
+/// A Result containing a vector of chassis names or an error.
+async fn get_all_chassis() -> Result<Vec<String>, Box<dyn std::error::Error>> {
+ let result = get_objects_with_interface(
+ Vec::new(),
+ "/xyz/openbmc_project",
+ "xyz.openbmc_project.Inventory.Item.Board",
+ )
+ .await?;
+ let mut chassis: Vec<String> = Vec::new();
+ for (_service, objects) in result {
+ for object_path in objects {
+ let path_segments: Vec<&str> = object_path.split('/').collect();
+ if let Some(last_segment) = path_segments.last() {
+ chassis.push(last_segment.to_string());
+ }
+ }
+ }
+ Ok(chassis)
+}
+
+/// Handles requests for the chassis collection.
+///
+/// # Arguments
+///
+/// * `state` - The application state.
+/// * `uri` - The original URI of the request.
+/// * `params` - Query parameters for expansion.
+async fn chassis_collection(
+ State(state): State<Arc<AppState>>,
+ uri: OriginalUri,
+ Query(params): Query<ExpandQuery>,
+) -> impl IntoResponse {
+ let params_str = params.expand.clone().unwrap_or_default();
+ let cache_key = format!("{}?{}", uri.path(), params_str);
+ if let Some(cached_response) = state.chassis_state.cache.get(&cache_key) {
+ return (StatusCode::OK, Json(cached_response.value().clone()));
+ }
+
+ let chassis_collection = get_all_chassis().await.unwrap_or_default();
+ let members = chassis_collection
+ .into_iter()
+ .map(|chassis_id| format!("{}/{}", uri.path(), chassis_id))
+ .map(|chassis_uri| odata_v4::IdRef {
+ odata_id: Some(odata_v4::Id(chassis_uri)),
+ })
+ .collect::<Vec<_>>();
+ let response = ChassisCollectionModel {
+ odata_id: odata_v4::Id(uri.path().to_string()),
+ members_odata_count: odata_v4::Count(members.len() as i64),
+ members,
+ name: resource::Name("Chassis Collection".to_string()),
+ ..Default::default()
+ };
+
+ let mut response_json = serde_json::to_value(response).unwrap_or_default();
+ let expand = params.expand.as_deref() == Some(".($levels=1)");
+ if expand {
+ let member_uris: Vec<String> = response_json["Members"]
+ .as_array()
+ .unwrap_or(&Vec::new())
+ .iter()
+ .filter_map(|m| m["@odata.id"].as_str().map(ToString::to_string))
+ .collect();
+
+ let tasks = member_uris
+ .into_iter()
+ .map(|odata_id| {
+ let state = state.clone();
+ tokio::spawn(async move {
+ let uri: Uri = odata_id.parse().unwrap_or(Uri::default());
+ let chassis_id = odata_id.split('/').last().unwrap_or("");
+ let response = chassis(
+ State(state),
+ OriginalUri(uri),
+ Path(chassis_id.to_string()),
+ Query::default(),
+ )
+ .await
+ .into_response();
+ let body_bytes = to_bytes(response.into_body(), usize::MAX)
+ .await
+ .unwrap_or_default();
+ serde_json::from_slice::<Value>(&body_bytes).unwrap_or_default()
+ })
+ })
+ .collect::<Vec<_>>();
+ let results = futures::future::join_all(tasks).await;
+
+ if let Some(members) = response_json
+ .get_mut("Members")
+ .and_then(|m| m.as_array_mut())
+ {
+ for (member, result) in members.iter_mut().zip(results) {
+ match result {
+ Ok(expanded_sensor) => *member = expanded_sensor,
+ Err(_e) => {
+ *member = json!({"error": "Failed to expand sensor"});
+ }
+ }
+ }
+ }
+ }
+
+ // TODO: create async task to monitoring the changes related to this cache entry
+ state
+ .chassis_state
+ .cache
+ .insert(cache_key, response_json.clone());
+ (StatusCode::OK, Json(response_json))
+}
+
+/// Handles requests for individual chassis.
+///
+/// # Arguments
+///
+/// * `state` - The application state.
+/// * `uri` - The original URI of the request.
+/// * `chassis_id` - The ID of the chassis.
+/// * `params` - Query parameters for expansion.
+async fn chassis(
+ State(state): State<Arc<AppState>>,
+ uri: OriginalUri,
+ Path(chassis_id): Path<String>,
+ Query(params): Query<ExpandQuery>,
+) -> impl IntoResponse {
+ let params_str = params.expand.clone().unwrap_or_default();
+ let cache_key = format!("{}?{}", uri.path(), params_str);
+ if let Some(cached_response) = state.chassis_state.cache.get(&cache_key) {
+ return (StatusCode::OK, Json(cached_response.value().clone()));
+ }
+
+ let response = ChassisModel {
+ odata_id: odata_v4::Id(uri.path().to_string()),
+ actions: Some(ChassisActions {
+ chassis_reset: Some(ChassisReset {
+ target: Some(format!("{}/{}", uri.path(), "Actions/Chassis.Reset")),
+ ..Default::default()
+ }),
+ ..Default::default()
+ }),
+ sensors: Some(odata_v4::IdRef {
+ odata_id: Some(odata_v4::Id(format!("{}/{}", uri.path(), "Sensors"))),
+ }),
+ chassis_type: ChassisType::RackMount,
+ id: resource::Id(chassis_id.clone()),
+ name: resource::Name(format!("Chassis {}", chassis_id)),
+ ..Default::default()
+ };
+
+ // TODO: create async task to monitoring the changes related to this cache entry
+ let response_json = serde_json::to_value(response).unwrap_or_default();
+ state
+ .chassis_state
+ .cache
+ .insert(cache_key, response_json.clone());
+ (StatusCode::OK, Json(response_json))
+}
+
+/// Retrieves the object path for all sensors under a chassis.
+///
+/// # Arguments
+///
+/// * `sensor_collection_uri` - The URI of the sensor collection.
+///
+/// # Returns
+///
+/// A string representing the object path for all sensors.
+fn get_all_sensors_object_path(sensor_collection_uri: &OriginalUri) -> String {
+ let path = sensor_collection_uri.path().to_string();
+ let parts: Vec<&str> = path.split('/').collect();
+ if let Some(chassis_index) = parts.iter().position(|&r| r == "Chassis") {
+ if let Some(chassis_name) = parts.get(chassis_index + 1) {
+ return format!(
+ "/xyz/openbmc_project/inventory/system/board/{}/all_sensors",
+ chassis_name
+ );
+ }
+ }
+ String::new()
+}
+
+/// Handles requests for the sensors collection under a chassis.
+///
+/// # Arguments
+///
+/// * `state` - The application state.
+/// * `uri` - The original URI of the request.
+/// * `chassis_id` - The ID of the chassis.
+/// * `params` - Query parameters for expansion.
+async fn sensors_collection(
+ State(state): State<Arc<AppState>>,
+ uri: OriginalUri,
+ Path(chassis_id): Path<String>,
+ Query(params): Query<ExpandQuery>,
+) -> impl IntoResponse {
+ let params_str = params.expand.clone().unwrap_or_default();
+ let cache_key = format!("{}?{}", uri.path(), params_str);
+ if let Some(cached_response) = state.chassis_state.cache.get(&cache_key) {
+ return (StatusCode::OK, Json(cached_response.value().clone()));
+ }
+
+ let sensors = get_sensors_under_a_chassis(&get_all_sensors_object_path(&uri))
+ .await
+ .unwrap_or_default();
+ let members = sensors
+ .into_iter()
+ .map(|sensor_id| format!("{}/{}", uri.path(), sensor_id))
+ .map(|sensor_uri| odata_v4::IdRef {
+ odata_id: Some(odata_v4::Id(sensor_uri)),
+ })
+ .collect::<Vec<_>>();
+
+ let response = SensorCollectionModel {
+ odata_id: odata_v4::Id(uri.path().to_string()),
+ members_odata_count: odata_v4::Count(members.len() as i64),
+ members,
+ name: resource::Name("Sensor Collection".to_string()),
+ ..Default::default()
+ };
+
+ let mut response_json = serde_json::to_value(response).unwrap_or_default();
+ let expand = params.expand.as_deref() == Some(".($levels=1)");
+ if expand {
+ let member_uris: Vec<String> = response_json["Members"]
+ .as_array()
+ .unwrap_or(&Vec::new())
+ .iter()
+ .filter_map(|m| m["@odata.id"].as_str().map(ToString::to_string))
+ .collect();
+
+ let tasks = member_uris
+ .into_iter()
+ .map(|odata_id| {
+ let state = state.clone();
+ let chassis_id = chassis_id.clone();
+ tokio::spawn(async move {
+ let uri: Uri = odata_id.parse().unwrap_or(Uri::default());
+ let sensor_id = odata_id.split('/').last().unwrap_or("");
+ let response = sensor(
+ State(state),
+ OriginalUri(uri),
+ Path((chassis_id.to_string(), sensor_id.to_string())),
+ Query::default(),
+ )
+ .await
+ .into_response();
+ let body_bytes = to_bytes(response.into_body(), usize::MAX)
+ .await
+ .unwrap_or_default();
+ serde_json::from_slice::<Value>(&body_bytes).unwrap_or_default()
+ })
+ })
+ .collect::<Vec<_>>();
+ let results = futures::future::join_all(tasks).await;
+
+ if let Some(members) = response_json
+ .get_mut("Members")
+ .and_then(|m| m.as_array_mut())
+ {
+ for (member, result) in members.iter_mut().zip(results) {
+ match result {
+ Ok(expanded_sensor) => *member = expanded_sensor,
+ Err(_e) => {
+ *member = json!({"error": "Failed to expand sensoris"});
+ }
+ }
+ }
+ }
+ }
+
+ // TODO: create async task to monitoring the changes related to this cache entry
+ state
+ .chassis_state
+ .cache
+ .insert(cache_key, response_json.clone());
+ (StatusCode::OK, Json(response_json))
+}
+
+/// Formats a sensor name by removing underscores and trimming.
+///
+/// # Arguments
+///
+/// * `sensor_name` - The original sensor name.
+///
+/// # Returns
+///
+/// A formatted string representing the sensor name.
+fn format_sensor_name(sensor_name: &str) -> String {
+ let trimmed_name = sensor_name
+ .find('_')
+ .map(|index| &sensor_name[index + 1..])
+ .unwrap_or(sensor_name);
+
+ trimmed_name.replace('_', " ")
+}
+
+/// Extracts the chassis ID from a base URL.
+///
+/// # Arguments
+///
+/// * `base_url` - The base URL containing the chassis ID.
+///
+/// # Returns
+///
+/// An Option containing the chassis ID as a String, if found.
+fn extract_chassis_id(base_url: &str) -> Option<String> {
+ let segments: Vec<&str> = base_url.split('/').collect();
+
+ segments
+ .iter()
+ .position(|&segment| segment == "Chassis")
+ .and_then(|index| segments.get(index + 1))
+ .map(|&segment| segment.to_string())
+}
+
+/// Handles a single request in the context of a composite query.
+///
+/// # Arguments
+///
+/// * `state` - The application state.
+/// * `sub_request` - The sub-request to handle.
+/// * `context` - The context for storing request results.
+///
+/// # Returns
+///
+/// A Result containing the RequestResponse or an Infallible error.
+async fn handle_one_request(
+ state: &Arc<AppState>,
+ sub_request: &mut SubRequest,
+ context: &mut Arc<DashMap<String, Value>>,
+) -> Result<RequestResponse, Infallible> {
+ let (base_url, _) = sub_request
+ .url
+ .split_at(sub_request.url.find('?').unwrap_or(sub_request.url.len()));
+ let segments: Vec<&str> = base_url.split('/').collect();
+
+ match sub_request.method.as_str() {
+ "GET" if base_url.ends_with("/Chassis") => {
+ if let Ok(parsed_url) = url::Url::parse(&format!("http://localhost{}", sub_request.url))
+ {
+ let query: String = parsed_url.query().unwrap_or("").to_string();
+ let expanded_query: ExpandQuery = serde_qs::from_str(&query).unwrap_or_default();
+ let url = sub_request.url.parse().unwrap_or_default();
+ let url = OriginalUri(url);
+ let response = chassis_collection(State(state.clone()), url, Query(expanded_query))
+ .await
+ .into_response();
+ let body_bytes = to_bytes(response.into_body(), usize::MAX)
+ .await
+ .unwrap_or_default();
+ let json = serde_json::from_slice::<Value>(&body_bytes).unwrap_or_default();
+ context.insert(sub_request.referenceId.clone(), json.clone());
+ Ok(RequestResponse::Json(json))
+ } else {
+ Ok(RequestResponse::Json(
+ json!({"error": "Invalid URL format"}),
+ ))
+ }
+ }
+ "GET" if segments.len() > 2 && segments[segments.len() - 2] == "Chassis" => {
+ let url = sub_request.url.parse().unwrap_or_default();
+ let url = OriginalUri(url);
+ let chassis_id = extract_chassis_id(base_url).unwrap_or_default();
+ let state = state.clone();
+ let response = chassis(
+ State(state),
+ url,
+ Path(chassis_id.to_string()),
+ Query::default(),
+ )
+ .await
+ .into_response();
+ let body_bytes = to_bytes(response.into_body(), usize::MAX)
+ .await
+ .unwrap_or_default();
+ let json = serde_json::from_slice::<Value>(&body_bytes).unwrap_or_default();
+ context.insert(sub_request.referenceId.clone(), json.clone());
+ Ok(RequestResponse::Json(json))
+ }
+ "GET" if base_url.ends_with("/Sensors") => {
+ if let Ok(parsed_url) = url::Url::parse(&format!("http://localhost{}", sub_request.url))
+ {
+ let query: String = parsed_url.query().unwrap_or("").to_string();
+ // Deserialize the query string into ExpandQuery
+ let expanded_query: ExpandQuery = serde_qs::from_str(&query).unwrap_or_default();
+ let url = sub_request.url.parse().unwrap_or_default();
+ let url = OriginalUri(url);
+ let chassis_id = extract_chassis_id(base_url).unwrap_or_default();
+ let response = sensors_collection(
+ State(state.clone()),
+ url,
+ Path(chassis_id),
+ Query(expanded_query),
+ )
+ .await
+ .into_response();
+ let body_bytes = to_bytes(response.into_body(), usize::MAX)
+ .await
+ .unwrap_or_default();
+ let json = serde_json::from_slice::<Value>(&body_bytes).unwrap_or_default();
+ context.insert(sub_request.referenceId.clone(), json.clone());
+ Ok(RequestResponse::Json(json))
+ } else {
+ Ok(RequestResponse::Json(
+ json!({"error": "Invalid URL format"}),
+ ))
+ }
+ }
+ "GET" if segments.len() > 2 && segments[segments.len() - 2] == "Sensors" => {
+ let url = sub_request.url.parse().unwrap_or_default();
+ let url = OriginalUri(url);
+ let chassis_id = extract_chassis_id(base_url).unwrap_or_default();
+ let sensor_id = base_url.split('/').last().unwrap_or("");
+ let state = state.clone();
+ let response = sensor(
+ State(state),
+ url,
+ Path((chassis_id.to_string(), sensor_id.to_string())),
+ Query::default(),
+ )
+ .await
+ .into_response();
+ let body_bytes = to_bytes(response.into_body(), usize::MAX)
+ .await
+ .unwrap_or_default();
+ let json = serde_json::from_slice::<Value>(&body_bytes).unwrap_or_default();
+ context.insert(sub_request.referenceId.clone(), json.clone());
+ Ok(RequestResponse::Json(json))
+ }
+ _ => Ok::<RequestResponse, Infallible>(RequestResponse::Json(
+ json!({"error": "Unsupported request method or URL"}),
+ )),
+ }
+}
+
+/// Retrieves a SensorModel, either from cache or by creating a new one.
+///
+/// # Arguments
+///
+/// * `state` - The application state.
+/// * `cache_key` - The key for caching the sensor model.
+/// * `odata_id` - The odata ID of the sensor.
+///
+/// # Returns
+///
+/// A Result containing the SensorModel or an Infallible error.
+async fn get_sensor_model(
+ state: &Arc<AppState>,
+ cache_key: &str,
+ odata_id: &str,
+) -> Result<SensorModel, Infallible> {
+ match state.chassis_state.cache.get(cache_key) {
+ Some(cached_response) => {
+ Ok(serde_json::from_value(cached_response.value().clone()).unwrap_or_default())
+ }
+ None => {
+ let uri: Uri = odata_id.parse().unwrap_or(Uri::default());
+ let uri = OriginalUri(uri);
+ let sensor_id = uri.path().split('/').last().unwrap_or_default().to_string();
+ let mut sensor_model = SensorModel {
+ odata_id: odata_v4::Id(odata_id.to_string()),
+ odata_type: odata_v4::Type("#Sensor.v1_7_0.Sensor".to_string()),
+ id: resource::Id(sensor_id.to_string()),
+ name: resource::Name(format_sensor_name(&sensor_id)),
+ ..Default::default()
+ };
+ get_one_sensor(&sensor_id, &mut sensor_model)
+ .await
+ .unwrap_or_default();
+ Ok(sensor_model)
+ }
+ }
+}
+
+/// Handles requests for individual sensors.
+///
+/// # Arguments
+///
+/// * `state` - The application state.
+/// * `uri` - The original URI of the request.
+/// * `_chassis_id` - The ID of the chassis (unused).
+/// * `_sensor_id` - The ID of the sensor (unused).
+/// * `_params` - Query parameters for expansion (unused).
+async fn sensor(
+ State(state): State<Arc<AppState>>,
+ uri: OriginalUri,
+ Path((_chassis_id, _sensor_id)): Path<(String, String)>,
+ Query(_params): Query<ExpandQuery>,
+) -> impl IntoResponse {
+ let cache_key = uri.path().to_string();
+ let sensor_model = get_sensor_model(&state, &cache_key, &cache_key)
+ .await
+ .unwrap_or_default();
+ let response_json = serde_json::to_value(sensor_model).unwrap_or_default();
+ state
+ .chassis_state
+ .cache
+ .insert(cache_key, response_json.clone());
+ (StatusCode::OK, Json(response_json))
+}
+
+/// Handles an XPath segment in the context of URL expansion.
+///
+/// # Arguments
+///
+/// * `urls` - The vector of URLs to modify.
+/// * `_segment` - The XPath segment (unused).
+/// * `state` - The application state.
+///
+/// # Returns
+///
+/// A Result containing a vector of RequestResponse objects or an Infallible error.
+pub async fn handle_xpath_segment(
+ urls: &mut Vec<String>,
+ _segment: &str,
+ state: &Arc<AppState>,
+) -> Result<Vec<RequestResponse>, Infallible> {
+ let context = Arc::new(DashMap::new());
+ let futures: Vec<_> = urls
+ .iter()
+ .map(|url| {
+ let state_clone = state.clone();
+ let mut context_clone = context.clone();
+ async move {
+ let mut request = SubRequest {
+ method: "GET".to_string(),
+ url: url.to_string(),
+ ..Default::default()
+ };
+
+ let response =
+ handle_one_request(&state_clone, &mut request, &mut context_clone).await;
+ let new_urls = match &response {
+ Ok(RequestResponse::Json(json)) => {
+ // TODO: use segment as query here
+ let query = "$.Members[*]['@odata.id']";
+ jsonpath_lib::select(json, query)
+ .unwrap_or_else(|_| vec![])
+ .iter()
+ .filter_map(|v| v.as_str().map(String::from))
+ .collect::<Vec<String>>()
+ }
+ _ => vec![],
+ };
+ (response, new_urls)
+ }
+ })
+ .collect();
+
+ let results = join_all(futures).await;
+
+ *urls = vec![];
+ let mut responses = vec![];
+ for (response, new_urls) in results {
+ urls.extend(new_urls);
+ responses.push(response.unwrap_or_default());
+ }
+
+ Ok(responses)
+}
+
+/// Appends a segment to each URL in the provided vector.
+///
+/// # Arguments
+///
+/// * `urls` - The vector of URLs to modify.
+/// * `segment` - The segment to append to each URL.
+pub fn append_to_urls(urls: &mut [String], segment: &str) {
+ urls.iter_mut().for_each(|url| {
+ if !url.ends_with('/') && !segment.starts_with('/') {
+ url.push('/');
+ }
+ url.push_str(segment);
+ });
+}
diff --git a/streaming_telemetry/src/handlers/mod.rs b/streaming_telemetry/src/handlers/mod.rs
new file mode 100644
index 0000000..bde0584
--- /dev/null
+++ b/streaming_telemetry/src/handlers/mod.rs
@@ -0,0 +1,2 @@
+pub mod chassis;
+pub mod xpath;
diff --git a/streaming_telemetry/src/handlers/xpath.rs b/streaming_telemetry/src/handlers/xpath.rs
new file mode 100644
index 0000000..f458165
--- /dev/null
+++ b/streaming_telemetry/src/handlers/xpath.rs
@@ -0,0 +1,53 @@
+//! This module contains functions and structures related to xpath handling
+
+use crate::app_state::AppState;
+use crate::handlers::chassis::{append_to_urls, handle_xpath_segment};
+use anyhow::Context;
+use std::collections::HashMap;
+use std::sync::Arc;
+
+/// Retrieves URLs based on the provided XPath-like segments.
+///
+/// # Arguments
+///
+/// * `state` - The application state.
+/// * `identifiers` - A map of identifiers and their corresponding values.
+/// * `segments` - A slice of string slices representing the XPath-like segments.
+///
+/// # Returns
+///
+/// A Result containing a vector of URLs or an error.
+pub async fn get_xpath_urls(
+ state: &Arc<AppState>,
+ identifiers: &HashMap<String, Vec<String>>,
+ segments: &[&str],
+) -> Result<Vec<String>, anyhow::Error> {
+ let mut urls = vec![String::from("http://localhost")];
+ for segment in segments {
+ if segment.starts_with('{') && segment.ends_with('}') {
+ // SAFETY: range access is checked
+ let placeholder = &segment[1..segment.len() - 1];
+ if let Some(values) = identifiers.get(placeholder) {
+ if values.contains(&"*".to_string()) {
+ if values.len() != 1 {
+ return Err(anyhow::anyhow!(
+ "Wildcard must be the only value for placeholder '{}'",
+ placeholder
+ ));
+ }
+ handle_xpath_segment(&mut urls, placeholder, state)
+ .await
+ .context(format!(
+ "Failed to handle XPath segment for placeholder '{}'",
+ placeholder
+ ))?;
+ // TODO: implement filter logic based on the protobuf message format
+ // urls = filter_urls(&urls, placeholder, filters, state).await?;
+ }
+ }
+ } else {
+ append_to_urls(&mut urls, segment);
+ }
+ }
+ Ok(urls)
+}
diff --git a/streaming_telemetry/src/telemetry_source_manager/sensor_configs.rs b/streaming_telemetry/src/telemetry_source_manager/sensor_configs.rs
index a14962a..a2e0b06 100644
--- a/streaming_telemetry/src/telemetry_source_manager/sensor_configs.rs
+++ b/streaming_telemetry/src/telemetry_source_manager/sensor_configs.rs
@@ -332,7 +332,6 @@
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"
