Implémentation pratique de l'algorithme Raft avec Typescript | de Radovan Stevanovic

Implémentation pratique de l’algorithme Raft avec Typescript | de Radovan Stevanovic | janvier 2023

Un outil puissant pour parvenir à un consensus dans un système distribué

L’algorithme Raft est un algorithme de consensus distribué conçu pour être facile à comprendre et à mettre en œuvre. Il est couramment utilisé dans les systèmes distribués, tels que les bases de données distribuées et les systèmes de fichiers, pour assurer la cohérence et la disponibilité des données.

Certaines applications réelles de l’algorithme Raft incluent etcdCockroachDB et TiDB.

Tout d’abord, configurons un nouveau projet Typescript à l’aide de Yarn. Dans votre terminal, exécutez les commandes suivantes :

yarn init -y
yarn add -D typescript
yarn tsc --init

Cela créera un nouveau package.json fichier, installez le compilateur Typescript en tant que dépendance de développement et créez un tsconfig.json fichier pour votre projet.

Avant de commencer à implémenter l’algorithme Raft, définissons quelques types et énumérations pour représenter les différents composants de l’algorithme.

type Term = number;
type LogIndex = number;enum NodeState {
FOLLOWER = "follower",
CANDIDATE = "candidate",
LEADER = "leader",
}
type NodeId = string;
type LogEntry = {
term: Term;
command: any;
};
type Log = LogEntry[];
type StateMachine = {
applyCommand: (command: any) => void;
};
type NodeConfiguration = {
id: NodeId;
stateMachine: StateMachine;
currentTerm: Term;
votedFor: NodeId | null;
log: Log;
};
type RequestVoteRPC = {
term: Term;
candidateId: NodeId;
lastLogIndex: LogIndex;
lastLogTerm: Term;
};
type RequestVoteResponse = {
term: Term;
voteGranted: boolean;
};
type AppendEntriesRPC = {
term: Term;
leaderId: NodeId;
prevLogIndex: LogIndex;
prevLogTerm: Term;
entries: LogEntry[];
leaderCommit: LogIndex;
};
type AppendEntriesResponse = {
term: Term;
success: boolean;
};

Ces types et énumérations représenteront les différents composants de l’algorithme Raft, tels que les termes, les journaux et les nœuds.

Tout d’abord, implémentons le requestVote fonction que les candidats utilisent pour demander des votes à leurs abonnés.

const requestVote = (
configuration: NodeConfiguration,
rpc: RequestVoteRPC
): RequestVoteResponse => {
// If the RPC term is less than the current term, return a response with voteGranted set to false
if (rpc.term < configuration.currentTerm) {
return {
term: configuration.currentTerm,
voteGranted: false,
};
}// If the RPC term is greater than the current term, update the current term and vote for the candidate
if (rpc.term > configuration.currentTerm) {
configuration.currentTerm = rpc.term;
configuration.votedFor = rpc.candidateId;
}
// If the voter has already voted for a candidate in this term, return a response with voteGranted set to false
if (configuration.votedFor && configuration.votedFor !== rpc.candidateId) {
return {
term: configuration.currentTerm,
voteGranted: false,
};
}
// If the candidate's log is not up-to-date, return a response with voteGranted set to false
const lastLogIndex = configuration.log.length - 1;
const lastLogTerm = configuration.log[lastLogIndex].term;
if (lastLogTerm > rpc.lastLogTerm || (lastLogTerm === rpc.lastLogTerm && lastLogIndex > rpc.lastLogIndex)) {
return {
term: configuration.currentTerm,
voteGranted: false,
};
}
// Otherwise, return a response with voteGranted set to true
configuration.votedFor = rpc.candidateId;
return {
term: configuration.currentTerm,
voteGranted: true,
};
};

Cette fonction prend un NodeConfiguration objet et un RequestVoteRPC objet en entrée et renvoie un RequestVoteResponse chose. Il vérifie le mandat RPC, le mandat actuel de l’électeur et le journal de l’électeur pour déterminer s’il convient d’accorder un vote au candidat.

Ensuite, laissez le appendEntries les leaders utilisent la fonction pour répliquer les entrées de journal et gérer l’index de validation.

