Merge pull request #850 from skhoroshavin/indy-1475

INDY-1475: Metrics improvements

Author: ashcherbakov

plenum/common/metrics_collector.py

@@ -18,6 +18,11 @@ class MetricsName(IntEnum):
     INCOMING_NODE_MESSAGE_SIZE = 6         # Incoming node message size, bytes
     OUTGOING_CLIENT_MESSAGE_SIZE = 7       # Outgoing client message size, bytes
     INCOMING_CLIENT_MESSAGE_SIZE = 8       # Incoming client message size, bytes
+    ORDERED_BATCH_SIZE = 9                 # Number of requests ordered
+    REQUEST_PROCESSING_TIME = 10           # Time spent on requests processing (including dynamic validation)
+    MASTER_3PC_BATCH_SIZE = 11             # Number of requests in one 3PC batch created on master instance
+    MASTER_ORDERED_BATCH_SIZE = 12         # Number of requests ordered on master instance
+    MASTER_REQUEST_PROCESSING_TIME = 13    # Time spent on requests processing on master instance
 
 
 MetricsEvent = NamedTuple('MetricsEvent', [('timestamp', datetime), ('name', MetricsName), ('value', float)])
@@ -42,12 +47,15 @@ class KvStoreMetricsFormat:
     seq_mask = (1 << seq_bits) - 1
 
     @staticmethod
-    def encode(event: MetricsEvent, seq_no: int = 0) -> (bytes, bytes):
-        int_ts = int(1000000 * event.timestamp.replace(tzinfo=timezone.utc).timestamp())
+    def encode_key(ts: datetime, seq_no: int):
+        int_ts = int(1000000 * ts.replace(tzinfo=timezone.utc).timestamp())
         int_ts = int_ts & KvStoreMetricsFormat.ts_mask
         seq_no = seq_no & KvStoreMetricsFormat.seq_mask
-        key = ((int_ts << KvStoreMetricsFormat.seq_bits) | seq_no).to_bytes(64, byteorder='big', signed=False)
+        return ((int_ts << KvStoreMetricsFormat.seq_bits) | seq_no).to_bytes(64, byteorder='big', signed=False)
 
+    @staticmethod
+    def encode(event: MetricsEvent, seq_no: int = 0) -> (bytes, bytes):
+        key = KvStoreMetricsFormat.encode_key(event.timestamp, seq_no)
         value = event.name.to_bytes(32, byteorder='big', signed=False) + struct.pack('d', event.value)
         return key, value
 

plenum/common/metrics_stats.py

@@ -106,7 +106,7 @@ def __init__(self):
     def add(self, id: MetricsName, value: float):
         self._stats[id].add(value)
 
-    def get(self, id: MetricsName):
+    def get(self, id: MetricsName) -> ValueAccumulator:
         return self._stats[id]
 
     def merge(self, other):
@@ -186,13 +186,11 @@ def load_metrics_from_kv_store(storage: KeyValueStorage,
                                step: timedelta = timedelta(minutes=1)) -> MetricsStats:
     result = MetricsStats(step)
 
-    # TODO: Implement faster filtering by timestamps
-    for k, v in storage.iterator():
+    start = KvStoreMetricsFormat.encode_key(min_ts, 0) if min_ts else None
+    for k, v in storage.iterator(start=start):
         ev = KvStoreMetricsFormat.decode(k, v)
-        if min_ts is not None and ev.timestamp < min_ts:
-            continue
         if max_ts is not None and ev.timestamp > max_ts:
-            continue
+            break
         result.add(ev.timestamp, ev.name, ev.value)
 
     return result

plenum/server/replica.py

@@ -704,6 +704,8 @@ def trackBatches(self, pp: PrePrepare, prevStateRootHash):
         self.logger.trace('{} tracking batch for {} with state root {}'.format(
             self, pp, prevStateRootHash))
         self.metrics.add_event(MetricsName.THREE_PC_BATCH_SIZE, len(pp.reqIdr))
+        if self.isMaster:
+            self.metrics.add_event(MetricsName.MASTER_3PC_BATCH_SIZE, len(pp.reqIdr))
         self.batches[(pp.viewNo, pp.ppSeqNo)] = [pp.ledgerId, pp.discarded,
                                                  pp.ppTime, prevStateRootHash]
 
@@ -816,6 +818,7 @@ def consume_req_queue_for_pre_prepare(self, ledger_id, view_no, pp_seq_no):
         validReqs = []
         inValidReqs = []
         rejects = []
+        start = time.perf_counter()
         while len(validReqs) + len(inValidReqs) < self.config.Max3PCBatchSize \
                 and self.requestQueues[ledger_id]:
             key = self.requestQueues[ledger_id].pop(0)
@@ -826,6 +829,10 @@ def consume_req_queue_for_pre_prepare(self, ledger_id, view_no, pp_seq_no):
             else:
                 self.logger.debug('{} found {} in its request queue but the '
                                   'corresponding request was removed'.format(self, key))
+        duration = time.perf_counter() - start
+        self.metrics.add_event(MetricsName.REQUEST_PROCESSING_TIME, duration)
+        if self.isMaster:
+            self.metrics.add_event(MetricsName.MASTER_REQUEST_PROCESSING_TIME, duration)
 
         return validReqs, inValidReqs, rejects, tm
 
@@ -1232,6 +1239,7 @@ def _apply_pre_prepare(self, pre_prepare: PrePrepare, sender: str) -> Optional[i
                 old_state_root,
                 old_txn_root))
 
+        start = time.perf_counter()
         for req_key in pre_prepare.reqIdr:
             req = self.requests[req_key].finalised
 
@@ -1240,6 +1248,10 @@ def _apply_pre_prepare(self, pre_prepare: PrePrepare, sender: str) -> Optional[i
                                        valid_reqs,
                                        invalid_reqs,
                                        rejects)
+        duration = time.perf_counter() - start
+        self.metrics.add_event(MetricsName.REQUEST_PROCESSING_TIME, duration)
+        if self.isMaster:
+            self.metrics.add_event(MetricsName.MASTER_REQUEST_PROCESSING_TIME, duration)
 
         def revert():
             self.revert(pre_prepare.ledgerId,
@@ -1735,6 +1747,9 @@ def order_3pc_key(self, key):
                          format(self, pp.viewNo, pp.ppSeqNo, pp.ledgerId,
                                 pp.stateRootHash, pp.txnRootHash, len(pp.reqIdr[:pp.discarded]),
                                 len(pp.reqIdr[pp.discarded:])))
+        self.metrics.add_event(MetricsName.ORDERED_BATCH_SIZE, pp.discarded)
+        if self.isMaster:
+            self.metrics.add_event(MetricsName.MASTER_ORDERED_BATCH_SIZE, pp.discarded)
 
         self.addToCheckpoint(pp.ppSeqNo, pp.digest, pp.ledgerId)