1347 lines
66 KiB
Rust
1347 lines
66 KiB
Rust
//! # 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<String, String>,
|
|
}
|
|
|
|
/// Aggregated performance metrics for an operation
|
|
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
|
|
pub struct PerformanceMetrics {
|
|
pub operation_id: String,
|
|
pub execution_time: Option<Duration>,
|
|
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<MetricPoint>,
|
|
}
|
|
|
|
/// 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<MetricPoint>,
|
|
labels: HashMap<String, String>,
|
|
}
|
|
|
|
/// Main telemetry collector that orchestrates metric collection
|
|
#[derive(Debug)]
|
|
pub struct TelemetryCollector {
|
|
active_operations: Arc<DashMap<String, TrackedOperation>>,
|
|
completed_metrics: Arc<RwLock<Vec<MetricPoint>>>,
|
|
aggregated_metrics: Arc<RwLock<HashMap<String, PerformanceMetrics>>>,
|
|
bottlenecks: Arc<RwLock<Vec<Bottleneck>>>,
|
|
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<Mutex<HashMap<String, f64>>>,
|
|
}
|
|
|
|
/// 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<String, String>
|
|
) -> 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<PerformanceMetrics> {
|
|
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<String, String>
|
|
) -> 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<Vec<MetricPoint>> {
|
|
let completed = self.completed_metrics.read();
|
|
let metrics: Vec<MetricPoint> = 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<PerformanceMetrics> {
|
|
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<MetricPoint> {
|
|
self.completed_metrics.read().clone()
|
|
}
|
|
|
|
/// Gets all aggregated performance metrics
|
|
pub fn get_all_performance_metrics(&self) -> HashMap<String, PerformanceMetrics> {
|
|
self.aggregated_metrics.read().clone()
|
|
}
|
|
|
|
/// Clears old metrics based on retention configuration
|
|
pub fn cleanup_old_metrics(&self) -> Result<usize> {
|
|
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<PerformanceMetrics> {
|
|
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::<u64>() / memory_values.len() as u64;
|
|
}
|
|
if !gpu_util_values.is_empty() {
|
|
performance.gpu_utilization_average = gpu_util_values.iter().sum::<f64>() / gpu_util_values.len() as f64;
|
|
}
|
|
if !bandwidth_values.is_empty() {
|
|
performance.bandwidth_average = bandwidth_values.iter().sum::<u64>() / bandwidth_values.len() as u64;
|
|
}
|
|
if !cache_hit_rates.is_empty() {
|
|
performance.cache_hit_rate = cache_hit_rates.iter().sum::<f64>() / cache_hit_rates.len() as f64;
|
|
}
|
|
if !throughput_values.is_empty() {
|
|
performance.throughput = throughput_values.iter().sum::<f64>() / 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<Bottleneck> {
|
|
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<Vec<Bottleneck>> {
|
|
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<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_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<Vec<String>> {
|
|
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<AnomalyInfo> {
|
|
if values.len() < 3 {
|
|
return None;
|
|
}
|
|
|
|
let mean = values.iter().sum::<f64>() / values.len() as f64;
|
|
let variance = values.iter()
|
|
.map(|x| (x - mean).powi(2))
|
|
.sum::<f64>() / 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<TrendInfo> {
|
|
if values.len() < 2 {
|
|
return None;
|
|
}
|
|
|
|
// Convert timestamps to seconds since first measurement
|
|
let first_time = values[0].0;
|
|
let time_values: Vec<f64> = values.iter()
|
|
.map(|(t, _)| t.duration_since(first_time).unwrap_or_default().as_secs_f64())
|
|
.collect();
|
|
let metric_values: Vec<f64> = 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<String> {
|
|
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<String> {
|
|
// 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<serde_json::Value> {
|
|
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);
|
|
}
|
|
}
|