Additional logging

Author: bakwc

mysql_ch_replicator/db_replicator.py

@@ -124,6 +124,7 @@ def __init__(self, config: Settings, database: str, target_database: str = None,
         self.last_touch_time = 0
 
     def run(self):
+        logger.info('launched db_replicator')
         if self.state.status == Status.RUNNING_REALTIME_REPLICATION:
             self.run_realtime_replication()
             return
@@ -226,6 +227,9 @@ def perform_initial_replication_table(self, table_name):
         primary_key_index = field_names.index(primary_key)
         primary_key_type = field_types[primary_key_index]
 
+        stats_number_of_records = 0
+        last_stats_dump_time = time.time()
+
         while True:
 
             query_start_value = max_primary_key
@@ -258,6 +262,14 @@ def perform_initial_replication_table(self, table_name):
             self.save_state_if_required()
             self.prevent_binlog_removal()
 
+            stats_number_of_records += len(records)
+            curr_time = time.time()
+            if curr_time - last_stats_dump_time >= 60.0:
+                last_stats_dump_time = curr_time
+                logger.info(
+                    f'replicating {table_name}, replicated {stats_number_of_records}, primary key: {max_primary_key}',
+                )
+
     def run_realtime_replication(self):
         if self.initial_only:
             logger.info('skip running realtime replication, only initial replication was requested')
@@ -277,6 +289,7 @@ def run_realtime_replication(self):
             if event is None:
                 time.sleep(DbReplicator.READ_LOG_INTERVAL)
                 self.upload_records_if_required(table_name=None)
+                self.log_stats_if_required()
                 continue
             assert event.db_name == self.database
             if self.database != self.target_database: