導入

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 アルゴリズム ライブラリに送信されます。

プロポーザルの内容には、新しい Key-Value ペアの追加、既存の Key-Value ペアの更新などが含まれます。

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



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

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



// raft.go
select {
    case prop, ok := 



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

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

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


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


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

現在リーダー ノードであるため、raft ライブラリは MsgApp タイプのメッセージを返します (論文の AppendEntries RPC に対応)。これらのメッセージをフォロワー ノードに送信する必要があります。ここでは、ノード通信に etcd によって提供される rafthttp を使用し、Send メソッドを使用してメッセージをフォロワー ノードに送信します。



// raft.go
case rd := 



次に、publishEntries メソッドを使用して、コミットされた raft ログ エントリをステート マシンに適用します。前述したように、raftexample では、kvstore モジュールがステート マシンとして機能します。 publishEntries メソッドでは、ステート マシンに適用する必要があるログ エントリを commitC に渡します。以前のproposalCと同様に、commitCは、raftモジュールがアプリケーションのためにkvstoreモジュールにコミットしたとみなしたログエントリをステートマシンに送信する責任を負います。



// raft.go
rc.commitC 



kvstore モジュールの readCommits メソッドでは、commitC から読み取られたメッセージが gob デコードされて、元のキーと値のペアが取得され、kvstore モジュール内のマップ構造に格納されます。



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




raft モジュールに戻り、Advance メソッドを使用して、Ready チャネルから読み取ったデータの処理が完了し、次のデータ バッチを処理する準備ができていることを raft ライブラリに通知します。

先ほど、リーダー ノードで、Send メソッドを使用して MsgApp タイプのメッセージをフォロワー ノードに送信しました。フォロワー ノードの rafthttp は、対応するポートをリッスンしてリクエストを受信し、応答を返します。フォロワー ノードが受信したリクエストであっても、リーダー ノードが受信したレスポンスであっても、Step メソッドを通じて処理するために raft ライブラリに送信されます。


raftNode は rafthttp に Raft インターフェイスを実装し、Raft インターフェイスの Process メソッドを呼び出して、受信したリクエストの内容 (MsgApp メッセージなど) を処理します。




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




上記は、raftexample での書き込みリクエストの完全な処理フローを説明しています。


  
  
  まとめ


この記事の内容は以上です。 raftexample の構造の概要を説明し、書き込みリクエストの処理フローを詳しく説明することで、etcd raft ライブラリを使用して独自の分散 KV ストレージ サービスを構築する方法をより深く理解できるようにしたいと考えています。

間違いや問題がある場合は、お気軽にコメントまたは直接メッセージを送ってください。ありがとう。


  
  
  参考文献

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

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

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