Merge pull request #43 from matburt/metrics_service

Adding a basic metrics service and some metrics to track

Author: matburt

Pipfile

@@ -12,6 +12,7 @@ flake8 = "*"
 
 [packages]
 python-dateutil = "*"
+prometheus-client = "*"
 
 [requires]
 python_version = "3.6"

Pipfile.lock

@@ -137,6 +137,21 @@ def __init__(self, args=None):
             value_type='bool',
             hint='Disable the server function and only connect to configured peers',
         )
+        self.add_config_option(
+            section='node',
+            key='stats_enable',
+            default_value=None,
+            set_value=True,
+            value_type='bool',
+            hint="Enable Prometheus style stats port",
+        )
+        self.add_config_option(
+            section='node',
+            key='stats_port',
+            default_value=8889,
+            value_type='int',
+            hint='Port to listen for requests to show stats',
+        )
         self.add_config_option(
             section='node',
             key='ping_interval',
@@ -192,6 +207,21 @@ def __init__(self, args=None):
             value_type='path',
             hint='Path to control socket for controller commands.',
         )
+        self.add_config_option(
+            section='controller',
+            key='stats_enable',
+            default_value=None,
+            set_value=True,
+            value_type='bool',
+            hint="Enable Prometheus style stats port",
+        )
+        self.add_config_option(
+            section='controller',
+            key='stats_port',
+            default_value=8889,
+            value_type='int',
+            hint='Port to listen for requests to show stats',
+        )
         self.add_config_option(
             section='ping',
             key='count',

receptor/config.py

@@ -4,6 +4,8 @@
 import sys
 import time
 
+from prometheus_client import start_http_server
+
 from .receptor import Receptor
 from . import controller
 from . import exceptions
@@ -15,12 +17,18 @@
 def run_as_controller(config):
     receptor = Receptor(config)
     logger.info(f'Starting up as node ID {receptor.node_id}')
+    if config.controller_stats_enable:
+        logger.info(f'Starting stats on port {config.controller_stats_port}')
+        start_http_server(config.controller_stats_port)
     controller.mainloop(receptor, config.controller_socket_path)
 
 
 def run_as_node(config):
     receptor = Receptor(config)
     logger.info(f'Running as Receptor node with ID: {receptor.node_id}')
+    if config.node_stats_enable:
+        logger.info(f'Starting stats on port {config.node_stats_port}')
+        start_http_server(config.node_stats_port)
     node.mainloop(receptor, config.node_ping_interval)
 
 

receptor/entrypoints.py

@@ -10,6 +10,7 @@
 
 from .messages import envelope
 from .exceptions import ReceptorBufferError
+from .stats import connected_peers_guage
 
 logger = logging.getLogger(__name__)
 
@@ -74,9 +75,11 @@ def connection_made(self, transport):
         self.transport = transport
         self.greeted = False
         self.incoming_buffer = DataBuffer()
+        connected_peers_guage.inc()
         self.loop.create_task(self.wait_greeting())
 
     def connection_lost(self, exc):
+        connected_peers_guage.dec()
         self.receptor.remove_connection(self)
 
     def data_received(self, data):
@@ -171,10 +174,12 @@ def __init__(self, receptor, loop):
 
     def connection_made(self, transport):
         self.transport = transport
+        connected_peers_guage.inc()
         if self not in self.receptor.controller_connections:
             self.receptor.controller_connections.append(self)
 
     def connection_lost(self, exc):
+        connected_peers_guage.dec()
         if self in self.receptor.controller_connections:
             self.receptor.controller_connections.remove(self)
 

receptor/protocol.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/receptor.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/router.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/stats.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):

receptor/work.py

@@ -19,6 +19,7 @@
     long_description_content_type='text/markdown',
     python_requires=">=3.6",
     install_requires=[
+        "prometheus_client==0.7.1",
     ],
     zip_safe=False,
     entry_points={