Introdução

raftexample é um exemplo fornecido pelo etcd que demonstra o uso da biblioteca de algoritmos de consenso da jangada etcd. Em última análise, o raftexample implementa um serviço distribuído de armazenamento de valor-chave que fornece uma API REST.

Este artigo irá ler e analisar o código do raftexample, na esperança de ajudar os leitores a entender melhor como usar a biblioteca raft etcd e a lógica de implementação da biblioteca raft.

Arquitetura

A arquitetura do raftexample é muito simples, com os arquivos principais sendo os seguintes:

• main.go: Responsável por organizar a interação entre o módulo raft, o módulo httpapi e o módulo kvstore;
• raft.go: Responsável por interagir com a biblioteca da jangada, incluindo envio de propostas, recebimento de mensagens RPC que precisam ser enviadas e realização de transmissão pela rede, etc.;
• httpapi.go: Responsável por fornecer a API REST, servindo como ponto de entrada para solicitações do usuário;
• kvstore.go: Responsável por armazenar persistentemente entradas de log confirmadas, equivalente à máquina de estado no protocolo raft.

O fluxo de processamento de uma solicitação de gravação

Uma solicitação de gravação chega ao método ServeHTTP do módulo httpapi por meio de uma solicitação HTTP PUT.

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

Depois de combinar o método de solicitação HTTP via switch, ele entra no fluxo de processamento do método PUT:

• Leia o conteúdo do corpo da solicitação HTTP (ou seja, o valor);
• Construa uma proposta através do método Propose do módulo kvstore (adicionando um par chave-valor com chave como chave e valor como valor);
• Como não há dados para retornar, responda ao cliente com 204 StatusNoContent;

A proposta é submetida à biblioteca de algoritmos de jangada por meio do método Propose fornecido pela biblioteca de algoritmos de jangada.

O conteúdo de uma proposta pode ser adicionar um novo par de valores-chave, atualizar um par de valores-chave existente, 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)

A seguir, vamos examinar o método Propose do módulo kvstore para ver como uma proposta é construída e processada.

No método Propose, primeiro codificamos o par chave-valor a ser escrito usando gob, e depois passamos o conteúdo codificado para proporC, canal responsável por transmitir propostas construídas pelo módulo kvstore para o módulo 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 



A proposta construída por kvstore e passada para proporC é recebida e processada pelo método serveChannels no módulo raft.

Depois de confirmar que o proporC não foi fechado, o módulo raft envia a proposta à biblioteca de algoritmos raft para processamento usando o método Propose fornecido pela biblioteca de algoritmos raft.



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



Depois que uma proposta é enviada, ela segue o processo do algoritmo de jangada. A proposta será eventualmente encaminhada para o nó líder (se o nó atual não for o líder e você permitir que os seguidores encaminhem propostas, controlado pela configuração DisableProposalForwarding). O líder adicionará a proposta como uma entrada de log ao seu log de balsa e a sincronizará com outros nós seguidores. Após ser considerado confirmado, será aplicado à máquina de estados e o resultado será devolvido ao usuário.

No entanto, como a própria biblioteca raft do etcd não lida com a comunicação entre nós, anexando ao log da raft, aplicando à máquina de estado, etc., a biblioteca raft apenas prepara os dados necessários para essas operações. As operações reais devem ser realizadas por nós.

Portanto, precisamos receber esses dados da biblioteca de jangadas e processá-los de acordo com seu tipo. O método Ready retorna um canal somente leitura através do qual podemos receber os dados que precisam ser processados.


Deve-se observar que os dados recebidos incluem vários campos, como instantâneos a serem aplicados, entradas de log a serem anexadas ao log da balsa, mensagens a serem transmitidas pela rede, etc.


Continuando com nosso exemplo de solicitação de gravação (nó líder), após receber os dados correspondentes, precisamos salvar persistentemente snapshots, HardState e entradas para lidar com problemas causados ​​por falhas no servidor (por exemplo, um seguidor votando em vários candidatos). HardState e Entries juntos constituem o estado Persistente em todos os servidores, conforme mencionado no artigo. Depois de salvá-los persistentemente, podemos aplicar o instantâneo e anexá-lo ao log da balsa.

Como atualmente somos o nó líder, a biblioteca raft retornará mensagens do tipo MsgApp para nós (correspondendo ao RPC AppendEntries no artigo). Precisamos enviar essas mensagens aos nós seguidores. Aqui, usamos o rafthttp fornecido pelo etcd para comunicação do nó e enviamos as mensagens aos nós seguidores usando o método Send.



// raft.go
case rd := 



Em seguida, usamos o métodopublishEntries para aplicar as entradas de log de raft confirmadas à máquina de estado. Conforme mencionado anteriormente, no exemplo raft, o módulo kvstore atua como a máquina de estado. No métodopublishEntries, passamos as entradas de log que precisam ser aplicadas à máquina de estado para commitC. Semelhante à propostaC anterior, commitC é responsável por transmitir as entradas de log que o módulo raft considerou comprometidas para o módulo kvstore para aplicação na máquina de estado.



// raft.go
rc.commitC 



No método readCommits do módulo kvstore, as mensagens lidas do commitC são decodificadas em gob para recuperar os pares chave-valor originais, que são então armazenados em uma estrutura de mapa dentro do módulo 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)
}




Voltando ao módulo raft, usamos o método Advance para notificar a biblioteca raft que terminamos de processar os dados lidos do canal Ready e estamos prontos para processar o próximo lote de dados.

Anteriormente, no nó líder, enviamos mensagens do tipo MsgApp para os nós seguidores usando o método Send. O rafthttp do nó seguido escuta na porta correspondente para receber solicitações e retornar respostas. Seja uma solicitação recebida por um nó seguidor ou uma resposta recebida por um nó líder, ela será enviada à biblioteca raft para processamento através do método Step.


raftNode implementa a interface Raft em rafthttp, e o método Process da interface Raft é chamado para lidar com o conteúdo da solicitação recebida (como mensagens MsgApp).




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




O texto acima descreve o fluxo completo de processamento de uma solicitação de gravação em raftexample.


  
  
  Resumo


Isso conclui o conteúdo deste artigo. Ao delinear a estrutura do raftexample e detalhar o fluxo de processamento de uma solicitação de gravação, espero ajudá-lo a entender melhor como usar a biblioteca etcd raft para construir seu próprio serviço de armazenamento KV distribuído.

Se houver algum erro ou problema, sinta-se à vontade para comentar ou me enviar uma mensagem diretamente. Obrigado.


  
  
  Referências

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

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

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