PostgreSQL Logical Replication Load-Balancing Example

example/01-logical-replication/db-node-rpc/class_mapping.py

@@ -0,0 +1,10 @@
+class_mapping = {
+    'System': 'System',
+    'NetworkTopology': 'NetworkTopology',
+    'NetIPv4': 'NetIPv4',
+    'NetIPv6': 'NetIPv6',
+    'HostNode': 'HostNode',
+    'Database': 'Database',
+    'Table': 'Table',
+    'Column': 'Column'
+}

example/01-logical-replication/db-node-rpc/class_reference.py

@@ -0,0 +1,43 @@
+references = {
+    'UpdateNetworkTopology': {
+        'System': {
+            'property_ref': 'System',
+            'children': {
+                'NetworkTopology': {
+                    'property_ref': 'NetworkTopology',
+                    'children': {
+                        'NetIPv4': {
+                            'property_ref': 'NetIPv4'
+                        },
+                        'NetIPv6': {
+                            'property_ref': 'NetIPv6'
+                        },
+                        'HostNode': {
+                            'property_ref': 'HostNode'
+                        }
+                    }
+                }
+            }
+        }
+    },
+    'InitDatabase': {
+        'Database': {
+            'property_ref': 'Database'
+        }
+    },
+    'CreateReplicaTable': {
+        'Database': {
+            'property_ref': 'Database',
+            'children': {
+                'Table': {
+                    'property_ref': 'Table',
+                    'children': {
+                        'Column': {
+                            'property_ref': 'Column'
+                        }
+                    }
+                }
+            }
+        }
+    }
+}

example/01-logical-replication/db-node-rpc/esbconfig.py

@@ -0,0 +1,12 @@
+import_classes = {
+    'service_implementation': [
+        'System',
+        'NetworkTopology',
+        'NetIPv4',
+        'NetIPv6',
+        'HostNode',
+        'Database',
+        'Table',
+        'Column'
+    ]
+}

example/01-logical-replication/db-node-rpc/json-rpc-server.py

@@ -0,0 +1,53 @@
+import sys
+import jsocket
+import logging
+import subprocess
+
+from microesb import microesb
+
+from class_reference import references as class_reference
+from service_properties import service_properties
+from class_mapping import class_mapping
+
+logging.getLogger().addHandler(
+    logging.StreamHandler(sys.stdout)
+)
+
+logging.getLogger().setLevel(
+    logging.DEBUG
+)
+
+
+def get_current_ip_address():
+    cmd_get_ip = 'ip -h addr show dev eth0 | grep inet | cut -d " " -f 6'
+    res = subprocess.run(cmd_get_ip, shell=True, capture_output=True)
+    raw_ip = res.stdout.strip()
+    raw_ip_sep = raw_ip.find(b'/')
+    return raw_ip[:raw_ip_sep]
+
+
+class JSONServer(jsocket.JsonServer):
+
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)
+
+    def _process_message(self, call_obj):
+        if isinstance(call_obj, dict):
+            print(call_obj)
+            class_mapper = microesb.ClassMapper(
+                class_references=class_reference[call_obj['SYSServiceID']],
+                class_mappings=class_mapping,
+                class_properties=service_properties
+            )
+            res = microesb.ServiceExecuter().execute_get_hierarchy(
+                class_mapper=class_mapper,
+                service_data=call_obj
+            )
+            return res[0]['System']['object_instance'].json_dict
+        return { "Status": "NoObject received" }
+
+
+server = JSONServer(
+    address=get_current_ip_address(),
+    port=64000
+).server_loop()

example/01-logical-replication/db-node-rpc/service_implementation.py

@@ -0,0 +1,68 @@
+import abc
+import logging
+import datetime
+
+from microesb import microesb
+
+logger = logging.getLogger(__name__)
+
+
+class System(microesb.ClassHandler):
+
+    def __init__(self):
+        super().__init__()
+
+
+class NetworkTopology(microesb.ClassHandler):
+
+    def __init__(self):
+        super().__init__()
+
+    def update(self):
+        pass
+
+
+class NetIPv4(microesb.ClassHandler):
+
+    def __init__(self):
+        super().__init__()
+
+
+class NetIPv6(microesb.ClassHandler):
+
+    def __init__(self):
+        super().__init__()
+
+
+class HostNode(microesb.MultiClassHandler):
+
+    def __init__(self):
+        super().__init__()
+
+
+class Database(microesb.ClassHandler):
+
+    def __init__(self):
+        super().__init__()
+
+    def init_db(self):
+        pass
+
+    def create_replica_table(self):
+        pass
+
+
+class Table(microesb.ClassHandler):
+
+    def __init__(self):
+        super().__init__()
+
+
+class Column(microesb.MultiClassHandler):
+
+    def __init__(self):
+        super().__init__()
+        self.primary_key = False
+        self.name = None
+        self.type = None
+        self.default = None

example/01-logical-replication/db-node-rpc/service_properties.py

