From Database to Memory

Der OTTO-Marktplatz wächst jede Woche um etwa 10.000 Produkte - Tendenz steigend. Eine besondere Herausforderung ist es daher mit solchen großen Datenmengen umzugehen und alle Systeme zeitnah mit den Daten anzureichern sowie diese stets aktuell zu halten.

Wir sind eines von etwa 20 Teams, das rund um den Online-Shop otto.de derzeit um die 15 Micro-Services betreibt. Um dem Wachstum des OTTO-Marktplatzes gerecht zu werden, ist es wichtig, unsere Infrastruktur entsprechend skalieren zu können. Sei es durch das Erstellen weiterer Micro-Services, die neue Fachlichkeit übernehmen, oder mittels einer horizontalen Skalierung. Bei beidem stellt sich in unserem Fall die Datenbank zunehmend als problematisch dar.

Ein Problem ist das Verteilen der Daten (vgl. Abb. 1): Durch die Unabhängigkeit der Micro-Services ist es mit zunehmender Anzahl von Produktdaten und den damit häufigeren Aktualisierungen von Preisen oder Verfügbarkeiten schwieriger die separaten Datenbanken auf dem gleichen Stand zu halten. Das ist allerdings sehr wichtig, da es andernfalls dazu kommt, dass auf verschiedenen Bereichen der Webseite Produktinformationen unterschiedlich angezeigt werden.

Ein weiteres Problem ist der Zugriff auf die Daten (vgl. Abb. 2): Der Anstieg von Produktdaten bedeutet gleichzeitig einen Anstieg von Aktualisierungen dieser Daten (Preise, Verfügbarkeiten, etc.) in der Datenbank. Diese vermehrten Schreiboperationen beeinträchtigen oft die Lesegeschwindigkeit. Gerade im Zusammenspiel mit einer horizontalen Skalierung, bei der weitere Instanzen lesend auf die Datenbank zugreifen, wird die Datenbank zu einem Engpass. Infolgedessen verschlechtern sich die Antwortzeiten und damit die Seitenladezeit für den Kunden.

Beide Probleme konnten wir lösen: Eine Umstellung unserer Datenversorgung, weg von einem Pull- hin zu einem Push-Prinzip, ermöglicht es nun Daten nahezu zeitgleich an alle relevanten Services zu verteilen. Das Verteilen geschieht mittels eines Messaging-Systems. Jede Instanz eines jeden Services ist als Empfänger registriert und hält die verarbeiteten Daten direkt im Speicher. Requests können so schnell ohne Datenbankanfrage beantwortet werden.

Im Folgenden werde ich noch einmal genauer auf die zwei genannten Probleme eingehen und anschließend unsere Lösung skizzieren.

Probleme der Datenversorgung

In einem verteilten System wie unserem, bestehend aus vielen kleinen unabhängigen Micro-Services, ist jeder einzelne Service für seine Datenversorgung selbst verantwortlich. Nach dem „Shared Nothing”-Prinzip teilt er nichts mit anderen Services und das System ist damit besonders skalierbar und ausfallsicher.

Auch Datenbanken werden demnach nicht zwischen Services geteilt und somit muss jeder Service sich selbst um dessen Aktualisierungen kümmern. Dies übernimmt bei uns ein regelmäßiger Prozess, der neue Daten asynchron im Hintergrund per Pull-Mechanismus von einem zentralen System holt (vgl. Abb. 3).

Erfährt ein bestimmter Service höhere Last, also eine höhere Anzahl von Aufrufen, lässt sich diese zielgerichtet durch eine horizontale Skalierung, also dem Einsatz weiterer Instanzen dieses Service, abfangen. Da sich die Instanzen jedoch gemeinsam eine Datenbank teilen, ist die mögliche Last wiederum durch diese limitiert. Auch hier kann durch Steigern der Node-Anzahl im Datenbank-Cluster eine horizontale Skalierung erreicht werden. Soweit die Theorie.

Wir verwenden MongoDB als Datenbank und haben eher schlechte Erfahrungen bei dieser Art von Skalierung gemacht. Die vermehrten Schreibzugriffe des Datenimports führen häufig zu globalen Locks, die dann Lesezugriffe zeitweilig blockieren und somit zu hohen Latenzen für den Kunden führen.

