-
Notifications
You must be signed in to change notification settings - Fork 1
/
Copy pathgenerate_load.py
55 lines (42 loc) · 1.72 KB
/
generate_load.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
49
50
51
52
53
54
55
import math
import random
import time
import requests
import sys
from typing import List
"""
Function to generate inter-arrival times using Poisson
distribution with exponential decay.
"""
def generate_inter_arrival_times(arrival_rate: int,
decay_rate: float,
duration: int) -> List[float]:
inter_arrival_times = []
time = 0
while time < duration:
# Calculate the arrival rate at the current time using exponential decay
current_rate = arrival_rate * math.exp(-decay_rate * time)
# Generate the inter-arrival time using the Poisson distribution
inter_arrival_time = random.expovariate(current_rate)
# Ensure larger values don't pollute the set.
if inter_arrival_time >= duration:
break
inter_arrival_times.append(inter_arrival_time)
time += inter_arrival_time
return inter_arrival_times
# Set up API calls...
# Configure inter arrival time and arrival rate semantics.
arrival_rate = 20 # Average arrival rate of 20 requests per second
decay_rate = 0.2 # Decay rate of 0.2 requests per second
duration = 120 # Duration of the load test in seconds
host, port = ('127.0.0.1', 18080)
# Override defaults if command-line arguments are provided
if len(sys.argv) > 2:
host, port = sys.argv[1], sys.argv[2]
inter_arrival_times = generate_inter_arrival_times(arrival_rate,
decay_rate,
duration)
for i, inter_arrival_time in enumerate(inter_arrival_times):
time.sleep(inter_arrival_time)
x = requests.get(f"http://{str(host)}:{str(port)}")
print(x.status_code)