Merge branch 'master' into multi_primary_key

Author: bakwc

README.md

@@ -3,7 +3,7 @@
 [![Release][release-image]][releases]
 [![License][license-image]][license]
 
-[release-image]: https://img.shields.io/badge/release-0.0.33-blue.svg?style=flat
+[release-image]: https://img.shields.io/badge/release-0.0.35-blue.svg?style=flat
 [releases]: https://github.com/bakwc/mysql_ch_replicator/releases
 
 [license-image]: https://img.shields.io/badge/license-MIT-blue.svg?style=flat
@@ -137,7 +137,8 @@ tables: '*'
 exclude_databases: ['database_10', 'database_*_42']   # optional
 exclude_tables: ['meta_table_*']                      # optional
 
-log_level: 'info'   # optional             
+log_level: 'info'         # optional       
+optimize_interval: 86400  # optional
 ```
 
 #### Required settings
@@ -152,6 +153,7 @@ log_level: 'info'   # optional
 - `exclude_databases` - databases to __exclude__, string or list, eg `'table1*'` or `['table2', 'table3*']`. If same database matches `databases` and `exclude_databases`, exclude has higher priority.
 - `exclude_tables` - databases to __exclude__, string or list. If same table matches `tables` and `exclude_tables`, exclude has higher priority.
 - `log_level` - log level, default is `info`, you can set to `debug` to get maximum information (allowed values are `debug`, `info`, `warning`, `error`, `critical`)
+- `optimize_interval` - interval (seconds) between automatic `OPTIMIZE table FINAL` calls. Default 86400 (1 day). This is required to perform all merges guaranteed and avoid increasing of used storage and decreasing performance.
 
 Few more tables / dbs examples:
 

mysql_ch_replicator/clickhouse_api.py

@@ -32,7 +32,8 @@
 class ClickhouseApi:
     MAX_RETRIES = 5
     RETRY_INTERVAL = 30
-    def __init__(self, database: str, clickhouse_settings: ClickhouseSettings):
+
+    def __init__(self, database: str | None, clickhouse_settings: ClickhouseSettings):
         self.database = database
         self.clickhouse_settings = clickhouse_settings
         self.client = clickhouse_connect.get_client(
@@ -175,10 +176,12 @@ def drop_database(self, db_name):
     def create_database(self, db_name):
         self.cursor.execute(f'CREATE DATABASE {db_name}')
 
-    def select(self, table_name, where=None):
+    def select(self, table_name, where=None, final=None):
         query = f'SELECT * FROM {table_name}'
         if where:
             query += f' WHERE {where}'
+        if final is not None:
+            query += f' SETTINGS final = {int(final)};'
         result = self.client.query(query)
         rows = result.result_rows
         columns = result.column_names

mysql_ch_replicator/config.py

@@ -81,6 +81,9 @@ def validate(self):
 
 
 class Settings:
+    DEFAULT_LOG_LEVEL = 'info'
+    DEFAULT_OPTIMIZE_INTERVAL = 86400
+    DEFAULT_CHECK_DB_UPDATED_INTERVAL = 120
 
     def __init__(self):
         self.mysql = MysqlSettings()
@@ -93,6 +96,8 @@ def __init__(self):
         self.settings_file = ''
         self.log_level = 'info'
         self.debug_log_level = False
+        self.optimize_interval = 0
+        self.check_db_updated_interval = 0
 
     def load(self, settings_file):
         data = open(settings_file, 'r').read()
@@ -105,7 +110,11 @@ def load(self, settings_file):
         self.tables = data.pop('tables', '*')
         self.exclude_databases = data.pop('exclude_databases', '')
         self.exclude_tables = data.pop('exclude_tables', '')
-        self.log_level = data.pop('log_level', 'info')
+        self.log_level = data.pop('log_level', Settings.DEFAULT_LOG_LEVEL)
+        self.optimize_interval = data.pop('optimize_interval', Settings.DEFAULT_OPTIMIZE_INTERVAL)
+        self.check_db_updated_interval = data.pop(
+            'check_db_updated_interval', Settings.DEFAULT_CHECK_DB_UPDATED_INTERVAL,
+        )
         assert isinstance(self.databases, str) or isinstance(self.databases, list)
         assert isinstance(self.tables, str) or isinstance(self.tables, list)
         self.binlog_replicator = BinlogReplicatorSettings(**data.pop('binlog_replicator'))

mysql_ch_replicator/converter.py

@@ -194,6 +194,8 @@ def convert_type(self, mysql_type, parameters):
             return 'String'
         if 'varbinary' in mysql_type:
             return 'String'
+        if 'binary' in mysql_type:
+            return 'String'
         raise Exception(f'unknown mysql type "{mysql_type}"')
 
     def convert_field_type(self, mysql_type, mysql_parameters):

mysql_ch_replicator/db_optimizer.py

@@ -0,0 +1,106 @@
+import pickle
+import os
+import time
+from logging import getLogger
+
+from .config import Settings
+from .mysql_api import MySQLApi
+from .clickhouse_api import ClickhouseApi
+from .utils import GracefulKiller
+
+
+logger = getLogger(__name__)
+
+
+class State:
+
+    def __init__(self, file_name):
+        self.file_name = file_name
+        self.last_process_time = {}
+        self.load()
+
+    def load(self):
+        file_name = self.file_name
+        if not os.path.exists(file_name):
+            return
+        data = open(file_name, 'rb').read()
+        data = pickle.loads(data)
+        self.last_process_time = data['last_process_time']
+
+    def save(self):
+        file_name = self.file_name
+        data = pickle.dumps({
+            'last_process_time': self.last_process_time,
+        })
+        with open(file_name + '.tmp', 'wb') as f:
+            f.write(data)
+        os.rename(file_name + '.tmp', file_name)
+
+
+class DbOptimizer:
+    def __init__(self, config: Settings):
+        self.state = State(os.path.join(
+            config.binlog_replicator.data_dir,
+            'db_optimizer.bin',
+        ))
+        self.config = config
+        self.mysql_api = MySQLApi(
+            database=None,
+            mysql_settings=config.mysql,
+        )
+        self.clickhouse_api = ClickhouseApi(
+            database=None,
+            clickhouse_settings=config.clickhouse,
+        )
+
+    def select_db_to_optimize(self):
+        databases = self.mysql_api.get_databases()
+        databases = [db for db in databases if self.config.is_database_matches(db)]
+        ch_databases = set(self.clickhouse_api.get_databases())
+
+        for db in databases:
+            if db not in ch_databases:
+                continue
+            last_process_time = self.state.last_process_time.get(db, 0.0)
+            if time.time() - last_process_time < self.config.optimize_interval:
+                continue
+            return db
+        return None
+
+    def optimize_table(self, db_name, table_name):
+        logger.info(f'Optimizing table {db_name}.{table_name}')
+        t1 = time.time()
+        self.clickhouse_api.execute_command(
+            f'OPTIMIZE TABLE {db_name}.{table_name} FINAL SETTINGS mutations_sync = 2'
+        )
+        t2 = time.time()
+        logger.info(f'Optimize finished in {int(t2-t1)} seconds')
+
+    def optimize_database(self, db_name):
+        self.mysql_api.set_database(db_name)
+        tables = self.mysql_api.get_tables()
+        tables = [table for table in tables if self.config.is_table_matches(table)]
+
+        self.clickhouse_api.execute_command(f'USE {db_name}')
+        ch_tables = set(self.clickhouse_api.get_tables())
+
+        for table in tables:
+            if table not in ch_tables:
+                continue
+            self.optimize_table(db_name, table)
+        self.state.last_process_time[db_name] = time.time()
+        self.state.save()
+
+    def run(self):
+        logger.info('running optimizer')
+        killer = GracefulKiller()
+        try:
+            while not killer.kill_now:
+                db_to_optimize = self.select_db_to_optimize()
+                if db_to_optimize is None:
+                    time.sleep(min(120, self.config.optimize_interval))
+                    continue
+                self.optimize_database(db_name=db_to_optimize)
+        except Exception as e:
+            logger.error(f'error {e}', exc_info=True)
+        logger.info('optimizer stopped')

mysql_ch_replicator/main.py

@@ -9,6 +9,7 @@
 from .config import Settings
 from .db_replicator import DbReplicator
 from .binlog_replicator import BinlogReplicator
+from .db_optimizer import DbOptimizer
 from .monitoring import Monitoring
 from .runner import Runner
 
@@ -97,6 +98,24 @@ def run_db_replicator(args, config: Settings):
     db_replicator.run()
 
 
+def run_db_optimizer(args, config: Settings):
+    data_dir = config.binlog_replicator.data_dir
+    if not os.path.exists(data_dir):
+        os.mkdir(data_dir)
+
+    log_file = os.path.join(
+        data_dir,
+        'db_optimizer.log',
+    )
+
+    set_logging_config(f'dbopt {args.db}', log_file=log_file, log_level_str=config.log_level)
+
+    db_optimizer = DbOptimizer(
+        config=config,
+    )
+    db_optimizer.run()
+
+
 def run_monitoring(args, config: Settings):
     set_logging_config('monitor', log_level_str=config.log_level)
     monitoring = Monitoring(args.db or '', config)
@@ -114,7 +133,7 @@ def main():
     parser.add_argument(
         "mode", help="run mode",
         type=str,
-        choices=["run_all", "binlog_replicator", "db_replicator", "monitoring"])
+        choices=["run_all", "binlog_replicator", "db_replicator", "monitoring", "db_optimizer"])
     parser.add_argument("--config", help="config file path", default='config.yaml', type=str)
     parser.add_argument("--db", help="source database(s) name", type=str)
     parser.add_argument("--target_db", help="target database(s) name, if not set will be same as source", type=str)
@@ -131,6 +150,8 @@ def main():
         run_binlog_replicator(args, config)
     if args.mode == 'db_replicator':
         run_db_replicator(args, config)
+    if args.mode == 'db_optimizer':
+        run_db_optimizer(args, config)
     if args.mode == 'monitoring':
         run_monitoring(args, config)
     if args.mode == 'run_all':

mysql_ch_replicator/mysql_api.py

@@ -17,9 +17,9 @@ def __init__(self, database: str, mysql_settings: MysqlSettings):
     def close(self):
         self.db.close()
 
-    def reconnect_if_required(self):
+    def reconnect_if_required(self, force=False):
         curr_time = time.time()
-        if curr_time - self.last_connect_time < MySQLApi.RECONNECT_INTERVAL:
+        if curr_time - self.last_connect_time < MySQLApi.RECONNECT_INTERVAL and not force:
             return
         conn_settings = dict(
             host=self.mysql_settings.host,
@@ -59,7 +59,7 @@ def set_database(self, database):
         self.cursor.execute(f'USE {self.database}')
 
     def get_databases(self):
-        self.reconnect_if_required()
+        self.reconnect_if_required(True)  # New database appear only after new connection
         self.cursor.execute('SHOW DATABASES')
         res = self.cursor.fetchall()
         tables = [x[0] for x in res]

mysql_ch_replicator/runner.py

@@ -25,6 +25,11 @@ def __init__(self, db_name, config_file):
         super().__init__(f'{sys.argv[0]} --config {config_file} --db {db_name} db_replicator')
 
 
+class DbOptimizerRunner(ProcessRunner):
+    def __init__(self, config_file):
+        super().__init__(f'{sys.argv[0]} --config {config_file} db_optimizer')
+
+
 class RunAllRunner(ProcessRunner):
     def __init__(self, db_name, config_file):
         super().__init__(f'{sys.argv[0]} --config {config_file} run_all --db {db_name}')
@@ -35,8 +40,9 @@ def __init__(self, config: Settings, wait_initial_replication: bool, databases:
         self.config = config
         self.databases = databases or config.databases
         self.wait_initial_replication = wait_initial_replication
-        self.runners: dict = {}
+        self.runners: dict[str: DbReplicatorRunner] = {}
         self.binlog_runner = None
+        self.db_optimizer = None
 
     def is_initial_replication_finished(self, db_name):
         state_path = os.path.join(
@@ -52,6 +58,28 @@ def restart_dead_processes(self):
             runner.restart_dead_process_if_required()
         if self.binlog_runner is not None:
             self.binlog_runner.restart_dead_process_if_required()
+        if self.db_optimizer is not None:
+            self.db_optimizer.restart_dead_process_if_required()
+
+    def check_databases_updated(self, mysql_api: MySQLApi):
+        logger.debug('check if databases were created / removed in mysql')
+        databases = mysql_api.get_databases()
+        logger.info(f'mysql databases: {databases}')
+        databases = [db for db in databases if self.config.is_database_matches(db)]
+        logger.info(f'mysql databases filtered: {databases}')
+        for db in databases:
+            if db in self.runners:
+                continue
+            logger.info(f'running replication for {db} (database created in mysql)')
+            runner = self.runners[db] = DbReplicatorRunner(db_name=db, config_file=self.config.settings_file)
+            runner.run()
+
+        for db in self.runners.keys():
+            if db in databases:
+                continue
+            logger.info(f'stop replication for {db} (database removed from mysql)')
+            self.runners[db].stop()
+            self.runners.pop(db)
 
     def run(self):
         mysql_api = MySQLApi(
@@ -65,6 +93,9 @@ def run(self):
         self.binlog_runner = BinlogReplicatorRunner(self.config.settings_file)
         self.binlog_runner.run()
 
+        self.db_optimizer = DbOptimizerRunner(self.config.settings_file)
+        self.db_optimizer.run()
+
         # First - continue replication for DBs that already finished initial replication
         for db in databases:
             if not self.is_initial_replication_finished(db_name=db):
@@ -90,16 +121,24 @@ def run(self):
 
         logger.info('all replicators launched')
 
+        last_check_db_updated = time.time()
         while not killer.kill_now:
             time.sleep(1)
             self.restart_dead_processes()
+            if time.time() - last_check_db_updated > self.config.check_db_updated_interval:
+                self.check_databases_updated(mysql_api=mysql_api)
+                last_check_db_updated = time.time()
 
         logger.info('stopping runner')
 
         if self.binlog_runner is not None:
             logger.info('stopping binlog replication')
             self.binlog_runner.stop()
 
+        if self.db_optimizer is not None:
+            logger.info('stopping db_optimizer')
+            self.db_optimizer.stop()
+
         for db_name, db_replication_runner in self.runners.items():
             logger.info(f'stopping replication for {db_name}')
             db_replication_runner.stop()

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "mysql-ch-replicator"
-version = "0.0.33"
+version = "0.0.35"
 description = "Tool for replication of MySQL databases to ClickHouse"
 authors = ["Filipp Ozinov <[email protected]>"]
 license = "MIT"