diff --git a/crates/health/src/otlp/convert.rs b/crates/health/src/otlp/convert.rs index 0b18e3c6c2..2b7d6db5ae 100644 --- a/crates/health/src/otlp/convert.rs +++ b/crates/health/src/otlp/convert.rs @@ -20,7 +20,7 @@ use std::time::SystemTime; use super::collector_logs::ExportLogsServiceRequest; use super::collector_metrics::ExportMetricsServiceRequest; -use super::common::{AnyValue, KeyValue, any_value}; +use super::common::{AnyValue, ArrayValue, KeyValue, KeyValueList, any_value}; use super::logs::{LogRecord as OtlpLogRecord, ResourceLogs, ScopeLogs, SeverityNumber}; use super::metrics::{ Gauge as OtlpGauge, Metric as OtlpMetric, NumberDataPoint, ResourceMetrics, ScopeMetrics, @@ -28,7 +28,12 @@ use super::metrics::{ }; use super::resource::Resource; use crate::endpoint::SwitchEndpointRole; -use crate::sink::{CollectorEvent, EventContext, MetricSample}; +use crate::sink::{ + CollectorEvent, EventContext, HealthReport, HealthReportAlert, HealthReportSuccess, + MetricSample, +}; + +const HEALTH_REPORT_SCHEMA_VERSION: &str = "v1"; fn severity_text_to_number(severity: &str) -> i32 { match severity.to_uppercase().as_str() { @@ -42,10 +47,14 @@ fn severity_text_to_number(severity: &str) -> i32 { } } -fn string_value(s: String) -> Option { - Some(AnyValue { +fn string_any_value(s: String) -> AnyValue { + AnyValue { value: Some(any_value::Value::StringValue(s)), - }) + } +} + +fn string_value(s: String) -> Option { + Some(string_any_value(s)) } fn int_value(value: i64) -> Option { @@ -68,6 +77,115 @@ fn int_kv(key: &str, value: i64) -> KeyValue { } } +fn any_kv(key: &str, value: AnyValue) -> KeyValue { + KeyValue { + key: key.to_string(), + value: Some(value), + } +} + +fn array_any_value(values: Vec) -> AnyValue { + AnyValue { + value: Some(any_value::Value::ArrayValue(ArrayValue { values })), + } +} + +fn kvlist_any_value(values: Vec) -> AnyValue { + AnyValue { + value: Some(any_value::Value::KvlistValue(KeyValueList { values })), + } +} + +fn health_report_success_value(success: &HealthReportSuccess) -> AnyValue { + let mut values = vec![kv("probe_id", success.probe_id.as_str().to_string())]; + if let Some(target) = &success.target { + values.push(kv("target", target.clone())); + } + + kvlist_any_value(values) +} + +fn health_report_alert_value(alert: &HealthReportAlert) -> AnyValue { + let mut values = vec![ + kv("probe_id", alert.probe_id.as_str().to_string()), + kv("message", alert.message.clone()), + any_kv( + "classifications", + array_any_value( + alert + .classifications + .iter() + .map(|classification| string_any_value(classification.as_str().to_string())) + .collect(), + ), + ), + ]; + if let Some(target) = &alert.target { + values.push(kv("target", target.clone())); + } + + kvlist_any_value(values) +} + +fn health_report_attributes(report: &HealthReport) -> Vec { + let mut attributes = vec![ + kv("event.type", "health_report".to_string()), + kv( + "health_report.schema_version", + HEALTH_REPORT_SCHEMA_VERSION.to_string(), + ), + kv("health_report.source", report.source.as_str().to_string()), + int_kv( + "health_report.success_count", + i64::try_from(report.successes.len()).unwrap_or(i64::MAX), + ), + int_kv( + "health_report.alert_count", + i64::try_from(report.alerts.len()).unwrap_or(i64::MAX), + ), + any_kv( + "health_report.successes", + array_any_value( + report + .successes + .iter() + .map(health_report_success_value) + .collect(), + ), + ), + any_kv( + "health_report.alerts", + array_any_value( + report + .alerts + .iter() + .map(health_report_alert_value) + .collect(), + ), + ), + ]; + if let Some(target) = report.target { + attributes.push(kv("health_report.target", target.as_str().to_string())); + } + if let Some(observed_at) = &report.observed_at { + attributes.push(kv( + "health_report.observed_at", + observed_at.to_rfc3339_opts(chrono::SecondsFormat::Nanos, true), + )); + } + + attributes +} + +fn health_report_event_time(report: &HealthReport, fallback_nanos: u64) -> u64 { + report + .observed_at + .as_ref() + .and_then(chrono::DateTime::timestamp_nanos_opt) + .and_then(|nanos| u64::try_from(nanos).ok()) + .unwrap_or(fallback_nanos) +} + fn resource_group_key(context: &EventContext) -> String { format!("{}|{}", context.endpoint_key, context.collector_type) } @@ -182,12 +300,12 @@ fn convert_event(event: &CollectorEvent, observed_nanos: u64) -> Option EventContext { @@ -399,6 +520,35 @@ mod tests { }) } + fn any_array_value(value: &AnyValue) -> Option<&[AnyValue]> { + match value.value.as_ref()? { + any_value::Value::ArrayValue(value) => Some(value.values.as_slice()), + _ => None, + } + } + + fn any_kvlist_value(value: &AnyValue) -> Option<&[KeyValue]> { + match value.value.as_ref()? { + any_value::Value::KvlistValue(value) => Some(value.values.as_slice()), + _ => None, + } + } + + fn any_string_value(value: &AnyValue) -> Option<&str> { + match value.value.as_ref()? { + any_value::Value::StringValue(value) => Some(value.as_str()), + _ => None, + } + } + + fn attr_array_value<'a>(attrs: &'a [KeyValue], key: &str) -> Option<&'a [AnyValue]> { + attrs + .iter() + .find(|attr| attr.key == key) + .and_then(|attr| attr.value.as_ref()) + .and_then(any_array_value) + } + #[test] fn resource_attributes_include_machine_metadata_when_present() { let domain_uuid = NvLinkDomainId::nil(); @@ -728,19 +878,26 @@ mod tests { } #[test] - fn health_report_converts_with_alert_severity() { + fn health_report_converts_with_structured_evidence() { let ctx = test_context(); + let observed_at = Utc + .with_ymd_and_hms(2026, 7, 31, 12, 34, 56) + .single() + .expect("valid timestamp"); let report = CollectorEvent::HealthReport( HealthReport { - source: ReportSource::BmcSensors, - target: None, - observed_at: None, - successes: vec![], + source: ReportSource::NvueLeakage, + target: Some(HealthReportTarget::Switch), + observed_at: Some(observed_at), + successes: vec![HealthReportSuccess { + probe_id: Probe::NvueLeakage, + target: Some("LEAK1".to_string()), + }], alerts: vec![HealthReportAlert { - probe_id: Probe::Sensor, - target: Some("Temp1".to_string()), - message: "critical".to_string(), - classifications: vec![Classification::SensorCritical], + probe_id: Probe::NvueLeakage, + target: Some("LEAK2".to_string()), + message: "NVUE leakage sensor LEAK2 reports leak".to_string(), + classifications: vec![Classification::Leak, Classification::SensorFailure], }], } .into(), @@ -748,7 +905,134 @@ mod tests { let request = build_export_request(&[(ctx, report)]); let records = &request.resource_logs[0].scope_logs[0].log_records; - assert_eq!(records[0].severity_text, "WARN"); + let record = &records[0]; + let attrs = record.attributes.as_slice(); + + assert_eq!(record.severity_text, "WARN"); + assert_eq!( + record.body.as_ref().and_then(any_string_value), + Some("health report: 1 alerts, 1 ok (source: NvueLeakage)") + ); + assert_eq!(attr_value(attrs, "event.type"), Some("health_report")); + assert_eq!( + record.time_unix_nano, + u64::try_from( + observed_at + .timestamp_nanos_opt() + .expect("timestamp has nanoseconds") + ) + .expect("timestamp is after the Unix epoch") + ); + assert_eq!( + attr_value(attrs, "health_report.schema_version"), + Some("v1") + ); + assert_eq!( + attr_value(attrs, "health_report.source"), + Some("nvue-leakage") + ); + assert_eq!(attr_value(attrs, "health_report.target"), Some("switch")); + assert_eq!( + attr_value(attrs, "health_report.observed_at"), + Some("2026-07-31T12:34:56.000000000Z") + ); + assert_eq!( + attr_int_value(attrs, "health_report.success_count"), + Some(1) + ); + assert_eq!(attr_int_value(attrs, "health_report.alert_count"), Some(1)); + + let successes = attr_array_value(attrs, "health_report.successes").expect("success array"); + let success = any_kvlist_value(&successes[0]).expect("structured success"); + assert_eq!(attr_value(success, "probe_id"), Some("NvueLeakage")); + assert_eq!(attr_value(success, "target"), Some("LEAK1")); + + let alerts = attr_array_value(attrs, "health_report.alerts").expect("alert array"); + let alert = any_kvlist_value(&alerts[0]).expect("structured alert"); + assert_eq!(attr_value(alert, "probe_id"), Some("NvueLeakage")); + assert_eq!(attr_value(alert, "target"), Some("LEAK2")); + assert_eq!( + attr_value(alert, "message"), + Some("NVUE leakage sensor LEAK2 reports leak") + ); + let classifications = + attr_array_value(alert, "classifications").expect("classification array"); + assert_eq!( + classifications + .iter() + .filter_map(any_string_value) + .collect::>(), + vec!["Leak", "SensorFailure"] + ); + } + + /// Export timestamp handed to `convert_event` so timestamp expectations stay + /// independent of the wall clock. + const EXPORT_NANOS: u64 = 1_784_500_000_000_000_000; + + #[derive(Debug, PartialEq)] + struct RecordTimestamps { + time_unix_nano: u64, + observed_time_unix_nano: u64, + } + + fn health_report_timestamps(observed_at: Option>) -> RecordTimestamps { + let event = CollectorEvent::HealthReport( + HealthReport { + source: ReportSource::BmcSensors, + target: Some(HealthReportTarget::Machine), + observed_at, + successes: vec![HealthReportSuccess { + probe_id: Probe::Sensor, + target: Some("Temp1".to_string()), + }], + alerts: vec![], + } + .into(), + ); + let record = convert_event(&event, EXPORT_NANOS).expect("health report converts"); + + RecordTimestamps { + time_unix_nano: record.time_unix_nano, + observed_time_unix_nano: record.observed_time_unix_nano, + } + } + + #[test] + fn health_report_event_time_prefers_the_observation_time() { + let observed_at = Utc + .with_ymd_and_hms(2026, 7, 31, 12, 34, 56) + .single() + .expect("valid timestamp"); + let observed_nanos = u64::try_from( + observed_at + .timestamp_nanos_opt() + .expect("timestamp has nanoseconds"), + ) + .expect("timestamp is after the Unix epoch"); + + value_scenarios!(health_report_timestamps: + "observation time present" { + Some(observed_at) => RecordTimestamps { + time_unix_nano: observed_nanos, + observed_time_unix_nano: EXPORT_NANOS, + }, + } + + "observation time absent" { + None => RecordTimestamps { + time_unix_nano: EXPORT_NANOS, + observed_time_unix_nano: EXPORT_NANOS, + }, + } + + "observation time before the Unix epoch" { + Utc.with_ymd_and_hms(1969, 12, 31, 23, 59, 59).single() => RecordTimestamps { + time_unix_nano: EXPORT_NANOS, + observed_time_unix_nano: EXPORT_NANOS, + }, + } + ); } #[test] diff --git a/crates/health/src/sink/events.rs b/crates/health/src/sink/events.rs index 2cab6dee9d..a9931b8400 100644 --- a/crates/health/src/sink/events.rs +++ b/crates/health/src/sink/events.rs @@ -42,6 +42,17 @@ pub enum HealthReportTarget { Switch, } +impl HealthReportTarget { + pub const fn as_str(self) -> &'static str { + match self { + Self::Machine => "machine", + Self::PowerShelf => "power-shelf", + Self::Rack => "rack", + Self::Switch => "switch", + } + } +} + #[derive(Clone, Debug)] pub struct EventContext { pub endpoint_key: String, @@ -862,6 +873,28 @@ mod tests { ); } + #[test] + fn report_target_strings() { + value_scenarios!( + run = HealthReportTarget::as_str; + "machine" { + HealthReportTarget::Machine => "machine", + } + + "power shelf" { + HealthReportTarget::PowerShelf => "power-shelf", + } + + "rack" { + HealthReportTarget::Rack => "rack", + } + + "switch" { + HealthReportTarget::Switch => "switch", + } + ); + } + #[test] fn probe_conversions() { value_scenarios!( diff --git a/docs/architecture/health_aggregation.md b/docs/architecture/health_aggregation.md index 3952aca333..c85d24409c 100644 --- a/docs/architecture/health_aggregation.md +++ b/docs/architecture/health_aggregation.md @@ -284,6 +284,40 @@ The publishing sinks expose that inventory context using the conventions of the - `[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.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. +#### OTLP health-report log contract + +OTLP health-report logs keep the existing human-readable summary body and add a versioned structured attribute contract. Match `health_report.schema_version` against `v1` before decoding the nested `health_report.successes` and `health_report.alerts` entries; only the scalar attributes below are guaranteed to keep their shape across schema versions. + +| Attribute | OTLP type | Presence | Value | +| --- | --- | --- | --- | +| `event.type` | string | Always | Always `health_report`. Select on it to isolate health reports from the other logs the sink exports. | +| `health_report.schema_version` | string | Always | Currently `v1`. | +| `health_report.source` | string | Always | The collector that assessed the report: `bmc-sensors`, `bmc-events`, `bmc-leak-detectors`, `tray-leak-detection`, `rack-leak-detection`, `nvue-leakage`, or `gpu-inventory`. | +| `health_report.target` | string | Optional | The kind of inventory object assessed: `machine`, `power-shelf`, `rack`, or `switch`. Omitted when the report names no target. | +| `health_report.observed_at` | string | Optional | Observation time as RFC 3339 UTC with nanosecond precision, for example `2026-07-31T12:34:56.000000000Z`. Omitted when the report carries no observation time. | +| `health_report.success_count` | int | Always | Number of entries in `health_report.successes`. | +| `health_report.alert_count` | int | Always | Number of entries in `health_report.alerts`. | +| `health_report.successes` | array of `kvlist` | Always | One entry per succeeded probe, empty when the report has none. | +| `health_report.alerts` | array of `kvlist` | Always | One entry per raised alert, empty when the report has none. | + +Each count always equals the length of the array it describes, so consumers can aggregate on the scalar without decoding the nested arrays. + +Every `health_report.successes` entry carries these fields: + +| Field | OTLP type | Presence | Value | +| --- | --- | --- | --- | +| `probe_id` | string | Always | The probe that ran: `BmcSensor`, `IntrusionSensorTriggered`, `BmcLeakDetection`, `NvueLeakage`, or `SkuValidation`. See [Health probe IDs](health/health_probe_ids.md) for the shared probe-ID catalogue. | +| `target` | string | Optional | The probed component, such as a sensor or leak-detector ID. Omitted when the probe ID fully describes what was tested. | + +Every `health_report.alerts` entry carries the same `probe_id` and optional `target` fields, plus these fields: + +| Field | OTLP type | Presence | Value | +| --- | --- | --- | --- | +| `message` | string | Always | Human-readable description of the alert. | +| `classifications` | array of string | Always | Zero or more of `SensorOk`, `SensorWarning`, `SensorCritical`, `SensorFatal`, `SensorFailure`, `PreventAllocations`, `Leak`, and `LeakDetector`; see [Health alert classifications](health/health_alert_classifications.md) for the classifications NICo interprets. Unlike the reports submitted to NICo API, this array carries only the classifications the collector raised and does not add the `Hardware` marker. | + +The two record timestamps carry different clocks. `time_unix_nano` carries the report's own observation time whenever the report supplies one that is representable as Unix nanoseconds, and otherwise falls back to the export time, so it is never zero. `observed_time_unix_nano` is always the export time, which keeps export order recoverable for reports whose observation time is older or absent. + ### BMC inventory monitoring The Site Explorer process within NICo Core periodically queries all Host and DPU BMCs in order to record certain BMC properties (e.g. components within a host and firmware versions).