v4.2.0

- Standardise on websockets and requests
|- Implement structured logging

Author: namikmesic

requirements.txt

@@ -3,6 +3,5 @@ prometheus-client==0.13.1
 pyyaml==6.0
 schema==0.7.5
 websockets==10.3
-web3==v6.0.0-beta.4
-solana==0.25.1
-conflux-web3==1.0.0b1
+structlog==22.1.0
+requests==2.28.1

src/collectors/bitcoin.py

@@ -1,39 +1,46 @@
-from helpers import check_protocol, strip_url, generate_labels_from_metadata
-from time import perf_counter
+from helpers import strip_url, generate_labels_from_metadata
 from settings import logger
 from metrics_processor import results
 import requests
+from collectors.https import https_connection
 
 
 class bitcoin_collector():
 
     def __init__(self, rpc_metadata):
-        self.url = rpc_metadata['url']
-        if check_protocol(self.url, "https"):
-            self.labels, self.labels_values = generate_labels_from_metadata(rpc_metadata)
-        else:
-            logger.error("Please provide https endpoint for {}".format(strip_url(self.url)))
-            exit(1)
+        self.url, self.stripped_url = rpc_metadata['url'], strip_url(rpc_metadata['url'])
+        self.labels, self.labels_values = generate_labels_from_metadata(rpc_metadata)
+        self.client = https_connection(self.url)
+
+    def _get_client_version(self):
+        payload = {"jsonrpc": "1.0", "id": "exporter", "method": "getnetworkinfo", "params": []}
+        response = requests.post(self.url, json=payload).json()['result']
+        version, subversion, protocolversion = response['version'], response['subversion'], response['protocolversion']
+        version = f"version:{version} subversion:{subversion} protocolversion:{protocolversion}"
+        return version
+
+    def _get_height_and_difficulty(self):
+        payload = {"jsonrpc": "1.0", "id": "exporter", "method": "getblockchaininfo", "params": []}
+        response = requests.post(self.url, json=payload).json()
+        block_height = response['result']['blocks']
+        total_difficulty = response['result']['difficulty']
+        return block_height, total_difficulty
 
     def probe(self) -> results:
         results.register(self.url, self.labels_values)
+        health_check_payload = {"jsonrpc": "1.0", "id": "exporter", "method": "getnetworkinfo"}
         try:
-            payload = {"jsonrpc": "1.0", "id": "exporter", "method": "getblockchaininfo", "params": []}
-            start = perf_counter()
-            response = requests.post(self.url, json=payload).json()
-            latency = (perf_counter() - start) * 1000
-
-            if response:
+            if self.client.is_connected_post_check(health_check_payload):
+                results.record_latency(self.url, self.client.get_latency(health_check_payload))
                 results.record_health(self.url, True)
-                results.record_latency(self.url, latency)
-                results.record_block_height(self.url, response['result']['blocks'])
-                results.record_total_difficulty(self.url, response['result']['difficulty'])
+
+                block_height, total_difficulty = self._get_height_and_difficulty()
+                results.record_client_version(self.url, self._get_client_version())
+                results.record_block_height(self.url, block_height)
+                results.record_total_difficulty(self.url, total_difficulty)
             else:
-                logger.error("Bad response from client {}: {}".format(strip_url(self.url), exc))
+                logger.error("The url did not pass liveness check.", self.stripped_url)
                 results.record_health(self.url, False)
-        except requests.RequestException as exc:
-            logger.error("Health check failed for {}: {}".format(strip_url(self.url), exc))
-            results.record_health(self.url, False)
         except Exception as exc:
-            logger.error("Health check failed for {}: {}".format(strip_url(self.url), exc))
+            logger.error(f"{exc}", url=self.stripped_url)
             results.record_health(self.url, False)

src/collectors/cardano.py

@@ -1,23 +1,20 @@
-from settings import logger
-import json
-from collectors.ws import websocket_collector
-from helpers import check_protocol, strip_url, generate_labels_from_metadata
+from settings import logger, cfg
+from helpers import strip_url, generate_labels_from_metadata, key_from_json_str
 from metrics_processor import results
+import json
+import websockets
+import asyncio
+from collectors.ws import fetch_latency
 
 
 class cardano_collector():
 
     def __init__(self, rpc_metadata):
-        self.url = rpc_metadata['url']
-        if check_protocol(self.url, "wss") or check_protocol(self.url, "ws"):
-            self.labels, self.labels_values = generate_labels_from_metadata(rpc_metadata)
-            self.ws_collector = websocket_collector(self.url)
-        else:
-            logger.error("Please provide https endpoint for {}".format(strip_url(self.url)))
-            exit(1)
+        self.url, self.stripped_url = rpc_metadata['url'], strip_url(rpc_metadata['url'])
+        self.labels, self.labels_values = generate_labels_from_metadata(rpc_metadata)
 
-    def _get_block_height(self):
-        blk_height_payload = {
+    async def _blockHeight(self, websocket):
+        payload = {
             "type": "jsonwsp/request",
             "version": "1.0",
             "servicename": "ogmios",
@@ -26,21 +23,24 @@ def _get_block_height(self):
                 "query": "blockHeight"
             }
         }
-        try:
-            return json.loads(self.ws_collector.query(blk_height_payload))['result']
-        except KeyError as err:
-            logger.error("Failed to fetch block height for {}, error: {}".format(strip_url(self.url), err))
+        await websocket.send(json.dumps(payload))
+        result = await websocket.recv()
+        return key_from_json_str(result, "result")
 
-    def probe(self) -> results:
+    async def _probe(self) -> results:
         results.register(self.url, self.labels_values)
         try:
