remove external binary deps on doublezero and solana

Author: alexpyattaev

README.md

@@ -15,11 +15,12 @@ This will not trigger on minor DZ packet loss, only substantial failures in the
 
 ## Installation
 
-Edit the `config.py` file to configure the parameters to your liking.
+You can run the script from an unpriviledged user account or as root.
+Sudo access to the `nft` and `ip` commands should be granted to use this as an
+unpriviledged user.
 
-It is recommended that you run this script from the sol user account (assuming it also has
-access to the `doublezero` command line). Sudo access to the `nft` command
-should be granted to use this as an unpriviledged user. Running this in tmux/zellij
+Edit the `config.py` file to configure the parameters to your liking.
+Running this in tmux/zellij and monitoring the output
 is a viable way to test that the parameters are chosen correctly.
 
 For permanent install it is recommended to have a systemd service configured to
@@ -29,15 +30,8 @@ A systemd unit `doublezero_monitor.service` is provided, install as appropriate
 sudo cp doublezero_monitor.service /etc/systemd/system/
 ```
 
-Keep in mind that when running as system service, the script will still need access to both `solana`
-and `doublezero` binaries to perform its function. To check, log in as a root user and verify
-that both commands can still be executed.
+Keep in mind that when running as system service, the script will run as root.
 
-In addition, you should make the DZ config available to the root user as follows:
-```bash
-mkdir -p /root/.config/doublezero/
-ln -s /home/sol/.config/doublezero/cli  /root/.config/doublezero/cli
-```
 
 ## For IBRL mode
 

config.py

@@ -1,20 +1,19 @@
-
 # Tunable parameters
 
-# Which cluster to connect to (fed to solana CLI)
-CLUSTER="testnet" # "mainnet-beta"
+# Which cluster to connect to (this is used to pick the correct RPC endpoint)
+CLUSTER = "testnet"  # "mainnet-beta"
 
 # how much stake do we need to observe to consider connection "good"
 STAKE_THRESHOLD: float = 0.9
 # How long do we accumulate packets before checking the counters
-PASSIVE_MONITORING_INTERVAL_SECONDS: float = 5.0
+PASSIVE_MONITORING_INTERVAL_SECONDS: float = 7.0
 # How long do we wait before consindering connection dead
-GRACE_PERIOD_SEC: float = 10.0
+GRACE_PERIOD_SEC: float = 14.0
 # Interval between refreshes of gossip tables via RPC
 NODE_REFRESH_INTERVAL_SECONDS: float = 60.0
 
 # Path to the admin RPC socket of the validator
-ADMIN_RPC_PATH="/home/sol/ledger/admin.rpc"
+ADMIN_RPC_PATH = "/home/sol/ledger/admin.rpc"
 
 # Parameters you should probably not tune
 

doublezero.py

@@ -1,26 +1,25 @@
 import asyncio
 import ipaddress
 
+
 async def doublezero_is_active() -> bool:
-    CMD = "doublezero status".split(" ")
+    CMD = "ip link show doublezero0 up".split(" ")
     proc = await asyncio.create_subprocess_exec(
-        *CMD,
-        stdout=asyncio.subprocess.DEVNULL,
-        stderr=asyncio.subprocess.DEVNULL
+        *CMD, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.DEVNULL
     )
-    return await proc.wait() == 0
+    out, _ = await proc.communicate()
+    return b"UP" in out
+
 
-async def get_doublezero_routes()->set[ipaddress.IPv4Address]:
+async def get_doublezero_routes() -> set[ipaddress.IPv4Address]:
     CMD = "ip route show table main".split(" ")
     proc = await asyncio.create_subprocess_exec(
-           *CMD,
-           stdout=asyncio.subprocess.PIPE,
-           stderr=asyncio.subprocess.DEVNULL
-       )
+        *CMD, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.DEVNULL
+    )
     output, _ = await proc.communicate()
-    reachable:set[ipaddress.IPv4Address] = set()
+    reachable: set[ipaddress.IPv4Address] = set()
     for line in output.decode().splitlines():
-        if 'dev doublezero0 proto bgp' in line:
-            line = line.split(' ')[0]
+        if "dev doublezero0 proto bgp" in line:
+            line = line.split(" ")[0]
             reachable.add(ipaddress.IPv4Address(line))
     return reachable

helpers.py

@@ -4,55 +4,96 @@
 from traceback import print_exc
 import json
 import os
+from typing import Any
 from config import *
+from urllib import request
+
 # Set sudo command to blank if in systemd (since then we are root)
 SUDO = "" if os.geteuid() == 0 else "sudo "
 
+
 def nft_add_table():
-    cmd=f"{SUDO}nft add table inet {NFT_TABLE}"
-    subprocess.check_call(cmd.split(" "))
-    cmd=f"{SUDO}nft add chain inet {NFT_TABLE} " + r" input { type filter hook input priority 0 \; }"
-    subprocess.check_call(cmd, shell=True)
+    cmd = f"{SUDO}nft add table inet {NFT_TABLE}"
+    _ = subprocess.check_call(cmd.split(" "))
+    cmd = (
+        f"{SUDO}nft add chain inet {NFT_TABLE} "
+        + r" input { type filter hook input priority 0 \; }"
+    )
+    _ = subprocess.check_call(cmd, shell=True)
+
 
 def nft_drop_table():
-    cmd=f"{SUDO}nft delete table inet {NFT_TABLE}"
-    subprocess.call(cmd.split(" "))
+    cmd = f"{SUDO}nft delete table inet {NFT_TABLE}"
+    _ = subprocess.call(cmd.split(" "))
+
 
-def get_nft_counters()->dict[ipaddress.IPv4Address, int]:
-    cmd=f"{SUDO}nft -j list chain inet {NFT_TABLE} input"
-    counters:dict[ipaddress.IPv4Address, int] = {}
+def get_nft_counters() -> dict[ipaddress.IPv4Address, int]:
+    cmd = f"{SUDO}nft -j list chain inet {NFT_TABLE} input"
+    counters: dict[ipaddress.IPv4Address, int] = {}
     try:
         (_status, output) = subprocess.getstatusoutput(cmd)
         x = json.loads(output)
-        for row in x['nftables']:
-            if 'rule' not in row:
+        for row in x["nftables"]:
+            if "rule" not in row:
                 continue
-            expr = row['rule']['expr']
-            source = ipaddress.IPv4Address(expr[0]['match']['right'])
-            counters[source] = expr[1]['counter']['packets']
+            expr = row["rule"]["expr"]
+            source = ipaddress.IPv4Address(expr[0]["match"]["right"])
+            counters[source] = expr[1]["counter"]["packets"]
     except:
         print_exc()
     finally:
         return counters
 
-def nft_add_counter(ip:ipaddress.IPv4Address):
+
+def nft_add_counter(ip: ipaddress.IPv4Address):
     cmd = f"{SUDO} nft add rule inet {NFT_TABLE} input ip saddr {ip} counter"
     (_status, _output) = subprocess.getstatusoutput(cmd)
 
-async def get_staked_nodes() ->dict[str, int]:
-    cmd = f"-u{CLUSTER} validators --output json"
-    proc = await asyncio.create_subprocess_exec("solana", *cmd.split(" "), stdout=asyncio.subprocess.PIPE,
-           stderr=asyncio.subprocess.PIPE)
-    stdout, _stderr = await proc.communicate()
-    output = json.loads(stdout)
-    return {v['identityPubkey']:v['activatedStake'] for v in output["validators"]  if v['activatedStake'] > MIN_STAKE_TO_CARE and not v['delinquent']}
-
-async def get_contact_infos()->dict[str, ipaddress.IPv4Address]:
-    cmd = f"-u{CLUSTER} gossip --output json"
-    proc = await asyncio.create_subprocess_exec("solana", *cmd.split(" "), stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE)
-    stdout, _stderr = await proc.communicate()
-    output = json.loads(stdout)
-    return  {v['identityPubkey']:ipaddress.IPv4Address( v['ipAddress']) for v in output if 'tpuPort' in v}
-
-def kill_dz_interface()->None:
-    subprocess.call(f"{SUDO}ip link set doublezero0 down", shell=True)
+
+async def get_staked_nodes() -> dict[str, int]:
+    output = await get_from_RPC("getVoteAccounts")
+    return {
+        v["nodePubkey"]: v["activatedStake"]
+        for v in output["current"]
+        if v["activatedStake"] > MIN_STAKE_TO_CARE
+    }
+
+
+async def get_contact_infos() -> dict[str, ipaddress.IPv4Address]:
+    output = await get_from_RPC("getClusterNodes")
+    return {
+        v["pubkey"]: ipaddress.IPv4Address(v["tpuQuic"].split(":")[0])
+        for v in output
+        if v.get("tpuQuic") is not None
+    }
+
+
+def kill_dz_interface() -> None:
+    _ = subprocess.call(f"{SUDO}ip link set doublezero0 down", shell=True)
+
+
+async def get_from_RPC(method: str) -> Any:
+    url = f"https://api.{CLUSTER}.solana.com"
+    headers = {"Content-Type": "application/json"}
+    data = json.dumps(
+        {
+            "jsonrpc": "2.0",
+            "id": 1,
+            "method": method,
+            "params": [],
+        }
+    ).encode()
+
+    loop = asyncio.get_event_loop()
+
+    def fetch() -> dict[str, Any]:
+        req = request.Request(url, data=data, headers=headers, method="POST")
+        with request.urlopen(req) as resp:
+            return json.load(resp)
+
+    try:
+        result = await loop.run_in_executor(None, fetch)
+        return result["result"]
+    except Exception as e:
+        print(f"Could not fetch data from RPC, error {e}")
+        return []