Maîtriser les Architectures Événementielles : Construire des Systèmes Réactifs et Scalables
Maîtriser les Architectures Événementielles : Construire des Systèmes Réactifs et Scalables

Gestion des Erreurs, Résilience et Observabilité dans les Architectures Événementielles

Contexte du cours : Maîtriser les Architectures Événementielles : Construire des Systèmes Réactifs et Scalables

Introduction : Naviguer dans le Monde Asynchrone des EDA

Les architectures événementielles (Event-Driven Architectures - EDA) sont devenues la pierre angulaire des systèmes modernes, réactifs et hautement scalables. En découplant les services via l'échange d'événements, elles offrent une flexibilité et une agilité considérables. Cependant, cette distribution et cet asynchronisme introduisent de nouveaux défis en matière de gestion des erreurs, de résilience et d'observabilité.

Contrairement aux architectures monolithiques où les erreurs peuvent souvent être gérées de manière synchrone au sein d'une seule transaction, les EDAs exigent une approche fondamentalement différente. Un événement peut déclencher une cascade d'opérations à travers de multiples services, potentiellement sur différents nœuds et dans des laps de temps variables. Comment s'assurer que le système reste robuste face aux pannes, qu'il puisse se remettre de situations inattendues et que nous puissions comprendre son comportement interne complexe ? C'est précisément l'objectif de cette leçon.

Nous explorerons comment ces trois piliers – la gestion des erreurs, la résilience et l'observabilité – sont non seulement interdépendants mais absolument essentiels pour construire et maintenir des systèmes événementiels fiables et performants.

Les Défis Spécifiques des Architectures Événementielles

Avant de plonger dans les solutions, il est crucial de comprendre les problèmes uniques que les EDAs posent :

  • Asynchronisme et Distribution : Les opérations ne se produisent pas séquentiellement ou dans le même processus. Un producteur d'événements ne sait pas si son événement sera traité avec succès par tous les consommateurs.
  • Absence de Pile d'Appels Directe : Lorsqu'un événement échoue chez un consommateur, il n'y a pas de trace de pile d'appels directe vers le producteur pour remonter l'erreur.
  • Dépendances Implicites : Les services sont faiblement couplés en surface, mais fortement couplés au schéma d'événements et aux conséquences des événements. Une modification ou une panne dans un service peut avoir des effets inattendus ailleurs.
  • Difficulté à Reproduire les États : La nature distribuée et le mouvement constant des données rendent difficile la reproduction d'un état spécifique du système pour diagnostiquer une erreur.
  • Latence et Incohérence Éventuelle : Les événements peuvent mettre du temps à être traités, et les différents services peuvent avoir des vues temporairement différentes de l'état global (consistance éventuelle).

Ces défis soulignent la nécessité d'une stratégie proactive et sophistiquée pour garantir la robustesse et la maintenabilité des EDAs.

Gestion des Erreurs : Anticiper et Réagir aux Imprévus

La gestion des erreurs dans les EDAs ne consiste pas seulement à try-catch ; c'est une stratégie holistique pour identifier, classer et réagir aux défaillances de manière appropriée, sans perturber le flux général du système.

Stratégies de Gestion des Erreurs

Il est essentiel de distinguer les types d'erreurs pour appliquer la bonne stratégie :

  • Erreurs Temporaires (Transient Errors) : Ce sont des erreurs qui peuvent se résoudre d'elles-mêmes après un court laps de temps, comme des problèmes réseau temporaires, une base de données temporairement indisponible, ou une surcharge de service.
    • Stratégie : Retenter l'opération (avec prudence).
  • Erreurs Permanentes (Permanent Errors) : Ce sont des erreurs qui ne se résoudront pas d'elles-mêmes, comme des données métier invalides, un schéma d'événement incorrect, une logique métier erronée, ou un service dépendant définitivement indisponible.
    • Stratégie : Échouer rapidement, enregistrer l'erreur, et potentiellement déplacer le message vers une file d'attente dédiée pour une analyse manuelle.
  • Erreurs Techniques vs. Erreurs Métier :
    • Techniques : Liées à l'infrastructure (DB down, réseau, mémoire).
    • Métier : Liées à la logique applicative (commande invalide, utilisateur inexistant). La gestion des erreurs métier peut nécessiter des transactions compensatoires.

