add Solana support in sync-lag-monitor

- change monitor name from head-lag-monitor to sync-lag-monitor
- add support for tracking Solana sync lag

Author: erwin-wee

README.md

@@ -195,12 +195,12 @@ Chainbench will keep the test data up to date by running a background process th
 
 ### Monitors
 Monitors are separate processes that run during the test to collect or process some additional data and metrics relevant to the test.
-For example, head-lag-monitor will collect the latest block information from the node under test, check the timestamp and compare it to current time to calculate how much the node lags behind.
+For example, sync-lag-monitor will collect the latest block information from the node under test, check the timestamp and compare it to current time to calculate how much the node lags behind.
 You may include monitors in your test by using the `-m` or `--monitor` option and specifying the name of the monitor. At the moment, monitors only work in headless mode.
 
 Here's an example:
 ```shell
-chainbench start --profile evm.light --users 50 --workers 2 --test-time 12h --target https://node-url --headless --autoquit -m head-lag-monitor
+chainbench start --profile evm.light --users 50 --workers 2 --test-time 12h --target https://node-url --headless --autoquit -m sync-lag-monitor
 ```
 
 

chainbench/main.py

@@ -8,8 +8,10 @@
 import click
 import gevent.pool
 from click import Context, Parameter
+from locust.argument_parser import parse_locustfile_paths
+from locust.util.load_locustfile import load_locustfile
 
