Auto restart db_replicator (#42)

Author: bakwc

README.md

@@ -140,10 +140,11 @@ tables: '*'
 exclude_databases: ['database_10', 'database_*_42']   # optional
 exclude_tables: ['meta_table_*']                      # optional
 
-log_level: 'info'         # optional       
-optimize_interval: 86400  # optional
+log_level: 'info'               # optional       
+optimize_interval: 86400        # optional
+auto_restart_interval: 3600     # optional
 
-indexes:                  # optional
+indexes:                        # optional
   - databases: '*'
     tables: ['test_table']
     index: 'INDEX name_idx name TYPE ngrambf_v1(5, 65536, 4, 0) GRANULARITY 1'
@@ -163,6 +164,7 @@ indexes:                  # optional
 - `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.
+- `auto_restart_interval` - interval (seconds) between automatic db_replicator restart. Default 3600 (1 hour). This is done to reduce memory usage.
 - `indexes` - you may want to add some indexes to accelerate performance, eg. ngram index for full-test search, etc. To apply indexes you need to start replication from scratch.
 
 Few more tables / dbs examples:

mysql_ch_replicator/config.py

@@ -91,6 +91,7 @@ class Settings:
     DEFAULT_LOG_LEVEL = 'info'
     DEFAULT_OPTIMIZE_INTERVAL = 86400
     DEFAULT_CHECK_DB_UPDATED_INTERVAL = 120
+    DEFAULT_AUTO_RESTART_INTERVAL = 3600
 
     def __init__(self):
         self.mysql = MysqlSettings()
@@ -106,6 +107,7 @@ def __init__(self):
         self.optimize_interval = 0
         self.check_db_updated_interval = 0
         self.indexes: list[Index] = []
+        self.auto_restart_interval = 0
 
     def load(self, settings_file):
         data = open(settings_file, 'r').read()
@@ -123,6 +125,9 @@ def load(self, settings_file):
         self.check_db_updated_interval = data.pop(
             'check_db_updated_interval', Settings.DEFAULT_CHECK_DB_UPDATED_INTERVAL,
         )
+        self.auto_restart_interval = data.pop(
+            'auto_restart_interval', Settings.DEFAULT_AUTO_RESTART_INTERVAL,
+        )
         indexes = data.pop('indexes', [])
         for index in indexes:
             self.indexes.append(

mysql_ch_replicator/db_replicator.py

@@ -132,6 +132,7 @@ def __init__(self, config: Settings, database: str, target_database: str = None,
         self.records_to_delete = defaultdict(set)  # table_name => {record_id, ...}
         self.last_records_upload_time = 0
         self.last_touch_time = 0
+        self.start_time = time.time()
 
     def create_state(self):
         return State(os.path.join(self.config.binlog_replicator.data_dir, self.database, 'state.pckl'))
@@ -359,6 +360,12 @@ def run_realtime_replication(self):
         killer = GracefulKiller()
 
         while not killer.kill_now:
+            if self.config.auto_restart_interval:
+                curr_time = time.time()
+                if curr_time - self.start_time >= self.config.auto_restart_interval:
+                    logger.info('process restart (check auto_restart_interval config option)')
+                    break
+
             event = self.data_reader.read_next_event()
             if event is None:
                 time.sleep(DbReplicator.READ_LOG_INTERVAL)