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
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
|
#!/usr/bin/env ruby
# Copyright 2016, Google Inc.
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are
# met:
#
# * Redistributions of source code must retain the above copyright
# notice, this list of conditions and the following disclaimer.
# * Redistributions in binary form must reproduce the above
# copyright notice, this list of conditions and the following disclaimer
# in the documentation and/or other materials provided with the
# distribution.
# * Neither the name of Google Inc. nor the names of its
# contributors may be used to endorse or promote products derived from
# this software without specific prior written permission.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
# A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
# OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
# DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
# THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
require 'optparse'
require 'thread'
require_relative '../pb/test/client'
require_relative './metrics_server'
require_relative '../lib/grpc'
class QpsGauge < Gauge
@query_count
@query_mutex
@start_time
def initialize
@query_count = 0
@query_mutex = Mutex.new
@start_time = Time.now
end
def increment_queries
@query_mutex.synchronize { @query_count += 1}
end
def get_name
'qps'
end
def get_type
'long'
end
def get_value
(@query_mutex.synchronize { @query_count / (Time.now - @start_time) }).to_i
end
end
def start_metrics_server(port)
host = "0.0.0.0:#{port}"
server = GRPC::RpcServer.new
server.add_http2_port(host, :this_port_is_insecure)
service = MetricsServiceImpl.new
server.handle(service)
server_thread = Thread.new { server.run_till_terminated }
[server, service, server_thread]
end
StressArgs = Struct.new(:server_addresses, :test_cases, :duration,
:channels_per_server, :concurrent_calls, :metrics_port)
def start(stress_args)
running = true
threads = []
qps_gauge = QpsGauge.new
metrics_server, metrics_service, metrics_thread =
start_metrics_server(stress_args.metrics_port)
metrics_service.register_gauge(qps_gauge)
stress_args.server_addresses.each do |address|
stress_args.channels_per_server.times do
client_args = Args.new
client_args.host, client_args.port = address.split(':')
client_args.secure = false
client_args.test_case = ''
stub = create_stub(client_args)
named_tests = NamedTests.new(stub, client_args)
stress_args.concurrent_calls.times do
threads << Thread.new do
while running
named_tests.method(stress_args.test_cases.sample).call
qps_gauge.increment_queries
end
end
end
end
end
if stress_args.duration >= 0
sleep stress_args.duration
running = false
metrics_server.stop
p "QPS: #{qps_gauge.get_value}"
threads.each { |thd| thd.join; }
end
metrics_thread.join
end
def parse_stress_args
stress_args = StressArgs.new
stress_args.server_addresses = ['localhost:8080']
stress_args.test_cases = []
stress_args.duration = -1
stress_args.channels_per_server = 1
stress_args.concurrent_calls = 1
stress_args.metrics_port = '8081'
OptionParser.new do |opts|
opts.on('--server_addresses [LIST]', Array) do |addrs|
stress_args.server_addresses = addrs
end
opts.on('--test_cases cases', Array) do |cases|
stress_args.test_cases = (cases.map do |item|
split = item.split(':')
[split[0]] * split[1].to_i
end).reduce([], :+)
end
opts.on('--test_duration_secs [INT]', OptionParser::DecimalInteger) do |time|
stress_args.duration = time
end
opts.on('--num_channels_per_server [INT]', OptionParser::DecimalInteger) do |channels|
stress_args.channels_per_server = channels
end
opts.on('--num_stubs_per_channel [INT]', OptionParser::DecimalInteger) do |stubs|
stress_args.concurrent_calls = stubs
end
opts.on('--metrics_port [port]') do |port|
stress_args.metrics_port = port
end
end.parse!
stress_args
end
def main
opts = parse_stress_args
start(opts)
end
if __FILE__ == $0
main
end
|