Für eine Skalierungsfähigkeit ist es zudem nicht nur wichtig die Last steigern zu können, sondern ebenfalls die Menge an Daten, die zeitnah an Services verteilt werden können.

Mit unserer Architektur wurde es ebenfalls immer schwieriger den Datenbestand über alle Services synchron zu halten. Durch die abweichenden Laufzeiten der Import-Prozesse, z.B. durch unterschiedliche Systemauslastung oder unterschiedliche Verarbeitungsschritte, kommt es zwischen den Services konzeptionell temporär zu Abweichungen. Bei der stetig wachsenden Anzahl an neuen Produkten und Produktdatenveränderungen werden diese Inkonsistenzen zunehmend verstärkt.

Alternative Datenversorgung per Messaging-System

Ein anderer Einflussfaktor für die Skalierungsfähigkeit und Robustheit von verteilten Systemen ist die Art der Kommunikation. Die vielen Services kommunizieren oft nicht direkt miteinander, sondern stattdessen asynchron durch das Verschicken von Nachrichten über einen Event-Bus. Nach dem Publish-Subscriber-Prinzip kann eine Nachricht mehr als einen Empfänger haben. Der oder die Empfänger müssen dabei nicht einmal bekannt sein.

Diese Form der Nachrichtenverteilung kann natürlich auch zur Datenversorgung genutzt werden. Wir setzen sie nun ein, da sie in unserem Fall folgenden Vorteil bietet: Es gibt nur noch einen zentralen regelmäßigen Prozess, der veränderte Produktdaten holt und diese dann über den Event-Bus an alle Abnehmer verteilt (vgl. Abb. 4). Dies geschieht mit geringer Latenz und so schaffen wir es trotz großer Datenmengen die Services relativ synchron zu halten.

Event-Sourcing

Diese Art der Datenversorgung alleine ersetzt jedoch keine Datenbank, denn sie ist nicht persistent. Ein Service würde nur die Daten kennen, die er ab dem Zeitpunkt seiner Registrierung an dem Event-Bus erhält. Jeder Neustart würde somit zum Datenverlust führen und eine erneute Verarbeitung aller Nachrichten erfordern, um wieder auf den aktuellen Stand zu gelangen.

Das entspricht einem wesentlichen Teil der Funktionsweise von Event-Sourcing. Anstatt wie in einer Datenbank den aktuellen Zustand zu speichern, werden alle Nachrichten, die zu diesem Zustand geführt haben, der Reihe nach in einem Message-Log gespeichert. Anschließend kann dann der finale Zustand wieder hergestellt werden, indem sequentiell alle Nachrichten erneut aus dem Message-Log gelesen und verarbeitet werden. Gleichzeitig ist das Message-Log auch der Event-Bus über das neue Nachrichten an alle registrierten Empfänger verteilt werden.

Durch den Einsatz von Event-Sourcing können wir unsere Services ohne Datenbank betreiben. Alle Daten, die wir beim Hochfahren aus dem Message-Log lesen, sowie die, die wir danach fortan empfangen, werden direkt im Speicher der jeweiligen Instanz abgelegt (vgl. Abb. 5).

Um Speicherplatz zu sparen, werden dabei selbstverständlich nur die für den Service relevanten Daten eines Produktes gespeichert. Ebenso nutzen wir nicht den Java-Heap- sondern einen Off-Heap-Speicher. Das hat zwei Gründe. Zum einen halbiert es nahezu den benötigten Speicherbedarf, da Strings mit ASCII-Zeichen nicht wie in Java mit zwei Bytes (UTF-16) sondern nur mit einem Byte (UTF-8) repräsentiert werden. Zum anderen (und sehr viel wichtigeren Punkt) erspart es dem Java-Garbage-Collector mehrere Gigabytes an Daten zu verwalten und verhindert so unnötiges Pausieren der Virtual Machine.

