avg speed in buckets/items/rows

Author: sunsingerus

main.py

@@ -29,7 +29,7 @@ def __init__(self):
         )
         super().__init__(pidfile=self.config.pid_file())
         logging.debug(pprint.pformat(self.config.config))
-        mp.set_start_method('forkserver')
+#        mp.set_start_method('forkserver')
 
     def run(self):
         pumper = Pumper(

src/pool/bbpool.py

@@ -32,11 +32,11 @@ class BBPool(Pool):
     }
 
     buckets_count = 0
-    buckets_content_count = 0;
+    items_count = 0;
 
     prev_time = None
     prev_buckets_count = 0
-    prev_buckets_content_count = 0;
+    prev_items_count = 0;
 
     def __init__(
             self,
@@ -133,18 +133,18 @@ def rotate_belt(self, belt_index, flush=False):
             last_bucket_size = len(self.belts[belt_index][buckets_num-1])
 
             self.buckets_count += 1
-            self.buckets_content_count += last_bucket_size
+            self.items_count += last_bucket_size
 
             logging.info('rot now:%d bktcnt:%d bktcontentcnt: %d index:%s reason:%s bktsonbelt:%d bktsize:%d beltnum:%d',
-                now,
-                self.buckets_count,
-                self.buckets_content_count,
-                str(belt_index),
-                rotate_reason,
-                buckets_num,
-                last_bucket_size,
-                len(self.belts)
-            )
+                         now,
+                         self.buckets_count,
+                         self.items_count,
+                         str(belt_index),
+                         rotate_reason,
+                         buckets_num,
+                         last_bucket_size,
+                         len(self.belts)
+                         )
 
             # time to flush data for specified key
             self.writer_builder.param('csv_file_path_suffix_parts', [str(now), str(self.buckets_count)])
@@ -160,17 +160,17 @@ def rotate_belt(self, belt_index, flush=False):
             # can calculate belt speed
             window_size = now - self.prev_time
             buckets_per_sec = (self.buckets_count - self.prev_buckets_count)/window_size
-            buckets_content_per_sec = (self.buckets_content_count - self.prev_buckets_content_count)/window_size
+            items_per_sec = (self.items_count - self.prev_items_count) / window_size
             logging.info(
-                'buckets per sec:%f buckets content per sec:%f for last %d sec',
+                'buckets_per_sec:%f items_per_sec:%f for last %d sec',
                  buckets_per_sec,
-                 buckets_content_per_sec,
+                 items_per_sec,
                  window_size
             )
 
         self.prev_time = now
         self.prev_buckets_count = self.buckets_count
-        self.prev_buckets_content_count = self.buckets_content_count
+        self.prev_items_count = self.items_count
 
         # belt rotated
         return True

src/reader/mysqlreader.py

@@ -73,13 +73,17 @@ def read(self):
         start_timestamp = int(time.time())
         # fetch events
         try:
+            prev_stat_time = time.time()
+            rows_num = 0
+
             while True:
-                logging.info('Check events in binlog stream')
+                logging.debug('Check events in binlog stream')
                 for mysql_event in self.binlog_stream:
                     if isinstance(mysql_event, WriteRowsEvent):
                         if self.subscribers('WriteRowsEvent'):
                             self.write_rows_event_num += 1
-                            logging.info('WriteRowsEvent #%d rows: %d', self.write_rows_event_num, len(mysql_event.rows))
+                            logging.debug('WriteRowsEvent #%d rows: %d', self.write_rows_event_num, len(mysql_event.rows))
+                            rows_num += len(mysql_event.rows)
                             event = Event()
                             event.schema = mysql_event.schema
                             event.table = mysql_event.table
@@ -90,8 +94,9 @@ def read(self):
 
                         if self.subscribers('WriteRowsEvent.EachRow'):
                             self.write_rows_event_each_row_num += 1
-                            logging.info('WriteRowsEvent.EachRow #%d', self.write_rows_event_each_row_num)
+                            logging.debug('WriteRowsEvent.EachRow #%d', self.write_rows_event_each_row_num)
                             for row in mysql_event.rows:
+                                rows_num += 1
                                 event = Event()
                                 event.schema = mysql_event.schema
                                 event.table = mysql_event.table
@@ -101,8 +106,21 @@ def read(self):
                         # skip non-insert events
                         pass
 
+                now = time.time()
+                if now > prev_stat_time + 60:
+                    # time to calc stat
+                    window_size = now - prev_stat_time
+                    rows_per_sec = rows_num / window_size
+                    logging.info(
+                        'rows_per_sec:%f for last %f sec',
+                        rows_per_sec,
+                        window_size
+                    )
+                    prev_stat_time = now
+                    rows_num = 0
+
                 if not self.blocking:
-                    break
+                    break # while True
 
                 # blocking
                 if self.nice_pause > 0:

src/writer/csvwriter.py

@@ -73,7 +73,7 @@ def insert(self, event_or_events):
         if len(events) < 1:
             return
 
-        logging.debug('class:%s insert %d rows', __class__, len(events))
+        logging.debug('class:%s insert %d events', __class__, len(events))
 
         if not self.opened():
             self.open()

src/writer/processwriter.py

@@ -24,13 +24,13 @@ def open(self):
         pass
 
     def process(self, event_or_events=None):
-        logging.debug('class:%s starting process', __class__)
+        logging.debug('class:%s process()', __class__)
         writer = self.next_writer_builder.get()
         writer.insert(event_or_events)
         writer.close()
         writer.push()
         writer.destroy()
-        logging.debug('class:%s ending process', __class__)
+        logging.debug('class:%s process() done', __class__)
 
     def insert(self, event_or_events=None):
         # event_or_events = [
@@ -44,11 +44,11 @@ def insert(self, event_or_events=None):
         logging.debug('class:%s insert', __class__)
         process = mp.Process(target=self.process, args=(event_or_events,))
 
-        logging.debug('class:%s insert - starting process', __class__)
+        logging.debug('class:%s insert.process.start()', __class__)
         process.start()
 
         #process.join()
-        logging.debug('class:%s insert - done process', __class__)
+        logging.debug('class:%s insert done', __class__)
         pass
 
     def flush(self):