use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
use std::sync::Arc;
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use tokio::sync::RwLock;
#[derive(Debug, Clone)]
pub struct MetricsCollector {
request_counts: Arc<RwLock<HashMap<String, AtomicU64>>>,
response_times: Arc<RwLock<HashMap<String, Vec<Duration>>>>,
error_counts: Arc<RwLock<HashMap<String, AtomicU64>>>,
resource_usage: Arc<ResourceUsageTracker>,
connection_metrics: Arc<ConnectionMetrics>,
health_status: Arc<RwLock<HealthStatus>>,
}
#[derive(Debug)]
pub struct ResourceUsageTracker {
memory_usage_mb: AtomicUsize,
peak_memory_mb: AtomicUsize,
cpu_samples: RwLock<Vec<f64>>,
active_requests: AtomicUsize,
peak_concurrent_requests: AtomicUsize,
}
#[derive(Debug)]
pub struct ConnectionMetrics {
total_connections: AtomicU64,
active_connections: AtomicUsize,
peak_connections: AtomicUsize,
connection_errors: AtomicU64,
connection_durations: RwLock<Vec<Duration>>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct HealthStatus {
pub health_score: u8,
pub components: HashMap<String, ComponentHealth>,
pub last_check: SystemTime,
pub alerts: Vec<HealthAlert>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ComponentHealth {
pub status: ComponentStatus,
pub message: Option<String>,
pub last_update: SystemTime,
pub metrics: HashMap<String, f64>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum ComponentStatus {
Healthy,
Warning,
Degraded,
Critical,
Unavailable,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct HealthAlert {
pub severity: AlertSeverity,
pub message: String,
pub component: String,
pub timestamp: SystemTime,
pub id: String,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum AlertSeverity {
Info,
Warning,
Error,
Critical,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct MetricsSnapshot {
pub timestamp: SystemTime,
pub requests: RequestMetrics,
pub resources: ResourceMetrics,
pub connections: ConnectionMetricsSnapshot,
pub health: HealthStatus,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RequestMetrics {
pub total_requests: u64,
pub requests_per_second: f64,
pub avg_response_time_ms: f64,
pub p95_response_time_ms: f64,
pub error_rate: f64,
pub top_request_types: Vec<(String, u64)>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ResourceMetrics {
pub current_memory_mb: usize,
pub peak_memory_mb: usize,
pub avg_cpu_usage: f64,
pub active_requests: usize,
pub peak_concurrent_requests: usize,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ConnectionMetricsSnapshot {
pub total_connections: u64,
pub active_connections: usize,
pub peak_connections: usize,
pub connection_error_rate: f64,
pub avg_connection_duration_ms: f64,
}
impl MetricsCollector {
pub fn new() -> Self {
Self {
request_counts: Arc::new(RwLock::new(HashMap::new())),
response_times: Arc::new(RwLock::new(HashMap::new())),
error_counts: Arc::new(RwLock::new(HashMap::new())),
resource_usage: Arc::new(ResourceUsageTracker::new()),
connection_metrics: Arc::new(ConnectionMetrics::new()),
health_status: Arc::new(RwLock::new(HealthStatus::new())),
}
}
pub async fn record_request(&self, method: &str, response_time: Duration) {
{
let mut counts = self.request_counts.write().await;
let counter = counts
.entry(method.to_string())
.or_insert_with(|| AtomicU64::new(0));
counter.fetch_add(1, Ordering::Relaxed);
}
{
let mut times = self.response_times.write().await;
let method_times = times.entry(method.to_string()).or_insert_with(Vec::new);
method_times.push(response_time);
if method_times.len() > 1000 {
method_times.drain(..method_times.len() - 1000);
}
}
}
pub async fn record_error(&self, error_type: &str) {
let mut counts = self.error_counts.write().await;
let counter = counts
.entry(error_type.to_string())
.or_insert_with(|| AtomicU64::new(0));
counter.fetch_add(1, Ordering::Relaxed);
}
pub fn update_memory_usage(&self, current_mb: usize) {
self.resource_usage
.memory_usage_mb
.store(current_mb, Ordering::Relaxed);
let current_peak = self.resource_usage.peak_memory_mb.load(Ordering::Relaxed);
if current_mb > current_peak {
self.resource_usage
.peak_memory_mb
.store(current_mb, Ordering::Relaxed);
}
}
pub async fn record_cpu_usage(&self, cpu_percent: f64) {
let mut samples = self.resource_usage.cpu_samples.write().await;
samples.push(cpu_percent);
let len = samples.len();
if len > 100 {
samples.drain(..len - 100);
}
}
pub fn start_request(&self) {
let current = self
.resource_usage
.active_requests
.fetch_add(1, Ordering::Relaxed)
+ 1;
let current_peak = self
.resource_usage
.peak_concurrent_requests
.load(Ordering::Relaxed);
if current > current_peak {
self.resource_usage
.peak_concurrent_requests
.store(current, Ordering::Relaxed);
}
}
pub fn complete_request(&self) {
self.resource_usage
.active_requests
.fetch_sub(1, Ordering::Relaxed);
}
pub fn record_connection(&self) {
self.connection_metrics
.total_connections
.fetch_add(1, Ordering::Relaxed);
let current = self
.connection_metrics
.active_connections
.fetch_add(1, Ordering::Relaxed)
+ 1;
let current_peak = self
.connection_metrics
.peak_connections
.load(Ordering::Relaxed);
if current > current_peak {
self.connection_metrics
.peak_connections
.store(current, Ordering::Relaxed);
}
}
pub fn record_connection_close(&self, duration: Duration) {
self.connection_metrics
.active_connections
.fetch_sub(1, Ordering::Relaxed);
let duration_holder = Arc::clone(&self.connection_metrics);
tokio::spawn(async move {
let mut durations_guard = duration_holder.connection_durations.write().await;
durations_guard.push(duration);
let len = durations_guard.len();
if len > 1000 {
durations_guard.drain(..len - 1000);
}
});
}
pub fn record_connection_error(&self) {
self.connection_metrics
.connection_errors
.fetch_add(1, Ordering::Relaxed);
}
pub async fn get_snapshot(&self) -> MetricsSnapshot {
let request_metrics = self.calculate_request_metrics().await;
let resource_metrics = self.calculate_resource_metrics().await;
let connection_metrics = self.calculate_connection_metrics().await;
let health_status = self.health_status.read().await.clone();
MetricsSnapshot {
timestamp: SystemTime::now(),
requests: request_metrics,
resources: resource_metrics,
connections: connection_metrics,
health: health_status,
}
}
pub async fn perform_health_check(&self) -> HealthStatus {
let mut health = HealthStatus::new();
health
.components
.insert("memory".to_string(), self.check_memory_health().await);
health.components.insert(
"connections".to_string(),
self.check_connection_health().await,
);
health
.components
.insert("requests".to_string(), self.check_request_health().await);
health
.components
.insert("errors".to_string(), self.check_error_health().await);
health.health_score = self.calculate_health_score(&health.components);
health.alerts = self.generate_health_alerts(&health.components).await;
*self.health_status.write().await = health.clone();
health
}
async fn calculate_request_metrics(&self) -> RequestMetrics {
let counts = self.request_counts.read().await;
let total_requests: u64 = counts.values().map(|c| c.load(Ordering::Relaxed)).sum();
let times = self.response_times.read().await;
let all_times: Vec<Duration> = times.values().flatten().cloned().collect();
let avg_response_time = if all_times.is_empty() {
0.0
} else {
all_times.iter().map(|d| d.as_millis() as f64).sum::<f64>() / all_times.len() as f64
};
let p95_response_time = if all_times.is_empty() {
0.0
} else {
let mut sorted_times = all_times.clone();
sorted_times.sort();
let p95_index = (sorted_times.len() as f64 * 0.95) as usize;
sorted_times
.get(p95_index.min(sorted_times.len() - 1))
.unwrap_or(&Duration::from_millis(0))
.as_millis() as f64
};
let error_counts = self.error_counts.read().await;
let total_errors: u64 = error_counts
.values()
.map(|c| c.load(Ordering::Relaxed))
.sum();
let error_rate = if total_requests > 0 {
(total_errors as f64 / total_requests as f64) * 100.0
} else {
0.0
};
let top_request_types: Vec<(String, u64)> = counts
.iter()
.map(|(method, count)| (method.clone(), count.load(Ordering::Relaxed)))
.collect();
RequestMetrics {
total_requests,
requests_per_second: 0.0, avg_response_time_ms: avg_response_time,
p95_response_time_ms: p95_response_time,
error_rate,
top_request_types,
}
}
async fn calculate_resource_metrics(&self) -> ResourceMetrics {
let cpu_samples = self.resource_usage.cpu_samples.read().await;
let avg_cpu_usage = if cpu_samples.is_empty() {
0.0
} else {
cpu_samples.iter().sum::<f64>() / cpu_samples.len() as f64
};
ResourceMetrics {
current_memory_mb: self.resource_usage.memory_usage_mb.load(Ordering::Relaxed),
peak_memory_mb: self.resource_usage.peak_memory_mb.load(Ordering::Relaxed),
avg_cpu_usage,
active_requests: self.resource_usage.active_requests.load(Ordering::Relaxed),
peak_concurrent_requests: self
.resource_usage
.peak_concurrent_requests
.load(Ordering::Relaxed),
}
}
async fn calculate_connection_metrics(&self) -> ConnectionMetricsSnapshot {
let durations = self.connection_metrics.connection_durations.read().await;
let avg_duration = if durations.is_empty() {
0.0
} else {
durations.iter().map(|d| d.as_millis() as f64).sum::<f64>() / durations.len() as f64
};
let total_connections = self
.connection_metrics
.total_connections
.load(Ordering::Relaxed);
let connection_errors = self
.connection_metrics
.connection_errors
.load(Ordering::Relaxed);
let error_rate = if total_connections > 0 {
(connection_errors as f64 / total_connections as f64) * 100.0
} else {
0.0
};
ConnectionMetricsSnapshot {
total_connections,
active_connections: self
.connection_metrics
.active_connections
.load(Ordering::Relaxed),
peak_connections: self
.connection_metrics
.peak_connections
.load(Ordering::Relaxed),
connection_error_rate: error_rate,
avg_connection_duration_ms: avg_duration,
}
}
async fn check_memory_health(&self) -> ComponentHealth {
let current_mb = self.resource_usage.memory_usage_mb.load(Ordering::Relaxed);
let peak_mb = self.resource_usage.peak_memory_mb.load(Ordering::Relaxed);
let status = if current_mb > 1000 {
ComponentStatus::Critical
} else if current_mb > 500 {
ComponentStatus::Warning
} else {
ComponentStatus::Healthy
};
let message = if current_mb > 1000 {
Some("High memory usage detected".to_string())
} else {
None
};
let mut metrics = HashMap::new();
metrics.insert("current_mb".to_string(), current_mb as f64);
metrics.insert("peak_mb".to_string(), peak_mb as f64);
ComponentHealth {
status,
message,
last_update: SystemTime::now(),
metrics,
}
}
async fn check_connection_health(&self) -> ComponentHealth {
let active = self
.connection_metrics
.active_connections
.load(Ordering::Relaxed);
let peak = self
.connection_metrics
.peak_connections
.load(Ordering::Relaxed);
let errors = self
.connection_metrics
.connection_errors
.load(Ordering::Relaxed);
let total = self
.connection_metrics
.total_connections
.load(Ordering::Relaxed);
let error_rate = if total > 0 {
(errors as f64 / total as f64) * 100.0
} else {
0.0
};
let status = if error_rate > 10.0 {
ComponentStatus::Critical
} else if error_rate > 5.0 || active > 100 {
ComponentStatus::Warning
} else {
ComponentStatus::Healthy
};
let message = if error_rate > 10.0 {
Some(format!("High connection error rate: {error_rate:.1}%"))
} else if active > 100 {
Some("High number of active connections".to_string())
} else {
None
};
let mut metrics = HashMap::new();
metrics.insert("active_connections".to_string(), active as f64);
metrics.insert("peak_connections".to_string(), peak as f64);
metrics.insert("error_rate".to_string(), error_rate);
ComponentHealth {
status,
message,
last_update: SystemTime::now(),
metrics,
}
}
async fn check_request_health(&self) -> ComponentHealth {
let active_requests = self.resource_usage.active_requests.load(Ordering::Relaxed);
let status = if active_requests > 1000 {
ComponentStatus::Critical
} else if active_requests > 500 {
ComponentStatus::Warning
} else {
ComponentStatus::Healthy
};
let message = if active_requests > 1000 {
Some("Very high request load".to_string())
} else if active_requests > 500 {
Some("High request load".to_string())
} else {
None
};
let mut metrics = HashMap::new();
metrics.insert("active_requests".to_string(), active_requests as f64);
ComponentHealth {
status,
message,
last_update: SystemTime::now(),
metrics,
}
}
async fn check_error_health(&self) -> ComponentHealth {
let error_counts = self.error_counts.read().await;
let total_errors: u64 = error_counts
.values()
.map(|c| c.load(Ordering::Relaxed))
.sum();
let request_counts = self.request_counts.read().await;
let total_requests: u64 = request_counts
.values()
.map(|c| c.load(Ordering::Relaxed))
.sum();
let error_rate = if total_requests > 0 {
(total_errors as f64 / total_requests as f64) * 100.0
} else {
0.0
};
let status = if error_rate > 10.0 {
ComponentStatus::Critical
} else if error_rate > 5.0 {
ComponentStatus::Warning
} else {
ComponentStatus::Healthy
};
let message = if error_rate > 10.0 {
Some(format!("High error rate: {error_rate:.1}%"))
} else if error_rate > 5.0 {
Some(format!("Elevated error rate: {error_rate:.1}%"))
} else {
None
};
let mut metrics = HashMap::new();
metrics.insert("error_rate".to_string(), error_rate);
metrics.insert("total_errors".to_string(), total_errors as f64);
ComponentHealth {
status,
message,
last_update: SystemTime::now(),
metrics,
}
}
fn calculate_health_score(&self, components: &HashMap<String, ComponentHealth>) -> u8 {
let total_components = components.len();
if total_components == 0 {
return 100;
}
let score_sum: u32 = components
.values()
.map(|health| match health.status {
ComponentStatus::Healthy => 100,
ComponentStatus::Warning => 75,
ComponentStatus::Degraded => 50,
ComponentStatus::Critical => 25,
ComponentStatus::Unavailable => 0,
})
.sum();
(score_sum / total_components as u32) as u8
}
async fn generate_health_alerts(
&self,
components: &HashMap<String, ComponentHealth>,
) -> Vec<HealthAlert> {
let mut alerts = Vec::new();
for (component_name, health) in components {
let (severity, should_alert) = match health.status {
ComponentStatus::Critical => (AlertSeverity::Critical, true),
ComponentStatus::Warning => (AlertSeverity::Warning, true),
ComponentStatus::Degraded => (AlertSeverity::Warning, true),
ComponentStatus::Unavailable => (AlertSeverity::Critical, true),
ComponentStatus::Healthy => (AlertSeverity::Info, false),
};
if should_alert {
let message = health
.message
.clone()
.unwrap_or_else(|| format!("{component_name} component is not healthy"));
alerts.push(HealthAlert {
severity,
message,
component: component_name.clone(),
timestamp: SystemTime::now(),
id: format!(
"{}_{}",
component_name,
SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or(Duration::from_secs(0))
.as_secs()
),
});
}
}
alerts
}
}
impl Default for MetricsCollector {
fn default() -> Self {
Self::new()
}
}
impl ResourceUsageTracker {
fn new() -> Self {
Self {
memory_usage_mb: AtomicUsize::new(0),
peak_memory_mb: AtomicUsize::new(0),
cpu_samples: RwLock::new(Vec::new()),
active_requests: AtomicUsize::new(0),
peak_concurrent_requests: AtomicUsize::new(0),
}
}
}
impl ConnectionMetrics {
fn new() -> Self {
Self {
total_connections: AtomicU64::new(0),
active_connections: AtomicUsize::new(0),
peak_connections: AtomicUsize::new(0),
connection_errors: AtomicU64::new(0),
connection_durations: RwLock::new(Vec::new()),
}
}
}
impl HealthStatus {
fn new() -> Self {
Self {
health_score: 100,
components: HashMap::new(),
last_check: SystemTime::now(),
alerts: Vec::new(),
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use tokio::time::Duration;
#[tokio::test]
async fn test_metrics_collection() {
let metrics = MetricsCollector::new();
metrics
.record_request("textDocument/completion", Duration::from_millis(100))
.await;
metrics
.record_request("textDocument/hover", Duration::from_millis(50))
.await;
metrics.record_error("timeout").await;
metrics.update_memory_usage(100);
metrics.record_cpu_usage(25.5).await;
metrics.record_connection();
let snapshot = metrics.get_snapshot().await;
assert!(snapshot.requests.total_requests > 0);
assert!(snapshot.resources.current_memory_mb == 100);
assert!(snapshot.connections.total_connections > 0);
}
#[tokio::test]
async fn test_health_check() {
let metrics = MetricsCollector::new();
metrics.update_memory_usage(100); metrics.record_cpu_usage(25.0).await;
let health = metrics.perform_health_check().await;
assert!(health.health_score > 50);
assert!(health.components.contains_key("memory"));
assert!(health.components.contains_key("connections"));
}
#[tokio::test]
async fn test_resource_limits() {
let metrics = MetricsCollector::new();
metrics.update_memory_usage(1200);
metrics
.resource_usage
.active_requests
.store(1500, Ordering::Relaxed);
let health = metrics.perform_health_check().await;
let memory_health = health.components.get("memory").unwrap();
let requests_health = health.components.get("requests").unwrap();
assert!(matches!(memory_health.status, ComponentStatus::Critical));
assert!(matches!(requests_health.status, ComponentStatus::Critical));
assert!(health.health_score < 75);
assert!(!health.alerts.is_empty());
}
}