Das Problem
Elasticsearch kann einem Client nicht sagen, was sich geändert hat. Issue #1242, "Changes API", ist seit August 2011 offen und hat 330 Reaktionen und 173 Kommentare. Gewünscht wird das für Cache-Invalidierung, den Abgleich mit anderen Datenbanken und Kopien für den Notfall.
Was es heute gibt, deckt den allgemeinen Fall nicht ab. Cross-Cluster-Replication liest die Shard-Historie, aber nur in einen anderen Elasticsearch-Cluster und nur mit Platinum- oder Enterprise-Lizenz. Abfragen über ein Zeitstempelfeld verlangen, dass jeder Schreiber das Feld setzt, können Dokumente verpassen, deren Zeitstempel hinter dem Abfragefenster liegt, und sehen keine Löschungen. Die im Issue verlinkten Plugins wurden zuletzt 2016 und 2021 geändert und lassen sich auf gehosteten Clustern nicht installieren.
Wie es arbeitet
Seit Version 6.0 bekommt jeder Schreibvorgang in einem Shard eine Sequenznummer, und jedes Dokument speichert die Nummer des Vorgangs, der es zuletzt geschrieben hat. es-changefeed merkt sich pro Primär-Shard die höchste ausgelieferte Sequenznummer. In jeder Runde liest es den globalen Checkpoint des Shards aus den Index-Statistiken, ruft bei Änderungen einen Refresh auf und sucht in diesem Shard nach Dokumenten mit einer Sequenznummer zwischen seinem Checkpoint und dem globalen Checkpoint, sortiert nach Sequenznummer.
Der globale Checkpoint macht das korrekt. Indexierungs-Threads werden in beliebiger Reihenfolge fertig, ein Dokument mit Sequenznummer 7 kann also vor einem mit 6 sichtbar werden. Ein Programm, das nach "größer als die höchste bisher gesehene Nummer" fragt, würde 6 überspringen. Alles bis zum globalen Checkpoint ist auf allen synchronen Kopien verarbeitet, nach einem Refresh fehlt davon also nichts. Vorgänge oberhalb des globalen Checkpoints können nach dem Ausfall eines Primärs zurückgerollt werden, und der Feed liest sie nie.
Löschungen hinterlassen nichts, was eine Suche finden könnte. Mit -deletes merkt sich der Feed, welche Dokumente er ausgeliefert hat, listet alle paar Minuten die noch vorhandenen IDs und meldet jedes ausgelieferte Dokument, das fehlt, als gelöscht. Der Abgleich läuft zwischen den Shard-Abfragen eines Index. Eine Neuanlage, die noch nicht sichtbar ist, hat deshalb eine höhere Sequenznummer als alles Ausgelieferte, und ihr Upsert kommt nach der Löschung.
Ausgaben gehen nach stdout, in eine Datei, an einen HTTP-Endpunkt oder per _bulk in einen anderen Cluster. Muster, Aliase und Data Streams werden über Rollover hinweg verfolgt. Ausgabe aus dem CLI-Test gegen Elasticsearch 9.5.5, nach einer Änderung und einer Löschung:
{"op":"upsert","index":"escf-cli","index_uuid":"BZfHMdwmRv-iskXKFEMGgA","shard":1,"id":"1","seq_no":2,"primary_term":1,"version":2,"source":{"title":"edited"}}
{"op":"delete","index":"escf-cli","index_uuid":"BZfHMdwmRv-iskXKFEMGgA","shard":1,"id":"2"}Wie es getestet ist
Die Unit-Tests laufen gegen einen Fake-Cluster im Speicher, der nachbildet, wovon der Algorithmus abhängt: Sequenznummern, die vergeben werden, bevor ein Vorgang fertig ist, lokale und globale Checkpoints und Sichtbarkeit, die sich nur beim Refresh ändert. Ein Zufallstest mischt Vorgänge außer der Reihe, nachlaufende Checkpoints, Routing, Löschungen und Neustarts und prüft, dass das Abspielen der Events die Quelle reproduziert.
Das CI führt dieselben Abläufe gegen echte Cluster aus: Elasticsearch 9.5.5, 8.19.23 und 7.17.28, OpenSearch 3.9.0 und 2.19.7 sowie Elasticsearch mit TLS und Passwort. Vier Schreiber indexieren, ändern und löschen acht Sekunden lang, während der Feed in einen zweiten Index spiegelt, danach müssen beide dieselben Dokumente enthalten. Ein Job tut das auf einem Cluster mit drei Knoten und Replikas und schießt mittendrin einen Knoten ab.
Alle Jobs sind grün. Die Spiegel enthielten zwischen 1573 und 2029 Dokumente und stimmten in jedem Job exakt mit der Quelle überein, auch im Job mit dem abgeschossenen Knoten. Die ersten Läufe gegen echte Cluster fanden zwei Fehler, die der Fake nicht gezeigt hatte. Dokumente wurden nur über die _id verfolgt, aber mit eigenem Routing kann dieselbe _id als zwei Dokumente auf zwei Shards liegen, deshalb werden sie jetzt pro Shard geführt. Und als der Test den Feed anhielt, während eine Anfrage an den Sink unterwegs war, kam die Anfrage trotzdem an. Der Feed hielt sie für gescheitert, und wurde das Dokument vor der Wiederholung gelöscht, ging nie eine Löschung raus. Die IDs werden jetzt vor dem Schreiben vermerkt. Beim Lesen der CI-Ausgabe fiel ein dritter, kleinerer Fehler auf: Das erste Dokument eines Shards hat die Sequenznummer 0, und der JSON-Encoder ließ das Feld weg.
Grenzen
Jedes Event ist der aktuelle Stand eines Dokuments. Ändert sich ein Dokument zwischen zwei Runden zweimal, wird nur die zweite Fassung ausgegeben. Die Erkennung von Löschungen hält alle ausgelieferten IDs im Speicher (in meiner Messung etwa 110 Byte pro ID) und listet pro Abgleich alle IDs eines Index. Nicht getestet habe ich Elastic Cloud und Amazon OpenSearch Service, Indizes mit mehr als einigen tausend Dokumenten und einen Failover, der tatsächlich Vorgänge zurückrollt.
Der Code liegt unter github.com/Jwrede/es-changefeed.