feat: Allow exposing Prometheus metrics in http port (#44)

* feat: Allow exposing Prometheus metrics in http port

* style: fix linters

* fix: default value in cli arg

* fix: improve update metrics task in prometheus

* fix: prometheus test

* fix: style

* fix: skip yapf in Makefile

* fix: tests not passing in python 3.6

* fix: styles

Author: luislhl

Makefile

@@ -61,7 +61,8 @@ fmt: yapf isort
 
 .PHONY: yapf
 yapf: $(py_sources) $(py_tests)
-	yapf -rip $^ -e \*_pb2.py,\*_pb2_grpc.py
+	echo "Skipping yapf because it's conflicting with flake8"
+#	yapf -rip $^ -e \*_pb2.py,\*_pb2_grpc.py
 
 .PHONY: isort
 isort: $(py_sources) $(py_tests)

tests/test_prometheus.py

@@ -9,6 +9,7 @@
 import tempfile
 import unittest
 
+from tests.utils import async_test
 from txstratum.manager import TxMiningManager
 from txstratum.prometheus import METRIC_INFO, MetricData, PrometheusExporter
 
@@ -27,9 +28,10 @@ def test_collect_metrics(self):
         metric_keys = set(MetricData._fields)
         self.assertEqual(description_keys, metric_keys)
 
-    def test_update_metrics(self):
+    @async_test
+    async def test_update_metrics(self):
         prometheus = PrometheusExporter(self.manager, self.tmpdir)
-        prometheus.update_metrics()
+        await prometheus.update_metrics()
         self.assertTrue(os.path.exists(prometheus.filepath))
 
         # Check if all metrics are in the file.

tests/utils.py

@@ -27,3 +27,14 @@ def enable(self) -> None:
 
     def disable(self) -> None:
         txstratum.time.set_time_function(None)
+
+
+# XXX: We could use unittest's IsolatedAsyncioTestCase instead, but it's available only in Python 3.8+.
+def async_test(coro):
+    def wrapper(*args, **kwargs):
+        loop = asyncio.new_event_loop()
+        try:
+            return loop.run_until_complete(coro(*args, **kwargs))
+        finally:
+            loop.close()
+    return wrapper

txstratum/cli.py

@@ -30,6 +30,7 @@ def create_parser() -> ArgumentParser:
     parser.add_argument('--tx-timeout', help='Tx mining timeout (seconds)', type=int, default=None)
     parser.add_argument('--fix-invalid-timestamp', action='store_true', help='Fix invalid timestamp to current time')
     parser.add_argument('--prometheus', help='Path to export metrics for Prometheus', type=str, default=None)
+    parser.add_argument('--prometheus-port', help='Enables exporting metrics in a http port', type=int, default=None)
     parser.add_argument('--testnet', action='store_true', help='Use testnet config parameters')
     parser.add_argument('--address', help='Mining address for blocks', type=str, default=None)
     parser.add_argument('--allow-non-standard-script', action='store_true', help='Accept mining non-standard tx')
@@ -102,6 +103,12 @@ def execute(args: Namespace) -> None:
         metrics = PrometheusExporter(manager, args.prometheus)
         metrics.start()
 
+    if args.prometheus_port:
+        from txstratum.prometheus import HttpPrometheusExporter
+        http_metrics = HttpPrometheusExporter(manager, args.prometheus_port)
+        http_metrics.start()
+        logger.info('Prometheus metrics server running at 0.0.0.0:{}...'.format(args.prometheus_port))
+
     if args.ban_addrs or args.ban_tx_ids:
         tx_filters.append(FileFilter.load_from_files(args.ban_tx_ids, args.ban_addrs))
 

txstratum/prometheus.py

@@ -2,7 +2,9 @@
 import os
 from typing import TYPE_CHECKING, Dict, NamedTuple
 
-from prometheus_client import CollectorRegistry, Gauge, write_to_textfile  # type: ignore
+from prometheus_client import CollectorRegistry, Gauge, start_http_server, write_to_textfile  # type: ignore
+
+from txstratum.utils import Periodic
 
 if TYPE_CHECKING:
     from txstratum.manager import TxMiningManager
@@ -50,24 +52,16 @@ def collect_metrics(manager: 'TxMiningManager') -> MetricData:
     )
 
 
