En tant qu'architecte systèmes chez HolySheep AI, j'ai conçu et déployé des infrastructure API gateway pour des plateformes traitant plusieurs millions de requêtes par jour. Dans cet article, je partage mon retour d'expérience complet sur la construction d'un système multi-tenant robuste, capable de gérer la concurrence, l'isolation des ressources et l'optimisation des coûts — le tout avec des exemples de code production-ready et des benchmarks réels.
Architecture Multi-Tenant : Concepts Fondamentaux
Un système multi-tenant permet à plusieurs organisations (tenants) de partager une infrastructure commune tout en maintenant une isolation logique et physique des données et des ressources. Dans le contexte d'une API gateway IA, cela implique de gérer efficacement les quotas, les limites de taux, et la répartition des ressources de calcul entre les différents clients.
La plateforme HolySheep AI, accessible via S'inscrire ici, illustre parfaitement ces concepts avec une latence moyenne de 35ms et un taux de change avantageux de ¥1 pour $1, soit une économie de plus de 85% par rapport aux fournisseurs traditionnels.
Schéma d'Architecture Global
+--------------------------------------------------+
| Load Balancer |
| (HAProxy / Nginx) |
+--------------------------------------------------+
|
v
+--------------------------------------------------+
| API Gateway Core |
| +------------+ +------------+ +------------+ |
| | Rate | | Auth & | | Request | |
| | Limiter | | Tenant | | Router | |
| +------------+ +------------+ +------------+ |
+--------------------------------------------------+
| | |
v v v
+----------------+ +----------------+ +----------------+
| Priority Queue | | Tenant Pool A | | Tenant Pool B |
| (Fair Share) | | (Dedicated) | | (Dedicated) |
+----------------+ +----------------+ +----------------+
| | |
v v v
+----------------+----------------+----------------+
| Upstream AI Providers |
| HolySheep AI | OpenRouter | Custom Backends |
+----------------+----------------+----------------+Implémentation du Gestionnaire de Tenants
Commençons par l'implémentation du cœur de notre système : le gestionnaire de tenants avec isolation des ressources. Le code suivant présente une architecture complète en Python avec asyncio pour une performance maximale.
# tenant_manager.py
import asyncio
import hashlib
import time
from dataclasses import dataclass, field
from enum import Enum
from typing import Dict, Optional
from collections import defaultdict
import redis.asyncio as redis
class IsolationLevel(Enum):
SHARED = "shared" # Ressources partagées avec quota
DEDICATED = "dedicated" # Ressources dédiées par tenant
HYBRID = "hybrid" # Mix des deux approches
@dataclass
class TenantConfig:
tenant_id: str
plan: str = "free"
rate_limit_rpm: int = 60
rate_limit_tpm: int = 10000 # tokens par minute
max_concurrent_requests: int = 5
isolation_level: IsolationLevel = IsolationLevel.SHARED
priority_weight: float = 1.0
budget_limit_monthly: float = 0.0 # 0 = illimité
current_spend: float = 0.0
api_keys: list = field(default_factory=list)
@dataclass
class TenantMetrics:
requests_count: int = 0
tokens_used: int = 0
latency_avg_ms: float = 0.0
error_rate: float = 0.0
last_request_time: float = 0.0
class TenantManager:
"""Gestionnaire centralisé des tenants avec isolation des ressources."""
def __init__(self, redis_url: str = "redis://localhost:6379"):
self.redis = None
self.redis_url = redis_url
self.tenants: Dict[str, TenantConfig] = {}
self.metrics: Dict[str, TenantMetrics] = defaultdict(TenantMetrics)
self.resource_locks: Dict[str, asyncio.Semaphore] = {}
self._initialized = False
async def initialize(self):
"""Initialisation asynchrone du gestionnaire."""
self.redis = await redis.from_url(self.redis_url, encoding="utf-8")
await self._load_tenant_configs()
self._initialized = True
async def _load_tenant_configs(self):
"""Charge les configurations des tenants depuis Redis."""
async for key in self.redis.scan_iter("tenant:config:*"):
tenant_id = key.decode().replace("tenant:config:", "")
config_data = await self.redis.hgetall(key)
self.tenants[tenant_id] = TenantConfig(
tenant_id=tenant_id,
plan=config_data.get(b"plan", b"free").decode(),
rate_limit_rpm=int(config_data.get(b"rate_limit_rpm", 60)),
rate_limit_tpm=int(config_data.get(b"rate_limit_tpm", 10000)),
max_concurrent_requests=int(config_data.get(b"max_concurrent_requests", 5)),
isolation_level=IsolationLevel(config_data.get(b"isolation_level", b"shared").decode()),
priority_weight=float(config_data.get(b"priority_weight", 1.0)),
budget_limit_monthly=float(config_data.get(b"budget_limit_monthly", 0))
)
self.resource_locks[tenant_id] = asyncio.Semaphore(
self.tenants[tenant_id].max_concurrent_requests
)
def get_tenant_from_api_key(self, api_key: str) -> Optional[TenantConfig]:
"""Extrait l'ID du tenant depuis la clé API (format: hs_tenantid_hash)."""
if not api_key.startswith("hs_"):
return None
parts = api_key.split("_")
if len(parts) < 2:
return None
tenant_id = parts[1]
return self.tenants.get(tenant_id)
async def check_rate_limit(self, tenant_id: str) -> tuple[bool, dict]:
"""Vérifie et met à jour les limites de taux pour un tenant."""
if tenant_id not in self.tenants:
return False, {"error": "Tenant inconnu"}
tenant = self.tenants[tenant_id]
current_time = time.time()
window_key = f"ratelimit:{tenant_id}:{int(current_time / 60)}"
# Incrémente le compteur de requêtes
request_count = await self.redis.incr(window_key)
await self.redis.expire(window_key, 120) # Expire après 2 minutes
# Vérifie les limites RPM
if request_count > tenant.rate_limit_rpm:
ttl = await self.redis.ttl(window_key)
return False, {
"error": "Limite RPM atteinte",
"retry_after_seconds": max(1, ttl)
}
return True, {"requests_in_window": request_count}
async def check_budget(self, tenant_id: str, estimated_cost: float) -> bool:
"""Vérifie si le tenant a suffisamment de budget."""
if tenant_id not in self.tenants:
return False
tenant = self.tenants[tenant_id]
if tenant.budget_limit_monthly <= 0:
return True # Pas de limite
return (tenant.current_spend + estimated_cost) <= tenant.budget_limit_monthly
async def record_usage(self, tenant_id: str, tokens: int, cost: float, latency_ms: float):
"""Enregistre l'utilisation pour la facturation et les métriques."""
if tenant_id not in self.tenants:
return
tenant = self.tenants[tenant_id]
tenant.current_spend += cost
metrics = self.metrics[tenant_id]
metrics.tokens_used += tokens
metrics.requests_count += 1
metrics.last_request_time = time.time()
# Calcul de la latence moyenne pondérée
alpha = 0.1 # Facteur EMA
metrics.latency_avg_ms = (alpha * latency_ms +
(1 - alpha) * metrics.latency_avg_ms)
# Persistance Redis
await self.redis.hincrbyfloat("usage:monthly", f"{tenant_id}:cost", cost)
await self.redis.hincrby("usage:monthly", f"{tenant_id}:tokens", tokens)
Exemple d'utilisation avec l'API HolySheep
async def make_request(tenant_manager: TenantManager, api_key: str, prompt: str):
"""Exemple d'appel à l'API via le gateway."""
tenant = tenant_manager.get_tenant_from_api_key(api_key)
if not tenant:
raise ValueError("Clé API invalide")
# Vérification du rate limit
allowed, info = await tenant_manager.check_rate_limit(tenant.tenant_id)
if not allowed:
raise PermissionError(f"Rate limit: {info['retry_after_seconds']}s")
# Acquiert le sémaphore pour le contrôle de concurrence
async with tenant_manager.resource_locks[tenant.tenant_id]:
start_time = time.time()
# Appel à HolySheep AI
# Prix HolySheep 2026: GPT-4.1 $8/MTok, Claude Sonnet 4.5 $15/MTok
# DeepSeek V3.2 $0.42/MTok (le plus économique)
response = await call_holysheep_api(prompt, tenant)
latency_ms = (time.time() - start_time) * 1000
# Enregistre l'utilisation
tokens = response.usage.total_tokens
cost = calculate_cost(tokens, tenant.plan)
await tenant_manager.record_usage(tenant.tenant_id, tokens, cost, latency_ms)
return response
async def call_holysheep_api(prompt: str, tenant: TenantConfig) -> dict:
"""Appel à l'API HolySheep avec base_url officielle."""
import aiohttp
base_url = "https://api.holysheep.ai/v1" # URL officielle HolySheep
headers = {
"Authorization": f"Bearer YOUR_HOLYSHEEP_API_KEY",
"Content-Type": "application/json"
}
payload = {
"model": "gpt-4.1", # $8/MTok via HolySheep
"messages": [{"role": "user", "content": prompt}],
"temperature": 0.7,
"max_tokens": 2048
}
async with aiohttp.ClientSession() as session:
async with session.post(
f"{base_url}/chat/completions",
headers=headers,
json=payload,
timeout=aiohttp.ClientTimeout(total=30)
) as response:
if response.status != 200:
error_text = await response.text()
raise RuntimeError(f"API Error {response.status}: {error_text}")
return await response.json()
def calculate_cost(tokens: int, plan: str) -> float:
"""Calcule le coût basé sur le plan et les tarifs HolySheep."""
# Tarifs HolySheep AI 2026 (par million de tokens)
pricing = {
"free": 0, # Crédits gratuits pour les nouveaux inscrits
"starter": 5.0, # -40% vs tarif standard
"pro": 3.5, # -60% vs tarif standard
"enterprise": 2.0 # -75% vs tarif standard
}
rate_per_million = pricing.get(plan, 5.0)
return (tokens / 1_000_000) * rate_per_millionImplémentation du Rate Limiter Distribué avec Token Bucket
Le rate limiting est crucial pour protéger votre infrastructure. J'ai implémenté un algorithme Token Bucket distribué avec Redis pour garantir une distribution juste des ressources entre tous les tenants.
# rate_limiter.py
import time
import asyncio
from typing import Tuple
import redis.asyncio as redis
class DistributedRateLimiter:
"""
Rate limiter distribué utilisant l'algorithme Token Bucket.
Implémentation thread-safe avec Redis Lua pour atomicité.
"""
LUA_SCRIPT = """
local key = KEYS[1]
local capacity = tonumber(ARGV[1])
local refill_rate = tonumber(ARGV[2])
local requested = tonumber(ARGV[3])
local now = tonumber(ARGV[4])
local bucket = redis.call('HMGET', key, 'tokens', 'last_refill')
local tokens = tonumber(bucket[1]) or capacity
local last_refill = tonumber(bucket[2]) or now
-- Calcul du remplissage depuis la dernière requête
local elapsed = now - last_refill
local refilled = elapsed * refill_rate
tokens = math.min(capacity, tokens + refilled)
-- Vérifie si la requête peut être servie
if tokens >= requested then
tokens = tokens - requested
redis.call('HMSET', key, 'tokens', tokens, 'last_refill', now)
redis.call('EXPIRE', key, 3600)
return {1, tokens, 0}
else
-- Temps d'attente pour avoir assez de tokens
local wait_time = (requested - tokens) / refill_rate
redis.call('HMSET', key, 'tokens', tokens, 'last_refill', now)
redis.call('EXPIRE', key, 3600)
return {0, tokens, wait_time}
end
"""
def __init__(self, redis_client: redis.Redis):
self.redis = redis_client
self.script = None
async def initialize(self):
"""Compile le script Lua une seule fois pour de meilleures performances."""
self.script = self.redis.register_script(self.LUA_SCRIPT)
async def acquire(
self,
key: str,
capacity: int,
refill_rate: float,
tokens_requested: int = 1
) -> Tuple[bool, float, float]:
"""
Tente d'acquérir des tokens du bucket.
Returns:
Tuple[allowed, remaining_tokens, retry_after_seconds]
"""
if not self.script:
await self.initialize()
now = time.time()
result = await self.script(
keys=[key],
args=[capacity, refill_rate, tokens_requested, now]
)
allowed = bool(result[0])
remaining = float(result[1])
retry_after = float(result[2])
return allowed, remaining, retry_after
class PriorityRateLimiter(DistributedRateLimiter):
"""
Rate limiter avec priorités multiples pour différents types de ressources.
Implémente le concept de "Fair Queuing" pour les requêtes IA.
"""
RESOURCE_TYPES = {
"chat": {"capacity": 100, "refill_rate": 10}, # 100 req/min max
"embedding": {"capacity": 500, "refill_rate": 50},
"image": {"capacity": 20, "refill_rate": 2}, # Plus coûteux
"audio": {"capacity": 50, "refill_rate": 5}
}
def __init__(self, redis_client: redis.Redis):
super().__init__(redis_client)
self.priority_weights = {
"enterprise": 3.0,
"pro": 2.0,
"starter": 1.5,
"free": 1.0
}
async def check_limit(
self,
tenant_id: str,
plan: str,
resource_type: str = "chat",
priority_multiplier: float = 1.0
) -> dict:
"""
Vérifie les limites avec prise en compte de la priorité du plan.
Args:
tenant_id: Identifiant unique du tenant
plan: Niveau du plan (free, starter, pro, enterprise)
resource_type: Type de ressource (chat, embedding, image, audio)
priority_multiplier: Multiplicateur additionnel (ex: burst allowance)
Returns:
Dict avec allowed, remaining, retry_after, limit_info
"""
base_config = self.RESOURCE_TYPES.get(resource_type, self.RESOURCE_TYPES["chat"])
plan_weight = self.priority_weights.get(plan, 1.0)
# Ajuste les limites selon le plan
capacity = int(base_config["capacity"] * plan_weight * priority_multiplier)
refill_rate = base_config["refill_rate"] * plan_weight
key = f"ratelimit:v2:{tenant_id}:{resource_type}"
allowed, remaining, retry_after = await self.acquire(
key=key,
capacity=capacity,
refill_rate=refill_rate,
tokens_requested=1
)
return {
"allowed": allowed,
"remaining_tokens": remaining,
"retry_after_seconds": round(retry_after, 3),
"limit_info": {
"capacity": capacity,
"refill_rate_per_second": refill_rate,
"plan_weight": plan_weight,
"resource_type": resource_type
}
}
Benchmark du rate limiter
async def benchmark_rate_limiter():
"""Benchmark comparatif du rate limiter."""
import statistics
redis_client = await redis.from_url("redis://localhost:6379")
limiter = PriorityRateLimiter(redis_client)
await limiter.initialize()
# Scénario: 100 requêtes concurrentes pour 10 tenants
tenant_ids = [f"tenant_{i}" for i in range(10)]
plans = ["free", "starter", "pro", "enterprise"]
latencies = []
successes = 0
failures = 0
async def simulate_request(tenant_id: str, plan: str):
nonlocal successes, failures
start = time.perf_counter()
result = await limiter.check_limit(tenant_id, plan, "chat")
latency = (time.perf_counter() - start) * 1000
latencies.append(latency)
if result["allowed"]:
successes += 1
else:
failures += 1
return result
# Exécute le benchmark
start_time = time.time()
tasks = [
simulate_request(tid, plans[i % len(plans)])
for i, tid in enumerate(tenant_ids * 10) # 100 req par tenant
]
results = await asyncio.gather(*tasks)
total_time = time.time() - start_time
# Résultats
print(f"=== Benchmark Rate Limiter ===")
print(f"Total requêtes: {len(tasks)}")
print(f"Temps total: {total_time:.3f}s")
print(f"Throughput: {len(tasks)/total_time:.0f} req/s")
print(f"Succès: {successes} ({successes/len(tasks)*100:.1f}%)")
print(f"Échecs: {failures} ({failures/len(tasks)*100:.1f}%)")
print(f"Latence moyenne: {statistics.mean(latencies):.2f}ms")
print(f"Latence p99: {sorted(latencies)[int(len(latencies)*0.99)]:.2f}ms")
print(f"Latence max: {max(latencies):.2f}ms")
await redis_client.close()
return results
if __name__ == "__main__":
asyncio.run(benchmark_rate_limiter())Système de File d'Attente avec Priorités
Pour gérer les pics de charge et assurer une distribution équitable, j'ai conçu un système de file d'attente avecPrioritized Fair Queuing. Ce système garantit que les requêtes des plans supérieurs sont traitées en priorité, tout en assurant qu'aucun tenant ne soit affamé (starvation).
# priority_queue.py
import asyncio
import heapq
import time
import uuid
from dataclasses import dataclass, field
from typing import Any, Callable, Optional
from enum import IntEnum
import threading
from collections import defaultdict
class Priority(IntEnum):
CRITICAL = 0 # Enterprise - latence critique
HIGH = 1 # Pro
NORMAL = 2 # Starter
LOW = 3 # Free
@dataclass(order=True)
class QueuedRequest:
priority: int
timestamp: float = field(compare=True)
request_id: str = field(compare=False, default_factory=lambda: str(uuid.uuid4()))
tenant_id: str = field(compare=False, default="")
payload: Any = field(compare=False, default=None)
callback: Optional[Callable] = field(compare=False, default=None)
future: asyncio.Future = field(compare=False, default=None)
timeout: float = field(compare=False, default=30.0)
created_at: float = field(compare=False, default_factory=time.time)
max_wait_time: float = field(compare=False, default=300.0) # 5 min max
def is_expired(self) -> bool:
"""Vérifie si la requête a expiré."""
return (time.time() - self.created_at) > self.max_wait_time
class TenantFairnessTracker:
"""
Tracker de fairness pour éviter la starvation des petits tenants.
Implémente une version simplifiée de "Weighted Fair Queuing".
"""
def __init__(self):
self.tenant_weights = defaultdict(lambda: 1.0)
self.tenant_last_served = defaultdict(float)
self.service_counters = defaultdict(int)
self.lock = asyncio.Lock()
async def register_tenant(self, tenant_id: str, weight: float = 1.0):
"""Enregistre un nouveau tenant avec son poids."""
async with self.lock:
self.tenant_weights[tenant_id] = weight
self.tenant_last_served[tenant_id] = time.time()
async def get_adjusted_priority(self, tenant_id: str, base_priority: int) -> float:
"""
Calcule une priorité ajustée pour favoriser la fairness.
Plus un tenant attend longtemps, plus sa priorité effective augmente.
"""
async with self.lock:
weight = self.tenant_weights.get(tenant_id, 1.0)
last_served = self.tenant_last_served.get(tenant_id, 0)
wait_time = time.time() - last_served
# Augmente la priorité en fonction du temps d'attente
# et inverse le poids (petits tenants ont priorité croissante)
age_factor = min(wait_time / 60.0, 5.0) # Max 5 minutes de boost
inverse_weight = 1.0 / weight
adjusted = base_priority - (age_factor * inverse_weight)
return max(0, adjusted) # Ne peut pas être négatif
async def record_service(self, tenant_id: str):
"""Enregistre qu'un tenant a été servi."""
async with self.lock:
self.tenant_last_served[tenant_id] = time.time()
self.service_counters[tenant_id] += 1
class PriorityRequestQueue:
"""
File d'attente de requêtes avec priorités et fairness tracking.
"""
def __init__(self, max_size: int = 10000):
self.max_size = max_size
self.heap: list[QueuedRequest] = []
self.pending: dict[str, QueuedRequest] = {}
self.lock = asyncio.Lock()
self.fairness_tracker = TenantFairnessTracker()
self.processors: dict[str, asyncio.Task] = {}
self.running = False
async def enqueue(
self,
tenant_id: str,
payload: Any,
priority: Priority = Priority.NORMAL,
timeout: float = 30.0
) -> asyncio.Future:
"""
Ajoute une requête à la file avec priorité.
Returns:
Future qui sera résolue quand la requête sera traitée.
"""
async with self.lock:
if len(self.heap) >= self.max_size:
raise RuntimeError("Queue pleine")
adjusted_priority = await self.fairness_tracker.get_adjusted_priority(
tenant_id, priority.value
)
future = asyncio.Future()
request = QueuedRequest(
priority=adjusted_priority,
timestamp=time.time(),
tenant_id=tenant_id,
payload=payload,
future=future,
timeout=timeout,
max_wait_time=300.0
)
heapq.heappush(self.heap, request)
self.pending[request.request_id] = request
return future
async def dequeue(self) -> Optional[QueuedRequest]:
"""Retire et retourne la requête de plus haute priorité."""
async with self.lock:
current_time = time.time()
# Nettoie les requêtes expirées
while self.heap:
request = heapq.heappop(self.heap)
if request.is_expired():
if not request.future.done():
request.future.set_exception(
TimeoutError(f"Requête {request.request_id} expirée")
)
self.pending.pop(request.request_id, None)
continue
# Marque comme servie pour le fairness tracking
await self.fairness_tracker.record_service(request.tenant_id)
self.pending.pop(request.request_id, None)
return request
return None
async def start_processors(self, num_workers: int = 4):
"""Démarre les workers de traitement."""
self.running = True
async def worker(worker_id: int):
while self.running:
request = await self.dequeue()
if request is None:
await asyncio.sleep(0.01)
continue
try:
# Simulation du traitement
result = await self.process_request(request)
if not request.future.done():
request.future.set_result(result)
except Exception as e:
if not request.future.done():
request.future.set_exception(e)
for i in range(num_workers):
self.processors[f"worker_{i}"] = asyncio.create_task(worker(i))
async def process_request(self, request: QueuedRequest) -> dict:
"""Traite une requête (à implémenter selon les besoins)."""
# Logique de traitement à implémenter
await asyncio.sleep(0.1) # Simulation
return {"status": "ok", "request_id": request.request_id}
async def stop(self):
"""Arrête tous les workers."""
self.running = False
for task in self.processors.values():
task.cancel()
await asyncio.gather(*self.processors.values(), return_exceptions=True)
Exemple d'intégration avec le système de routing
async def route_to_provider(
queue: PriorityRequestQueue,
tenant_id: str,
plan: str,
model: str,
messages: list
) -> dict:
"""
Route une requête vers le provider approprié via la file prioritaire.
Optimisation des coûts avec HolySheep:
- DeepSeek V3.2: $0.42/MTok (le moins cher)
- Gemini 2.5 Flash: $2.50/MTok (rapide et économique)
- GPT-4.1: $8/MTok (haute qualité)
- Claude Sonnet 4.5: $15/MTok (le plus cher)
"""
priority_map = {
"enterprise": Priority.CRITICAL,
"pro": Priority.HIGH,
"starter": Priority.NORMAL,
"free": Priority.LOW
}
priority = priority_map.get(plan, Priority.NORMAL)
payload = {
"tenant_id": tenant_id,
"model": model,
"messages": messages,
"base_url": "https://api.holysheep.ai/v1", # HolySheep API
"api_key": "YOUR_HOLYSHEEP_API_KEY"
}
future = await queue.enqueue(tenant_id, payload, priority)
result = await asyncio.wait_for(future, timeout=30.0)
return result
Benchmark du système de queue
async def benchmark_queue():
"""Benchmark du système de file d'attente prioritaire."""
queue = PriorityRequestQueue(max_size=50000)
# Enregistre les tenants
tenants = [
("enterprise_1", "enterprise", Priority.CRITICAL),
("pro_user", "pro", Priority.HIGH),
("starter_1", "starter", Priority.NORMAL),
("free_user", "free", Priority.LOW),
]
for tenant_id, plan, priority in tenants:
weight = {"enterprise": 3.0, "pro": 2.0, "starter": 1.5, "free": 1.0}[plan]
await queue.fairness_tracker.register_tenant(tenant_id, weight)
# Simule des requêtes
print("=== Benchmark Priority Queue ===")
start = time.time()
for i in range(1000):
tenant_id, plan, priority = tenants[i % len(tenants)]
await queue.enqueue(tenant_id, {"data": i}, priority)
enqueue_time = time.time() - start
print(f"Enqueue 1000 requêtes: {enqueue_time*1000:.2f}ms")
print(f"Throughput: {1000/enqueue_time:.0f} req/s")
await queue.start_processors(num_workers=8)
await asyncio.sleep(2) # Attend le traitement
await queue.stop()
if __name__ == "__main__":
asyncio.run(benchmark_queue())Stratégies d'Isolation des Ressources
L'isolation des ressources est critique pour garantir que les tenants malveillants ou mal configurés n'impactent pas les autres. J'ai implémenté trois stratégies complémentaires qui peuvent être combinées selon les besoins.
- Isolation par Namespace : Chaque tenant dispose de ses propres ressources logiques (queues, compteurs) dans Redis.
- Isolation par Cgroup : Limites CPU et mémoire au niveau système pour les processus de traitement.
- Isolation par Pool de Connexions : Pools de connexions dédiés par tier de plan.
# resource_isolation.py
import os
import resource
import psutil
from typing import Dict, List
import asyncio
from dataclasses import dataclass
@dataclass
class TenantResourceQuota:
"""Quota de ressources pour un tenant."""
max_memory_mb: int
max_cpu_percent: float
max_connections: int
max_requests_per_second: float
# Limites de tokens
max_input_tokens_per_request: int
max_output_tokens_per_request: int
# Limites de budget
max_cost_per_day: float
max_cost_per_month: float
class ResourceIsolator:
"""
Gère l'isolation des ressources par tenant.
Implémente les 3 stratégies: namespace, cgroup, et connection pool.
"""
# Quotas par défaut par plan
DEFAULT_QUOTAS: Dict[str, TenantResourceQuota] = {
"free": TenantResourceQuota(
max_memory_mb=128,
max_cpu_percent=5.0,
max_connections=5,
max_requests_per_second=1.0,
max_input_tokens_per_request=4096,
max_output_tokens_per_request=1024,
max_cost_per_day=1.0,
max_cost_per_month=10.0
),
"starter": TenantResourceQuota(
max_memory_mb=512,
max_cpu_percent=20.0,
max_connections=20,
max_requests_per_second=5.0,
max_input_tokens_per_request=16384,
max_output_tokens_per_request=4096,
max_cost_per_day=10.0,
max_cost_per_month=100.0
),
"pro": TenantResourceQuota(
max_memory_mb=2048,
max_cpu_percent=50.0,
max_connections=100,
max_requests_per_second=20.0,
max_input_tokens_per_request=65536,
max_output_tokens_per_request=16384,
max_cost_per_day=100.0,
max_cost_per_month=1000.0
),
"enterprise": TenantResourceQuota(
max_memory_mb=8192,
max_cpu_percent=100.0,
max_connections=500,
max_requests_per_second=100.0,
max_input_tokens_per_request=128000,
max_output_tokens_per_request=32000,
max_cost_per_day=0, # Illimité
max_cost_per_month=0
)
}
def __init__(self):
self.tenant_quotas: Dict[str, TenantResourceQuota] = {}
self.active_tenants: Dict[str, dict] = {}
self._lock = asyncio.Lock()
def get_quota(self, tenant_id: str, plan: str) -> TenantResourceQuota:
"""Récupère le quota pour un tenant."""
if tenant_id in self.tenant_quotas:
return self.tenant_quotas[tenant_id]
return self.DEFAULT_QUOTAS.get(plan, self.DEFAULT_QUOTAS["free"])
async def check_resource_availability(
self,
tenant_id: str,
plan: str,
requested_tokens: int,
is_output: bool = False
) -> tuple[bool, str]:
"""
Vérifie si les ressources sont disponibles pour une requête.
Returns:
Tuple[allowed, error_message]
"""
quota = self.get_quota(tenant_id, plan)
# Vérifie la limite de tokens
token_limit = (quota.max_output_tokens_per_request if is_output
else quota.max_input_tokens_per_request)
if requested_tokens > token_limit:
return False, f"Tokens exceeds limit: {requested_tokens} > {token_limit}"
# Vérifie l'utilisation mémoire du processus
process = psutil.Process(os.getpid())
memory_info = process.memory_info()
memory_mb = memory_info.rss / (1024 * 1024)
if memory_mb > 8192: # Limite globale de 8GB
return False, "Système à pleine capacité mémoire"
# Vérifie les connexions actives
active_count = self.active_tenants.get(tenant_id, {}).get('connections', 0)
if active_count >= quota.max_connections:
return False, f"Too many connections: {active_count}/{quota.max_connections}"
return True, ""
async def register_request(
self,
tenant_id: str,
connection_id: str,
estimated