use crate::error::{MlError, Result};
use std::collections::HashMap;
use std::path::PathBuf;
use std::sync::{Arc, RwLock};
use tracing::{debug, info};
#[derive(Debug, Clone)]
pub struct ModelVersion {
pub version: String,
pub path: PathBuf,
pub deployed_at: std::time::SystemTime,
pub metadata: HashMap<String, String>,
pub metrics: VersionMetrics,
}
#[derive(Debug, Clone, Default)]
pub struct VersionMetrics {
pub requests: u64,
pub avg_latency_ms: f32,
pub success_rate: f32,
pub avg_cpu_usage: f32,
pub avg_memory_mb: f32,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum DeploymentStrategy {
Replace,
BlueGreen,
Canary {
initial_percent: u8,
step_percent: u8,
},
ABTest {
split_percent: u8,
},
Shadow,
}
#[derive(Debug, Clone)]
pub struct ServerConfig {
pub max_concurrent: usize,
pub timeout_ms: u64,
pub enable_queue: bool,
pub queue_size: usize,
pub health_check: bool,
pub health_check_interval_s: u64,
}
impl Default for ServerConfig {
fn default() -> Self {
Self {
max_concurrent: 100,
timeout_ms: 30000,
enable_queue: true,
queue_size: 1000,
health_check: true,
health_check_interval_s: 30,
}
}
}
pub struct ModelServer {
config: ServerConfig,
versions: Arc<RwLock<HashMap<String, ModelVersion>>>,
active_version: Arc<RwLock<String>>,
routing: Arc<RwLock<RoutingStrategy>>,
}
#[derive(Debug, Clone)]
enum RoutingStrategy {
Single {
version: String,
},
Weighted {
weights: HashMap<String, u8>,
},
Canary {
stable: String,
canary: String,
canary_percent: u8,
},
}
impl ModelServer {
#[must_use]
pub fn new(config: ServerConfig) -> Self {
info!("Initializing model server");
Self {
config,
versions: Arc::new(RwLock::new(HashMap::new())),
active_version: Arc::new(RwLock::new(String::new())),
routing: Arc::new(RwLock::new(RoutingStrategy::Single {
version: String::new(),
})),
}
}
pub fn register_version(
&mut self,
version_id: &str,
model_path: PathBuf,
metadata: HashMap<String, String>,
) -> Result<()> {
info!("Registering model version: {}", version_id);
if !model_path.exists() {
return Err(MlError::InvalidConfig(format!(
"Model file not found: {}",
model_path.display()
)));
}
let version = ModelVersion {
version: version_id.to_string(),
path: model_path,
deployed_at: std::time::SystemTime::now(),
metadata,
metrics: VersionMetrics::default(),
};
if let Ok(mut versions) = self.versions.write() {
versions.insert(version_id.to_string(), version);
}
Ok(())
}
pub fn deploy(&mut self, version_id: &str, strategy: DeploymentStrategy) -> Result<()> {
info!(
"Deploying version {} with strategy {:?}",
version_id, strategy
);
let version_exists = self
.versions
.read()
.map(|v| v.contains_key(version_id))
.unwrap_or(false);
if !version_exists {
return Err(MlError::InvalidConfig(format!(
"Version not found: {}",
version_id
)));
}
match strategy {
DeploymentStrategy::Replace => self.deploy_replace(version_id),
DeploymentStrategy::BlueGreen => self.deploy_blue_green(version_id),
DeploymentStrategy::Canary {
initial_percent,
step_percent,
} => self.deploy_canary(version_id, initial_percent, step_percent),
DeploymentStrategy::ABTest { split_percent } => {
self.deploy_ab_test(version_id, split_percent)
}
DeploymentStrategy::Shadow => self.deploy_shadow(version_id),
}
}
pub fn rollback(&mut self, version_id: &str) -> Result<()> {
info!("Rolling back to version: {}", version_id);
self.deploy_replace(version_id)
}
#[must_use]
pub fn version_metrics(&self) -> HashMap<String, VersionMetrics> {
self.versions
.read()
.map(|versions| {
versions
.iter()
.map(|(k, v)| (k.clone(), v.metrics.clone()))
.collect()
})
.unwrap_or_default()
}
#[must_use]
pub fn active_version(&self) -> String {
self.active_version
.read()
.map(|v| v.clone())
.unwrap_or_default()
}
#[must_use]
pub fn health_check(&self) -> HealthStatus {
if !self.config.health_check {
return HealthStatus::Unknown;
}
let has_models = self.versions.read().map(|v| !v.is_empty()).unwrap_or(false);
if !has_models {
return HealthStatus::Unhealthy;
}
let active_version = self.active_version();
if active_version.is_empty() {
return HealthStatus::Degraded;
}
let version_exists = self
.versions
.read()
.map(|v| v.contains_key(&active_version))
.unwrap_or(false);
if !version_exists {
return HealthStatus::Unhealthy;
}
if let Ok(memory_info) = Self::get_memory_usage() {
if memory_info.usage_percent > 90.0 {
return HealthStatus::Degraded;
}
if memory_info.usage_percent > 95.0 {
return HealthStatus::Unhealthy;
}
}
HealthStatus::Healthy
}
fn get_memory_usage() -> Result<MemoryInfo> {
#[cfg(target_os = "linux")]
{
Self::get_memory_usage_linux()
}
#[cfg(target_os = "macos")]
{
Self::get_memory_usage_macos()
}
#[cfg(target_os = "windows")]
{
Self::get_memory_usage_windows()
}
#[cfg(not(any(target_os = "linux", target_os = "macos", target_os = "windows")))]
{
Ok(MemoryInfo {
total_mb: 0,
used_mb: 0,
available_mb: 0,
usage_percent: 0.0,
})
}
}
#[cfg(target_os = "linux")]
fn get_memory_usage_linux() -> Result<MemoryInfo> {
use std::fs;
let meminfo = fs::read_to_string("/proc/meminfo")
.map_err(|e| MlError::InvalidConfig(format!("Failed to read meminfo: {}", e)))?;
let mut total = 0u64;
let mut available = 0u64;
for line in meminfo.lines() {
if let Some(rest) = line.strip_prefix("MemTotal:") {
total = rest
.trim()
.split_whitespace()
.next()
.and_then(|s| s.parse::<u64>().ok())
.unwrap_or(0);
} else if let Some(rest) = line.strip_prefix("MemAvailable:") {
available = rest
.trim()
.split_whitespace()
.next()
.and_then(|s| s.parse::<u64>().ok())
.unwrap_or(0);
}
}
let total_mb = total / 1024;
let available_mb = available / 1024;
let used_mb = total_mb.saturating_sub(available_mb);
let usage_percent = if total_mb > 0 {
(used_mb as f32 / total_mb as f32) * 100.0
} else {
0.0
};
Ok(MemoryInfo {
total_mb,
used_mb,
available_mb,
usage_percent,
})
}
#[cfg(target_os = "macos")]
fn get_memory_usage_macos() -> Result<MemoryInfo> {
Ok(MemoryInfo {
total_mb: 16384, used_mb: 8192, available_mb: 8192,
usage_percent: 50.0,
})
}
#[cfg(target_os = "windows")]
fn get_memory_usage_windows() -> Result<MemoryInfo> {
Ok(MemoryInfo {
total_mb: 16384, used_mb: 8192, available_mb: 8192,
usage_percent: 50.0,
})
}
fn deploy_replace(&mut self, version_id: &str) -> Result<()> {
debug!("Deploying with replace strategy");
if let Ok(mut active) = self.active_version.write() {
*active = version_id.to_string();
}
if let Ok(mut routing) = self.routing.write() {
*routing = RoutingStrategy::Single {
version: version_id.to_string(),
};
}
info!("Version {} deployed successfully", version_id);
Ok(())
}
fn deploy_blue_green(&mut self, version_id: &str) -> Result<()> {
debug!("Deploying with blue-green strategy");
self.deploy_replace(version_id)
}
fn deploy_canary(
&mut self,
version_id: &str,
initial_percent: u8,
_step_percent: u8,
) -> Result<()> {
debug!(
"Deploying with canary strategy ({}% initial)",
initial_percent
);
let stable_version = self.active_version();
if let Ok(mut routing) = self.routing.write() {
*routing = RoutingStrategy::Canary {
stable: stable_version,
canary: version_id.to_string(),
canary_percent: initial_percent,
};
}
info!("Canary deployment started for version {}", version_id);
Ok(())
}
fn deploy_ab_test(&mut self, version_id: &str, split_percent: u8) -> Result<()> {
debug!("Deploying with A/B test ({}% split)", split_percent);
let stable_version = self.active_version();
let mut weights = HashMap::new();
weights.insert(stable_version, 100 - split_percent);
weights.insert(version_id.to_string(), split_percent);
if let Ok(mut routing) = self.routing.write() {
*routing = RoutingStrategy::Weighted { weights };
}
info!("A/B test started for version {}", version_id);
Ok(())
}
fn deploy_shadow(&mut self, version_id: &str) -> Result<()> {
debug!("Deploying in shadow mode");
info!("Version {} deployed in shadow mode", version_id);
Ok(())
}
pub fn increase_canary_traffic(&mut self, increment: u8) -> Result<()> {
let mut routing = self
.routing
.write()
.map_err(|_| MlError::InvalidConfig("Failed to acquire routing lock".to_string()))?;
match &mut *routing {
RoutingStrategy::Canary { canary_percent, .. } => {
*canary_percent = (*canary_percent + increment).min(100);
info!("Increased canary traffic to {}%", canary_percent);
Ok(())
}
_ => Err(MlError::InvalidConfig(
"Not in canary deployment mode".to_string(),
)),
}
}
pub fn promote_canary(&mut self) -> Result<()> {
let routing = self
.routing
.read()
.map_err(|_| MlError::InvalidConfig("Failed to acquire routing lock".to_string()))?;
if let RoutingStrategy::Canary { canary, .. } = &*routing {
let canary_version = canary.clone();
drop(routing); self.deploy_replace(&canary_version)?;
info!("Canary promoted to stable");
Ok(())
} else {
Err(MlError::InvalidConfig(
"Not in canary deployment mode".to_string(),
))
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum HealthStatus {
Healthy,
Degraded,
Unhealthy,
Unknown,
}
#[derive(Debug, Clone)]
struct MemoryInfo {
total_mb: u64,
used_mb: u64,
available_mb: u64,
usage_percent: f32,
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_server_config_default() {
let config = ServerConfig::default();
assert_eq!(config.max_concurrent, 100);
assert_eq!(config.timeout_ms, 30000);
assert!(config.enable_queue);
}
#[test]
fn test_deployment_strategy_variants() {
let strategies = vec![
DeploymentStrategy::Replace,
DeploymentStrategy::BlueGreen,
DeploymentStrategy::Canary {
initial_percent: 10,
step_percent: 10,
},
DeploymentStrategy::ABTest { split_percent: 50 },
DeploymentStrategy::Shadow,
];
for strategy in strategies {
let _ = format!("{:?}", strategy);
}
}
#[test]
fn test_model_server_creation() {
let config = ServerConfig::default();
let server = ModelServer::new(config);
assert_eq!(server.active_version(), "");
}
#[test]
fn test_health_status() {
assert_eq!(HealthStatus::Healthy, HealthStatus::Healthy);
assert_ne!(HealthStatus::Healthy, HealthStatus::Degraded);
}
#[test]
fn test_version_metrics() {
let metrics = VersionMetrics {
requests: 1000,
avg_latency_ms: 50.0,
success_rate: 0.99,
avg_cpu_usage: 45.0,
avg_memory_mb: 512.0,
};
assert_eq!(metrics.requests, 1000);
assert!((metrics.success_rate - 0.99).abs() < 0.01);
}
}