Patterns de Gestion des Erreurs

Pour aborder ces défis, plusieurs patterns ont émergé :

1. Dead Letter Queue (DLQ - File d'Attente des Messages Morts)

Une DLQ est une file d'attente spéciale où les messages qui ne peuvent pas être traités avec succès sont déplacés.

  • Fonctionnement : Lorsqu'un consommateur échoue de manière répétée à traiter un message (après un certain nombre de tentatives), ou s'il rencontre une erreur permanente, le message est automatiquement ou manuellement déplacé vers la DLQ.
  • Avantages :
    • Empêche les messages défectueux de bloquer la file d'attente principale.
    • Permet l'inspection, la correction et le re-traitement manuel des messages.
    • Isole les erreurs, améliorant la résilience du système principal.
  • Cas d'usage : Messages mal formés, erreurs de validation métier non anticipées, dépendances externes inaccessibles de manière permanente.

2. Retries (Nouvelles Tentatives)

La stratégie de re-tentative consiste à essayer de traiter un message plusieurs fois avant de le considérer comme défaillant.

  • Types de Retries :
    • Simple : Réessayer immédiatement après un échec. Généralement déconseillé car peut aggraver une situation de surcharge.
    • Avec Backoff Exponentiel : Augmenter progressivement le délai entre chaque tentative (ex: 1s, 2s, 4s, 8s...). Cela donne au système dépendant plus de temps pour récupérer et réduit la charge sur celui-ci.
    • Avec Jitter : Ajouter une petite variation aléatoire au délai de backoff exponentiel pour éviter que tous les consommateurs ne réessayent en même temps, ce qui pourrait provoquer un "thundering herd problem".
  • Considérations :
    • Limiter le nombre maximal de tentatives pour éviter les boucles infinies.
    • S'assurer que l'opération est idempotente (voir ci-dessous).
  • Cas d'usage : Erreurs réseau intermittentes, verrous de base de données temporaires, microservices temporairement indisponibles.

3. Idempotence

Une opération est idempotente si elle peut être appliquée plusieurs fois sans modifier le résultat au-delà de la première application. C'est crucial pour les retries et le re-traitement des messages dans les EDAs.

  • Pourquoi ? Si un message est re-traité, il ne doit pas créer d'effets secondaires indésirables (ex: facturer un client deux fois, créer des doublons de commandes).
  • Implémentation :
    • Utiliser un identifiant unique (clé d'idempotence) associé à chaque opération.
    • Vérifier cet identifiant avant d'appliquer l'opération. Si l'identifiant a déjà été traité, ignorer ou retourner le résultat précédent.
  • Exemple : Lors de la création d'une commande, un order_id unique généré par le client peut servir de clé d'idempotence. Si une requête de création de commande avec le même order_id est reçue à nouveau, le système ignore la création et renvoie l'état de la commande existante.

4. Transactions Compensatoires (Compensating Transactions)

Lorsque des opérations métier distribuées échouent après avoir effectué des actions irréversibles (par exemple, un paiement a été traité mais la livraison a échoué), les transactions compensatoires permettent de "défaire" les actions précédentes.

  • Fonctionnement : Chaque étape d'une transaction distribuée est conçue pour avoir une action compensatoire correspondante. Si une étape échoue, les actions compensatoires des étapes précédentes sont exécutées pour ramener le système à un état cohérent.
  • Pattern : Souvent associé au Saga Pattern, où une série d'opérations locales sont gérées par un coordinateur. Si une opération échoue, le saga déclenche les opérations de compensation nécessaires.
  • Cas d'usage : Processus de commande complexe (paiement, stock, livraison, notification), réservation de voyages (vol, hôtel, voiture).

Résilience : S'Adapter et Survivre aux Pannes

La résilience est la capacité d'un système à récupérer des défaillances et à continuer de fonctionner, même de manière dégradée. Dans les EDAs, cela signifie concevoir des services qui peuvent tolérer les pannes des autres services et de l'infrastructure.

Qu'est-ce que la Résilience ?

Un système résilient ne prévient pas nécessairement toutes les pannes, mais il est conçu pour les anticiper, les contenir et s'en remettre rapidement, minimisant ainsi l'impact sur l'utilisateur final.

Patterns de Résilience

Voici quelques patterns clés pour construire des systèmes événementiels résilients :

1. Circuit Breaker (Disjoncteur)

Le pattern Circuit Breaker empêche un service d'essayer continuellement de se connecter à un service défaillant, ce qui pourrait épuiser les ressources et propager la panne.

  • Fonctionnement :
    • Closed (Fermé) : Les appels au service cible passent normalement. Si les échecs dépassent un seuil, le circuit s'ouvre.
    • Open (Ouvert) : Les appels sont immédiatement refusés, et une erreur est retournée sans essayer le service cible. Après un délai configuré, le circuit passe en Half-Open.
    • Half-Open (Semi-Ouvert) : Quelques appels sont autorisés à passer pour tester si le service cible a récupéré. Si ces appels réussissent, le circuit retourne à l'état Closed. Sinon, il revient à l'état Open.
  • Avantages :
    • Évite la surcharge des services défaillants.
    • Préserve les ressources du service appelant.
    • Permet au service défaillant de récupérer sans être constamment sollicité.
  • Implémentation : Souvent implémenté via des bibliothèques comme Hystrix (Java, déprécié), Resilience4j (Java) ou Polly (.NET).

2. Bulkhead (Cloison Étanches)

Inspiré des cloisons étanches des navires, ce pattern isole les composants d'un système de sorte qu'une panne dans l'un ne puisse pas couler le navire entier.

  • Fonctionnement : Allouer des pools de ressources (threads, connexions, mémoire) séparés pour différents types d'appels ou de clients. Si un pool est épuisé ou rencontre une erreur, cela n'affecte pas les autres.
  • Avantages :
    • Contient les pannes à des sections spécifiques du système.
    • Empêche les échecs en cascade.
    • Améliore la disponibilité globale.
  • Cas d'usage : Utiliser des pools de threads séparés pour différents services externes, isoler les consommateurs de différents types d'événements.

3. Timeout (Délai d'Attente)

Fixer une limite de temps maximale pour la complétion d'une opération. Si l'opération ne se termine pas dans le temps imparti, elle est annulée.

  • Fonctionnement : Empêche les threads et les requêtes de rester bloqués indéfiniment en attendant une réponse d'un service lent ou non réactif.
  • Avantages :
    • Libère des ressources.
    • Empêche les goulots d'étranglement et les retards prolongés.
    • Peut être combiné avec les retries ou un fallback.

4. Rate Limiting (Limitation de Débit)

Contrôler le nombre de requêtes qu'un service peut traiter sur une période donnée afin d'éviter la surcharge et de garantir la stabilité.

  • Fonctionnement : Les requêtes excédentaires sont rejetées ou mises en file d'attente. Peut être appliqué côté client ou serveur.
  • Avantages :
    • Protège les services contre les pics de trafic inattendus.
    • Assure une performance stable pour les requêtes légitimes.
    • Utile pour protéger les services externes payants ou sensibles.

5. Fallback (Stratégie de Repli)

Lorsqu'une opération échoue (après retries, timeouts, ou circuit breaker ouvert), fournir une réponse alternative ou un comportement dégradé plutôt que de planter entièrement.

  • Fonctionnement : Au lieu de retourner une erreur à l'utilisateur, le système peut renvoyer des données en cache, une valeur par défaut, ou une réponse partielle.
  • Avantages :
    • Améliore l'expérience utilisateur en fournissant une réponse, même si elle n'est pas optimale.
    • Maintient la fonctionnalité essentielle du système.
  • Exemple : Si le service de recommandation est en panne, afficher des produits populaires par défaut.

Observabilité : Comprendre l'Incompréhensible

Dans un système distribué et asynchrone comme une EDA, il est presque impossible de comprendre ce qui se passe simplement en exécutant le code. L'observabilité est la capacité de déduire l'état interne du système à partir des données qu'il génère (logs, métriques, traces). Elle est cruciale pour diagnostiquer les problèmes, comprendre les flux d'événements et optimiser les performances.

Qu'est-ce que l'Observabilité ? (vs. Monitoring)

  • Monitoring : Vous savez ce que vous cherchez. Vous définissez des seuils et des alertes pour des métriques connues (CPU, mémoire, requêtes/seconde). C'est pour les "known unknowns".
  • Observabilité : Vous pouvez poser des questions arbitraires sur votre système sans avoir à déployer de nouveau code. C'est pour les "unknown unknowns". Elle vous donne les outils pour explorer et comprendre des comportements imprévus.

Les Piliers de l'Observabilité

L'observabilité repose sur trois types de données télémétriques :

1. Logs (Journaux)

Les logs enregistrent des événements discrets qui se sont produits dans le système.

  • Dans les EDAs : Les logs doivent être structurés (JSON est idéal) et inclure des identifiants de corrélation (Correlation ID) pour relier les événements à travers les différents services.
  • Bonnes pratiques :
    • Inclure des informations contextuelles (ID de l'événement, ID de l'utilisateur, nom du service).
    • Utiliser des niveaux de log appropriés (DEBUG, INFO, WARN, ERROR, FATAL).
    • Centraliser les logs (ex: ELK Stack, Grafana Loki) pour une recherche et une analyse efficaces.

2. Metrics (Métriques)

Les métriques sont des valeurs numériques agrégées au fil du temps, utilisées pour suivre la santé et les performances du système.

  • Types :
    • Compteurs : Augmentent seulement (nombre d'événements traités, nombre d'erreurs).
    • Jauges : Valeurs qui peuvent monter et descendre (taille de la file d'attente, utilisation CPU).
    • Histogrammes/Summaries : Mesurent la distribution d'une série de valeurs (latence de traitement des événements).
  • Golden Signals pour les Services :
    • Latence : Temps de traitement des événements.
    • Trafic : Nombre d'événements par seconde.
    • Erreurs : Taux d'événements traités avec erreur.
    • Saturation : Charge des ressources (CPU, mémoire, file d'attente).
  • Outils : Prometheus, Grafana, Datadog, New Relic.

3. Traces Distribuées (Distributed Tracing)

Les traces distribuées permettent de suivre le chemin d'une requête ou d'un événement à travers tous les services participants dans une architecture distribuée.

  • Fonctionnement : Chaque opération (appel API, traitement d'événement) est un span. Les spans sont liés entre eux par un Trace ID unique et un Parent Span ID pour former une trace complète.
  • Dans les EDAs : Lorsqu'un service publie un événement, il doit inclure le Trace ID et potentiellement son Span ID dans les métadonnées de l'événement. Le service consommateur extrait ces IDs et les utilise pour démarrer de nouveaux spans, créant ainsi une chaîne de traçabilité à travers les microservices.
  • Avantages :
    • Visualiser le flux complet d'un événement.
    • Identifier les goulots d'étranglement et les services lents.
    • Faciliter le diagnostic des erreurs distribuées.
  • Outils : OpenTelemetry (standardisation), Jaeger, Zipkin, AWS X-Ray, Google Cloud Trace.

Implémentation de l'Observabilité dans les EDA

La clé de l'observabilité dans les EDAs est la propagation du contexte.

  • Correlation ID (ID de Corrélation) : Un ID unique généré au début d'un flux d'événements et propagé dans tous les événements et logs subséquents. Cela permet de regrouper tous les logs et les traces liés à un même processus métier.
  • Trace ID / Span ID : Utilisés spécifiquement pour le tracing distribué, souvent gérés par des bibliothèques dédiées (ex: OpenTelemetry).

Un producteur d'événements doit injecter ces IDs dans les en-têtes ou le payload des messages. Chaque consommateur doit extraire ces IDs et les propager dans tout nouvel événement qu'il publie ou dans les logs qu'il génère.

Exemple Pratique : Gestion d'Erreurs et Tracing Conceptuel

Illustrons certains concepts avec des extraits de code Python.

Code 1 : Consommateur d'Événements avec Retries et DLQ Conceptuelle

Cet exemple simule un consommateur d'événements qui essaie de traiter un message. En cas d'échec temporaire, il retente avec un backoff exponentiel. Après un nombre défini de tentatives, il envoie le message vers une DLQ conceptuelle.

import time
import random
import json
import logging

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

class EventConsumer:
    def __init__(self, max_retries=3, initial_backoff=1):
        self.max_retries = max_retries
        self.initial_backoff = initial_backoff
        logging.info(f"Consumer initialized with max_retries={max_retries}, initial_backoff={initial_backoff}s")

    def _process_event_logic(self, event_data):
        """Simule la logique de traitement d'un événement.
        Peut échouer aléatoirement pour la démonstration.
        """
        # Simule une erreur temporaire 70% du temps
        if random.random() < 0.7:
            raise ConnectionError("Simulated temporary database connection error.")
        
        # Simule une erreur permanente 10% du temps
        if event_data.get("type") == "invalid":
            raise ValueError("Simulated permanent invalid event data.")

        logging.info(f"Successfully processed event: {event_data.get('id')}")
        return True

    def send_to_dlq(self, event_message, reason):
        """Simule l'envoi d'un message à une Dead Letter Queue."""
        logging.error(f"Event failed after all retries or due to permanent error. Moving to DLQ. Reason: {reason}. Message: {event_message}")
        # Dans un vrai système, vous publieriez ceci vers une file d'attente DLQ de votre broker
        # ex: self.broker.publish_to_dlq(event_message)

    def consume_message(self, event_message):
        event_data = json.loads(event_message['payload'])
        event_id = event_data.get('id', 'N/A')
        correlation_id = event_message.get('correlation_id', 'N/A')

        logging.info(f"[{correlation_id}] Attempting to process event ID: {event_id}")

        current_backoff = self.initial_backoff
        for attempt in range(1, self.max_retries + 2): # +1 pour la tentative initiale + les retries
            try:
                self._process_event_logic(event_data)
                logging.info(f"[{correlation_id}] Event ID: {event_id} processed successfully on attempt {attempt}.")
                return True
            except ValueError as e: # Erreur permanente
                logging.error(f"[{correlation_id}] Permanent error for event ID: {event_id}. Reason: {e}. Moving to DLQ.")
                self.send_to_dlq(event_message, f"Permanent error: {e}")
                return False
            except Exception as e: # Autres erreurs (temporaires ou inattendues)
                logging.warning(f"[{correlation_id}] Failed to process event ID: {event_id} on attempt {attempt}. Error: {e}")
                if attempt <= self.max_retries:
                    sleep_time = current_backoff + random.uniform(0, current_backoff * 0.5) # Ajouter du jitter
                    logging.info(f"[{correlation_id}] Retrying event ID: {event_id} in {sleep_time:.2f}s...")
                    time.sleep(sleep_time)
                    current_backoff *= 2 # Backoff exponentiel
                else:
                    logging.error(f"[{correlation_id}] Max retries ({self.max_retries}) exceeded for event ID: {event_id}.")
                    self.send_to_dlq(event_message, f"Max retries exceeded: {e}")
                    return False
        return False # Devrait être couvert par les retours ci-dessus, mais pour être sûr

# --- Simulation d'utilisation ---
if __name__ == "__main__":
    consumer = EventConsumer()

    # Générer un correlation_id pour suivre le flux
    def generate_correlation_id():
        return f"corr-{random.randint(1000, 9999)}"

    # Message qui devrait réussir après quelques retries
    message_success = {
        "correlation_id": generate_correlation_id(),
        "payload": json.dumps({"id": "order-123", "type": "new_order", "value": 100})
    }
    consumer.consume_message(message_success)

    print("\n" + "="*50 + "\n")

    # Message qui devrait échouer après retries et aller en DLQ
    message_retry_fail = {
        "correlation_id": generate_correlation_id(),
        "payload": json.dumps({"id": "order-456", "type": "update_order", "value": 200})
    }
    consumer.consume_message(message_retry_fail)

    print("\n" + "="*50 + "\n")

    # Message avec une erreur permanente
    message_permanent_fail = {
        "correlation_id": generate_correlation_id(),
        "payload": json.dumps({"id": "order-789", "type": "invalid", "value": "abc"})
    }
    consumer.consume_message(message_permanent_fail)

Explication du code :

  • Le EventConsumer simule un consommateur d'événements.
  • _process_event_logic est la fonction qui traite l'événement. Elle simule des échecs temporaires (ConnectionError) et permanents (ValueError).
  • consume_message est la boucle principale de traitement. Elle contient la logique de retry :
    • Pour chaque tentative, elle essaie de traiter l'événement.
    • Si une ValueError (erreur permanente) est attrapée, le message est immédiatement envoyé à la DLQ.
    • Si une autre Exception est attrapée (simulant une erreur temporaire), le consommateur attend avec un exponential backoff (le délai double à chaque tentative) et un jitter (une variation aléatoire pour éviter les collisions), puis retente.
    • Si le nombre maximal de tentatives est dépassé, le message est envoyé à la DLQ.
  • send_to_dlq est une fonction illustrative qui imprimerait le message en cas réel, il l'enverrait à une véritable file d'attente dédiée.
  • Le correlation_id est généré et utilisé pour suivre un message unique à travers ses logs.

Code 2 : Propagation de Contexte pour le Tracing Distribué Conceptuel

Cet exemple montre comment un Trace ID et un Span ID (simplifiés ici en correlation_id) seraient propagés à travers les métadonnées d'un événement pour permettre le tracing distribué.

import json
import uuid
import time
import logging

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

# --- Service Producteur ---
class EventProducer:
    def publish_event(self, event_type, payload, parent_context=None):
        # Générer un nouvel ID de trace si ce n'est pas déjà un enfant d'une trace existante
        trace_id = parent_context.get("trace_id") if parent_context else str(uuid.uuid4())
        
        # Chaque opération est un span. Ici, la publication est un span.
        # En réalité, OpenTelemetry gérerait ça avec des objets Span.
        span_id = str(uuid.uuid4()) 
        
        event_data = {
            "id": str(uuid.uuid4()),
            "type": event_type,
            "timestamp": int(time.time()),
            "payload": payload,
            "metadata": {
                "trace_id": trace_id,
                "span_id": span_id,
                # Si c'est un enfant, nous voudrions aussi le parent_span_id
                "parent_span_id": parent_context.get("span_id") if parent_context else None
            }
        }
        logging.info(f"[Producer][{event_data['metadata']['trace_id']}] Publishing event {event_data['id']} of type {event_type} with span {event_data['metadata']['span_id']}")
        
        # En réalité, ici on enverrait l'événement à un broker (Kafka, RabbitMQ, etc.)
        return event_data # Retourne l'événement pour la simulation

# --- Service Consommateur ---
class EventProcessor:
    def process(self, event):
        metadata = event.get("metadata", {})
        trace_id = metadata.get("trace_id", "N/A")
        parent_span_id = metadata.get("span_id", "N/A") # Le span du producteur devient le parent

        # Démarrer un nouveau span pour le traitement de cet événement
        current_span_id = str(uuid.uuid4())
        
        logging.info(f"[Processor][{trace_id}][{current_span_id}] Processing event {event['id']} (Parent Span: {parent_span_id})")
        time.sleep(random.uniform(0.1, 0.5)) # Simule le travail
        
        # Simule la publication d'un événement de suivi
        if event["type"] == "order_created":
            logging.info(f"[Processor][{trace_id}][{current_span_id}] Order created, publishing 'order_shipped' event.")
            # Passer le contexte actuel au nouvel événement publié
            context_for_next_event = {
                "trace_id": trace_id,
                "span_id": current_span_id # Ce span devient le parent du prochain span
            }
            producer.publish_event("order_shipped", {"order_id": event["id"], "status": "shipped"}, context_for_next_event)
        
        logging.info(f"[Processor][{trace_id}][{current_span_id}] Finished processing event {event['id']}.")

# --- Flux d'événements simulé ---
if __name__ == "__main__":
    producer = EventProducer()
    processor = EventProcessor()

    # 1. Le producteur publie un événement "order_created"
    order_event = producer.publish_event("order_created", {"user_id": 1, "product_id": 101, "quantity": 2})

    print("\n" + "="*50 + "\n")

    # 2. Le processeur reçoit et traite l'événement "order_created"
    processor.process(order_event)

    print("\n" + "="*50 + "\n")

    # 3. Supposons que l'événement "order_shipped" soit ensuite consommé par un autre service (ici le même processeur pour la démo)
    # L'événement "order_shipped" a été publié par le "processor", donc il contient le trace_id et le span_id du processeur comme parent.
    # Pour simuler, nous allons récupérer le dernier événement publié par le "producer" (qui est dans cet exemple le EventProcessor)
    # Dans un vrai système, un autre service consommerait l'événement 'order_shipped' directement du broker.
    # Pour cette démo, nous allons simuler un deuxième traitement avec un événement manquant ici,
    # car 'publish_event' retourne l'événement mais ne le stocke pas globalement.

    # Relancer un flow pour montrer le tracing avec un événement qui déclenche un autre
    print("\n--- Nouveau Flux avec Tracing ---")
    initial_event_context = None # Pas de parent_context pour le premier événement
    order_event_2 = producer.publish_event("order_created", {"user_id": 2, "product_id": 102, "quantity": 1}, initial_event_context)
    processor.process(order_event_2)

    # Pour un suivi plus précis, il faudrait un broker réel ou une structure pour les événements en mémoire
    # Ici, l'événement "order_shipped" est directement créé par l'appel à producer.publish_event dans processor.process,
    # ce qui montre bien la propagation du contexte.

Explication du code :

  • EventProducer et EventProcessor sont des services distincts.
  • Lorsqu'un événement est publié, un trace_id unique est créé s'il n'y a pas de contexte parent. Un span_id est également généré pour l'opération de publication.
  • Ces trace_id et span_id sont inclus dans les metadata de l'événement.
  • Lorsque EventProcessor reçoit l'événement, il extrait le trace_id et le span_id du parent (ici, le span_id du producteur devient parent_span_id).
  • Le EventProcessor crée ensuite son propre current_span_id pour son opération de traitement.
  • Si le processeur publie un nouvel événement (par exemple, order_shipped après order_created), il transmet son propre trace_id et current_span_id comme parent_context au nouvel événement.
  • Chaque log inclut le trace_id et le span_id pertinents, ce qui permet de reconstruire l'enchaînement des opérations à travers les services.

Conclusion

La gestion des erreurs, la résilience et l'observabilité ne sont pas de simples "bonnes pratiques" dans les architectures événementielles ; elles en sont les fondations mêmes. Ignorer ces aspects revient à construire sur du sable.

  • La gestion des erreurs nous permet de classer les défaillances et d'appliquer des stratégies adaptées (retries, DLQ, idempotence, transactions compensatoires) pour maintenir l'intégrité des données et la fluidité des processus.
  • La résilience assure que notre système peut survivre aux pannes de composants (Circuit Breaker, Bulkhead, Timeout, Fallback), minimisant l'impact sur l'utilisateur final et prévenant les défaillances en cascade.
  • L'observabilité nous donne les outils pour comprendre le comportement de notre système distribué (Logs, Metrics, Traces distribuées), permettant un diagnostic rapide et une optimisation continue.

En intégrant ces principes dès la conception de nos EDAs, nous construisons des systèmes non seulement puissants et scalables, mais aussi fiables, maintenables et faciles à opérer, capables de s'adapter aux réalités complexes du monde réel. C'est un investissement essentiel pour la robustesse et la pérennité de vos architectures événementielles.