github.com/120dev/elasticsearch-postgres

PostgreSQL To ElasticSearch in Go

Requirements

Project in Go + SQL to create an ElasticSearch index from PostgreSQL database.

This procedure is for a Linux operating system.

The following dependencies needs to be previously installed :

• PostgreSQL (database)
• ElasticSearch (index)
• Docker (containerizer)
• PG Admin (Sql manager)

Two ways to start listening:

• Via PostGresToES.go: this will save the raw data from the database
• Via PostGresToES_ApiGateway.go: this will query the API in order to retrieve JSON which will be indexed in ES

Nothing changes in terms of triggers

Installation

• Docker : https://docs.docker.com/install/
• PostgreSQL

docker pull postgres
docker run --name abcd -e POSTGRES_PASSWORD=abcd -d postgres

• PG Admin

docker pull thajeztah/pgadmin4
docker run --restart=always -p 5050:5050 thajeztah/pgadmin4

• ElasticSearch

docker pull elasticsearch
docker run --restart=always -d -p 9200:9200 -p 9300:9300 -it -h elasticsearch --name elasticsearch elasticsearch

Configure :

• Create index :

PUT http://localhost:9200/<INDEX_NAME>
{
    "settings" : {
        "index" : {
            "number_of_shards" : 5,
            "number_of_replicas" : 2
        }
    }
}
OR :
curl -XPUT  'http://localhost:9200/<INDEX_NAME>' -d '{"settings" : {"number_of_shards" : 5, "number_of_replicas" : 2}}'

• Create the PostgreSql function :

CREATE OR REPLACE FUNCTION public.notify_event()
    RETURNS trigger
    LANGUAGE 'plpgsql'
    COST 100
    VOLATILE NOT LEAKPROOF
AS $BODY$

    DECLARE
        data json;
        notification json;
        id integer;
    BEGIN

        -- Convert the old or new row to JSON, based on the kind of action.
        -- Action = DELETE?             -> OLD row
        -- Action = INSERT or UPDATE?   -> NEW row
        IF (TG_OP = 'DELETE') THEN
            data = row_to_json(OLD);
            id = OLD.id;
        ELSE
            data = row_to_json(NEW);
            id = NEW.id;
        END IF;

        -- Contruct the notification as a JSON string.
        notification = json_build_object(
                          'table',TG_TABLE_NAME,
                          'action', TG_OP,
                          'id', id,
                          'data', data);

        -- Execute pg_notify(channel, notification)
        PERFORM pg_notify('events',notification::text);

        -- Result is ignored since this is an AFTER trigger
        RETURN NULL;
    END;
$BODY$;

• Create the PostgreSql trigger :

CREATE TRIGGER products_notify_event
    AFTER INSERT OR DELETE OR UPDATE
    ON public.users
    FOR EACH ROW
    EXECUTE PROCEDURE public.notify_event();

• Install Go https://go.dev/doc/install
• Init go.mod go mod init PostGresToES
• Install dependencies go mod tidy
• Setup (auth, indexName) : https://github.com/120dev/ElasticSearch-postgres/blob/master/PostGresToES.go#L20
• if PostGresToES_ApiGateway.go : setup conf.yaml
• Build go build PostGresToES.go && chmod +x ./PostGresToES or go build PostGresToES_ApiGateway.go && chmod +x ./PostGresToES_ApiGateway
• Run Go script

./PostGresToES or ./PostGresToES_ApiGateway

And wait, all events are logged.

UML diagrams

You can render UML diagrams using Mermaid. For example, this will produce a sequence diagram:

sequenceDiagram
PostgreSQL ->> Trigger: Event crud in Json
Trigger ->> Go: Read and set ES _id
Go ->> ES: Post Json to ES