const appendEntries = (
configuration: NodeConfiguration,
rpc: AppendEntriesRPC
): AppendEntriesResponse => {
// If the RPC term is less than the current term, return a response with success set to false
if (rpc.term < configuration.currentTerm) {
return {
term: configuration.currentTerm,
success: false,
};
}// If the RPC term is greater than the current term, update the current term and set the node's state to follower
if (rpc.term > configuration.currentTerm) {
configuration.currentTerm = rpc.term;
configuration.state = NodeState.FOLLOWER;
}
// If the previous log index and term don't match the node's log, return a response with success set to false
const prevLogIndex = rpc.prevLogIndex;
const prevLogTerm = rpc.prevLogTerm;
if (configuration.log[prevLogIndex]?.term !== prevLogTerm) {
return {
term: configuration.currentTerm,
success: false,
};
}
// Otherwise, append the new entries to the log and return a response with success set to true
configuration.log = [...configuration.log.slice(0, prevLogIndex + 1), ...rpc.entries];
configuration.commitIndex = Math.min(rpc.leaderCommit, configuration.log.length - 1);
return {
term: configuration.currentTerm,
success: true,
};
};

Cette fonction prend un NodeConfiguration objet et un AppendEntriesRPC objet en entrée et renvoie un AppendEntriesResponse chose. Il vérifie le terme RPC, l’index et le terme du journal précédent, ainsi que le journal du nœud pour déterminer s’il convient d’ajouter les nouvelles entrées au journal.

Enfin, implémentons le startElection fonction, que les partisans utilisent pour lancer une nouvelle élection lorsqu’ils n’ont reçu aucune communication d’un dirigeant.

const startElection = (configuration: NodeConfiguration): void => {
// Increment the current term and set the node's state to candidate
configuration.currentTerm++;
configuration.state = NodeState.CANDIDATE;// Reset the votedFor field
configuration.votedFor = null;
// Request votes from other nodes
sendRequestVoteRPC(configuration);
};
const sendRequestVoteRPC = (configuration: NodeConfiguration): void => {
// Implementation omitted for brevity
// Sends a RequestVoteRPC to other nodes in the cluster
};

Cette fonction prend un NodeConfiguration objet en tant qu’entrée et incrémente le terme actuel, définit l’état du nœud sur un candidat et envoie RequestVoteRPCs aux autres nœuds du cluster.

Maintenant que nous avons implémenté les principales fonctions de l’algorithme Raft, ajoutons quelques fonctions supplémentaires pour compléter l’implémentation.

Ajoutons d’abord le handleRequestVoteRPC fonction, qui est appelée lorsqu’un nœud reçoit un RequestVoteRPC.

const handleRequestVoteRPC = (
configuration: NodeConfiguration,
rpc: RequestVoteRPC
): RequestVoteResponse => {
// If the RPC term is less than the current term, return a response with voteGranted set to false
if (rpc.term < configuration.currentTerm) {
return {
term: configuration.currentTerm,
voteGranted: false,
};
}// If the RPC term is greater than the current term, update the current term and set the node's state to follower
if (rpc.term > configuration.currentTerm) {
configuration.currentTerm = rpc.term;
configuration.state = NodeState.FOLLOWER;
}
// If the node is already a leader or a candidate, return a response with voteGranted set to false
if (configuration.state === NodeState.LEADER || configuration.state === NodeState.CANDIDATE) {
return {
term: configuration.currentTerm,
voteGranted: false,
};
}
// If the node has already voted for another candidate in this term, return a response with voteGranted set to false
if (configuration.votedFor && configuration.votedFor !== rpc.candidateId) {
return {
term: configuration.currentTerm,
voteGranted: false,
};
}
// Otherwise, return the result of the requestVote function
return requestVote(configuration, rpc);
};

Ensuite, ajoutons le handleAppendEntriesRPC fonction, qui est appelée lorsqu’un nœud reçoit un AppendEntriesRPC.

