mockforge_chaos/

multi_cluster.rs

//! Multi-cluster orchestration support
//!
//! Execute chaos orchestrations across multiple Kubernetes clusters simultaneously.

use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
use std::collections::HashMap;

/// Multi-cluster orchestration configuration
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct MultiClusterOrchestration {
    pub name: String,
    pub description: Option<String>,
    pub clusters: Vec<ClusterTarget>,
    pub synchronization: SyncMode,
    pub orchestration: serde_json::Value,
    pub failover_policy: FailoverPolicy,
}

/// Target cluster configuration
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ClusterTarget {
    pub name: String,
    pub context: String,
    pub namespace: String,
    pub region: Option<String>,
    pub priority: u32,
    pub enabled: bool,
}

/// Synchronization mode for multi-cluster execution
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[serde(rename_all = "lowercase")]
pub enum SyncMode {
    Parallel,   // Execute on all clusters simultaneously
    Sequential, // Execute one cluster at a time
    Rolling,    // Rolling execution with configurable window
    Canary,     // Execute on canary cluster first, then others
}

/// Failover policy
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct FailoverPolicy {
    pub enabled: bool,
    pub max_failures: usize,
    pub continue_on_cluster_failure: bool,
}

/// Multi-cluster execution status
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct MultiClusterStatus {
    pub orchestration_name: String,
    pub start_time: DateTime<Utc>,
    pub end_time: Option<DateTime<Utc>>,
    pub overall_status: ExecutionStatus,
    pub cluster_statuses: HashMap<String, ClusterExecutionStatus>,
    pub total_clusters: usize,
    pub successful_clusters: usize,
    pub failed_clusters: usize,
}

/// Execution status
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[serde(rename_all = "lowercase")]
pub enum ExecutionStatus {
    Pending,
    Running,
    Completed,
    Failed,
    PartialSuccess,
}

/// Status for individual cluster
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ClusterExecutionStatus {
    pub cluster_name: String,
    pub status: ExecutionStatus,
    pub start_time: Option<DateTime<Utc>>,
    pub end_time: Option<DateTime<Utc>>,
    pub progress: f64,
    pub error: Option<String>,
    pub metrics: ClusterMetrics,
}

/// Metrics for cluster execution
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
pub struct ClusterMetrics {
    pub steps_completed: usize,
    pub steps_total: usize,
    pub failures: usize,
    pub avg_latency_ms: f64,
    pub error_rate: f64,
}

/// Multi-cluster orchestrator
pub struct MultiClusterOrchestrator {
    orchestrations: HashMap<String, MultiClusterOrchestration>,
    statuses: HashMap<String, MultiClusterStatus>,
}

impl MultiClusterOrchestrator {
    /// Create a new multi-cluster orchestrator
    pub fn new() -> Self {
        Self {
            orchestrations: HashMap::new(),
            statuses: HashMap::new(),
        }
    }

    /// Register a multi-cluster orchestration
    pub fn register(&mut self, orchestration: MultiClusterOrchestration) {
        self.orchestrations.insert(orchestration.name.clone(), orchestration);
    }

    /// Execute a multi-cluster orchestration
    pub async fn execute(&mut self, name: &str) -> Result<MultiClusterStatus, String> {
        let orchestration = self
            .orchestrations
            .get(name)
            .ok_or_else(|| format!("Orchestration '{}' not found", name))?
            .clone();

        let start_time = Utc::now();
        let mut cluster_statuses = HashMap::new();

        // Initialize cluster statuses
        for cluster in &orchestration.clusters {
            if cluster.enabled {
                cluster_statuses.insert(
                    cluster.name.clone(),
                    ClusterExecutionStatus {
                        cluster_name: cluster.name.clone(),
                        status: ExecutionStatus::Pending,
                        start_time: None,
                        end_time: None,
                        progress: 0.0,
                        error: None,
                        metrics: ClusterMetrics::default(),
                    },
                );
            }
        }

        let mut status = MultiClusterStatus {
            orchestration_name: name.to_string(),
            start_time,
            end_time: None,
            overall_status: ExecutionStatus::Running,
            cluster_statuses: cluster_statuses.clone(),
            total_clusters: orchestration.clusters.iter().filter(|c| c.enabled).count(),
            successful_clusters: 0,
            failed_clusters: 0,
        };

        // Execute based on synchronization mode
        match orchestration.synchronization {
            SyncMode::Parallel => {
                self.execute_parallel(&orchestration, &mut status).await?;
            }
            SyncMode::Sequential => {
                self.execute_sequential(&orchestration, &mut status).await?;
            }
            SyncMode::Rolling => {
                self.execute_rolling(&orchestration, &mut status).await?;
            }
            SyncMode::Canary => {
                self.execute_canary(&orchestration, &mut status).await?;
            }
        }

        status.end_time = Some(Utc::now());

        // Determine overall status
        if status.failed_clusters == 0 {
            status.overall_status = ExecutionStatus::Completed;
        } else if status.successful_clusters > 0 {
            status.overall_status = ExecutionStatus::PartialSuccess;
        } else {
            status.overall_status = ExecutionStatus::Failed;
        }

        self.statuses.insert(name.to_string(), status.clone());

        Ok(status)
    }

