-
Notifications
You must be signed in to change notification settings - Fork 41
/
LocustGrafanaMonitoringExample.py
48 lines (39 loc) · 1.53 KB
/
LocustGrafanaMonitoringExample.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
from locust import HttpLocust, TaskSet, task
import locust.events
import time
import socket
import atexit
class FlightSearchTest(TaskSet):
@task
def open_login_page(self):
self.client.get("/login")
@task
def find_flight_between_Paris_and_Buenos_Aires(self):
self.client.post("/reserve.php", {
'fromPort': 'Paris', 'toPort': 'Buenos+Aires'
})
@task
def purchase_flight_between_Paris_and_Buenos_Aires(self):
self.client.post("/purchase.php", {
'fromPort': 'Paris', 'toPort': 'Buenos+Aires',
'airline': 'Virgin+America','flight': 43,
'price': 472.56
})
class MyLocust(HttpLocust):
task_set = FlightSearchTest
host = "http://blazedemo.com"
sock = None
def __init__(self):
super(MyLocust, self).__init__()
self.sock = socket.socket()
self.sock.connect( ("localhost", 2003) )
locust.events.request_success += self.hook_request_success
locust.events.request_failure += self.hook_request_fail
atexit.register(self.exit_handler)
def hook_request_success(self, request_type, name, response_time, response_length):
self.sock.send("%s %d %d\n" % ("performance." + name.replace('.', '-'), response_time, time.time()))
def hook_request_fail(self, request_type, name, response_time, exception):
self.request_fail_stats.append([name, request_type, response_time, exception])
def exit_handler(self):
self.sock.shutdown(socket.SHUT_RDWR)
self.sock.close()