Implementando o algoritmo de consenso RAFT para armazenamento KV distribuído em Java

Descrição geral da solução

• Seguidor (seguidor). Aceita solicitações de leitura do cliente. Leva um batimento cardíaco do líder
• Candidato (candidato). Aceita solicitações de leitura do cliente. Envia solicitações de voto para outros nós
• Líder Aceita pedidos de leitura e gravação. Envia solicitações de pulsação para outros nós. Envia anexar dados de solicitações para outros nós.

Armazenamento de dados

public interface OperationsLog {
   void append(Operation operation);
   Operation get(Integer index);
   List<Operation> all();

   Long getTerm(Integer index);
   Integer getLastIndex();
   Long getLastTerm();

   void removeAllFromIndex(int newOperationIndex);
}

public interface Storage {
   List<Entry> all();
   String get(Long key);
   void insert(Long key, String val);
   void update(Long key, String val);
   void delete(Long key);
}

Temporizadores

• votar ao conduzir uma rodada de votação
• anexar, também conhecido como pulsação (se não houver dados), para replicar dados de log para seguidores e impedir o início de uma nova rodada de votação.

• voto. Para iniciar uma rodada de votação. Cada nó tem seu próprio intervalo, após o qual tentará iniciar uma nova votação. A contagem regressiva começa novamente ao receber um batimento cardíaco do líder.
• batimento cardiaco. Para enviar uma solicitação aos seguidores pelo líder do anexo. Se o nó não recebe uma pulsação e o cronômetro de votação expirou, ele se torna um candidato e inicia as eleições, aumenta o número da rodada de votação e envia solicitações de votação para outros nós. Se o nó coletar a maioria dos votos, ele se tornará o líder e começará a enviar batimentos cardíacos.

O estado atual do nó

public interface Context {
   Integer getId(); //    
   State getState();//: , ,  
   Integer getVotedFor(); 
               //          
   Long getCurrentTerm(); //  
   Integer getCommitIndex(); //    
   List<Peer> getPeers(); //      
}

public interface Peer {
   Integer getId(); //  
   Integer getNextIndex(); //  ,    
   Integer getMatchIndex();//   
   Boolean getVoteGranted(); //     
}

Votação

public interface ElectionService {
   void processElection();
   AnswerVoteDTO vote(RequestVoteDTO requestVoteDTO);
} 

Enviando um pedido de votação

//1
context.setState(CANDIDATE); 
Long term = context.incCurrentTerm(); 
context.setVotedFor(context.getId()); 

List<Integer> peersIds = context.getPeers().stream().map(Peer::getId).collect(Collectors.toList());
long voteGrantedCount = 1L;
long voteRevokedCount = 0L;