    /// Execute on all clusters in parallel
    async fn execute_parallel(
        &self,
        orchestration: &MultiClusterOrchestration,
        status: &mut MultiClusterStatus,
    ) -> Result<(), String> {
        // In real implementation, spawn tasks for each cluster
        for cluster in &orchestration.clusters {
            if !cluster.enabled {
                continue;
            }

            match self.execute_on_cluster(cluster, &orchestration.orchestration).await {
                Ok(cluster_status) => {
                    status.cluster_statuses.insert(cluster.name.clone(), cluster_status);
                    status.successful_clusters += 1;
                }
                Err(e) => {
                    if let Some(cluster_status) = status.cluster_statuses.get_mut(&cluster.name) {
                        cluster_status.status = ExecutionStatus::Failed;
                        cluster_status.error = Some(e.clone());
                    }
                    status.failed_clusters += 1;

                    if !orchestration.failover_policy.continue_on_cluster_failure {
                        return Err(format!("Cluster {} failed: {}", cluster.name, e));
                    }
                }
            }
        }

        Ok(())
    }

    /// Execute on clusters sequentially
    async fn execute_sequential(
        &self,
        orchestration: &MultiClusterOrchestration,
        status: &mut MultiClusterStatus,
    ) -> Result<(), String> {
        for cluster in &orchestration.clusters {
            if !cluster.enabled {
                continue;
            }

            match self.execute_on_cluster(cluster, &orchestration.orchestration).await {
                Ok(cluster_status) => {
                    status.cluster_statuses.insert(cluster.name.clone(), cluster_status);
                    status.successful_clusters += 1;
                }
                Err(e) => {
                    if let Some(cluster_status) = status.cluster_statuses.get_mut(&cluster.name) {
                        cluster_status.status = ExecutionStatus::Failed;
                        cluster_status.error = Some(e.clone());
                    }
                    status.failed_clusters += 1;

                    if !orchestration.failover_policy.continue_on_cluster_failure {
                        return Err(format!("Cluster {} failed: {}", cluster.name, e));
                    }
                }
            }
        }

        Ok(())
    }

    /// Execute with rolling deployment
    async fn execute_rolling(
        &self,
        orchestration: &MultiClusterOrchestration,
        status: &mut MultiClusterStatus,
    ) -> Result<(), String> {
        // Rolling execution with window size of 1-2 clusters at a time
        let window_size = 2;
        let mut enabled_clusters: Vec<_> =
            orchestration.clusters.iter().filter(|c| c.enabled).collect();

        enabled_clusters.sort_by_key(|c| c.priority);

        for window in enabled_clusters.chunks(window_size) {
            for cluster in window {
                match self.execute_on_cluster(cluster, &orchestration.orchestration).await {
                    Ok(cluster_status) => {
                        status.cluster_statuses.insert(cluster.name.clone(), cluster_status);
                        status.successful_clusters += 1;
                    }
                    Err(e) => {
                        if let Some(cluster_status) = status.cluster_statuses.get_mut(&cluster.name)
                        {
                            cluster_status.status = ExecutionStatus::Failed;
                            cluster_status.error = Some(e.clone());
                        }
                        status.failed_clusters += 1;

                        if !orchestration.failover_policy.continue_on_cluster_failure {
                            return Err(format!("Cluster {} failed: {}", cluster.name, e));
                        }
                    }
                }
            }

            // Small delay between windows
            tokio::time::sleep(tokio::time::Duration::from_secs(5)).await;
        }

        Ok(())
    }

