So erstellen Sie Ihr eigenes verteiltes KV-Speichersystem mithilfe der etcd Raft-Bibliothek

Einführung

raftexample ist ein von etcd bereitgestelltes Beispiel, das die Verwendung der etcd-Raft-Konsensalgorithmus-Bibliothek demonstriert. Raftexample implementiert letztendlich einen verteilten Schlüsselwert-Speicherdienst, der eine REST-API bereitstellt.

In diesem Artikel wird der Code von „raftexample“ gelesen und analysiert, in der Hoffnung, den Lesern dabei zu helfen, die Verwendung der etcd-Raft-Bibliothek und die Implementierungslogik der Raft-Bibliothek besser zu verstehen.

Architektur

Die Architektur von RaftExample ist sehr einfach, mit den Hauptdateien wie folgt:

• main.go: Verantwortlich für die Organisation der Interaktion zwischen dem Raft-Modul, dem httpapi-Modul und dem kvstore-Modul;
• raft.go: Verantwortlich für die Interaktion mit der Raft-Bibliothek, einschließlich der Einreichung von Vorschlägen, dem Empfang von RPC-Nachrichten, die gesendet werden müssen, und der Durchführung von Netzwerkübertragungen usw.;
• httpapi.go: Verantwortlich für die Bereitstellung der REST-API, die als Einstiegspunkt für Benutzeranfragen dient;
• kvstore.go: Verantwortlich für die dauerhafte Speicherung festgeschriebener Protokolleinträge, äquivalent zur Zustandsmaschine im Raft-Protokoll.

Der Verarbeitungsablauf einer Schreibanforderung

Eine Schreibanforderung kommt über eine HTTP-PUT-Anfrage in der ServeHTTP-Methode des httpapi-Moduls an.

curl -L http://127.0.0.1:12380/key -XPUT -d value

Nachdem die HTTP-Anforderungsmethode über den Schalter abgeglichen wurde, gelangt sie in den Verarbeitungsablauf der PUT-Methode:

• Lesen Sie den Inhalt aus dem HTTP-Anfragetext (d. h. den Wert);
• Erstellen Sie einen Vorschlag mithilfe der Propose-Methode des kvstore-Moduls (Hinzufügen eines Schlüssel-Wert-Paares mit Schlüssel als Schlüssel und Wert als Wert);
• Da keine Daten zurückgegeben werden müssen, antworten Sie dem Client mit 204 StatusNoContent;

Der Vorschlag wird über die von der Raft-Algorithmusbibliothek bereitgestellte Propose-Methode an die Raft-Algorithmusbibliothek übermittelt.

Der Inhalt eines Vorschlags kann das Hinzufügen eines neuen Schlüssel-Wert-Paares, das Aktualisieren eines vorhandenen Schlüssel-Wert-Paares usw. sein.

// httpapi.go
v, err := io.ReadAll(r.Body)
if err != nil {
    log.Printf("Failed to read on PUT (%v)\n", err)
    http.Error(w, "Failed on PUT", http.StatusBadRequest)
    return
}
h.store.Propose(key, string(v))
w.WriteHeader(http.StatusNoContent)

Als nächstes schauen wir uns die Propose-Methode des kvstore-Moduls an, um zu sehen, wie ein Vorschlag erstellt und verarbeitet wird.

In der Propose-Methode codieren wir zunächst das zu schreibende Schlüssel-Wert-Paar mit gob und übergeben dann den codierten Inhalt an ProposeC, einen Kanal, der für die Übertragung der vom KVStore-Modul erstellten Vorschläge an das Raft-Modul verantwortlich ist.

// kvstore.go
func (s *kvstore) Propose(k string, v string) {
    var buf strings.Builder
    if err := gob.NewEncoder(&buf).Encode(kv{k, v}); err != nil {
        log.Fatal(err)
    }
    s.proposeC <- buf.String()
}

Der von kvstore erstellte und an ProposC übergebene Vorschlag wird von der ServeChannels-Methode im Raft-Modul empfangen und verarbeitet.

Nachdem bestätigt wurde, dass „proposC“ nicht geschlossen wurde, übermittelt das Raft-Modul den Vorschlag an die Raft-Algorithmusbibliothek zur Verarbeitung mithilfe der von der Raft-Algorithmusbibliothek bereitgestellten Propose-Methode.

