Als wir den zweiten Server zugefügt haben, wurden verteilte Systeme zu einem festen Bestandteil von Amazon. Als ich 1999 bei Amazon anfing, hatten wir so wenige Server, dass wir einigen von ihnen so einprägsame Namen wie "fishy" oder "online-01" geben konnten. Doch schon 1999 war die verteilte Verarbeitung kein einfaches Unterfangen. Damals wie heute drehten sich die Herausforderungen bei verteilten Systemen um Latenz, Skalierung, Know-how zu Netzwerk-APIs, Marshalling und Unmarshalling von Daten und die Komplexität von Algorithmen (Paxos). Die Systeme wurden bald immer größer und verteilter und theoretische Szenarien wurden zu regelmäßigen Fällen.

Es ist eine schwierige Aufgabe, verteilte Dienstleistungen zu entwickeln – das gilt für ein zuverlässiges Telefonnetz wie auch für Amazon Web Services (AWS). Außerdem ist die verteilte Verarbeitung seltsamer und weniger intuitiv als andere Verarbeitungsansätze. Dafür gibt es zwei zusammenhängende Ursachen. Unabhängige Ausfälle und Nichtdeterminismus verursachen die schwerwiegendsten Probleme bei verteilten Systemen. Bei verteilten Systemen treten nicht nur die typischen von Entwicklern gewohnten Ausfälle auf, sondern noch viele weitere. Doch es kommt noch schlimmer: Es ist nicht immer möglich, zu wissen, ob etwas ausgefallen ist.

In der Amazon Builders' Library gehen wir immer wieder darauf ein, wie AWS mit den komplexen Entwicklungs- und Betriebsproblemen fertig wird, die mit verteilten Systemen einhergehen. Bevor Sie die Lösungen in anderen Artikeln studieren, sollten Sie die Ursachen dafür kennenlernen, dass die verteilte Verarbeitung eben so, nun ja, seltsam ist. Fangen wir mit den Typen der verteilten Systeme an.

Typen von verteilten Systemen

Verteilte Systeme sind unterschiedlich schwierig zu implementieren. Am einen Ende des Spektrums befinden sich die verteilten Systeme des Typs Offline. Dazu zählen Systeme für die Batch-Verarbeitung, Cluster für Big-Data-Analysen, Renderfarms für Filmeffekte, Cluster für die Berechnung von Proteinfaltungen und Ähnliches. Diese Systeme sind zwar keineswegs trivial zu implementieren, doch sie besitzen fast alle Vorteile der verteilten Verarbeitung (Skalierbarkeit und Fehlertoleranz) und fast keine der Nachteile (komplexe Fehlermodi und Nichtdeterminismus).
 
In der Mitte des Spektrums liegen die verteilten Systeme des Typs Soft Real-Time. Das sind kritische Systeme, die kontinuierlich Ergebnisse erzeugen oder aktualisieren müssen, doch sie haben relativ viel Zeit dafür. Dazu zählen zum Beispiel bestimmte Search Index Builders, Systeme auf der Suche nach beeinträchtigten Servern, Rollen für Amazon Elastic Compute Cloud (Amazon EC2) und so weiter. Ein Search Indexer könnte (je nach Anwendung) 10 Minuten oder sogar mehrere Stunden offline sein, ohne dass es sich auf den Kunden auswirkt. Rollen für Amazon EC2 müssen aktualisierte Anmeldeinformationen an (praktisch) jede EC2-Instance übertragen, doch bis dahin dürfen mehrere Stunden vergehen, denn die alten Anmeldeinformationen verfallen erst nach einiger Zeit.
 
Am anderen (und komplexen) Ende des Spektrums befinden sich die verteilten Systeme des Typs Hard Real-Time, oder kurz: verteilte HRT-Systeme. Sie werden oft als Request/Reply-Services bezeichnet. Wenn wir bei Amazon an die Entwicklung eines verteilten Systems denken, kommt uns zuerst ein HRT-System in den Sinn. Leider sind verteilte HRT-Systeme am schwierigsten korrekt umzusetzen. Das Schwierige daran ist, dass die Requests unvorhersehbar eintreffen und die Replies schnell zurückgesendet werden müssen (zum Beispiel weil der Kunde auf die Reply wartet). Dazu zählen beispielsweise Front-End-Webserver, die Pipeline bei der Bestellung, Kreditkarten-Transaktionen, jede AWS-API, Telefonie usw. Verteilte HRT-Systeme sind das zentrale Thema dieses Artikels.

Verteilte HRT-Systeme sind einfach seltsam

