Un schéma assez typique lors du développement d'un système, lorsque la logique de traitement principale est concentrée dans l'application (dans notre cas, Erlang), et que les données de cette application (paramètres, profils utilisateur, etc.) se trouvent dans la base de données (PostgreSQL). L'application Erlang met en cache les paramètres dans ETS pour accélérer le traitement et réduire la charge sur la base de données en rejetant les requêtes persistantes. Dans ce cas, la modification de ces données se produit via un service distinct (éventuellement externe).

Dans de telles situations, le défi se pose de maintenir à jour les données mises en cache. Il existe différentes approches pour résoudre ce problème. L'un d'eux est la réplication logique PostgreSQL. À ce sujet et sera discuté ci-dessous.

Protocole de réplication logique de flux

La réplication logique utilise le protocole de réplication en continu PostgreSQL pour recevoir les modifications de données dans les tables PostgreSQL en lisant les journaux WAL, en filtrant les tables dont vous avez besoin et en envoyant ces modifications à l'abonné. Ce mécanisme est similaire à celui utilisé pour la réplication physique pour créer une base de données de secours.

La réplication logique offre les avantages suivants:

recevoir les modifications sans délai en temps réel;
filtrage des modifications par tables et opérations (INSERT / DELETE / UPDATE);
exhaustivité et intégrité des données reçues par l'abonné. L'abonné reçoit les modifications dans le même ordre que celles survenues dans la base de données;
pas de perte de données en cas d'arrêt temporaire de l'abonné. PostgreSQL se souvient où la réplication s'est arrêtée;

Préparation de la base de données

Pour travailler avec la réplication logique, vous avez besoin d'un plugin qui décode les enregistrements WAL du serveur dans un format plus pratique.
Avant PostgreSQL 10, vous pouvez utiliser le plugin / extension pglogical_output .
À partir du plugin pgoutput PostgreSQL 10.
Cet article couvrira le plugin pgoutput.

Côté PostgreSQL, vous devez effectuer les étapes suivantes:

Définissez les paramètres pour prendre en charge la réplication logique dans
postgresql.conf
```
wal_level = 'logical' max_replication_slots = 5 max_wal_senders = 5 
```
Créez un rôle à utiliser pour la réplication. Le rôle doit avoir l' SUPERUSER REPLICATION ou SUPERUSER .
```
 CREATE USER epgl_test WITH REPLICATION PASSWORD 'epgl_test'; 
```
Autoriser l'accès à ce rôle dans pg_hba.conf avec database = replication
```
 host replication epgl_test 127.0.0.1/32 trust 
```

Créez une publication . Lors de la création d'une publication, nous indiquons les tableaux que nous prévoyons de recevoir dans l'application Erlang

 CREATE PUBLICATION epgl_test FOR TABLE public.test_table1, public.test_table3; ALTER PUBLICATION epgl_test ADD TABLE public.test_table2; --

Partie Erlang

Il n'y a pas si longtemps, la prise en charge du protocole de réplication en continu a été ajoutée à la populaire bibliothèque Erlang pour travailler avec PostgreSQL EPGSQL . Sur la base de cette bibliothèque, nous allons construire la logique de réception des modifications dans Erlang.
Étant donné que le format des données directement dans le message XlogData du protocole dépend du plug-in utilisé pour l'emplacement de réplication, la bibliothèque EPGSQL décode pas les données, mais appelle la méthode de rappel ou envoie le message au processus de manière asynchrone.

Connexion DB

Une connexion de réplication spéciale à la base de données doit être créée, pour cela, vous devez passer l'indicateur de replication .
Dans une connexion à une base de données de réplication, seules les commandes de réplication peuvent être exécutées (par exemple, DROP_REPLICATION_SLOT, CREATE_REPLICATION_SLOT).
Vous ne pouvez pas exécuter une demande régulière via cette connexion.

Créer un emplacement de réplication

L'emplacement de réplication est utilisé pour suivre la position actuelle du journal WAL transféré.
Lors de la création d'un emplacement de réplication, un plug-in de décodage est spécifié.

PostgreSQL 10 introduit la possibilité de créer des emplacements de réplication temporaires qui sont automatiquement supprimés lorsque la connexion de réplication est fermée.

