Move stats under a single module, hook up more stats

Author: matburt

receptor/config.py

@@ -5,8 +5,6 @@
 import os
 import ssl
 
-from prometheus_client import Counter, Gauge
-
 from .entrypoints import run_as_node, run_as_controller, run_as_ping, run_as_send
 from .exceptions import ReceptorRuntimeError, ReceptorConfigError
 
@@ -32,12 +30,6 @@
     },
 }
 
-messages_received_counter = Counter("incoming_messages", "Messages received from Receptor Peers")
-connected_peers_guage = Gauge("connected_peers", "Number of active peer connections")
-work_counter = Counter("work_events", "A count of the number of work events that have been received")
-active_work_gauge = Gauge("active_work", "Amount of work currently being performed")
-route_counter = Counter("route_events", "A count of the number of messages that have been routed elsewhere in the mesh")
-
 
 def py_class(class_spec):
     if class_spec not in SINGLETONS:

receptor/protocol.py

@@ -10,7 +10,7 @@
 
 from .messages import envelope
 from .exceptions import ReceptorBufferError
-from .config import connected_peers_guage
+from .stats import connected_peers_guage
 
 logger = logging.getLogger(__name__)
 

receptor/receptor.py

@@ -9,6 +9,7 @@
 from .router import MeshRouter
 from .work import WorkManager
 from .messages import envelope, directive
+from .stats import messages_received_counter
 from . import exceptions
 
 RECEPTOR_DIRECTIVE_NAMESPACE = 'receptor'
@@ -203,6 +204,7 @@ async def handle_message(self, msg):
             directive=self.handle_directive,
             response=self.handle_response,
         )
+        messages_received_counter.inc()
         outer_env = envelope.OuterEnvelope(**msg)
         next_hop = self.router.next_hop(outer_env.recipient)
         if next_hop:

receptor/router.py

@@ -9,6 +9,7 @@
 from dateutil import parser
 from .messages import envelope
 from .exceptions import UnrouteableError, ReceptorBufferError
+from .stats import route_counter
 
 logger = logging.getLogger(__name__)
 
@@ -145,6 +146,7 @@ async def forward(self, outer_envelope, next_hop):
         outer_envelope.route_list.append(self.node_id)
         logger.debug(f'Forwarding frame {outer_envelope.frame_id} to {next_hop}')
         try:
+            route_counter.inc()
             buffer_obj.push(outer_envelope.serialize().encode("utf-8"))
         except ReceptorBufferError as e:
             logger.exception("Receptor Buffer Write Error forwarding message to {}: {}".format(next_hop, e))

receptor/stats.py

@@ -0,0 +1,7 @@
+from prometheus_client import Counter, Gauge
+
+messages_received_counter = Counter("incoming_messages", "Messages received from Receptor Peers")
+connected_peers_guage = Gauge("connected_peers", "Number of active peer connections")
+work_counter = Counter("work_events", "A count of the number of work events that have been received")
+active_work_gauge = Gauge("active_work", "Amount of work currently being performed")
+route_counter = Counter("route_events", "A count of the number of messages that have been routed elsewhere in the mesh")

receptor/work.py

@@ -4,6 +4,9 @@
 
 from . import exceptions
 from .messages import envelope
+from .stats import work_counter, active_work_gauge
+
+
 logger = logging.getLogger(__name__)
 
 
@@ -27,13 +30,16 @@ def get_work(self):
         return self.active_work
 
     def add_work(self, env):
+        work_counter.inc()
+        active_work_gauge.inc()
         self.active_work.append(dict(id=env.message_id,
                                      directive=env.directive,
                                      sender=env.sender))
 
     def remove_work(self, env):
         for work in self.active_work:
             if env.message_id == work["id"]:
+                active_work_gauge.dec()
                 self.active_work.remove(work)
 
     async def handle(self, inner_env):