    /// Execute with canary strategy
    async fn execute_canary(
        &self,
        orchestration: &MultiClusterOrchestration,
        status: &mut MultiClusterStatus,
    ) -> Result<(), String> {
        // Find canary cluster (highest priority)
        let mut enabled_clusters: Vec<_> =
            orchestration.clusters.iter().filter(|c| c.enabled).collect();

        enabled_clusters.sort_by_key(|c| std::cmp::Reverse(c.priority));

        if enabled_clusters.is_empty() {
            return Err("No enabled clusters".to_string());
        }

        // Execute on canary first
        let canary = enabled_clusters[0];
        match self.execute_on_cluster(canary, &orchestration.orchestration).await {
            Ok(cluster_status) => {
                status.cluster_statuses.insert(canary.name.clone(), cluster_status);
                status.successful_clusters += 1;
            }
            Err(e) => {
                return Err(format!("Canary cluster {} failed: {}", canary.name, e));
            }
        }

        // If canary succeeded, execute on remaining clusters
        for cluster in &enabled_clusters[1..] {
            match self.execute_on_cluster(cluster, &orchestration.orchestration).await {
                Ok(cluster_status) => {
                    status.cluster_statuses.insert(cluster.name.clone(), cluster_status);
                    status.successful_clusters += 1;
                }
                Err(e) => {
                    if let Some(cluster_status) = status.cluster_statuses.get_mut(&cluster.name) {
                        cluster_status.status = ExecutionStatus::Failed;
                        cluster_status.error = Some(e.clone());
                    }
                    status.failed_clusters += 1;

                    if !orchestration.failover_policy.continue_on_cluster_failure {
                        return Err(format!("Cluster {} failed: {}", cluster.name, e));
                    }
                }
            }
        }

        Ok(())
    }

    /// Execute orchestration on a single cluster
    async fn execute_on_cluster(
        &self,
        cluster: &ClusterTarget,
        _orchestration: &serde_json::Value,
    ) -> Result<ClusterExecutionStatus, String> {
        // In real implementation:
        // 1. Connect to cluster using context
        // 2. Deploy orchestration
        // 3. Monitor execution
        // 4. Collect metrics

        // Simulated execution
        Ok(ClusterExecutionStatus {
            cluster_name: cluster.name.clone(),
            status: ExecutionStatus::Completed,
            start_time: Some(Utc::now()),
            end_time: Some(Utc::now()),
            progress: 1.0,
            error: None,
            metrics: ClusterMetrics {
                steps_completed: 5,
                steps_total: 5,
                failures: 0,
                avg_latency_ms: 125.0,
                error_rate: 0.0,
            },
        })
    }

    /// Get status of a multi-cluster orchestration
    pub fn get_status(&self, name: &str) -> Option<&MultiClusterStatus> {
        self.statuses.get(name)
    }

    /// List all registered orchestrations
    pub fn list_orchestrations(&self) -> Vec<String> {
        self.orchestrations.keys().cloned().collect()
    }
}

impl Default for MultiClusterOrchestrator {
    fn default() -> Self {
        Self::new()
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_multi_cluster_orchestrator_creation() {
        let orchestrator = MultiClusterOrchestrator::new();
        assert_eq!(orchestrator.list_orchestrations().len(), 0);
    }

    #[test]
    fn test_register_orchestration() {
        let mut orchestrator = MultiClusterOrchestrator::new();

        let orch = MultiClusterOrchestration {
            name: "test-orch".to_string(),
            description: Some("Test".to_string()),
            clusters: vec![ClusterTarget {
                name: "cluster-1".to_string(),
                context: "kind-cluster-1".to_string(),
                namespace: "default".to_string(),
                region: Some("us-east-1".to_string()),
                priority: 1,
                enabled: true,
            }],
            synchronization: SyncMode::Parallel,
            orchestration: serde_json::json!({}),
            failover_policy: FailoverPolicy {
                enabled: true,
                max_failures: 1,
                continue_on_cluster_failure: true,
            },
        };

        orchestrator.register(orch);
        assert_eq!(orchestrator.list_orchestrations().len(), 1);
    }
}