PostgreSQL Logical Replication Load-Balancing Example

example/01-logical-replication/REPLICATION-ORDER.md

@@ -0,0 +1,13 @@
+# Precise Replication Order
+
+| NodeID  | SrcNode | Function          | ProcOn | Command                       |
+| ------- | ------- | ----------------- | ------ | ----------------------------- |
+| Node1   | node1   | node_pub_self     | node   | create pub 'pub-node1-table1' |
+| Node2   | node2   | node_pub_self     | node   | create pub 'pub-node2-table1' |
+|         | node2   | node_sub_2others  | node   | create sub 'sub-node1-table1' |
+|         | node1   | others_sub_2node  | orch   | create sub 'sub-node2-table1' |
+| Node3   | node3   | node_pub_self     | node   | create pub 'pub-node3-table1' |
+|         | node3   | node_sub_2others  | node   | create sub 'sub-node2-table1' |
+|         | node3   | node_sub_2others  | node   | create sub 'sub-node1-table1' |
+|         | node2   | others_sub_2node  | orch   | create sub 'sub-node3-table1' |
+|         | node1   | others_sub_2node  | orch   | create sub 'sub-node3-table1' |

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

@@ -0,0 +1,11 @@
+class_mapping = {
+    'System': 'System',
+    'Network': 'Network',
+    'NetworkTopology': 'NetworkTopology',
+    'NetIPv4': 'NetIPv4',
+    'NetIPv6': 'NetIPv6',
+    'TopologyHost': 'TopologyHost',
+    'Database': 'Database',
+    'Table': 'Table',
+    'Column': 'Column'
+}

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

@@ -0,0 +1,46 @@
+references = {
+    'UpdateNetworkTopology': {
+        'System': {
+            'property_ref': 'System',
+            'children': {
+                'Network': {
+                    'property_ref': 'Network'
+                },
+                'NetworkTopology': {
+                    'property_ref': 'NetworkTopology',
+                    'children': {
+                        'NetIPv4': {
+                            'property_ref': 'NetIPv4'
+                        },
+                        'NetIPv6': {
+                            'property_ref': 'NetIPv6'
+                        },
+                        'TopologyHost': {
+                            'property_ref': 'TopologyHost'
+                        }
+                    }
+                }
+            }
+        }
+    },
+    '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,13 @@
+import_classes = {
+    'service_implementation': [
+        'System',
+        'Network',
+        'NetworkTopology',
+        'TopologyHost',
+        'NetIPv4',
+        'NetIPv6',
+        '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:{}'.format(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(
+                class_mapper=class_mapper,
+                service_data=call_obj
+            )
+            return { "Status": "ok" }
+        return { "Status": "error - objtype not dict()" }
+
+
+server = JSONServer(
+    address=get_current_ip_address(),
+    port=64000
+).server_loop()

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

@@ -0,0 +1,81 @@
+import json
+import logging
+
+from microesb import microesb
+
+logger = logging.getLogger(__name__)
+
+
+class System(microesb.ClassHandler):
+
+    def __init__(self):
+        super().__init__()
+
+    def update_network_topology(self):
+
+        self.json_transform()
+
+        net_config = {}
+        net_config['Network'] = self.Network.json_dict
+        net_config['NetworkTopology'] = self.NetworkTopology.TopologyHost.json_dict
+
+        with open('/tmp/net-config.json', 'w')as fh:
+            fh.write(json.dumps(net_config))
+
+
+class Network(microesb.ClassHandler):
+
+    def __init__(self):
+        super().__init__()
+
+
+class NetworkTopology(microesb.ClassHandler):
+
+    def __init__(self):
+        super().__init__()
+
+
+class NetIPv4(microesb.ClassHandler):
+
+    def __init__(self):
+        super().__init__()
+
+
+class NetIPv6(microesb.ClassHandler):
+
+    def __init__(self):
+        super().__init__()
+
+
+class TopologyHost(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