-from chainbench.user import get_subclass_tasks
+from chainbench.user import EvmUser, SolanaUser, get_subclass_tasks
 from chainbench.user.common import all_method_classes, all_methods
 from chainbench.util.cli import (
     ContextData,
@@ -148,7 +150,7 @@ def validate_profile(ctx: Context, param: Parameter, value: str) -> str:
     default=[],
     help="Add a monitor to collect additional data or metrics. "
     "You may specify this option multiple times for different monitors",
-    type=click.Choice(["head-lag-monitor"], case_sensitive=False),
+    type=click.Choice(["sync-lag-monitor"], case_sensitive=False),
     multiple=True,
 )
 @click.option(
@@ -251,18 +253,16 @@ def start(
         click.echo(f"Profile path {profile_path} does not exist.")
         sys.exit(1)
 
+    user_classes = {}
+    for locustfile in parse_locustfile_paths([profile_path.__str__()]):
+        _, _user_classes, _ = load_locustfile(locustfile)
+        for key, value in _user_classes.items():
+            user_classes[key] = value
+    test_data_types = set()
+    for user_class in user_classes.values():
+        test_data_types.add(type(getattr(user_class, "test_data")).__name__)
+
     if test_by_directory:
-        from locust.argument_parser import parse_locustfile_paths
-        from locust.util.load_locustfile import load_locustfile
-
-        user_classes = {}
-        test_data_types = set()
-        for locustfile in parse_locustfile_paths([profile_path.__str__()]):
-            _, _user_classes, _ = load_locustfile(locustfile)
-            for key, value in _user_classes.items():
-                user_classes[key] = value
-        for user_class in user_classes.values():
-            test_data_types.add(type(getattr(user_class, "test_data")).__name__)
         if len(test_data_types) > 1:
             click.echo(
                 "Error occurred: Multiple test data types detected. "
@@ -384,9 +384,14 @@ def start(
             tags=["loudspeaker"],
         )
 
+    if list(test_data_types)[0] == "SolanaTestData":
+        user_class = SolanaUser
+    else:
+        user_class = EvmUser
+
     unique_monitors: set[str] = set(monitor)
     for m in unique_monitors:
-        p = Process(target=monitors[m], args=(target, results_path, test_time))
+        p = Process(target=monitors[m], args=(user_class, target, results_path, test_time))
         click.echo(f"Starting monitor {m}")
         p.start()
         ctx.obj.monitors.append(p)
@@ -505,7 +510,7 @@ def shapes() -> None:
 )
 def methods() -> None:
     for method_class in all_method_classes:
-        click.echo(f"\nMethods for {method_class.__name__}:")
+        click.echo(f"\nMethods for {method_class.__name__}: ")
         task_list = get_subclass_tasks(method_class)
         for task in task_list:
             click.echo(f"- {method_class.task_to_method(task.name)}")

chainbench/util/monitor.py

@@ -1,12 +1,14 @@
 import csv
 import logging
+import typing
 from datetime import datetime, timedelta
 from pathlib import Path
 from time import sleep
 
 from locust.util.timespan import parse_timespan
 from orjson import JSONDecodeError
 
+from ..user import SolanaUser
 from .http import HttpClient
 
 logger = logging.getLogger(__name__)
@@ -24,43 +26,81 @@ def calculate_lag(current_timestamp: datetime, block_timestamp: datetime) -> int
     return max(int((current_timestamp - block_timestamp).total_seconds()), 0)
 
 
-def head_lag_monitor(endpoint: str, result_path: Path, duration: str):
-    data = {
-        "id": 1,
+# EVM
+def eth_get_block_by_number(http_client: HttpClient) -> dict:
+    body = {
         "jsonrpc": "2.0",
+        "id": 1,
         "method": "eth_getBlockByNumber",
         "params": ["latest", False],
     }
+    response = http_client.post(data=body)
+    return response.json["result"]
+
+
+# Solana
+
+
+def get_slot(http_client: HttpClient) -> int:
+    response = http_client.post(data={"jsonrpc": "2.0", "id": 1, "method": "getSlot", "params": []})
+    return response.json["result"]
+
+
+def get_block(http_client: HttpClient, slot: int) -> dict:
+    body = {
+        "jsonrpc": "2.0",
+        "id": 1,
+        "method": "getBlock",
+        "params": [
+            slot,
+            {
+                "encoding": "jsonParsed",
+                "transactionDetails": "none",
+                "rewards": False,
+                "maxSupportedTransactionVersion": 0,
+            },
+        ],
+    }
+    response = http_client.post(data=body)
+    return response.json["result"]
+
+
+def sync_lag_monitor(user_class: typing.Any, endpoint: str, result_path: Path, duration: str):
     end_time = datetime.now() + timedelta(seconds=parse_timespan(duration))
     http = HttpClient(endpoint)
     with open(
-        file=f"{result_path}/head_lag.csv",
+        file=f"{result_path}/sync_lag.csv",
         mode="a",
         encoding="utf-8-sig",
         newline="",
     ) as csv_file:
-        logger.info("Start monitoring head lag")
+        logger.info("Start monitoring sync lag")
         csv_writer = csv.writer(csv_file)
         csv_writer.writerow(["timestamp", "lag (s)", "block number"])
         while datetime.now() < end_time:
             current_timestamp = datetime.now()
-            response = http.post(data=data)
             try:
-                block_timestamp = datetime.fromtimestamp(int(response.json["result"]["timestamp"], 0))
-                block_number = int(response.json["result"]["number"], 0)
+                if user_class == SolanaUser:
+                    block_number = get_slot(http)
+                    block = get_block(http, block_number)
+                    block_timestamp = datetime.fromtimestamp(block["blockTime"])
+                else:
+                    block = eth_get_block_by_number(http)
+                    block_timestamp = datetime.fromtimestamp(int(block["timestamp"], 0))
+                    block_number = int(block["number"], 0)
                 csv_writer.writerow(
                     [
                         current_timestamp,
                         f"{calculate_lag(current_timestamp, block_timestamp)}",
                         block_number,
                     ]
                 )
-                logger.info("Written 1 row to head_lag.csv")
+                logger.info("Written 1 row to sync_lag.csv")
                 sleep(10)
             except (KeyError, JSONDecodeError):
                 logger.error("Error decoding JSON or key not found")
                 sleep(1)
-    logger.info("Finished monitoring head lag")
+    logger.info("Finished monitoring sync lag")
 
 
-monitors = {"head-lag-monitor": head_lag_monitor}
+monitors = {"sync-lag-monitor": sync_lag_monitor}