Comment créer votre propre système de stockage KV distribué à l'aide de la bibliothèque Raft etcd

Introduction

raftexample est un exemple fourni par etcd qui démontre l'utilisation de la bibliothèque d'algorithmes de consensus etcd raft. raftexample implémente finalement un service de stockage clé-valeur distribué qui fournit une API REST.

Cet article lira et analysera le code de raftexample, dans l'espoir d'aider les lecteurs à mieux comprendre comment utiliser la bibliothèque raft etcd et la logique d'implémentation de la bibliothèque raft.

Architecture

L'architecture de raftexample est très simple, avec les fichiers principaux comme suit :

• main.go : Responsable de l'organisation de l'interaction entre le module raft, le module httpapi et le module kvstore ;
• raft.go : Responsable de l'interaction avec la bibliothèque raft, y compris la soumission de propositions, la réception des messages RPC qui doivent être envoyés et la transmission réseau, etc. ;
• httpapi.go : Responsable de fournir l'API REST, servant de point d'entrée aux demandes des utilisateurs ;
• kvstore.go : Responsable du stockage persistant des entrées de journal validées, équivalent à la machine à états dans le protocole raft.

Le flux de traitement d'une demande d'écriture

Une requête d'écriture arrive dans la méthode ServeHTTP du module httpapi via une requête HTTP PUT.

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

Après avoir fait correspondre la méthode de requête HTTP via switch, elle entre dans le flux de traitement de la méthode PUT :

• Lire le contenu du corps de la requête HTTP (c'est-à-dire la valeur) ;
• Construire une proposition via la méthode Propose du module kvstore (en ajoutant une paire clé-valeur avec clé comme clé et valeur comme valeur) ;
• Comme il n'y a aucune donnée à renvoyer, répondez au client avec 204 StatusNoContent ;

La proposition est soumise à la bibliothèque d'algorithmes raft via la méthode Propose fournie par la bibliothèque d'algorithmes raft.

Le contenu d'une proposition peut consister à ajouter une nouvelle paire clé-valeur, à mettre à jour une paire clé-valeur existante, etc.

// 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)

Ensuite, examinons la méthode Propose du module kvstore pour voir comment une proposition est construite et traitée.

Dans la méthode Propose, nous encodons d'abord la paire clé-valeur à écrire en utilisant gob, puis transmettons le contenu encodé à proposerC, un canal chargé de transmettre les propositions construites par le module kvstore au module raft.

// 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()
}

La proposition construite par kvstore et transmise à proposeC est reçue et traitée par la méthode serveChannels dans le module raft.

Après avoir confirmé que proposeC n'a pas été fermé, le module raft soumet la proposition à la bibliothèque d'algorithmes raft pour traitement à l'aide de la méthode Propose fournie par la bibliothèque d'algorithmes raft.

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

Une fois qu'une proposition est soumise, elle suit le processus de l'algorithme du radeau. La proposition sera finalement transmise au nœud leader (si le nœud actuel n'est pas le leader et que vous autorisez les abonnés à transmettre des propositions, contrôlé par la configuration DisableProposalForwarding). Le leader ajoutera la proposition en tant qu'entrée de journal à son journal de radeau et la synchronisera avec d'autres nœuds suiveurs. Après avoir été réputé validé, il sera appliqué à la machine à états et le résultat sera renvoyé à l'utilisateur.

Cependant, étant donné que la bibliothèque raft etcd elle-même ne gère pas la communication entre les nœuds, l'ajout au journal raft, l'application à la machine d'état, etc., la bibliothèque raft ne prépare que les données requises pour ces opérations. Les opérations proprement dites doivent être réalisées par nos soins.

Par conséquent, nous devons recevoir ces données de la bibliothèque raft et les traiter en conséquence en fonction de leur type. La méthode Ready renvoie un canal en lecture seule à travers lequel nous pouvons recevoir les données à traiter.

Il convient de noter que les données reçues comprennent plusieurs champs, tels que les instantanés à appliquer, les entrées de journal à ajouter au journal du radeau, les messages à transmettre sur le réseau, etc.

En continuant avec notre exemple de demande d'écriture (nœud leader), après avoir reçu les données correspondantes, nous devons enregistrer de manière persistante les instantanés, le HardState et les entrées pour gérer les problèmes causés par des pannes de serveur (par exemple, un suiveur votant pour plusieurs candidats). HardState et Entries constituent ensemble l’état persistant sur tous les serveurs, comme mentionné dans le document. Après les avoir sauvegardés de manière persistante, nous pouvons appliquer l'instantané et l'ajouter au journal du radeau.

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

Ce qui précède est le contenu détaillé de. pour plus d'informations, suivez d'autres articles connexes sur le site Web de PHP en chinois!