run_kute.py

# Create argument parser
import argparse
import os
import subprocess
import tempfile
import time
from pathlib import Path

LOKI_ENDPOINT_ENV_VAR = "LOKI_ENDPOINT"
PROMETHEUS_ENDPOINT_ENV_VAR = "PROMETHEUS_ENDPOINT"
PROMETHEUS_USERNAME_ENV_VAR = "PROMETHEUS_USERNAME"
PROMETHEUS_PASSWORD_ENV_VAR = "PROMETHEUS_PASSWORD"

executables = {
    "kute": "./nethermind/tools/artifacts/bin/Nethermind.Tools.Kute/release/Nethermind.Tools.Kute"
}


def _quote(path: Path) -> str:
    return f"\"{path.as_posix()}\""


def resolve_kute_command(preferred_path: str) -> str:
    candidates = []
    if preferred_path:
        candidates.append(Path(preferred_path))

    artifacts_root = Path("nethermind/tools/artifacts/bin")
    if artifacts_root.exists():
        patterns = (
            "Nethermind.Tools.Kute",
            "Nethermind.Tools.Kute.exe",
            "Kute",
            "Kute.exe",
            "Nethermind.Tools.Kute.dll",
            "Kute.dll",
        )
        for pattern in patterns:
            candidates.extend(sorted(artifacts_root.rglob(pattern)))

    seen = set()
    for candidate in candidates:
        if not candidate or not candidate.is_file():
            continue
        resolved = candidate.resolve()
        key = str(resolved)
        if key in seen:
            continue
        seen.add(key)
        if resolved.suffix.lower() == ".dll":
            return f"dotnet {_quote(resolved)}"
        return _quote(resolved)

    raise FileNotFoundError(
        "Unable to find a Kute executable under nethermind/tools/artifacts/bin. "
        "Run `make prepare_tools` first."
    )

def get_command_env(
    client: str,
    test_case_file: str,
):
    command_env = os.environ.copy()

    loki_endpoint = command_env.get(LOKI_ENDPOINT_ENV_VAR, "")
    prometheus_endpoint = command_env.get(PROMETHEUS_ENDPOINT_ENV_VAR, "")
    prometheus_username = command_env.get(PROMETHEUS_USERNAME_ENV_VAR, "")
    prometheus_password = command_env.get(PROMETHEUS_PASSWORD_ENV_VAR, "")

    test_case_name = os.path.splitext(os.path.split(test_case_file)[-1])[0]

    command_env["GA_LOKI_REMOTE_WRITE_URL"] = loki_endpoint
    command_env["GA_PROMETHEUS_REMOTE_WRITE_URL"] = prometheus_endpoint
    command_env["GA_PROMETHEUS_REMOTE_WRITE_USERNAME"] = prometheus_username
    command_env["GA_PROMETHEUS_REMOTE_WRITE_PASSWORD"] = prometheus_password
    command_env["GA_METRICS_LABELS_INSTANCE"] = f"{client}-{test_case_name}"
    command_env["GA_METRICS_LABELS_TESTNET"] = "gas-benchmarks-testnet"
    command_env["GA_METRICS_LABELS_EXECUTION_CLIENT"] = client

    return command_env


def run_command(
    client,
    test_case_file,
    jwt_secret,
    response,
    ec_url,
    kute_extra_arguments,
    skip_forkchoice=False,
    rerun_syncing=False,
):
    def response_contains_syncing(response_path: str) -> bool:
        if not os.path.exists(response_path):
            return False
        try:
            with open(response_path, "r", encoding="utf-8") as handle:
                payload = handle.read()
        except OSError:
            return False
        return "syncing" in payload.lower()

    def stderr_looks_like_syncing_payload(stderr_text: str) -> bool:
        lowered = (stderr_text or "").lower()
        if "syncing" in lowered:
            return True
        # Erigon can briefly return non-JSON payloads (for example plain "syncing")
        # that Kute fails to parse as JSON.
        return "jsonreaderexception" in lowered and "invalid start of a value" in lowered

    input_path = test_case_file
    temp_path = None
    if skip_forkchoice:
        try:
            with open(test_case_file, "r", encoding="utf-8") as original:
                lines = original.readlines()
        except OSError:
            lines = []
        filtered_lines = [line for line in lines if "engine_forkchoiceUpdated" not in line]
        if len(filtered_lines) != len(lines):
            temp_file = tempfile.NamedTemporaryFile(
                mode="w", delete=False, encoding="utf-8", suffix=".txt"
            )
            try:
                temp_file.writelines(filtered_lines)
            finally:
                temp_file.close()
            input_path = temp_file.name
            temp_path = temp_file.name
    # Add logic here to run the appropriate command for each client
    command = (
        f"{executables['kute']} -i \"{input_path}\" -s \"{jwt_secret}\" -r \"{response}\" -a \"{ec_url}\" "
        f"{kute_extra_arguments} "
    )
    # Prepare env variables
    command_env = get_command_env(
        client,
        test_case_file,
    )

    attempts = 0
    max_attempts = 30
    retry_backoff_sec = 1
    while True:
        results = subprocess.run(
            command, shell=True, capture_output=True, text=True, env=command_env
        )
        syncing_from_response = response_contains_syncing(response)
        syncing_from_stderr = stderr_looks_like_syncing_payload(results.stderr)
        if rerun_syncing and \
                attempts < max_attempts and \
                (syncing_from_response or syncing_from_stderr):
            attempts += 1
            retry_reason = "syncing response" if syncing_from_response else "non-JSON syncing-like response"
            print(f"Rerunning {retry_reason} {attempts} times out of {max_attempts} max with {retry_backoff_sec} seconds backoff")
            if os.path.exists(response):
                os.remove(response)
            time.sleep(retry_backoff_sec)
        else:
            break
    if temp_path and os.path.exists(temp_path):
        try:
            os.remove(temp_path)
        except OSError:
            pass
    print(results.stderr, end="")
    if results.returncode != 0:
        raise SystemExit(results.returncode)
    return results.stdout

