Введение

raftexample — это пример, предоставленный etcd, который демонстрирует использование библиотеки алгоритмов консенсуса etcd raft. В конечном итоге raftexample реализует распределенную службу хранения значений ключей, которая предоставляет REST API.

В этой статье будет прочитан и проанализирован код raftexample, в надежде помочь читателям лучше понять, как использовать библиотеку etcd raft и логику реализации библиотеки raft.

Архитектура

Архитектура raftexample очень проста, основные файлы выглядят следующим образом:

• main.go: Отвечает за организацию взаимодействия модуля raft, модуля httpapi и модуля kvstore;
• raft.go: Отвечает за взаимодействие с библиотекой raft, включая отправку предложений, получение сообщений RPC, которые необходимо отправить, выполнение сетевой передачи и т. д.;
• httpapi.go: Отвечает за предоставление REST API, служащего точкой входа для пользовательских запросов;
• kvstore.go: Отвечает за постоянное хранение зафиксированных записей журнала, что эквивалентно конечному автомату в протоколе raft.

Поток обработки запроса на запись

Запрос на запись поступает в метод ServeHTTP модуля httpapi через HTTP-запрос PUT.

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

После сопоставления метода HTTP-запроса через переключатель он входит в поток обработки метода PUT:

• Прочитайте содержимое тела HTTP-запроса (т. е. значение);
• Создайте предложение с помощью метода Propose модуля kvstore (добавив пару ключ-значение с ключом в качестве ключа и значением в качестве значения);
• Поскольку данных для возврата нет, ответьте клиенту с помощью 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)

Далее давайте рассмотрим метод Propose модуля kvstore, чтобы увидеть, как создается и обрабатывается предложение.

В методе Propose мы сначала кодируем пару ключ-значение для записи с помощью gob, а затем передаем закодированный контент в OfferC, канал, ответственный за передачу предложений, созданных модулем kvstore, в модуль 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 



Предложение, созданное kvstore и переданное в OfferC, принимается и обрабатывается методом serveChannels в модуле raft.

После подтверждения того, что OfferC не был закрыт, модуль raft отправляет предложение в библиотеку алгоритмов raft для обработки с использованием метода Propose, предоставленного библиотекой алгоритмов raft.



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



После подачи предложения оно следует алгоритму плота. Предложение в конечном итоге будет перенаправлено узлу-лидеру (если текущий узел не является лидером и вы разрешаете последователям пересылать предложения, что контролируется конфигурацией DisableProposalForwarding). Лидер добавит предложение в виде записи в свой основной журнал и синхронизирует его с другими ведомыми узлами. После того, как оно считается зафиксированным, оно будет применено к конечному автомату, и результат будет возвращен пользователю.

Однако, поскольку библиотека etcd raft сама по себе не занимается связью между узлами, добавлением в журнал raft, применением к конечному автомату и т. д., библиотека raft только подготавливает данные, необходимые для этих операций. Фактические операции должны выполняться нами.

Поэтому нам нужно получить эти данные из библиотеки raft и соответствующим образом обработать их в зависимости от их типа. Метод Ready возвращает канал только для чтения, через который мы можем получать данные, которые необходимо обработать.


Следует отметить, что полученные данные включают в себя несколько полей, таких как снимки, которые нужно применить, записи журнала, которые нужно добавить в плотный журнал, сообщения, которые будут передаваться по сети и т. д.


Продолжая наш пример запроса на запись (ведущий узел), после получения соответствующих данных нам необходимо постоянно сохранять снимки, HardState и записи для решения проблем, вызванных сбоями сервера (например, ведомое голосование за нескольких кандидатов). HardState и Entries вместе составляют постоянное состояние на всех серверах, как упоминалось в статье. После их постоянного сохранения мы можем применить снимок и добавить его в журнал плота.

Поскольку в настоящее время мы являемся ведущим узлом, библиотека raft будет возвращать нам сообщения типа MsgApp (соответствующие AppendEntries RPC в статье). Нам нужно отправить эти сообщения последующим узлам. Здесь мы используем rafthttp, предоставленный etcd, для связи узлов и отправляем сообщения узлам-последователям с помощью метода Send.



// raft.go
case rd := 



Далее мы используем методPublishEntries, чтобы применить зафиксированные записи журнала плота к конечному автомату. Как упоминалось ранее, в примере raftexample модуль kvstore действует как конечный автомат. В методе publicEntries мы передаем записи журнала, которые необходимо применить к конечному автомату, в commitC. Подобно более раннему предложениюC, commitC отвечает за передачу записей журнала, которые модуль raft считает зафиксированными, в модуль kvstore для применения в конечном автомате.



// raft.go
rc.commitC 



В методе readCommits модуля kvstore сообщения, считанные из commitC, декодируются для получения исходных пар ключ-значение, которые затем сохраняются в структуре карты в модуле 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, чтобы уведомить библиотеку raft о том, что мы завершили обработку данных, считанных из канала Ready, и готовы к обработке следующего пакета данных.

Ранее на ведущем узле мы отправляли сообщения типа MsgApp на ведомые узлы с помощью метода Send. rafthttp ведомого узла прослушивает соответствующий порт, чтобы получать запросы и возвращать ответы. Будь то запрос, полученный узлом-ведомым, или ответ, полученный узлом-лидером, он будет отправлен в библиотеку raft для обработки с помощью метода Step.


raftNode реализует интерфейс Raft в rafthttp, а метод Process интерфейса Raft вызывается для обработки полученного содержимого запроса (например, сообщений 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