758 lines
25 KiB
Rust
758 lines
25 KiB
Rust
//! Multi-region orchestration module for RustyTorch++ Platform
|
|
//! Handles region management, health monitoring, failover, and cross-region coordination
|
|
|
|
use crate::error::RegionError;
|
|
use crate::{PlatformError, PlatformResult};
|
|
use dashmap::DashMap;
|
|
use std::collections::HashMap;
|
|
use std::sync::{
|
|
Arc,
|
|
atomic::{AtomicBool, Ordering},
|
|
};
|
|
use std::time::{Duration, Instant};
|
|
use tokio::sync::{RwLock as AsyncRwLock, broadcast};
|
|
use uuid::Uuid;
|
|
|
|
/// Regional configuration
|
|
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
|
|
pub struct RegionConfig {
|
|
pub id: String,
|
|
pub name: String,
|
|
pub endpoint: String,
|
|
pub capacity_limits: HashMap<String, u64>,
|
|
pub availability_zone_count: u32,
|
|
pub latency_targets_ms: HashMap<String, u64>,
|
|
}
|
|
|
|
/// Region status enumeration
|
|
#[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)]
|
|
pub enum RegionStatus {
|
|
Available,
|
|
Degraded,
|
|
Unavailable,
|
|
Maintenance,
|
|
}
|
|
|
|
/// Region health information
|
|
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
|
|
pub enum RegionHealth {
|
|
Healthy {
|
|
uptime: Duration,
|
|
cpu_usage: f64,
|
|
memory_usage: f64,
|
|
#[serde(skip, default = "std::time::Instant::now")]
|
|
last_check: std::time::Instant,
|
|
},
|
|
Degraded {
|
|
issues: Vec<String>,
|
|
#[serde(skip, default = "std::time::Instant::now")]
|
|
last_check: std::time::Instant,
|
|
},
|
|
Unhealthy {
|
|
errors: Vec<String>,
|
|
#[serde(skip, default = "std::time::Instant::now")]
|
|
last_check: std::time::Instant,
|
|
},
|
|
}
|
|
|
|
/// Cross-region operation definition
|
|
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
|
|
pub struct CrossRegionOp {
|
|
pub id: Uuid,
|
|
pub tenant_id: Uuid,
|
|
pub source_region: String,
|
|
pub target_regions: Vec<String>,
|
|
pub operation_type: String,
|
|
pub data_size_bytes: u64,
|
|
pub priority: u32,
|
|
pub timeout_ms: u64,
|
|
}
|
|
|
|
/// Region capacity information
|
|
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
|
|
pub struct RegionCapacity {
|
|
pub region_id: String,
|
|
pub total_gpu: u64,
|
|
pub available_gpu: u64,
|
|
pub reserved_gpu: u64,
|
|
pub used_gpu: u64,
|
|
pub total_memory_gb: u64,
|
|
pub available_memory_gb: u64,
|
|
pub reserved_memory_gb: u64,
|
|
pub used_memory_gb: u64,
|
|
}
|
|
|
|
/// Operation scheduling request
|
|
#[derive(Debug, Clone)]
|
|
pub struct OperationRequest {
|
|
pub tenant_id: Uuid,
|
|
pub preferred_region: String,
|
|
pub operation_type: String,
|
|
pub data_size_bytes: u64,
|
|
pub resource_requirements: HashMap<String, u64>,
|
|
}
|
|
|
|
/// Operation assignment result
|
|
#[derive(Debug, Clone)]
|
|
pub struct OperationAssignment {
|
|
pub operation_id: Uuid,
|
|
pub tenant_id: Uuid,
|
|
pub assigned_region: String,
|
|
pub is_failover: bool,
|
|
pub estimated_start_time: chrono::DateTime<chrono::Utc>,
|
|
}
|
|
|
|
/// Cross-region coordination result
|
|
#[derive(Debug, Clone)]
|
|
pub struct CrossRegionResult {
|
|
pub operation_id: Uuid,
|
|
pub participating_regions: Vec<String>,
|
|
pub latency_ms: u64,
|
|
pub success: bool,
|
|
}
|
|
|
|
/// Latency-sensitive request
|
|
#[derive(Debug, Clone)]
|
|
pub struct LatencySensitiveRequest {
|
|
pub tenant_id: Uuid,
|
|
pub required_regions: Vec<String>,
|
|
pub max_latency_ms: u64,
|
|
pub operation_type: String,
|
|
pub data_size_bytes: u64,
|
|
}
|
|
|
|
/// Latency-optimized assignment
|
|
#[derive(Debug, Clone)]
|
|
pub struct LatencyOptimizedAssignment {
|
|
pub operation_id: Uuid,
|
|
pub participating_regions: Vec<String>,
|
|
pub max_inter_region_latency_ms: u64,
|
|
pub latency_matrix: HashMap<String, HashMap<String, f64>>,
|
|
}
|
|
|
|
/// Data locality request
|
|
#[derive(Debug, Clone)]
|
|
pub struct DataLocalityRequest {
|
|
pub tenant_id: Uuid,
|
|
pub dataset_id: String,
|
|
pub operation_type: String,
|
|
pub resource_types: Vec<String>,
|
|
pub resource_quantity: u64,
|
|
}
|
|
|
|
/// Data locality optimized assignment
|
|
#[derive(Debug, Clone)]
|
|
pub struct DataLocalityAssignment {
|
|
pub operation_id: Uuid,
|
|
pub assigned_region: String,
|
|
pub data_transfer_required: bool,
|
|
pub estimated_data_transfer_time_ms: u64,
|
|
}
|
|
|
|
/// Capacity reservation
|
|
#[derive(Debug, Clone)]
|
|
pub struct CapacityReservation {
|
|
pub reservation_id: Uuid,
|
|
pub tenant_id: Uuid,
|
|
pub region_id: String,
|
|
pub resources: HashMap<String, u64>,
|
|
pub expires_at: chrono::DateTime<chrono::Utc>,
|
|
}
|
|
|
|
/// Region manager for multi-region orchestration
|
|
#[derive(Debug)]
|
|
pub struct RegionManager {
|
|
regions: Arc<DashMap<String, RegionConfig>>,
|
|
region_status: Arc<DashMap<String, RegionStatus>>,
|
|
region_health: Arc<DashMap<String, RegionHealth>>,
|
|
region_capacity: Arc<DashMap<String, RegionCapacity>>,
|
|
reservations: Arc<DashMap<Uuid, CapacityReservation>>,
|
|
data_locality: Arc<DashMap<String, HashMap<Uuid, Vec<String>>>>, // dataset_id -> tenant_id -> regions
|
|
latency_matrix: Arc<AsyncRwLock<HashMap<String, HashMap<String, f64>>>>,
|
|
health_monitor_running: Arc<AtomicBool>,
|
|
shutdown_tx: Option<broadcast::Sender<()>>,
|
|
client: reqwest::Client,
|
|
}
|
|
|
|
impl RegionManager {
|
|
/// Create new RegionManager
|
|
pub async fn new(config: &crate::PlatformConfig) -> PlatformResult<Self> {
|
|
let regions = Arc::new(DashMap::new());
|
|
let region_status = Arc::new(DashMap::new());
|
|
let region_health = Arc::new(DashMap::new());
|
|
let region_capacity = Arc::new(DashMap::new());
|
|
let reservations = Arc::new(DashMap::new());
|
|
let data_locality = Arc::new(DashMap::new());
|
|
let latency_matrix = Arc::new(AsyncRwLock::new(HashMap::new()));
|
|
let health_monitor_running = Arc::new(AtomicBool::new(false));
|
|
let client = reqwest::Client::new();
|
|
|
|
// Initialize regions
|
|
for (region_id, region_config) in &config.regions {
|
|
regions.insert(region_id.clone(), region_config.clone());
|
|
region_status.insert(region_id.clone(), RegionStatus::Available);
|
|
|
|
// Initialize capacity based on config
|
|
let capacity = RegionCapacity {
|
|
region_id: region_id.clone(),
|
|
total_gpu: region_config
|
|
.capacity_limits
|
|
.get("gpu")
|
|
.copied()
|
|
.unwrap_or(0),
|
|
available_gpu: region_config
|
|
.capacity_limits
|
|
.get("gpu")
|
|
.copied()
|
|
.unwrap_or(0),
|
|
reserved_gpu: 0,
|
|
used_gpu: 0,
|
|
total_memory_gb: region_config
|
|
.capacity_limits
|
|
.get("memory_gb")
|
|
.copied()
|
|
.unwrap_or(0),
|
|
available_memory_gb: region_config
|
|
.capacity_limits
|
|
.get("memory_gb")
|
|
.copied()
|
|
.unwrap_or(0),
|
|
reserved_memory_gb: 0,
|
|
used_memory_gb: 0,
|
|
};
|
|
region_capacity.insert(region_id.clone(), capacity);
|
|
|
|
// Initialize health as healthy
|
|
region_health.insert(
|
|
region_id.clone(),
|
|
RegionHealth::Healthy {
|
|
uptime: Duration::from_secs(0),
|
|
cpu_usage: 0.1,
|
|
memory_usage: 0.1,
|
|
last_check: Instant::now(),
|
|
},
|
|
);
|
|
}
|
|
|
|
// Initialize latency matrix from config
|
|
let mut matrix = HashMap::new();
|
|
for (source_id, source_config) in &config.regions {
|
|
let mut targets = HashMap::new();
|
|
for (target_id, latency) in &source_config.latency_targets_ms {
|
|
targets.insert(target_id.clone(), *latency as f64);
|
|
}
|
|
// Self-latency is 0
|
|
targets.insert(source_id.clone(), 0.0);
|
|
matrix.insert(source_id.clone(), targets);
|
|
}
|
|
*latency_matrix.write().await = matrix;
|
|
|
|
Ok(Self {
|
|
regions,
|
|
region_status,
|
|
region_health,
|
|
region_capacity,
|
|
reservations,
|
|
data_locality,
|
|
latency_matrix,
|
|
health_monitor_running,
|
|
shutdown_tx: None,
|
|
client,
|
|
})
|
|
}
|
|
|
|
/// Start region manager services
|
|
pub async fn start(&mut self) -> PlatformResult<()> {
|
|
self.start_health_monitoring().await?;
|
|
tracing::info!("RegionManager started");
|
|
Ok(())
|
|
}
|
|
|
|
/// Shutdown region manager
|
|
pub async fn shutdown(&mut self) -> PlatformResult<()> {
|
|
self.health_monitor_running.store(false, Ordering::Relaxed);
|
|
if let Some(tx) = self.shutdown_tx.take() {
|
|
let _ = tx.send(());
|
|
}
|
|
tracing::info!("RegionManager shutdown");
|
|
Ok(())
|
|
}
|
|
|
|
/// Get region count
|
|
pub async fn get_region_count(&self) -> usize {
|
|
self.regions.len()
|
|
}
|
|
|
|
/// Get region configuration
|
|
pub async fn get_region(&self, region_id: &str) -> PlatformResult<RegionConfig> {
|
|
self.regions
|
|
.get(region_id)
|
|
.map(|r| r.clone())
|
|
.ok_or_else(|| {
|
|
PlatformError::Region(RegionError::NotFound {
|
|
region_id: region_id.to_string(),
|
|
})
|
|
})
|
|
}
|
|
|
|
/// Start health monitoring
|
|
pub async fn start_health_monitoring(&mut self) -> PlatformResult<()> {
|
|
if self.health_monitor_running.load(Ordering::Relaxed) {
|
|
return Ok(());
|
|
}
|
|
|
|
self.health_monitor_running.store(true, Ordering::Relaxed);
|
|
let (shutdown_tx, mut shutdown_rx) = broadcast::channel(1);
|
|
self.shutdown_tx = Some(shutdown_tx);
|
|
|
|
let region_health = self.region_health.clone();
|
|
let region_status = self.region_status.clone();
|
|
let regions = self.regions.clone();
|
|
let health_running = self.health_monitor_running.clone();
|
|
let client = self.client.clone();
|
|
|
|
tokio::spawn(async move {
|
|
let mut interval = tokio::time::interval(Duration::from_secs(10));
|
|
|
|
while health_running.load(Ordering::Relaxed) {
|
|
tokio::select! {
|
|
_ = interval.tick() => {
|
|
// Perform health checks on all regions
|
|
for region_entry in regions.iter() {
|
|
let region_id = region_entry.key();
|
|
let region_config = region_entry.value();
|
|
|
|
if let Ok(health) = Self::check_region_health(&client, region_config).await {
|
|
region_health.insert(region_id.clone(), health);
|
|
region_status.insert(region_id.clone(), RegionStatus::Available);
|
|
} else {
|
|
region_health.insert(region_id.clone(), RegionHealth::Unhealthy {
|
|
errors: vec!["Health check failed".to_string()],
|
|
last_check: Instant::now(),
|
|
});
|
|
region_status.insert(region_id.clone(), RegionStatus::Unavailable);
|
|
}
|
|
}
|
|
}
|
|
_ = shutdown_rx.recv() => {
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
});
|
|
|
|
Ok(())
|
|
}
|
|
|
|
/// Check health of a specific region
|
|
async fn check_region_health(
|
|
client: &reqwest::Client,
|
|
region_config: &RegionConfig,
|
|
) -> PlatformResult<RegionHealth> {
|
|
// Simulate health check - in real implementation would call region endpoint
|
|
let _response = client
|
|
.get(format!("{}/health", region_config.endpoint))
|
|
.timeout(Duration::from_secs(5))
|
|
.send()
|
|
.await
|
|
.map_err(PlatformError::Network)?;
|
|
|
|
Ok(RegionHealth::Healthy {
|
|
uptime: Duration::from_secs(3600), // 1 hour uptime
|
|
cpu_usage: 0.3,
|
|
memory_usage: 0.4,
|
|
last_check: Instant::now(),
|
|
})
|
|
}
|
|
|
|
/// Get region health
|
|
pub async fn get_region_health(&self, region_id: &str) -> PlatformResult<RegionHealth> {
|
|
self.region_health
|
|
.get(region_id)
|
|
.map(|h| h.clone())
|
|
.ok_or_else(|| {
|
|
PlatformError::Region(RegionError::NotFound {
|
|
region_id: region_id.to_string(),
|
|
})
|
|
})
|
|
}
|
|
|
|
/// Coordinate cross-region operation
|
|
pub async fn coordinate_cross_region_operation(
|
|
&self,
|
|
operation: CrossRegionOp,
|
|
) -> PlatformResult<CrossRegionResult> {
|
|
let start_time = Instant::now();
|
|
|
|
// Verify all target regions are available
|
|
for region_id in &operation.target_regions {
|
|
if !self.regions.contains_key(region_id) {
|
|
return Err(PlatformError::Region(RegionError::NotFound {
|
|
region_id: region_id.clone(),
|
|
}));
|
|
}
|
|
|
|
let status = self
|
|
.region_status
|
|
.get(region_id)
|
|
.map_or(RegionStatus::Unavailable, |s| s.clone());
|
|
|
|
if status != RegionStatus::Available {
|
|
return Err(PlatformError::Region(RegionError::Unavailable {
|
|
region_id: region_id.clone(),
|
|
}));
|
|
}
|
|
}
|
|
|
|
// Calculate maximum latency between participating regions
|
|
let latency_matrix = self.latency_matrix.read().await;
|
|
let mut max_latency = 0u64;
|
|
|
|
let all_regions = vec![operation.source_region.clone()]
|
|
.into_iter()
|
|
.chain(operation.target_regions.iter().cloned())
|
|
.collect::<Vec<_>>();
|
|
|
|
for source in &all_regions {
|
|
for target in &all_regions {
|
|
if source != target
|
|
&& let Some(source_latencies) = latency_matrix.get(source)
|
|
&& let Some(&latency) = source_latencies.get(target)
|
|
{
|
|
max_latency = max_latency.max(latency as u64);
|
|
}
|
|
}
|
|
}
|
|
|
|
let coordination_latency = start_time.elapsed().as_millis() as u64 + max_latency;
|
|
|
|
Ok(CrossRegionResult {
|
|
operation_id: operation.id,
|
|
participating_regions: all_regions,
|
|
latency_ms: coordination_latency,
|
|
success: true,
|
|
})
|
|
}
|
|
|
|
/// Create operation request
|
|
pub fn create_operation_request(
|
|
&self,
|
|
tenant_id: Uuid,
|
|
preferred_region: &str,
|
|
operation_type: &str,
|
|
data_size_bytes: u64,
|
|
) -> OperationRequest {
|
|
OperationRequest {
|
|
tenant_id,
|
|
preferred_region: preferred_region.to_string(),
|
|
operation_type: operation_type.to_string(),
|
|
data_size_bytes,
|
|
resource_requirements: HashMap::new(),
|
|
}
|
|
}
|
|
|
|
/// Schedule operation with failover support
|
|
pub async fn schedule_operation(
|
|
&self,
|
|
request: OperationRequest,
|
|
) -> PlatformResult<OperationAssignment> {
|
|
let operation_id = Uuid::new_v4();
|
|
|
|
// Check if preferred region is available
|
|
let preferred_status = self
|
|
.region_status
|
|
.get(&request.preferred_region)
|
|
.map_or(RegionStatus::Unavailable, |s| s.clone());
|
|
|
|
let (assigned_region, is_failover) = if preferred_status == RegionStatus::Available {
|
|
(request.preferred_region.clone(), false)
|
|
} else {
|
|
// Find alternative region
|
|
let alternative = self
|
|
.find_alternative_region(&request.preferred_region)
|
|
.await?;
|
|
(alternative, true)
|
|
};
|
|
|
|
Ok(OperationAssignment {
|
|
operation_id,
|
|
tenant_id: request.tenant_id,
|
|
assigned_region,
|
|
is_failover,
|
|
estimated_start_time: chrono::Utc::now(),
|
|
})
|
|
}
|
|
|
|
/// Mark region as unavailable
|
|
pub async fn mark_region_unavailable(
|
|
&self,
|
|
region_id: &str,
|
|
reason: &str,
|
|
) -> PlatformResult<()> {
|
|
if !self.regions.contains_key(region_id) {
|
|
return Err(PlatformError::Region(RegionError::NotFound {
|
|
region_id: region_id.to_string(),
|
|
}));
|
|
}
|
|
|
|
self.region_status
|
|
.insert(region_id.to_string(), RegionStatus::Unavailable);
|
|
self.region_health.insert(
|
|
region_id.to_string(),
|
|
RegionHealth::Unhealthy {
|
|
errors: vec![reason.to_string()],
|
|
last_check: Instant::now(),
|
|
},
|
|
);
|
|
|
|
tracing::warn!("Region {} marked unavailable: {}", region_id, reason);
|
|
Ok(())
|
|
}
|
|
|
|
/// Find alternative region for failover
|
|
async fn find_alternative_region(&self, failed_region: &str) -> PlatformResult<String> {
|
|
// Find first available region that's not the failed one
|
|
for region_entry in self.region_status.iter() {
|
|
let region_id = region_entry.key();
|
|
let status = region_entry.value();
|
|
|
|
if region_id != failed_region && *status == RegionStatus::Available {
|
|
return Ok(region_id.clone());
|
|
}
|
|
}
|
|
|
|
Err(PlatformError::Region(RegionError::Unavailable {
|
|
region_id: "no_alternatives".to_string(),
|
|
}))
|
|
}
|
|
|
|
/// Get region capacity
|
|
pub async fn get_region_capacity(&self, region_id: &str) -> PlatformResult<RegionCapacity> {
|
|
self.region_capacity
|
|
.get(region_id)
|
|
.map(|c| c.clone())
|
|
.ok_or_else(|| {
|
|
PlatformError::Region(RegionError::NotFound {
|
|
region_id: region_id.to_string(),
|
|
})
|
|
})
|
|
}
|
|
|
|
/// Reserve capacity in a region
|
|
pub async fn reserve_capacity(
|
|
&self,
|
|
region_id: &str,
|
|
tenant_id: Uuid,
|
|
resources: HashMap<String, u64>,
|
|
) -> PlatformResult<CapacityReservation> {
|
|
let mut capacity = self.region_capacity.get_mut(region_id).ok_or_else(|| {
|
|
PlatformError::Region(RegionError::NotFound {
|
|
region_id: region_id.to_string(),
|
|
})
|
|
})?;
|
|
|
|
// Check if resources are available
|
|
for (resource_type, requested) in &resources {
|
|
match resource_type.as_str() {
|
|
"gpu" => {
|
|
if capacity.available_gpu < *requested {
|
|
return Err(PlatformError::Region(RegionError::CapacityExhausted {
|
|
region_id: region_id.to_string(),
|
|
}));
|
|
}
|
|
}
|
|
"memory_gb" => {
|
|
if capacity.available_memory_gb < *requested {
|
|
return Err(PlatformError::Region(RegionError::CapacityExhausted {
|
|
region_id: region_id.to_string(),
|
|
}));
|
|
}
|
|
}
|
|
_ => {} // Ignore unknown resource types for now
|
|
}
|
|
}
|
|
|
|
// Reserve the resources
|
|
for (resource_type, requested) in &resources {
|
|
match resource_type.as_str() {
|
|
"gpu" => {
|
|
capacity.available_gpu -= requested;
|
|
capacity.reserved_gpu += requested;
|
|
}
|
|
"memory_gb" => {
|
|
capacity.available_memory_gb -= requested;
|
|
capacity.reserved_memory_gb += requested;
|
|
}
|
|
_ => {}
|
|
}
|
|
}
|
|
|
|
let reservation = CapacityReservation {
|
|
reservation_id: Uuid::new_v4(),
|
|
tenant_id,
|
|
region_id: region_id.to_string(),
|
|
resources: resources.clone(),
|
|
expires_at: chrono::Utc::now() + chrono::Duration::hours(1),
|
|
};
|
|
|
|
self.reservations
|
|
.insert(reservation.reservation_id, reservation.clone());
|
|
Ok(reservation)
|
|
}
|
|
|
|
/// Create latency-sensitive request
|
|
pub fn create_latency_sensitive_request(
|
|
&self,
|
|
tenant_id: Uuid,
|
|
required_regions: Vec<String>,
|
|
max_latency_ms: u64,
|
|
operation_type: &str,
|
|
data_size_bytes: u64,
|
|
) -> LatencySensitiveRequest {
|
|
LatencySensitiveRequest {
|
|
tenant_id,
|
|
required_regions,
|
|
max_latency_ms,
|
|
operation_type: operation_type.to_string(),
|
|
data_size_bytes,
|
|
}
|
|
}
|
|
|
|
/// Optimize operation for latency
|
|
pub async fn optimize_for_latency(
|
|
&self,
|
|
request: LatencySensitiveRequest,
|
|
) -> PlatformResult<LatencyOptimizedAssignment> {
|
|
let operation_id = Uuid::new_v4();
|
|
let latency_matrix = self.latency_matrix.read().await;
|
|
|
|
// Verify latency constraints can be met
|
|
let mut max_latency: f64 = 0.0;
|
|
let mut assignment_matrix = HashMap::new();
|
|
|
|
for source in &request.required_regions {
|
|
let mut target_latencies = HashMap::new();
|
|
|
|
for target in &request.required_regions {
|
|
if source != target {
|
|
let latency = latency_matrix
|
|
.get(source)
|
|
.and_then(|targets| targets.get(target))
|
|
.copied()
|
|
.unwrap_or(1000.0); // Default high latency if unknown
|
|
|
|
target_latencies.insert(target.clone(), latency);
|
|
max_latency = max_latency.max(latency);
|
|
}
|
|
}
|
|
assignment_matrix.insert(source.clone(), target_latencies);
|
|
}
|
|
|
|
if max_latency > request.max_latency_ms as f64 {
|
|
return Err(PlatformError::Region(RegionError::CrossRegionFailed {
|
|
src: "optimization".to_string(),
|
|
dst: "latency_constraint_violation".to_string(),
|
|
}));
|
|
}
|
|
|
|
Ok(LatencyOptimizedAssignment {
|
|
operation_id,
|
|
participating_regions: request.required_regions,
|
|
max_inter_region_latency_ms: max_latency as u64,
|
|
latency_matrix: assignment_matrix,
|
|
})
|
|
}
|
|
|
|
/// Register data locality information
|
|
pub async fn register_data_locality(
|
|
&self,
|
|
tenant_id: Uuid,
|
|
dataset_id: String,
|
|
regions: Vec<String>,
|
|
_data_size_bytes: u64,
|
|
) -> PlatformResult<()> {
|
|
let mut tenant_data = HashMap::new();
|
|
tenant_data.insert(tenant_id, regions);
|
|
self.data_locality.insert(dataset_id, tenant_data);
|
|
Ok(())
|
|
}
|
|
|
|
/// Create data locality request
|
|
pub fn create_data_locality_request(
|
|
&self,
|
|
tenant_id: Uuid,
|
|
dataset_id: String,
|
|
operation_type: &str,
|
|
resource_types: Vec<String>,
|
|
resource_quantity: u64,
|
|
) -> DataLocalityRequest {
|
|
DataLocalityRequest {
|
|
tenant_id,
|
|
dataset_id,
|
|
operation_type: operation_type.to_string(),
|
|
resource_types,
|
|
resource_quantity,
|
|
}
|
|
}
|
|
|
|
/// Optimize operation for data locality
|
|
pub async fn optimize_for_data_locality(
|
|
&self,
|
|
request: DataLocalityRequest,
|
|
) -> PlatformResult<DataLocalityAssignment> {
|
|
let operation_id = Uuid::new_v4();
|
|
|
|
// Find regions where data is available
|
|
let data_regions = if let Some(tenant_data) = self.data_locality.get(&request.dataset_id) {
|
|
tenant_data
|
|
.get(&request.tenant_id)
|
|
.cloned()
|
|
.unwrap_or_else(Vec::new)
|
|
} else {
|
|
Vec::new()
|
|
};
|
|
|
|
if data_regions.is_empty() {
|
|
// Data not found, assign to first available region
|
|
let first_available = self
|
|
.region_status
|
|
.iter()
|
|
.find(|entry| *entry.value() == RegionStatus::Available)
|
|
.map(|entry| entry.key().clone())
|
|
.ok_or_else(|| {
|
|
PlatformError::Region(RegionError::Unavailable {
|
|
region_id: "no_available_regions".to_string(),
|
|
})
|
|
})?;
|
|
|
|
return Ok(DataLocalityAssignment {
|
|
operation_id,
|
|
assigned_region: first_available,
|
|
data_transfer_required: true,
|
|
estimated_data_transfer_time_ms: 30000, // 30 seconds estimate
|
|
});
|
|
}
|
|
|
|
// Prefer first region where data exists and region is available
|
|
let assigned_region = data_regions
|
|
.into_iter()
|
|
.find(|region| {
|
|
self.region_status
|
|
.get(region)
|
|
.is_some_and(|status| *status == RegionStatus::Available)
|
|
})
|
|
.ok_or_else(|| {
|
|
PlatformError::Region(RegionError::Unavailable {
|
|
region_id: "no_available_data_regions".to_string(),
|
|
})
|
|
})?;
|
|
|
|
Ok(DataLocalityAssignment {
|
|
operation_id,
|
|
assigned_region,
|
|
data_transfer_required: false,
|
|
estimated_data_transfer_time_ms: 0,
|
|
})
|
|
}
|
|
}
|