Merge pull request #29 from cardano-foundation/develop

add cardano governor metrics from prometheus endpoint to mqtt payload

src/blockperf/app.py

@@ -8,6 +8,7 @@
 import time
 from datetime import datetime, timedelta, timezone
 from pathlib import Path
+from urllib.request import urlopen
 
 from blockperf import __version__ as blockperf_version
 from blockperf.blocksample import BlockSample, slot_time_of
@@ -101,6 +102,19 @@ def print_block_stats(self, blocksample: BlockSample) -> None:
         )
         logger.info("\n" + msg)
 
+    def prometheus_metrics(self, endpoint) -> dict:
+        """Retrieves metrics from given endpoint and returns as dictionary"""
+        metrics = None
+        try:
+            with urlopen(endpoint) as response:
+                raw_metrics = response.read().decode()
+                metrics = {
+                    k[0]: k[1] for k in [m.split(" ") for m in raw_metrics.splitlines()]
+                }
+        except Exception as e:
+            logger.error("error retrieving metrics from prometheus: %s", e)
+        return metrics
+
     def mqtt_payload_from(self, sample: BlockSample) -> dict:
         """Returns a dictionary for use as payload when publishing the sample."""
         payload = {
@@ -122,6 +136,16 @@ def mqtt_payload_from(self, sample: BlockSample) -> dict:
             "blockLocalPort": str(self.app_config.relay_public_port),
             "blockG": str(sample.block_g),
         }
+
+        if endpoint := self.app_config.prometheus_endpoint:
+            metrics = self.prometheus_metrics(endpoint)
+            if metrics:
+                payload["inboundGovernor_hot"] = int(
+                    metrics.get("cardano_node_metrics_inboundGovernor_hot")
+                )
+                payload["inboundGovernor_warm"] = int(
+                    metrics.get("cardano_node_metrics_inboundGovernor_warm")
+                )
         return payload
 
     def ensure_maxblocks(self):

src/blockperf/config.py

@@ -306,3 +306,15 @@ def masked_addresses(self) -> list:
                     f"Given address {addr} is not a valid ip address"
                 ) from exc
         return validated_addresses
+
+    @property
+    def prometheus_endpoint(self) -> [str, None]:
+        has_prometheus = self.node_config.get("hasPrometheus")
+        if (
+            not has_prometheus
+            or not isinstance(has_prometheus, list)
+            or not len(has_prometheus) == 2
+            or not isinstance(has_prometheus[1], int)
+        ):
+            return None
+        return f"http://127.0.0.1:{has_prometheus[1]}/metrics"