En tant qu'ingénieur qui a géré des infrastructureserves critiques 处理 des milliers de requêtes par seconde vers les API d'IA, j'ai vécu firsthand les catastrophes qu'une absence de protection peut provoquer. L'année dernière, notre plateforme a subi une panne de 3 heures après qu'un batch mal configuré ait épuisé notre quota API en 45 minutes, coûtant des milliers d'euros en opportunités perdues. Aujourd'hui, je partage mon expérience complète pour vous épargner ces erreurs.
Tableau comparatif : HolySheep vs API officielles vs Services relais
| Critère | HolySheep AI | API Officielles (OpenAI/Anthropic) | Services relais tiers |
|---|---|---|---|
| Latence moyenne | <50ms | 150-300ms | 80-200ms |
| GPT-4.1 ($/MTok) | $8.00 | $15.00 | $10-12 |
| Claude Sonnet 4.5 ($/MTok) | $15.00 | $22.00 | $18-20 |
| Gemini 2.5 Flash ($/MTok) | $2.50 | $4.50 | $3.50 |
| DeepSeek V3.2 ($/MTok) | $0.42 | N/A | $0.60-0.80 |
| Taux de change | ¥1 = $1 USD | Dollar américain uniquement | Variables |
| Paiement | WeChat Pay, Alipay, Stripe | Carte internationale uniquement | Limité |
| Crédits gratuits | Oui — inclus | Limité | Rarement |
| Rate Limiting intégré | Avancé | Basique | Variable |
| Circuit Breaker | Native support | Non | Développeur-dépendant |
| Économie vs officiel | 85%+ | Référence | 30-50% |
Après avoir testé tous les providers majeurs pendant 6 mois, HolySheep s'est imposé comme le choix optimal pour les architectures modernes. Passons maintenant à l'implémentation technique.
Comprendre les concepts fondamentaux
Qu'est-ce que le Rate Limiting ?
Le rate limiting est une technique qui contrôle le nombre de requêtes qu'un client peut faire dans un laps de temps donné. Pour les API IA, c'est crucial car les coûts peuvent exploser rapidement et les providers imposent des limites strictes.
Types de limites常见的
- Token Bucket :允许突发流量，同时限制平均速率
- Leaky Bucket :文件请求速率恒定
- Fixed Window :按固定时间段计数
- Sliding Window :平滑的时间窗口
Qu'est-ce que le Circuit Breaker ?
Le pattern Circuit Breaker, popularisé par Michael Nygard dans "Release It!", empêche les appels en cascade vers un service défaillant. Au lieu d'attendre un timeout, le circuit "ouvre" après un nombre défini d'échecs.
Implémentation complète en Python
1. Rate Limiter avec Token Bucket
"""
HolySheep AI - Rate Limiter avec Token Bucket Pattern
Implémentation production-ready pour gérer les quotas API
"""
import time
import threading
from collections import deque
from dataclasses import dataclass
from typing import Optional, Dict
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
@dataclass
class RateLimitConfig:
"""Configuration du rate limiting par endpoint"""
requests_per_second: int = 10
burst_size: int = 20
tokens_per_request: int = 1
class TokenBucketRateLimiter:
"""
Rate Limiter utilisant le pattern Token Bucket
Avantages:
- Permet les pics de traffic (burst)
- Limite la moyenne de requêtes
- Thread-safe pour environnement concurrentiel
"""
def __init__(self, config: RateLimitConfig):
self.config = config
self.tokens = float(config.burst_size)
self.last_update = time.time()
self.lock = threading.Lock()
self.request_history = deque(maxlen=1000)
def _refill_tokens(self):
"""Remplit les tokens selon le temps écoulé"""
now = time.time()
elapsed = now - self.last_update
new_tokens = elapsed * self.config.requests_per_second
self.tokens = min(
self.config.burst_size,
self.tokens + new_tokens
)
self.last_update = now
def acquire(self, tokens: int = 1, blocking: bool = False) -> bool:
"""
Acquérir des tokens pour une requête
Args:
tokens: Nombre de tokens nécessaires
blocking: Si True, attend que des tokens soient disponibles
Returns:
True si les tokens ont été acquis
"""
with self.lock:
self._refill_tokens()
if self.tokens >= tokens:
self.tokens -= tokens
self.request_history.append(time.time())
logger.debug(f"Token acquis. Tokens restants: {self.tokens:.2f}")
return True
if blocking:
# Calculer le temps d'attente
wait_time = (tokens - self.tokens) / self.config.requests_per_second
time.sleep(wait_time)
self._refill_tokens()
self.tokens -= tokens
self.request_history.append(time.time())
return True
return False
def get_wait_time(self) -> float:
"""Retourne le temps d'attente estimé en secondes"""
with self.lock:
self._refill_tokens()
tokens_needed = self.config.tokens_per_request
if self.tokens >= tokens_needed:
return 0.0
return (tokens_needed - self.tokens) / self.config.requests_per_second
def get_stats(self) -> Dict:
"""Statistiques d'utilisation"""
with self.lock:
now = time.time()
recent_requests = sum(1 for t in self.request_history if now - t < 60)
return {
"tokens_disponibles": round(self.tokens, 2),
"requests_derniere_minute": recent_requests,
"limite_rps": self.config.requests_per_second,
"burst_size": self.config.burst_size
}
Exemple d'utilisation avec HolySheep API
if __name__ == "__main__":
# Configuration pour différents modèles
configs = {
"gpt4": RateLimitConfig(requests_per_second=10, burst_size=20),
"claude": RateLimitConfig(requests_per_second=5, burst_size=10),
"gemini": RateLimitConfig(requests_per_second=50, burst_size=100),
}
limiter = TokenBucketRateLimiter(configs["gpt4"])
# Simuler des requêtes
for i in range(25):
if limiter.acquire():
print(f"Requête {i+1}: ✓ | Stats: {limiter.get_stats()}")
else:
wait = limiter.get_wait_time()
print(f"Requête {i+1}: Rate limité, attente {wait:.2f}s")2. Circuit Breaker Pattern complet
"""
HolySheep AI - Circuit Breaker Implementation
Protection contre les échecs en cascade
"""
import time
import threading
from enum import Enum
from typing import Callable, Any, Optional
from dataclasses import dataclass
import asyncio
class CircuitState(Enum):
"""États du Circuit Breaker"""
CLOSED = "closed" # Fonctionnement normal
OPEN = "open" # Circuit ouvert, requêtes bloquées
HALF_OPEN = "half_open" # Test de récupération
@dataclass
class CircuitBreakerConfig:
"""Configuration du Circuit Breaker"""
failure_threshold: int = 5 # Échecs avant ouverture
success_threshold: int = 3 # Succès pour fermer
timeout: float = 30.0 # Timeout avant demi-ouverture (secondes)
half_open_max_calls: int = 3 # Appels max en demi-ouverture
class CircuitBreakerOpen(Exception):
"""Exception levée quand le circuit est ouvert"""
def __init__(self, circuit_name: str, retry_after: float):
self.circuit_name = circuit_name
self.retry_after = retry_after
super().__init__(
f"Circuit '{circuit_name}' est ouvert. "
f"Réessayez dans {retry_after:.1f} secondes."
)
class CircuitBreaker:
"""
Circuit Breaker Pattern pour HolySheep API
Comportement:
1. CLOSED: Surveillance normale des erreurs
2. OPEN: Toutes les requêtes échouent immédiatement
3. HALF_OPEN: Test de récupération avec requêtes limitées
Paramètres recommandés:
- failure_threshold: 5 (ouvre après 5 erreurs consécutives)
- timeout: 30s (attendre avant de tester)
- success_threshold: 3 (ferme après 3 succès)
"""
def __init__(self, name: str, config: Optional[CircuitBreakerConfig] = None):
self.name = name
self.config = config or CircuitBreakerConfig()
self._state = CircuitState.CLOSED
self._failure_count = 0
self._success_count = 0
self._last_failure_time: Optional[float] = None
self._lock = threading.RLock()
self._half_open_calls = 0
# Métriques
self.total_calls = 0
self.successful_calls = 0
self.failed_calls = 0
self.rejected_calls = 0
@property
def state(self) -> CircuitState:
"""Retourne l'état actuel en vérifiant le timeout"""
with self._lock:
if self._state == CircuitState.OPEN:
if self._should_attempt_reset():
self._transition_to_half_open()
return self._state
def _should_attempt_reset(self) -> bool:
"""Vérifie si assez de temps s'est écoulé pour réessayer"""
if self._last_failure_time is None:
return True
return time.time() - self._last_failure_time >= self.config.timeout
def _transition_to_half_open(self):
"""Transition vers l'état half-open"""
self._state = CircuitState.HALF_OPEN
self._half_open_calls = 0
self._success_count = 0
logger.info(f"Circuit '{self.name}': Transition vers HALF_OPEN")
def _transition_to_open(self):
"""Transition vers l'état open"""
self._state = CircuitState.OPEN
self._last_failure_time = time.time()
logger.warning(
f"Circuit '{self.name}': OUVERT après {self._failure_count} échecs"
)
def _transition_to_closed(self):
"""Transition vers l'état closed"""
self._state = CircuitState.CLOSED
self._failure_count = 0
self._success_count = 0
logger.info(f"Circuit '{self.name}': FERMé - Service récupéré")
def record_success(self):
"""Enregistre un succès"""
with self._lock:
self.successful_calls += 1
if self._state == CircuitState.HALF_OPEN:
self._success_count += 1
self._half_open_calls += 1
if self._success_count >= self.config.success_threshold:
self._transition_to_closed()
elif self._state == CircuitState.CLOSED:
self._failure_count = max(0, self._failure_count - 1)
def record_failure(self):
"""Enregistre un échec"""
with self._lock:
self.failed_calls += 1
self._failure_count += 1
if self._state == CircuitState.CLOSED:
if self._failure_count >= self.config.failure_threshold:
self._transition_to_open()
elif self._state == CircuitState.HALF_OPEN:
self._transition_to_open()
def can_execute(self) -> bool:
"""Vérifie si une requête peut être exécutée"""
with self._lock:
if self._state == CircuitState.CLOSED:
return True
if self._state == CircuitState.HALF_OPEN:
return self._half_open_calls < self.config.half_open_max_calls
return False
def call(self, func: Callable, *args, **kwargs) -> Any:
"""
Exécute une fonction avec protection du circuit breaker
Raises:
CircuitBreakerOpen: Si le circuit est ouvert
Exception: Toute exception de la fonction originale
"""
self.total_calls += 1
if not self.can_execute():
self.rejected_calls += 1
retry_after = self.config.timeout
if self._last_failure_time:
retry_after = max(
0,
self.config.timeout - (time.time() - self._last_failure_time)
)
raise CircuitBreakerOpen(self.name, retry_after)
try:
result = func(*args, **kwargs)
self.record_success()
return result
except Exception as e:
self.record_failure()
raise
def get_metrics(self) -> dict:
"""Retourne les métriques du circuit breaker"""
return {
"name": self.name,
"state": self.state.value,
"failure_count": self._failure_count,
"total_calls": self.total_calls,
"successful_calls": self.successful_calls,
"failed_calls": self.failed_calls,
"rejected_calls": self.rejected_calls,
"success_rate": (
self.successful_calls / self.total_calls * 100
if self.total_calls > 0 else 0
)
}
Démonstration avec HolySheep API
if __name__ == "__main__":
import requests
# Configuration du circuit breaker
config = CircuitBreakerConfig(
failure_threshold=3,
success_threshold=2,
timeout=10.0
)
breaker = CircuitBreaker("holy_sheep_chat", config)
def call_holysheep_api(messages):
"""Appel simulé à l'API HolySheep"""
response = requests.post(
"https://api.holysheep.ai/v1/chat/completions",
headers={
"Authorization": f"Bearer YOUR_HOLYSHEEP_API_KEY",
"Content-Type": "application/json"
},
json={
"model": "gpt-4.1",
"messages": messages,
"max_tokens": 100
},
timeout=10
)
return response.json()
# Test du circuit breaker
test_messages = [{"role": "user", "content": "Test"}]
print("=== Test Circuit Breaker ===")
for i in range(10):
try:
result = breaker.call(call_holysheep_api, test_messages)
print(f"Appel {i+1}: Succès")
except CircuitBreakerOpen as e:
print(f"Appel {i+1}: Bloqué - {e}")
except Exception as e:
print(f"Appel {i+1}: Erreur - {e}")
print(f"Métriques: {breaker.get_metrics()}")
print("-" * 50)3. Client API HolySheep complet avec protection
"""
HolySheep AI - Client API complet avec Rate Limiting et Circuit Breaker
Production-ready implementation
"""
import time
import threading
import logging
from typing import List, Dict, Any, Optional
from dataclasses import dataclass, field
from queue import Queue, Empty
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
logger = logging.getLogger(__name__)
@dataclass
class HolySheepClientConfig:
"""Configuration du client HolySheep"""
api_key: str
base_url: str = "https://api.holysheep.ai/v1"
# Rate Limiting
requests_per_second: float = 10.0
burst_size: int = 20
max_queue_size: int = 1000
# Retry
max_retries: int = 3
retry_delay: float = 1.0
exponential_backoff: bool = True
# Circuit Breaker
cb_failure_threshold: int = 5
cb_timeout: float = 30.0
cb_success_threshold: int = 3
class HolySheepAIClient:
"""
Client API HolySheep avec protection complète
Fonctionnalités:
- Rate Limiting automatique
- Circuit Breaker
- Retry avec backoff exponentiel
- Queue de requêtes asynchrones
- Métriques de monitoring
"""
def __init__(self, config: HolySheepClientConfig):
self.config = config
self.api_key = config.api_key
# Rate Limiter
self._tokens = float(config.burst_size)
self._last_refill = time.time()
self._rate_lock = threading.Lock()
# Request queue
self._request_queue: Queue = Queue(maxsize=config.max_queue_size)
self._worker_thread: Optional[threading.Thread] = None
self._shutdown_event = threading.Event()
# Circuit Breaker
self._circuit_breaker = CircuitBreaker(
"holy_sheep_api",
CircuitBreakerConfig(
failure_threshold=config.cb_failure_threshold,
timeout=config.cb_timeout,
success_threshold=config.cb_success_threshold
)
)
# Session HTTP optimisée
self._session = self._create_session()
# Métriques
self._metrics = {
"total_requests": 0,
"successful_requests": 0,
"failed_requests": 0,
"rate_limited": 0,
"circuit_open": 0,
"total_tokens_used": 0
}
self._metrics_lock = threading.Lock()
def _create_session(self) -> requests.Session:
"""Crée une session HTTP optimisée"""
session = requests.Session()
retry_strategy = Retry(
total=self.config.max_retries,
backoff_factor=1 if self.config.exponential_backoff else 0,
status_forcelist=[429, 500, 502, 503, 504],
allowed_methods=["POST", "GET"]
)
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount("http://", adapter)
session.mount("https://", adapter)
return session
def _refill_tokens(self):
"""Remplit les tokens selon le temps écoulé"""
now = time.time()
elapsed = now - self._last_refill
new_tokens = elapsed * self.config.requests_per_second
self._tokens = min(
self.config.burst_size,
self._tokens + new_tokens
)
self._last_refill = now
def _acquire_token(self, blocking: bool = True, timeout: float = 30.0) -> bool:
"""Acquiert un token pour la requête"""
start_time = time.time()
while True:
with self._rate_lock:
self._refill_tokens()
if self._tokens >= 1:
self._tokens -= 1
return True
if not blocking:
return False
# Calculer le temps d'attente
wait_time = 1.0 / self.config.requests_per_second
if time.time() - start_time >= timeout:
return False
time.sleep(min(wait_time, timeout - (time.time() - start_time)))
def chat_completions(
self,
messages: List[Dict[str, str]],
model: str = "gpt-4.1",
**kwargs
) -> Dict[str, Any]:
"""
Appelle l'endpoint /chat/completions
Args:
messages: Liste des messages
model: Modèle à utiliser (gpt-4.1, claude-sonnet-4.5, etc.)
**kwargs: Paramètres additionnels (temperature, max_tokens, etc.)
Returns:
Réponse de l'API
"""
with self._metrics_lock:
self._metrics["total_requests"] += 1
# Rate Limiting
if not self._acquire_token(blocking=True):
with self._metrics_lock:
self._metrics["rate_limited"] += 1
raise Exception("Rate limit atteint - timeout d'acquisition")
# Circuit Breaker
if not self._circuit_breaker.can_execute():
with self._metrics_lock:
self._metrics["circuit_open"] += 1
raise CircuitBreakerOpen("holy_sheep_api", self.config.cb_timeout)
# Préparer la requête
payload = {
"model": model,
"messages": messages,
**kwargs
}
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
try:
response = self._session.post(
f"{self.config.base_url}/chat/completions",
headers=headers,
json=payload,
timeout=kwargs.get("timeout", 60)
)
response.raise_for_status()
result = response.json()
# Extraire les tokens utilisés
if "usage" in result:
tokens_used = result["usage"].get("total_tokens", 0)
with self._metrics_lock:
self._metrics["total_tokens_used"] += tokens_used
self._circuit_breaker.record_success()
with self._metrics_lock:
self._metrics["successful_requests"] += 1
return result
except requests.exceptions.HTTPError as e:
self._circuit_breaker.record_failure()
with self._metrics_lock:
self._metrics["failed_requests"] += 1
if e.response.status_code == 429:
raise Exception("Rate limit API atteint - réduire le rythme")
elif e.response.status_code == 401:
raise Exception("Clé API invalide")
elif e.response.status_code == 400:
raise Exception(f"Requête invalide: {e.response.text}")
else:
raise
except requests.exceptions.RequestException as e:
self._circuit_breaker.record_failure()
with self._metrics_lock:
self._metrics["failed_requests"] += 1
raise Exception(f"Erreur de connexion: {str(e)}")
def get_metrics(self) -> Dict[str, Any]:
"""Retourne les métriques complètes"""
with self._metrics_lock:
metrics = self._metrics.copy()
metrics["circuit_breaker"] = self._circuit_breaker.get_metrics()
metrics["success_rate"] = (
metrics["successful_requests"] / metrics["total_requests"] * 100
if metrics["total_requests"] > 0 else 0
)
return metrics
def close(self):
"""Ferme le client et libère les ressources"""
self._shutdown_event.set()
if self._worker_thread and self._worker_thread.is_alive():
self._worker_thread.join(timeout=5)
self._session.close()
Exemple d'utilisation
if __name__ == "__main__":
# Configuration
config = HolySheepClientConfig(
api_key="YOUR_HOLYSHEEP_API_KEY",
requests_per_second=10,
burst_size=20,
cb_failure_threshold=5
)
client = HolySheepAIClient(config)
try:
# Exemple d'appel
messages = [
{"role": "system", "content": "Tu es un assistant utile."},
{"role": "user", "content": "Explique le rate limiting en une phrase."}
]
response = client.chat_completions(
messages=messages,
model="gpt-4.1",
temperature=0.7,
max_tokens=150
)
print("=== Réponse ===")
print(response["choices"][0]["message"]["content"])
print("\n=== Métriques ===")
import json
print(json.dumps(client.get_metrics(), indent=2))
except CircuitBreakerOpen as e:
print(f"Circuit breaker ouvert: {e}")
except Exception as e:
print(f"Erreur: {e}")
finally:
client.close()Architecture de surveillance et monitoring
"""
HolySheep AI - Système de monitoring et alertes
Dashboard de supervision en temps réel
"""
import time
import threading
from dataclasses import dataclass, field
from typing import Dict, List
from collections import deque
import json
@dataclass
class MetricPoint:
"""Un point de métrique"""
timestamp: float
value: float
labels: Dict[str, str] = field(default_factory=dict)
class MetricsCollector:
"""
Collecteur de métriques temps réel
Collecte et agrège:
- Latence des requêtes
- Taux d'erreur
- Utilisation des tokens
- État des circuits
"""
def __init__(self, retention_seconds: int = 3600):
self.retention = retention_seconds
self._metrics: Dict[str, deque] = {}
self._lock = threading.Lock()
self._start_time = time.time()
def record(self, metric_name: str, value: float, labels: Dict[str, str] = None):
"""Enregistre une métrique"""
with self._lock:
if metric_name not in self._metrics:
self._metrics[metric_name] = deque(maxlen=10000)
point = MetricPoint(
timestamp=time.time(),
value=value,
labels=labels or {}
)
self._metrics[metric_name].append(point)
# Cleanup old points
cutoff = time.time() - self.retention
while self._metrics[metric_name] and self._metrics[metric_name][0].timestamp < cutoff:
self._metrics[metric_name].popleft()
def get_stats(self, metric_name: str, window_seconds: int = 60) -> Dict:
"""Calcule les statistiques pour une fenêtre de temps"""
with self._lock:
if metric_name not in self._metrics:
return {}
cutoff = time.time() - window_seconds
points = [
p for p in self._metrics[metric_name]
if p.timestamp >= cutoff
]
if not points:
return {"count": 0}
values = [p.value for p in points]
return {
"count": len(values),
"min": min(values),
"max": max(values),
"avg": sum(values) / len(values),
"p50": self._percentile(values, 50),
"p95": self._percentile(values, 95),
"p99": self._percentile(values, 99),
"window_seconds": window_seconds
}
def _percentile(self, values: List[float], percentile: int) -> float:
"""Calcule un percentile"""
sorted_values = sorted(values)
index = int(len(sorted_values) * percentile / 100)
return sorted_values[min(index, len(sorted_values) - 1)]
def get_all_metrics(self) -> Dict:
"""Retourne toutes les métriques agrégées"""
return {
"uptime_seconds": time.time() - self._start_time,
"latency": self.get_stats("latency_ms", 60),
"error_rate": self.get_stats("error", 60),
"tokens_used": self.get_stats("tokens", 60),
"rate_limit_hits": self.get_stats("rate_limit", 60)
}
def generate_prometheus_format(self) -> str:
"""Exporte les métriques au format Prometheus"""
lines = []
stats = self.get_all_metrics()
for metric_name, values in stats.items():
if isinstance(values, dict) and "avg" in values:
lines.append(f"# HELP holy_sheep_{metric_name} {metric_name}")
lines.append(f"# TYPE holy_sheep_{metric_name} gauge")
lines.append(f"holy_sheep_{metric_name}_avg {values['avg']:.2f}")
lines.append(f"holy_sheep_{metric_name}_p95 {values['p95']:.2f}")
return "\n".join(lines)
class AlertManager:
"""Gestionnaire d'alertes pour conditions critiques"""
def __init__(self, client: HolySheepAIClient):
self.client = client
self.alerts: List[Dict] = []
self._lock = threading.Lock()
def check_conditions(self):
"""Vérifie les conditions d'alerte"""
metrics = self.client.get_metrics()
alerts = []
# Circuit breaker ouvert
if metrics["circuit_breaker"]["state"] == "open":
alerts.append({
"severity": "critical",
"message": "Circuit breaker ouvert",
"circuit": metrics["circuit_breaker"]["name"]
})
# Taux d'erreur élevé
if metrics["successful_requests"] > 0:
error_rate = metrics["failed_requests"] / metrics["total_requests"]
if error_rate > 0.1:
alerts.append({
"severity": "warning",
"message": f"Taux d'erreur élevé: {error_rate*100:.1f}%"
})
with self._lock:
self.alerts = alerts
return alerts
Exemple de monitoring continu
if __name__ == "__main__":
collector = MetricsCollector()
config = HolySheepClientConfig(api_key="YOUR_HOLYSHEEP_API_KEY")
client = HolySheepAIClient(config)
alert_manager = AlertManager(client)
print("=== Monitoring HolySheep API ===")
print("Collecte des métriques pendant 10 secondes...\n")
start = time.time()
while time.time() - start < 10:
try:
# Simuler des appels API
messages = [{"role": "user", "content": "Test"}]
req_start = time.time()
response = client.chat_completions(messages, model="gpt-4.1")
latency = (time.time() - req_start) * 1000
collector.record("latency_ms", latency)
collector.record("success", 1)
if "usage" in response:
collector.record(
"tokens",
response["usage"].get("total_tokens", 0)
)
except Exception as e:
collector.record("error", 1)
print(f"Erreur: {e}")
time.sleep(0.5)
# Afficher les statistiques
print("\n=== Statistiques ===")
stats = collector.get_all_metrics()
print(json.dumps(stats, indent=2))
# Afficher les alertes
print("\n=== Alertes ===")
alerts = alert_manager.check_conditions()
if alerts:
for alert in alerts:
print(f"[{alert['severity'].upper()}] {alert['message']}")
else:
print("Aucune alerte")
# Export Prometheus
print("\n=== Format Prometheus ===")
print(collector.generate_prometheus_format())
client.close()Stratégies de déploiement en production
Configuration recommandée selon le use case
| Use Case | requests_per_second | burst_size | <
|---|