Fix timeouts (#27)

Author: namikmesic

src/collectors/cardano.py

@@ -24,7 +24,7 @@ async def _blockHeight(self, websocket):
             }
         }
         await websocket.send(json.dumps(payload))
-        result = await websocket.recv()
+        result = await asyncio.wait_for(websocket.recv(), timeout=cfg.response_timeout)
         return key_from_json_str(result, "result")
 
     async def _probe(self) -> results:

src/collectors/conflux.py

@@ -23,13 +23,13 @@ def __init__(self, rpc_metadata):
     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()
+        result = await asyncio.wait_for(websocket.recv(), timeout=cfg.response_timeout)
         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()
+        result = await asyncio.wait_for(websocket.recv(), timeout=cfg.response_timeout)
         return hex_to_int(key_from_json_str(result, "result"))
 
     async def _probe(self) -> results:
@@ -50,7 +50,7 @@ async def _probe(self) -> results:
             logger.error(
                 f"Timed out while trying to establish websocket connection. Current open_timeout value in config: {cfg.open_timeout}.",
                 url=self.stripped_url)
-            results.record_health(self.url, False)        
+            results.record_health(self.url, False)
         except Exception as exc:
             results.record_health(self.url, False)
             logger.error(f"{exc}", url=self.stripped_url)

src/collectors/evm.py

@@ -1,11 +1,10 @@
 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 json
 import websockets
 import asyncio
-import json
 
 
 class evm_collector():
@@ -23,13 +22,13 @@ def __init__(self, rpc_metadata):
     async def _web3_clientVersion(self, websocket):
         payload = {"jsonrpc": "2.0", "method": "web3_clientVersion", "params": [], "id": self.chain_id}
         await websocket.send(json.dumps(payload))
-        result = await websocket.recv()
+        result = await asyncio.wait_for(websocket.recv(), timeout=cfg.response_timeout)
         return key_from_json_str(result, "result")
 
     async def _eth_blockNumber(self, websocket):
         payload = {"jsonrpc": "2.0", "method": "eth_blockNumber", "params": [], "id": self.chain_id}
         await websocket.send(json.dumps(payload))
-        result = await websocket.recv()
+        result = await asyncio.wait_for(websocket.recv(), timeout=cfg.response_timeout)
         return hex_to_int(key_from_json_str(result, "result"))
 
     async def _probe(self) -> results:
@@ -49,7 +48,7 @@ async def _probe(self) -> results:
                 results.record_disconnects(self.url, self.sub.disconnects)
         except asyncio.exceptions.TimeoutError:
             logger.error(
-                f"Timed out while trying to establish websocket connection. Current open_timeout value in config: {cfg.open_timeout}.",
+                f"Timed out while trying to establish websocket connection. Current response_timeout value in config: {cfg.response_timeout}.",
                 url=self.stripped_url)
             results.record_health(self.url, False)
         except Exception as exc:

src/collectors/https.py

@@ -2,6 +2,7 @@
 from settings import logger
 from helpers import strip_url
 from time import perf_counter
+from settings import cfg
 
 
 class https_connection():
@@ -11,7 +12,7 @@ def __init__(self, url):
 
     def is_connected_post_check(self, payload):
         try:
-            response = requests.post(self.url, json=payload)
+            response = requests.post(self.url, json=payload, timeout=cfg.response_timeout)
             response.raise_for_status()
         except (IOError, requests.HTTPError) as exc:
             logger.error(exc, url=self.stripped_url)

src/collectors/ws.py

@@ -48,5 +48,5 @@ async def _subscribe(self):
 async def fetch_latency(websocket):
     start = perf_counter()
     pong = await websocket.ping()
-    await pong
+    await asyncio.wait_for(pong, timeout=cfg.response_timeout)
     return (perf_counter() - start) * 1000

src/settings.py

@@ -43,6 +43,7 @@ def __init__(self, config_file_path: str, validation_file_path: str):
         except KeyError:
             self.ping_timeout = 7
             logger.info(f"connection_parameters.ping_timeout not set, defaulting to {self.ping_timeout}")
+
     def _populate_chain_id_metadata(self):
         # Conditionally add chain_id based on the colelctor type to each rpc item.
         if self.configuration['collector'] not in ['cardano', 'solana', 'bitcoin', 'doge', 'filecoin', 'starkware']: