Load Testing with Locust

Rob McBroom

 

 

 

 

 

 

GitHub & App.net: Skurfer
Twitter: RobMcBroom

Locust

Define user behavior with Python code, and swarm your system with millions of simultaneous users.

Classes

Client

  • Understands how to talk to the service you're testing
  • Responsible for reporting stats and errors
  • Only HTTP supported by default

Locust

A simulated user

TaskSet

The behavior of your locusts

Basic Example (HTTP)

from locust import HttpLocust, TaskSet, task


class UserBehavior(TaskSet):
    @task
    def world(self):
        self.client.get('/world')

    @task
    def opinion(self):
        self.client.get('/opinion')

    @task
    def tech(self):
        self.client.get('/tech')


class WebsiteUser(HttpLocust):
    task_set = UserBehavior
    min_wait = 2000
    max_wait = 9000

Demo

$ locust --host http://www.cnn.com
$ locust --host http://www.foxnews.com

Things to Note

  • Stats reset after all users are spawned
  • It can be difficult to hit a specific number of requests
  • Locust doesn't act like a browser

Defining Tasks

Inside of a TaskSet…

Decorator

@task(3)
def tech(self):
    self.client.get('/tech')

Collection

tasks = [world, opinion, tech]

or

tasks = {world: 2, opinion: 2, tech: 5}

Scheduling

self.schedule_task(get_url, args=(url,))

Multiple Requests Per Task

@task
def doin_it_all(self):
    self.client.get('/world')
    self.client.get('/opinion')
    self.client.get('/tech')

On Start

Called once when a locust begins its task set.

def on_start(self):
    self.client.post(
        'https://some.example.app/login',
        {'login': 'iroh', 'password': 'White Lotus'},
    )

Other Features

Distributed Testing

Host A

$ locust --master

Host B, C, D, E…

$ locust --slave --master-ip=192.168.1.10

Nested Task Sets

class UserBehavior(TaskSet):
    tasks = {AnotherTaskSet: 5}

    @task
    def world(self):
        self.client.get('/world')

Custom Clients

import time
import dns.resolver
from locust import Locust, events

class DnsClient(object):
    def __init__(self, nameserver=None):
        self.resolver = dns.resolver.get_default_resolver()
        if nameserver is not None:
            self.resolver.nameservers = [nameserver]

    def query(self, qtext, qtype='A'):
        result = None
        start_time = time.time()
        try:
            result = self.resolver.query(qtext, qtype)
        except Exception as e:
            total_time = int((time.time() - start_time) * 1000)
            events.request_failure.fire(
                request_type=qtype, name=qtext,
                response_time=total_time, exception=e
            )
        else:
            total_time = int((time.time() - start_time) * 1000)
            length = len(result.response.to_text())
            events.request_success.fire(
                request_type=qtype, name=qtext,
                response_time=total_time, response_length=length
            )
        return result

Custom Clients (Continued)

class DnsLocust(Locust):
    def __init__(self, *args, **kwargs):
        super(DnsLocust, self).__init__(*args, **kwargs)
        self.client = DnsClient(nameserver=self.host)

Links