From 17d3c6637578226c5a75545dca837a187c4c9432 Mon Sep 17 00:00:00 2001 From: Jay Zhu Date: Fri, 31 Jul 2026 18:11:13 -0600 Subject: [PATCH 1/2] feat(health): add NVLink domain UUID to switch telemetry Read switch domain UUIDs from Carbide API and propagate them to metrics, OTLP attributes, events, and structured logs. Restart collectors when a switch changes domains so cached metadata and registered metric labels are replaced before collection resumes. Signed-off-by: Jay Zhu --- crates/health/src/api_client.rs | 54 ++++++- .../nvue/gnmi/on_change_processor.rs | 1 + .../collectors/nvue/gnmi/sample_processor.rs | 1 + crates/health/src/discovery/cleanup.rs | 136 +++++++++++++++++- crates/health/src/discovery/context.rs | 35 +++++ crates/health/src/discovery/iteration.rs | 11 +- crates/health/src/discovery/spawn.rs | 1 + crates/health/src/endpoint/model.rs | 7 + crates/health/src/endpoint/sources.rs | 2 + crates/health/src/otlp/convert.rs | 12 ++ crates/health/src/sink/events.rs | 12 +- crates/health/src/sink/log_file.rs | 57 +++++++- crates/health/src/sink/prometheus.rs | 9 ++ 13 files changed, 326 insertions(+), 12 deletions(-) diff --git a/crates/health/src/api_client.rs b/crates/health/src/api_client.rs index c745e1bfbd..fac7fcc5f4 100644 --- a/crates/health/src/api_client.rs +++ b/crates/health/src/api_client.rs @@ -25,6 +25,7 @@ use std::net::IpAddr; use std::str::FromStr; use std::sync::{Arc, Mutex}; +use carbide_uuid::nvlink::NvLinkDomainId; use carbide_uuid::rack::RackId; use carbide_uuid::switch::SwitchId; use forge_tls::client_config::ClientCert; @@ -275,6 +276,9 @@ fn switch_endpoint_metadata( .placement_in_rack .as_ref() .and_then(|placement| placement.tray_index), + nvlink_domain_uuid: switch + .nvlink_domain_uuid + .filter(|domain_uuid| domain_uuid != &NvLinkDomainId::nil()), endpoint_role, is_primary: switch.is_primary, nmxc_enabled: config.enable_nmxc, @@ -755,7 +759,8 @@ impl From for BmcCredentials { mod tests { use std::sync::atomic::{AtomicUsize, Ordering}; - use carbide_test_support::value_scenarios; + use carbide_test_support::{Check, check_values, value_scenarios}; + use carbide_uuid::nvlink::NvLinkDomainId; use carbide_uuid::switch::{SwitchId, SwitchIdSource, SwitchType}; use nv_redfish::bmc_http::reqwest::ClientParams as ReqwestClientParams; @@ -847,6 +852,53 @@ mod tests { ); } + #[test] + fn switch_endpoint_metadata_uses_non_nil_api_domain() { + let domain = NvLinkDomainId::from_str("9f4b45ec-705a-4af4-89f7-a112bc9c8f4e") + .expect("valid domain UUID"); + + check_values( + [ + Check { + scenario: "domain is missing", + input: None, + expect: None, + }, + Check { + scenario: "nil domain is absent", + input: Some(NvLinkDomainId::nil()), + expect: None, + }, + Check { + scenario: "non-nil API switch field", + input: Some(domain), + expect: Some(domain), + }, + ], + |nvlink_domain_uuid| { + let metadata = switch_endpoint_metadata( + &rpc::forge::Switch { + config: Some(rpc::forge::SwitchConfig { + name: "switch-a".to_string(), + ..Default::default() + }), + nvlink_domain_uuid, + ..Default::default() + }, + SwitchEndpointRole::Bmc, + false, + ) + .expect("switch metadata"); + + let EndpointMetadata::Switch(switch) = metadata else { + panic!("expected switch metadata"); + }; + + switch.nvlink_domain_uuid + }, + ); + } + #[tokio::test] async fn cache_returns_existing_client_on_matching_kind() { let mut cache: HashMap = HashMap::new(); diff --git a/crates/health/src/collectors/nvue/gnmi/on_change_processor.rs b/crates/health/src/collectors/nvue/gnmi/on_change_processor.rs index 647da3c540..2c8a99590e 100644 --- a/crates/health/src/collectors/nvue/gnmi/on_change_processor.rs +++ b/crates/health/src/collectors/nvue/gnmi/on_change_processor.rs @@ -786,6 +786,7 @@ mod tests { serial: "SN-SWITCH-001".to_string(), slot_number: Some(7), tray_index: Some(3), + nvlink_domain_uuid: None, endpoint_role: SwitchEndpointRole::Host, is_primary: false, nmxc_enabled: false, diff --git a/crates/health/src/collectors/nvue/gnmi/sample_processor.rs b/crates/health/src/collectors/nvue/gnmi/sample_processor.rs index 1bb9fd2e02..12f33bb76d 100644 --- a/crates/health/src/collectors/nvue/gnmi/sample_processor.rs +++ b/crates/health/src/collectors/nvue/gnmi/sample_processor.rs @@ -1174,6 +1174,7 @@ mod tests { serial: "SN-SWITCH-001".to_string(), slot_number: Some(7), tray_index: Some(3), + nvlink_domain_uuid: None, endpoint_role: SwitchEndpointRole::Host, is_primary: false, nmxc_enabled: false, diff --git a/crates/health/src/discovery/cleanup.rs b/crates/health/src/discovery/cleanup.rs index 9cb4f374fb..e39c910c3d 100644 --- a/crates/health/src/discovery/cleanup.rs +++ b/crates/health/src/discovery/cleanup.rs @@ -17,13 +17,19 @@ use std::borrow::Cow; use std::collections::HashSet; +use std::sync::Arc; + +use futures::future::join_all; use super::context::{CollectorKind, DiscoveryLoopContext}; +use crate::collectors::Collector; +use crate::endpoint::BmcEndpoint; #[derive(Clone, Copy)] enum CollectorStopReason { EndpointRemoved, SwitchEndpointNoLongerEligible, + SwitchDomainChanged, } impl std::fmt::Display for CollectorStopReason { @@ -31,17 +37,74 @@ impl std::fmt::Display for CollectorStopReason { f.write_str(match self { Self::EndpointRemoved => "endpoint removed", Self::SwitchEndpointNoLongerEligible => "switch endpoint is no longer eligible", + Self::SwitchDomainChanged => "switch NVLink domain changed", }) } } -fn stop_collectors_for_keys( +/// Restarts collectors whose captured switch domain no longer matches discovery. +/// +/// Collectors snapshot endpoint metadata when they start. A domain change keeps +/// the same endpoint key, so removed-endpoint cleanup cannot detect it and the +/// idempotent spawn path would otherwise leave collectors on the old domain. +/// The collectors are removed and fully stopped before same-pass discovery +/// respawns them with current metadata. +pub(super) async fn stop_stale_switch_collectors( + ctx: &mut DiscoveryLoopContext, + endpoints: &[Arc], +) { + // Build both sets together because endpoint keys allocate. This keeps + // reconciliation linear without allocating the same key twice per switch. + let mut active_switch_endpoints = HashSet::with_capacity(endpoints.len()); + let mut changed_endpoints = HashSet::new(); + + for endpoint in endpoints { + let Some(switch) = endpoint.switch_data() else { + continue; + }; + + let key = Cow::Owned(endpoint.key()); + if ctx + .collectors + .observe_switch_domain(&key, switch.nvlink_domain_uuid) + { + changed_endpoints.insert(key.clone()); + } + + active_switch_endpoints.insert(key); + } + + ctx.collectors + .retain_switch_domains(&active_switch_endpoints); + + let collectors = CollectorKind::ALL + .into_iter() + .flat_map(|kind| { + take_collectors_for_keys( + ctx, + kind, + &changed_endpoints, + CollectorStopReason::SwitchDomainChanged, + ) + }) + .collect::>(); + + // Collector shutdown emits CollectorRemoved, which unregisters cached + // Prometheus stream metrics. Wait for that boundary before same-pass + // respawn can register the updated label set. Respawn uses the normal + // endpoint, credential, and TLS providers; this state tracks only telemetry + // attribution metadata. + join_all(collectors.into_iter().map(Collector::stop)).await; +} + +fn take_collectors_for_keys( ctx: &mut DiscoveryLoopContext, kind: CollectorKind, removed_keys: &HashSet>, stop_reason: CollectorStopReason, -) { +) -> Vec { let collectors = ctx.collectors.map_mut(kind); + let mut removed = Vec::new(); for key in removed_keys { if let Some(collector) = collectors.remove(key) { tracing::info!( @@ -51,11 +114,23 @@ fn stop_collectors_for_keys( remaining_collector_count = collectors.len(), "Stopping collector" ); - tokio::spawn(async move { - collector.stop().await; - }); + removed.push(collector); } } + removed +} + +fn stop_collectors_for_keys( + ctx: &mut DiscoveryLoopContext, + kind: CollectorKind, + removed_keys: &HashSet>, + stop_reason: CollectorStopReason, +) { + for collector in take_collectors_for_keys(ctx, kind, removed_keys, stop_reason) { + tokio::spawn(async move { + collector.stop().await; + }); + } } pub(super) fn stop_removed_bmc_collectors( @@ -134,6 +209,8 @@ mod tests { use super::*; use crate::collectors::Collector; use crate::config::Config; + use crate::endpoint::test_support::{mac, test_endpoint}; + use crate::endpoint::{EndpointMetadata, SwitchData, SwitchEndpointRole}; use crate::limiter::{NoopLimiter, RateLimiter}; use crate::metrics::MetricsManager; @@ -220,4 +297,53 @@ mod tests { .contains(CollectorKind::Nmxt, "ineligible-switch") ); } + + #[tokio::test] + async fn switch_domain_change_restarts_collectors_for_same_endpoint_key() { + let mut ctx = context("switch_domain_change_restarts_collectors"); + let mut endpoint = test_endpoint(mac("00:11:22:33:44:55")); + endpoint.metadata = Some(EndpointMetadata::Switch(SwitchData { + id: None, + serial: "switch-1".to_string(), + slot_number: None, + tray_index: None, + nvlink_domain_uuid: None, + endpoint_role: SwitchEndpointRole::Host, + is_primary: true, + nmxc_enabled: true, + nmxt_enabled: true, + })); + let key = endpoint.key(); + let mut endpoint = Arc::new(endpoint); + + ctx.collectors.insert( + CollectorKind::NvueRest, + Cow::Owned(key.clone()), + noop_collector(), + ); + stop_stale_switch_collectors(&mut ctx, std::slice::from_ref(&endpoint)).await; + assert!(ctx.collectors.contains(CollectorKind::NvueRest, &key)); + + let expected_domain = carbide_uuid::nvlink::NvLinkDomainId::new(); + let Some(EndpointMetadata::Switch(switch)) = Arc::make_mut(&mut endpoint).metadata.as_mut() + else { + panic!("test endpoint should contain switch metadata"); + }; + switch.nvlink_domain_uuid = Some(expected_domain); + + stop_stale_switch_collectors(&mut ctx, std::slice::from_ref(&endpoint)).await; + + assert!(!ctx.collectors.contains(CollectorKind::NvueRest, &key)); + + let updated_context = crate::sink::EventContext::from_endpoint(&endpoint, "nvue_rest"); + assert_eq!(updated_context.nvlink_domain_uuid(), Some(expected_domain)); + + ctx.collectors.insert( + CollectorKind::NvueRest, + Cow::Owned(key.clone()), + noop_collector(), + ); + stop_stale_switch_collectors(&mut ctx, &[endpoint]).await; + assert!(ctx.collectors.contains(CollectorKind::NvueRest, &key)); + } } diff --git a/crates/health/src/discovery/context.rs b/crates/health/src/discovery/context.rs index 9811c6341b..e1af754dc2 100644 --- a/crates/health/src/discovery/context.rs +++ b/crates/health/src/discovery/context.rs @@ -21,6 +21,7 @@ use std::sync::Arc; use std::time::Duration; use arc_swap::ArcSwapOption; +use carbide_uuid::nvlink::NvLinkDomainId; use prometheus::{Histogram, HistogramOpts}; use crate::HealthError; @@ -83,6 +84,7 @@ pub(super) struct CollectorState { nvue_gnmi: HashMap, Collector>, gpu_inventory: HashMap, Collector>, inventories: HashMap, SharedInventory>, + switch_domain_uuids: HashMap, Option>, } impl CollectorState { @@ -100,6 +102,7 @@ impl CollectorState { nvue_gnmi: HashMap::new(), gpu_inventory: HashMap::new(), inventories: HashMap::new(), + switch_domain_uuids: HashMap::new(), } } @@ -153,6 +156,38 @@ impl CollectorState { self.inventories.remove(key); } + /// Records the latest switch domain and reports whether it changed. + /// + /// The first observation establishes a baseline without forcing a restart. + /// Later transitions between absent and present values, or between two UUIDs, + /// require a restart because running collectors retain their startup metadata. + pub(super) fn observe_switch_domain( + &mut self, + key: &str, + domain_uuid: Option, + ) -> bool { + match self.switch_domain_uuids.get_mut(key) { + Some(previous) if *previous != domain_uuid => { + *previous = domain_uuid; + true + } + Some(_) => false, + None => { + self.switch_domain_uuids + .insert(Cow::Owned(key.to_string()), domain_uuid); + false + } + } + } + + pub(super) fn retain_switch_domains( + &mut self, + active_switch_endpoints: &HashSet>, + ) { + self.switch_domain_uuids + .retain(|key, _| active_switch_endpoints.contains(key)); + } + pub(super) fn contains(&self, kind: CollectorKind, key: &str) -> bool { self.map(kind).contains_key(key) } diff --git a/crates/health/src/discovery/iteration.rs b/crates/health/src/discovery/iteration.rs index 78b32b8b4a..84bdabec94 100644 --- a/crates/health/src/discovery/iteration.rs +++ b/crates/health/src/discovery/iteration.rs @@ -23,7 +23,9 @@ use std::time::Instant; use futures::{StreamExt, stream}; use super::DiscoveryIterationStats; -use super::cleanup::{stop_ineligible_nmxc_collectors, stop_removed_bmc_collectors}; +use super::cleanup::{ + stop_ineligible_nmxc_collectors, stop_removed_bmc_collectors, stop_stale_switch_collectors, +}; use super::context::{CollectorKind, DiscoveryLoopContext}; use super::identity::ensure_primary_system_uuid; use super::spawn::{spawn_collectors_for_endpoint, switch_supports_nmxc_subscription}; @@ -107,6 +109,11 @@ pub async fn run_discovery_iteration( // prune before respawn so downgraded auto-mode endpoints get replaced ctx.collectors.prune_finished_logs(); + // A domain change does not change the endpoint key, so the idempotent spawn + // path would preserve collectors with stale EventContext metadata. Stop them + // first and wait for metric registrations to be released before respawn. + stop_stale_switch_collectors(ctx, &sharded_endpoints).await; + for endpoint in &sharded_endpoints { spawn_collectors_for_endpoint(ctx, endpoint, data_sink.clone(), metrics_prefix)?; } @@ -160,6 +167,7 @@ mod tests { serial: format!("serial-{mac}"), slot_number: None, tray_index: None, + nvlink_domain_uuid: None, endpoint_role: SwitchEndpointRole::Host, is_primary: false, nmxc_enabled: false, @@ -208,6 +216,7 @@ mod tests { serial: format!("serial-{mac}"), slot_number: None, tray_index: None, + nvlink_domain_uuid: None, endpoint_role, is_primary, nmxc_enabled, diff --git a/crates/health/src/discovery/spawn.rs b/crates/health/src/discovery/spawn.rs index 13364ee027..eb00d1e807 100644 --- a/crates/health/src/discovery/spawn.rs +++ b/crates/health/src/discovery/spawn.rs @@ -770,6 +770,7 @@ mod tests { serial: serial.to_string(), slot_number: None, tray_index: None, + nvlink_domain_uuid: None, endpoint_role, is_primary, nmxc_enabled, diff --git a/crates/health/src/endpoint/model.rs b/crates/health/src/endpoint/model.rs index f6eb67e9a6..657b54c7f6 100644 --- a/crates/health/src/endpoint/model.rs +++ b/crates/health/src/endpoint/model.rs @@ -203,6 +203,13 @@ pub struct SwitchData { pub serial: String, pub slot_number: Option, pub tray_index: Option, + + /// NVLink domain UUID associated with the switch, when known. + /// + /// Discovery restarts collectors when this value changes so subsequent + /// telemetry uses current metadata. + pub nvlink_domain_uuid: Option, + pub endpoint_role: SwitchEndpointRole, pub is_primary: bool, pub nmxc_enabled: bool, diff --git a/crates/health/src/endpoint/sources.rs b/crates/health/src/endpoint/sources.rs index 85d1a1388c..8c63f758bc 100644 --- a/crates/health/src/endpoint/sources.rs +++ b/crates/health/src/endpoint/sources.rs @@ -114,6 +114,7 @@ impl StaticEndpointSource { serial, slot_number: switch.slot_number, tray_index: switch.tray_index, + nvlink_domain_uuid: None, endpoint_role, is_primary: switch.is_primary, nmxc_enabled, @@ -351,6 +352,7 @@ mod tests { assert_eq!(s.serial, "SN-001"); assert_eq!(s.slot_number, Some(7)); assert_eq!(s.tray_index, Some(3)); + assert_eq!(s.nvlink_domain_uuid, None); assert_eq!(s.endpoint_role, SwitchEndpointRole::Host); assert!(s.is_primary); assert!(s.nmxc_enabled); diff --git a/crates/health/src/otlp/convert.rs b/crates/health/src/otlp/convert.rs index 0b18e3c6c2..619da74a34 100644 --- a/crates/health/src/otlp/convert.rs +++ b/crates/health/src/otlp/convert.rs @@ -487,6 +487,9 @@ mod tests { fn resource_attributes_include_switch_placement_metadata_when_present() { let switch_id = test_switch_id("switch-a"); let switch_id_attr = switch_id.to_string(); + let nvlink_domain_uuid = NvLinkDomainId::new(); + let nvlink_domain_uuid_attr = nvlink_domain_uuid.to_string(); + let context = EventContext { endpoint_key: "11:22:33:44:55:66".to_string(), addr: BmcAddr { @@ -501,6 +504,7 @@ mod tests { serial: "SN-SWITCH-001".to_string(), slot_number: Some(7), tray_index: Some(3), + nvlink_domain_uuid: Some(nvlink_domain_uuid), endpoint_role: SwitchEndpointRole::Host, is_primary: false, nmxc_enabled: false, @@ -519,6 +523,11 @@ mod tests { assert_eq!(attr_value(&attrs, "component.type"), Some("nvlink_switch")); assert_eq!(attr_int_value(&attrs, "switch.slot_number"), Some(7)); assert_eq!(attr_int_value(&attrs, "switch.tray_index"), Some(3)); + + assert_eq!( + attr_value(&attrs, "nvlink.domain.uuid"), + Some(nvlink_domain_uuid_attr.as_str()) + ); } #[test] @@ -539,6 +548,7 @@ mod tests { serial: "SN-SWITCH-001".to_string(), slot_number: Some(7), tray_index: Some(3), + nvlink_domain_uuid: None, endpoint_role: SwitchEndpointRole::Host, is_primary: true, nmxc_enabled: true, @@ -592,6 +602,7 @@ mod tests { serial: "SN-SWITCH-BMC-001".to_string(), slot_number: Some(8), tray_index: Some(4), + nvlink_domain_uuid: None, endpoint_role: SwitchEndpointRole::Bmc, is_primary: false, nmxc_enabled: false, @@ -878,6 +889,7 @@ mod tests { serial: "SN-SWITCH-001".to_string(), slot_number: Some(7), tray_index: Some(3), + nvlink_domain_uuid: None, endpoint_role: SwitchEndpointRole::Host, is_primary: true, nmxc_enabled: true, diff --git a/crates/health/src/sink/events.rs b/crates/health/src/sink/events.rs index 2cab6dee9d..639d7b823a 100644 --- a/crates/health/src/sink/events.rs +++ b/crates/health/src/sink/events.rs @@ -122,10 +122,13 @@ impl EventContext { .and_then(|machine| machine.tray_index) } - /// Returns the NVLink domain UUID when the machine participates in one. + /// Returns the NVLink domain UUID associated with the endpoint, when known. pub fn nvlink_domain_uuid(&self) -> Option { - self.machine_metadata() - .and_then(|machine| machine.nvlink_domain_uuid) + match &self.metadata { + Some(EndpointMetadata::Machine(machine)) => machine.nvlink_domain_uuid, + Some(EndpointMetadata::Switch(switch)) => switch.nvlink_domain_uuid, + _ => None, + } } pub fn switch_id(&self) -> Option { @@ -667,6 +670,7 @@ mod tests { serial: "SW-001".to_string(), slot_number: Some(9), tray_index: Some(4), + nvlink_domain_uuid: Some(nvlink_domain_id()), endpoint_role: SwitchEndpointRole::Host, is_primary: true, nmxc_enabled: true, @@ -1167,7 +1171,7 @@ mod tests { machine_id: None, slot_number: None, tray_index: None, - nvlink_domain_uuid: None, + nvlink_domain_uuid: Some(nvlink_domain_id().to_string()), machine_serial: None, driver_version: None, component_type: Some("nvlink_switch"), diff --git a/crates/health/src/sink/log_file.rs b/crates/health/src/sink/log_file.rs index 8e40620b47..776a0a062d 100644 --- a/crates/health/src/sink/log_file.rs +++ b/crates/health/src/sink/log_file.rs @@ -281,7 +281,7 @@ mod tests { use mac_address::MacAddress; use super::*; - use crate::endpoint::{BmcAddr, EndpointMetadata, MachineData}; + use crate::endpoint::{BmcAddr, EndpointMetadata, MachineData, SwitchData, SwitchEndpointRole}; use crate::sink::DiagnosticLogRecord; /// Builds a base log context without endpoint metadata. @@ -511,6 +511,61 @@ mod tests { ); } + #[test] + fn test_writes_switch_nvlink_domain_uuid_as_jsonl_field() { + let dir = tempfile::tempdir().expect("tempdir"); + + let config = LogFileSinkConfig { + include_diagnostics: false, + output_dir: dir.path().to_string_lossy().into_owned(), + max_file_size: 1024 * 1024, + max_backups: 2, + }; + + let sink = LogFileSink::new(&config).expect("sink"); + + let nvlink_domain_uuid = NvLinkDomainId::from_str("9f4b45ec-705a-4af4-89f7-a112bc9c8f4e") + .expect("valid NVLink domain UUID"); + + let mut ctx = test_context(); + + ctx.metadata = Some(EndpointMetadata::Switch(SwitchData { + id: None, + serial: "SN-SWITCH-001".to_string(), + slot_number: Some(7), + tray_index: Some(3), + nvlink_domain_uuid: Some(nvlink_domain_uuid), + endpoint_role: SwitchEndpointRole::Host, + is_primary: true, + nmxc_enabled: true, + nmxt_enabled: true, + })); + + let event = CollectorEvent::Log( + LogRecord { + body: "switch event".to_string(), + severity: "WARN".to_string(), + attributes: Vec::new(), + diagnostic_record: None, + } + .into(), + ); + + sink.handle_event(&ctx, &event); + + let log_path = dir.path().join("health_logs.jsonl"); + let contents = fs::read_to_string(log_path).expect("read log"); + let line = contents.lines().next().expect("one JSONL record"); + let parsed: serde_json::Value = serde_json::from_str(line).expect("valid json"); + + assert_eq!( + parsed["nvlink_domain_uuid"], + "9f4b45ec-705a-4af4-89f7-a112bc9c8f4e" + ); + + assert_eq!(parsed["component_type"], "nvlink_switch"); + } + #[test] fn test_rotation_creates_backups() { let dir = tempfile::tempdir().expect("tempdir"); diff --git a/crates/health/src/sink/prometheus.rs b/crates/health/src/sink/prometheus.rs index a3cfeba8eb..5f4cf54269 100644 --- a/crates/health/src/sink/prometheus.rs +++ b/crates/health/src/sink/prometheus.rs @@ -329,6 +329,9 @@ mod tests { fn test_stream_static_labels_includes_switch_placement_metadata() { let switch_id = test_switch_id("switch-a"); let switch_id_label = switch_id.to_string(); + let nvlink_domain_uuid = NvLinkDomainId::new(); + let nvlink_domain_uuid_label = nvlink_domain_uuid.to_string(); + let context = EventContext { endpoint_key: "11:22:33:44:55:66".to_string(), addr: BmcAddr { @@ -343,6 +346,7 @@ mod tests { serial: "SN-SWITCH-001".to_string(), slot_number: Some(7), tray_index: Some(3), + nvlink_domain_uuid: Some(nvlink_domain_uuid), endpoint_role: SwitchEndpointRole::Host, is_primary: false, nmxc_enabled: false, @@ -363,5 +367,10 @@ mod tests { assert_eq!(label_value("rack_id"), Some("RACK_2")); assert_eq!(label_value("switch_slot_number"), Some("7")); assert_eq!(label_value("switch_tray_index"), Some("3")); + + assert_eq!( + label_value("nvlink_domain_uuid"), + Some(nvlink_domain_uuid_label.as_str()) + ); } } From 076c5e1a4bcb90b149214cfdb93156007012839d Mon Sep 17 00:00:00 2001 From: Jay Zhu Date: Sat, 1 Aug 2026 12:12:40 -0600 Subject: [PATCH 2/2] feat(health): support static switch domain UUID Allow static switch endpoints to provide NVLink domain UUID metadata. Ignore invalid and nil values so telemetry never publishes unusable labels. Preserve first-source precedence when duplicate endpoint keys are discovered, preventing false domain changes and unnecessary collector restarts. Signed-off-by: Jay Zhu --- crates/health/example/config.example.toml | 6 +- crates/health/src/config.rs | 13 ++- crates/health/src/discovery/cleanup.rs | 79 ++++++++++++++---- crates/health/src/discovery/iteration.rs | 6 +- crates/health/src/endpoint/sources.rs | 97 +++++++++++++++++++---- crates/health/src/sink/events.rs | 4 +- docs/architecture/health_aggregation.md | 8 +- 7 files changed, 170 insertions(+), 43 deletions(-) diff --git a/crates/health/example/config.example.toml b/crates/health/example/config.example.toml index 436a464e3d..8ef0785810 100644 --- a/crates/health/example/config.example.toml +++ b/crates/health/example/config.example.toml @@ -49,7 +49,9 @@ port = 443 mac = "11:22:33:44:55:66" username = "admin" password = "secret" -switch = { id = "fsw100htjtiaehv1n5vh67tbmqq4eabcjdng40f7jupsadbedhruh6rag1l0", serial = "SN-SWITCH-BMC-001", endpoint_role = "bmc", slot_number = 7, tray_index = 3 } +# Configure the same domain UUID on every static endpoint for the switch. +# Invalid or nil values are omitted from telemetry. +switch = { id = "fsw100htjtiaehv1n5vh67tbmqq4eabcjdng40f7jupsadbedhruh6rag1l0", serial = "SN-SWITCH-BMC-001", endpoint_role = "bmc", slot_number = 7, tray_index = 3, nvlink_domain_uuid = "9f4b45ec-705a-4af4-89f7-a112bc9c8f4e" } [[endpoint_sources.static_bmc_endpoints]] ip = "10.0.1.2" @@ -60,7 +62,7 @@ password = "secret" # For static switch host endpoints, nmxc_enabled controls direct NMX-C # Subscribe eligibility after the endpoint_role="host" and is_primary=true # checks. If omitted, it defaults to is_primary. -switch = { id = "fsw100htjtiaehv1n5vh67tbmqq4eabcjdng40f7jupsadbedhruh6rag1l0", serial = "SN-SWITCH-HOST-001", endpoint_role = "host", is_primary = true, nmxc_enabled = true, slot_number = 7, tray_index = 3 } +switch = { id = "fsw100htjtiaehv1n5vh67tbmqq4eabcjdng40f7jupsadbedhruh6rag1l0", serial = "SN-SWITCH-HOST-001", endpoint_role = "host", is_primary = true, nmxc_enabled = true, slot_number = 7, tray_index = 3, nvlink_domain_uuid = "9f4b45ec-705a-4af4-89f7-a112bc9c8f4e" } [[endpoint_sources.static_bmc_endpoints]] ip = "10.0.2.1" diff --git a/crates/health/src/config.rs b/crates/health/src/config.rs index 2f024f68de..32bdbee76b 100644 --- a/crates/health/src/config.rs +++ b/crates/health/src/config.rs @@ -255,6 +255,11 @@ pub struct StaticSwitchEndpoint { pub slot_number: Option, #[serde(alias = "compute_tray_index")] pub tray_index: Option, + + /// Optional non-nil NVLink domain UUID associated with this switch. + /// Invalid or nil values are omitted from telemetry. + pub nvlink_domain_uuid: Option, + #[serde(default = "default_static_switch_endpoint_role")] pub endpoint_role: StaticSwitchEndpointRole, #[serde(default)] @@ -1918,6 +1923,7 @@ mod tests { serial: Some("switch-serial".to_string()), slot_number: None, tray_index: None, + nvlink_domain_uuid: None, endpoint_role: StaticSwitchEndpointRole::Host, is_primary: false, nmxc_enabled: None, @@ -3866,7 +3872,7 @@ ip = "10.0.1.2" mac = "11:22:33:44:55:77" username = "admin" password = "pass" -switch = { id = "fsw100htjtiaehv1n5vh67tbmqq4eabcjdng40f7jupsadbedhruh6rag1l0", serial = "SN-SW-002", endpoint_role = "host", is_primary = false, nmxc_enabled = true, nmxt_enabled = true } +switch = { id = "fsw100htjtiaehv1n5vh67tbmqq4eabcjdng40f7jupsadbedhruh6rag1l0", serial = "SN-SW-002", endpoint_role = "host", is_primary = false, nmxc_enabled = true, nmxt_enabled = true, nvlink_domain_uuid = "9f4b45ec-705a-4af4-89f7-a112bc9c8f4e" } "#; let config: Config = Figment::new() @@ -3884,6 +3890,11 @@ switch = { id = "fsw100htjtiaehv1n5vh67tbmqq4eabcjdng40f7jupsadbedhruh6rag1l0", assert!(!switch.is_primary); assert_eq!(switch.nmxc_enabled, Some(true)); assert_eq!(switch.nmxt_enabled, Some(true)); + + assert_eq!( + switch.nvlink_domain_uuid.as_deref(), + Some("9f4b45ec-705a-4af4-89f7-a112bc9c8f4e") + ); } #[test] diff --git a/crates/health/src/discovery/cleanup.rs b/crates/health/src/discovery/cleanup.rs index e39c910c3d..96bb379653 100644 --- a/crates/health/src/discovery/cleanup.rs +++ b/crates/health/src/discovery/cleanup.rs @@ -42,19 +42,18 @@ impl std::fmt::Display for CollectorStopReason { } } -/// Restarts collectors whose captured switch domain no longer matches discovery. +/// Restarts switch collectors when discovery reports a different NVLink domain. /// -/// Collectors snapshot endpoint metadata when they start. A domain change keeps -/// the same endpoint key, so removed-endpoint cleanup cannot detect it and the -/// idempotent spawn path would otherwise leave collectors on the old domain. -/// The collectors are removed and fully stopped before same-pass discovery -/// respawns them with current metadata. +/// Collectors retain the endpoint metadata captured at startup. Because the +/// endpoint key does not change with the domain UUID, removed-endpoint cleanup +/// cannot refresh that metadata. This function removes affected collectors and +/// waits for their shutdown before discovery respawns them. pub(super) async fn stop_stale_switch_collectors( ctx: &mut DiscoveryLoopContext, endpoints: &[Arc], ) { - // Build both sets together because endpoint keys allocate. This keeps - // reconciliation linear without allocating the same key twice per switch. + // Keep one domain observation per collector key. The active set also + // identifies which saved observations remain valid after this pass. let mut active_switch_endpoints = HashSet::with_capacity(endpoints.len()); let mut changed_endpoints = HashSet::new(); @@ -64,6 +63,14 @@ pub(super) async fn stop_stale_switch_collectors( }; let key = Cow::Owned(endpoint.key()); + + // Collector spawning uses the first endpoint for a key. Apply the same + // precedence here so a later source cannot create a false domain change + // for the collector that was spawned from the first endpoint. + if active_switch_endpoints.contains(&key) { + continue; + } + if ctx .collectors .observe_switch_domain(&key, switch.nvlink_domain_uuid) @@ -74,10 +81,14 @@ pub(super) async fn stop_stale_switch_collectors( active_switch_endpoints.insert(key); } + // Forget observations for switches absent from this discovery pass. If a + // switch returns later, its current domain establishes a fresh baseline. ctx.collectors .retain_switch_domains(&active_switch_endpoints); - let collectors = CollectorKind::ALL + // Remove every collector kind before awaiting shutdown. Same-pass spawning + // can then create replacements with the updated endpoint metadata. + let stale_collectors = CollectorKind::ALL .into_iter() .flat_map(|kind| { take_collectors_for_keys( @@ -89,12 +100,9 @@ pub(super) async fn stop_stale_switch_collectors( }) .collect::>(); - // Collector shutdown emits CollectorRemoved, which unregisters cached - // Prometheus stream metrics. Wait for that boundary before same-pass - // respawn can register the updated label set. Respawn uses the normal - // endpoint, credential, and TLS providers; this state tracks only telemetry - // attribution metadata. - join_all(collectors.into_iter().map(Collector::stop)).await; + // CollectorRemoved unregisters the old Prometheus label set. Wait for that + // cleanup before replacement collectors register the new domain UUID. + join_all(stale_collectors.into_iter().map(Collector::stop)).await; } fn take_collectors_for_keys( @@ -346,4 +354,45 @@ mod tests { stop_stale_switch_collectors(&mut ctx, &[endpoint]).await; assert!(ctx.collectors.contains(CollectorKind::NvueRest, &key)); } + + #[tokio::test] + async fn duplicate_switch_domains_use_first_source_without_restarts() { + let mut ctx = context("duplicate_switch_domains_use_first_source"); + let mut first = test_endpoint(mac("00:11:22:33:44:55")); + + first.metadata = Some(EndpointMetadata::Switch(SwitchData { + id: None, + serial: "switch-1".to_string(), + slot_number: None, + tray_index: None, + nvlink_domain_uuid: None, + endpoint_role: SwitchEndpointRole::Host, + is_primary: true, + nmxc_enabled: true, + nmxt_enabled: true, + })); + + let key = first.key(); + let first = Arc::new(first); + let mut duplicate = first.as_ref().clone(); + + let Some(EndpointMetadata::Switch(switch)) = duplicate.metadata.as_mut() else { + panic!("test endpoint should contain switch metadata"); + }; + + switch.nvlink_domain_uuid = Some(carbide_uuid::nvlink::NvLinkDomainId::new()); + + let endpoints = [first, Arc::new(duplicate)]; + stop_stale_switch_collectors(&mut ctx, &endpoints).await; + + ctx.collectors.insert( + CollectorKind::NvueRest, + Cow::Owned(key.clone()), + noop_collector(), + ); + + stop_stale_switch_collectors(&mut ctx, &endpoints).await; + + assert!(ctx.collectors.contains(CollectorKind::NvueRest, &key)); + } } diff --git a/crates/health/src/discovery/iteration.rs b/crates/health/src/discovery/iteration.rs index 84bdabec94..beb6aff069 100644 --- a/crates/health/src/discovery/iteration.rs +++ b/crates/health/src/discovery/iteration.rs @@ -109,9 +109,9 @@ pub async fn run_discovery_iteration( // prune before respawn so downgraded auto-mode endpoints get replaced ctx.collectors.prune_finished_logs(); - // A domain change does not change the endpoint key, so the idempotent spawn - // path would preserve collectors with stale EventContext metadata. Stop them - // first and wait for metric registrations to be released before respawn. + // A domain change keeps the same endpoint key and collector type. Complete + // old collector cleanup before respawn so a late CollectorRemoved cannot + // unregister the replacement's metrics. stop_stale_switch_collectors(ctx, &sharded_endpoints).await; for endpoint in &sharded_endpoints { diff --git a/crates/health/src/endpoint/sources.rs b/crates/health/src/endpoint/sources.rs index 8c63f758bc..d6d4333a52 100644 --- a/crates/health/src/endpoint/sources.rs +++ b/crates/health/src/endpoint/sources.rs @@ -33,6 +33,24 @@ use crate::endpoint::{ PowerShelfData, SharedSystemUuid, SwitchData, SwitchEndpointRole, }; +fn parse_static_nvlink_domain_uuid( + value: Option<&str>, + endpoint_kind: &str, +) -> Option { + value.and_then(|value| match NvLinkDomainId::from_str(value) { + Ok(domain_uuid) => Some(domain_uuid), + Err(error) => { + tracing::warn!( + ?error, + nvlink_domain_uuid = ?value, + "Invalid {endpoint_kind}.nvlink_domain_uuid in static endpoint config" + ); + + None + } + }) +} + pub struct StaticEndpointSource { endpoints: Vec>, } @@ -101,6 +119,11 @@ impl StaticEndpointSource { .clone() .or_else(|| switch.id.clone()) .unwrap_or_else(|| cfg.mac.clone()); + + let nvlink_domain_uuid = + parse_static_nvlink_domain_uuid(switch.nvlink_domain_uuid.as_deref(), "switch") + .filter(|domain_uuid| domain_uuid != &NvLinkDomainId::nil()); + let endpoint_role = match switch.endpoint_role { StaticSwitchEndpointRole::Bmc => SwitchEndpointRole::Bmc, StaticSwitchEndpointRole::Host => SwitchEndpointRole::Host, @@ -114,7 +137,7 @@ impl StaticEndpointSource { serial, slot_number: switch.slot_number, tray_index: switch.tray_index, - nvlink_domain_uuid: None, + nvlink_domain_uuid, endpoint_role, is_primary: switch.is_primary, nmxc_enabled, @@ -128,20 +151,11 @@ impl StaticEndpointSource { None } }); - let nvlink_domain_uuid = - machine.nvlink_domain_uuid.as_ref().and_then( - |id| match NvLinkDomainId::from_str(id) { - Ok(id) => Some(id), - Err(error) => { - tracing::warn!( - ?error, - nvlink_domain_uuid = ?id, - "Invalid machine.nvlink_domain_uuid in static endpoint config" - ); - None - } - }, - ); + + let nvlink_domain_uuid = parse_static_nvlink_domain_uuid( + machine.nvlink_domain_uuid.as_deref(), + "machine", + ); let driver_version = machine .driver_version @@ -320,6 +334,8 @@ mod tests { #[tokio::test] async fn test_static_endpoint_with_switch_serial_sets_metadata() { let switch_id = test_switch_id("switch-a"); + let nvlink_domain_uuid = NvLinkDomainId::new(); + let configs = vec![StaticBmcEndpoint { ip: ip("10.0.1.1"), port: Some(443), @@ -333,6 +349,7 @@ mod tests { serial: Some("SN-001".to_string()), slot_number: Some(7), tray_index: Some(3), + nvlink_domain_uuid: Some(nvlink_domain_uuid.to_string()), endpoint_role: StaticSwitchEndpointRole::Host, is_primary: true, nmxc_enabled: None, @@ -352,7 +369,7 @@ mod tests { assert_eq!(s.serial, "SN-001"); assert_eq!(s.slot_number, Some(7)); assert_eq!(s.tray_index, Some(3)); - assert_eq!(s.nvlink_domain_uuid, None); + assert_eq!(s.nvlink_domain_uuid, Some(nvlink_domain_uuid)); assert_eq!(s.endpoint_role, SwitchEndpointRole::Host); assert!(s.is_primary); assert!(s.nmxc_enabled); @@ -362,6 +379,54 @@ mod tests { } } + #[tokio::test] + async fn test_static_switch_endpoint_omits_invalid_or_nil_domain_uuid() { + let nil_domain_uuid = NvLinkDomainId::nil().to_string(); + + let cases = [ + ("10.0.1.2", "11:22:33:44:55:67", "not-a-uuid"), + ("10.0.1.3", "11:22:33:44:55:68", nil_domain_uuid.as_str()), + ]; + + let configs = cases + .iter() + .copied() + .map(|(ip_address, mac_address, domain_uuid)| StaticBmcEndpoint { + ip: ip(ip_address), + port: Some(443), + mac: mac_address.to_string(), + username: "cumulus".to_string(), + password: Some("pass".to_string()), + machine: None, + power_shelf: None, + switch: Some(StaticSwitchEndpoint { + id: None, + serial: Some(mac_address.to_string()), + slot_number: None, + tray_index: None, + nvlink_domain_uuid: Some(domain_uuid.to_string()), + endpoint_role: StaticSwitchEndpointRole::Host, + is_primary: false, + nmxc_enabled: None, + nmxt_enabled: None, + }), + rack_id: None, + labels: Default::default(), + }) + .collect::>(); + + let source = StaticEndpointSource::from_config(&configs, &reqwest(), None, 10); + let endpoints = source.fetch_bmc_hosts().await.unwrap(); + + assert_eq!(endpoints.len(), cases.len()); + + assert!(endpoints.iter().all(|endpoint| { + endpoint + .switch_data() + .is_some_and(|switch| switch.nvlink_domain_uuid.is_none()) + })); + } + #[tokio::test] async fn test_static_endpoint_with_power_shelf_metadata() { let power_shelf_id = test_power_shelf_id("power-shelf-a"); diff --git a/crates/health/src/sink/events.rs b/crates/health/src/sink/events.rs index 639d7b823a..5832409396 100644 --- a/crates/health/src/sink/events.rs +++ b/crates/health/src/sink/events.rs @@ -641,8 +641,8 @@ mod tests { } fn nvlink_domain_id() -> NvLinkDomainId { - NvLinkDomainId::from_str("00000000-0000-0000-0000-000000000000") - .expect("valid NVLink domain id") + NvLinkDomainId::from_str("9f4b45ec-705a-4af4-89f7-a112bc9c8f4e") + .expect("valid non-nil NVLink domain id") } fn addr() -> BmcAddr { diff --git a/docs/architecture/health_aggregation.md b/docs/architecture/health_aggregation.md index 3952aca333..34a80c7dad 100644 --- a/docs/architecture/health_aggregation.md +++ b/docs/architecture/health_aggregation.md @@ -269,19 +269,19 @@ ranges or by interpreting the `health_ok` values provided by BMCs. Machine endpoints carry the inventory metadata needed to interpret hardware health in fleet context. This includes machine ID, primary Redfish system UUID, serial number, rack ID, rack placement, and NVLink domain UUID when present. -Switch endpoints carry switch ID, serial number, and rack placement when present. +Switch endpoints carry switch ID, serial number, rack placement, and NVLink domain UUID when present. **For local and test deployments**, you can configure explicit machine, switch, or power-shelf identity with `[[endpoint_sources.static_bmc_endpoints]]`. Note the following: - Static machine endpoints can include the same serial number, rack placement, and NVLink domain UUID metadata -- Static switch endpoints can include serial number and rack placement metadata +- Static switch endpoints can include serial number, rack placement, and NVLink domain UUID metadata - All static endpoints can provide `rack_id` and validated custom telemetry `labels` - The primary Redfish system UUID remains BMC-derived and cannot be overridden by a custom label The publishing sinks expose that inventory context using the conventions of the target backend: -- `[sinks.prometheus]` adds _machine_ metadata as metric labels named `machine_id`, `system_uuid`, `serial_number`, `rack_id`, `machine_slot_number`, `machine_tray_index`, and `nvlink_domain_uuid` _Switch_ metadata labels are `switch_id`, `serial_number`, `rack_id`, `switch_slot_number`, and `switch_tray_index`. Static endpoint custom labels keep their configured names. -- `[sinks.otlp]` adds _machine_ metadata as OTLP resource attributes named `machine.id`, `system.uuid`, `rack.id`, integer `machine.slot_number`, integer `machine.tray_index`, and `nvlink.domain.uuid`. _Switch_ metadata labels are `switch.id`, `rack.id`, integer `switch.slot_number`, and integer `switch.tray_index`. Static endpoint custom labels keep their configured names. +- `[sinks.prometheus]` adds _machine_ metadata as metric labels named `machine_id`, `system_uuid`, `serial_number`, `rack_id`, `machine_slot_number`, `machine_tray_index`, and `nvlink_domain_uuid`. _Switch_ metadata labels are `switch_id`, `serial_number`, `rack_id`, `switch_slot_number`, `switch_tray_index`, and `nvlink_domain_uuid`. Static endpoint custom labels keep their configured names. +- `[sinks.otlp]` adds the string resource attributes `collector.type` and either `bmc.endpoint` and `bmc.ip`, or `switch.endpoint` and `switch.ip` for host-side switch collection. Typed inventory adds the strings `component.type` and, when present, `rack.id`. _Machine_ metadata attributes are the strings `machine.id`, `system.uuid`, `machine.serial`, `driver.version`, and `nvlink.domain.uuid`, plus the integers `machine.slot_number` and `machine.tray_index`. _Switch_ metadata attributes are the strings `switch.id`, `switch.serial_number`, `switch.endpoint_role`, and `nvlink.domain.uuid`, the boolean `switch.is_primary`, and the integers `switch.slot_number` and `switch.tray_index`. Static endpoint custom labels are string resource attributes and keep their configured names. - `[sinks.health_report]`, `[sinks.rack_health_report]`, `[sinks.switch_health_report]`, and `[sinks.power_shelf_health_report]` use the same event context when submitting assessed health reports back to NICo API. The persisted `HealthReport` and `HealthProbeAlert` schemas remain the probe success/alert model described above. ### BMC inventory monitoring