use async_trait::async_trait;
use chrono::{DateTime, Utc};
use rand::Rng;
use serde::{Deserialize, Serialize};
use std::collections::{HashMap, VecDeque};
use std::hash::{Hash, Hasher};
use std::sync::Arc;
use tokio::sync::RwLock;
use tracing::{debug, error, info, warn};
use super::error::{ActuatorError, ActuatorResult, DeploymentState};
use super::traits::{CanaryDeployment, HealthMonitor, TrafficSplitter};
use super::types::{
CanaryConfig, DeploymentMetrics, DeploymentRequest, DeploymentStatus, HealthCheckResult,
HealthStatus, RollbackReason, RollbackRequest, RollbackResult, StatisticalAnalysis,
SuccessCriteria, TrafficSplitStrategy,
};
#[derive(Debug)]
struct CanaryDeploymentState {
request: DeploymentRequest,
config: CanaryConfig,
current_phase: usize,
phase_start_time: DateTime<Utc>,
deployment_start_time: DateTime<Utc>,
state: DeploymentState,
metrics: DeploymentMetrics,
health: HealthStatus,
errors: Vec<String>,
paused: bool,
traffic_split_counter: u64,
last_updated: DateTime<Utc>,
}
#[derive(Clone)]
pub struct CanaryDeploymentEngine {
deployments: Arc<RwLock<HashMap<String, CanaryDeploymentState>>>,
health_monitor: Option<Arc<dyn HealthMonitor>>,
default_config: CanaryConfig,
metrics_buffer: Arc<RwLock<HashMap<String, MetricsBuffer>>>,
}
impl std::fmt::Debug for CanaryDeploymentEngine {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("CanaryDeploymentEngine")
.field("deployments", &self.deployments)
.field("health_monitor", &self.health_monitor.as_ref().map(|_| "Arc<dyn HealthMonitor>"))
.field("default_config", &self.default_config)
.field("metrics_buffer", &self.metrics_buffer)
.finish()
}
}
#[derive(Debug, Clone)]
struct MetricsBuffer {
canary_latencies: VecDeque<f64>,
control_latencies: VecDeque<f64>,
canary_successes: u64,
canary_failures: u64,
control_successes: u64,
control_failures: u64,
max_samples: usize,
}
impl MetricsBuffer {
fn new(max_samples: usize) -> Self {
Self {
canary_latencies: VecDeque::with_capacity(max_samples),
control_latencies: VecDeque::with_capacity(max_samples),
canary_successes: 0,
canary_failures: 0,
control_successes: 0,
control_failures: 0,
max_samples,
}
}
fn add_canary_latency(&mut self, latency: f64) {
if self.canary_latencies.len() >= self.max_samples {
self.canary_latencies.pop_front();
}
self.canary_latencies.push_back(latency);
}
fn add_control_latency(&mut self, latency: f64) {
if self.control_latencies.len() >= self.max_samples {
self.control_latencies.pop_front();
}
self.control_latencies.push_back(latency);
}
fn record_canary_result(&mut self, success: bool) {
if success {
self.canary_successes += 1;
} else {
self.canary_failures += 1;
}
}
fn record_control_result(&mut self, success: bool) {
if success {
self.control_successes += 1;
} else {
self.control_failures += 1;
}
}
fn calculate_percentile(values: &VecDeque<f64>, percentile: f64) -> f64 {
if values.is_empty() {
return 0.0;
}
let mut sorted: Vec<f64> = values.iter().copied().collect();
sorted.sort_by(|a, b| a.partial_cmp(b).unwrap());
let index = ((percentile / 100.0) * (sorted.len() - 1) as f64).round() as usize;
sorted[index]
}
fn get_metrics(&self) -> (f64, f64, f64, f64, f64, f64) {
let canary_p95 = Self::calculate_percentile(&self.canary_latencies, 95.0);
let canary_p99 = Self::calculate_percentile(&self.canary_latencies, 99.0);
let control_p95 = Self::calculate_percentile(&self.control_latencies, 95.0);
let control_p99 = Self::calculate_percentile(&self.control_latencies, 99.0);
let canary_total = self.canary_successes + self.canary_failures;
let control_total = self.control_successes + self.control_failures;
let canary_success_rate = if canary_total > 0 {
(self.canary_successes as f64 / canary_total as f64) * 100.0
} else {
0.0
};
let control_success_rate = if control_total > 0 {
(self.control_successes as f64 / control_total as f64) * 100.0
} else {
0.0
};
(
canary_success_rate,
control_success_rate,
canary_p95,
control_p95,
canary_p99,
control_p99,
)
}
}
impl CanaryDeploymentEngine {
pub fn new(default_config: CanaryConfig) -> Self {
Self {
deployments: Arc::new(RwLock::new(HashMap::new())),
health_monitor: None,
default_config,
metrics_buffer: Arc::new(RwLock::new(HashMap::new())),
}
}
pub fn with_health_monitor(
default_config: CanaryConfig,
health_monitor: Arc<dyn HealthMonitor>,
) -> Self {
Self {
deployments: Arc::new(RwLock::new(HashMap::new())),
health_monitor: Some(health_monitor),
default_config,
metrics_buffer: Arc::new(RwLock::new(HashMap::new())),
}
}
async fn should_auto_promote(&self, deployment_id: &str) -> ActuatorResult<bool> {
let deployments = self.deployments.read().await;
let state = deployments
.get(deployment_id)
.ok_or_else(|| ActuatorError::DeploymentFailed("Deployment not found".to_string()))?;
if !state.config.auto_promote {
return Ok(false);
}
let now = Utc::now();
let phase_duration = state.config.phase_duration;
let elapsed = now
.signed_duration_since(state.phase_start_time)
.to_std()
.unwrap_or_default();
if elapsed < phase_duration {
debug!(
"Phase duration not elapsed: {:?}/{:?}",
elapsed, phase_duration
);
return Ok(false);
}
self.validate_success_criteria(deployment_id).await
}
async fn validate_success_criteria(&self, deployment_id: &str) -> ActuatorResult<bool> {
let deployments = self.deployments.read().await;
let state = deployments
.get(deployment_id)
.ok_or_else(|| ActuatorError::DeploymentFailed("Deployment not found".to_string()))?;
let criteria = &state.config.success_criteria;
let metrics = &state.metrics;
if (metrics.canary_requests as usize) < criteria.min_sample_size {
debug!(
"Insufficient sample size: {} < {}",
metrics.canary_requests, criteria.min_sample_size
);
return Ok(false);
}
if metrics.canary_success_rate < criteria.min_success_rate {
warn!(
"Canary success rate {} below threshold {}",
metrics.canary_success_rate, criteria.min_success_rate
);
return Ok(false);
}
if metrics.canary_error_rate > criteria.max_error_rate {
warn!(
"Canary error rate {} exceeds threshold {}",
metrics.canary_error_rate, criteria.max_error_rate
);
return Ok(false);
}
if metrics.canary_p95_latency_ms > criteria.max_p95_latency_ms {
warn!(
"Canary P95 latency {} exceeds threshold {}",
metrics.canary_p95_latency_ms, criteria.max_p95_latency_ms
);
return Ok(false);
}
if metrics.canary_p99_latency_ms > criteria.max_p99_latency_ms {
warn!(
"Canary P99 latency {} exceeds threshold {}",
metrics.canary_p99_latency_ms, criteria.max_p99_latency_ms
);
return Ok(false);
}
if metrics.canary_cost_per_request > 0.0 && metrics.control_cost_per_request > 0.0 {
let cost_increase = ((metrics.canary_cost_per_request
- metrics.control_cost_per_request)
/ metrics.control_cost_per_request)
* 100.0;
if cost_increase > criteria.max_cost_increase_pct {
warn!(
"Cost increase {}% exceeds threshold {}%",
cost_increase, criteria.max_cost_increase_pct
);
return Ok(false);
}
}
if let Some(analysis) = &metrics.statistical_analysis {
if analysis.significant_difference && analysis.success_rate_p_value < 0.05 {
if metrics.canary_success_rate < metrics.control_success_rate {
warn!("Statistically significant degradation detected");
return Ok(false);
}
}
}
info!("All success criteria met for deployment {}", deployment_id);
Ok(true)
}
async fn should_auto_rollback(
&self,
deployment_id: &str,
) -> ActuatorResult<Option<RollbackReason>> {
let deployments = self.deployments.read().await;
let state = deployments
.get(deployment_id)
.ok_or_else(|| ActuatorError::DeploymentFailed("Deployment not found".to_string()))?;
if !state.config.auto_rollback {
return Ok(None);
}
let criteria = &state.config.success_criteria;
let metrics = &state.metrics;
if (metrics.canary_requests as usize) < criteria.min_sample_size {
return Ok(None);
}
if metrics.canary_error_rate > criteria.max_error_rate * 2.0 {
return Ok(Some(RollbackReason::HighErrorRate));
}
if metrics.canary_p99_latency_ms > criteria.max_p99_latency_ms * 2.0 {
return Ok(Some(RollbackReason::HighLatency));
}
if !state.health.healthy && state.health.checks_failed > 2 {
return Ok(Some(RollbackReason::HealthCheckFailed));
}
let now = Utc::now();
let total_elapsed = now
.signed_duration_since(state.deployment_start_time)
.to_std()
.unwrap_or_default();
if total_elapsed > state.config.max_duration {
return Ok(Some(RollbackReason::Timeout));
}
Ok(None)
}
pub async fn update_metrics(
&self,
deployment_id: &str,
is_canary: bool,
latency_ms: f64,
success: bool,
cost: Option<f64>,
) -> ActuatorResult<()> {
{
let mut buffers = self.metrics_buffer.write().await;
let buffer = buffers
.entry(deployment_id.to_string())
.or_insert_with(|| MetricsBuffer::new(10000));
if is_canary {
buffer.add_canary_latency(latency_ms);
buffer.record_canary_result(success);
} else {
buffer.add_control_latency(latency_ms);
buffer.record_control_result(success);
}
}
let mut deployments = self.deployments.write().await;
let state = deployments.get_mut(deployment_id).ok_or_else(|| {
ActuatorError::DeploymentFailed("Deployment not found".to_string())
})?;
state.metrics.total_requests += 1;
if is_canary {
state.metrics.canary_requests += 1;
if let Some(cost_value) = cost {
let n = state.metrics.canary_requests as f64;
state.metrics.canary_cost_per_request =
((n - 1.0) * state.metrics.canary_cost_per_request + cost_value) / n;
}
} else {
state.metrics.control_requests += 1;
if let Some(cost_value) = cost {
let n = state.metrics.control_requests as f64;
state.metrics.control_cost_per_request =
((n - 1.0) * state.metrics.control_cost_per_request + cost_value) / n;
}
}
let buffers = self.metrics_buffer.read().await;
if let Some(buffer) = buffers.get(deployment_id) {
let (canary_sr, control_sr, canary_p95, control_p95, canary_p99, control_p99) =
buffer.get_metrics();
state.metrics.canary_success_rate = canary_sr;
state.metrics.control_success_rate = control_sr;
state.metrics.canary_error_rate = 100.0 - canary_sr;
state.metrics.control_error_rate = 100.0 - control_sr;
state.metrics.canary_p95_latency_ms = canary_p95;
state.metrics.control_p95_latency_ms = control_p95;
state.metrics.canary_p99_latency_ms = canary_p99;
state.metrics.control_p99_latency_ms = control_p99;
if buffer.canary_successes + buffer.canary_failures
>= state.config.success_criteria.min_sample_size as u64
&& buffer.control_successes + buffer.control_failures
>= state.config.success_criteria.min_sample_size as u64
{
state.metrics.statistical_analysis = Some(self.perform_statistical_analysis(buffer));
}
}
state.last_updated = Utc::now();
Ok(())
}
fn perform_statistical_analysis(&self, buffer: &MetricsBuffer) -> StatisticalAnalysis {
let canary_total = buffer.canary_successes + buffer.canary_failures;
let control_total = buffer.control_successes + buffer.control_failures;
let canary_success_rate =
buffer.canary_successes as f64 / canary_total.max(1) as f64;
let control_success_rate =
buffer.control_successes as f64 / control_total.max(1) as f64;
let pooled_rate = (buffer.canary_successes + buffer.control_successes) as f64
/ (canary_total + control_total) as f64;
let se = (pooled_rate * (1.0 - pooled_rate)
* (1.0 / canary_total as f64 + 1.0 / control_total as f64))
.sqrt();
let z_score = (canary_success_rate - control_success_rate) / se.max(0.0001);
let success_rate_p_value = 2.0 * (1.0 - normal_cdf(z_score.abs()));
let latency_p_value = if !buffer.canary_latencies.is_empty()
&& !buffer.control_latencies.is_empty()
{
let canary_mean: f64 = buffer.canary_latencies.iter().sum::<f64>()
/ buffer.canary_latencies.len() as f64;
let control_mean: f64 = buffer.control_latencies.iter().sum::<f64>()
/ buffer.control_latencies.len() as f64;
let canary_var: f64 = buffer
.canary_latencies
.iter()
.map(|x| (x - canary_mean).powi(2))
.sum::<f64>()
/ buffer.canary_latencies.len().max(1) as f64;
let control_var: f64 = buffer
.control_latencies
.iter()
.map(|x| (x - control_mean).powi(2))
.sum::<f64>()
/ buffer.control_latencies.len().max(1) as f64;
let se_latency = (canary_var / buffer.canary_latencies.len() as f64
+ control_var / buffer.control_latencies.len() as f64)
.sqrt();
let t_stat = (canary_mean - control_mean) / se_latency.max(0.0001);
2.0 * (1.0 - normal_cdf(t_stat.abs()))
} else {
1.0
};
let significant_difference = success_rate_p_value < 0.05 || latency_p_value < 0.05;
StatisticalAnalysis {
success_rate_p_value,
latency_p_value,
significant_difference,
confidence_level: 0.95,
canary_sample_size: canary_total as usize,
control_sample_size: control_total as usize,
}
}
pub async fn pause(&self, deployment_id: &str) -> ActuatorResult<()> {
let mut deployments = self.deployments.write().await;
let state = deployments.get_mut(deployment_id).ok_or_else(|| {
ActuatorError::DeploymentFailed("Deployment not found".to_string())
})?;
state.paused = true;
state.last_updated = Utc::now();
info!("Deployment {} paused", deployment_id);
Ok(())
}
pub async fn resume(&self, deployment_id: &str) -> ActuatorResult<()> {
let mut deployments = self.deployments.write().await;
let state = deployments.get_mut(deployment_id).ok_or_else(|| {
ActuatorError::DeploymentFailed("Deployment not found".to_string())
})?;
state.paused = false;
state.phase_start_time = Utc::now(); state.last_updated = Utc::now();
info!("Deployment {} resumed", deployment_id);
Ok(())
}
pub async fn is_paused(&self, deployment_id: &str) -> ActuatorResult<bool> {
let deployments = self.deployments.read().await;
let state = deployments
.get(deployment_id)
.ok_or_else(|| ActuatorError::DeploymentFailed("Deployment not found".to_string()))?;
Ok(state.paused)
}
pub async fn get_config(&self, deployment_id: &str) -> ActuatorResult<CanaryConfig> {
let deployments = self.deployments.read().await;
let state = deployments
.get(deployment_id)
.ok_or_else(|| ActuatorError::DeploymentFailed("Deployment not found".to_string()))?;
Ok(state.config.clone())
}
pub async fn update_health(
&self,
deployment_id: &str,
health: HealthStatus,
) -> ActuatorResult<()> {
let mut deployments = self.deployments.write().await;
let state = deployments.get_mut(deployment_id).ok_or_else(|| {
ActuatorError::DeploymentFailed("Deployment not found".to_string())
})?;
state.health = health;
state.last_updated = Utc::now();
Ok(())
}
}
#[async_trait]
impl CanaryDeployment for CanaryDeploymentEngine {
async fn start_canary(
&mut self,
request: DeploymentRequest,
) -> ActuatorResult<DeploymentStatus> {
let deployment_id = request.id.clone();
info!("Starting canary deployment: {}", deployment_id);
if request.config_changes.is_empty() {
return Err(ActuatorError::ValidationFailed(
"No configuration changes specified".to_string(),
));
}
let now = Utc::now();
let config = self.default_config.clone();
let initial_health = HealthStatus {
healthy: true,
checks_passed: 0,
checks_failed: 0,
check_results: vec![],
last_check_time: now,
};
let state = CanaryDeploymentState {
request: request.clone(),
config: config.clone(),
current_phase: 0,
phase_start_time: now,
deployment_start_time: now,
state: DeploymentState::Validating,
metrics: DeploymentMetrics::default(),
health: initial_health.clone(),
errors: vec![],
paused: false,
traffic_split_counter: 0,
last_updated: now,
};
{
let mut deployments = self.deployments.write().await;
deployments.insert(deployment_id.clone(), state);
}
{
let mut buffers = self.metrics_buffer.write().await;
buffers.insert(
deployment_id.clone(),
MetricsBuffer::new(10000),
);
}
{
let mut deployments = self.deployments.write().await;
if let Some(state) = deployments.get_mut(&deployment_id) {
state.state = DeploymentState::RollingOut;
state.last_updated = Utc::now();
}
}
info!(
"Canary deployment {} started with {} phases",
deployment_id,
config.phases.len()
);
self.get_status(&deployment_id).await
}
async fn promote(&mut self, deployment_id: &str) -> ActuatorResult<DeploymentStatus> {
info!("Promoting canary deployment: {}", deployment_id);
if !self.can_promote(deployment_id).await? {
return Err(ActuatorError::ValidationFailed(
"Cannot promote: success criteria not met".to_string(),
));
}
let mut deployments = self.deployments.write().await;
let state = deployments.get_mut(deployment_id).ok_or_else(|| {
ActuatorError::DeploymentFailed("Deployment not found".to_string())
})?;
if state.current_phase < state.config.phases.len() - 1 {
state.current_phase += 1;
state.phase_start_time = Utc::now();
state.state = DeploymentState::RollingOut;
state.last_updated = Utc::now();
let traffic_percent = state.config.phases[state.current_phase].traffic_percent;
info!(
"Promoted deployment {} to phase {} ({}% traffic)",
deployment_id, state.current_phase, traffic_percent
);
} else {
state.state = DeploymentState::Completed;
state.last_updated = Utc::now();
info!("Deployment {} completed", deployment_id);
}
drop(deployments);
self.get_status(deployment_id).await
}
async fn complete(&mut self, deployment_id: &str) -> ActuatorResult<DeploymentStatus> {
info!("Completing canary deployment: {}", deployment_id);
let mut deployments = self.deployments.write().await;
let state = deployments.get_mut(deployment_id).ok_or_else(|| {
ActuatorError::DeploymentFailed("Deployment not found".to_string())
})?;
state.current_phase = state.config.phases.len() - 1;
state.state = DeploymentState::Completed;
state.last_updated = Utc::now();
info!("Deployment {} completed at 100%", deployment_id);
drop(deployments);
self.get_status(deployment_id).await
}
async fn abort(&mut self, deployment_id: &str) -> ActuatorResult<RollbackResult> {
info!("Aborting canary deployment: {}", deployment_id);
let request = RollbackRequest {
deployment_id: deployment_id.to_string(),
reason: RollbackReason::Manual,
timestamp: Utc::now(),
automatic: false,
metadata: HashMap::new(),
};
self.execute_rollback(&request).await
}
async fn get_status(&self, deployment_id: &str) -> ActuatorResult<DeploymentStatus> {
let deployments = self.deployments.read().await;
let state = deployments
.get(deployment_id)
.ok_or_else(|| ActuatorError::DeploymentFailed("Deployment not found".to_string()))?;
let traffic_percent = if state.current_phase < state.config.phases.len() {
state.config.phases[state.current_phase].traffic_percent
} else {
100.0
};
Ok(DeploymentStatus {
deployment_id: deployment_id.to_string(),
state: state.state,
current_phase: Some(state.current_phase),
traffic_percent,
start_time: state.deployment_start_time,
phase_start_time: Some(state.phase_start_time),
health: state.health.clone(),
metrics: state.metrics.clone(),
errors: state.errors.clone(),
last_updated: state.last_updated,
})
}
async fn can_promote(&self, deployment_id: &str) -> ActuatorResult<bool> {
let deployments = self.deployments.read().await;
let state = deployments
.get(deployment_id)
.ok_or_else(|| ActuatorError::DeploymentFailed("Deployment not found".to_string()))?;
if state.paused {
return Ok(false);
}
if state.state != DeploymentState::RollingOut {
return Ok(false);
}
if state.current_phase >= state.config.phases.len() - 1 {
return Ok(false);
}
drop(deployments);
self.validate_success_criteria(deployment_id).await
}
}
impl CanaryDeploymentEngine {
async fn execute_rollback(&self, request: &RollbackRequest) -> ActuatorResult<RollbackResult> {
let start_time = Utc::now();
let deployment_id = &request.deployment_id;
let mut deployments = self.deployments.write().await;
let state = deployments.get_mut(deployment_id).ok_or_else(|| {
ActuatorError::RollbackFailed("Deployment not found".to_string())
})?;
state.state = DeploymentState::RollingBack;
state.last_updated = Utc::now();
let reverted_changes = state.request.config_changes.clone();
state.state = DeploymentState::RolledBack;
state.last_updated = Utc::now();
let duration = Utc::now()
.signed_duration_since(start_time)
.to_std()
.unwrap_or_default();
info!(
"Rollback completed for deployment {} in {:?}",
deployment_id, duration
);
Ok(RollbackResult {
deployment_id: deployment_id.clone(),
success: true,
error: None,
reverted_changes,
timestamp: Utc::now(),
duration,
})
}
}
#[async_trait]
impl TrafficSplitter for CanaryDeploymentEngine {
async fn route_request(&self, request_id: &str, canary_percent: f64) -> ActuatorResult<bool> {
let deployments = self.deployments.read().await;
let active_deployment = deployments
.values()
.find(|d| d.state == DeploymentState::RollingOut);
if let Some(deployment) = active_deployment {
match deployment.config.traffic_split_strategy {
TrafficSplitStrategy::WeightedRandom => {
let mut rng = rand::thread_rng();
let random: f64 = rng.gen();
Ok(random < (canary_percent / 100.0))
}
TrafficSplitStrategy::ConsistentHash => {
let mut hasher = std::collections::hash_map::DefaultHasher::new();
request_id.hash(&mut hasher);
let hash = hasher.finish();
Ok((hash % 100) < canary_percent as u64)
}
TrafficSplitStrategy::RoundRobin => {
let counter = deployment.traffic_split_counter;
Ok((counter % 100) < canary_percent as u64)
}
TrafficSplitStrategy::Geographic => {
let mut hasher = std::collections::hash_map::DefaultHasher::new();
request_id.hash(&mut hasher);
let hash = hasher.finish();
Ok((hash % 100) < canary_percent as u64)
}
}
} else {
Ok(false)
}
}
fn get_routing_stats(&self) -> HashMap<String, u64> {
HashMap::new()
}
}
fn normal_cdf(x: f64) -> f64 {
0.5 * (1.0 + erf(x / std::f64::consts::SQRT_2))
}
fn erf(x: f64) -> f64 {
let a1 = 0.254829592;
let a2 = -0.284496736;
let a3 = 1.421413741;
let a4 = -1.453152027;
let a5 = 1.061405429;
let p = 0.3275911;
let sign = if x < 0.0 { -1.0 } else { 1.0 };
let x = x.abs();
let t = 1.0 / (1.0 + p * x);
let y = 1.0 - (((((a5 * t + a4) * t) + a3) * t + a2) * t + a1) * t * (-x * x).exp();
sign * y
}
#[cfg(test)]
mod tests {
use super::*;
use std::time::Duration as StdDuration;
fn create_test_request(id: &str) -> DeploymentRequest {
DeploymentRequest {
id: id.to_string(),
timestamp: Utc::now(),
decision: crate::decision::Decision {
id: format!("decision-{}", id),
timestamp: Utc::now(),
decision_type: crate::decision::DecisionType::PromptOptimization,
criteria: crate::decision::DecisionCriteria {
metric_name: "latency".to_string(),
operator: crate::decision::ComparisonOperator::LessThan,
threshold: 100.0,
priority: 1.0,
},
config_changes: vec![],
expected_impact: crate::decision::ExpectedImpact::default(),
rollback_plan: crate::decision::RollbackPlan::default(),
safety_checks: vec![],
metadata: HashMap::new(),
},
strategy: super::super::types::DeploymentStrategy::Canary,
config_changes: vec![],
expected_duration: StdDuration::from_secs(300),
metadata: HashMap::new(),
}
}
#[tokio::test]
async fn test_start_canary_deployment() {
let config = CanaryConfig::default();
let mut engine = CanaryDeploymentEngine::new(config);
let mut request = create_test_request("test-deploy-1");
request.config_changes = vec![crate::decision::ConfigChange {
config_type: crate::decision::ConfigType::Provider,
key: "model".to_string(),
old_value: serde_json::json!("gpt-3.5-turbo"),
new_value: serde_json::json!("gpt-4"),
reason: "Testing".to_string(),
}];
let status = engine.start_canary(request).await.unwrap();
assert_eq!(status.deployment_id, "test-deploy-1");
assert_eq!(status.state, DeploymentState::RollingOut);
assert_eq!(status.current_phase, Some(0));
assert_eq!(status.traffic_percent, 1.0);
}
#[tokio::test]
async fn test_cannot_start_without_config_changes() {
let config = CanaryConfig::default();
let mut engine = CanaryDeploymentEngine::new(config);
let request = create_test_request("test-deploy-2");
let result = engine.start_canary(request).await;
assert!(result.is_err());
assert!(matches!(
result.unwrap_err(),
ActuatorError::ValidationFailed(_)
));
}
#[tokio::test]
async fn test_update_metrics() {
let config = CanaryConfig::default();
let mut engine = CanaryDeploymentEngine::new(config);
let mut request = create_test_request("test-deploy-3");
request.config_changes = vec![crate::decision::ConfigChange {
config_type: crate::decision::ConfigType::Provider,
key: "model".to_string(),
old_value: serde_json::json!("gpt-3.5-turbo"),
new_value: serde_json::json!("gpt-4"),
reason: "Testing".to_string(),
}];
engine.start_canary(request).await.unwrap();
for i in 0..100 {
let is_canary = i % 2 == 0;
engine
.update_metrics("test-deploy-3", is_canary, 50.0, true, Some(0.001))
.await
.unwrap();
}
let status = engine.get_status("test-deploy-3").await.unwrap();
assert_eq!(status.metrics.total_requests, 100);
assert_eq!(status.metrics.canary_requests, 50);
assert_eq!(status.metrics.control_requests, 50);
}
#[tokio::test]
async fn test_traffic_splitting_weighted_random() {
let config = CanaryConfig {
traffic_split_strategy: TrafficSplitStrategy::WeightedRandom,
..Default::default()
};
let mut engine = CanaryDeploymentEngine::new(config);
let mut request = create_test_request("test-deploy-4");
request.config_changes = vec![crate::decision::ConfigChange {
config_type: crate::decision::ConfigType::Provider,
key: "model".to_string(),
old_value: serde_json::json!("gpt-3.5-turbo"),
new_value: serde_json::json!("gpt-4"),
reason: "Testing".to_string(),
}];
engine.start_canary(request).await.unwrap();
let mut canary_count = 0;
let iterations = 1000;
for i in 0..iterations {
let route_to_canary = engine
.route_request(&format!("req-{}", i), 50.0)
.await
.unwrap();
if route_to_canary {
canary_count += 1;
}
}
let canary_percent = (canary_count as f64 / iterations as f64) * 100.0;
assert!(canary_percent > 40.0 && canary_percent < 60.0);
}
#[tokio::test]
async fn test_traffic_splitting_consistent_hash() {
let config = CanaryConfig {
traffic_split_strategy: TrafficSplitStrategy::ConsistentHash,
..Default::default()
};
let mut engine = CanaryDeploymentEngine::new(config);
let mut request = create_test_request("test-deploy-5");
request.config_changes = vec![crate::decision::ConfigChange {
config_type: crate::decision::ConfigType::Provider,
key: "model".to_string(),
old_value: serde_json::json!("gpt-3.5-turbo"),
new_value: serde_json::json!("gpt-4"),
reason: "Testing".to_string(),
}];
engine.start_canary(request).await.unwrap();
let request_id = "consistent-request-123";
let route1 = engine.route_request(request_id, 50.0).await.unwrap();
let route2 = engine.route_request(request_id, 50.0).await.unwrap();
let route3 = engine.route_request(request_id, 50.0).await.unwrap();
assert_eq!(route1, route2);
assert_eq!(route2, route3);
}
#[tokio::test]
async fn test_pause_and_resume() {
let config = CanaryConfig::default();
let mut engine = CanaryDeploymentEngine::new(config);
let mut request = create_test_request("test-deploy-6");
request.config_changes = vec![crate::decision::ConfigChange {
config_type: crate::decision::ConfigType::Provider,
key: "model".to_string(),
old_value: serde_json::json!("gpt-3.5-turbo"),
new_value: serde_json::json!("gpt-4"),
reason: "Testing".to_string(),
}];
engine.start_canary(request).await.unwrap();
engine.pause("test-deploy-6").await.unwrap();
assert!(engine.is_paused("test-deploy-6").await.unwrap());
let can_promote = engine.can_promote("test-deploy-6").await.unwrap();
assert!(!can_promote);
engine.resume("test-deploy-6").await.unwrap();
assert!(!engine.is_paused("test-deploy-6").await.unwrap());
}
#[tokio::test]
async fn test_validation_with_insufficient_samples() {
let config = CanaryConfig::default();
let mut engine = CanaryDeploymentEngine::new(config);
let mut request = create_test_request("test-deploy-7");
request.config_changes = vec![crate::decision::ConfigChange {
config_type: crate::decision::ConfigType::Provider,
key: "model".to_string(),
old_value: serde_json::json!("gpt-3.5-turbo"),
new_value: serde_json::json!("gpt-4"),
reason: "Testing".to_string(),
}];
engine.start_canary(request).await.unwrap();
for i in 0..10 {
engine
.update_metrics("test-deploy-7", true, 50.0, true, Some(0.001))
.await
.unwrap();
}
let can_promote = engine.can_promote("test-deploy-7").await.unwrap();
assert!(!can_promote);
}
#[tokio::test]
async fn test_validation_with_sufficient_samples() {
let mut config = CanaryConfig::default();
config.success_criteria.min_sample_size = 50;
config.phase_duration = StdDuration::from_millis(1);
let mut engine = CanaryDeploymentEngine::new(config);
let mut request = create_test_request("test-deploy-8");
request.config_changes = vec![crate::decision::ConfigChange {
config_type: crate::decision::ConfigType::Provider,
key: "model".to_string(),
old_value: serde_json::json!("gpt-3.5-turbo"),
new_value: serde_json::json!("gpt-4"),
reason: "Testing".to_string(),
}];
engine.start_canary(request).await.unwrap();
for _ in 0..100 {
engine
.update_metrics("test-deploy-8", true, 50.0, true, Some(0.001))
.await
.unwrap();
}
tokio::time::sleep(StdDuration::from_millis(10)).await;
let can_promote = engine.can_promote("test-deploy-8").await.unwrap();
assert!(can_promote);
}
#[tokio::test]
async fn test_high_error_rate_prevents_promotion() {
let mut config = CanaryConfig::default();
config.success_criteria.min_sample_size = 50;
config.success_criteria.max_error_rate = 5.0;
config.phase_duration = StdDuration::from_millis(1);
let mut engine = CanaryDeploymentEngine::new(config);
let mut request = create_test_request("test-deploy-9");
request.config_changes = vec![crate::decision::ConfigChange {
config_type: crate::decision::ConfigType::Provider,
key: "model".to_string(),
old_value: serde_json::json!("gpt-3.5-turbo"),
new_value: serde_json::json!("gpt-4"),
reason: "Testing".to_string(),
}];
engine.start_canary(request).await.unwrap();
for i in 0..100 {
let success = i % 10 != 0; engine
.update_metrics("test-deploy-9", true, 50.0, success, Some(0.001))
.await
.unwrap();
}
tokio::time::sleep(StdDuration::from_millis(10)).await;
let can_promote = engine.can_promote("test-deploy-9").await.unwrap();
assert!(!can_promote);
}
#[tokio::test]
async fn test_promote_through_phases() {
let mut config = CanaryConfig::default();
config.success_criteria.min_sample_size = 10;
config.phase_duration = StdDuration::from_millis(1);
let mut engine = CanaryDeploymentEngine::new(config);
let mut request = create_test_request("test-deploy-10");
request.config_changes = vec![crate::decision::ConfigChange {
config_type: crate::decision::ConfigType::Provider,
key: "model".to_string(),
old_value: serde_json::json!("gpt-3.5-turbo"),
new_value: serde_json::json!("gpt-4"),
reason: "Testing".to_string(),
}];
engine.start_canary(request).await.unwrap();
let status = engine.get_status("test-deploy-10").await.unwrap();
assert_eq!(status.current_phase, Some(0));
assert_eq!(status.traffic_percent, 1.0);
for _ in 0..20 {
engine
.update_metrics("test-deploy-10", true, 50.0, true, Some(0.001))
.await
.unwrap();
}
tokio::time::sleep(StdDuration::from_millis(10)).await;
let status = engine.promote("test-deploy-10").await.unwrap();
assert_eq!(status.current_phase, Some(1));
assert_eq!(status.traffic_percent, 5.0);
}
#[tokio::test]
async fn test_complete_deployment() {
let config = CanaryConfig::default();
let mut engine = CanaryDeploymentEngine::new(config);
let mut request = create_test_request("test-deploy-11");
request.config_changes = vec![crate::decision::ConfigChange {
config_type: crate::decision::ConfigType::Provider,
key: "model".to_string(),
old_value: serde_json::json!("gpt-3.5-turbo"),
new_value: serde_json::json!("gpt-4"),
reason: "Testing".to_string(),
}];
engine.start_canary(request).await.unwrap();
let status = engine.complete("test-deploy-11").await.unwrap();
assert_eq!(status.state, DeploymentState::Completed);
assert_eq!(status.traffic_percent, 100.0);
}
#[tokio::test]
async fn test_abort_deployment() {
let config = CanaryConfig::default();
let mut engine = CanaryDeploymentEngine::new(config);
let mut request = create_test_request("test-deploy-12");
request.config_changes = vec![crate::decision::ConfigChange {
config_type: crate::decision::ConfigType::Provider,
key: "model".to_string(),
old_value: serde_json::json!("gpt-3.5-turbo"),
new_value: serde_json::json!("gpt-4"),
reason: "Testing".to_string(),
}];
engine.start_canary(request).await.unwrap();
let result = engine.abort("test-deploy-12").await.unwrap();
assert!(result.success);
assert_eq!(result.deployment_id, "test-deploy-12");
let status = engine.get_status("test-deploy-12").await.unwrap();
assert_eq!(status.state, DeploymentState::RolledBack);
}
#[tokio::test]
async fn test_metrics_buffer_percentile_calculation() {
let mut buffer = MetricsBuffer::new(1000);
for i in 1..=100 {
buffer.add_canary_latency(i as f64);
}
let (_, _, canary_p95, _, canary_p99, _) = buffer.get_metrics();
assert!(canary_p95 >= 94.0 && canary_p95 <= 96.0);
assert!(canary_p99 >= 98.0 && canary_p99 <= 100.0);
}
#[tokio::test]
async fn test_statistical_analysis() {
let mut buffer = MetricsBuffer::new(1000);
for _ in 0..200 {
buffer.add_canary_latency(50.0);
buffer.add_control_latency(50.0);
buffer.record_canary_result(true);
buffer.record_control_result(true);
}
let config = CanaryConfig::default();
let engine = CanaryDeploymentEngine::new(config);
let analysis = engine.perform_statistical_analysis(&buffer);
assert!(!analysis.significant_difference);
assert_eq!(analysis.canary_sample_size, 200);
assert_eq!(analysis.control_sample_size, 200);
}
}