use tracing::{info, warn, error, debug};
use std::sync::RwLock;
use std::hash::Hash;
#[derive(Debug, Clone, Hash, Eq, PartialEq, serde::Serialize, serde::Deserialize)]
pub struct TenantId(pub String);
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct Tenant {
pub id: TenantId,
pub name: String,
pub email: String,
pub created_at: chrono::DateTime<chrono::Utc>,
pub status: TenantStatus,
pub resource_quota: ResourceQuota,
pub security_context: SecurityContext,
}
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq)]
pub enum TenantStatus {
Active,
Suspended,
Terminated,
}
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct ResourceQuota {
pub max_clusters: u32,
pub max_replicas_per_cluster: u32,
pub max_memory_mb: u64,
pub max_cpu_cores: f32,
pub max_storage_gb: u64,
pub max_concurrent_executions: u32,
}
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct SecurityContext {
pub network_policy: NetworkPolicy,
pub rbac_roles: Vec<RbacRole>,
pub resource_limits: ResourceLimits,
pub allowed_apis: Vec<String>,
}
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct NetworkPolicy {
pub ingress_allowed: bool,
pub egress_allowed: bool,
pub allowed_sources: Vec<String>,
pub allowed_destinations: Vec<String>,
pub blocked_ports: Vec<u16>,
}
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct RbacRole {
pub name: String,
pub permissions: Vec<Permission>,
}
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub enum Permission {
CreateCluster,
DeleteCluster,
UpdateCluster,
ViewCluster,
ExecuteScript,
ManageResources,
}
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct ResourceLimits {
pub cpu_limit_per_cluster: f32,
pub memory_limit_per_cluster_mb: u64,
pub storage_limit_per_cluster_gb: u64,
pub network_bandwidth_limit_mbps: u64,
}
#[derive(Debug, Clone)]
pub struct ExecutionContext {
pub tenant_id: TenantId,
pub cluster_name: String,
pub resource_limits: ResourceLimits,
pub security_context: SecurityContext,
pub execution_id: String,
}
pub struct TenancyManager {
tenants: Arc<RwLock<std::collections::HashMap<TenantId, Tenant>>>,
executions: Arc<RwLock<std::collections::HashMap<String, ExecutionContext>>>,
resource_usage: Arc<RwLock<std::collections::HashMap<TenantId, ResourceUsage>>>,
}
#[derive(Debug, Clone, Default)]
pub struct ResourceUsage {
pub active_clusters: u32,
pub total_replicas: u32,
pub memory_usage_mb: u64,
pub cpu_usage_cores: f32,
pub storage_usage_gb: u64,
pub concurrent_executions: u32,
}
impl TenancyManager {
pub fn new() -> Self {
TenancyManager {
tenants: Arc::new(Mutex::new(std::collections::HashMap::new()))
executions: Arc::new(Mutex::new(std::collections::HashMap::new()))
resource_usage: Arc::new(Mutex::new(std::collections::HashMap::new()))
}
}
pub async fn create_tenant(
&self,
name: String,
email: String,
resource_quota: ResourceQuota,
) -> Result<TenantId, Box<dyn std::error::Error>> {
let tenant_id: _ = TenantId(uuid::Uuid::new_v4().to_string());
info!("Creating tenant: {} ({})", name, tenant_id.0);
let tenant: _ = Tenant {
id: tenant_id.clone(),
name,
email,
created_at: chrono::Utc::now(),
status: TenantStatus::Active,
resource_quota: resource_quota.clone(),
security_context: SecurityContext {
network_policy: NetworkPolicy {
ingress_allowed: true,
egress_allowed: true,
allowed_sources: vec!["*".to_string()],
allowed_destinations: vec!["*".to_string()],
blocked_ports: vec![],
},
rbac_roles: vec![],
resource_limits: ResourceLimits {
cpu_limit_per_cluster: 1.0,
memory_limit_per_cluster_mb: 1024,
storage_limit_per_cluster_gb: 10,
network_bandwidth_limit_mbps: 100,
},
allowed_apis: vec!["*".to_string()],
},
};
let mut tenants = self.tenants.write().await;
tenants.insert(tenant_id.clone(), tenant);
let mut resource_usage = self.resource_usage.write().await;
resource_usage.insert(tenant_id.clone(), ResourceUsage::default());
info!("Successfully created tenant: {} ({})", tenant.name, tenant_id.0);
Ok(tenant_id)
}
pub async fn get_tenant(&self, tenant_id: &TenantId) -> Option<Tenant> {
let tenants: _ = self.tenants.read().await;
tenants.get(tenant_id).cloned()
}
pub async fn update_tenant(
&self,
tenant_id: &TenantId,
updates: TenantUpdates,
) -> Result<(), Box<dyn std::error::Error>> {
let mut tenants = self.tenants.write().await;
if let Some(tenant) = tenants.get_mut(tenant_id) {
if let Some(name) = updates.name {
tenant.name = name;
}
if let Some(email) = updates.email {
tenant.email = email;
}
if let Some(status) = updates.status {
tenant.status = status;
}
if let Some(resource_quota) = updates.resource_quota {
tenant.resource_quota = resource_quota;
}
Ok(())
} else {
Err(format!("Tenant not found: {}", tenant_id.0).into())
}
}
pub async fn delete_tenant(&self, tenant_id: &TenantId) -> Result<(), Box<dyn std::error::Error>> {
info!("Deleting tenant: {}", tenant_id.0);
let mut tenants = self.tenants.write().await;
tenants.remove(tenant_id);
let mut resource_usage = self.resource_usage.write().await;
resource_usage.remove(tenant_id);
info!("Successfully deleted tenant: {}", tenant_id.0);
Ok(())
}
pub async fn create_execution_context(
&self,
tenant_id: &TenantId,
cluster_name: String,
) -> Result<ExecutionContext, Box<dyn std::error::Error>> {
let tenants: _ = self.tenants.read().await;
if let Some(tenant) = tenants.get(tenant_id) {
let mut resource_usage = self.resource_usage.write().await;
let usage: _ = resource_usage.entry(tenant_id.clone()).or_insert_with(ResourceUsage::default);
if usage.concurrent_executions >= tenant.resource_quota.max_concurrent_executions {
return Err("Maximum concurrent executions reached".into());
}
usage.concurrent_executions += 1;
let execution_id: _ = uuid::Uuid::new_v4().to_string();
let context: _ = ExecutionContext {
tenant_id: tenant_id.clone(),
cluster_name,
resource_limits: tenant.security_context.resource_limits.clone(),
security_context: tenant.security_context.clone(),
execution_id,
};
let mut executions = self.executions.write().await;
executions.insert(context.execution_id.clone(), context.clone());
info!("Created execution context for tenant: {} (execution: {})", tenant_id.0, execution_id);
Ok(context)
} else {
Err(format!("Tenant not found: {}", tenant_id.0).into())
}
}
pub async fn cleanup_execution(&self, execution_id: &str) -> Result<(), Box<dyn std::error::Error>> {
let mut executions = self.executions.write().await;
if let Some(context) = executions.remove(execution_id) {
let mut resource_usage = self.resource_usage.write().await;
if let Some(usage) = resource_usage.get_mut(&context.tenant_id) {
if usage.concurrent_executions > 0 {
usage.concurrent_executions -= 1;
}
}
info!("Cleaned up execution context: {}", execution_id);
Ok(())
} else {
Err(format!("Execution not found: {}", execution_id).into())
}
}
pub async fn get_resource_usage(&self, tenant_id: &TenantId) -> Option<ResourceUsage> {
let resource_usage: _ = self.resource_usage.read().await;
resource_usage.get(tenant_id).cloned()
}
pub async fn list_tenants(&self) -> Vec<Tenant> {
let tenants: _ = self.tenants.read().await;
tenants.values().cloned().collect()
}
pub async fn check_quota_exceeded(&self, tenant_id: &TenantId) -> Result<bool, Box<dyn std::error::Error>> {
let tenants: _ = self.tenants.read().await;
if let Some(tenant) = tenants.get(tenant_id) {
let resource_usage: _ = self.resource_usage.read().await;
if let Some(usage) = resource_usage.get(tenant_id) {
let exceeded: _ = usage.active_clusters > tenant.resource_quota.max_clusters
|| usage.total_replicas > tenant.resource_quota.max_replicas_per_cluster
|| usage.memory_usage_mb > tenant.resource_quota.max_memory_mb
|| usage.cpu_usage_cores > tenant.resource_quota.max_cpu_cores
|| usage.storage_usage_gb > tenant.resource_quota.max_storage_gb
|| usage.concurrent_executions > tenant.resource_quota.max_concurrent_executions;
Ok(exceeded)
} else {
Ok(false)
}
} else {
Err(format!("Tenant not found: {}", tenant_id.0).into())
}
}
}
#[derive(Debug, Default)]
pub struct TenantUpdates {
pub name: Option<String>,
pub email: Option<String>,
pub status: Option<TenantStatus>,
pub resource_quota: Option<ResourceQuota>,
}
impl TenantUpdates {
pub fn new() -> Self {
Self::default()
}
pub fn with_name(mut self, name: String) -> Self {
self.name = Some(name);
self
}
pub fn with_email(mut self, email: String) -> Self {
self.email = Some(email);
self
}
pub fn with_status(mut self, status: TenantStatus) -> Self {
self.status = Some(status);
self
}
pub fn with_resource_quota(mut self, resource_quota: ResourceQuota) -> Self {
self.resource_quota = Some(resource_quota);
self
}
}
#[cfg(test)]
mod tests {
use super::*;
#[tokio::test]
async fn test_create_tenant() {
let manager: _ = TenancyManager::new();
let resource_quota: _ = ResourceQuota {
max_clusters: 10,
max_replicas_per_cluster: 5,
max_memory_mb: 8192,
max_cpu_cores: 4.0,
max_storage_gb: 100,
max_concurrent_executions: 20,
};
let tenant_id: _ = manager
.create_tenant("test-tenant".to_string(), "test@example.com".to_string(), resource_quota)
.await
.unwrap();
let tenant: _ = manager.get_tenant(&tenant_id).await.unwrap();
assert_eq!(tenant.name, "test-tenant");
assert_eq!(tenant.email, "test@example.com");
assert_eq!(tenant.status, TenantStatus::Active);
}
#[tokio::test]
async fn test_execution_context_creation() {
let manager: _ = TenancyManager::new();
let resource_quota: _ = ResourceQuota {
max_clusters: 10,
max_replicas_per_cluster: 5,
max_memory_mb: 8192,
max_cpu_cores: 4.0,
max_storage_gb: 100,
max_concurrent_executions: 2,
};
let tenant_id: _ = manager
.create_tenant("test-tenant".to_string(), "test@example.com".to_string(), resource_quota)
.await
.unwrap();
let context1: _ = manager
.create_execution_context(&tenant_id, "cluster-1".to_string())
.await
.unwrap();
assert_eq!(context1.tenant_id, tenant_id);
let context2: _ = manager
.create_execution_context(&tenant_id, "cluster-2".to_string())
.await
.unwrap();
assert_eq!(context2.tenant_id, tenant_id);
let result: _ = manager
.create_execution_context(&tenant_id, "cluster-3".to_string())
.await;
assert!(result.is_err());
}
}
use std::collections::{BTreeMap, HashMap};
use std::sync::{Arc, Mutex};