const handleAppendEntriesRPC = (
configuration: NodeConfiguration,
rpc: AppendEntriesRPC
): AppendEntriesResponse => {
// If the RPC term is less than the current term, return a response with success set to false
if (rpc.term < configuration.currentTerm) {
return {
term: configuration.currentTerm,
success: false,
};
}// If the RPC term is greater than the current term, update the current term and set the node's state to follower
if (rpc.term > configuration.currentTerm) {
configuration.currentTerm = rpc.term;
configuration.state = NodeState.FOLLOWER;
}
// If the node is a leader, return a response with success set to false
if (configuration.state === NodeState.LEADER) {
return {
term: configuration.currentTerm,
success: false,
};
}
// Otherwise, return the result of the appendEntries function
return appendEntries(configuration, rpc);
};

Cette fonction prend un NodeConfiguration objet et un AppendEntriesRPC objet en entrée et renvoie un AppendEntriesResponse chose. Il vérifie d’abord le terme RPC et l’état du nœud, puis renvoie une réponse avec succès défini sur faux ou appelle le appendEntries fonction pour déterminer le résultat.

Enfin, ajoutons le advanceCommitIndex fonction, qui est appelée lorsqu’un nœud leader reçoit un AppendEntriesResponse d’une majorité d’adeptes.

const advanceCommitIndex = (
configuration: NodeConfiguration,
responses: AppendEntriesResponse[]
): void => {
// Sort the responses by term and index
responses.sort((a, b) => a.term !== b.term ? a.term - b.term : a.index - b.index);// Find the highest index that is included in a majority of responses
const majority = Math.floor(responses.length / 2) + 1;
let commitIndex = 0;
for (let i = 0; i < responses.length; i++) {
if (responses.slice(0, i + 1).filter((r) => r.success).length >= majority) {
commitIndex = responses[i].index;
}
}
// Set the commit index to the highest index that is included in a majority of responses
configuration.commitIndex = commitIndex;
};

Cette fonction prend un NodeConfiguration objet et un tableau de AppendEntriesResponse objets en entrée et met à jour l’index de validation du nœud. Il trie les réponses par terme et indice, puis trouve l’indice le plus élevé inclus dans la plupart des réponses.

Pour compiler le code TypeScript en JavaScript, ajoutez le script suivant au scripts champ dans le package.json dossier:

"scripts": {
"build": "tsc"
}

Pour compiler le code, exécutez la commande suivante :

yarn build

Vous pouvez également utiliser le --watch Drapeau pour recompiler automatiquement le code lorsque des modifications sont apportées :

yarn build --watch

Pour utiliser les fonctions que nous avons implémentées dans les sections précédentes, importez-les dans votre code et transmettez les arguments nécessaires. Par example:

import { NodeConfiguration, NodeState } from './nodeConfiguration';
import { requestVote, handleRequestVoteRPC } from './requestVote';
import { appendEntries, handleAppendEntriesRPC } from './appendEntries';
import { startElection } from './startElection';
import { advanceCommitIndex } from './advanceCommitIndex';// Create a new NodeConfiguration object and set its state to FOLLOWER
const configuration = new NodeConfiguration();
configuration.state = NodeState.FOLLOWER;
// Create a requestVote RPC object
const rpc = {
term: 1,
candidateId: 'node-1',
lastLogIndex: 0,
lastLogTerm: 1,
};
console.log(requestVote(configuration, rpc));
console.log(handleRequestVoteRPC(configuration, rpc));
const entries = [
{
term: 1,
command: 'foo',
},
{
term: 1,
command: 'bar',
},
];
const appendEntriesRPC = {
term: 1,
leaderId: 'node-2',
prevLogIndex: 0,
prevLogTerm: 1,
entries,
leaderCommit: 1,
};
console.log(appendEntries(configuration, appendEntriesRPC));
console.log(handleAppendEntriesRPC(configuration, appendEntriesRPC));
startElection(configuration);
const responses = [
{
term: 1,
index: 2,
success: true,
},
{
term: 1,
index: 2,
success: true,
},
{
term: 1,
index: 1,
success: true,
},
];
advanceCommitIndex(configuration, responses);
/*
Output:
{ term: 1, voteGranted: true }
{ term: 1, voteGranted: true }
{ term: 1, success: true }
{ term: 1, success: true }
null
null
*/

Pour tester les fonctions que nous avons implémentées dans ce projet, nous allons utiliser le jest cadre de test.