Bei einem Handlungsstrang der Superman-Comics trifft er auf ein Alter Ego namens Bizarro, der auf einem Planeten lebt (Bizarro-Welt), auf der alles ins Gegenteil verkehrt ist. Bizarros Aussehen mag zwar Superman ähneln, doch er ist böse. Bei verteilten HRT-Systemen verhält es sich ähnlich. Sie erwecken den Eindruck einer normalen Datenverarbeitung, aber ehrlich gesagt sind sie nicht nur anders, sondern auch böse.

Es gibt einen zentralen Grund dafür, warum die Entwicklung verteilter HRT-Systeme bizarr ist: der Austausch von Requests/Replies. Ich meine nicht die feinen Details von TCP/IP, DNS, Sockets oder anderen Protokollen dieser Art. Diese Themen sind zwar teilweise schwierig zu verstehen, aber sie ähneln auch anderen komplexen Problemen in der Datenverarbeitung.

Das wirklich Schwierige an verteilten HRT-Systemen ist die Tatsache, dass Nachrichten von einer fehlerhaften Domäne an eine andere gesendet werden können. Eine Nachricht wird gesendet, das hört sich eigentlich harmlos an. Doch mit dem Versenden von Nachrichten wird alles sehr viel komplizierter.

Sehen Sie sich einfach einmal das folgende Code-Snippet einer Pac-Man-Implementierung an. Es soll auf einem einzigen Computer ausgeführt werden und versendet keine Nachrichten über ein Netzwerk.

board.move(pacman, user.joystickDirection())
ghosts = board.findAll(":ghost")
for (ghost in ghosts)
  if board.overlaps(pacman, ghost)
    user.slayBy(":ghost")
    board.remove(pacman)
    return

Stellen wir uns jetzt vor, wir schreiben eine Netzwerkversion dieses Codes, wobei der Status des Spielfeld-Objekts von einem separaten Server verwaltet wird. Jedes Mal, wenn das Spielfeld-Objekt aufgerufen wird, z. B. findAll(), müssen Nachrichten zwischen den beiden Servern hin- und hergesendet werden.

Wenn ein Request/Reply-Austausch zwischen zwei Servern stattfindet, müssen immer mindestens acht Schritte durchlaufen werden. Rufen wir uns einmal die Grundlagen zu Requests/Replies ins Gedächtnis, um den vernetzten Pac-Man-Code zu verstehen. 

Requests/Replies in einem Netzwerk

Request/Reply-Zyklen erfordern immer die gleichen Schritte. Auf der nachfolgenden Abbildung sendet der CLIENT eine Request-NACHRICHT über das NETZWERK an den SERVER, der über das NETZWERK eine REPLY-Nachricht sendet.

Im Idealfall gibt es keine Probleme und die folgenden Schritte werden durchlaufen:

1. REQUEST SENDEN: CLIENT sendet NACHRICHT ins NETZWERK.
2. REQUEST BEREITSTELLEN: NETZWERK stellt NACHRICHT für SERVER bereit.
3. REQUEST VALIDIEREN: SERVER validiert NACHRICHT.
4. SERVERSTATUS AKTUALISIEREN: SERVER aktualisiert bei Bedarf seinen Status auf Grundlage der NACHRICHT.
5. REPLY SENDEN: SERVER sendet REPLY ins NETZWERK.
6. REPLY BEREITSTELLEN: NETZWERK stellt REPLY für CLIENT bereit.
7. REPLY VALIDIEREN: CLIENT validiert REPLY.
8. CLIENT-STATUS AKTUALISIEREN: CLIENT aktualisiert bei Bedarf seinen Status auf Grundlage der REPLY.

Sind das nicht ganz schön viele Schritte für eine einfache Aufgabe? Doch es bleibt dabei: Diese Schritte sind die Definition einer Request/Reply-Kommunikation in einem Netzwerk. Keiner dieser Schritte kann ausgelassen werden. Zum Beispiel können Sie nicht einfach Schritt 1 überspringen. Der Client muss die NACHRICHT ja irgendwie ins NETZWERK senden. Physisch betrachtet werden Pakete über eine Netzwerkkarte gesendet, wodurch elektrische Signale über Kabel und Verteiler durch das Netzwerk zwischen CLIENT und SERVER übertragen werden. Dieser Vorgang gehört noch nicht zu Schritt 2, denn der zweite Schritt könnte aus anderen Gründen fehlschlagen, zum Beispiel, wenn die Stromversorgung vom SERVER abbricht und er die eingehenden Pakete nicht empfangen kann. Die gleiche Logik gilt auch für die restlichen Schritte.