def save_to(output_folder, file_name, content):
    output_path = os.path.join(output_folder, file_name)
    with open(output_path, "w") as file:
        file.write(content)

def main():
    parser = argparse.ArgumentParser(description="Benchmark script")
    parser.add_argument(
        "--testsPath", type=str, help="Path to test case folder", default="small_tests"
    )
    parser.add_argument("--client", type=str, help="Name of the client we are testing")
    parser.add_argument(
        "--run", type=int, help="Number of times the test was run", default=0
    )
    parser.add_argument(
        "--jwtPath",
        type=str,
        help="Path to the JWT secret used to communicate with the client you want to test",
    )
    parser.add_argument(
        "--output",
        type=str,
        help="Output folder for metrics charts generation. If the folder does "
        "not exist will be created.",
        default="results",
    )
    # Executables path
    parser.add_argument(
        "--dotnetPath",
        type=str,
        help="Path to dotnet executable, needed if testing nethermind and "
        "you need to use something different to dotnet.",
        default="dotnet",
    )
    parser.add_argument(
        "--kutePath",
        type=str,
        help="Path to kute executable.",
        default=executables["kute"],
    )
    parser.add_argument(
        "--kuteArguments", type=str, help="Extra arguments for Kute.", default=""
    )
    parser.add_argument(
        "--ecURL",
        type=str,
        help="Execution client where we will be running kute url.",
        default="http://localhost:8551",
    )
    parser.add_argument(
        "--warmupPath", type=str, help="Set path to warm up file.", default=""
    )
    parser.add_argument(
        "--skipForkchoice",
        action="store_true",
        help="Ignore engine_forkchoiceUpdated requests contained in the test input.",
    )
    parser.add_argument(
        "--rerunSyncing",
        action="store_true",
        help="Rerun a syncing response. Useful when sending preparation payloads to clients that may return SYNCING while still initialising",
    )

    # Parse command-line arguments
    args = parser.parse_args()

    # Get client name and test case folder from command-line arguments
    tests_paths = args.testsPath
    jwt_path = args.jwtPath
    execution_url = args.ecURL
    output_folder = args.output
    executables["dotnet"] = args.dotnetPath
    try:
        executables["kute"] = resolve_kute_command(args.kutePath)
    except FileNotFoundError as exc:
        raise SystemExit(str(exc)) from exc
    kute_arguments = args.kuteArguments
    warmup_file = args.warmupPath
    client = args.client
    run = args.run

    # Create the output folder if it doesn't exist
    if not os.path.exists(output_folder):
        os.makedirs(output_folder)

    if warmup_file != "":
        warmup_response_file = os.path.join(
            output_folder, f"warmup_{client}_response_{run}.txt"
        )
        warmup_response = run_command(
            client,
            warmup_file,
            jwt_path,
            warmup_response_file,
            execution_url,
            kute_arguments,
            skip_forkchoice=args.skipForkchoice,
        )
        save_to(output_folder, f"warmup_{client}_results_{run}.txt", warmup_response)

    # if test case path is a folder, run all the test cases in the folder
    if os.path.isdir(tests_paths):
        tests_cases = []
        for root, _, files in os.walk(tests_paths):
            if len(files) == 0:
                continue
            for file in files:
                if file.endswith("metadata.txt"):
                    continue
                tests_cases.append(os.path.join(root, file))
        for test_case_path in tests_cases:
            name = test_case_path.split("/")[-1].split(".")[0]
            response_file = os.path.join(
                output_folder, f"{client}_response_{run}_{name}.txt"
            )
            response = run_command(
                client,
                test_case_path,
                jwt_path,
                response_file,
                execution_url,
                kute_arguments,
                skip_forkchoice=args.skipForkchoice,
                rerun_syncing=args.rerunSyncing,
            )
            save_to(output_folder, f"{client}_results_{run}_{name}.txt", response)
        return
    else:        
        test_case_without_extension = os.path.splitext(tests_paths.split('/')[-1])[0]
        response_file = os.path.join(output_folder, f'{client}_response_{run}_{test_case_without_extension}.txt')
        response = run_command(
            client,
            tests_paths,
            jwt_path,
            response_file,
            execution_url,
            kute_arguments,
            skip_forkchoice=args.skipForkchoice,
            rerun_syncing=args.rerunSyncing,
        )
        save_to(output_folder, f'{client}_results_{run}_{test_case_without_extension}.txt',
                response)


if __name__ == '__main__':
    main()