Introduce better domain protocol validation (#19)

Author: namikmesic

src/collectors/bitcoin.py

@@ -1,16 +1,19 @@
 import asyncio
 from bitcoinrpc import BitcoinRPC
-from helpers import strip_url, validate_protocol, generate_labels_from_metadata
+from helpers import check_protocol, strip_url, generate_labels_from_metadata
 from time import perf_counter
 from settings import logger
 
 
 class bitcoin_collector():
 
     def __init__(self, rpc_metadata):
-        validate_protocol(rpc_metadata['url'], "https")
         self.url = rpc_metadata['url']
-        self.labels, self.labels_values = generate_labels_from_metadata(rpc_metadata)
+        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)
 
     async def _probe(self, metrics):
         try:

src/collectors/cardano.py

@@ -1,17 +1,19 @@
 from settings import logger
 import json
 from collectors.ws import websocket_collector
-from helpers import strip_url, validate_protocol, generate_labels_from_metadata
+from helpers import check_protocol, strip_url, generate_labels_from_metadata
 
 
 class cardano_collector():
 
     def __init__(self, rpc_metadata):
-        validate_protocol(rpc_metadata['url'], "wss")
         self.url = rpc_metadata['url']
-
-        self.labels, self.labels_values = generate_labels_from_metadata(rpc_metadata)
-        self.ws_collector = websocket_collector(self.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)
 
     def _get_block_height(self):
         blk_height_payload = {

src/collectors/conflux.py

@@ -1,28 +1,32 @@
 from settings import cfg, logger
 from conflux_web3 import Web3
 import asyncio
-from helpers import strip_url, validate_protocol, generate_labels_from_metadata
+from helpers import strip_url, check_protocol, generate_labels_from_metadata
 from collectors.ws import websocket_collector
 
 
 class conflux_collector():
 
     def __init__(self, rpc_metadata):
-        validate_protocol(rpc_metadata['url'], "wss")
         self.url = rpc_metadata['url']
-        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)
+        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()
+            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, metrics):
         try:
@@ -42,6 +46,8 @@ def probe(self, metrics):
                     )
 
                 metrics['brpc_gas_price'].add_metric(self.labels_values, self.client.cfx.gas_price)
+                metrics['brpc_client_version'].add_metric(self.labels_values,
+                                                          value={"client_version": self.client.clientVersion})
             else:
                 logger.info("Client is not connected to {}".format(strip_url(self.url)))
                 metrics['brpc_health'].add_metric(self.labels_values, False)

src/collectors/dogecoin.py

@@ -1,4 +1,4 @@
-from helpers import strip_url, validate_protocol, generate_labels_from_metadata
+from helpers import strip_url, check_protocol, generate_labels_from_metadata
 from settings import logger
 from time import perf_counter
 import requests
@@ -7,9 +7,12 @@
 class doge_collector():
 
     def __init__(self, rpc_metadata):
-        validate_protocol(rpc_metadata['url'], "https")
         self.url = rpc_metadata['url']
-        self.labels, self.labels_values = generate_labels_from_metadata(rpc_metadata)
+        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)
 
     def probe(self, metrics):
         try:

src/collectors/evm.py

@@ -1,30 +1,32 @@
 from settings import cfg, logger
 from web3 import Web3
 import asyncio
-from helpers import strip_url, validate_protocol, generate_labels_from_metadata
+from helpers import strip_url, check_protocol, generate_labels_from_metadata
 from collectors.ws import websocket_collector
 from websockets.exceptions import WebSocketException
 
 
 class evm_collector():
 
     def __init__(self, rpc_metadata):
-        validate_protocol(rpc_metadata['url'], "wss")
         self.url = rpc_metadata['url']
-        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": "eth_subscribe",
-                                                    "jsonrpc": "2.0",
-                                                    "id": rpc_metadata['chain_id'],
-                                                    "params": ["newHeads"]
-                                                })
-        self.net_peer_enabled = True
-        self.ws_collector.setDaemon(True)
-        self.ws_collector.start()
+        if check_protocol(rpc_metadata['url'], "wss") or check_protocol(rpc_metadata['url'], 'ws'):
+            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": "eth_subscribe",
+                                                        "jsonrpc": "2.0",
+                                                        "id": rpc_metadata['chain_id'],
+                                                        "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, metrics):
         try:
@@ -35,21 +37,13 @@ def probe(self, metrics):
                 metrics['brpc_latency'].add_metric(self.labels_values, self.ws_collector.get_latency())
                 metrics['brpc_block_height'].add_metric(self.labels_values, self.client.eth.block_number)
                 metrics['brpc_total_difficulty'].add_metric(self.labels_values,
-                                                              self.client.eth.get_block('latest')['totalDifficulty'])
+                                                            self.client.eth.get_block('latest')['totalDifficulty'])
                 metrics['brpc_difficulty'].add_metric(self.labels_values,
-                                                        self.client.eth.get_block('latest')['difficulty'])
-
-                try:
-                    if self.net_peer_enabled:
-                        metrics['brpc_net_peer_count'].add_metric(self.labels_values, self.client.net.peer_count)
-                except ValueError:
-                    logger.error(
-                        "Net peer function is not supported for this chain, the collector will ignore this from this point on."
-                    )
-                    self.net_peer_enabled = False
-
+                                                      self.client.eth.get_block('latest')['difficulty'])
                 metrics['brpc_gas_price'].add_metric(self.labels_values, self.client.eth.gas_price)
                 metrics['brpc_max_priority_fee'].add_metric(self.labels_values, self.client.eth.max_priority_fee)
