Loadtest with Vegeta

loadtest.py

# Script for loadtesting a server with https://github.com/tsenart/vegeta
#
# Sample output. The server under test breaks down at 10k queries per second:
# its error rate goes from 0% to 23%, its 99th latency percentile goes
# from 0.488 milliseconds (at 1000 qps) to 71.18 ms (at 10000 qps).
#
# qps	errors	50th	95th	99th
# 1000	0.0	0.182	0.222	0.488
# 2000	0.0	0.144	0.224	0.431
# 3000	0.0	0.141	0.244	0.466
# [...]
# 9000	0.0	0.155	0.481	0.622
# 10000	23.0	17.901	51.785	71.18

from __future__ import print_function
import json, subprocess

QPS = list(range(1000, 30000, 1000))
VEGETA_BINARY = '/Users/sascha/go/bin/vegeta'
LOAD = """GET http://localhost:8080/collections/castles/items                   
"""



def run_loadtest(qps, load):
    command = [
        VEGETA_BINARY, 'attack',
        '-rate=%d/s' % qps, '-duration=10s', '-output=loadtest-%s' % qps]
    p = subprocess.Popen(command,
                         stdout=subprocess.PIPE, stdin=subprocess.PIPE,
                         stderr=subprocess.STDOUT)
    p.communicate(input=load)


def get_result(qps):
    """target-qps --> (error-rate, 50th, 95th, 99th percentile [ms])"""
    command = [
        VEGETA_BINARY, 'report', '-type=json', 'loadtest-%d' % qps]
    report = json.loads(subprocess.check_output(command))
    lat = report['latencies']
    return (qps, round(100.0 * (1.0 - report['success']), 1),
            round(lat['50th'] / 1e6, 3),
            round(lat['95th'] / 1e6, 3),
            round(lat['99th'] / 1e6, 3))


print('qps\terrors\t50th\t95th\t99th')
for qps in QPS:
    run_loadtest(qps, LOAD)
    r = get_result(qps)
    print('\t'.join([str(x) for x in r]))