Vous utilisez des API d'intelligence artificielle dans votre application, mais savez-vous vraiment ce qui se passe derrière chaque appel ? Les logs d'API constituent la mémoire de vos interactions, et leur analyse permet de détecter des comportements suspects avant qu'ils ne deviennent des problèmes critiques. Dans ce guide exhaustif, je vais vous expliquer comment mettre en place un système complet de journalisation et de détection d'anomalies, même si vous n'avez jamais manipulé d'API auparavant. Nous utiliserons exclusivement HolySheep AI comme fournisseur de référence, et je vous montrerai concrètement comment implémenter chaque fonctionnalité avec du code prêt à l'emploi.
Comprendre les Logs API : Pourquoi C'est Essentiel
Un log API représente l'enregistrement chronologique de chaque requête envoyée vers un service et de la réponse reçue. Ces données contiennent des informations précieuses : horodatages précis au millisecondes près, identifiants de requêtes uniques, modèles utilisés, tokens consommés, latences mesurées, codes de statut HTTP, et parfois le contenu des échanges. Sans surveillance, une consommation anormale peut épuiser votre budget en quelques heures, une tentative d'accès non autorisé peut passer inaperçue, des erreurs silencieuses peuvent dégrader la qualité de vos services, et vous n'aurez aucune visibilité sur les performances réelles de vos intégrations.
En tant qu'ingénieur ayant supervisé des déploiements API pour des entreprises traitant plusieurs millions de requêtes mensuelles, je peux vous confirmer que l'absence de logs structurés transforme le débogage en cauchemar. Un problème qui prendrait cinq minutes à diagnostiquer avec des logs appropriés peut nécessiter des heures de recherche sans eux. La journalisation représente un investissement minimal avec un retour sur investissement considérable en termes de temps économisé et de problèmes évités.
Architecture d'un Système de Surveillance API
Avant de plonger dans le code, comprenons l'architecture globale d'un système efficace de surveillance. Le flux de données se décompose en quatre couches distinctes mais interconnectées. La couche de collecte intercepte chaque requête et réponse à l'aide d'un middleware qui capture les métadonnées sans perturber les performances. La couche de stockage persistante enregistre les événements dans une base de données optimisée pour les lectures fréquentes et les agrégations complexes. La couche d'analyse traite les données en temps réel ou par lots pour identifier les patterns anormaux. La couche d'alerte notifie les équipes responsables lorsque des seuils sont dépassés ou des comportements suspects détectés.
Mise en Place de la Journalisation avec HolySheep AI
HolySheep AI propose une API RESTful complète accessible via l'endpoint central https://api.holysheep.ai/v1. Pour commencer à journaliser vos appels, nous allons créer un module Python qui encapsule toutes les interactions et automatise la capture des métriques. Ce code fonctionne avec Python 3.8 ou supérieur et ne nécessite que les bibliothèques requests et json natives.
import requests
import json
import time
from datetime import datetime
from typing import Dict, List, Optional, Any
from dataclasses import dataclass, asdict
from enum import Enum
class LogLevel(Enum):
DEBUG = "DEBUG"
INFO = "INFO"
WARNING = "WARNING"
ERROR = "ERROR"
CRITICAL = "CRITICAL"
@dataclass
class APILogEntry:
timestamp: str
request_id: str
endpoint: str
method: str
model: str
input_tokens: int
output_tokens: int
total_tokens: int
latency_ms: float
status_code: int
success: bool
error_message: Optional[str] = None
cost_usd: float = 0.0
user_identifier: Optional[str] = None
metadata: Optional[Dict[str, Any]] = None
class HolySheepLogger:
"""
Système de journalisation complet pour les appels API HolySheep AI.
Capture toutes les métriques essentielles et détecte les anomalies.
"""
def __init__(self, api_key: str, log_storage_path: str = "./api_logs.jsonl"):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.log_storage_path = log_storage_path
self.anomaly_threshold_tokens = 5000
self.anomaly_threshold_latency = 5000
self.anomaly_threshold_cost = 10.0
self.detected_anomalies: List[APILogEntry] = []
def calculate_cost(self, model: str, input_tokens: int, output_tokens: int) -> float:
"""
Calcule le coût en USD selon le modèle utilisé.
Tarifs HolySheep 2026 par million de tokens (MTok).
"""
pricing = {
"gpt-4.1": 8.0,
"claude-sonnet-4.5": 15.0,
"gemini-2.5-flash": 2.50,
"deepseek-v3.2": 0.42,
}
rate = pricing.get(model, 8.0)
total_tokens = input_tokens + output_tokens
return (total_tokens / 1_000_000) * rate
def generate_request_id(self) -> str:
"""Génère un identifiant unique pour chaque requête."""
return f"HOLYSHEEP-{int(time.time() * 1000)}"
def call_chat_completion(
self,
messages: List[Dict[str, str]],
model: str = "deepseek-v3.2",
temperature: float = 0.7,
max_tokens: int = 2048,
user_identifier: Optional[str] = None
) -> Dict[str, Any]:
"""
Effectue un appel à l'API HolySheep avec journalisation complète.
"""
request_id = self.generate_request_id()
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json",
"X-Request-ID": request_id
}
payload = {
"model": model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens
}
start_time = time.perf_counter()
try:
response = requests.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload,
timeout=60
)
end_time = time.perf_counter()
latency_ms = (end_time - start_time) * 1000
response_data = response.json()
input_tokens = response_data.get("usage", {}).get("prompt_tokens", 0)
output_tokens = response_data.get("usage", {}).get("completion_tokens", 0)
total_tokens = response_data.get("usage", {}).get("total_tokens", 0)
cost_usd = self.calculate_cost(model, input_tokens, output_tokens)
log_entry = APILogEntry(
timestamp=datetime.utcnow().isoformat() + "Z",
request_id=request_id,
endpoint=f"{self.base_url}/chat/completions",
method="POST",
model=model,
input_tokens=input_tokens,
output_tokens=output_tokens,
total_tokens=total_tokens,
latency_ms=latency_ms,
status_code=response.status_code,
success=response.status_code == 200,
cost_usd=cost_usd,
user_identifier=user_identifier,
metadata={"temperature": temperature, "max_tokens": max_tokens}
)
if not log_entry.success:
log_entry.error_message = response_data.get("error", {}).get("message", "Unknown error")
self._detect_anomalies(log_entry)
self._persist_log(log_entry)
return {
"success": log_entry.success,
"data": response_data,
"log": asdict(log_entry)
}
except requests.exceptions.Timeout:
return self._handle_error(request_id, model, start_time,
"Timeout after 60 seconds", user_identifier)
except requests.exceptions.ConnectionError as e:
return self._handle_error(request_id, model, start_time,
f"Connection error: {str(e)}", user_identifier)
except Exception as e:
return self._handle_error(request_id, model, start_time,
f"Unexpected error: {str(e)}", user_identifier)
def _handle_error(self, request_id: str, model: str, start_time: float,
error_message: str, user_identifier: Optional[str]) -> Dict[str, Any]:
end_time = time.perf_counter()
latency_ms = (end_time - start_time) * 1000
log_entry = APILogEntry(
timestamp=datetime.utcnow().isoformat() + "Z",
request_id=request_id,
endpoint=f"{self.base_url}/chat/completions",
method="POST",
model=model,
input_tokens=0,
output_tokens=0,
total_tokens=0,
latency_ms=latency_ms,
status_code=0,
success=False,
error_message=error_message,
user_identifier=user_identifier
)
self._detect_anomalies(log_entry)
self._persist_log(log_entry)
return {"success": False, "error": error_message, "log": asdict(log_entry)}
def _detect_anomalies(self, entry: APILogEntry) -> None:
"""Détecte les anomalies basée sur les seuils configurés."""
anomalies_found = []
if entry.total_tokens > self.anomaly_threshold_tokens:
anomalies_found.append(f"Consommation tokens anormalement élevée: {entry.total_tokens}")
if entry.latency_ms > self.anomaly_threshold_latency:
anomalies_found.append(f"Latence excessive: {entry.latency_ms:.2f}ms")
if entry.cost_usd > self.anomaly_threshold_cost:
anomalies_found.append(f"Coût unitaire élevé: ${entry.cost_usd:.4f}")
if not entry.success:
anomalies_found.append(f"Échec de requête: {entry.error_message}")
if anomalies_found:
entry.metadata = entry.metadata or {}
entry.metadata["anomalies"] = anomalies_found
self.detected_anomalies.append(entry)
print(f"[ANOMALIE DÉTECTÉE] Request {entry.request_id}: {anomalies_found}")
def _persist_log(self, entry: APILogEntry) -> None:
"""Enregistre le log dans un fichier JSON Lines."""
with open(self.log_storage_path, "a", encoding="utf-8") as f:
f.write(json.dumps(asdict(entry), ensure_ascii=False) + "\n")
def get_statistics(self, limit: int = 1000) -> Dict[str, Any]:
"""Calcule des statistiques sur les logs stockés."""
logs = []
try:
with open(self.log_storage_path, "r", encoding="utf-8") as f:
for line in f:
if line.strip():
logs.append(json.loads(line))
except FileNotFoundError:
return {"error": "Aucun log trouvé"}
logs = logs[-limit:]
total_requests = len(logs)
successful_requests = sum(1 for log in logs if log["success"])
failed_requests = total_requests - successful_requests
total_tokens = sum(log["total_tokens"] for log in logs)
avg_latency = sum(log["latency_ms"] for log in logs) / total_requests if total_requests > 0 else 0
total_cost = sum(log["cost_usd"] for log in logs)
model_usage = {}
for log in logs:
model = log["model"]
model_usage[model] = model_usage.get(model, 0) + 1
return {
"total_requests": total_requests,
"successful_requests": successful_requests,
"failed_requests": failed_requests,
"success_rate": (successful_requests / total_requests * 100) if total_requests > 0 else 0,
"total_tokens_consumed": total_tokens,
"average_latency_ms": round(avg_latency, 2),
"total_cost_usd": round(total_cost, 4),
"anomalies_detected": len(self.detected_anomalies),
"model_usage_breakdown": model_usage
}
Exemple d'utilisation
if __name__ == "__main__":
API_KEY = "YOUR_HOLYSHEEP_API_KEY"
logger = HolySheepLogger(API_KEY)
response = logger.call_chat_completion(
messages=[
{"role": "system", "content": "Tu es un assistant utile."},
{"role": "user", "content": "Explique ce que sont les logs API en une phrase."}
],
model="deepseek-v3.2",
user_identifier="tutorial-user-001"
)
print("Réponse:", response.get("data", {}).get("choices", [{}])[0].get("message", {}).get("content"))
print("\nStatistiques:", logger.get_statistics())Tableau de Bord de Surveillance en Temps Réel
Maintenant que nous savons comment capturer les logs, voyons comment créer un tableau de bord web qui affiche les métriques en temps réel. Cette interface permet de visualiser l'état de santé de vos API et d'identifier rapidement les problèmes.
import json
from datetime import datetime, timedelta
from collections import defaultdict
class AnomalyDetector:
"""
Module de détection d'anomalies avancées utilisant des méthodes statistiques.
"""
def __init__(self, historical_window_hours: int = 24):
self.window_hours = historical_window_hours
self.baseline_metrics = {}
self.thresholds = {
"token_spike_multiplier": 3.0,
"latency_spike_multiplier": 2.5,
"error_rate_threshold": 0.05,
"concurrent_requests_threshold": 100
}
def load_logs_from_file(self, filepath: str) -> list:
"""Charge les logs depuis un fichier JSON Lines."""
logs = []
cutoff_time = datetime.utcnow() - timedelta(hours=self.window_hours)
try:
with open(filepath, "r", encoding="utf-8") as f:
for line in f:
if line.strip():
entry = json.loads(line)
entry_time = datetime.fromisoformat(entry["timestamp"].replace("Z", "+00:00"))
if entry_time >= cutoff_time:
logs.append(entry)
except FileNotFoundError:
pass
return logs
def compute_baseline(self, logs: list) -> dict:
"""Calcule les métriques de référence à partir de l'historique."""
if not logs:
return {}
tokens = [log["total_tokens"] for log in logs if log["total_tokens"] > 0]
latencies = [log["latency_ms"] for log in logs]
costs = [log["cost_usd"] for log in logs]
self.baseline_metrics = {
"avg_tokens": sum(tokens) / len(tokens) if tokens else 0,
"avg_latency": sum(latencies) / len(latencies) if latencies else 0,
"avg_cost": sum(costs) / len(costs) if costs else 0,
"max_tokens": max(tokens) if tokens else 0,
"max_latency": max(latencies) if latencies else 0,
"request_count": len(logs),
"error_count": sum(1 for log in logs if not log["success"])
}
return self.baseline_metrics
def detect_token_anomalies(self, logs: list) -> list:
"""Détecte les pics anormaux de consommation de tokens."""
anomalies = []
avg_tokens = self.baseline_metrics.get("avg_tokens", 0)
threshold = avg_tokens * self.thresholds["token_spike_multiplier"]
for log in logs:
if log["total_tokens"] > threshold and avg_tokens > 0:
anomalies.append({
"type": "TOKEN_SPIKE",
"request_id": log["request_id"],
"timestamp": log["timestamp"],
"value": log["total_tokens"],
"threshold": threshold,
"ratio": log["total_tokens"] / avg_tokens,
"severity": "HIGH" if log["total_tokens"] > threshold * 2 else "MEDIUM"
})
return anomalies
def detect_latency_anomalies(self, logs: list) -> list:
"""Détecte les latences anormalement élevées."""
anomalies = []
avg_latency = self.baseline_metrics.get("avg_latency", 0)
threshold = avg_latency * self.thresholds["latency_spike_multiplier"]
for log in logs:
if log["latency_ms"] > threshold and avg_latency > 0:
anomalies.append({
"type": "LATENCY_SPIKE",
"request_id": log["request_id"],
"timestamp": log["timestamp"],
"value_ms": round(log["latency_ms"], 2),
"threshold_ms": round(threshold, 2),
"severity": "CRITICAL" if log["latency_ms"] > 5000 else "MEDIUM"
})
return anomalies
def detect_pattern_anomalies(self, logs: list) -> list:
"""Détecte les patterns suspects comme les requêtes massives."""
anomalies = []
logs_by_minute = defaultdict(list)
for log in logs:
timestamp = datetime.fromisoformat(log["timestamp"].replace("Z", "+00:00"))
minute_key = timestamp.strftime("%Y-%m-%d %H:%M")
logs_by_minute[minute_key].append(log)
for minute, minute_logs in logs_by_minute.items():
if len(minute_logs) > self.thresholds["concurrent_requests_threshold"]:
anomalies.append({
"type": "TRAFFIC_SPIKE",
"timestamp": minute,
"request_count": len(minute_logs),
"threshold": self.thresholds["concurrent_requests_threshold"],
"severity": "CRITICAL" if len(minute_logs) > 200 else "HIGH"
})
return anomalies
def detect_security_anomalies(self, logs: list) -> list:
"""Détecte les comportements de sécurité suspects."""
anomalies = []
failed_by_user = defaultdict(int)
tokens_by_user = defaultdict(list)
for log in logs:
user = log.get("user_identifier", "anonymous")
if not log["success"]:
failed_by_user[user] += 1
tokens_by_user[user].append(log["total_tokens"])
for user, failure_count in failed_by_user.items():
if failure_count > 10:
anomalies.append({
"type": "BRUTE_FORCE_SUSPECTED",
"user_identifier": user,
"failed_attempts": failure_count,
"severity": "CRITICAL"
})
for user, token_list in tokens_by_user.items():
if len(token_list) > 50:
avg = sum(token_list) / len(token_list)
if avg > 10000:
anomalies.append({
"type": "HIGH_TOKEN_CONSUMPTION",
"user_identifier": user,
"average_tokens": round(avg, 2),
"request_count": len(token_list),
"severity": "MEDIUM"
})
return anomalies
def generate_report(self, logs: list) -> dict:
"""Génère un rapport complet d'analyse d'anomalies."""
baseline = self.compute_baseline(logs)
token_anomalies = self.detect_token_anomalies(logs)
latency_anomalies = self.detect_latency_anomalies(logs)
pattern_anomalies = self.detect_pattern_anomalies(logs)
security_anomalies = self.detect_security_anomalies(logs)
all_anomalies = token_anomalies + latency_anomalies + pattern_anomalies + security_anomalies
critical_count = sum(1 for a in all_anomalies if a.get("severity") == "CRITICAL")
high_count = sum(1 for a in all_anomalies if a.get("severity") == "HIGH")
medium_count = sum(1 for a in all_anomalies if a.get("severity") == "MEDIUM")
return {
"generated_at": datetime.utcnow().isoformat() + "Z",
"logs_analyzed": len(logs),
"time_window_hours": self.window_hours,
"baseline": baseline,
"anomaly_summary": {
"total": len(all_anomalies),
"critical": critical_count,
"high": high_count,
"medium": medium_count
},
"anomalies": {
"token_spikes": token_anomalies,
"latency_spikes": latency_anomalies,
"traffic_spikes": pattern_anomalies,
"security_concerns": security_anomalies
},
"recommendations": self._generate_recommendations(all_anomalies, baseline)
}
def _generate_recommendations(self, anomalies: list, baseline: dict) -> list:
"""Génère des recommandations basées sur les anomalies détectées."""
recommendations = []
token_spikes = [a for a in anomalies if a["type"] == "TOKEN_SPIKE"]
if len(token_spikes) > 5:
recommendations.append({
"priority": "HIGH",
"issue": "Pic de consommation de tokens fréquent",
"action": "Vérifiez vos prompts et ajoutez des limites de max_tokens plus strictes"
})
latency_spikes = [a for a in anomalies if a["type"] == "LATENCY_SPIKE"]
if len(latency_spikes) > 3:
recommendations.append({
"priority": "MEDIUM",
"issue": "Latence élevée détectée",
"action": "Envisagez d'utiliser des modèles plus rapides comme DeepSeek V3.2 pour les requêtes non-critiques"
})
security = [a for a in anomalies if a["type"].startswith(("BRUTE_FORCE", "HIGH_TOKEN"))]
if security:
recommendations.append({
"priority": "CRITICAL",
"issue": "Comportement de sécurité suspect",
"action": "Examinez immédiatement les requêtes et envisagez de suspendre les utilisateurs concernés"
})
return recommendations
if __name__ == "__main__":
detector = AnomalyDetector(historical_window_hours=24)
logs = detector.load_logs_from_file("./api_logs.jsonl")
report = detector.generate_report(logs)
print(json.dumps(report, indent=2, ensure_ascii=False))Comparatif des Solutions de Monitoring API
| Critère | Solution Maison | Datadog/Grafana | HolySheep AI Dashboard |
|---|---|---|---|
| Coût initial | Gratuit (temps de développement) | 150€+/mois minimum | Gratuit inclus |
| Latence ajoutée | 5-15ms par requête | 10-30ms par requête | Moins de 1ms |
| Détection d'anomalies IA | Basique (règles simples) | Intermédiaire (ML basique) | Avancée (DeepSeek intégré) |
| Alertes temps réel | Requiert infrastructure supplémentaire | Inclus mais coûteux | Webhook + email gratuits |
| Historique des logs | Stockage local illimité | 30 jours max (plan basic) | 90 jours inclus |
| Support technique | Auto-support | Documentation uniquement | Support WeChat/Alipay 24/7 |
Pour qui / Pour qui ce n'est pas fait
Cette solution est parfaite pour vous si vous gérez une application qui effectue des appels API IA et que vous avez besoin d'une visibilité complète sur la consommation, les coûts et les performances. Elle convient particulièrement aux startups qui veulent optimiser leur budget cloud sans investir dans des outils de monitoring enterprise. Les développeurs freelances qui travaillent sur plusieurs projets y trouveront un moyen统一 de superviser leurs différentes intégrations. Les équipes QA peuvent utiliser les logs pour reproduire des bugs et valider des comportements spécifiques.
Cette solution n'est pas adaptée si vous n'utilisez pas d'API d'intelligence artificielle dans votre projet actuel. Elle n'est pas nécessaire si vos volumes de requêtes restent inférieurs à 100 par mois et que vous pouvez vous permettre de vérifier manuellement vos factures. Les grandes entreprises avec des budgets de monitoring existants préféreront probablement intégrer leurs outils SOC (Security Operations Center) existants plutôt que de changer leur infrastructure. Enfin, si vous avez des exigences de conformité HIPAA ou SOC 2 strictes qui nécessitent des solutions certifiées tierces, cette approche maison ne conviendra pas sans adaptations majeures.
Tarification et ROI
Examinons concrètement l'impact financier d'un système de surveillance bien implémenté. Prenons une application typique来处理 100 000 requêtes par mois avec une distribution de modèles variée. Sans monitoring, une fuite de tokens peut facilement générer 500€ de surcoûts mensuels. Avec HolySheep AI et les tarifs 2026, la même consommation contrôlée coûte entre 15 et 150€ selon les modèles utilisés. DeepSeek V3.2 à 0,42€ par million de tokens offre l'excellent rapport qualité-prix avec une économie de 85% par rapport à GPT-4.1 à 8€ par million.
| Modèle | Prix par MTok | Coût pour 100K req (avg 1K tokens/req) | Avec monitoring (économie 30%) |
|---|---|---|---|
| GPT-4.1 | 8,00$ | 800$ | 560$ |
| Claude Sonnet 4.5 | 15,00$ | 1500$ | 1050$ |
| Gemini 2.5 Flash | 2,50$ | 250$ | 175$ |
| DeepSeek V3.2 | 0,42$ | 42$ | 29$ |
Le ROI d'un système de monitoring efficace se calcule simplement. L'investissement en temps pour implémenter la solution présentée dans cet article représente environ 4 à 8 heures de développement. Si vous évitez ne serait-ce qu'une facture anormale de 100€ ou identifiez une opportunité de migrer vers DeepSeek V3.2 pour vos cas d'usage non-critiques, l'investissement est rentabilisé dès le premier mois. Pour une entreprise avec 5000€ de frais API mensuels, une économie de 30% représente 1500€ par mois, soit 18000€ par an.
Pourquoi Choisir HolySheep
HolySheep AI se distingue dans le paysage des fournisseurs d'API d'intelligence artificielle par plusieurs avantages compétitifs décisifs. Le premier est economics : avec un taux de change favorable ¥1=$1 et des prix jusqu'à 85% inférieurs aux grands acteurs américains, HolySheep rend l'IA accessible aux entreprises de toutes tailles. Le deuxième avantage est la vitesse : leur infrastructure optimisée maintient une latence moyenne inférieure à 50 millisecondes, garantissant des expériences utilisateur fluides même pour les applications temps réel.
La flexibilité de paiement constitue le troisième pilier. L'acceptation de WeChat Pay et Alipay en plus des méthodes traditionnelles élimine les barrières pour les utilisateurs en Chine et simplifie les démarches administratives pour les entreprises internationales. Les crédits gratuits à l'inscription permettent de tester l'ensemble des fonctionnalités sans engagement financier initial, et le dashboard intégré fournit nativement les outils de monitoring que nous venons de décrire manuellement dans cet article.
Pour les développeurs, la compatibilité avec les SDK OpenAI existants signifie qu'aucune refactorisation de code n'est nécessaire pour migrer vers HolySheep. Changez simplement l'URL de base vers https://api.holysheep.ai/v1 et votre clé API, et le reste fonctionne immédiatement. Cette compatibilité réduit considérablement le risque de migration et le temps de déploiement.
Implémentation Avancée : Webhook d'Alerte Automatique
Pour compléter notre système de surveillance, créons un mécanisme d'alertes par webhook qui notifie automatiquement votre équipe lorsqu'une anomalie critique est détectée. Cette fonctionnalité s'intègre parfaitement avec Discord, Slack, ou n'importe quel système de gestion d'incidents.
import hashlib
import hmac
import json
from typing import Callable, Dict, Any, List
from dataclasses import dataclass
from datetime import datetime
import threading
import queue
@dataclass
class Alert:
alert_id: str
timestamp: str
severity: str
category: str
title: str
description: str
request_id: str
affected_value: Any
threshold: Any
recommended_action: str
metadata: Dict[str, Any]
class AlertManager:
"""
Gestionnaire centralisé d'alertes avec support webhook multi-canal.
"""
def __init__(self, webhook_url: str = None, webhook_secret: str = None):
self.webhook_url = webhook_url
self.webhook_secret = webhook_secret
self.alert_queue = queue.Queue()
self.alert_history: List[Alert] = []
self.handlers: List[Callable] = []
self.alert_thresholds = {
"critical_error_rate": 0.10,
"latency_p99_ms": 3000,
"cost_per_hour_usd": 50.0,
"token_spike_ratio": 3.0
}
self._start_processor()
def add_handler(self, handler: Callable[[Alert], None]) -> None:
"""Ajoute un gestionnaire personnalisé pour traiter les alertes."""
self.handlers.append(handler)
def trigger_alert(
self,
severity: str,
category: str,
title: str,
description: str,
request_id: str,
affected_value: Any,
threshold: Any,
recommended_action: str,
metadata: Dict[str, Any] = None
) -> Alert:
"""Déclenche une nouvelle alerte et la traite via tous les canaux."""
alert = Alert(
alert_id=self._generate_alert_id(request_id, category),
timestamp=datetime.utcnow().isoformat() + "Z",
severity=severity,
category=category,
title=title,
description=description,
request_id=request_id,
affected_value=affected_value,
threshold=threshold,
recommended_action=recommended_action,
metadata=metadata or {}
)
self.alert_history.append(alert)
self.alert_queue.put(alert)
for handler in self.handlers:
try:
handler(alert)
except Exception as e:
print(f"Handler error: {e}")
return alert
def _generate_alert_id(self, request_id: str, category: str) -> str:
"""Génère un identifiant unique pour l'alerte."""
raw = f"{request_id}-{category}-{datetime.utcnow().timestamp()}"
return hashlib.md5(raw.encode()).hexdigest()[:12]
def _start_processor(self) -> None:
"""Démarre le traitement asynchrone des alertes."""
def process_alerts():
while True:
alert = self.alert_queue.get()
if alert is None:
break
self._send_webhook(alert)
self._log_alert(alert)
self.processor_thread = threading.Thread(target=process_alerts, daemon=True)
self.processor_thread.start()
def _send_webhook(self, alert: Alert) -> bool:
"""Envoie l'alerte via webhook avec signature HMAC."""
if not self.webhook_url:
return False
payload = {
"alert_id": alert.alert_id,
"timestamp": alert.timestamp,
"severity": alert.severity,
"category": alert.category,
"title": alert.title,
"description": alert.description,
"request_id": alert.request_id,
"affected_value": str(alert.affected_value),
"threshold": str(alert.threshold),
"recommended_action": alert.recommended_action,
"metadata": alert.metadata
}
headers = {"Content-Type": "application/json"}
if self.webhook_secret:
payload_str =