+                metrics['brpc_client_version'].add_metric(self.labels_values,
+                                                          value={"client_version": self.client.clientVersion})
             else:
                 logger.info("Client is not connected to {}".format(strip_url(self.url)))
                 metrics['brpc_health'].add_metric(self.labels_values, False)

src/collectors/filecoin.py

@@ -1,4 +1,4 @@
-from helpers import strip_url, validate_protocol, generate_labels_from_metadata
+from helpers import strip_url, check_protocol, generate_labels_from_metadata
 from settings import logger
 from time import perf_counter
 import requests
@@ -7,10 +7,12 @@
 class filecoin_collector():
 
     def __init__(self, rpc_metadata):
-        validate_protocol(rpc_metadata['url'], "https")
         self.url = rpc_metadata['url']
-
-        self.labels, self.labels_values = generate_labels_from_metadata(rpc_metadata)
+        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)
 
     def probe(self, metrics):
         try:

src/collectors/solana.py

@@ -1,13 +1,22 @@
 from settings import cfg, logger
 from solana.rpc.api import Client
-from helpers import strip_url, url_join, validate_protocol, generate_labels_from_metadata
+from helpers import strip_url, url_join, check_protocol, generate_labels_from_metadata
 from collectors.ws import websocket_collector
 import requests
 
 
 class solana_collector():
 
     def __init__(self, rpc_metadata):
+        self.url = rpc_metadata['url']
+        if check_protocol(self.url, "https"):
+            self.health_uri = url_join(self.url, "/health")
+            self.client = Client(self.url, timeout=cfg.response_timeout)
+
+            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)
 
         try:
             self.subscribe_url = rpc_metadata['subscribe_url']
@@ -16,21 +25,18 @@ def __init__(self, rpc_metadata):
                 "Please note that solana collector requires subscribe_url websocket endpoint on top of regular url. Please refer to example configuration for example"
             )
 
-        validate_protocol(rpc_metadata['subscribe_url'], 'wss')
-        validate_protocol(rpc_metadata['url'], 'https')
-        self.url = rpc_metadata['url']
-        self.health_uri = url_join(self.url, "/health")
-        self.client = Client(self.url, timeout=cfg.response_timeout)
-
-        self.labels, self.labels_values = generate_labels_from_metadata(rpc_metadata)
-        self.ws_collector = websocket_collector(self.subscribe_url,
-                                                sub_payload={
-                                                    "jsonrpc": "2.0",
-                                                    "id": 1,
-                                                    "method": "slotSubscribe"
-                                                })
-        self.ws_collector.setDaemon(True)
-        self.ws_collector.start()
+        if check_protocol(self.subscribe_url, "wss") or check_protocol(self.subscribe_url, 'ws'):
+            self.ws_collector = websocket_collector(self.subscribe_url,
+                                                    sub_payload={
+                                                        "jsonrpc": "2.0",
+                                                        "id": 1,
+                                                        "method": "slotSubscribe"
+                                                    })
+            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 is_connected(self) -> bool:
         """Health check."""

src/collectors/starkware.py

@@ -1,16 +1,18 @@
 from settings import logger
-from helpers import strip_url, validate_protocol, generate_labels_from_metadata
+from helpers import strip_url, check_protocol, generate_labels_from_metadata
 from time import perf_counter
 import requests
 
 
 class starkware_collector():
 
     def __init__(self, rpc_metadata):
-        validate_protocol(rpc_metadata['url'], "https")
         self.url = rpc_metadata['url']
-
-        self.labels, self.labels_values = generate_labels_from_metadata(rpc_metadata)
+        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)
 
     def probe(self, metrics):
         try:

src/exporter.py

@@ -1,6 +1,6 @@
 from concurrent.futures import ThreadPoolExecutor
 from prometheus_client import REGISTRY, make_wsgi_app
-from prometheus_client.metrics_core import GaugeMetricFamily, CounterMetricFamily
+from prometheus_client.metrics_core import GaugeMetricFamily, CounterMetricFamily, InfoMetricFamily
 from wsgiref.simple_server import make_server
 from helpers import strip_url
 from collectors.evm import evm_collector
@@ -65,7 +65,6 @@ def _instantiate_bitcoin(self):
         for item in cfg.endpoints:
             logger.info("Initializing bitcoin node {}".format(strip_url(item['url'])))
             self.collectors.append(bitcoin_collector(item))
-            print(self.collectors[0].labels)
             self.labels = self.collectors[0].labels
 
     def _instantiate_doge(self):
@@ -117,7 +116,11 @@ def collect(self):
             "brpc_net_peer_count":
             GaugeMetricFamily('brpc_net_peer_count',
                               'Number of peers currently connected to the client.',
-                              labels=self.labels)
+                              labels=self.labels),
+            "brpc_client_version":
+            InfoMetricFamily('brpc_client_version',
+                             'Client version for the particular RPC endpoint.',
+                             labels=self.labels)
         }
 
         def write_metrics(prom_collector, metrics):

src/helpers.py

@@ -8,15 +8,8 @@ def strip_url(url):
     return urllib.parse.urlparse(url).hostname
 
 
-def validate_protocol(url, protocol):
-    if urllib.parse.urlparse(url).scheme == protocol:
-        pass
-    else:
-        logger.error("Non {} endpoint provided for {}".format(protocol, strip_url(url)))
-        exit(1)
-
-
 def check_protocol(url, protocol):
+    # Allow both ws and wss if either ws or wss is used to check protocols
     return urllib.parse.urlparse(url).scheme == protocol