etcd Raft ライブラリを使用して独自の分散 KV ストレージ システムを構築する方法

導入

raftexample は、etcd raft コンセンサス アルゴリズム ライブラリの使用法を示す etcd によって提供される例です。 raftexample は最終的に、REST API を提供する分散キー値ストレージ サービスを実装します。

この記事では、読者が etcd raft ライブラリの使用方法と raft ライブラリの実装ロジックをよりよく理解できるように、raftexample のコードを読んで分析します。

建築

raftexample のアーキテクチャは非常にシンプルで、主なファイルは次のとおりです。

• main.go: raft モジュール、httpapi モジュール、kvstore モジュール間の対話を整理する責任を負います。
• raft.go: 提案の送信、送信する必要がある RPC メッセージの受信、ネットワーク送信の実行など、raft ライブラリとの対話を担当します。
• httpapi.go: REST API の提供を担当し、ユーザーリクエストのエントリポイントとして機能します。
• kvstore.go: コミットされたログ エントリを永続的に保存する役割を果たします。これは、raft プロトコルのステート マシンに相当します。

書き込みリクエストの処理フロー

書き込みリクエストは、HTTP PUT リクエストを介して httpapi モジュールの ServeHTTP メソッドに到着します。

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

スイッチを介して HTTP リクエスト メソッドを照合した後、PUT メソッドの処理フローに入ります。

• HTTP リクエストの本文 (つまり、値) からコンテンツを読み取ります。
• kvstore モジュールの Propose メソッドを使用して提案を構築します (キーをキー、値を値として持つキーと値のペアを追加します);
• 返すデータがないため、204 StatusNoContent;
でクライアントに応答します。

提案は、raft アルゴリズム ライブラリによって提供される Propose メソッドを通じて raft アルゴリズム ライブラリに送信されます。

提案の内容には、新しいキーと値のペアの追加、既存のキーと値のペアの更新などが含まれます。

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

次に、kvstore モジュールの Propose メソッドを調べて、提案がどのように構築され、処理されるかを見てみましょう。

Propose メソッドでは、最初に gob を使用して書き込まれるキーと値のペアをエンコードし、次にエンコードされたコンテンツを、kvstore モジュールによって構築された提案を raft モジュールに送信する役割を担うチャネルである ProposalC に渡します。

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

kvstore によって構築され、proposalC に渡されたプロポーザルは、raft モジュールのserveChannels メソッドによって受信され、処理されます。

proposalC が閉じられていないことを確認した後、raft モジュールは、raft アルゴリズム ライブラリによって提供される Propose メソッドを使用して処理するために、提案を raft アルゴリズム ライブラリに送信します。

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

提案が送信されると、raft アルゴリズムのプロセスに従います。プロポーザルは最終的にリーダー ノードに転送されます (現在のノードがリーダーではなく、DisableProposalForwarding 構成によって制御され、フォロワーにプロポーザルの転送を許可している場合)。リーダーは、提案をログ エントリとしてラフト ログに追加し、他のフォロワー ノードと同期します。コミットされたとみなされた後、ステート マシンに適用され、結果がユーザーに返されます。

ただし、etcd raft ライブラリ自体はノード間の通信、raft ログへの追加、ステートマシンへの適用などを処理しないため、raft ライブラリはこれらの操作に必要なデータのみを準備します。実際の操作は弊社で行う必要があります。

したがって、このデータを raft ライブラリから受信し、そのタイプに基づいて適切に処理する必要があります。 Ready メソッドは、処理が必要なデータを受信できる読み取り専用チャネルを返します。

受信したデータには、適用されるスナップショット、raft ログに追加されるログエントリ、ネットワーク経由で送信されるメッセージなどの複数のフィールドが含まれることに注意してください。

書き込みリクエストの例 (リーダー ノード) を続けます。対応するデータを受信した後、サーバーのクラッシュによって引き起こされる問題 (例: 複数の候補者に投票するフォロワー) に対処するために、スナップショット、HardState、およびエントリを永続的に保存する必要があります。この論文で説明されているように、HardState と Entries は、すべてのサーバー上で Persistent 状態を構成します。それらを永続的に保存した後、スナップショットを適用して raft ログに追加できます。

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

以上がetcd Raft ライブラリを使用して独自の分散 KV ストレージ システムを構築する方法の詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。