Um Verkettungen von Events/Anfragen zu analysieren oder Fehlerstellen zu identifizieren, kann man in verteilten Systemem mit Correlation IDs arbeiten. In diesem Artikel wollen wir auf Möglichkeiten des Setzens von Correlation IDs im Stream Processing eingehen und eine generische Lösung zum Festhalten der Event Times vorschlagen, um die Gesamt- oder Teilverarbeitungsdauer bestimmen zu können.

Fallbeispiel

Wir betrachten einen Teil einer E-Commerce Plattform, die mit Microservices aufgezogen worden ist. Verschiedene Services kümmern sich um Teilaspekte. Die Erfüllung eines Usecases kann dabei mehrere Services betreffen.

Im folgenden Beispiel betrachten wir die Änderung von Kundendaten und ihre Auswirkung auf die weiteren Services.

Die Abbildung stellt eine Architektur eines möglichen Stream Processings dar:

• Es gibt einen customer-service, welcher Daten produziert, wenn sich Kundendaten geändert haben.
• Der address-service hört auf den Channel changed_customer_data und validiert Adressen. Das Ergebnis wird in den Channel “validated_address_changes” gespeichert.
• Der billing-service hört auf den Channel validated_address_changes und persistiert validierte neue Adressen, da er sie beim Erzeugen von Rechnungen braucht.
• Der voucher-service hört ebenfalls auf den Channel validated_address_changes, um zu überprüfen, ob die neue Adresse in ein Kampagnengebiet fällt. Entsprechend würde die Verkettung weitergehen.

Die Abbildung beinhaltet zwei Verarbeitungsstränge:

1. Persistieren neuer Kundendaten
2. Fällt die geänderte Adresse in ein Kampagnengebiet, soll ein Gutschein erstellt werden

Wenn ein Fehler vorkommt, stellen sich folgenden Fragen:

• An welcher Stelle trat der Fehler auf?
• Ist es ein Service Fehler?
• Waren die Daten invalide?
• Kann der Service mit einer bestimmten Kombination der Daten nicht umgehen?
  • Wie ist solch eine Kombination der Daten möglich?

Was kann man bisher und was sind die Probleme?

Anhand der Logs eines Services bekommt man Informationen, dass zum Beispiel ein Fehler vorgekommen ist. Wenn man entsprechend granular loggt, kann man die Daten des Events einsehen. Dann aber kann es sein, dass man im Service nachvollziehen möchte, welche Schritte davor passiert sind und da mehrere Events gleichzeitig abgearbeitet werden können, ist die Reihenfolge der Logs in dem Fall nicht gesichert. Das Problem wird noch gravierender, wenn man die Verarbeitungskette über mehrere Services hinweg analysieren möchte. Hierbei muss es sich auch nicht unbedingt um eine Fehleranalyse handeln. Wie kann man Events über mehrere Services hinweg miteinander korrelieren?

Correlation IDs

Correlation IDs sind eindeutig vergebene IDs, die als Metainformation an ein Event oder eine Anfrage angehangen werden.
Bei verteilten Systemen, man denke an eine Microservice Architektur, hat man in der Regel mehrere Aufrufe gegen mehrere Systeme, um einen Use Case zu erfüllen.
Um Fehler identifizieren zu können oder auch um eine Verkettung der Events zu repräsentieren, nutzt man Correlation IDs.

Im Falle von Microservices, die über REST miteinander kommunizieren, kann man mit einer Correlation ID arbeiten und den kompletten Verarbeitungsstrang mit nur einer Correlation ID repräsentieren. Im Stream Processing bekommt jedes Event eine eigene ID, weil die Reaktion auf ein Event vom Generierer des Events nicht bekannt ist und zu beliebig vielen Events führen kann. Um verschiedene Verarbeitungsstränge dann unterscheiden zu können, braucht man auch verschiedene Correlation IDs die lediglich auf das gleiche Auslöser Event referenzieren.

Brain Storming

Ein Event stellen wir in diesem Beispiel als Briefumschlag(Metainformationen zum Event) mit Inhalt(Payload) dar:

• Jedes Event hat eine ID, das in den folgenden Events als correlation_id referenziert werden kann
• Jedes Event hat eine Event Time, welche dem Zeitpunkt der Generierung entspricht. Empfohlen ist eine Zeitangabe nach ISO Standard mit einer Zeitzone.