-class PrometheusExporter:
-    """Class that sends hathor metrics to a node exporter that will be read by Prometheus."""
+class BasePrometheusExporter:
+    """Base class for prometheus exporters."""
 
-    def __init__(self, manager: 'TxMiningManager', path: str, filename: str = 'tx-mining-service.prom'):
-        """Init PrometheusExporter.
+    def __init__(self, manager: 'TxMiningManager'):
+        """Init BasePrometheusExporter.
 
         :param manager: Manager where the metrics will be collected from
-        :param path: Path to save the prometheus file
-        :param filename: Name of the prometheus file (must end in .prom)
         """
         self.manager = manager
 
-        # Create full directory, if does not exist
-        os.makedirs(path, exist_ok=True)
-
-        # Full filepath with filename
-        self.filepath: str = os.path.join(path, filename)
-
         # Stores all Gauge objects for each metric (key is the metric name)
         self.metric_gauges: Dict[str, Gauge] = {}
 
@@ -77,8 +71,8 @@ def __init__(self, manager: 'TxMiningManager', path: str, filename: str = 'tx-mi
         # Interval in which the write data method will be called (in seconds)
         self.call_interval: int = 5
 
-        # If exporter is running
-        self.running: bool = False
+        # Periodic task to update metrics
+        self.update_metrics_task: Periodic = Periodic(self.update_metrics, self.call_interval)
 
     def _initial_setup(self) -> None:
         """Start a collector registry to send data to node exporter."""
@@ -87,28 +81,61 @@ def _initial_setup(self) -> None:
         for name, comment in METRIC_INFO.items():
             self.metric_gauges[name] = Gauge(name, comment, registry=self.registry)
 
-    def start(self) -> None:
-        """Start exporter."""
-        self.running = True
-        self._schedule_and_write_data()
-
-    def update_metrics(self) -> None:
+    async def update_metrics(self) -> None:
         """Update metric_gauges dict with new data from metrics."""
         data = collect_metrics(self.manager)
         for metric_name in METRIC_INFO.keys():
             self.metric_gauges[metric_name].set(getattr(data, metric_name))
 
+    def start(self) -> None:
+        """Start exporter."""
+        asyncio.ensure_future(self.update_metrics_task.start())
+
+    def stop(self) -> None:
+        """Stop exporter."""
+        asyncio.ensure_future(self.update_metrics_task.stop())
+
+
+class PrometheusExporter(BasePrometheusExporter):
+    """Class that sends hathor metrics to a node exporter that will be read by Prometheus."""
+
+    def __init__(self, manager: 'TxMiningManager', path: str, filename: str = 'tx-mining-service.prom'):
+        """Init PrometheusExporter.
+
+        :param manager: Manager where the metrics will be collected from
+        :param path: Path to save the prometheus file
+        :param filename: Name of the prometheus file (must end in .prom)
+        """
+        super().__init__(manager)
+
+        # Create full directory, if does not exist
+        os.makedirs(path, exist_ok=True)
+
+        # Full filepath with filename
+        self.filepath: str = os.path.join(path, filename)
+
+    async def update_metrics(self) -> None:
+        """Update metric_gauges dict with new data from metrics."""
+        await super().update_metrics()
+
         write_to_textfile(self.filepath, self.registry)
 
-    def _schedule_and_write_data(self) -> None:
-        """Update metrics and schedule to be called again."""
-        if self.running:
-            self.update_metrics()
 
-            # Schedule next call
-            loop = asyncio.get_event_loop()
-            loop.call_later(self.call_interval, self._schedule_and_write_data)
+class HttpPrometheusExporter(BasePrometheusExporter):
+    """Class that exposes metrics in a http endpoint."""
 
-    def stop(self) -> None:
-        """Stop exporter."""
-        self.running = False
+    def __init__(self, manager: 'TxMiningManager', port: int):
+        """Init HttpPrometheusExporter.
+
+        :param manager: Manager where the metrics will be collected from
+        :param port: Port to expose the metrics
+        """
+        super().__init__(manager)
+
+        self.port = port
+
+    def start(self) -> None:
+        """Start exporter."""
+        super().start()
+
+        start_http_server(self.port, registry=self.registry)