Si l'application lit l'état initial des tables à chaque démarrage, je recommande d'utiliser des emplacements de réplication temporaires, auquel cas vous n'aurez pas à vous soucier de supprimer les emplacements de réplication créés (DROP_REPLICATION_SLOT). La suppression des emplacements de réplication anciens / inutilisés est extrêmement importante car PostgreSQL ne supprime pas les journaux WAL jusqu'à ce que les abonnés de tous les emplacements de réplication reçoivent la modification. S'il reste un emplacement de réplication inactif, les journaux WAL commenceront à s'accumuler et tôt ou tard le système de fichiers débordera.

Obtention de l'état initial des tables

Lors de la création d'un emplacement de réplication (voir l'étape précédente), un instantané est automatiquement créé qui montre l'état de la base de données au moment où l'emplacement a été créé. Cet instantané peut être utilisé pour charger l'état initial des tables, qui était au début de la réplication.

L'instantané n'est disponible que jusqu'à la CREATE_REPLICATION_SLOT la connexion de réplication dans laquelle la commande CREATE_REPLICATION_SLOT été exécutée.

Pour charger les données initiales, une nouvelle connexion régulière / sans réplication à la base de données doit être créée, car SELECT ne peut pas être effectué sur la connexion de réplication. Dans cette connexion, définissez le snapshot SET TRANSACTION SNAPSHOT SnapshotName et extrayez les données nécessaires.

Lancer la réplication

Nous commençons la réplication pour l'emplacement de réplication créé. Lors du démarrage de la réplication, nous transmettons des paramètres supplémentaires pour le plugin, pour pgoutput c'est le nom de la publication créée.

Toutes les étapes ensemble

 start_replication() -> %%    {ok, ReplConn} = epgsql:connect(Host, User, Password, [{database, DBName}, {port, Port}, {replication, "database"}]), %%    {ok, _, [{_, _, SnapshotName}|_]} = epgsql:squery(ReplConn, "CREATE_REPLICATION_SLOT epgl_repl_slot TEMPORARY LOGICAL pgoutput"). %%     {ok, NormalConn} = epgsql:connect(Host, User, Password, [{database, DBName}, {port, Port}]), {ok, _, _} = epgsql:squery(NormalConn, "BEGIN TRANSACTION ISOLATION LEVEL REPEATABLE READ"), {ok, _, _} = epgsql:squery(NormalConn, ["SET TRANSACTION SNAPSHOT '", SnapshotName, "'"]), %% select/load data epgsql:equery(NormalConn,... epgsql:close(NormalConn), %%   ReplSlot = "epgl_repl_slot", Callback = ?MODULE, CbInitState = #{}, WALPosition = "0/0", PluginOpts = "proto_version '1', publication_names '\"epgl_test\"'", ok = epgsql:start_replication(ReplConn, ReplSlot, Callback, CbInitState, WALPosition, PluginOpts). handle_x_log_data(StartLSN, EndLSN, Data, CbState) -> io:format("~p~n", [{StartLSN, EndLSN, Data}]), {ok, EndLSN, EndLSN, CbState}.

Il existe deux options pour interagir avec la bibliothèque EPGSQL :

Synchrone. Le nom du module est transmis en tant que rappel. La bibliothèque des données reçues appellera la fonction CallbackModule:handle_x_log_data . La fonction doit renvoyer LastFlushedLSN, LastAppliedLSN, qui est envoyé dans la réponse PostgreSQL pour suivre la position actuelle de l'emplacement de réplication. Dans nos projets, nous utilisons uniquement cette option;
Asynchrone. Le rappel est le pid du processus, qui recevra les messages de la forme {epgsql, self(), {x_log_data, StartLSN, EndLSN, WALRecord}} . Après le traitement, le processus doit signaler le LSN traité via un appel epgsql:standby_status_update(Conn, FlushedLSN, AppliedLSN) ;

Au lieu d'une conclusion

De plus, pour utiliser l'approche décrite, il est nécessaire d'implémenter le décodage des messages du format de plug-in de slot de réplication dans des structures plus familières à Erlang. Ou utilisez la bibliothèque avec GitHub , qui implémente le décodage pour deux plug-ins et simplifie l'exécution des commandes de réplication.

Réplication logique de PostgreSQL vers Erlang