Deshalb wird ein einziger Request/Reply-Vorgang über das Netzwerk (Abruf einer Methode) auf einmal zu acht Vorgängen. CLIENT, SERVER und NETZWERK können völlig unabhängig voneinander Fehler verursachen, was die Angelegenheit noch riskanter macht. Der Entwicklercode müsste deshalb für den Fehler bei jedem einzelnen Schritt eine Lösung bereithalten. Doch das ist bei der Entwicklung nur selten der Fall. Der folgende Ausdruck von der Codeversion für einen einzigen Computer verdeutlicht die Ursachen dafür.

board.find("pacman")

Technisch gesehen könnte der Code aus verschiedenen komischen Gründen während der Ausführung fehlschlagen, selbst wenn die Implementierung von spielfeld.find keine Bugs beinhaltet. Zum Beispiel könnte die CPU während der Ausführung plötzlich überhitzen. Auch die Stromversorgung des Computers könnte unerwartet abreißen. Es könnte eine Kernel Panic auftreten. Der Arbeitsspeicher könnte voll sein und schon kann ein Objekt, das spielfeld.find erstellen möchte, nicht erstellt werden. Oder der Festplattenspeicher ist voll, weshalb spielfeld.find eine Statistikdatei nicht aktualisieren kann und einen Fehler ausgibt, obwohl sie das wahrscheinlich nicht tun sollte. Ein Gammastrahl könnte den Server treffen und den Wert eines Bits im RAM ändern. Aber meistens machen sich die Entwickler einfach keine Gedanken um solche Fälle. Zum Beispiel wird bei Gerätetests das Szenario eines CPU-Ausfalls niemals und das eines überfüllten Arbeitsspeichers nur selten behandelt.

Typischerweise treten solche Ausfälle nur bei einem einzigen Computer auf. Es gibt also nur eine einzige fehlerhafte Domäne. Wenn zum Beispiel die spielfeld.find-Methode nicht funktioniert, weil die CPU spontan überhitzt, ist davon auszugehen, dass der gesamte Computer nicht mehr läuft. Ein solcher Fehler lässt sich einfach nicht vermeiden. Ähnliches gilt für die anderen zuvor erwähnten Fehlertypen. Zugegeben: Für einige diese Fälle könnten Sie einen Code schreiben, aber würde sich das wirklich lohnen? Wenn diese Ausfälle auftreten, dann ist davon auszugehen, dass alles andere ebenfalls ausfällt. In der Informatik nennt man das auch Shared Fate (geteiltes Schicksal). Durch "Fate Sharing" wird die Anzahl der verschiedenen Fehlermodi für den Entwickler deutlich reduziert.

Fehlermodi in verteilten HRT-Systemen behandeln

Entwickler müssen bei verteilten HRT-Systemen alle Aspekte der Netzwerkausfälle testen, denn Server und Netzwerk haben kein Shared Fate. Im Gegensatz zum Beispiel mit dem einzelnen Computer kann hier das Netzwerk ausfallen, doch das Client-Gerät weiterhin funktionieren. Auch wenn der Server ausfällt, funktioniert das Client-Gerät weiterhin, und so weiter.

Um alle Fehlerszenarien der zuvor beschriebenen Request/Reply-Schritte zu testen, müssen Entwickler davon ausgehen, dass jeder einzelne Schritt fehlschlagen könnte. Und sie müssen sicherstellen, dass sich der Code (auf dem Client und dem Server) immer korrekt verhält, wenn einer dieser Fehler auftritt.
Sehen wir uns einmal eine Request/Reply-Aktion an, die nicht nach Plan verläuft:

1. POST-ANFORDERUNG schlägt fehl: Entweder das NETZWERK konnte die Nachricht nicht bereitstellen (z. B. weil ein Verteiler zum falschen Zeitpunkt ausgefallen ist) oder der SERVER hat sie explizit abgelehnt.
2. REQUEST BEREITSTELLEN schlägt fehl: Das NETZWERK stellt die NACHRICHT erfolgreich für den SERVER bereit, aber der SERVER fällt kurz nach Erhalt der NACHRICHT aus.
3. REQUEST VALIDIEREN schlägt fehl: Der SERVER entscheidet, dass die NACHRICHT ungültig ist. Dafür sind viele verschiedene Gründe möglich. Zum Beispiel beschädigte Pakete, inkompatible Softwareversionen oder Bugs auf dem Client oder Server.
4. SERVERSTATUS AKTUALISIEREN schlägt fehl: Der SERVER versucht seinen Status zu aktualisieren, doch es funktioniert nicht.
5. REPLY SENDEN schlägt fehl: Der SERVER konnte die Reply nicht senden, unabhängig davon, wie die Reply ausgefallen wäre. Zum Beispiel könnte die Netzwerkkarte genau im falschen Moment ausfallen.
6. REPLY BEREITSTELLEN schlägt fehl: Wie zuvor bereits beschrieben, kann es passieren, dass das NETZWERK die REPLY nicht für den CLIENT bereitstellen kann, auch wenn das NETZWERK bei einem früheren Schritt noch funktioniert hat.
7. REPLY VALIDIEREN schlägt fehl: Der CLIENT entscheidet, dass die REPLY ungültig ist.
8. CLIENT-STATUS AKTUALISIEREN schlägt fehl: Der CLIENT empfängt die REPLY-Nachricht, aber er kann den eigenen Status nicht aktualisieren, die Nachricht nicht verstehen (Inkompatibilität) oder die Aufgabe aus einem anderen Grund nicht erfüllen.

