Skip to content

Instantly share code, notes, and snippets.

@mjumbewu
Last active October 9, 2016 00:58
Show Gist options
  • Save mjumbewu/686dd3e488bb5b699f7375bc83c23c36 to your computer and use it in GitHub Desktop.
Save mjumbewu/686dd3e488bb5b699f7375bc83c23c36 to your computer and use it in GitHub Desktop.
Simple load tester that continuously sends GET requests to a list of URLs for a given duration of time. Requires python libraries click and requests.
#!/usr/bin/env python
from __future__ import print_function, division
import click
import csv
from datetime import timedelta
from itertools import cycle
from queue import Queue
import requests
from threading import Thread
from time import sleep, time
import sys
def load_urls(urllistfile):
return cycle(url.strip() for url in urllistfile)
def log_requests(queue, logfile):
writer = csv.writer(logfile)
writer.writerow(['Address', 'Start Time', 'Duration', 'Response Code'])
while True:
request_data = queue.get()
if request_data is None:
break
writer.writerow(request_data)
def make_request(url, queue):
start_time = time()
response = requests.get(url)
request_duration = time() - start_time
queue.put([url, start_time, request_duration, response.status_code])
def do_load_test(duration, count, urllistfile, logfile):
average_step = (duration / count).total_seconds()
url_it = iter(load_urls(urllistfile))
response_queue = Queue()
logger_thread = Thread(target=lambda q=response_queue, f=logfile: log_requests(q, f))
logger_thread.start()
request_threads = []
for _ in range(count):
# start a new request in a thread
request_func = lambda a=next(url_it), q=response_queue: make_request(a, q)
thread = Thread(target=request_func)
thread.start()
request_threads.append(thread)
# wait until starting the next request
sleep(average_step)
# wait for all the threads to finish
for t in request_threads:
t.join()
logger_thread.join()
@click.command('loadtest')
@click.option('--hours', '-h', type=int, default=0)
@click.option('--minutes', '-m', type=int, default=0)
@click.option('--seconds', '-s', type=int, default=0)
@click.option('--numrequests', '-n', type=int, required=True,
help='The number of requests to send; these will be spread over '
'the duration of the test.')
@click.option('--urls', '-i', type=click.File('rU'), default=sys.stdin,
help='A file containing a list of URLs, one on each line. The '
'load tester will cycle through the addresses in this file. '
'[default: stdin].')
@click.option('--outfile', '-o', type=click.File('w'), default=sys.stdout,
help='The file to dump the results. [default: stdout].')
def _loadtest(hours, minutes, seconds, numrequests, urls, outfile):
'''
Simple load tester that continuously sends GET requests to a list of URLs
for a given duration of time.
'''
if not hours and not minutes and not seconds:
raise click.UsageError('You must specify some non-zero duration of '
'HOURS, MINUTES, and/or SECONDS.')
duration = timedelta(hours=hours, minutes=minutes, seconds=seconds)
do_load_test(duration, numrequests, urls, outfile)
if __name__ == '__main__':
_loadtest()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment