diff options
Diffstat (limited to 'src/python/grpcio_tests/tests/stress')
| -rw-r--r-- | src/python/grpcio_tests/tests/stress/client.py | 214 | ||||
| -rw-r--r-- | src/python/grpcio_tests/tests/stress/metrics_server.py | 41 | ||||
| -rw-r--r-- | src/python/grpcio_tests/tests/stress/test_runner.py | 59 |
3 files changed, 164 insertions, 150 deletions
diff --git a/src/python/grpcio_tests/tests/stress/client.py b/src/python/grpcio_tests/tests/stress/client.py index 390ea13021..61f9e1c6b1 100644 --- a/src/python/grpcio_tests/tests/stress/client.py +++ b/src/python/grpcio_tests/tests/stress/client.py @@ -26,7 +26,6 @@ # 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. - """Entry point for running stress tests.""" import argparse @@ -46,115 +45,132 @@ from tests.stress import test_runner def _args(): - parser = argparse.ArgumentParser(description='gRPC Python stress test client') - parser.add_argument( - '--server_addresses', - help='comma seperated list of hostname:port to run servers on', - default='localhost:8080', type=str) - parser.add_argument( - '--test_cases', - help='comma seperated list of testcase:weighting of tests to run', - default='large_unary:100', - type=str) - parser.add_argument( - '--test_duration_secs', - help='number of seconds to run the stress test', - default=-1, type=int) - parser.add_argument( - '--num_channels_per_server', - help='number of channels per server', - default=1, type=int) - parser.add_argument( - '--num_stubs_per_channel', - help='number of stubs to create per channel', - default=1, type=int) - parser.add_argument( - '--metrics_port', - help='the port to listen for metrics requests on', - default=8081, type=int) - parser.add_argument( - '--use_test_ca', - help='Whether to use our fake CA. Requires --use_tls=true', - default=False, type=bool) - parser.add_argument( - '--use_tls', - help='Whether to use TLS', default=False, type=bool) - parser.add_argument( - '--server_host_override', default="foo.test.google.fr", - help='the server host to which to claim to connect', type=str) - return parser.parse_args() + parser = argparse.ArgumentParser( + description='gRPC Python stress test client') + parser.add_argument( + '--server_addresses', + help='comma seperated list of hostname:port to run servers on', + default='localhost:8080', + type=str) + parser.add_argument( + '--test_cases', + help='comma seperated list of testcase:weighting of tests to run', + default='large_unary:100', + type=str) + parser.add_argument( + '--test_duration_secs', + help='number of seconds to run the stress test', + default=-1, + type=int) + parser.add_argument( + '--num_channels_per_server', + help='number of channels per server', + default=1, + type=int) + parser.add_argument( + '--num_stubs_per_channel', + help='number of stubs to create per channel', + default=1, + type=int) + parser.add_argument( + '--metrics_port', + help='the port to listen for metrics requests on', + default=8081, + type=int) + parser.add_argument( + '--use_test_ca', + help='Whether to use our fake CA. Requires --use_tls=true', + default=False, + type=bool) + parser.add_argument( + '--use_tls', help='Whether to use TLS', default=False, type=bool) + parser.add_argument( + '--server_host_override', + default="foo.test.google.fr", + help='the server host to which to claim to connect', + type=str) + return parser.parse_args() def _test_case_from_arg(test_case_arg): - for test_case in methods.TestCase: - if test_case_arg == test_case.value: - return test_case - else: - raise ValueError('No test case {}!'.format(test_case_arg)) + for test_case in methods.TestCase: + if test_case_arg == test_case.value: + return test_case + else: + raise ValueError('No test case {}!'.format(test_case_arg)) def _parse_weighted_test_cases(test_case_args): - weighted_test_cases = {} - for test_case_arg in test_case_args.split(','): - name, weight = test_case_arg.split(':', 1) - test_case = _test_case_from_arg(name) - weighted_test_cases[test_case] = int(weight) - return weighted_test_cases + weighted_test_cases = {} + for test_case_arg in test_case_args.split(','): + name, weight = test_case_arg.split(':', 1) + test_case = _test_case_from_arg(name) + weighted_test_cases[test_case] = int(weight) + return weighted_test_cases + def _get_channel(target, args): - if args.use_tls: - if args.use_test_ca: - root_certificates = resources.test_root_certificates() + if args.use_tls: + if args.use_test_ca: + root_certificates = resources.test_root_certificates() + else: + root_certificates = None # will load default roots. + channel_credentials = grpc.ssl_channel_credentials( + root_certificates=root_certificates) + options = (( + 'grpc.ssl_target_name_override', + args.server_host_override,),) + channel = grpc.secure_channel( + target, channel_credentials, options=options) else: - root_certificates = None # will load default roots. - channel_credentials = grpc.ssl_channel_credentials( - root_certificates=root_certificates) - options = (('grpc.ssl_target_name_override', args.server_host_override,),) - return grpc.secure_channel( - target, channel_credentials, options=options) - else: - return grpc.insecure_channel(target) + channel = grpc.insecure_channel(target) + + # waits for the channel to be ready before we start sending messages + grpc.channel_ready_future(channel).result() + return channel + def run_test(args): - test_cases = _parse_weighted_test_cases(args.test_cases) - test_server_targets = args.server_addresses.split(',') - # Propagate any client exceptions with a queue - exception_queue = queue.Queue() - stop_event = threading.Event() - hist = histogram.Histogram(1, 1) - runners = [] - - server = grpc.server(futures.ThreadPoolExecutor(max_workers=25)) - metrics_pb2.add_MetricsServiceServicer_to_server( - metrics_server.MetricsServer(hist), server) - server.add_insecure_port('[::]:{}'.format(args.metrics_port)) - server.start() - - for test_server_target in test_server_targets: - for _ in xrange(args.num_channels_per_server): - channel = _get_channel(test_server_target, args) - for _ in xrange(args.num_stubs_per_channel): - stub = test_pb2.TestServiceStub(channel) - runner = test_runner.TestRunner(stub, test_cases, hist, - exception_queue, stop_event) - runners.append(runner) - - for runner in runners: - runner.start() - try: - timeout_secs = args.test_duration_secs - if timeout_secs < 0: - timeout_secs = None - raise exception_queue.get(block=True, timeout=timeout_secs) - except queue.Empty: - # No exceptions thrown, success - pass - finally: - stop_event.set() + test_cases = _parse_weighted_test_cases(args.test_cases) + test_server_targets = args.server_addresses.split(',') + # Propagate any client exceptions with a queue + exception_queue = queue.Queue() + stop_event = threading.Event() + hist = histogram.Histogram(1, 1) + runners = [] + + server = grpc.server(futures.ThreadPoolExecutor(max_workers=25)) + metrics_pb2.add_MetricsServiceServicer_to_server( + metrics_server.MetricsServer(hist), server) + server.add_insecure_port('[::]:{}'.format(args.metrics_port)) + server.start() + + for test_server_target in test_server_targets: + for _ in xrange(args.num_channels_per_server): + channel = _get_channel(test_server_target, args) + for _ in xrange(args.num_stubs_per_channel): + stub = test_pb2.TestServiceStub(channel) + runner = test_runner.TestRunner(stub, test_cases, hist, + exception_queue, stop_event) + runners.append(runner) + for runner in runners: - runner.join() - runner = None - server.stop(None) + runner.start() + try: + timeout_secs = args.test_duration_secs + if timeout_secs < 0: + timeout_secs = None + raise exception_queue.get(block=True, timeout=timeout_secs) + except queue.Empty: + # No exceptions thrown, success + pass + finally: + stop_event.set() + for runner in runners: + runner.join() + runner = None + server.stop(None) + if __name__ == '__main__': - run_test(_args()) + run_test(_args()) diff --git a/src/python/grpcio_tests/tests/stress/metrics_server.py b/src/python/grpcio_tests/tests/stress/metrics_server.py index 33dd1d6f2a..3a4cbc27ba 100644 --- a/src/python/grpcio_tests/tests/stress/metrics_server.py +++ b/src/python/grpcio_tests/tests/stress/metrics_server.py @@ -26,7 +26,6 @@ # 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. - """MetricsService for publishing stress test qps data.""" import time @@ -38,23 +37,23 @@ GAUGE_NAME = 'python_overall_qps' class MetricsServer(metrics_pb2.MetricsServiceServicer): - def __init__(self, histogram): - self._start_time = time.time() - self._histogram = histogram - - def _get_qps(self): - count = self._histogram.get_data().count - delta = time.time() - self._start_time - self._histogram.reset() - self._start_time = time.time() - return int(count/delta) - - def GetAllGauges(self, request, context): - qps = self._get_qps() - return [metrics_pb2.GaugeResponse(name=GAUGE_NAME, long_value=qps)] - - def GetGauge(self, request, context): - if request.name != GAUGE_NAME: - raise Exception('Gauge {} does not exist'.format(request.name)) - qps = self._get_qps() - return metrics_pb2.GaugeResponse(name=GAUGE_NAME, long_value=qps) + def __init__(self, histogram): + self._start_time = time.time() + self._histogram = histogram + + def _get_qps(self): + count = self._histogram.get_data().count + delta = time.time() - self._start_time + self._histogram.reset() + self._start_time = time.time() + return int(count / delta) + + def GetAllGauges(self, request, context): + qps = self._get_qps() + return [metrics_pb2.GaugeResponse(name=GAUGE_NAME, long_value=qps)] + + def GetGauge(self, request, context): + if request.name != GAUGE_NAME: + raise Exception('Gauge {} does not exist'.format(request.name)) + qps = self._get_qps() + return metrics_pb2.GaugeResponse(name=GAUGE_NAME, long_value=qps) diff --git a/src/python/grpcio_tests/tests/stress/test_runner.py b/src/python/grpcio_tests/tests/stress/test_runner.py index 88f13727e3..258abe9c21 100644 --- a/src/python/grpcio_tests/tests/stress/test_runner.py +++ b/src/python/grpcio_tests/tests/stress/test_runner.py @@ -26,7 +26,6 @@ # 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. - """Thread that sends random weighted requests on a TestService stub.""" import random @@ -36,38 +35,38 @@ import traceback def _weighted_test_case_generator(weighted_cases): - weight_sum = sum(weighted_cases.itervalues()) + weight_sum = sum(weighted_cases.itervalues()) - while True: - val = random.uniform(0, weight_sum) - partial_sum = 0 - for case in weighted_cases: - partial_sum += weighted_cases[case] - if val <= partial_sum: - yield case - break + while True: + val = random.uniform(0, weight_sum) + partial_sum = 0 + for case in weighted_cases: + partial_sum += weighted_cases[case] + if val <= partial_sum: + yield case + break class TestRunner(threading.Thread): - def __init__(self, stub, test_cases, hist, exception_queue, stop_event): - super(TestRunner, self).__init__() - self._exception_queue = exception_queue - self._stop_event = stop_event - self._stub = stub - self._test_cases = _weighted_test_case_generator(test_cases) - self._histogram = hist + def __init__(self, stub, test_cases, hist, exception_queue, stop_event): + super(TestRunner, self).__init__() + self._exception_queue = exception_queue + self._stop_event = stop_event + self._stub = stub + self._test_cases = _weighted_test_case_generator(test_cases) + self._histogram = hist - def run(self): - while not self._stop_event.is_set(): - try: - test_case = next(self._test_cases) - start_time = time.time() - test_case.test_interoperability(self._stub, None) - end_time = time.time() - self._histogram.add((end_time - start_time)*1e9) - except Exception as e: - traceback.print_exc() - self._exception_queue.put( - Exception("An exception occured during test {}" - .format(test_case), e)) + def run(self): + while not self._stop_event.is_set(): + try: + test_case = next(self._test_cases) + start_time = time.time() + test_case.test_interoperability(self._stub, None) + end_time = time.time() + self._histogram.add((end_time - start_time) * 1e9) + except Exception as e: + traceback.print_exc() + self._exception_queue.put( + Exception("An exception occured during test {}" + .format(test_case), e)) |