Der Wegfall des Datenbankzugriffs verbessert die Antwortzeiten des Services und erlaubt gleichzeitig eine kosteneffiziente und eine echte lineare Skalierung. Linear bedeutet, dass zweimal so viele Instanzen jetzt genau das Doppelte an Last verarbeiten können; 100 Instanzen das hundertfache an Last, usw. - sofern man die begrenzte Netzwerkbandbreite vernachlässigt. Das Datenbankcluster ist kein limitierender Faktor mehr und es ist günstiger, Instanzen mit viel Speicher auszustatten, als zusätzliche Nodes für ein Datenbankcluster zu betreiben.

Es gibt sogar weitere Vorteile. Da Nachrichten in einem Message-Log immutable und "append-only" sind, kann nicht nur der aktuelle sondern jeder beliebige Zustand für einen Zeitpunkt wiederhergestellt werden. Man erhält dadurch quasi automatisch eine Versionierung. Zum Debugging kann jede Änderung am System einzeln nachvollzogen werden und Schritt für Schritt wiederholt werden.

Wo Licht ist, ist auch Schatten

Abgesehen davon, dass Event-Sourcing eher unbekannt und nicht so verbreitet ist, es keine Frameworks gibt und die Komplexität recht hoch ist, verletzt es ein wichtiges Merkmal von Micro-Services: Kurze Startup-Zeiten (siehe Twelve-Factor Apps).

Dem entgegen steht das zeitintensive Lesen des gesamten Event-Logs zum Wiederherstellen des Datenbestands. Da für einen Service in der Regel lediglich der aktuelle Zustand von Interesse ist, gibt es verschiedene Ansätze diesen möglichst schnell zu erreichen:

• Variante a) Das Event-Log kann rückwärts gelesen werden. Es ist jedoch äußerst schwierig festzustellen wie weit man rückwärts lesen muss, um alle Nachrichten zu erhalten, die für den korrekten Zustand nötig sind.
• Variante b) Der aktuelle Zustand und der Zeitpunkt der letzten Nachricht wird lokal in einer Datenbank persistiert. Nach einem Neustart muss das Event-Log nicht von Anfang an, sondern lediglich ab dem gespeicherten Zeitpunkt gelesen werden.
• Variante c) Von dem Event-Log wird regelmäßig (z.B. alle zwei Stunden) ein kompakter Snapshot erstellt, der jeweils nur die letzte Version einer Nachricht beinhaltet (siehe Abb. 6). Nach einem Neustart muss dann nur ein reduziertes Event-Log in Form eines Snapshots sowie die Nachrichten ab dem Zeitpunkt der Snapshoterstellung gelesen werden.

Wir nutzen Variante „c“, da weniger Nachrichten als bei Variante „a“ gelesen werden müssen - selbst wenn man genau wüsste, wie weit zurück gelesen werden muss. Variante „b“ scheidet aus, weil es unser Ziel war auf eine Datenbank zu verzichten. Außerdem würde ein neu entwickelter Service keine Nachrichten vor seiner Inbetriebnahme empfangen und der Sender müsste diese erneut senden. Dies verletzt das Publish-Subscriber-Prinzip, da Empfänger und Sender eigentlich nicht voneinander wissen sollten.

Wenn Apache Kafka als Middleware für die Umsetzung von Event-Sourcing bzw. eines Event-Logs eingesetzt wird, kann eine sogenannte "Log-Compaction" genutzten werden. Die Funktionsweise einer Compaction ähnelt der einer Snapshoterstellung, so dass lediglich die jeweils letzte Version einer Nachricht vorgehalten wird (siehe Abb. 6).

Da wir bei OTTO vor kurzem unsere komplette Infrastruktur in die AWS migriert haben, wollten wir erst Erfahrungen sammeln bevor wir dort Kafka in Eigenregie betreiben. Zur Umsetzung von Event-Sourcing als neue Datenversorgung verwenden wir deswegen vorerst ausschließlich AWS-Technologien, wie AWS-Kinesis und AWS-SQS zusammen mit einer eigens entwickelten Open-Source-Library names Synapse.
Synapse hat zum Ziel das Entwickeln von neuen Micro-Services mit einer Event-Sourcing-Anbindung zu erleichtern. Dieses Thema wollen wir euch in einem folgenden Blog-Eintrag detaillierter vorstellen.