Erweitern wir unser Fallbeispiel um IDs zu den Events im Diagramm:

• customer-service erzeugt Event mit ID 95ada853-eeaa-4ebf-91b2-94ee4c81961d
• address-service erzeugt Event mit ID 252f3bfb-98ec-450a-ad5f-f3f2c2e3660d
• voucher-service erzeugt Event mit ID 28f7e282-8c51-4ed8-89fa-20889d8be7c9
• billing-service erzeugt Event mit ID 4de4d74d-6b28-461e-a1cc-d46b49dc4697

1.) Vorschlag

Speichere die ID des vorhergehenden Events in das Feld der correlation_id. Das erzeugte Event vom billing-service beinhaltet also die vorhergehende ID, nämlich die vom address-service.

Bei der Erzeugung eines neuen Events auf Basis des Events, welches in der Form vom billing-service ankommt, würde man die ID des Input-Events als Wert für das Feld correlation_id setzen. In das Feld id trägt man eine für das neue Event erzeugte ID ein.

Man hat also immer eine einfache Korrelation zum vorherigen Event. Damit bei diesem Vorschlag auch alle Events nachvollzogen werden können, müssen die Events zwischengespeichert und abgefragt werden können, um die komplette Verkettung nachvollziehen zu können.

2.) Vorschlag

Speichere in der Liste der correlation_id alle IDs der vorherigen Events. Das erzeugte Event vom voucher-service beinhaltet also alle vorhergehenden Event-IDs.

Bei der Erzeugung eines neuen Events auf Basis des Events, welches in der Form vom voucher-service ankommt, würde man die ID des Input-Events ans Ende der correlation_id Liste hinzufügen. Beim Einhalten der Reihenfolge, kann man die Verkettung nach der Erzeugung repräsentieren. Das neue Event erstellt seine eigene ID, die es in das Feld id einträgt. Gleichzeitig hat man mit der Liste der Correlation IDs eine Liste zum Filtern in den Logs, wenn die Logs so angepasst wurden, dass immer die ID als Präfix mitgeloggt wird.

Diese Variante ermöglicht die direkte Einsicht der Verkettung nach den IDs anhand der Liste.

Ergänzender Vorschlag

Wenn erst mehrere Events zur Erzeugung eines neuen Events führen und man das darstellen möchte, müsste man überlegen anstatt nur einer ID ein Tupelobjekt als correlation Objekt einzufügen. Meist kann man aber davon ausgehen, dass ein Event das auslösende Event gewesen ist. Außer man arbeitet mit Joins im Streaming. Denn dann sind beide Events gleichwertig auslösend.

Was bringt die Event Time

Man unterscheidet drei Arten von Zeitpunkten

• event time (Zeitpunkt der Generierung)
• ingestion time (Zeitpunkt des Eintritts in das verarbeitende System)
• processing time (Zeitpunkt der Verarbeitung)

Um beispielsweise auszuwerten, ob ein Anwendungsfall performant ausgeführt werden konnte, müsste man zwischen der Verarbeitung der kompletten Verarbeitungskette die Event-Time von der Processing-Time abziehen und insgesamt zusammen addieren. Damit man das machen kann, muss man aber auch pro Event die Event-Time mitgeben. Alternativ möchte man auch nur die Dauer einer Teilkettenverarbeitung ausrechnen. Auch hierfür bräuchte man auf Eventebene die Eventtime. Ein genereller Vorschlag um die Verarbeitungsdauer mehrerer Events ausrechnen zu können, wäre neben der ID der vorhergehenden Events auch die Event-Times zu schreiben.

Empfehlung

Der erste Vorschlag ist einfach umgesetzt und erlaubt, sei es auch über Umwege, seine verketteten Events zu analysieren. Zumindest kann man sich von einem Event immer weiter zurückhangeln. Um die Events dauerhaft analysieren zu können, braucht man aber eine Persistenz.

Der zweite Vorschlag ist im Vergleich zum ersten Vorschlag mit leicht höherem Aufwand umzusetzen. Jedoch erlaubt es die direkte Einsicht seiner Verkettung mit entsprechender Reihenfolge und das auch ohne Persistenz.

Verbunden mit den Event-Times hat man anhand der Metainformationen schon eine gute Übersicht einer Verkettung und ihrer Verarbeitungsdauer.