Skip to content

Instantly share code, notes, and snippets.

@laurenancona
Forked from mjumbewu/loadtest.py
Created October 7, 2016 21:16
Show Gist options
  • Save laurenancona/5240d9f85b9464911d53746d36db5213 to your computer and use it in GitHub Desktop.
Save laurenancona/5240d9f85b9464911d53746d36db5213 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