Tambah database mirror status

CHANGES.txt

+0.7 2-7-2021
+------------
+- Tambah mirror_status
+
 0.6 5-6-2021
 0.6 5-6-2021
 ------------
 ------------
 - Bug fixed is_mirror() saat Postgres versi 13
 - Bug fixed is_mirror() saat Postgres versi 13

maintenance.ini

@@ -25,6 +25,12 @@ database_backup_dir = /var/backups/pg
 # Jumlah masing-masing database yang tersimpan
 # Jumlah masing-masing database yang tersimpan
 database_rotate = 5
 database_rotate = 5
 
 
+[mirror_status]
+status_dir = /tmp/mirror-status
+
+# Perintah yang dijalankan ketika ada perubahan status
+after = /usr/local/bin/kabari-admin {status_file} 
+
 [formatter_generic]
 [formatter_generic]
 format = %(asctime)s %(levelname)s %(message)s
 format = %(asctime)s %(levelname)s %(message)s
 
 

mirror_status/__init__.py

+import sys
+import os
+import re
+from argparse import ArgumentParser
+from configparser import ConfigParser
+from subprocess import (
+    PIPE,
+    Popen,
+    )
+from maintenance import is_mirror
+from .structure import (
+    RE_OK,
+    RE_TIMEOUT,
+    RE_NO_ROUTE,
+    RE_REMOVED,
+    RE_INVALID,
+    RE_ANY,
+    )
+
+
+REC_OK = re.compile(RE_OK)
+REC_TIMEOUT = re.compile(RE_TIMEOUT)
+REC_NO_ROUTE = re.compile(RE_NO_ROUTE)
+REC_REMOVED = re.compile(RE_REMOVED)
+REC_INVALID = re.compile(RE_INVALID)
+REC_ANY = re.compile(RE_ANY)
+
+PARSER = [
+    (REC_OK, 'OK'),
+    (REC_TIMEOUT, 'network bermasalah'),
+    (REC_NO_ROUTE, 'network bermasalah'),
+    (REC_REMOVED, 'FATAL, harus dibuat ulang'),
+    (REC_INVALID, 'WARNING, invalid record length'),
+    (REC_ANY, 10)]
+
+
+def parse_log(s):
+    for regex, message in PARSER:
+        match = regex.search(s)
+        if not match:
+            continue
+        if isinstance(message, int):
+            return match.group(message)
+        return message
+
+
+def get_option(argv):
+    parser = ArgumentParser()
+    parser.add_argument('config')
+    return parser.parse_args(argv)
+
+
+def mkdir(name):
+    if not os.path.exists(name):
+        os.mkdir(name)
+
+
+def main(argv=sys.argv[1:]):
+    option = get_option(argv)
+    conf_file = option.config
+    conf = ConfigParser()
+    conf.read(conf_file)
+    status_dir = conf.get('mirror_status', 'status_dir')
+    status_change_cmd = conf.get('mirror_status', 'after')
+    mkdir(status_dir)
+    pg_conf_dir = conf.get('main', 'pg_conf_dir')
+    for version in os.listdir(pg_conf_dir):
+        version_conf_dir = os.path.join(pg_conf_dir, version)
+        for cluster in os.listdir(version_conf_dir):
+            cluster_conf_dir = os.path.join(version_conf_dir, cluster)
+            pg_conf = os.path.join(cluster_conf_dir, 'postgresql.conf')
+            if not is_mirror(pg_conf):
+                continue
+            log_file = '/var/log/postgresql/'\
+                       f'postgresql-{version}-{cluster}.log'
+            c1 = ['grep', '^20', log_file]
+            c2 = ['tail', '-n', '1']
+            p1 = Popen(c1, stdout=PIPE)
+            p2 = Popen(c2, stdin=p1.stdout, stdout=PIPE)
+            new_status, stderr = p2.communicate()
+            new_status = parse_log(new_status.decode('utf8'))
+            if not new_status:
+                continue
+            print(f'{cluster} {new_status}')
+            status_file = f'postgresql-{version}-{cluster}.txt'
+            status_file = os.path.join(status_dir, status_file)
+            if os.path.exists(status_file):
+                with open(status_file, 'rb') as f:
+                    old_status = f.read()
+                old_status = old_status.decode('utf8')
+            else:
+                old_status = None
+            with open(status_file, 'wb') as f:
+                f.write(new_status.encode('utf8'))
+            if new_status != old_status:
+                d = dict(status_file=status_file)
+                cmd = status_change_cmd.format(**d)
+                print(cmd)
+                os.system(cmd)

mirror_status/structure.py

+RE_PREFIX = r'([\d]*)-([\d]*)-([\d]*) ([\d]*):([\d]*):([\d]*).([\d]*) '\
+            r'([A-Z]*) .([\d]*). '
+
+# 2022-01-03 07:04:42.728 WIB [1268975] LOG:  started streaming WAL from
+# primary at 3/B6000000 on timeline 1
+RE_OK = RE_PREFIX + 'LOG:  started streaming WAL'
+
+# 2022-01-03 07:04:10.757 WIB [1258171] FATAL:  terminating walreceiver due to
+# timeout
+RE_TIMEOUT = RE_PREFIX + 'FATAL:  terminating walreceiver due to timeout'
+
+# 2022-01-06 08:55:58.797 WIB [1983436] FATAL:  could not connect to the
+# primary server: connection to server at "10.8.18.1", port 5432 failed:
+# Koneksi kehabisan waktu
+RE_NO_ROUTE = RE_PREFIX + 'FATAL:  could not connect to'
+
+# 2022-01-03 07:27:16.064 WIB [22592] FATAL:  could not receive data from WAL
+# stream: ERROR:  requested WAL segment 00000001000000A5000000A9 has already
+# been removed
+RE_REMOVED = RE_PREFIX + '(.*) removed'
+
+# 2022-01-05 04:22:58.394 WIB [1237] LOG:  invalid record length
+RE_INVALID = RE_PREFIX + 'LOG:  invalid'
+
+RE_ANY = RE_PREFIX + 'FATAL: (.*)'

setup.py

@@ -45,6 +45,7 @@ setup(
             'log_table_trigger = log_table.set_trigger:main',
             'log_table_trigger = log_table.set_trigger:main',
             'log2db_init = log2db:init',
             'log2db_init = log2db:init',
             'log2db = log2db:main',
             'log2db = log2db:main',
+            'mirror_status = mirror_status:main',
         ]
         ]
     },
     },
 )
 )