Introduction

Once I heard about logical replication on Postgres, I started thinking about all the issues I face with data aggregation.

To play around with logical replication and see how it works I created The Collector, an example using a Django App to illustrate how to use it.

The Collector

The main idea of the project is to have as many publishers as you like and one master subscriber which aggregates all the information.

All of the data created uses UUID as primary key, Postgres logical replication still doesn't offer support for conflict resolution. To make things simpler Publishers are only able to create data and not edit.

The image below illustrates how it does work:

Publisher

The publisher has one API endpoint that accepts data as JSON and stores it as JSONB in Postgres. There is no validation or control to add the data, but you can't edit it.

Subscriber

The APIs at the subscriber is protected with authentication, and it has two endpoints:

/data/

At the subscriber, data endpoint is read-only, and it aggregates all of the data added by all the publishers.

/annotation/

In the subscriber, we offer the ability to annotate data matching the UUID of the data generated by the publishers.

Running

In this example I'm going to be using docker to show it, let's first create the networks.

docker network create collector
docker-compose build

Create Publishers

First let's create a small shell script function which will help us setup the containers.

Using scale in docker-compose won't work because docker has a load-balacing feature which will confuse Django on connecting to the right database

# Add the number of publishers
export PUBLISHERS_COUNT=2
fake_scale() {
  for ((i=1; i<=${PUBLISHERS_COUNT}; i++)); do
    docker-compose -p pub$i $@
  done
}

# All these commands will run in all the publishers
fake_scale up -d
fake_scale exec database psql -U postgres -c 'CREATE DATABASE collector'
fake_scale exec django python manage.py migrate
fake_scale exec database psql -U postgres collector -c 'CREATE PUBLICATION pub_data FOR TABLE data_data'

Create Subscriber

# Put subscriber up
docker-compose -p sub -f docker-compose.subscriber.yml up -d

docker-compose -p sub -f docker-compose.subscriber.yml exec database psql -U postgres -c 'CREATE DATABASE collector'
docker-compose -p sub -f docker-compose.subscriber.yml exec subscriber python manage.py migrate

# Subscribe to the publications
for ((i=1; i<=${PUBLISHERS_COUNT}; i++)); do
docker-compose -p sub -f docker-compose.yml -f docker-compose.subscriber.yml exec database psql -U postgres collector -c \
  "CREATE SUBSCRIPTION sub_collector_${i} CONNECTION 'host=pub${i}_database_1 port=5432 dbname=collector' PUBLICATION pub_data"
done

# User admin1234 for the test
docker-compose -p sub -f docker-compose.subscriber.yml exec subscriber python manage.py createsuperuser --username admin --email [email protected]

Testing

# same password as before
curl -d "username=admin" -d "password=admin1234" -X POST  http://localhost:8000/login/
# Grab the token and test the API endpoint
curl http://localhost:8000/data/ -H "Authorization: Token <token>"| python -m json.tool

# Add data
fake_scale exec django curl -H "Content-Type: application/json" -X POST -d '{"data": {"email":"[email protected]","amount":123}}' http://localhost:8000/data/

# Check the results, you should see 2 entries
curl http://localhost:8000/data/ -H "Authorization: Token f319df4c2de377c95b4e62e7625e3a0f8eccd114"| python -m json.tool

Cleaning up

# stop and remove containers
docker-compose -p sub -f docker-compose.subscriber.yml down
fake_scale down