Auto restart db_replicator

Author: bakwc

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)