Diese Fehlermodi sind der Grund dafür, warum verteilte Verarbeitung so schwierig ist. Ich nenne sie die acht apokalyptischen Fehlermodi. Sehen wir uns noch einmal den Ausdruck des Pac-Man-Codes unter Berücksichtigung dieser Fehlermodi an.

board.find("pacman")

Der Ausdruck lässt sich in die folgenden Client-Aktivitäten aufschlüsseln:

1. Nachricht, z. B. {action: "find", name: "pacman", userId: "8765309"}, an den Spielfeld-Server adressieren und ins Netzwerk senden.
2. Wenn das Netzwerk nicht verfügbar ist oder die Verbindung zum Spielfeld-Server explizit abgelehnt wird, Fehler auslösen. Dieser Fall ist etwas speziell, denn der Client weiß deterministisch, dass die Request auf keinen Fall vom Server erhalten wurde.
3. Auf Reply warten.
4. Wenn keine Reply empfangen wird, Timeout auslösen. In diesem Schritt bedeutet der Timeout, dass das Request-Ergebnis UNBEKANNT ist. Vielleicht ist die Request angekommen, vielleicht auch nicht. Der Client muss korrekt mit UNBEKANNT umgehen.
5. Wenn eine Reply erhalten wird, feststellen, ob sie einen Fehler oder einen Erfolg meldet, oder ob sie nicht auswertbar/beschädigt ist.
6. Wenn kein Fehler gemeldet wird, Unmarshalling mit Reply durchführen und sie in einen Objektcode umwandeln, den der Code versteht.
7. Wenn die Reply einen Fehler meldet oder sie nicht auswertbar ist, Ausnahme auslösen.
8. Die Methode für die Ausnahme muss entscheiden, ob die Request wiederholt werden soll oder ob die Aufgabe verworfen und das Spiel beendet werden soll.

Der Ausdruck startet auch die folgenden Server-Aktivitäten:

1. Request erhalten (das könnte schon nicht geschehen).
2. Request validieren.
3. Benutzer suchen, um festzustellen, ob er noch im Spiel ist. (Der Server kann den Benutzer aufgeben, wenn er zu lange keine Nachrichten von ihm erhält.)
4. Keepalive-Tabelle für den Benutzer aktualisieren, damit der Server weiß, dass er (wahrscheinlich) noch im Spiel ist.
5. Position des Benutzers überprüfen.
6. Eine Reply mit Informationen wie {xPos: 23, yPos: 92, clock: 23481984134} senden.
7. Jede weitere Serverlogik muss mit den zukünftigen Auswirkungen des Clients korrekt umgehen. Zum Beispiel, wenn die Nachricht nicht erhalten wird, wenn sie nicht verstanden wird, wenn der Server nach Erhalt abstürzt oder wenn er sie erfolgreich verarbeitet.

Aus einem Ausdruck im normalen Code werden also fünfzehn Zusatzschritte im Code des verteilten HRT-Systems. Diese Zunahme ist durch die acht verschiedenen Stationen bedingt, an denen jede einzelne Kommunikation zwischen Client und Server fehlschlagen kann. Jeder einzelne Ausdruck, der eine solche Netzwerkkommunikation auslöst, wie spielfeld.find("pacman"), führt zu Folgendem.

(error, reply) = network.send(remote, actionData)
switch error
  case POST_FAILED:
    // handle case where you know server didn't get it
  case RETRYABLE:
    // handle case where server got it but reported transient failure
  case FATAL:
    // handle case where server got it and definitely doesn't like it
  case UNKNOWN: // i.e., time out
    // handle case where the *only* thing you know is that the server received
    // the message; it may have been trying to report SUCCESS, FATAL, or RETRYABLE
  case SUCCESS:
    if validate(reply)
      // do something with reply object
    else
      // handle case where reply is corrupt/incompatible

Die Komplexität lässt sich nicht vermeiden. Wenn der Code nicht alle Fälle korrekt behandelt, treten bizarre Fehler beim Service auf. Stellen Sie sich vor, Sie müssten Tests für alle Fehlermodi schreiben, die bei einem Client/Server-System wie im Pac-Man-Beispiel auftreten könnten!