@@ -0,0 +1,149 @@
+service_properties = {
+    'System': {
+        'properties': {
+            'id': {
+                'type': 'str',
+                'default': None,
+                'required': True,
+                'description': 'System id'
+            }
+        },
+        'methods': [
+            'update_network_topology'
+        ]
+    },
+    'NetworkTopology': {
+        'properties': {
+            'type': {
+                'type': 'str',
+                'default': 'un-partitioned',
+                'required': False,
+                'description': 'Network topology type'
+            }
+        }
+    },
+    'NetIPv4': {
+        'properties': {
+            'subnet': {
+                'type': 'str',
+                'default': None,
+                'required': True,
+                'description': 'IPv4 subnet'
+            },
+            'netmask': {
+                'type': 'str',
+                'default': None,
+                'required': True,
+                'description': 'IPv4 netmask'
+            },
+            'netbits': {
+                'type': 'int',
+                'default': None,
+                'required': True,
+                'description': 'IPv4 netmask bits'
+            },
+            'gateway': {
+                'type': 'str',
+                'default': None,
+                'required': True,
+                'description': 'IPv4 gateway address'
+            },
+            'hostaddress': {
+                'type': 'str',
+                'default': None,
+                'required': False,
+                'description': 'IPv4 docker host address'
+            }
+        }
+    },
+    'NetIPv6': {
+        'properties': {
+        }
+    },
+    'HostNode': {
+        'properties': {
+            'name': {
+                'type': 'str',
+                'default': None,
+                'required': True,
+                'description': 'Hostname'
+            },
+            'ipv4': {
+                'type': 'str',
+                'default': None,
+                'required': False,
+                'description': 'IPv4 address'
+            },
+            'ipv6': {
+                'type': 'str',
+                'default': None,
+                'required': False,
+                'description': 'IPv6 address'
+            }
+        }
+    },
+    'Database': {
+        'properties': {
+            'id': {
+                'type': 'str',
+                'default': None,
+                'required': True,
+                'description': 'Database id'
+            }
+        },
+        'methods': [
+            'init_db',
+            'create_replica_table'
+        ]
+    },
+    'Table': {
+        'properties': {
+            'name': {
+                'type': 'str',
+                'default': None,
+                'required': True,
+                'description': 'Table name'
+            },
+            'add_timestamp_cols': {
+                'type': 'bool',
+                'default': True,
+                'required': False,
+                'description': 'Automatically add timestamp columns for insert and update'
+            },
+            'attach_replication_trigger':  {
+                'type': 'bool',
+                'default': True,
+                'required': False,
+                'description': 'Automatically attach replication check trigger (currently only update)'
+            },
+        }
+    },
+    'Column': {
+        'properties': {
+            'primary_key': {
+                'type': 'bool',
+                'default': False,
+                'required': False,
+                'description': 'Column primary key flag'
+            },
+            'name': {
+                'type': 'str',
+                'default': None,
+                'required': True,
+                'description': 'Column id'
+            },
+            'type': {
+                'type': 'str',
+                'default': None,
+                'required': True,
+                'description': 'Column type'
+            },
+            'default': {
+                'type': 'str',
+                'default': None,
+                'required': False,
+                'description': 'Column default value'
+            }
+        }
+    }
+}

example/01-logical-replication/db-node-rpc/start-server.sh

@@ -0,0 +1,2 @@
+#!/bin/sh
+cd /json-rpc-server && python3 ./json-rpc-server.py &

example/01-logical-replication/db-node.dockerfile

@@ -0,0 +1,21 @@
+FROM postgres:18-bookworm
+MAINTAINER Claus Prüfer
+
+RUN apt-get -qq update -y
+RUN apt-get -qq install iproute2 iputils-ping net-tools python3-pip python3-psycopg2 -y
+
+COPY ./packages/jsocket-1.9.5.tar.gz /
+
+RUN pip3 install microesb --break-system-packages
+RUN pip3 install ./jsocket-1.9.5.tar.gz --break-system-packages
+
+RUN mkdir /json-rpc-server
+COPY ./db-node-rpc/*.py /json-rpc-server/
+COPY ./db-node-rpc/*.sh /json-rpc-server/
+
+ENV POSTGRES_USER postgres
+ENV POSTGRES_PASSWORD password
+ENV POSTGRES_DB lb-test
+
+EXPOSE 5432
+EXPOSE 64000

example/01-logical-replication/docker-build.sh

@@ -0,0 +1,4 @@
+#!/bin/sh
+
+# build docker database node
+docker build -t db-node --file ./db-node.dockerfile .

example/01-logical-replication/docker-daemon/daemon.json

@@ -0,0 +1,3 @@
+{
+	"iptables": false
+}

example/01-logical-replication/docker-network.sh

@@ -0,0 +1,3 @@
+#!/bin/sh
+
+docker network create --subnet=172.16.1.0/24 --gateway=172.16.1.254 -o com.docker.network.bridge.enable_ip_masquerade=false -o com.docker.network.bridge.name=dbr0 dbpool-net