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
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
|
# Copyright (c) Authors of Clover
#
# All rights reserved. This program and the accompanying materials
# are made available under the terms of the Apache License, Version 2.0
# which accompanies this distribution, and is available at
# http://www.apache.org/licenses/LICENSE-2.0
from flask import Blueprint, request, Response
import grpc
import jmeter_pb2
import jmeter_pb2_grpc
import pickle
import logging
jmeter = Blueprint('jmeter', __name__)
grpc_port = '50054'
pod_name = 'clover-jmeter-master.default'
jmeter_grpc = pod_name + ':' + grpc_port
channel = grpc.insecure_channel(jmeter_grpc)
stub = jmeter_pb2_grpc.ControllerStub(channel)
@jmeter.route("/jmeter/gen", methods=['GET', 'POST'])
@jmeter.route("/jmeter/create", methods=['GET', 'POST'])
def gentest():
try:
p = request.json
u_list = []
u_names = []
u_methods = []
u_agents = []
try:
for u in p['url_list']:
u_list.append(u['url'])
u_names.append(u['name'])
u_methods.append(u['method'])
u_agents.append(u['user-agent'])
url_list = pickle.dumps(u_list)
url_names = pickle.dumps(u_names)
url_methods = pickle.dumps(u_methods)
url_agents = pickle.dumps(u_agents)
num_threads = p['load_spec']['num_threads']
ramp_time = p['load_spec']['ramp_time']
duration = p['load_spec']['duration']
loops = p['load_spec']['loops']
except (KeyError, ValueError) as e:
logging.debug(e)
return Response('Invalid value in test plan json/yaml', status=400)
response = stub.GenTest(jmeter_pb2.ConfigJmeter(
url_list=url_list, url_names=url_names, url_methods=url_methods,
num_threads=str(num_threads), ramp_time=str(ramp_time),
url_agents=url_agents, duration=str(duration), loops=str(loops)))
except Exception as e:
logging.debug(e)
if e.__class__.__name__ == "_Rendezvous":
return Response("Error connecting to jmeter via gRPC", status=400)
else:
return Response("Error generating test plan", status=400)
return response.message
@jmeter.route("/jmeter/start", methods=['GET', 'POST'])
def start():
try:
p = request.json
if not p:
slave_list = ''
num_slaves = '0'
else:
slave_list = p['slave_list']
num_slaves = p['num_slaves']
response = stub.StartTest(jmeter_pb2.TestParams(
num_slaves=num_slaves, test_plan='test.jmx',
slave_ips=slave_list))
except Exception as e:
logging.debug(e)
if e.__class__.__name__ == "_Rendezvous":
return Response("Error connecting to jmeter via gRPC", status=400)
else:
return Response("Error starting jmeter test", status=400)
return response.message
@jmeter.route("/jmeter/results/<r_type>", methods=['GET'])
def results(r_type):
try:
if not r_type:
r_file = 'results'
else:
r_file = r_type
response = stub.GetResults(jmeter_pb2.JResults(
r_format='csv', r_file=r_file))
except Exception as e:
logging.debug(e)
if e.__class__.__name__ == "_Rendezvous":
return Response("Error connecting to jmeter via gRPC", status=400)
else:
return Response("Error returning results", status=400)
return response.message
@jmeter.route("/jmeter/test")
def test():
return "<h1 style='color:blue'>Jmeter API Test Response</h1>"
|