// raft.go
select {
    case prop, ok := <-rc.proposeC:
    if !ok {
        rc.proposeC = nil
    } else {
        rc.node.Propose(context.TODO(), []byte(prop))
    }

Nachdem ein Vorschlag eingereicht wurde, folgt er dem Raft-Algorithmus-Prozess. Der Vorschlag wird schließlich an den führenden Knoten weitergeleitet (wenn der aktuelle Knoten nicht der führende Knoten ist und Sie Followern erlauben, Vorschläge weiterzuleiten, gesteuert durch die DisableProposalForwarding-Konfiguration). Der Anführer fügt den Vorschlag als Protokolleintrag zu seinem Raft-Protokoll hinzu und synchronisiert ihn mit anderen Folgeknoten. Nachdem es als festgeschrieben gilt, wird es auf die Zustandsmaschine angewendet und das Ergebnis wird an den Benutzer zurückgegeben.

Da sich die etcd-Raft-Bibliothek selbst jedoch nicht um die Kommunikation zwischen Knoten, das Anhängen an das Raft-Protokoll, die Anwendung auf die Zustandsmaschine usw. kümmert, bereitet die Raft-Bibliothek nur die für diese Vorgänge erforderlichen Daten vor. Die eigentlichen Arbeiten müssen von uns durchgeführt werden.

Daher müssen wir diese Daten von der Floßbibliothek erhalten und sie je nach Typ entsprechend verarbeiten. Die Ready-Methode gibt einen schreibgeschützten Kanal zurück, über den wir die zu verarbeitenden Daten empfangen können.

Es ist zu beachten, dass die empfangenen Daten mehrere Felder umfassen, z. B. anzuwendende Snapshots, Protokolleinträge, die an das Raft-Protokoll angehängt werden sollen, über das Netzwerk zu übertragende Nachrichten usw.

Um mit unserem Schreibanforderungsbeispiel (Leader-Knoten) fortzufahren, müssen wir nach Erhalt der entsprechenden Daten Snapshots, HardState und Einträge dauerhaft speichern, um Probleme zu bewältigen, die durch Serverabstürze verursacht werden (z. B. wenn ein Follower für mehrere Kandidaten stimmt). HardState und Entries bilden zusammen den Persistent-Status auf allen Servern, wie im Dokument erwähnt. Nachdem wir sie dauerhaft gespeichert haben, können wir den Schnappschuss anwenden und an das Raft-Protokoll anhängen.

Since we are currently the leader node, the raft library will return MsgApp type messages to us (corresponding to AppendEntries RPC in the paper). We need to send these messages to the follower nodes. Here, we use the rafthttp provided by etcd for node communication and send the messages to follower nodes using the Send method.

// raft.go
case rd := <-rc.node.Ready():
    if !raft.IsEmptySnap(rd.Snapshot) {
        rc.saveSnap(rd.Snapshot)
    }
    rc.wal.Save(rd.HardState, rd.Entries)
    if !raft.IsEmptySnap(rd.Snapshot) {
        rc.raftStorage.ApplySnapshot(rd.Snapshot)
        rc.publishSnapshot(rd.Snapshot)
    }
    rc.raftStorage.Append(rd.Entries)
    rc.transport.Send(rc.processMessages(rd.Messages))
    applyDoneC, ok := rc.publishEntries(rc.entriesToApply(rd.CommittedEntries))
    if !ok {
        rc.stop()
        return
    }
    rc.maybeTriggerSnapshot(applyDoneC)
    rc.node.Advance()

Next, we use the publishEntries method to apply the committed raft log entries to the state machine. As mentioned earlier, in raftexample, the kvstore module acts as the state machine. In the publishEntries method, we pass the log entries that need to be applied to the state machine to commitC. Similar to the earlier proposeC, commitC is responsible for transmitting the log entries that the raft module has deemed committed to the kvstore module for application to the state machine.

// raft.go
rc.commitC <- &commit{data, applyDoneC}

In the readCommits method of the kvstore module, messages read from commitC are gob-decoded to retrieve the original key-value pairs, which are then stored in a map structure within the kvstore module.

// kvstore.go
for commit := range commitC {
    ...
    for _, data := range commit.data {
        var dataKv kv
        dec := gob.NewDecoder(bytes.NewBufferString(data))
        if err := dec.Decode(&dataKv); err != nil {
            log.Fatalf("raftexample: could not decode message (%v)", err)
        }
        s.mu.Lock()
        s.kvStore[dataKv.Key] = dataKv.Val
        s.mu.Unlock()
    }
    close(commit.applyDoneC)
}

Returning to the raft module, we use the Advance method to notify the raft library that we have finished processing the data read from the Ready channel and are ready to process the next batch of data.

Earlier, on the leader node, we sent MsgApp type messages to the follower nodes using the Send method. The follower node's rafthttp listens on the corresponding port to receive requests and return responses. Whether it's a request received by a follower node or a response received by a leader node, it will be submitted to the raft library for processing through the Step method.

raftNode implements the Raft interface in rafthttp, and the Process method of the Raft interface is called to handle the received request content (such as MsgApp messages).

// raft.go
func (rc *raftNode) Process(ctx context.Context, m raftpb.Message) error {
    return rc.node.Step(ctx, m)
}

The above describes the complete processing flow of a write request in raftexample.

Summary

This concludes the content of this article. By outlining the structure of raftexample and detailing the processing flow of a write request, I hope to help you better understand how to use the etcd raft library to build your own distributed KV storage service.

If there are any mistakes or issues, please feel free to comment or message me directly. Thank you.

References

• https://github.com/etcd-io/etcd/tree/main/contrib/raftexample

• https://github.com/etcd-io/raft

• https://raft.github.io/raft.pdf

Das obige ist der detaillierte Inhalt vonSo erstellen Sie Ihr eigenes verteiltes KV-Speichersystem mithilfe der etcd Raft-Bibliothek. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!