Verteilte HRT-Systeme testen

Die Version des Code-Snippets von Pac-Man auf einem einzigen Computer lässt sich recht einfach testen. Sie erstellen verschiedene Spielfeld-Objekte, geben ihnen verschiedene Status, erstellen verschiedene Benutzer-Objekte in verschiedenen Status und so weiter. Entwickler würden sich Gedanken über Randbedingungen machen und vielleicht Generative Testing oder Fuzzing einsetzen.

Im Pac-Man-Code gibt es vier Stellen, bei denen das Spielfeld-Objekt verwendet wird. Im verteilten Pac-Man gibt es vier Stellen im Code mit fünf unterschiedlichen möglichen Ergebnissen, wie zuvor dargestellt (SENDEN_FEHLGESCHLAGEN, WIEDERHOLEN, AUSFALL, UNBEKANNT oder ERFOLG). Dadurch vervielfachen sich die zu testenden Status. Zum Beispiel müssen Entwickler von verteilten HRT-Systemen Mechanismen für viele verschiedene Permutationen einrichten. Nehmen wir einmal an, dass spielfeld.find() fehlschlägt und SENDEN_FEHLGESCHLAGEN ausgegeben wird. Dann müssen Sie testen, was passiert, wenn der Vorgang fehlschlägt und WIEDERHOLEN ausgegeben wird, dann müssen Sie testen, was passiert, wenn der Vorgang fehlschlägt und AUSFALL ausgegeben wird – und so weiter.

Doch diese aufwendigen Tests reichen noch nicht. Bei normalem Code gehen Entwickler davon aus, dass wenn spielfeld.find() funktioniert, auch der nächste Request, z. B. spielfeld.move(), funktionieren wird. Bei der Entwicklung verteilter HRT-Systeme ist das nicht garantiert. Der Server könnte jederzeit unabhängig ausfallen. Deshalb müssen Entwickler zu jedem Spielfeld-Zugriff für alle fünf Fälle Tests schreiben. Nehmen wir einmal an, dass einem Entwickler bei der Pac-Man-Version auf einem einzigen Computer 10 Testszenarien eingefallen sind. Bei der Version mit den verteilten Systemen muss er jedes dieser Szenarien 20 Mal testen. Somit wächst die Testmatrix von 10 auf 200 an!

Aber das ist noch nicht alles. Der Entwickler könnte auch für den Servercode zuständig sein. Er muss bei jeder erdenklichen Kombination aus Client-, Netzwerk- und Server-Fehlern testen, dass weder der Client noch der Server einen Ausfall verursacht. Der Servercode könnte etwa wie folgt aussehen.

handleFind(channel, message)
  if !validate(message)
    channel.send(INVALID_MESSAGE)
    return
  if !userThrottle.ok(message.user())
    channel.send(RETRYABLE_ERROR)
    return
  location = database.lookup(message.user())
  if location.error()
    channel.send(USER_NOT_FOUND)
    return
  else
    channel.send(SUCCESS, location)

handleMove(...)
  ...

handleFindAll(...)
  ...

handleRemove(...)
  ...

Vier Server-Funktionen müssen getestet werden. Nehmen wir einmal an, dass für jede Funktion bei einem einzigen Computer fünf Tests erforderlich sind. Das ergibt 20 Tests. Clients senden mehrere Nachrichten an denselben Server, deshalb sollten Tests auch Abfolgen mit mehreren Requests simulieren und damit die Zuverlässigkeit des Servers überprüfen. Typische Requests sind "find", "move", "remove" und "findAll".

Nehmen wir an, dass es für ein Konstrukt 10 verschiedene Szenarien mit durchschnittlich drei Anfragen pro Szenario gibt. Das ergibt 30 weitere Tests. Aber in einem Szenario müssen auch die Fehler getestet werden. Sie müssen für jeden dieser Tests simulieren, was passiert, wenn der Client einen der vier Fehlertypen erhalten hat (SENDEN_FEHLGESCHLAGEN, WIEDERHOLEN, AUSFALL und UNBEKANNT) und dann noch einmal eine ungültige Request an den Server sendet. Beispielsweise könnte ein "find"-Request erfolgreich sein, doch ein "move"-Request manchmal "UNBEKANNT" an den Client ausgeben. Danach könnte der Client aus bestimmten Gründen wieder einen "find"-Request senden. Geht der Server mit diesem Fall korrekt um? Wahrscheinlich, aber nur ein Test kann Gewissheit bringen. Also nicht nur die Testmatrix des Clients, sondern auch die des Servers wächst extrem an.