//2
while (checkCurrentElectionStatus(term)) {
   List<AnswerVoteDTO> answers = getVoteFromAllPeers(term, peersIds);
   peersIds = new ArrayList<>();
   for (AnswerVoteDTO answer : answers) {
       //3
       if (answer.getStatusCode().equals(OK)) {
           //4
           if (answer.getTerm()>context.getCurrentTerm()) {
               context.setTermGreaterThenCurrent(answer.getTerm());
               return;
           }
           if (answer.isVoteGranted()) {
               //5 
               context.getPeer(answer.getId()).setVoteGranted(true);
               voteGrantedCount++;
           } else
               //6 
               voteRevokedCount++;
       } else {
          peersIds.add(answer.getId());
       }
   }
  //7
  if (voteGrantedCount >= context.getQuorum()) {
       winElection(term);
       return;
   } else if (voteRevokedCount >= context.getQuorum()) {
       loseElection(term);
       return;
   } 

1. Defina o status de "Candidato". Aumente o número da rodada e vote em nós mesmos.
2. , ( ). - , , heartbeat .
3. - , . , , -.
4. , . , heartbeat .
5. O nó votou em nós! Aumentamos o número de nós que votam em nós e corrigimos que esse nó votou em nós.
6. Votou não para nós, também acreditamos.
7. Se o quorum for coletado e o nó vencer a eleição, estabelecemos o status de "Líder". Caso contrário, nos tornamos seguidores e esperamos.

  private void winElection(Long term) {
       context.setState(LEADER);
       context.getPeers().forEach(peer ->
               peer.setNextIndex(operationsLog.getLastIndex()+1)

       );
   }

Processamento de solicitação de votação

class RequestVoteDTO {
   private final Long term; //     
   private final Integer candidateId; //  
   private final Integer lastLogIndex; //     
   private final Long lastLogTerm; //       
}

public AnswerVoteDTO vote(RequestVoteDTO dto) {
   
       boolean termCheck;
       //1
       if (dto.getTerm() < context.getCurrentTerm())
           return new AnswerVoteDTO(context.getId(),context.getCurrentTerm(),false);
       else //2
       if (dto.getTerm().equals(context.getCurrentTerm())) {
           termCheck = (context.getVotedFor() == null||
                          context.getVotedFor().equals(dto.getCandidateId()));
       }
       else
       {   //3
           termCheck = true;
             context.setTermGreaterThenCurrent(dto.getTerm());
       }

       //4  
       boolean logCheck = !((operationsLog.getLastTerm() > dto.getLastLogTerm()) ||
               ((operationsLog.getLastTerm().equals(dto.getLastLogTerm())) &&
                       (operationsLog.getLastIndex() > dto.getLastLogIndex())));


       boolean voteGranted = termCheck&&logCheck;

       //5
       if (voteGranted) {
           context.setVotedFor(dto.getCandidateId());
       }
       //6   
       return new AnswerVoteDTO(context.getId(),context.getCurrentTerm(),voteGranted);
   }

1. Se a atual rodada do nó for maior que a do candidato, recusamos, porque este é um pedido de um nó atrasado, que, aparentemente, ficou fora do cluster por algum tempo e iniciou o processo de eleição porque não viu o líder atual.
2. , , , , , , ; . — .
3. ,
4. . , , , , .
5. Com um resultado positivo, corrigimos o fato de que o nó participou das eleições e votamos no candidato.
6. Envie o resultado de volta ao candidato

Replicação

class RequestAppendDTO {
   private final Long term; //   
   private final Integer leaderId; //   

   private final Integer prevLogIndex;//   
   private final Long prevLogTerm;//   
   private final Integer leaderCommit;//      
   private final Operation operation; //
}

public interface ReplicationService {
   void appendRequest();
   AnswerAppendDTO append(RequestAppendDTO requestAppendDTO);
}

Enviar uma solicitação de alteração de dados

private CompletableFuture<AnswerAppendDTO> sendAppendForOnePeer(Integer id) {
   return CompletableFuture.supplyAsync(() -> {
       try {
           //1
           Peer peer = context.getPeer(id);

           Operation operation;
           Integer prevIndex;
           //2    
           if (peer.getNextIndex() <= operationsLog.getLastIndex()) {
               operation = operationsLog.get(peer.getNextIndex());
               prevIndex = peer.getNextIndex() - 1;
           } else 
           //3  
           {
               operation = null;
               prevIndex = operationsLog.getLastIndex();
           }


           RequestAppendDTO requestAppendDTO = new RequestAppendDTO(
                   context.getCurrentTerm(), //   
                   context.getId(), //  
                   prevIndex,//      
                   operationsLog.getTerm(prevIndex),//  
                   context.getCommitIndex(),
                               //      
                   Operation //
           );

...
/*   http     */
}

1. Metadados do seguidor
2. , . ( ), , , , . , , , ,
3. , , ; , , ,

public void appendRequest() {
       List<Integer> peersIds = context.getPeers().stream().map(Peer::getId).collect(Collectors.toList());

       //1 
       while (peersIds.size() > 0) {
           //2 
           List<AnswerAppendDTO> answers = sendAppendToAllPeers(peersIds);
           peersIds = new ArrayList<>();
           for (AnswerAppendDTO answer : answers) {
               //3
               if (answer.getStatusCode().equals(OK)) {
                   //4
                   if (answer.getTerm() > context.getCurrentTerm()) {
                        context.setTermGreaterThenCurrent(answer.getTerm());
                       return;
                   }
                   Peer peer = context.getPeer(answer.getId());
                   //5     
                   if (answer.getSuccess()) {                      
                       peer.setNextIndex(answer.getMatchIndex() + 1);
                       peer.setMatchIndex(answer.getMatchIndex());
                       if (peer.getNextIndex() <= operationsLog.getLastIndex())
                           peersIds.add(answer.getId());
                   //6      
                   } else {
                       peer.decNextIndex();
                       peersIds.add(answer.getId());
                   }
               }
           }
           //7
           tryToCommit();
       }
}

1. Repetimos a solicitação até recebermos uma resposta de todos os seguidores de que a replicação foi bem-sucedida. Como uma operação é enviada por solicitação, pode levar várias iterações para sincronizar os logs dos seguidores
2. Envie solicitações para todos os seguidores e obtenha uma lista com respostas
3. Consideramos respostas apenas de seguidores disponíveis
4. Se a rodada de um dos seguidores for maior que a do líder, paramos tudo e nos tornamos seguidores
5. Se o seguidor responder que tudo foi bem-sucedido, atualizamos os metadados do seguidor: salvamos o último índice do log do seguidor e o índice da próxima operação esperada pelo seguidor.
6. , , , , . , , . , . , .
7. , . .

public AnswerAppendDTO append(RequestAppendDTO dto) {
     
       //1     
       if (dto.getTerm() < context.getCurrentTerm()) {
           return new AnswerAppendDTO(context.getId(),context.getCurrentTerm(),false, null);
       } else if (dto.getTerm() > context.getCurrentTerm()) {
           //2 
           context.setCurrentTerm(dto.getTerm());
           context.setVotedFor(null);
       }
       //3  
       applicationEventPublisher.publishEvent(new ResetElectionTimerEvent(this));

       if (!context.getState().equals(FOLLOWER)) {
           context.setState(FOLLOWER);
       }
        
       //4  
       if ((dto.getPrevLogIndex() > operationsLog.getLastIndex()) ||                                                                                        !dto.getPrevLogTerm().equals(operationsLog.getTerm(dto.getPrevLogIndex()))) {
                      return new AnswerAppendDTO(context.getId(), context.getCurrentTerm(), false, null);
       }


       Operation newOperation = dto.getOperation();
       if (newOperation != null) {
           int newOperationIndex = dto.getPrevLogIndex() + 1;
           
         synchronized (this) {
               //5
               if ((newOperationIndex <= operationsLog.getLastIndex()) &&
                      (!newOperation.getTerm().equals(operationsLog.getTerm(newOperationIndex)))){
                   operationsLog.removeAllFromIndex(newOperationIndex);
               }
               //6
               if (newOperationIndex <= operationsLog.getLastIndex())
               {
                 return new AnswerAppendDTO(context.getId(), context.getCurrentTerm(), true,      operationsLog.getLastIndex());
               }
               //7
               operationsLog.append(newOperation);
           }
        }
        //8 
        if (dto.getLeaderCommit() > context.getCommitIndex()) {
           context.setCommitIndex(Math.min(dto.getLeaderCommit(), operationsLog.getLastIndex()));
       }

                 
       return new AnswerAppendDTO(context.getId(), context.getCurrentTerm(), true, operationsLog.getLastIndex());
   }

1. Se a rodada do líder for menor que a do seguidor, enviaremos uma rodada ao líder e um sinal de que sua solicitação foi rejeitada. Assim que o líder receber uma rodada maior que a dele em resposta, ele se tornará um seguidor
2. Se a rodada do líder for maior que a do seguidor, defina-a para o seguidor.
3. Como a solicitação foi recebida do líder, independentemente de haver ou não dados, redefinimos o cronômetro de votação e, se não fossemos um seguidor, o tornamos
4. , , , , , , . , ,
5. , . . , , - , , , , . , .
6. , . ,
7. ,
8. , , , .

  private void tryToCommit() {
       while (true) {
           //1
           int N = context.getCommitIndex() + 1;
           //2
           Supplier<Long> count = () ->
               context.getPeers().stream().map(Peer::getMatchIndex).
                       filter(matchIndex -> matchIndex >= N).count() + 1;

           //3 
           if (operationsLog.getLastIndex() >= N &&
                   operationsLog.getTerm(N).equals(context.getCurrentTerm())&&
                      count.get()>=context.getQuorum()
           )
           {
               context.setCommitIndex(N);
           } else
               return;
       }
   }

1. Obtemos o seguinte índice da operação aplicada ao banco de dados
2. Contamos quantos seguidores têm uma operação com esse índice em seus logs e não esquecemos de adicionar um líder
3. Se o número de seguidores é quorum e a operação com esse índice está no log do líder, e a rodada dessa operação é equivalente à atual, o líder aplica a operação ao banco de dados e aumenta o índice da última operação aplicada. As operações da rodada anterior não podem ser aplicadas, porque outro líder foi responsável por elas e um conflito pode surgir. Cada líder aplica operações apenas de sua rodada atual.

Conclusão

Referências

• Repositório .
• Especificação
• Pequena descrição