-            alive = self.ws_collector.get_liveliness()
-            if alive:
+            async with websockets.connect(self.url,
+                                          open_timeout=cfg.open_timeout,
+                                          close_timeout=cfg.close_timeout,
+                                          ping_interval=cfg.ping_interval,
+                                          ping_timeout=cfg.ping_timeout) as websocket:
+                results.record_latency(self.url, await fetch_latency(websocket))
                 results.record_health(self.url, True)
-                results.record_latency(self.url, self.ws_collector.get_latency())
-                results.record_block_height(self.url, self._get_block_height())
-            else:
-                results.record_health(self.url, False)
+                results.record_block_height(self.url, await self._blockHeight(websocket))
         except Exception as exc:
-            logger.error("Failed probing {} with error: {}".format(strip_url(self.url), exc))
             results.record_health(self.url, False)
+            logger.error(f"{exc}", url=self.stripped_url)
+
+    def probe(self):
+        asyncio.run(self._probe())

src/collectors/conflux.py

@@ -1,58 +1,54 @@
-from settings import cfg, logger
-from conflux_web3 import Web3
-import asyncio
-from helpers import strip_url, check_protocol, generate_labels_from_metadata
-from collectors.ws import websocket_collector
+from settings import logger, cfg
+from helpers import strip_url, generate_labels_from_metadata, hex_to_int, key_from_json_str
 from metrics_processor import results
 
+from collectors.ws import subscription, fetch_latency
+import websockets
+import asyncio
+import json
+
 
 class conflux_collector():
 
     def __init__(self, rpc_metadata):
-        self.url = rpc_metadata['url']
-        if check_protocol(self.url, "wss") or check_protocol(self.url, 'ws'):
-            self.labels, self.labels_values = generate_labels_from_metadata(rpc_metadata)
-            self.labels, self.labels_values = generate_labels_from_metadata(rpc_metadata)
-            self.client = Web3(Web3.WebsocketProvider(self.url, websocket_timeout=cfg.response_timeout))
-            self.labels, self.labels_values = generate_labels_from_metadata(rpc_metadata)
-
-            self.ws_collector = websocket_collector(self.url,
-                                                    sub_payload={
-                                                        "method": "cfx_subscribe",
-                                                        "jsonrpc": "2.0",
-                                                        "id": 1,
-                                                        "params": ["newHeads"]
-                                                    })
-            self.ws_collector.setDaemon(True)
-            self.ws_collector.start()
-        else:
-            logger.error("Please provide wss/ws endpoint for {}".format(strip_url(self.url)))
-            exit(1)
-
-    def probe(self) -> results:
+        self.url, self.chain_id, self.stripped_url = rpc_metadata['url'], rpc_metadata['chain_id'], strip_url(
+            rpc_metadata['url'])
+        self.labels, self.labels_values = generate_labels_from_metadata(rpc_metadata)
+
+        sub_payload = {"method": "cfx_subscribe", "jsonrpc": "2.0", "id": 1, "params": ["newHeads"]}
+        self.sub = subscription(self.url, sub_payload)
+        self.sub.isDaemon()
+        self.sub.start()
+
+    async def _cfx_clientVersion(self, websocket):
+        payload = {"jsonrpc": "2.0", "method": "cfx_clientVersion", "params": [], "id": 1}
+        await websocket.send(json.dumps(payload))
+        result = await websocket.recv()
+        return key_from_json_str(result, "result")
+
+    async def _cfx_epochNumber(self, websocket):
+        payload = {"jsonrpc": "2.0", "method": "cfx_epochNumber", "params": [], "id": 1}
+        await websocket.send(json.dumps(payload))
+        result = await websocket.recv()
+        return hex_to_int(key_from_json_str(result, "result"))
+
+    async def _probe(self) -> results:
         results.register(self.url, self.labels_values)
         try:
-            if self.client.isConnected():
+            async with websockets.connect(self.url,
+                                          open_timeout=cfg.open_timeout,
+                                          close_timeout=cfg.close_timeout,
+                                          ping_interval=cfg.ping_interval,
+                                          ping_timeout=cfg.ping_timeout) as websocket:
+                results.record_latency(self.url, await fetch_latency(websocket))
                 results.record_health(self.url, True)
-                results.record_head_count(self.url, self.ws_collector.message_counter)
-                results.record_disconnects(self.url, self.ws_collector.disconnects_counter)
-                results.record_latency(self.url, self.ws_collector.get_latency())
-                results.record_block_height(self.url, self.client.cfx.epoch_number)
-                try:
-                    difficulty = self.client.cfx.get_block_by_hash(self.client.cfx.get_best_block_hash())['difficulty']
-                    results.record_difficulty(self.url, difficulty)
-                except TypeError:
-                    logger.error(
-                        "RPC Endpoint sent faulty response type when querying for difficulty. This is most likely issue with RPC endpoint."
-                    )
-                results.record_gas_price(self.url, self.client.cfx.gas_price)
-                results.record_client_version(self.url, self.client.clientVersion)
-            else:
-                logger.info("Client is not connected to {}".format(strip_url(self.url)))
-                results.record_health(self.url, False)
-        except asyncio.exceptions.TimeoutError:
-            logger.info("Client timed out for {}".format(strip_url(self.url)))
-            results.record_health(self.url, False)
+                results.record_block_height(self.url, await self._cfx_epochNumber(websocket))
+                results.record_client_version(self.url, await self._cfx_clientVersion(websocket))
+                results.record_head_count(self.url, self.sub.head_counter)
+                results.record_disconnects(self.url, self.sub.disconnects)
         except Exception as exc:
-            logger.error("Failed probing {} with error: {}".format(strip_url(self.url), exc))
             results.record_health(self.url, False)
+            logger.error(f"{exc}", url=self.stripped_url)
+
+    def probe(self):
+        asyncio.run(self._probe())