Unbekannte Faktoren berücksichtigen

Der bloße Gedanke an alle möglichen Permutationen, die bei einem verteilten System auftreten können, ist schwindelerregend – gerade bei mehreren Requests. Einer unserer Ansätze bei der Entwicklung verteilter Systeme ist es, allem zu misstrauen. Jede Codezeile, außer sie kann auf keinen Fall eine Netzwerkkommunikation auslösen, könnte Probleme verursachen.

Das Schwierigste daran ist vermutlich der richtige Umgang mit dem Fehlertyp "UNBEKANNT", der zuvor bereits erwähnt wurde. Der Client weiß nicht immer, ob die Request erfolgreich war. Vielleicht hat die Request Pac-Man bewegt (oder bei einem Bankingservice Geld vom Konto des Benutzers abgehoben), vielleicht auch nicht. Wie sollen Entwickler diesen Problemen begegnen? Entwickler sind auch nur Menschen, und Menschen kommen meist nicht besonders gut zurecht mit der Ungewissheit. Menschen sind Code wie den folgenden gewohnt.

bool isEven(number)
  switch number % 2
    case 0
      return true
    case 1
      return false

Menschen verstehen diesen Code, denn er macht genau das, was er auch zu machen scheint. Menschen haben ihre Probleme mit der verteilten Version des Codes, der einen Teil der Arbeit an einen Service abgibt.

bool distributedIsEven(number)
  switch mathServer.mod(number, 2)
    case 0
      return true
    case 1
      return false
    case UNKNOWN
      return WHAT_THE_FARG?

Es ist für einen Menschen auch fast unmöglich, festzustellen, wie er mit dem UNBEKANNTEN richtig umgehen sollte. Was bedeutet UNBEKANNT eigentlich? Sollte der Code den Vorgang wiederholen? Wenn ja, wie oft? Wie lange soll der Code zwischen den Wiederholungen warten? Und es wird noch schlimmer, wenn der Code weitere Auswirkungen hat. Bei einer Budget-Anwendung auf einem einzigen Gerät ist es einfach, Geld von einem Konto abzuheben, wie das nachfolgende Beispiel zeigt.

class Teller
  bool doWithdraw(account, amount)
    switch account.withdraw(amount)
      case SUCCESS
        return true
      case INSUFFICIENT_FUNDS
        return false

Doch das UNBEKANNTE macht die verteilte Version dieser Anwendung seltsam.

class DistributedTeller
  bool doWithdraw(account, amount)
    switch this.accountService.withdraw(account, amount)
      case SUCCESS
        return true
      case INSUFFICIENT_FUNDS
        return false
      case UNKNOWN
        return WHAT_THE_FARG?

Der richtige Umgang mit dem Fehlertyp "UNBEKANNT" ist einer der Gründe dafür, dass bei der Entwicklung verteilter Systeme der Schein trügen kann.

Gruppen aus verteilten HRT-Systemen

Die acht apokalyptischen Fehlermodi können bei jeder Abstraktionsebene eines verteilten Systems auftauchen. Das vorherige Beispiel begrenzte sich auf einen einzigen Client, ein Netzwerk und einen einzigen Server. Selbst in diesem vereinfachten Szenario zeigte sich schon die zunehmende Komplexität der Fehlerstatus-Matrix. Bei echten verteilten Systemen sind die Fehlerstatus-Matrizen komplizierter als bei dem Beispiel mit dem einzelnen Client. Echte verteilte Systeme bestehen aus mehreren Servern, die auf mehreren Abstraktionsebenen betrachtet werden können:

1. Einzelne Server
2. Server-Gruppen
3. Server-Gruppen-Gruppen
4. Und so weiter (theoretisch)

Zum Beispiel könnten verschiedene Server, die Ressourcen innerhalb einer bestimmten Availability Zone verwalten, von einem auf AWS basierenden Service in Gruppen zusammengefasst werden. Es könnte auch zwei weitere Servergruppen geben, die zwei andere Availability Zones verwalten. Diese Gruppen könnten wiederum in eine AWS Region-Gruppe zusammengefasst werden. Und diese Region-Gruppe könnte mit anderen Region-Gruppen (logisch) kommunizieren. Leider gelten die gesamten Probleme auch auf dieser höheren, logischeren Ebene.

Nehmen wir an, dass ein Service mehrere Server in eine einzige logische Gruppe zusammengefasst hat: GRUPPE1. GRUPPE1 könnte manchmal Nachrichten an eine andere Servergruppe "GRUPPE2" senden. Das ist ein Beispiel für die rekursive verteilte Programmierung. Alle zuvor beschriebenen Netzwerk-Fehlermodi gelten auch hier. GRUPPE1 möchte zum Beispiel eine Request an GRUPPE2 senden. Wie die nachfolgende Abbildung zeigt, ist die Request/Reply-Interaktion zwischen den beiden Servern wie die zuvor besprochene des einzelnen Servers.

