jcc-blockchain/scripts/devnet_simulator.py

#!/usr/bin/env python3
"""
devnet_simulator.py — deterministic IBFT 2.0 consensus + EVM execution simulation
Generates Prometheus-style metrics that can be scraped via node_exporter textfile
or pushed to a real Prometheus (when available).

Usage:
    python3 devnet_simulator.py --validators 4 --block-time 3 --duration 3600
"""

import argparse
import asyncio
import json
import logging
import random
import signal
import sys
import time
from collections import deque
from dataclasses import dataclass, field
from pathlib import Path
from typing import Deque, List

# Deterministic seed for reproducibility
random.seed(42)

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger("simulator")


@dataclass
class Validator:
    id: int
    address: str
    is_faulty: bool = False
    peers: int = 0
    tx_pool: int = 0
    head_block: int = 0
    head_tx_count: int = 0
    block_times: Deque[float] = field(default_factory=lambda: deque(maxlen=1000))
    total_tx: int = 0

    def reset_faulty(self):
        self.is_faulty = False


def ibft_extra_data(addresses: List[str]) -> str:
    """Generate a deterministic extraData string for the genesis."""
    vanity = b"\x00" * 32
    # RLP list of addresses (20 bytes each)
    items = b"".join([bytes.fromhex(a.replace("0x", "")) for a in addresses])
    # RLP list prefix
    if len(items) <= 55:
        rlp_list = bytes([0xC0 + len(items)]) + items
    else:
        length_bytes = len(items).to_bytes((len(items).bit_length() + 7) // 8, "big")
        rlp_list = bytes([0xF7 + len(length_bytes)]) + length_bytes + items
    round_bytes = b"\x00" * 8
    seal = b"\x00" * 96
    extra = vanity + rlp_list + round_bytes + seal
    return "0x" + extra.hex()


class BesuDevnetSimulator:
    def __init__(self, validators: int, block_time: float, duration: float, fault_fraction: float = 0.0):
        self.block_time = block_time
        self.duration = duration
        self.fault_fraction = fault_fraction
        self.addresses = [f"0x{'%040x' % i}" for i in range(validators)]
        self.extra_data = ibft_extra_data(self.addresses)
        self.validators: List[Validator] = [
            Validator(id=i, address=self.addresses[i], is_faulty=(random.random() < fault_fraction))
            for i in range(validators)
        ]
        self.genesis = {
            "config": {
                "chainId": 13371337,
                "ibft2": {
                    "blockperiodseconds": int(block_time),
                    "epochlength": 30000,
                    "requesttimeoutseconds": max(2, int(block_time)),
                }
            },
            "extraData": self.extra_data,
            "alloc": {},
        }
        self.running = False
        self.started_at: float = 0.0
        self.metrics_dir = Path("monitoring/prometheus/metrics")
        self.metrics_dir.mkdir(parents=True, exist_ok=True)

        # Counter contract simulation state
        self.counter_value = 0
        self.counter_owner = self.addresses[0]

    async def _simulate_block(self, proposer: Validator, ts: float) -> None:
        # Simulate block production time with small jitter
        jitter = random.gauss(0, 0.15)
        actual_block_time = max(self.block_time + jitter, 0.1)
        proposer.block_times.append(actual_block_time)
        proposer.head_block += 1
        # Simulate tx inclusion
        tx_in_block = random.randint(0, int(600 * self.block_time))  # targeting peak TPS
        proposer.head_tx_count = tx_in_block
        proposer.total_tx += tx_in_block
        proposer.tx_pool = max(0, proposer.tx_pool - tx_in_block + random.randint(0, 10))

        # Simulate simple consensus log line
        logger.info("component=consensus node=%s block=%d txs=%d time=%.3fs",
                    proposer.address[:10], proposer.head_block, tx_in_block, actual_block_time)

        # Simulate execution layer (increment counter)
        if random.random() < 0.1:
            self.counter_value += 1

        await asyncio.sleep(actual_block_time)

    async def _update_metrics(self) -> None:
        """Write Prometheus exposition format to textfile collector."""
        lines: List[str] = []
        for v in self.validators:
            labels = f'node="validator-{v.id}",component="consensus"'
            lines.append(f'ethereum_blockchain_height{{{labels}}} {v.head_block}')
            lines.append(f'ethereum_peer_count{{{labels}}} {v.peers}')
            lines.append(f'ethereum_blockchain_transaction_pool_size{{{labels}}} {v.tx_pool}')
            lines.append(f'ethereum_blockchain_head_transactions_added_total{{{labels}}} {v.total_tx}')
            if v.block_times:
                for bucket in [0.5, 1.0, 2.0, 3.0, 4.0, 5.0, 10.0]:
                    count = sum(1 for t in v.block_times if t <= bucket)
                    lines.append(
                        f'ethereum_blockchain_new_head_block_time_seconds_bucket{{{labels},le="{bucket}"}} {count}'
                    )
                lines.append(
                    f'ethereum_blockchain_new_head_block_time_seconds_sum{{{labels}}} {sum(v.block_times)}'
                )
                lines.append(
                    f'ethereum_blockchain_new_head_block_time_seconds_count{{{labels}}} {len(v.block_times)}'
                )
        # Write atomically by temp + rename
        out = self.metrics_dir / "besu.prom"
        tmp = self.metrics_dir / "besu.prom.tmp"
        tmp.write_text("\n".join(lines) + "\n")
        tmp.rename(out)

    async def run(self) -> None:
        self.running = True
        self.started_at = time.monotonic()
        proposer_idx = 0
        logger.info("Starting IBFT devnet simulation: validators=%d block_time=%.1fs duration=%.0fs",
                    len(self.validators), self.block_time, self.duration)
        while self.running:
            elapsed = time.monotonic() - self.started_at
            if elapsed >= self.duration:
                break
            proposer = self.validators[proposer_idx % len(self.validators)]
            if proposer.is_faulty:
                logger.warning("node=%s is faulty; skipping proposal this round", proposer.address[:10])
                # BFT safety: faulty nodes are tolerated as long as minority
                proposer_idx += 1
                await asyncio.sleep(self.block_time)
                continue
            await self._simulate_block(proposer, time.time())
            proposer_idx += 1
            if proposer_idx % 30 == 0:
                await self._update_metrics()
        await self._update_metrics()
        logger.info("Simulation complete after %.1fs", time.monotonic() - self.started_at)

    def signal_handler(self, _sig, _frame):
        self.running = False
        logger.info("Shutdown requested, finishing current block...")


async def main():
    parser = argparse.ArgumentParser(description="Besu IBFT 2.0 devnet simulator")
    parser.add_argument("--validators", type=int, default=4)
    parser.add_argument("--block-time", type=float, default=3.0, help="Target block time in seconds")
    parser.add_argument("--duration", type=float, default=3600.0, help="Simulation duration in seconds")
    parser.add_argument("--fault-fraction", type=float, default=0.0, help="Fraction of validators to mark faulty")
    args = parser.parse_args()

    sim = BesuDevnetSimulator(args.validators, args.block_time, args.duration, args.fault_fraction)
    signal.signal(signal.SIGINT, sim.signal_handler)
    signal.signal(signal.SIGTERM, sim.signal_handler)
    await sim.run()

    # Print summary
    print("\n=== Simulation Summary ===")
    for v in sim.validators:
        avg_bt = sum(v.block_times) / len(v.block_times) if v.block_times else 0.0
        print(f"  validator-{v.id}: blocks={v.head_block} avg_block_time={avg_bt:.3f}s total_tx={v.total_tx}")
    with open(sim.metrics_dir / "besu.prom") as f:
        print("\n=== Prometheus metrics snippet (first 10 lines) ===")
        for line in list(f)[:10]:
            print("  ", line.strip())


if __name__ == "__main__":
    asyncio.run(main())