diff --git a/.gitignore b/.gitignore index b26d6a52d8d..e1074be3031 100644 --- a/.gitignore +++ b/.gitignore @@ -107,3 +107,7 @@ orchagent/p4orch/tests/*_tr.xml build-env/.env build-env/custom-setup.sh + +# Perf / flamegraph profiling artifacts # +################### +perf.data* diff --git a/Cargo.toml b/Cargo.toml index 5f59681d703..0713070273c 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -5,6 +5,11 @@ members = [ ] exclude = [] +# Enable debug symbols in the bench profile so profilers (cargo-flamegraph, perf) +# can resolve function names. +[profile.bench] +debug = true + [workspace.package] version = "0.1.0" authors = ["SONiC"] diff --git a/crates/countersyncd/benches/otel_actor_perf.rs b/crates/countersyncd/benches/otel_actor_perf.rs index e6333c551c2..80033dd4b4d 100644 --- a/crates/countersyncd/benches/otel_actor_perf.rs +++ b/crates/countersyncd/benches/otel_actor_perf.rs @@ -81,7 +81,7 @@ fn build_stats_message(counters: usize, seed: u64) -> SAIStatsMessage { Arc::new(SAIStats::new(seed, stats)) } -async fn run_stream(prepared: PreparedDataset, endpoint: String) -> (std::time::Duration, usize) { +async fn run_stream(messages: Vec, total_counters: usize, endpoint: String) -> (std::time::Duration, usize) { let (tx, rx) = mpsc::channel(1024); let (shutdown_tx, _shutdown_rx) = oneshot::channel(); @@ -97,14 +97,11 @@ async fn run_stream(prepared: PreparedDataset, endpoint: String) -> (std::time:: let handle = tokio::spawn(async move { actor.run().await }); - let total_counters = prepared.expected_counters; let start = std::time::Instant::now(); - for tmpl in prepared.templates.iter() { - for msg_idx in 0..tmpl.records { - let msg = build_stats_message(tmpl.spec.counters, msg_idx as u64); - tx.send(msg).await.expect("send stats"); - } + // Only sending + actor conversion/encode/send is timed + for msg in messages { + tx.send(msg).await.expect("send stats"); } drop(tx); // close channel so actor exits after processing @@ -154,16 +151,32 @@ fn bench_otel_actor(c: &mut Criterion) { b.to_async(&rt).iter_batched( { let spec = spec.clone(); - move || PreparedDataset::new(spec.clone()) + move || { + // Build all input messages outside the profiled window so the + // flamegraph reflects only conversion + protobuf + gRPC send. + let prepared = PreparedDataset::new(spec.clone()); + let total_counters = prepared.expected_counters; + let mut messages = Vec::new(); + for tmpl in prepared.templates.iter() { + for msg_idx in 0..tmpl.records { + messages.push(build_stats_message( + tmpl.spec.counters, + msg_idx as u64, + )); + } + } + (messages, total_counters) + } }, - move |prepared| { + move |(messages, total_counters)| { let endpoint = endpoint.clone(); let exports_counter = exports_counter.clone(); let spec = spec.clone(); async move { let exports_before = exports_counter.load(Ordering::Relaxed); - let (elapsed, counters) = run_stream(prepared, endpoint.clone()).await; + let (elapsed, counters) = + run_stream(messages, total_counters, endpoint.clone()).await; let exports_after = exports_counter.load(Ordering::Relaxed); let exported = exports_after.saturating_sub(exports_before); @@ -174,7 +187,7 @@ fn bench_otel_actor(c: &mut Criterion) { ); } }, - BatchSize::SmallInput, + BatchSize::PerIteration, ) }); } diff --git a/crates/countersyncd/src/actor/otel.rs b/crates/countersyncd/src/actor/otel.rs index 7c68180d3b5..1615a336601 100644 --- a/crates/countersyncd/src/actor/otel.rs +++ b/crates/countersyncd/src/actor/otel.rs @@ -23,7 +23,6 @@ use opentelemetry_proto::tonic::{ KeyValue as ProtoKeyValue, }, metrics::v1::{ - Gauge as ProtoGauge, Metric, ResourceMetrics, ScopeMetrics, @@ -31,8 +30,8 @@ use opentelemetry_proto::tonic::{ resource::v1::Resource as ProtoResource, }; use crate::message::{ - otel::OtelMetrics, - saistats::SAIStatsMessage, + otel::{sai_stats_to_proto_metrics, DisplaySaiStats}, + saistats::{SAIStats, SAIStatsMessage}, }; use crate::utilities::{record_comm_stats, ChannelLabel}; @@ -89,8 +88,9 @@ pub struct OtelActor { resource: ProtoResource, instrumentation_scope: InstrumentationScope, - // Batching - buffer: Vec, + // Batching — buffers the raw SAI messages; each is converted straight to + // its final protobuf form at flush time + buffer: Vec, buffered_counters: usize, flush_deadline: TokioInstant, @@ -227,15 +227,14 @@ impl OtelActor { let was_empty = self.buffer.is_empty(); - // Convert to OTel format using message types and buffer - let otel_metrics = OtelMetrics::from_sai_stats(&stats); let counters_in_message = stats.stats.len(); + // The export path converts the buffered SAI message directly if log::log_enabled!(log::Level::Debug) { - self.print_otel_metrics(&otel_metrics).await; + self.print_stats_report(&stats); } - self.buffer.push(otel_metrics); + self.buffer.push(stats); self.buffered_counters += counters_in_message; // Start timeout when buffer transitions from empty to non-empty @@ -252,41 +251,20 @@ impl OtelActor { Ok(()) } - async fn print_otel_metrics(&mut self, otel_metrics: &OtelMetrics) { + fn print_stats_report(&mut self, stats: &SAIStats) { self.console_reports += 1; debug!( - "[OTel Report #{}] Service: {}, Scope: {} v{}, Total Gauges: {}, Messages Received: {}, Exports: {} (Failures: {})", + "[OTel Report #{}] Service: countersyncd, Scope: countersyncd v1.0, Counters: {}, Messages Received: {}, Exports: {} (Failures: {})", self.console_reports, - otel_metrics.service_name, - otel_metrics.scope_name, - otel_metrics.scope_version, - otel_metrics.len(), + stats.stats.len(), self.messages_received, self.exports_performed, self.export_failures ); - if !otel_metrics.is_empty() { - debug!("Gauge Metrics:"); - for (index, gauge) in otel_metrics.gauges.iter().enumerate() { - let data_point = &gauge.data_points[0]; - - debug!("[{:3}] Gauge: {}", index + 1, gauge.name); - debug!("Value: {}", data_point.value); - debug!("Unit: {}", gauge.unit); - debug!("Time: {}ns", data_point.time_unix_nano); - debug!("Description: {}", gauge.description); - - if !data_point.attributes.is_empty() { - debug!("Attributes:"); - for attr in &data_point.attributes { - debug!(" - {}={}", attr.key, attr.value); - } - } - - debug!("Raw Gauge: {:#?}", gauge); - } + if !stats.stats.is_empty() { + debug!("SAI counters:\n{}", DisplaySaiStats(stats)); } } @@ -314,23 +292,27 @@ impl OtelActor { self.client.as_mut() } - async fn send_request( - &mut self, - request: ExportMetricsServiceRequest, - ) -> Result<(), Box> { + async fn send_request(&mut self) -> Result<(), Box> { for attempt in 1..=MAX_EXPORT_RETRIES { - // Ensure we have a client - let client = match self.get_client() { - Some(c) => c, // Use existing or newly created client - _none => { // Failed to create client - self.client = None; - self.backoff(attempt).await; // Wait before retrying - continue; - } + // Ensure a client can be created before doing request + // construction; when get_client() fails the build below is skipped. + if self.get_client().is_none() { + self.client = None; + self.backoff(attempt).await; // Wait before retrying + continue; + } + + // Client is available, build a fresh request + let request = match self.build_export_request() { + Some(r) => r, + None => return Ok(()), }; + // Re-borrow the client for the actual send + let client = self.client.as_mut().expect("client ensured above"); + // Attempt to send the request - match client.export(request.clone()).await { + match client.export(request).await { Ok(_) => { // Successful export self.exports_performed += 1; self.consecutive_failures = 0; @@ -349,38 +331,18 @@ impl OtelActor { Err(Box::new(OtelActorExportError("Max export retries exceeded".to_string()))) } - // Export buffered metrics to OpenTelemetry collector - async fn flush_buffer(&mut self) -> Result<(), Box> { - if self.buffer.is_empty() { - return Ok(()); - } - + /// Build an export request from the currently buffered metrics. + /// Returns `None` when there is nothing to export (empty metrics). + /// The buffer is left intact so it can be rebuilt on retry. + fn build_export_request(&self) -> Option { let mut proto_metrics: Vec = Vec::new(); - for otel_metrics in &self.buffer { - for gauge in &otel_metrics.gauges { - let proto_data_points = gauge.data_points.iter() - .map(|dp| dp.to_proto()) - .collect(); - - let proto_gauge = ProtoGauge { - data_points: proto_data_points, - }; - - proto_metrics.push(Metric { - name: gauge.name.clone(), - description: gauge.description.clone(), - metadata: vec![], - data: Some(opentelemetry_proto::tonic::metrics::v1::metric::Data::Gauge(proto_gauge)), - ..Default::default() - }); - } + for stats in &self.buffer { + proto_metrics.extend(sai_stats_to_proto_metrics(stats)); } if proto_metrics.is_empty() { - self.buffer.clear(); - self.buffered_counters = 0; - return Ok(()); + return None; } let resource_metrics = ResourceMetrics { @@ -393,12 +355,20 @@ impl OtelActor { schema_url: String::new(), }; - let request = ExportMetricsServiceRequest { + Some(ExportMetricsServiceRequest { resource_metrics: vec![resource_metrics], - }; + }) + } + + // Export buffered metrics to OpenTelemetry collector + async fn flush_buffer(&mut self) -> Result<(), Box> { + if self.buffer.is_empty() { + return Ok(()); + } - // Send the export request - let result = self.send_request(request).await; + // The request is built lazily inside send_request from the intact + // buffer (rebuilt per retry) + let result = self.send_request().await; if let Err(e) = &result { self.export_failures += 1; diff --git a/crates/countersyncd/src/message/otel.rs b/crates/countersyncd/src/message/otel.rs index 63f532e6c28..13c407903b7 100644 --- a/crates/countersyncd/src/message/otel.rs +++ b/crates/countersyncd/src/message/otel.rs @@ -3,414 +3,487 @@ //! This module defines data structures for converting SAI statistics //! to OpenTelemetry gauge format for export to observability systems. -use crate::message::saistats::{SAIStat, SAIStats}; +use std::borrow::Cow; +use std::collections::HashMap; +use std::fmt; + +use crate::message::saistats::SAIStats; +use crate::sai::{ + SaiBufferPoolStat, SaiIngressPriorityGroupStat, SaiObjectType, SaiPortStat, SaiQueueStat, +}; use opentelemetry_proto::tonic::{ common::v1::{KeyValue as ProtoKeyValue, AnyValue, any_value::Value}, - metrics::v1::{NumberDataPoint, number_data_point}, + metrics::v1::{NumberDataPoint, number_data_point, Gauge as ProtoGauge, Metric, metric}, }; -/// OpenTelemetry Gauge representation for SAI statistics -/// -/// This struct represents an OpenTelemetry gauge metric following the OTLP protocol. -/// Each gauge contains data points with attributes, timestamps, and values derived -/// from SAI statistics. -#[derive(Debug, Clone, PartialEq)] -pub struct OtelGauge { - /// Metric name (e.g., "sai_counter_type_100_stat_200") - pub name: String, - /// Description of the metric - pub description: String, - /// Unit of measurement (typically "1" for counters) - pub unit: String, - /// Data points for this gauge - pub data_points: Vec, -} - -/// OpenTelemetry Data Point for a single measurement -/// -/// Represents a single measurement point in time for a gauge metric, -/// converted from a SAI statistic entry. -#[derive(Debug, Clone, PartialEq)] -pub struct OtelDataPoint { - /// Attributes (labels) for this data point - pub attributes: Vec, - /// Timestamp in nanoseconds since Unix epoch - pub time_unix_nano: u64, - /// The gauge value (converted from SAI counter) - pub value: i64, -} - -/// OpenTelemetry Attribute (Key-Value Pair) -/// -/// Represents a single attribute/label attached to a metric data point. -#[derive(Debug, Clone, PartialEq)] -pub struct OtelAttribute { - /// Attribute key - pub key: String, - /// Attribute value - pub value: String, +/// Returns the readable SAI object-type name for a `type_id` +/// (e.g. `1` -> `"SAI_OBJECT_TYPE_PORT"`). Unknown ids fall back to a +/// synthetic name so no information is lost. +fn sai_type_name(type_id: u32) -> Cow<'static, str> { + match SaiObjectType::from_u32(type_id) { + Some(object_type) => Cow::Borrowed(object_type.to_c_name()), + None => Cow::Owned(format!("SAI_OBJECT_TYPE_UNKNOWN_{}", type_id)), + } } -impl OtelAttribute { - /// Creates a new OtelAttribute - pub fn new(key: impl Into, value: impl Into) -> Self { - Self { - key: key.into(), - value: value.into(), +/// Returns the readable SAI stat name for a `(type_id, stat_id)` pair +/// (e.g. `(1, 1)` -> `"SAI_PORT_STAT_IF_IN_UCAST_PKTS"`), dispatching on the +/// object type. Unknown ids fall back to a synthetic name. +fn sai_stat_name(type_id: u32, stat_id: u32) -> Cow<'static, str> { + let name = SaiObjectType::from_u32(type_id).and_then(|object_type| match object_type { + SaiObjectType::Port => SaiPortStat::from_u32(stat_id).map(|s| s.to_c_name()), + SaiObjectType::Queue => SaiQueueStat::from_u32(stat_id).map(|s| s.to_c_name()), + SaiObjectType::BufferPool => SaiBufferPoolStat::from_u32(stat_id).map(|s| s.to_c_name()), + SaiObjectType::IngressPriorityGroup => { + SaiIngressPriorityGroupStat::from_u32(stat_id).map(|s| s.to_c_name()) } - } + _ => None, + }); - /// Converts to OpenTelemetry protobuf KeyValue - pub fn to_proto(&self) -> ProtoKeyValue { - ProtoKeyValue { - key: self.key.clone(), - value: Some(AnyValue { - value: Some(Value::StringValue(self.value.clone())), - }), - } + match name { + Some(c_name) => Cow::Borrowed(c_name), + None => Cow::Owned(format!("SAI_STAT_UNKNOWN_TYPE_{}_STAT_{}", type_id, stat_id)), } } -impl OtelDataPoint { - /// Creates a new OtelDataPoint from SAI statistic - pub fn from_sai_stat(sai_stat: &SAIStat, observation_time_nano: u64) -> Self { - let attributes = vec![ - OtelAttribute::new("object_name", &sai_stat.object_name), - OtelAttribute::new("sai_type_id", sai_stat.type_id.to_string()), - OtelAttribute::new("sai_stat_id", sai_stat.stat_id.to_string()), - ]; - - Self { - attributes, - time_unix_nano: observation_time_nano, - value: sai_stat.counter as i64, - } - } - - /// Converts to OpenTelemetry protobuf NumberDataPoint - pub fn to_proto(&self) -> NumberDataPoint { - NumberDataPoint { - time_unix_nano: self.time_unix_nano, - value: Some(number_data_point::Value::AsInt(self.value)), - attributes: self.attributes.iter().map(|attr| attr.to_proto()).collect(), - ..Default::default() - } +/// Builds an OTLP protobuf `KeyValue` attribute with a string value. +fn proto_string_attr(key: &'static str, value: String) -> ProtoKeyValue { + ProtoKeyValue { + key: key.to_string(), + value: Some(AnyValue { + value: Some(Value::StringValue(value)), + }), } } -impl OtelGauge { - /// Creates a new OtelGauge from SAI statistic - pub fn from_sai_stat(sai_stat: &SAIStat, observation_time_nano: u64) -> Self { - let name = format!("sai_counter_type_{}_stat_{}", sai_stat.type_id, sai_stat.stat_id); - let description = format!( - "SAI counter for object {} (type:{}, stat:{})", - sai_stat.object_name, sai_stat.type_id, sai_stat.stat_id - ); - - let data_point = OtelDataPoint::from_sai_stat(sai_stat, observation_time_nano); - - Self { - name, - description, - unit: "1".to_string(), - data_points: vec![data_point], - } - } +/// Converts a SAI statistics collection directly into OTLP protobuf `Metric`s, +/// grouping stats that share the same `(type_id, stat_id)` into a single gauge +/// with one data point per object. +pub fn sai_stats_to_proto_metrics(sai_stats: &SAIStats) -> Vec { + let observation_time_nano = sai_stats.observation_time; - /// Creates multiple OtelGauges from SAI statistics collection - pub fn from_sai_stats(sai_stats: &SAIStats) -> Vec { - // Use the observation_time from the SAI statistics - let observation_time_nano = sai_stats.observation_time; + let mut index: HashMap<(u32, u32), usize> = HashMap::new(); + let mut metrics: Vec = Vec::new(); - sai_stats.stats - .iter() - .map(|stat| Self::from_sai_stat(stat, observation_time_nano)) - .collect() - } -} + for stat in &sai_stats.stats { + let key = (stat.type_id, stat.stat_id); -/// Collection of OpenTelemetry gauges with metadata -/// -/// This structure represents a collection of OpenTelemetry gauges -/// derived from SAI statistics, ready for export to collectors. -#[derive(Debug, Clone)] -pub struct OtelMetrics { - /// Service name for resource attribution - pub service_name: String, - /// Instrumentation scope name - pub scope_name: String, - /// Instrumentation scope version - pub scope_version: String, - /// Collection of gauge metrics - pub gauges: Vec, -} + let data_point = NumberDataPoint { + time_unix_nano: observation_time_nano, + value: Some(number_data_point::Value::AsInt(stat.counter as i64)), + attributes: vec![ + proto_string_attr("object_name", stat.object_name.clone()), + proto_string_attr("sai_type_name", sai_type_name(stat.type_id).into_owned()), + proto_string_attr( + "sai_stat_name", + sai_stat_name(stat.type_id, stat.stat_id).into_owned(), + ), + ], + ..Default::default() + }; -impl OtelMetrics { - /// Creates OtelMetrics from SAI statistics - pub fn from_sai_stats(sai_stats: &SAIStats) -> Self { - let gauges = OtelGauge::from_sai_stats(sai_stats); + let metric_index = match index.get(&key).copied() { + Some(i) => i, + None => { + let i = metrics.len(); + index.insert(key, i); + + let type_name = sai_type_name(stat.type_id); + let stat_name = sai_stat_name(stat.type_id, stat.stat_id); + metrics.push(Metric { + name: stat_name.clone().into_owned(), + description: format!("{} / {}", type_name, stat_name), + metadata: vec![], + data: Some(metric::Data::Gauge(ProtoGauge { + data_points: Vec::new(), + })), + ..Default::default() + }); + + i + } + }; - Self { - service_name: "countersyncd".to_string(), - scope_name: "countersyncd".to_string(), - scope_version: "1.0".to_string(), - gauges, + if let Some(metric::Data::Gauge(gauge)) = metrics[metric_index].data.as_mut() { + gauge.data_points.push(data_point); } } - /// Returns the number of gauges in this collection - pub fn len(&self) -> usize { - self.gauges.len() - } + metrics +} - /// Returns true if this collection is empty - pub fn is_empty(&self) -> bool { - self.gauges.is_empty() +/// Human-readable rendering of a [`SAIStats`] batch for debug logging. +pub struct DisplaySaiStats<'a>(pub &'a SAIStats); + +impl fmt::Display for DisplaySaiStats<'_> { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + let stats = self.0; + writeln!( + f, + "SAIStats @ {}ns ({} counters)", + stats.observation_time, + stats.stats.len() + )?; + for stat in &stats.stats { + writeln!( + f, + " {} {} / {} = {}", + stat.object_name, + sai_type_name(stat.type_id), + sai_stat_name(stat.type_id, stat.stat_id), + stat.counter + )?; + } + Ok(()) } } #[cfg(test)] mod tests { use super::*; - use std::sync::Arc; - use crate::message::saistats::{SAIStat, SAIStats}; - use log::{info, debug}; - - /// Helper function to create test SAI statistics (similar to saistats.rs pattern) - fn create_test_sai_stats(observation_time: u64, stat_count: usize) -> SAIStats { - let stats = (0..stat_count) - .map(|i| SAIStat { - object_name: format!("Ethernet{}", i), - type_id: (i * 100 + 1) as u32, - stat_id: (i * 10 + 1) as u32, - counter: (i * 1000 + 500) as u64, - }) - .collect(); + use crate::message::saistats::SAIStat; - SAIStats::new(observation_time, stats) + /// Returns the `Gauge` data points of a proto `Metric` + fn gauge_points(metric: &Metric) -> &Vec { + match metric.data.as_ref() { + Some(metric::Data::Gauge(gauge)) => &gauge.data_points, + _ => panic!("metric {} is not a gauge", metric.name), + } } - #[test] - fn test_otel_attribute_creation() { - let attr = OtelAttribute::new("object_name", "Ethernet0"); - assert_eq!(attr.key, "object_name"); - assert_eq!(attr.value, "Ethernet0"); - - let attr2 = OtelAttribute::new("sai_type_id", "100"); - assert_eq!(attr2.key, "sai_type_id"); - assert_eq!(attr2.value, "100"); + /// Reads a string attribute value by key from a proto data point. + fn attr<'a>(dp: &'a NumberDataPoint, key: &str) -> &'a str { + dp.attributes + .iter() + .find(|kv| kv.key == key) + .and_then(|kv| kv.value.as_ref()) + .and_then(|v| match v.value.as_ref() { + Some(Value::StringValue(s)) => Some(s.as_str()), + _ => None, + }) + .unwrap_or_else(|| panic!("string attribute `{key}` missing")) } - #[test] - fn test_otel_data_point_from_sai_stat() { - let sai_stat = SAIStat { - object_name: "Ethernet0".to_string(), - type_id: 100, - stat_id: 200, - counter: 1500, - }; - - let observation_time_nano = 0u64; // 1970-01-01 00:00:00 UTC - let data_point = OtelDataPoint::from_sai_stat(&sai_stat, observation_time_nano); - - assert_eq!(data_point.time_unix_nano, observation_time_nano); - assert_eq!(data_point.value, 1500); - assert_eq!(data_point.attributes.len(), 3); - - // Check attributes - let object_name_attr = data_point.attributes.iter() - .find(|attr| attr.key == "object_name").unwrap(); - assert_eq!(object_name_attr.value, "Ethernet0"); - - let type_id_attr = data_point.attributes.iter() - .find(|attr| attr.key == "sai_type_id").unwrap(); - assert_eq!(type_id_attr.value, "100"); - - let stat_id_attr = data_point.attributes.iter() - .find(|attr| attr.key == "sai_stat_id").unwrap(); - assert_eq!(stat_id_attr.value, "200"); + /// Reads the integer value of a proto data point. + fn as_int(dp: &NumberDataPoint) -> i64 { + match dp.value { + Some(number_data_point::Value::AsInt(v)) => v, + _ => panic!("data point value is not AsInt"), + } } #[test] - fn test_otel_gauge_from_sai_stat() { - let sai_stat = SAIStat { - object_name: "BufferPool1".to_string(), - type_id: 24, - stat_id: 2, - counter: 5000, - }; - - let observation_time_nano = 0u64; // 1970-01-01 00:00:00 UTC - let gauge = OtelGauge::from_sai_stat(&sai_stat, observation_time_nano); - - assert_eq!(gauge.name, "sai_counter_type_24_stat_2"); - assert_eq!(gauge.description, "SAI counter for object BufferPool1 (type:24, stat:2)"); - assert_eq!(gauge.unit, "1"); - assert_eq!(gauge.data_points.len(), 1); - - let data_point = &gauge.data_points[0]; - assert_eq!(data_point.value, 5000); - assert_eq!(data_point.time_unix_nano, observation_time_nano); + fn test_sai_stats_to_proto_metrics_empty() { + let sai_stats = SAIStats::new(0, vec![]); + assert!(sai_stats_to_proto_metrics(&sai_stats).is_empty()); } #[test] - fn test_otel_gauge_from_sai_stats_collection() { - let sai_stats = create_test_sai_stats(1672531200, 3); - let gauges = OtelGauge::from_sai_stats(&sai_stats); + fn test_sai_stats_to_proto_metrics_single_stat_fields() { + let observation_time = 1_700_000_000u64; + let sai_stats = SAIStats::new( + observation_time, + vec![SAIStat { + object_name: "Ethernet0".to_string(), + type_id: 1, // SAI_OBJECT_TYPE_PORT + stat_id: 1, // SAI_PORT_STAT_IF_IN_UCAST_PKTS + counter: 12345, + }], + ); - assert_eq!(gauges.len(), 3); + let metrics = sai_stats_to_proto_metrics(&sai_stats); + assert_eq!(metrics.len(), 1, "one (type_id, stat_id) key -> one gauge"); - // Check first gauge - let first_gauge = &gauges[0]; - assert_eq!(first_gauge.name, "sai_counter_type_1_stat_1"); - assert!(first_gauge.description.contains("Ethernet0")); - assert_eq!(first_gauge.data_points[0].value, 500); + let metric = &metrics[0]; + assert_eq!(metric.name, "SAI_PORT_STAT_IF_IN_UCAST_PKTS"); + assert_eq!( + metric.description, + "SAI_OBJECT_TYPE_PORT / SAI_PORT_STAT_IF_IN_UCAST_PKTS" + ); - let expected_time_nano = 1672531200u64; - for gauge in &gauges { - assert_eq!(gauge.data_points[0].time_unix_nano, expected_time_nano); - } + let points = gauge_points(metric); + assert_eq!(points.len(), 1); + let dp = &points[0]; + assert_eq!(as_int(dp), 12345, "value comes from the SAI counter"); + assert_eq!(dp.time_unix_nano, observation_time); + + // Exactly the three expected attributes. + assert_eq!(dp.attributes.len(), 3); + assert_eq!(attr(dp, "object_name"), "Ethernet0"); + assert_eq!(attr(dp, "sai_type_name"), "SAI_OBJECT_TYPE_PORT"); + assert_eq!(attr(dp, "sai_stat_name"), "SAI_PORT_STAT_IF_IN_UCAST_PKTS"); } #[test] - fn test_otel_metrics_from_sai_stats() { + fn test_sai_stats_to_proto_metrics_groups_by_type_and_stat() { + let observation_time = 42u64; let sai_stats = SAIStats::new( - 1234567890, + observation_time, vec![ + // Two objects sharing the same (type_id, stat_id) must merge + // into a single gauge, one data point per object. SAIStat { object_name: "Ethernet0".to_string(), type_id: 1, stat_id: 1, - counter: 12345, + counter: 10, }, + SAIStat { + object_name: "Ethernet1".to_string(), + type_id: 1, + stat_id: 1, + counter: 20, + }, + // A different (type_id, stat_id) gets its own gauge. SAIStat { object_name: "BufferPool1".to_string(), type_id: 24, stat_id: 2, - counter: 67890, + counter: 30, }, ], ); - let otel_metrics = OtelMetrics::from_sai_stats(&sai_stats); - - assert_eq!(otel_metrics.service_name, "countersyncd"); - assert_eq!(otel_metrics.scope_name, "countersyncd"); - assert_eq!(otel_metrics.scope_version, "1.0"); - assert_eq!(otel_metrics.len(), 2); - assert!(!otel_metrics.is_empty()); - - // Check individual gauges - let port_gauge = otel_metrics.gauges.iter() - .find(|g| g.name == "sai_counter_type_1_stat_1").unwrap(); - assert_eq!(port_gauge.data_points[0].value, 12345); + let metrics = sai_stats_to_proto_metrics(&sai_stats); + assert_eq!( + metrics.len(), + 2, + "two distinct (type_id, stat_id) keys -> two gauges" + ); - let buffer_gauge = otel_metrics.gauges.iter() - .find(|g| g.name == "sai_counter_type_24_stat_2").unwrap(); - assert_eq!(buffer_gauge.data_points[0].value, 67890); + // First gauge: the shared (1, 1) key, one data point per object in + // input order. + let shared = &metrics[0]; + assert_eq!(shared.name, "SAI_PORT_STAT_IF_IN_UCAST_PKTS"); + let shared_points = gauge_points(shared); + assert_eq!(shared_points.len(), 2, "both objects merged into one gauge"); + assert_eq!(attr(&shared_points[0], "object_name"), "Ethernet0"); + assert_eq!(as_int(&shared_points[0]), 10); + assert_eq!(attr(&shared_points[1], "object_name"), "Ethernet1"); + assert_eq!(as_int(&shared_points[1]), 20); + + // Second gauge: the distinct (24, 2) key. + let other = &metrics[1]; + assert_eq!(other.name, "SAI_BUFFER_POOL_STAT_DROPPED_PACKETS"); + let other_points = gauge_points(other); + assert_eq!(other_points.len(), 1); + assert_eq!(attr(&other_points[0], "object_name"), "BufferPool1"); + assert_eq!(as_int(&other_points[0]), 30); } #[test] - fn test_otel_metrics_message_creation() { - let sai_stats = create_test_sai_stats(555555, 2); + fn test_sai_stats_to_proto_metrics_covers_all_supported_object_types() { + // (object_name, type_id, stat_id, expected_type_name, expected_stat_name) + // Spans every object type the converter dispatches on (port, queue, + // buffer pool, ingress priority group) with several distinct stat ids + // so both the type-name and stat-name lookups are exercised. + let cases: &[(&str, u32, u32, &str, &str)] = &[ + ( + "Ethernet0", + 1, + 1, + "SAI_OBJECT_TYPE_PORT", + "SAI_PORT_STAT_IF_IN_UCAST_PKTS", + ), + ( + "Ethernet0:Queue0", + 21, + 0, + "SAI_OBJECT_TYPE_QUEUE", + "SAI_QUEUE_STAT_PACKETS", + ), + ( + "Ethernet0:Queue1", + 21, + 2, + "SAI_OBJECT_TYPE_QUEUE", + "SAI_QUEUE_STAT_DROPPED_PACKETS", + ), + ( + "BufferPool0", + 24, + 0, + "SAI_OBJECT_TYPE_BUFFER_POOL", + "SAI_BUFFER_POOL_STAT_CURR_OCCUPANCY_BYTES", + ), + ( + "BufferPool0", + 24, + 2, + "SAI_OBJECT_TYPE_BUFFER_POOL", + "SAI_BUFFER_POOL_STAT_DROPPED_PACKETS", + ), + ( + "Ethernet0:PG0", + 26, + 0, + "SAI_OBJECT_TYPE_INGRESS_PRIORITY_GROUP", + "SAI_INGRESS_PRIORITY_GROUP_STAT_PACKETS", + ), + ( + "Ethernet0:PG0", + 26, + 8, + "SAI_OBJECT_TYPE_INGRESS_PRIORITY_GROUP", + "SAI_INGRESS_PRIORITY_GROUP_STAT_DROPPED_PACKETS", + ), + ]; - // Wrap metrics in Arc manually for sharing scenarios - let message1 = Arc::new(OtelMetrics::from_sai_stats(&sai_stats)); - let message2 = OtelMetrics::from_sai_stats(&sai_stats); + let observation_time = 9_000u64; + let stats: Vec = cases + .iter() + .enumerate() + .map(|(i, (obj, type_id, stat_id, _, _))| SAIStat { + object_name: (*obj).to_string(), + type_id: *type_id, + stat_id: *stat_id, + counter: (i as u64 + 1) * 100, + }) + .collect(); + let sai_stats = SAIStats::new(observation_time, stats); + + let metrics = sai_stats_to_proto_metrics(&sai_stats); + assert_eq!( + metrics.len(), + cases.len(), + "each distinct (type_id, stat_id) becomes its own gauge" + ); - assert_eq!(message1.service_name, message2.service_name); - assert_eq!(message1.len(), message2.len()); - assert_eq!(message1.gauges.len(), 2); + for (i, (obj, _type_id, _stat_id, type_name, stat_name)) in cases.iter().enumerate() { + let metric = &metrics[i]; + assert_eq!(metric.name, *stat_name, "gauge name for case {i}"); + assert_eq!( + metric.description, + format!("{type_name} / {stat_name}"), + "description for case {i}" + ); + + let points = gauge_points(metric); + assert_eq!(points.len(), 1, "case {i} has a single object"); + let dp = &points[0]; + assert_eq!(as_int(dp), (i as i64 + 1) * 100, "value for case {i}"); + assert_eq!(dp.time_unix_nano, observation_time); + assert_eq!(attr(dp, "object_name"), *obj, "object_name for case {i}"); + assert_eq!(attr(dp, "sai_type_name"), *type_name, "type name for case {i}"); + assert_eq!(attr(dp, "sai_stat_name"), *stat_name, "stat name for case {i}"); + } } #[test] - fn test_otel_data_point_proto_conversion() { - let sai_stat = SAIStat { - object_name: "TestInterface".to_string(), - type_id: 999, - stat_id: 888, - counter: 777, - }; + fn test_sai_stats_to_proto_metrics_unknown_ids_fallback() { + let sai_stats = SAIStats::new( + 7u64, + vec![ + // Unknown object type -> synthetic type and stat names. + SAIStat { + object_name: "Mystery0".to_string(), + type_id: 99_999, + stat_id: 5, + counter: 1, + }, + // Known type (Port) but unknown stat id -> real type name, + // synthetic stat name. + SAIStat { + object_name: "Ethernet0".to_string(), + type_id: 1, + stat_id: 888_888, + counter: 2, + }, + ], + ); - let data_point = OtelDataPoint::from_sai_stat(&sai_stat, 123456789); - let proto_point = data_point.to_proto(); + let metrics = sai_stats_to_proto_metrics(&sai_stats); + assert_eq!(metrics.len(), 2); - assert_eq!(proto_point.time_unix_nano, 123456789); - match proto_point.value.unwrap() { - number_data_point::Value::AsInt(val) => assert_eq!(val, 777), - _ => panic!("Expected integer value"), - } - assert_eq!(proto_point.attributes.len(), 3); - - // Check one attribute conversion - let object_attr = &proto_point.attributes[0]; - assert_eq!(object_attr.key, "object_name"); - if let Some(AnyValue { value: Some(Value::StringValue(val)) }) = &object_attr.value { - assert_eq!(val, "TestInterface"); - } else { - panic!("Expected string value"); - } + let unknown_type = &metrics[0]; + assert_eq!(unknown_type.name, "SAI_STAT_UNKNOWN_TYPE_99999_STAT_5"); + assert_eq!( + unknown_type.description, + "SAI_OBJECT_TYPE_UNKNOWN_99999 / SAI_STAT_UNKNOWN_TYPE_99999_STAT_5" + ); + let p0 = &gauge_points(unknown_type)[0]; + assert_eq!(attr(p0, "sai_type_name"), "SAI_OBJECT_TYPE_UNKNOWN_99999"); + assert_eq!(attr(p0, "sai_stat_name"), "SAI_STAT_UNKNOWN_TYPE_99999_STAT_5"); + + let unknown_stat = &metrics[1]; + assert_eq!(unknown_stat.name, "SAI_STAT_UNKNOWN_TYPE_1_STAT_888888"); + let p1 = &gauge_points(unknown_stat)[0]; + // The type resolves to a real name; only the stat id is unknown. + assert_eq!(attr(p1, "sai_type_name"), "SAI_OBJECT_TYPE_PORT"); + assert_eq!(attr(p1, "sai_stat_name"), "SAI_STAT_UNKNOWN_TYPE_1_STAT_888888"); } -#[test] -fn test_sai_to_otel_gauge_conversion() { - let test_stats = vec![ - SAIStat { object_name: "Ethernet0".to_string(), type_id: 1, stat_id: 1, counter: 1000000 }, - SAIStat { object_name: "Ethernet0".to_string(), type_id: 1, stat_id: 2, counter: 2000000 }, - SAIStat { object_name: "Ethernet1".to_string(), type_id: 1, stat_id: 1, counter: 1500000 }, - SAIStat { object_name: "BufferPool_ingress_lossless_pool".to_string(), type_id: 24, stat_id: 1, counter: 500000 }, - ]; - - let sai_stats = SAIStats::new(1672531200, test_stats); - let otel_metrics = OtelMetrics::from_sai_stats(&sai_stats); - - for (index, gauge) in otel_metrics.gauges.iter().enumerate() { - let data_point = &gauge.data_points[0]; - info!("[{}] Gauge: {}", index + 1, gauge.name); - info!("Value: {}, Unit: {}, Timestamp: {}ns", data_point.value, gauge.unit, data_point.time_unix_nano); - info!("Description: {}", gauge.description); - - if !data_point.attributes.is_empty() { - for attr in &data_point.attributes { - debug!(" - {}={}", attr.key, attr.value); - } - } - info!("Raw gauge: {:#?}", gauge); + #[test] + fn test_sai_type_name_known_and_unknown() { + assert_eq!(sai_type_name(1).as_ref(), "SAI_OBJECT_TYPE_PORT"); + assert_eq!(sai_type_name(21).as_ref(), "SAI_OBJECT_TYPE_QUEUE"); + assert_eq!(sai_type_name(24).as_ref(), "SAI_OBJECT_TYPE_BUFFER_POOL"); + assert_eq!( + sai_type_name(26).as_ref(), + "SAI_OBJECT_TYPE_INGRESS_PRIORITY_GROUP" + ); + assert_eq!(sai_type_name(99_999).as_ref(), "SAI_OBJECT_TYPE_UNKNOWN_99999"); } - assert_eq!(otel_metrics.len(), 4); - - // Verify port stats conversion - let port_stats: Vec<_> = otel_metrics.gauges.iter() - .filter(|g| g.description.contains("Ethernet")) - .collect(); - assert_eq!(port_stats.len(), 3); - - // Verify buffer pool stats conversion - let buffer_stats: Vec<_> = otel_metrics.gauges.iter() - .filter(|g| g.description.contains("BufferPool")) - .collect(); - assert_eq!(buffer_stats.len(), 1); - - // Check that all metrics have proper timestamps - let expected_time = 1672531200u64; - for gauge in &otel_metrics.gauges { - assert_eq!(gauge.data_points[0].time_unix_nano, expected_time); + #[test] + fn test_sai_stat_name_known_and_unknown() { + assert_eq!(sai_stat_name(1, 1).as_ref(), "SAI_PORT_STAT_IF_IN_UCAST_PKTS"); + assert_eq!(sai_stat_name(21, 0).as_ref(), "SAI_QUEUE_STAT_PACKETS"); + assert_eq!(sai_stat_name(21, 2).as_ref(), "SAI_QUEUE_STAT_DROPPED_PACKETS"); + assert_eq!( + sai_stat_name(24, 0).as_ref(), + "SAI_BUFFER_POOL_STAT_CURR_OCCUPANCY_BYTES" + ); + assert_eq!( + sai_stat_name(24, 2).as_ref(), + "SAI_BUFFER_POOL_STAT_DROPPED_PACKETS" + ); + assert_eq!( + sai_stat_name(26, 0).as_ref(), + "SAI_INGRESS_PRIORITY_GROUP_STAT_PACKETS" + ); + assert_eq!( + sai_stat_name(26, 8).as_ref(), + "SAI_INGRESS_PRIORITY_GROUP_STAT_DROPPED_PACKETS" + ); + // Unknown stat id on a known type, and an unknown type entirely. + assert_eq!( + sai_stat_name(1, 888_888).as_ref(), + "SAI_STAT_UNKNOWN_TYPE_1_STAT_888888" + ); + assert_eq!( + sai_stat_name(99_999, 5).as_ref(), + "SAI_STAT_UNKNOWN_TYPE_99999_STAT_5" + ); } - // Verify metric naming - let port_rx_metric = otel_metrics.gauges.iter() - .find(|g| g.name == "sai_counter_type_1_stat_1").unwrap(); - assert!(port_rx_metric.description.contains("type:1, stat:1")); -} - #[test] - fn test_empty_sai_stats_to_otel() { - let empty_stats = SAIStats::new(1111111111, vec![]); - let otel_metrics = OtelMetrics::from_sai_stats(&empty_stats); + fn test_display_sai_stats_debug_format() { + let sai_stats = SAIStats::new( + 1000, + vec![ + SAIStat { + object_name: "Ethernet0".to_string(), + type_id: 1, + stat_id: 1, + counter: 42, + }, + // Unknown ids still render via the synthetic-name fallback. + SAIStat { + object_name: "Mystery0".to_string(), + type_id: 99_999, + stat_id: 5, + counter: 7, + }, + ], + ); - assert_eq!(otel_metrics.len(), 0); - assert!(otel_metrics.is_empty()); - assert_eq!(otel_metrics.service_name, "countersyncd"); + let rendered = DisplaySaiStats(&sai_stats).to_string(); + assert!(rendered.contains("SAIStats @ 1000ns (2 counters)")); + assert!(rendered + .contains("Ethernet0 SAI_OBJECT_TYPE_PORT / SAI_PORT_STAT_IF_IN_UCAST_PKTS = 42")); + assert!(rendered.contains( + "Mystery0 SAI_OBJECT_TYPE_UNKNOWN_99999 / SAI_STAT_UNKNOWN_TYPE_99999_STAT_5 = 7" + )); } }