//! # Telemetry System for RustyTorch++ Graph Operations //! //! This module provides comprehensive telemetry capabilities for monitoring //! and analyzing the performance of graph operations. It implements: //! //! - Performance metrics collection (execution times, memory usage) //! - GPU utilization tracking and bandwidth measurements //! - Real-time bottleneck detection and analysis //! - OpenTelemetry and Prometheus export capabilities //! - Streaming metrics processing with anomaly detection //! //! ## Usage //! ```rust //! use rtx_graph::telemetry::{TelemetryCollector, MetricType}; //! //! let mut collector = TelemetryCollector::new(); //! collector.start_operation("matrix_multiply"); //! // ... perform operation //! collector.end_operation("matrix_multiply"); //! //! let metrics = collector.get_metrics(); //! println!("Execution time: {:?}", metrics.execution_time); //! ``` use std::collections::HashMap; use std::sync::{Arc, Mutex}; use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; use serde::{Deserialize, Serialize}; use uuid::Uuid; use anyhow::{Result, Context}; use dashmap::DashMap; use parking_lot::RwLock; use tracing::{info, warn, error, debug}; /// Core telemetry data structures and collection logic /// Following strict TDD - tests written first, real implementations only /// Represents different types of metrics we can collect #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] pub enum MetricType { /// Execution time metrics ExecutionTime { duration: Duration }, /// Memory usage metrics in bytes MemoryUsage { bytes: u64 }, /// GPU utilization percentage (0.0 - 100.0) GpuUtilization { percentage: f64 }, /// Bandwidth measurements in bytes/second Bandwidth { bytes_per_second: u64 }, /// Cache hit rate percentage (0.0 - 100.0) CacheHitRate { percentage: f64 }, /// Operation throughput (operations per second) Throughput { ops_per_second: f64 }, /// Error count ErrorCount { count: u64 }, } /// A single metric data point with timestamp and metadata #[derive(Debug, Clone, Serialize, Deserialize)] pub struct MetricPoint { pub id: Uuid, pub timestamp: SystemTime, pub operation_id: String, pub metric_type: MetricType, pub labels: HashMap, } /// Aggregated performance metrics for an operation #[derive(Debug, Clone, Default, Serialize, Deserialize)] pub struct PerformanceMetrics { pub operation_id: String, pub execution_time: Option, pub memory_peak: u64, pub memory_average: u64, pub gpu_utilization_peak: f64, pub gpu_utilization_average: f64, pub bandwidth_peak: u64, pub bandwidth_average: u64, pub cache_hit_rate: f64, pub throughput: f64, pub error_count: u64, pub sample_count: u64, } /// Detected bottleneck information #[derive(Debug, Clone, Serialize, Deserialize)] pub struct Bottleneck { pub id: Uuid, pub bottleneck_type: BottleneckType, pub operation_id: String, pub severity: BottleneckSeverity, pub description: String, pub detected_at: SystemTime, pub metrics: Vec, } /// Types of bottlenecks we can detect #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] pub enum BottleneckType { /// CPU computation is the limiting factor Compute, /// Memory bandwidth or capacity is limiting Memory, /// GPU utilization is limiting Gpu, /// I/O bandwidth is limiting Bandwidth, /// Cache misses are causing performance issues Cache, } /// Severity levels for bottlenecks #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] pub enum BottleneckSeverity { Low, Medium, High, Critical, } /// Represents an ongoing operation being tracked #[derive(Debug)] struct TrackedOperation { operation_id: String, start_time: Instant, metrics: Vec, labels: HashMap, } /// Main telemetry collector that orchestrates metric collection #[derive(Debug)] pub struct TelemetryCollector { active_operations: Arc>, completed_metrics: Arc>>, aggregated_metrics: Arc>>, bottlenecks: Arc>>, config: TelemetryConfig, } /// Configuration for telemetry collection #[derive(Debug, Clone)] pub struct TelemetryConfig { pub enable_gpu_monitoring: bool, pub enable_memory_tracking: bool, pub enable_bandwidth_monitoring: bool, pub bottleneck_detection_enabled: bool, pub metrics_retention_duration: Duration, pub sampling_interval: Duration, pub export_prometheus: bool, pub export_opentelemetry: bool, } impl Default for TelemetryConfig { fn default() -> Self { Self { enable_gpu_monitoring: true, enable_memory_tracking: true, enable_bandwidth_monitoring: true, bottleneck_detection_enabled: true, metrics_retention_duration: Duration::from_secs(3600), // 1 hour sampling_interval: Duration::from_millis(100), export_prometheus: false, export_opentelemetry: false, } } } /// Bottleneck detector that analyzes metrics for performance issues pub struct BottleneckDetector { thresholds: BottleneckThresholds, } /// Thresholds for detecting different types of bottlenecks #[derive(Debug, Clone)] pub struct BottleneckThresholds { pub high_execution_time_ms: u64, pub critical_execution_time_ms: u64, pub high_memory_usage_percent: f64, pub critical_memory_usage_percent: f64, pub low_gpu_utilization_percent: f64, pub high_bandwidth_usage_percent: f64, pub low_cache_hit_rate_percent: f64, } impl Default for BottleneckThresholds { fn default() -> Self { Self { high_execution_time_ms: 1000, critical_execution_time_ms: 5000, high_memory_usage_percent: 80.0, critical_memory_usage_percent: 95.0, low_gpu_utilization_percent: 20.0, high_bandwidth_usage_percent: 90.0, low_cache_hit_rate_percent: 50.0, } } } /// Real-time metrics analyzer for streaming data processing pub struct RealTimeAnalyzer { window_size: Duration, anomaly_threshold: f64, trend_detection_enabled: bool, } /// Prometheus metrics exporter pub struct PrometheusExporter { endpoint: String, metrics_registry: Arc>>, } /// OpenTelemetry exporter pub struct OpenTelemetryExporter { service_name: String, endpoint: String, } // Implementation follows - all functions are real implementations, no mocks impl TelemetryCollector { /// Creates a new telemetry collector with default configuration pub fn new() -> Self { Self::with_config(TelemetryConfig::default()) } /// Creates a new telemetry collector with custom configuration pub fn with_config(config: TelemetryConfig) -> Self { Self { active_operations: Arc::new(DashMap::new()), completed_metrics: Arc::new(RwLock::new(Vec::new())), aggregated_metrics: Arc::new(RwLock::new(HashMap::new())), bottlenecks: Arc::new(RwLock::new(Vec::new())), config, } } /// Starts tracking an operation pub fn start_operation(&self, operation_id: &str) -> Result<()> { self.start_operation_with_labels(operation_id, HashMap::new()) } /// Starts tracking an operation with custom labels pub fn start_operation_with_labels( &self, operation_id: &str, labels: HashMap ) -> Result<()> { let tracked_op = TrackedOperation { operation_id: operation_id.to_string(), start_time: Instant::now(), metrics: Vec::new(), labels, }; self.active_operations.insert(operation_id.to_string(), tracked_op); debug!("Started tracking operation: {}", operation_id); Ok(()) } /// Ends tracking an operation and records final metrics pub fn end_operation(&self, operation_id: &str) -> Result { let tracked_op = self.active_operations.remove(operation_id) .ok_or_else(|| anyhow::anyhow!("Operation not found: {}", operation_id))?; let execution_time = tracked_op.1.start_time.elapsed(); // Record execution time metric let execution_metric = MetricPoint { id: Uuid::new_v4(), timestamp: SystemTime::now(), operation_id: operation_id.to_string(), metric_type: MetricType::ExecutionTime { duration: execution_time }, labels: tracked_op.1.labels.clone(), }; // Store completed metrics (release lock immediately) { let mut completed = self.completed_metrics.write(); completed.push(execution_metric); completed.extend(tracked_op.1.metrics); } // Lock is released here // Calculate aggregated metrics (without holding any locks) let performance_metrics = self.calculate_performance_metrics(operation_id)?; // Store aggregated metrics { let mut aggregated = self.aggregated_metrics.write(); aggregated.insert(operation_id.to_string(), performance_metrics.clone()); } // Lock is released here debug!("Ended tracking operation: {} in {:?}", operation_id, execution_time); Ok(performance_metrics) } /// Records a metric point during an active operation pub fn record_metric(&self, operation_id: &str, metric_type: MetricType) -> Result<()> { self.record_metric_with_labels(operation_id, metric_type, HashMap::new()) } /// Records a metric point with custom labels during an active operation pub fn record_metric_with_labels( &self, operation_id: &str, metric_type: MetricType, labels: HashMap ) -> Result<()> { let metric_point = MetricPoint { id: Uuid::new_v4(), timestamp: SystemTime::now(), operation_id: operation_id.to_string(), metric_type, labels, }; if let Some(mut tracked_op) = self.active_operations.get_mut(operation_id) { tracked_op.metrics.push(metric_point); debug!("Recorded metric for operation: {}", operation_id); Ok(()) } else { anyhow::bail!("Operation not being tracked: {}", operation_id); } } /// Gets all metrics for a specific operation pub fn get_operation_metrics(&self, operation_id: &str) -> Result> { let completed = self.completed_metrics.read(); let metrics: Vec = completed .iter() .filter(|m| m.operation_id == operation_id) .cloned() .collect(); Ok(metrics) } /// Gets aggregated performance metrics for an operation pub fn get_performance_metrics(&self, operation_id: &str) -> Result { let aggregated = self.aggregated_metrics.read(); aggregated.get(operation_id) .cloned() .ok_or_else(|| anyhow::anyhow!("No metrics found for operation: {}", operation_id)) } /// Gets all completed metrics across all operations pub fn get_all_metrics(&self) -> Vec { self.completed_metrics.read().clone() } /// Gets all aggregated performance metrics pub fn get_all_performance_metrics(&self) -> HashMap { self.aggregated_metrics.read().clone() } /// Clears old metrics based on retention configuration pub fn cleanup_old_metrics(&self) -> Result { let retention_cutoff = SystemTime::now() - self.config.metrics_retention_duration; let mut completed = self.completed_metrics.write(); let original_count = completed.len(); completed.retain(|metric| metric.timestamp > retention_cutoff); let removed_count = original_count - completed.len(); if removed_count > 0 { info!("Cleaned up {} old metrics", removed_count); } Ok(removed_count) } /// Calculates aggregated performance metrics for an operation fn calculate_performance_metrics(&self, operation_id: &str) -> Result { let completed = self.completed_metrics.read(); let metrics: Vec<&MetricPoint> = completed .iter() .filter(|m| m.operation_id == operation_id) .collect(); if metrics.is_empty() { return Ok(PerformanceMetrics { operation_id: operation_id.to_string(), ..Default::default() }); } let mut performance = PerformanceMetrics { operation_id: operation_id.to_string(), sample_count: metrics.len() as u64, ..Default::default() }; let mut memory_values = Vec::new(); let mut gpu_util_values = Vec::new(); let mut bandwidth_values = Vec::new(); let mut cache_hit_rates = Vec::new(); let mut throughput_values = Vec::new(); for metric in &metrics { match &metric.metric_type { MetricType::ExecutionTime { duration } => { performance.execution_time = Some(*duration); } MetricType::MemoryUsage { bytes } => { memory_values.push(*bytes); performance.memory_peak = performance.memory_peak.max(*bytes); } MetricType::GpuUtilization { percentage } => { gpu_util_values.push(*percentage); performance.gpu_utilization_peak = performance.gpu_utilization_peak.max(*percentage); } MetricType::Bandwidth { bytes_per_second } => { bandwidth_values.push(*bytes_per_second); performance.bandwidth_peak = performance.bandwidth_peak.max(*bytes_per_second); } MetricType::CacheHitRate { percentage } => { cache_hit_rates.push(*percentage); } MetricType::Throughput { ops_per_second } => { throughput_values.push(*ops_per_second); } MetricType::ErrorCount { count } => { performance.error_count += count; } } } // Calculate averages if !memory_values.is_empty() { performance.memory_average = memory_values.iter().sum::() / memory_values.len() as u64; } if !gpu_util_values.is_empty() { performance.gpu_utilization_average = gpu_util_values.iter().sum::() / gpu_util_values.len() as f64; } if !bandwidth_values.is_empty() { performance.bandwidth_average = bandwidth_values.iter().sum::() / bandwidth_values.len() as u64; } if !cache_hit_rates.is_empty() { performance.cache_hit_rate = cache_hit_rates.iter().sum::() / cache_hit_rates.len() as f64; } if !throughput_values.is_empty() { performance.throughput = throughput_values.iter().sum::() / throughput_values.len() as f64; } Ok(performance) } } impl BottleneckDetector { /// Creates a new bottleneck detector with default thresholds pub fn new() -> Self { Self::with_thresholds(BottleneckThresholds::default()) } /// Creates a new bottleneck detector with custom thresholds pub fn with_thresholds(thresholds: BottleneckThresholds) -> Self { Self { thresholds } } /// Analyzes performance metrics to detect bottlenecks pub fn detect_bottlenecks(&self, metrics: &PerformanceMetrics) -> Vec { let mut bottlenecks = Vec::new(); let now = SystemTime::now(); // Check execution time bottlenecks if let Some(execution_time) = metrics.execution_time { let execution_ms = execution_time.as_millis() as u64; if execution_ms >= self.thresholds.critical_execution_time_ms { bottlenecks.push(Bottleneck { id: Uuid::new_v4(), bottleneck_type: BottleneckType::Compute, operation_id: metrics.operation_id.clone(), severity: BottleneckSeverity::Critical, description: format!( "Critical execution time: {}ms (threshold: {}ms)", execution_ms, self.thresholds.critical_execution_time_ms ), detected_at: now, metrics: Vec::new(), }); } else if execution_ms >= self.thresholds.high_execution_time_ms { bottlenecks.push(Bottleneck { id: Uuid::new_v4(), bottleneck_type: BottleneckType::Compute, operation_id: metrics.operation_id.clone(), severity: BottleneckSeverity::High, description: format!( "High execution time: {}ms (threshold: {}ms)", execution_ms, self.thresholds.high_execution_time_ms ), detected_at: now, metrics: Vec::new(), }); } } // Check GPU utilization bottlenecks if metrics.gpu_utilization_average < self.thresholds.low_gpu_utilization_percent { let severity = if metrics.gpu_utilization_average < self.thresholds.low_gpu_utilization_percent / 2.0 { BottleneckSeverity::High } else { BottleneckSeverity::Medium }; bottlenecks.push(Bottleneck { id: Uuid::new_v4(), bottleneck_type: BottleneckType::Gpu, operation_id: metrics.operation_id.clone(), severity, description: format!( "Low GPU utilization: {:.1}% (threshold: {:.1}%)", metrics.gpu_utilization_average, self.thresholds.low_gpu_utilization_percent ), detected_at: now, metrics: Vec::new(), }); } // Check cache hit rate bottlenecks if metrics.cache_hit_rate < self.thresholds.low_cache_hit_rate_percent { let severity = if metrics.cache_hit_rate < self.thresholds.low_cache_hit_rate_percent / 2.0 { BottleneckSeverity::High } else { BottleneckSeverity::Medium }; bottlenecks.push(Bottleneck { id: Uuid::new_v4(), bottleneck_type: BottleneckType::Cache, operation_id: metrics.operation_id.clone(), severity, description: format!( "Low cache hit rate: {:.1}% (threshold: {:.1}%)", metrics.cache_hit_rate, self.thresholds.low_cache_hit_rate_percent ), detected_at: now, metrics: Vec::new(), }); } bottlenecks } /// Updates bottleneck thresholds pub fn update_thresholds(&mut self, thresholds: BottleneckThresholds) { self.thresholds = thresholds; } /// Gets current bottleneck thresholds pub fn get_thresholds(&self) -> &BottleneckThresholds { &self.thresholds } } impl RealTimeAnalyzer { /// Creates a new real-time analyzer pub fn new() -> Self { Self { window_size: Duration::from_secs(60), // 1 minute window anomaly_threshold: 2.0, // 2 standard deviations trend_detection_enabled: true, } } /// Analyzes a stream of metrics for anomalies pub fn analyze_metrics_stream(&self, metrics: &[MetricPoint]) -> Result> { if metrics.len() < 2 { return Ok(Vec::new()); } let mut anomalies = Vec::new(); let now = SystemTime::now(); // Group metrics by type and operation let mut grouped_metrics: HashMap<(String, String), Vec> = HashMap::new(); for metric in metrics { let key = (metric.operation_id.clone(), format!("{:?}", metric.metric_type)); let value = self.extract_numeric_value(&metric.metric_type); grouped_metrics.entry(key).or_insert_with(Vec::new).push(value); } // Detect anomalies in each group for ((operation_id, metric_type_str), values) in grouped_metrics { if let Some(anomaly) = self.detect_anomaly(&values) { anomalies.push(Bottleneck { id: Uuid::new_v4(), bottleneck_type: self.infer_bottleneck_type(&metric_type_str), operation_id, severity: BottleneckSeverity::Medium, description: format!( "Anomaly detected in {}: value {:.2} is {:.2} std devs from mean", metric_type_str, anomaly.value, anomaly.std_devs ), detected_at: now, metrics: Vec::new(), }); } } Ok(anomalies) } /// Detects performance trends in metrics pub fn detect_performance_trends(&self, metrics: &[MetricPoint]) -> Result> { if !self.trend_detection_enabled || metrics.len() < 10 { return Ok(Vec::new()); } let mut trends = Vec::new(); // Group metrics by operation and type let mut grouped: HashMap<(String, String), Vec<(SystemTime, f64)>> = HashMap::new(); for metric in metrics { let key = (metric.operation_id.clone(), format!("{:?}", metric.metric_type)); let value = self.extract_numeric_value(&metric.metric_type); grouped.entry(key).or_insert_with(Vec::new).push((metric.timestamp, value)); } // Analyze trends in each group for ((operation_id, metric_type), values) in grouped { if let Some(trend) = self.calculate_trend(&values) { if trend.slope.abs() > 0.1 { // Significant trend threshold let trend_direction = if trend.slope > 0.0 { "increasing" } else { "decreasing" }; trends.push(format!( "Performance trend detected in {}: {} {} at rate {:.3}/sec", operation_id, metric_type, trend_direction, trend.slope )); } } } Ok(trends) } /// Extracts numeric value from metric type for analysis fn extract_numeric_value(&self, metric_type: &MetricType) -> f64 { match metric_type { MetricType::ExecutionTime { duration } => duration.as_secs_f64(), MetricType::MemoryUsage { bytes } => *bytes as f64, MetricType::GpuUtilization { percentage } => *percentage, MetricType::Bandwidth { bytes_per_second } => *bytes_per_second as f64, MetricType::CacheHitRate { percentage } => *percentage, MetricType::Throughput { ops_per_second } => *ops_per_second, MetricType::ErrorCount { count } => *count as f64, } } /// Detects anomalies in a series of values fn detect_anomaly(&self, values: &[f64]) -> Option { if values.len() < 3 { return None; } let mean = values.iter().sum::() / values.len() as f64; let variance = values.iter() .map(|x| (x - mean).powi(2)) .sum::() / values.len() as f64; let std_dev = variance.sqrt(); if std_dev < 1e-10 { // Avoid division by zero return None; } // Check the most recent value let last_value = values[values.len() - 1]; let z_score = (last_value - mean) / std_dev; if z_score.abs() > self.anomaly_threshold { Some(AnomalyInfo { value: last_value, std_devs: z_score, }) } else { None } } /// Calculates trend information from time series data fn calculate_trend(&self, values: &[(SystemTime, f64)]) -> Option { if values.len() < 2 { return None; } // Convert timestamps to seconds since first measurement let first_time = values[0].0; let time_values: Vec = values.iter() .map(|(t, _)| t.duration_since(first_time).unwrap_or_default().as_secs_f64()) .collect(); let metric_values: Vec = values.iter().map(|(_, v)| *v).collect(); // Simple linear regression to calculate slope let n = time_values.len() as f64; let sum_x: f64 = time_values.iter().sum(); let sum_y: f64 = metric_values.iter().sum(); let sum_xy: f64 = time_values.iter().zip(&metric_values).map(|(x, y)| x * y).sum(); let sum_x2: f64 = time_values.iter().map(|x| x * x).sum(); let denominator = n * sum_x2 - sum_x * sum_x; if denominator.abs() < 1e-10 { return None; } let slope = (n * sum_xy - sum_x * sum_y) / denominator; let intercept = (sum_y - slope * sum_x) / n; Some(TrendInfo { slope, intercept }) } /// Infers bottleneck type from metric type string fn infer_bottleneck_type(&self, metric_type_str: &str) -> BottleneckType { if metric_type_str.contains("Memory") { BottleneckType::Memory } else if metric_type_str.contains("Gpu") { BottleneckType::Gpu } else if metric_type_str.contains("Bandwidth") { BottleneckType::Bandwidth } else if metric_type_str.contains("Cache") { BottleneckType::Cache } else { BottleneckType::Compute } } } /// Helper struct for anomaly detection #[derive(Debug)] struct AnomalyInfo { value: f64, std_devs: f64, } /// Helper struct for trend analysis #[derive(Debug)] struct TrendInfo { slope: f64, intercept: f64, } impl PrometheusExporter { /// Creates a new Prometheus exporter pub fn new(endpoint: String) -> Self { Self { endpoint, metrics_registry: Arc::new(Mutex::new(HashMap::new())), } } /// Exports metrics in Prometheus format pub fn export_metrics(&self, metrics: &[MetricPoint]) -> Result { let mut registry = self.metrics_registry.lock().unwrap(); registry.clear(); // Convert metrics to Prometheus format for metric in metrics { let metric_name = self.format_metric_name(&metric.metric_type); let metric_value = self.extract_metric_value(&metric.metric_type); let key = format!("{}_{}", metric_name, metric.operation_id); registry.insert(key, metric_value); } // Generate Prometheus exposition format let mut output = String::new(); for (name, value) in registry.iter() { output.push_str(&format!("# TYPE {} gauge\n", name)); output.push_str(&format!("{} {}\n", name, value)); } Ok(output) } /// Gets the Prometheus endpoint URL pub fn endpoint(&self) -> &str { &self.endpoint } /// Formats metric type into Prometheus metric name fn format_metric_name(&self, metric_type: &MetricType) -> String { match metric_type { MetricType::ExecutionTime { .. } => "execution_time_seconds".to_string(), MetricType::MemoryUsage { .. } => "memory_usage_bytes".to_string(), MetricType::GpuUtilization { .. } => "gpu_utilization_percent".to_string(), MetricType::Bandwidth { .. } => "bandwidth_bytes_per_second".to_string(), MetricType::CacheHitRate { .. } => "cache_hit_rate_percent".to_string(), MetricType::Throughput { .. } => "throughput_ops_per_second".to_string(), MetricType::ErrorCount { .. } => "error_count_total".to_string(), } } /// Extracts numeric value from metric type fn extract_metric_value(&self, metric_type: &MetricType) -> f64 { match metric_type { MetricType::ExecutionTime { duration } => duration.as_secs_f64(), MetricType::MemoryUsage { bytes } => *bytes as f64, MetricType::GpuUtilization { percentage } => *percentage, MetricType::Bandwidth { bytes_per_second } => *bytes_per_second as f64, MetricType::CacheHitRate { percentage } => *percentage, MetricType::Throughput { ops_per_second } => *ops_per_second, MetricType::ErrorCount { count } => *count as f64, } } } impl OpenTelemetryExporter { /// Creates a new OpenTelemetry exporter pub fn new(service_name: String, endpoint: String) -> Self { Self { service_name, endpoint } } /// Exports metrics in OpenTelemetry format pub fn export_metrics(&self, metrics: &[MetricPoint]) -> Result { // This is a simplified OpenTelemetry export - in production would use the official SDK let mut traces = Vec::new(); for metric in metrics { let trace_data = serde_json::json!({ "resourceSpans": [{ "resource": { "attributes": [{ "key": "service.name", "value": { "stringValue": self.service_name } }] }, "scopeSpans": [{ "spans": [{ "traceId": format!("{:032x}", metric.id.as_u128()), "spanId": format!("{:016x}", metric.id.as_u128() & 0xFFFFFFFFFFFFFFFF), "name": metric.operation_id, "startTimeUnixNano": metric.timestamp .duration_since(UNIX_EPOCH) .unwrap_or_default() .as_nanos(), "endTimeUnixNano": metric.timestamp .duration_since(UNIX_EPOCH) .unwrap_or_default() .as_nanos(), "attributes": self.format_metric_attributes(&metric.metric_type) }] }] }] }); traces.push(trace_data); } Ok(serde_json::to_string_pretty(&traces)?) } /// Gets the OpenTelemetry endpoint URL pub fn endpoint(&self) -> &str { &self.endpoint } /// Gets the service name pub fn service_name(&self) -> &str { &self.service_name } /// Formats metric type into OpenTelemetry attributes fn format_metric_attributes(&self, metric_type: &MetricType) -> Vec { match metric_type { MetricType::ExecutionTime { duration } => vec![ serde_json::json!({ "key": "duration_ms", "value": { "intValue": duration.as_millis() as i64 } }) ], MetricType::MemoryUsage { bytes } => vec![ serde_json::json!({ "key": "memory_bytes", "value": { "intValue": *bytes as i64 } }) ], MetricType::GpuUtilization { percentage } => vec![ serde_json::json!({ "key": "gpu_utilization_percent", "value": { "doubleValue": *percentage } }) ], MetricType::Bandwidth { bytes_per_second } => vec![ serde_json::json!({ "key": "bandwidth_bps", "value": { "intValue": *bytes_per_second as i64 } }) ], MetricType::CacheHitRate { percentage } => vec![ serde_json::json!({ "key": "cache_hit_rate_percent", "value": { "doubleValue": *percentage } }) ], MetricType::Throughput { ops_per_second } => vec![ serde_json::json!({ "key": "throughput_ops", "value": { "doubleValue": *ops_per_second } }) ], MetricType::ErrorCount { count } => vec![ serde_json::json!({ "key": "error_count", "value": { "intValue": *count as i64 } }) ], } } } impl Default for TelemetryCollector { fn default() -> Self { Self::new() } } impl Default for BottleneckDetector { fn default() -> Self { Self::new() } } impl Default for RealTimeAnalyzer { fn default() -> Self { Self::new() } } // Tests follow strict TDD approach - written first, then implementation #[cfg(test)] mod tests { use super::*; use std::thread; #[test] fn test_telemetry_collector_creation() { let collector = TelemetryCollector::new(); assert_eq!(collector.active_operations.len(), 0); assert_eq!(collector.get_all_metrics().len(), 0); } #[test] fn test_start_and_end_operation() { let collector = TelemetryCollector::new(); // Start operation collector.start_operation("test_op").unwrap(); assert_eq!(collector.active_operations.len(), 1); // End operation (no sleep needed for basic functionality test) let metrics = collector.end_operation("test_op").unwrap(); assert_eq!(collector.active_operations.len(), 0); assert!(metrics.execution_time.is_some()); // Just verify execution time is positive, don't check specific duration assert!(metrics.execution_time.unwrap().as_nanos() > 0); assert_eq!(metrics.operation_id, "test_op"); } #[test] fn test_record_metric_during_operation() { let collector = TelemetryCollector::new(); collector.start_operation("test_op").unwrap(); // Record various metrics collector.record_metric("test_op", MetricType::MemoryUsage { bytes: 1024 }).unwrap(); collector.record_metric("test_op", MetricType::GpuUtilization { percentage: 85.5 }).unwrap(); collector.record_metric("test_op", MetricType::Bandwidth { bytes_per_second: 1_000_000 }).unwrap(); let performance = collector.end_operation("test_op").unwrap(); // Verify metrics were recorded assert_eq!(performance.memory_peak, 1024); assert_eq!(performance.memory_average, 1024); assert_eq!(performance.gpu_utilization_peak, 85.5); assert_eq!(performance.gpu_utilization_average, 85.5); assert_eq!(performance.bandwidth_peak, 1_000_000); assert_eq!(performance.bandwidth_average, 1_000_000); assert_eq!(performance.sample_count, 4); // 3 recorded + 1 execution time } #[test] fn test_operation_not_found_error() { let collector = TelemetryCollector::new(); // Try to end operation that was never started let result = collector.end_operation("nonexistent_op"); assert!(result.is_err()); assert!(result.unwrap_err().to_string().contains("Operation not found")); // Try to record metric for nonexistent operation let result = collector.record_metric("nonexistent_op", MetricType::MemoryUsage { bytes: 100 }); assert!(result.is_err()); assert!(result.unwrap_err().to_string().contains("Operation not being tracked")); } #[test] fn test_multiple_operations_tracking() { let collector = TelemetryCollector::new(); // Start multiple operations collector.start_operation("op1").unwrap(); collector.start_operation("op2").unwrap(); collector.start_operation("op3").unwrap(); assert_eq!(collector.active_operations.len(), 3); // Record different metrics for each collector.record_metric("op1", MetricType::MemoryUsage { bytes: 1000 }).unwrap(); collector.record_metric("op2", MetricType::MemoryUsage { bytes: 2000 }).unwrap(); collector.record_metric("op3", MetricType::MemoryUsage { bytes: 3000 }).unwrap(); // End operations let metrics1 = collector.end_operation("op1").unwrap(); let metrics2 = collector.end_operation("op2").unwrap(); let metrics3 = collector.end_operation("op3").unwrap(); assert_eq!(collector.active_operations.len(), 0); assert_eq!(metrics1.memory_peak, 1000); assert_eq!(metrics2.memory_peak, 2000); assert_eq!(metrics3.memory_peak, 3000); } #[test] fn test_metric_aggregation() { let collector = TelemetryCollector::new(); collector.start_operation("agg_test").unwrap(); // Record multiple memory usage metrics collector.record_metric("agg_test", MetricType::MemoryUsage { bytes: 1000 }).unwrap(); collector.record_metric("agg_test", MetricType::MemoryUsage { bytes: 1500 }).unwrap(); collector.record_metric("agg_test", MetricType::MemoryUsage { bytes: 800 }).unwrap(); // Record multiple GPU utilization metrics collector.record_metric("agg_test", MetricType::GpuUtilization { percentage: 70.0 }).unwrap(); collector.record_metric("agg_test", MetricType::GpuUtilization { percentage: 90.0 }).unwrap(); collector.record_metric("agg_test", MetricType::GpuUtilization { percentage: 60.0 }).unwrap(); let performance = collector.end_operation("agg_test").unwrap(); // Verify aggregation assert_eq!(performance.memory_peak, 1500); assert_eq!(performance.memory_average, 1100); // (1000 + 1500 + 800) / 3 assert_eq!(performance.gpu_utilization_peak, 90.0); assert_eq!(performance.gpu_utilization_average, 73.33333333333333); // (70 + 90 + 60) / 3 } #[test] fn test_bottleneck_detector_creation() { let detector = BottleneckDetector::new(); let thresholds = detector.get_thresholds(); assert_eq!(thresholds.high_execution_time_ms, 1000); assert_eq!(thresholds.critical_execution_time_ms, 5000); } #[test] fn test_compute_bottleneck_detection() { let detector = BottleneckDetector::new(); // Create metrics with high execution time but normal other values let metrics = PerformanceMetrics { operation_id: "slow_op".to_string(), execution_time: Some(Duration::from_millis(2000)), // Above high threshold gpu_utilization_average: 75.0, // Above low threshold cache_hit_rate: 90.0, // Above low threshold ..Default::default() }; let bottlenecks = detector.detect_bottlenecks(&metrics); assert_eq!(bottlenecks.len(), 1); assert_eq!(bottlenecks[0].bottleneck_type, BottleneckType::Compute); assert_eq!(bottlenecks[0].severity, BottleneckSeverity::High); assert!(bottlenecks[0].description.contains("High execution time")); } #[test] fn test_critical_compute_bottleneck_detection() { let detector = BottleneckDetector::new(); // Create metrics with critical execution time but normal other values let metrics = PerformanceMetrics { operation_id: "critical_slow_op".to_string(), execution_time: Some(Duration::from_millis(6000)), // Above critical threshold gpu_utilization_average: 75.0, // Above low threshold cache_hit_rate: 90.0, // Above low threshold ..Default::default() }; let bottlenecks = detector.detect_bottlenecks(&metrics); assert_eq!(bottlenecks.len(), 1); assert_eq!(bottlenecks[0].bottleneck_type, BottleneckType::Compute); assert_eq!(bottlenecks[0].severity, BottleneckSeverity::Critical); assert!(bottlenecks[0].description.contains("Critical execution time")); } #[test] fn test_gpu_utilization_bottleneck_detection() { let detector = BottleneckDetector::new(); // Create metrics with low GPU utilization but normal other values let metrics = PerformanceMetrics { operation_id: "low_gpu_op".to_string(), execution_time: Some(Duration::from_millis(500)), // Below high threshold gpu_utilization_average: 15.0, // Below threshold of 20% cache_hit_rate: 90.0, // Above low threshold ..Default::default() }; let bottlenecks = detector.detect_bottlenecks(&metrics); assert_eq!(bottlenecks.len(), 1); assert_eq!(bottlenecks[0].bottleneck_type, BottleneckType::Gpu); assert_eq!(bottlenecks[0].severity, BottleneckSeverity::Medium); assert!(bottlenecks[0].description.contains("Low GPU utilization")); } #[test] fn test_cache_hit_rate_bottleneck_detection() { let detector = BottleneckDetector::new(); // Create metrics with low cache hit rate but normal other values let metrics = PerformanceMetrics { operation_id: "low_cache_op".to_string(), execution_time: Some(Duration::from_millis(500)), // Below high threshold gpu_utilization_average: 75.0, // Above low threshold cache_hit_rate: 30.0, // Below threshold of 50% ..Default::default() }; let bottlenecks = detector.detect_bottlenecks(&metrics); assert_eq!(bottlenecks.len(), 1); assert_eq!(bottlenecks[0].bottleneck_type, BottleneckType::Cache); assert_eq!(bottlenecks[0].severity, BottleneckSeverity::Medium); assert!(bottlenecks[0].description.contains("Low cache hit rate")); } #[test] fn test_no_bottlenecks_detected() { let detector = BottleneckDetector::new(); // Create metrics within normal ranges let metrics = PerformanceMetrics { operation_id: "normal_op".to_string(), execution_time: Some(Duration::from_millis(500)), // Below thresholds gpu_utilization_average: 75.0, // Above threshold cache_hit_rate: 90.0, // Above threshold ..Default::default() }; let bottlenecks = detector.detect_bottlenecks(&metrics); assert_eq!(bottlenecks.len(), 0); } #[test] fn test_real_time_analyzer_creation() { let analyzer = RealTimeAnalyzer::new(); assert!(analyzer.trend_detection_enabled); assert_eq!(analyzer.window_size, Duration::from_secs(60)); assert_eq!(analyzer.anomaly_threshold, 2.0); } #[test] fn test_anomaly_detection_insufficient_data() { let analyzer = RealTimeAnalyzer::new(); // Create insufficient data points let metrics = vec![ MetricPoint { id: Uuid::new_v4(), timestamp: SystemTime::now(), operation_id: "test_op".to_string(), metric_type: MetricType::ExecutionTime { duration: Duration::from_millis(100) }, labels: HashMap::new(), } ]; let anomalies = analyzer.analyze_metrics_stream(&metrics).unwrap(); assert_eq!(anomalies.len(), 0); } #[test] fn test_anomaly_detection_normal_data() { let analyzer = RealTimeAnalyzer::new(); // Create normal data points (similar values) let metrics = vec![ MetricPoint { id: Uuid::new_v4(), timestamp: SystemTime::now(), operation_id: "test_op".to_string(), metric_type: MetricType::ExecutionTime { duration: Duration::from_millis(100) }, labels: HashMap::new(), }, MetricPoint { id: Uuid::new_v4(), timestamp: SystemTime::now(), operation_id: "test_op".to_string(), metric_type: MetricType::ExecutionTime { duration: Duration::from_millis(105) }, labels: HashMap::new(), }, MetricPoint { id: Uuid::new_v4(), timestamp: SystemTime::now(), operation_id: "test_op".to_string(), metric_type: MetricType::ExecutionTime { duration: Duration::from_millis(95) }, labels: HashMap::new(), } ]; let anomalies = analyzer.analyze_metrics_stream(&metrics).unwrap(); assert_eq!(anomalies.len(), 0); } #[test] fn test_prometheus_exporter_creation() { let exporter = PrometheusExporter::new("http://localhost:9090".to_string()); assert_eq!(exporter.endpoint(), "http://localhost:9090"); } #[test] fn test_prometheus_export_format() { let exporter = PrometheusExporter::new("http://localhost:9090".to_string()); let metrics = vec![ MetricPoint { id: Uuid::new_v4(), timestamp: SystemTime::now(), operation_id: "test_op".to_string(), metric_type: MetricType::ExecutionTime { duration: Duration::from_millis(123) }, labels: HashMap::new(), }, MetricPoint { id: Uuid::new_v4(), timestamp: SystemTime::now(), operation_id: "test_op".to_string(), metric_type: MetricType::MemoryUsage { bytes: 1024 }, labels: HashMap::new(), } ]; let exported = exporter.export_metrics(&metrics).unwrap(); assert!(exported.contains("# TYPE execution_time_seconds_test_op gauge")); assert!(exported.contains("execution_time_seconds_test_op 0.123")); assert!(exported.contains("# TYPE memory_usage_bytes_test_op gauge")); assert!(exported.contains("memory_usage_bytes_test_op 1024")); } #[test] fn test_opentelemetry_exporter_creation() { let exporter = OpenTelemetryExporter::new( "rustytorch-graph".to_string(), "http://localhost:4317".to_string() ); assert_eq!(exporter.service_name(), "rustytorch-graph"); assert_eq!(exporter.endpoint(), "http://localhost:4317"); } #[test] fn test_opentelemetry_export_format() { let exporter = OpenTelemetryExporter::new( "rustytorch-graph".to_string(), "http://localhost:4317".to_string() ); let metrics = vec![ MetricPoint { id: Uuid::new_v4(), timestamp: SystemTime::now(), operation_id: "test_op".to_string(), metric_type: MetricType::ExecutionTime { duration: Duration::from_millis(123) }, labels: HashMap::new(), } ]; let exported = exporter.export_metrics(&metrics).unwrap(); assert!(exported.contains("resourceSpans")); assert!(exported.contains("service.name")); assert!(exported.contains("rustytorch-graph")); assert!(exported.contains("test_op")); assert!(exported.contains("duration_ms")); } #[test] fn test_metrics_cleanup() { let mut config = TelemetryConfig::default(); config.metrics_retention_duration = Duration::from_nanos(1); // Immediate expiration let collector = TelemetryCollector::with_config(config); // Start and end operation to create metrics collector.start_operation("cleanup_test").unwrap(); collector.record_metric("cleanup_test", MetricType::MemoryUsage { bytes: 1024 }).unwrap(); collector.end_operation("cleanup_test").unwrap(); // Verify metrics exist assert_eq!(collector.get_all_metrics().len(), 2); // 1 memory + 1 execution time // Cleanup should remove all metrics due to immediate expiration let removed_count = collector.cleanup_old_metrics().unwrap(); assert_eq!(removed_count, 2); assert_eq!(collector.get_all_metrics().len(), 0); } #[test] fn test_telemetry_with_labels() { let collector = TelemetryCollector::new(); let mut labels = HashMap::new(); labels.insert("device".to_string(), "gpu0".to_string()); labels.insert("model".to_string(), "llama2".to_string()); collector.start_operation_with_labels("labeled_op", labels.clone()).unwrap(); collector.record_metric_with_labels( "labeled_op", MetricType::GpuUtilization { percentage: 95.0 }, labels.clone() ).unwrap(); let performance = collector.end_operation("labeled_op").unwrap(); assert_eq!(performance.gpu_utilization_peak, 95.0); // Verify labels are preserved in metrics let all_metrics = collector.get_all_metrics(); let gpu_metric = all_metrics.iter() .find(|m| matches!(m.metric_type, MetricType::GpuUtilization { .. })) .unwrap(); assert_eq!(gpu_metric.labels.get("device").unwrap(), "gpu0"); assert_eq!(gpu_metric.labels.get("model").unwrap(), "llama2"); } #[test] fn test_threshold_updates() { let mut detector = BottleneckDetector::new(); let new_thresholds = BottleneckThresholds { high_execution_time_ms: 500, critical_execution_time_ms: 2000, ..Default::default() }; detector.update_thresholds(new_thresholds.clone()); let current_thresholds = detector.get_thresholds(); assert_eq!(current_thresholds.high_execution_time_ms, 500); assert_eq!(current_thresholds.critical_execution_time_ms, 2000); } }