Ein Server von GRUPPE1 muss eine an GRUPPE2 (logisch) adressierte Nachricht ins NETZWERK senden. Ein Server von GRUPPE2 muss die Request verarbeiten und so weiter. GRUPPE1 und GRUPPE2 sind zwar Servergruppen, aber es gelten trotzdem die gleichen Prinzipien. GRUPPE1, GRUPPE2 und das NETZWERK können immer noch unabhängig voneinander Fehler verursachen.

Doch das ist nur die Betrachtung auf der Gruppen-Ebene. Es gibt außerdem noch die Interaktionen auf der Ebene einzelner Server innerhalb der Gruppe. Zum Beispiel könnte GRUPPE2 wie in der nachfolgenden Abbildung strukturiert sein.

Zuerst wird über den Load Balancer eine Nachricht an einen Server (nennen wir ihn "S20") der GRUPPE2 gesendet. Die Entwickler des Systems wissen, dass S20 während der Phase "STATUS AKTUALISIEREN" ausfallen könnte. Deshalb muss S20 die Nachricht an mindestens einen weiteren Server weiterleiten, der derselben oder einer anderen Gruppe angehören kann. Wie bewerkstelligt S20 das? Indem er eine Request/Reply-Nachricht (zum Beispiel) an S25 sendet, so wie es auch auf der nachfolgenden Abbildung dargestellt ist.

Folglich agiert S20 im Netzwerk rekursiv. Es können wieder alle acht Fehler unabhängig voneinander auftreten. Die verteilte Entwicklung wird nicht mehr einmal, sondern zweimal angewandt. Die Nachricht von GRUPPE1 an GRUPPE2 kann auf der logischen Ebene in allen acht Schritten fehlschlagen. Die Nachricht löst eine weitere Nachricht aus, deren Übertragung unabhängig und wieder in allen acht genannten Schritten fehlschlagen kann. Dieses Szenario müsste mindestens den folgenden Tests unterzogen werden:

• Ein Test für alle acht Schritte, in denen die Übertragung der Gruppen-Ebenen-Nachricht von GRUPPE1 zu GRUPPE2 fehlschlagen kann.
• Ein Test für alle acht Schritte, in denen die Übertragung der Server-Ebenen-Nachricht von S20 zu S25 fehlschlagen kann.

Das Beispiel mit den Request/Reply-Nachrichten verdeutlicht, warum es so schwierig ist, verteilte Systeme zu testen – selbst nach über 20 Jahren Erfahrung. Es gibt unzählige Sonderfälle, deshalb sind die Tests so herausfordernd, aber zugleich sind sie bei diesen Systemen besonders wichtig. Bugs können erst sehr spät nach der Systembereitstellung in Erscheinung treten. Und die Bugs können unvorhersehbare schwere Auswirkungen auf ein System und die benachbarten Systeme haben.

Verteilte Bugs sind oft latent

Ein "vorprogrammierter" Fehler sollte besser zu früh als zu spät auftreten. Wenn zum Beispiel ein Skalierungsproblem bei einem Service vorliegt, dessen Behebung sechs Monate dauert, ist es natürlich besser, dieses Problem spätestens sechs Monate vor dem Zeitpunkt zu finden, an dem der Service die entsprechende Skalierungsgröße erreicht. Bugs sollten ja auch gefunden werden, bevor die Lösung in die Produktion übergeht. Wenn die Bugs erst in der Produktion auftreten, ist es besser, sie zu finden, bevor sie sich auf viele Kunden auswirken oder andere negative Folgen haben.

Verteilte Bugs treten auf, wenn nicht alle Permutationen der acht apokalyptischen Fehlermodi korrekt behandelt werden, und diese sind oft besonders schwerwiegend. Es gibt unzählige Beispiele für latente Störungen in großen verteilten Systemen, von Telekommunikationssystemen bis hin zu grundlegenden Internetsystemen. Diese Ausfälle sind nicht nur weitverbreitet und teuer, sie können auch von Bugs verursacht werden, die Monate vorher in der Produktion bereitgestellt wurden. Es dauert dann eine Weile, bis die Kombination aus den verschiedenen Szenarien auftritt, die diese Bugs auslöst (und auf das gesamte System verbreitet).

Verteilte Bugs verbreiten sich wie eine Epidemie

Es gibt noch ein weiteres inhärentes Problem von verteilten Bugs:

1. Verteilte Bugs treten per Definition in vernetzten Systemen auf.
2. Die Wahrscheinlichkeit ist bei verteilten Bugs höher, dass sie sich auf andere Server (oder Servergruppen) ausbreiten, sie tragen es ja bereits im Namen.

Auch Amazon hat bereits Bekanntschaft mit den verteilten Bugs gemacht. Ein altes, aber relevantes Beispiel dafür ist der vollständige Ausfall von www.amazon.com. Das gesamte System ist ausgefallen, weil ein einziger Server im Remote-Servicekatalog ausgefallen ist, als seine Festplatte voll wurde.

Die Fehlerbedingung wurde falsch behandelt und der Remote-Katalog-Server sendete leere Replies für alle erhaltenen Requests. Außerdem wurden die Replies sehr schnell gesendet, denn "nichts" lässt sich sehr viel schneller zurückgeben als "etwas" (oder zumindest war das in diesem Fall so). Währenddessen erkannte der Load Balancer zwischen der Website und dem Remote-Katalogservice nicht, dass alle Replies eine Länge von null hatten. Doch der Load Balancer erkannte, dass die Replies deutlich schneller als die der anderen Remote-Katalogserver waren. Deshalb leitete der Load Balancer einen riesigen Anteil des Traffics von www.amazon.com auf den einen Remote-Katalogserver, dessen Festplatte voll war. Im Endeffekt ist die gesamte Website ausgefallen, weil ein Remote-Server keine Produktinformationen mehr anzeigen konnte.

Wir haben den fehlerhaften Server schnell gefunden und ihn aus dem Service entfernt, damit die Website wieder funktionierte. Danach folgte unser üblicher Prozess, bei dem wir die Ursachen feststellen und die Probleme identifizieren, um ein erneutes Auftreten des Fehlers zu verhindern. Wir haben alle relevanten Mitarbeiter bei Amazon über unser Fazit informiert, damit das gleiche Problem nicht noch einmal bei einem anderen System auftritt. Wir konnten nicht nur wichtige Informationen über diesen speziellen Fehlermodus, sondern auch über die schnelle und unvorhersehbare Ausbreitung von Fehlermodi in verteilten Systemen gewinnen.

Fazit zu den Problemen in verteilten Systemen

Verteilte Systeme sind schwierig zu entwickeln. Das hat die folgenden Gründe:

• Entwickler können Fehlerbedingungen nicht kombinieren. Stattdessen müssen sie viele verschiedene Fehler-Permutationen berücksichtigen. Die meisten Fehler können jederzeit und unabhängig von (und deshalb möglicherweise auch in Kombination mit) jeder anderen Fehlerbedingung auftreten.
• Das Ergebnis von jeder Netzwerkoperation kann UNBEKANNT sein, das heißt der Request könnte erfolgreich gewesen sein, fehlgeschlagen sein oder angekommen sein, aber nicht verarbeitet worden sein.
• Verteilte Probleme können auf jeder logischen Ebene eines verteilten Systems vorkommen, nicht nur auf unterster Hardwareebene.
• Rekursion führt dazu, dass verteilte Probleme auf höheren Systemebenen immer schlimmer werden.
• Verteilte Bugs zeigen sich oft erst lange nach der Bereitstellung auf einem System.
• Verteilte Bugs können das gesamte System befallen.
• Viele der oben genannten Probleme beruhen auf physikalischen Gesetzen, die für Netzwerke gelten und nicht geändert werden können.

Die verteilte Verarbeitung ist komplex, und auch seltsam, doch es gibt Lösungen für diese Probleme. In der Amazon Builders’ Library wird an vielen verschiedenen Stellen erklärt, wie AWS die verteilten Systeme meistert. Wir hoffen, dass wir Ihnen mit einigen unserer Erkenntnisse dabei helfen können, zuverlässige Systeme für Ihre Kunden zu entwickeln.


Über den Autor

Jacob Gabrielson ist Senior Principal Engineer bei Amazon Web Services. Er arbeitet seit 17 Jahren für Amazon, hauptsächlich an den internen Mikroservice-Plattformen. In den vergangenen 8 Jahren hat er sich der Arbeit an EC2 und ECS gewidmet, darunter an Systemen zur Softwarebereitstellung, Services für Steuerebenen, dem Spot-Markt, Lightsail und jüngst auch Containern. Jacob beschäftigt sich gerne mit Systemprogrammierung, Programmiersprachen und verteilter Verarbeitung. Er ist nicht besonders begeistert vom bimodalen Systemverhalten, besonders in Fehlerzuständen. Er hat seinen Bachelor in Informatik an der University of Washington in Seattle abgeschlossen.

Timeouts, Wiederholungsversuche und Backoff mit Jitter Vermeiden von Fallback in verteilten Systemen