In this blog, I am going to share a simple code snippet that I have used to run a load tests on geth node. To get things done, I am using the locust python library, and to connect to get node, I am using web3 library. This piece of code finds 10 addresses at the initialization step. After initialization, it spawns a predefined number of users and tries to find the balance of 1 of those 10 addresses.
from functools import wraps
from time import time
from locust import events, task
from locust.user.users import User
from web3 import Web3
import random
def geth_locust_task(f):
'''
Simple timing wrapper which fires off the necessary
success and failure events for locust.
'''
@wraps(f)
def wrapped(*args, **kwargs):
start_time = time()
try:
result = f(*args, **kwargs)
print(result)
except Exception as e:
print('Exception in {}'.format(f.__name__))
total_time = int((time() - start_time) * 1000)
events.request_failure.fire(
request_type="web3",
name=f.__name__,
response_time=total_time,
exception=e)
return False
else:
total_time = int((time() - start_time) * 1000)
events.request_success.fire(
request_type="web3",
name=f.__name__,
response_time=total_time,
response_length=0)
return result
return wrapped
class GethUser(User):
def __init__(self, *args, **kwargs):
super(GethUser, self).__init__(*args, **kwargs)
server = args[0].host
self.client = Web3(Web3.HTTPProvider(server))
c=1
block='latest'
self.addresses = []
while c<10:
tnx = self.client.eth.get_block(block)
self.addresses.append(tnx.miner)
block = tnx.parentHash
c+=1
print(self.addresses)
@geth_locust_task
@task
def get_balance(self):
target_addr = random.choice(self.addresses)
bal = self.client.eth.get_balance(target_addr)
return bal
Then you can run the tests using `locust -f locustfile.py`. It would run a server on port 8089. On browser you have to open localhost:8089 and then you have to provide the number of users you want to spawn, the spawn rate and the host of your geth node.
`