min/max and windows size

Author: sunsingerus

src/pool/bbpool.py

@@ -159,14 +159,17 @@ def rotate_belt(self, belt_index, flush=False):
             # have previous time - meaning this is at least second rotate
             # can calculate belt speed
             window_size = now - self.prev_time
-            buckets_per_sec = (self.buckets_count - self.prev_buckets_count)/window_size
-            items_per_sec = (self.items_count - self.prev_items_count) / window_size
-            logging.info(
-                'buckets_per_sec:%f items_per_sec:%f for last %d sec',
-                 buckets_per_sec,
-                 items_per_sec,
-                 window_size
-            )
+            if window_size > 0:
+                buckets_per_sec = (self.buckets_count - self.prev_buckets_count)/window_size
+                items_per_sec = (self.items_count - self.prev_items_count) / window_size
+                logging.info(
+                    'PERF - buckets_per_sec:%f items_per_sec:%f for last %d sec',
+                     buckets_per_sec,
+                     items_per_sec,
+                     window_size
+                )
+            else:
+                logging.info("PERF - buckets window size=0 can not calc performance for this window")
 
         self.prev_time = now
         self.prev_buckets_count = self.buckets_count

src/reader/mysqlreader.py

@@ -69,22 +69,26 @@ def __init__(
             resume_stream=self.resume_stream,
         )
 
-    def performance_report(self, start, rows_num, rows_per_event_min=-1, rows_per_event_max=-1, now=None):
+    def performance_report(self, start, rows_num, rows_num_per_event_min=None, rows_num_per_event_max=None, now=None):
         # time to calc stat
 
         if now is None:
             now = time.time()
 
         window_size = now - start
-        rows_per_sec = rows_num / window_size
-        logging.info(
-            'rows_per_sec:%f rows_per_event_min: %d rows_per_event_max: %d for last %d rows %f sec',
-            rows_per_sec,
-            rows_per_event_min,
-            rows_per_event_max,
-            rows_num,
-            window_size,
-        )
+        if window_size > 0:
+            rows_per_sec = rows_num / window_size
+            logging.info(
+                'PERF - rows_per_sec:%f rows_per_event_min: %d rows_per_event_max: %d for last %d rows %f sec',
+                rows_per_sec,
+                rows_num_per_event_min if rows_num_per_event_min is not None else -1,
+                rows_num_per_event_max if rows_num_per_event_max is not None else -1,
+                rows_num,
+                window_size,
+            )
+        else:
+            logging.info("PERF - rows window size=0 can not calc performance for this window")
+
 
     def read(self):
         start_timestamp = int(time.time())
@@ -96,20 +100,20 @@ def read(self):
                 start = time.time()
                 rows_num = 0
                 rows_num_since_interim_performance_report = 0
-                rows_per_event = 0
-                rows_per_event_min = 0
-                rows_per_event_max = 0
+                rows_num_per_event = 0
+                rows_num_per_event_min = None
+                rows_num_per_event_max = None
 
 
                 # fetch available events from MySQL
                 for mysql_event in self.binlog_stream:
                     if isinstance(mysql_event, WriteRowsEvent):
 
-                        rows_per_event = len(mysql_event.rows)
-                        if rows_per_event < rows_per_event_min:
-                            rows_per_event_min = rows_per_event
-                        if rows_per_event > rows_per_event_max:
-                            rows_per_event_max = rows_per_event
+                        rows_num_per_event = len(mysql_event.rows)
+                        if (rows_num_per_event_min is None) or (rows_num_per_event < rows_num_per_event_min):
+                            rows_num_per_event_min = rows_num_per_event
+                        if (rows_num_per_event_max is None) or (rows_num_per_event > rows_num_per_event_max):
+                            rows_num_per_event_max = rows_num_per_event
 
                         if self.subscribers('WriteRowsEvent'):
                             self.write_rows_event_num += 1
@@ -139,12 +143,12 @@ def read(self):
                             self.performance_report(
                                 start=start,
                                 rows_num=rows_num,
-                                rows_per_event_min=rows_per_event_min,
-                                rows_per_event_max=rows_per_event_max,
+                                rows_num_per_event_min=rows_num_per_event_min,
+                                rows_num_per_event_max=rows_num_per_event_max,
                             )
                             rows_num_since_interim_performance_report = 0
-                            rows_per_event_min = 0
-                            rows_per_event_max = 0
+                            rows_num_per_event_min = None
+                            rows_num_per_event_max = None
                     else:
                         # skip non-insert events
                         pass