Tout d’abord, installez jest en tant que dépendance de développement :

yarn add -D jest

Ensuite, créez un tests répertoire à la racine du projet et créer un nodeConfiguration.test.ts fichier à l’intérieur.

Dans le nodeConfiguration.test.ts fichier, importez le NodeConfiguration classe et écrire un test pour s’assurer qu’un nouveau NodeConfiguration objet est créé avec les valeurs par défaut correctes :

import { NodeConfiguration } from '../nodeConfiguration';test('creates a new NodeConfiguration object with the correct default values', () => {
const configuration = new NodeConfiguration();
expect(configuration.id).toBeDefined();
expect(configuration.currentTerm).toBe(0);
expect(configuration.votedFor).toBeNull();
expect(configuration.log).toEqual([]);
});

Pour exécuter les tests, ajoutez le script suivant au scripts champ dans le package.json dossier:

"scripts": {
"test": "jest"
}

Ensuite, lancez les tests en exécutant la commande suivante :

yarn test

Cela exécutera tous les tests dans le tests répertoire et afficher les résultats.

Ensuite, écrivez des tests pour le requestVote et handleRequestVoteRPC les fonctions. Pour ce faire, créez un nouveau fichier appelé requestVote.test.ts dans le tests répertoire et ajoutez les tests suivants :

import { NodeConfiguration, NodeState } from '../nodeConfiguration';
import { requestVote, handleRequestVoteRPC } from '../requestVote';test('requestVote function returns voteGranted as true if RPC term is equal to current term and the node has not voted for another candidate in this term', () => {
const configuration = new NodeConfiguration();
configuration.state = NodeState.FOLLOWER;
configuration.currentTerm = 1;
const rpc = {
term: 1,
candidateId: 'node-1',
lastLogIndex: 0,
lastLogTerm: 1,
};
expect(requestVote(configuration, rpc)).toEqual({ term: 1, voteGranted: true });
})
test('handleRequestVoteRPC function does not update the node configuration if RPC term is less than current term', () => {
const configuration = new NodeConfiguration();
configuration.state = NodeState.FOLLOWER;
configuration.currentTerm = 2;
configuration.votedFor = 'node-2';
const rpc = {
term: 1,
candidateId: 'node-1',
lastLogIndex: 0,
lastLogTerm: 1,
};
handleRequestVoteRPC(configuration, rpc);
expect(configuration.currentTerm).toBe(2);
expect(configuration.votedFor).toBe('node-2');
});
test('handleRequestVoteRPC function does not update the node configuration if the node has already voted for another candidate in the current term', () => {
const configuration = new NodeConfiguration();
configuration.state = NodeState.FOLLOWER;
configuration.currentTerm = 1;
configuration.votedFor = 'node-2';
const rpc = {
term: 1,
candidateId: 'node-1',
lastLogIndex: 0,
lastLogTerm: 1,
};
handleRequestVoteRPC(configuration, rpc);
expect(configuration.currentTerm).toBe(1);
expect(configuration.votedFor).toBe('node-2');
});

Il existe de nombreux autres tests unitaires à ajouter suivant le même schéma.

En conclusion, l’algorithme Raft est un outil puissant pour parvenir à un consensus dans un système distribué. En suivant le protocole Raft, les nœuds du système peuvent s’entendre sur un ensemble cohérent de commandes, même face aux pannes et aux partitions du réseau.

Cet article a fourni une implémentation pratique de l’algorithme Raft dans TypeScript. Bien que cette implémentation ne soit qu’un exemple de la manière dont l’algorithme Raft peut être implémenté, elle sert de référence utile pour comprendre les concepts sous-jacents à l’algorithme et comment ils peuvent être appliqués dans la pratique.

Nous avons également inclus des explications techniques détaillées et des références d’applications réelles tout au long de l’article pour mieux comprendre l’algorithme Raft et son importance dans les systèmes distribués.

Dans l’ensemble, l’algorithme Raft est un élément clé de tout système distribué, et nous espérons que cet article a fourni une ressource utile pour comprendre sa mise en œuvre et son application.

Télécharger ici

A propos de l'auteur