Implementación del algoritmo de consenso RAFT para almacenamiento distribuido de KV en Java

Descripción general de la solución

• Seguidor (seguidor). Acepta solicitudes de lectura del cliente. Toma un latido del líder
• Candidato (candidato). Acepta solicitudes de lectura del cliente. Envía solicitudes de voto a otros nodos
• Líder Acepta solicitudes de lectura y escritura. Envía solicitudes de latidos a otros nodos. Envía datos de solicitudes de anexos a otros nodos.

Almacenamiento de datos

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 cuando se realiza una ronda de votación
• anexe, también conocido como latido del corazón (si no tiene datos), para replicar los datos de registro a los seguidores y evitar el inicio de una nueva ronda de votación.

• votar. Para comenzar una ronda de votación. Cada nodo tiene su propio intervalo, después del cual intentará iniciar una nueva votación. La cuenta regresiva comienza de nuevo cuando recibe un latido del líder.
• latido del corazón. Para enviar una solicitud a los seguidores por parte del líder adjunto. Si el nodo no recibe un latido y el temporizador de votación ha expirado, se convierte en candidato e inicia elecciones, aumenta el número de la ronda de votación y envía solicitudes de votación a otros nodos. Si el nodo recoge la mayoría de los votos, se convierte en el líder y comienza a enviar latidos.

El estado actual del nodo.

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

Votación

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

Enviar una solicitud de votación

//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. Establecer el estado de "Candidato". Eleve el número redondo y vote por nosotros mismos.
2. , ( ). - , , heartbeat .
3. - , . , , -.
4. , . , heartbeat .
5. ¡El nodo votó por nosotros! Aumentamos el número de nodos que emiten votos por nosotros y corregimos que este nodo votó por nosotros.
6. Votados no por nosotros, también creemos.
7. Si se reúne el quórum y el nodo gana la elección, establecemos el estado de "Líder". De lo contrario, nos convertimos en seguidores y esperamos.

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

       );
   }

Procesamiento de solicitud de votación

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. Si la ronda actual del nudo es mayor que la ronda del candidato, nos negamos, porque esta es una solicitud de algún nudo rezagado, que, aparentemente, estuvo fuera del grupo durante algún tiempo y comenzó el proceso de elección porque no vio al líder titular.
2. , , , , , , ; . — .
3. ,
4. . , , , , .
5. Con un resultado positivo, arreglamos el hecho de que el nodo participó en las elecciones y votó por el candidato.
6. Enviar el resultado al candidato

Replicación

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 una solicitud de cambio de datos

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. Metadatos del 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 la solicitud hasta que recibamos una respuesta de todos los seguidores de que la replicación fue exitosa. Dado que se envía una operación por solicitud, puede llevar varias iteraciones sincronizar los registros de seguidores
2. Enviar solicitudes a todos los seguidores y obtener una lista con respuestas
3. Consideramos respuestas solo de seguidores disponibles
4. Si resulta que la ronda de uno de los seguidores es más que la ronda del líder, detenemos todo y nos convertimos en seguidores.
5. Si el seguidor respondió que todo fue exitoso, actualizamos los metadatos del seguidor: guardamos el último índice del registro del seguidor y el índice de la próxima operación esperada por el 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. Si la ronda del líder es menor que la ronda del seguidor, entonces enviamos al líder una ronda y una señal de que su solicitud ha sido rechazada. Tan pronto como el líder reciba una ronda más grande que la suya en respuesta, se convertirá en un seguidor.
2. Si la ronda del líder es más que la ronda del seguidor, establezca esta ronda para el seguidor.
3. Dado que la solicitud fue recibida del líder, independientemente de si hay datos allí o no, restablecemos el temporizador de votación y, si no éramos seguidores, nos convertimos en él.
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. Obtenemos el siguiente índice de la operación aplicada a la base de datos.
2. Contamos cuántos seguidores tienen una operación con dicho índice en sus registros, y no olvidemos agregar un líder
3. Si el número de seguidores es quórum y la operación con dicho índice está en el registro del líder, y la ronda de esta operación es equivalente a la actual, entonces el líder aplica la operación a la base de datos y aumenta el índice de la última operación aplicada. Las operaciones de la ronda anterior no pueden aplicarse, porque otro líder fue responsable de ellas y podría surgir un conflicto. Cada líder aplica operaciones solo de su ronda actual.

Conclusión

Referencias

• Repositorio .
• Especificación
• Breve descripción