diff --git a/tools/gke/kubernetes_api.py b/tools/gke/kubernetes_api.py
index d14c26ad6ac5fd021ed72c6924a10ba2c0dda395..e1017e9da63189b95641d264e14b7f4e96c38840 100755
--- a/tools/gke/kubernetes_api.py
+++ b/tools/gke/kubernetes_api.py
@@ -50,7 +50,8 @@ def _make_pod_config(pod_name, image_name, container_port_list, cmd_list,
'name': pod_name,
'image': image_name,
'ports': [{'containerPort': port,
- 'protocol': 'TCP'} for port in container_port_list],
+ 'protocol': 'TCP'}
+ for port in container_port_list],
'imagePullPolicy': 'Always'
}
]
@@ -222,3 +223,47 @@ def delete_pod(kube_host, kube_port, namespace, pod_name):
del_url = 'http://%s:%d/api/v1/namespaces/%s/pods/%s' % (kube_host, kube_port,
namespace, pod_name)
return _do_delete(del_url, 'Delete Pod')
+
+
+def create_pod_and_service(kube_host, kube_port, namespace, pod_name,
+ image_name, container_port_list, cmd_list, arg_list,
+ env_dict, is_headless_service):
+ """A simple helper function that creates a pod and a service (if pod creation was successful)."""
+ is_success = create_pod(kube_host, kube_port, namespace, pod_name, image_name,
+ container_port_list, cmd_list, arg_list, env_dict)
+ if not is_success:
+ print 'Error in creating Pod'
+ return False
+
+ is_success = create_service(
+ kube_host,
+ kube_port,
+ namespace,
+ pod_name, # Use pod_name for service
+ pod_name,
+ container_port_list, # Service port list same as container port list
+ container_port_list,
+ is_headless_service)
+ if not is_success:
+ print 'Error in creating Service'
+ return False
+
+ print 'Successfully created the pod/service %s' % pod_name
+ return True
+
+
+def delete_pod_and_service(kube_host, kube_port, namespace, pod_name):
+ """ A simple helper function that calls delete_pod and delete_service """
+ is_success = delete_pod(kube_host, kube_port, namespace, pod_name)
+ if not is_success:
+ print 'Error in deleting pod %s' % pod_name
+ return False
+
+ # Note: service name assumed to the the same as pod name
+ is_success = delete_service(kube_host, kube_port, namespace, pod_name)
+ if not is_success:
+ print 'Error in deleting service %s' % pod_name
+ return False
+
+ print 'Successfully deleted the Pod/Service: %s' % pod_name
+ return True
diff --git a/tools/gke/run_stress_tests_on_gke.py b/tools/gke/run_stress_tests_on_gke.py
index 065b11e91c08b86c2e290b0ae1ae685333cce692..cf0ef595a6eba4141d21acb1aff0b89a48ec2f79 100755
--- a/tools/gke/run_stress_tests_on_gke.py
+++ b/tools/gke/run_stress_tests_on_gke.py
@@ -27,6 +27,7 @@
# 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.
+import argparse
import datetime
import os
import subprocess
@@ -40,17 +41,52 @@ from stress_test_utils import BigQueryHelper
import kubernetes_api
-GRPC_ROOT = os.path.abspath(os.path.join(os.path.dirname(sys.argv[0]), '../..'))
-os.chdir(GRPC_ROOT)
+_GRPC_ROOT = os.path.abspath(os.path.join(
+ os.path.dirname(sys.argv[0]), '../..'))
+os.chdir(_GRPC_ROOT)
+# num of seconds to wait for the GKE image to start and warmup
+_GKE_IMAGE_WARMUP_WAIT_SECS = 60
-class BigQuerySettings:
+_SERVER_POD_NAME = 'stress-server'
+_CLIENT_POD_NAME_PREFIX = 'stress-client'
+_DATASET_ID_PREFIX = 'stress_test'
+_SUMMARY_TABLE_ID = 'summary'
+_QPS_TABLE_ID = 'qps'
- def __init__(self, run_id, dataset_id, summary_table_id, qps_table_id):
- self.run_id = run_id
- self.dataset_id = dataset_id
- self.summary_table_id = summary_table_id
- self.qps_table_id = qps_table_id
+_DEFAULT_DOCKER_IMAGE_NAME = 'grpc_stress_test'
+
+# The default port on which the kubernetes proxy server is started on localhost
+# (i.e kubectl proxy --port=<port>)
+_DEFAULT_KUBERNETES_PROXY_PORT = 8001
+
+# How frequently should the stress client wrapper script (running inside a GKE
+# container) poll the health of the stress client (also running inside the GKE
+# container) and upload metrics to BigQuery
+_DEFAULT_STRESS_CLIENT_POLL_INTERVAL_SECS = 60
+
+# The default setting for stress test server and client
+_DEFAULT_STRESS_SERVER_PORT = 8080
+_DEFAULT_METRICS_PORT = 8081
+_DEFAULT_TEST_CASES_STR = 'empty_unary:1,large_unary:1,client_streaming:1,server_streaming:1,empty_stream:1'
+_DEFAULT_NUM_CHANNELS_PER_SERVER = 5
+_DEFAULT_NUM_STUBS_PER_CHANNEL = 10
+_DEFAULT_METRICS_COLLECTION_INTERVAL_SECS = 30
+
+# Number of stress client instances to launch
+_DEFAULT_NUM_CLIENTS = 3
+
+# How frequently should this test monitor the health of Stress clients and
+# Servers running in GKE
+_DEFAULT_TEST_POLL_INTERVAL_SECS = 60
+
+# Default run time for this test (2 hour)
+_DEFAULT_TEST_DURATION_SECS = 7200
+
+# The number of seconds it would take a GKE pod to warm up (i.e get to 'Running'
+# state from the time of creation). Ideally this is something the test should
+# automatically determine by using Kubernetes API to poll the pods status.
+_DEFAULT_GKE_WARMUP_SECS = 60
class KubernetesProxy:
@@ -76,11 +112,74 @@ class KubernetesProxy:
def __del__(self):
if self.p is not None:
+ print 'Shutting down Kubernetes proxy..'
self.p.kill()
+class TestSettings:
+
+ def __init__(self, build_docker_image, test_poll_interval_secs,
+ test_duration_secs, kubernetes_proxy_port):
+ self.build_docker_image = build_docker_image
+ self.test_poll_interval_secs = test_poll_interval_secs
+ self.test_duration_secs = test_duration_secs
+ self.kubernetes_proxy_port = kubernetes_proxy_port
+
+
+class GkeSettings:
+
+ def __init__(self, project_id, docker_image_name):
+ self.project_id = project_id
+ self.docker_image_name = docker_image_name
+ self.tag_name = 'gcr.io/%s/%s' % (project_id, docker_image_name)
+
+
+class BigQuerySettings:
+
+ def __init__(self, run_id, dataset_id, summary_table_id, qps_table_id):
+ self.run_id = run_id
+ self.dataset_id = dataset_id
+ self.summary_table_id = summary_table_id
+ self.qps_table_id = qps_table_id
+
+
+class StressServerSettings:
+
+ def __init__(self, server_pod_name, server_port):
+ self.server_pod_name = server_pod_name
+ self.server_port = server_port
+
+
+class StressClientSettings:
+
+ def __init__(self, num_clients, client_pod_name_prefix, server_pod_name,
+ server_port, metrics_port, metrics_collection_interval_secs,
+ stress_client_poll_interval_secs, num_channels_per_server,
+ num_stubs_per_channel, test_cases_str):
+ self.num_clients = num_clients
+ self.client_pod_name_prefix = client_pod_name_prefix
+ self.server_pod_name = server_pod_name
+ self.server_port = server_port
+ self.metrics_port = metrics_port
+ self.metrics_collection_interval_secs = metrics_collection_interval_secs
+ self.stress_client_poll_interval_secs = stress_client_poll_interval_secs
+ self.num_channels_per_server = num_channels_per_server
+ self.num_stubs_per_channel = num_stubs_per_channel
+ self.test_cases_str = test_cases_str
+
+ # == Derived properties ==
+ # Note: Client can accept a list of server addresses (a comma separated list
+ # of 'server_name:server_port'). In this case, we only have one server
+ # address to pass
+ self.server_addresses = '%s.default.svc.cluster.local:%d' % (
+ server_pod_name, server_port)
+ self.client_pod_names_list = ['%s-%d' % (client_pod_name_prefix, i)
+ for i in range(1, num_clients + 1)]
+
+
def _build_docker_image(image_name, tag_name):
""" Build the docker image and add a tag """
+ print 'Building docker image: %s' % image_name
os.environ['INTEROP_IMAGE'] = image_name
# Note that 'BASE_NAME' HAS to be 'grpc_interop_stress_cxx' since the script
# build_interop_stress_image.sh invokes the following script:
@@ -93,6 +192,7 @@ def _build_docker_image(image_name, tag_name):
print 'Error in building docker image'
return False
+ print 'Adding an additional tag %s to the image %s' % (tag_name, image_name)
cmd = ['docker', 'tag', '-f', image_name, tag_name]
p = subprocess.Popen(args=cmd)
retcode = p.wait()
@@ -115,144 +215,86 @@ def _push_docker_image_to_gke_registry(docker_tag_name):
return True
-def _launch_image_on_gke(kubernetes_api_server, kubernetes_api_port, namespace,
- pod_name, image_name, port_list, cmd_list, arg_list,
- env_dict, is_headless_service):
- """Creates a GKE Pod and a Service object for a given image by calling Kubernetes API"""
- is_success = kubernetes_api.create_pod(
- kubernetes_api_server,
- kubernetes_api_port,
- namespace,
- pod_name,
- image_name,
- port_list, # The ports to be exposed on this container/pod
- cmd_list, # The command that launches the stress server
- arg_list,
- env_dict # Environment variables to be passed to the pod
- )
- if not is_success:
- print 'Error in creating Pod'
- return False
-
- is_success = kubernetes_api.create_service(
- kubernetes_api_server,
- kubernetes_api_port,
- namespace,
- pod_name, # Use the pod name for service name as well
- pod_name,
- port_list, # Service port list
- port_list, # Container port list (same as service port list)
- is_headless_service)
- if not is_success:
- print 'Error in creating Service'
- return False
-
- print 'Successfully created the pod/service %s' % pod_name
- return True
-
-
-def _delete_image_on_gke(kubernetes_proxy, pod_name_list):
- """Deletes a GKE Pod and Service object for given list of Pods by calling Kubernetes API"""
- if not kubernetes_proxy.is_started:
- print 'Kubernetes proxy must be started before calling this function'
- return False
-
- is_success = True
- for pod_name in pod_name_list:
- is_success = kubernetes_api.delete_pod(
- 'localhost', kubernetes_proxy.get_port(), 'default', pod_name)
- if not is_success:
- print 'Error in deleting pod %s' % pod_name
- break
-
- is_success = kubernetes_api.delete_service(
- 'localhost', kubernetes_proxy.get_port(), 'default',
- pod_name) # service name same as pod name
- if not is_success:
- print 'Error in deleting service %s' % pod_name
- break
-
- if is_success:
- print 'Successfully deleted the Pods/Services: %s' % ','.join(pod_name_list)
-
- return is_success
-
-
-def _launch_server(gcp_project_id, docker_image_name, bq_settings,
- kubernetes_proxy, server_pod_name, server_port):
+def _launch_server(gke_settings, stress_server_settings, bq_settings,
+ kubernetes_proxy):
""" Launches a stress test server instance in GKE cluster """
if not kubernetes_proxy.is_started:
print 'Kubernetes proxy must be started before calling this function'
return False
+ # This is the wrapper script that is run in the container. This script runs
+ # the actual stress test server
server_cmd_list = [
'/var/local/git/grpc/tools/run_tests/stress_test/run_server.py'
- ] # Process that is launched
- server_arg_list = [] # run_server.py does not take any args (for now)
+ ]
- # == Parameters to the server process launched in GKE ==
+ # run_server.py does not take any args from the command line. The args are
+ # instead passed via environment variables (see server_env below)
+ server_arg_list = []
+
+ # The parameters to the script run_server.py are injected into the container
+ # via environment variables
server_env = {
'STRESS_TEST_IMAGE_TYPE': 'SERVER',
'STRESS_TEST_IMAGE': '/var/local/git/grpc/bins/opt/interop_server',
- 'STRESS_TEST_ARGS_STR': '--port=%s' % server_port,
+ 'STRESS_TEST_ARGS_STR': '--port=%s' % stress_server_settings.server_port,
'RUN_ID': bq_settings.run_id,
- 'POD_NAME': server_pod_name,
- 'GCP_PROJECT_ID': gcp_project_id,
+ 'POD_NAME': stress_server_settings.server_pod_name,
+ 'GCP_PROJECT_ID': gke_settings.project_id,
'DATASET_ID': bq_settings.dataset_id,
'SUMMARY_TABLE_ID': bq_settings.summary_table_id,
'QPS_TABLE_ID': bq_settings.qps_table_id
}
# Launch Server
- is_success = _launch_image_on_gke(
+ is_success = kubernetes_api.create_pod_and_service(
'localhost',
kubernetes_proxy.get_port(),
- 'default',
- server_pod_name,
- docker_image_name,
- [server_port], # Port that should be exposed on the container
+ 'default', # Use 'default' namespace
+ stress_server_settings.server_pod_name,
+ gke_settings.tag_name,
+ [stress_server_settings.server_port], # Port that should be exposed
server_cmd_list,
server_arg_list,
server_env,
- True # Headless = True for server. Since we want DNS records to be greated by GKE
+ True # Headless = True for server. Since we want DNS records to be created by GKE
)
return is_success
-def _launch_client(gcp_project_id, docker_image_name, bq_settings,
- kubernetes_proxy, num_instances, client_pod_name_prefix,
- server_pod_name, server_port):
+def _launch_client(gke_settings, stress_server_settings, stress_client_settings,
+ bq_settings, kubernetes_proxy):
""" Launches a configurable number of stress test clients on GKE cluster """
if not kubernetes_proxy.is_started:
print 'Kubernetes proxy must be started before calling this function'
return False
- server_address = '%s.default.svc.cluster.local:%d' % (server_pod_name,
- server_port)
- #TODO(sree) Make the whole client args configurable
- test_cases_str = 'empty_unary:1,large_unary:1'
stress_client_arg_list = [
- '--server_addresses=%s' % server_address,
- '--test_cases=%s' % test_cases_str, '--num_stubs_per_channel=10'
+ '--server_addresses=%s' % stress_client_settings.server_addresses,
+ '--test_cases=%s' % stress_client_settings.test_cases_str,
+ '--num_stubs_per_channel=%d' %
+ stress_client_settings.num_stubs_per_channel
]
+ # This is the wrapper script that is run in the container. This script runs
+ # the actual stress client
client_cmd_list = [
'/var/local/git/grpc/tools/run_tests/stress_test/run_client.py'
]
- # run_client.py takes no args. All args are passed as env variables
- client_arg_list = []
- # TODO(sree) Make this configurable (and also less frequent)
- poll_interval_secs = 30
+ # run_client.py takes no args. All args are passed as env variables (see
+ # client_env)
+ client_arg_list = []
- metrics_port = 8081
- metrics_server_address = 'localhost:%d' % metrics_port
+ metrics_server_address = 'localhost:%d' % stress_client_settings.metrics_port
metrics_client_arg_list = [
'--metrics_server_address=%s' % metrics_server_address,
'--total_only=true'
]
+ # The parameters to the script run_client.py are injected into the container
+ # via environment variables
client_env = {
'STRESS_TEST_IMAGE_TYPE': 'CLIENT',
'STRESS_TEST_IMAGE': '/var/local/git/grpc/bins/opt/stress_test',
@@ -260,27 +302,28 @@ def _launch_client(gcp_project_id, docker_image_name, bq_settings,
'METRICS_CLIENT_IMAGE': '/var/local/git/grpc/bins/opt/metrics_client',
'METRICS_CLIENT_ARGS_STR': ' '.join(metrics_client_arg_list),
'RUN_ID': bq_settings.run_id,
- 'POLL_INTERVAL_SECS': str(poll_interval_secs),
- 'GCP_PROJECT_ID': gcp_project_id,
+ 'POLL_INTERVAL_SECS':
+ str(stress_client_settings.stress_client_poll_interval_secs),
+ 'GCP_PROJECT_ID': gke_settings.project_id,
'DATASET_ID': bq_settings.dataset_id,
'SUMMARY_TABLE_ID': bq_settings.summary_table_id,
'QPS_TABLE_ID': bq_settings.qps_table_id
}
- for i in range(1, num_instances + 1):
- pod_name = '%s-%d' % (client_pod_name_prefix, i)
+ for pod_name in stress_client_settings.client_pod_names_list:
client_env['POD_NAME'] = pod_name
- is_success = _launch_image_on_gke(
- 'localhost',
+ is_success = kubernetes_api.create_pod_and_service(
+ 'localhost', # Since proxy is running on localhost
kubernetes_proxy.get_port(),
- 'default',
+ 'default', # default namespace
pod_name,
- docker_image_name,
- [metrics_port], # Client pods expose metrics port
+ gke_settings.tag_name,
+ [stress_client_settings.metrics_port
+ ], # Client pods expose metrics port
client_cmd_list,
client_arg_list,
client_env,
- False # Client is not a headless service.
+ False # Client is not a headless service
)
if not is_success:
print 'Error in launching client %s' % pod_name
@@ -289,20 +332,17 @@ def _launch_client(gcp_project_id, docker_image_name, bq_settings,
return True
-def _launch_server_and_client(bq_settings, gcp_project_id, docker_image_name,
- num_client_instances):
+def _launch_server_and_client(gke_settings, stress_server_settings,
+ stress_client_settings, bq_settings,
+ kubernetes_proxy_port):
# Start kubernetes proxy
- kubernetes_api_port = 9001
- kubernetes_proxy = KubernetesProxy(kubernetes_api_port)
+ print 'Kubernetes proxy'
+ kubernetes_proxy = KubernetesProxy(kubernetes_proxy_port)
kubernetes_proxy.start()
- # num of seconds to wait for the GKE image to start and warmup
- image_warmp_secs = 60
-
- server_pod_name = 'stress-server'
- server_port = 8080
- is_success = _launch_server(gcp_project_id, docker_image_name, bq_settings,
- kubernetes_proxy, server_pod_name, server_port)
+ print 'Launching server..'
+ is_success = _launch_server(gke_settings, stress_server_settings, bq_settings,
+ kubernetes_proxy)
if not is_success:
print 'Error in launching server'
return False
@@ -310,116 +350,217 @@ def _launch_server_and_client(bq_settings, gcp_project_id, docker_image_name,
# Server takes a while to start.
# TODO(sree) Use Kubernetes API to query the status of the server instead of
# sleeping
- print 'Waiting for %s seconds for the server to start...' % image_warmp_secs
- time.sleep(image_warmp_secs)
+ print 'Waiting for %s seconds for the server to start...' % _GKE_IMAGE_WARMUP_WAIT_SECS
+ time.sleep(_GKE_IMAGE_WARMUP_WAIT_SECS)
# Launch client
- server_address = '%s.default.svc.cluster.local:%d' % (server_pod_name,
- server_port)
client_pod_name_prefix = 'stress-client'
- is_success = _launch_client(gcp_project_id, docker_image_name, bq_settings,
- kubernetes_proxy, num_client_instances,
- client_pod_name_prefix, server_pod_name,
- server_port)
+ is_success = _launch_client(gke_settings, stress_server_settings,
+ stress_client_settings, bq_settings,
+ kubernetes_proxy)
+
if not is_success:
print 'Error in launching client(s)'
return False
- print 'Waiting for %s seconds for the client images to start...' % image_warmp_secs
- time.sleep(image_warmp_secs)
+ print 'Waiting for %s seconds for the client images to start...' % _GKE_IMAGE_WARMUP_WAIT_SECS
+ time.sleep(_GKE_IMAGE_WARMUP_WAIT_SECS)
return True
-def _delete_server_and_client(num_client_instances):
- kubernetes_api_port = 9001
- kubernetes_proxy = KubernetesProxy(kubernetes_api_port)
+def _delete_server_and_client(stress_server_settings, stress_client_settings,
+ kubernetes_proxy_port):
+ kubernetes_proxy = KubernetesProxy(kubernetes_proxy_port)
kubernetes_proxy.start()
# Delete clients first
- client_pod_names = ['stress-client-%d' % i
- for i in range(1, num_client_instances + 1)]
-
- is_success = _delete_image_on_gke(kubernetes_proxy, client_pod_names)
- if not is_success:
- return False
+ is_success = True
+ for pod_name in stress_client_settings.client_pod_names_list:
+ is_success = kubernetes_api.delete_pod_and_service(
+ 'localhost', kubernetes_proxy_port, 'default', pod_name)
+ if not is_success:
+ return False
# Delete server
- server_pod_name = 'stress-server'
- return _delete_image_on_gke(kubernetes_proxy, [server_pod_name])
+ is_success = kubernetes_api.delete_pod_and_service(
+ 'localhost', kubernetes_proxy_port, 'default',
+ stress_server_settings.server_pod_name)
+ return is_success
-def _build_and_push_docker_image(gcp_project_id, docker_image_name, tag_name):
- is_success = _build_docker_image(docker_image_name, tag_name)
- if not is_success:
- return False
- return _push_docker_image_to_gke_registry(tag_name)
-
+def run_test_main(test_settings, gke_settings, stress_server_settings,
+ stress_client_clients):
+ is_success = True
-# TODO(sree): This is just to test the above APIs. Rewrite this to make
-# everything configurable (like image names / number of instances etc)
-def run_test(skip_building_image, gcp_project_id, image_name, tag_name,
- num_client_instances, poll_interval_secs, total_duration_secs):
- if not skip_building_image:
- is_success = _build_docker_image(image_name, tag_name)
+ if test_settings.build_docker_image:
+ is_success = _build_docker_image(gke_settings.docker_image_name,
+ gke_settings.tag_name)
if not is_success:
return False
- is_success = _push_docker_image_to_gke_registry(tag_name)
+ is_success = _push_docker_image_to_gke_registry(gke_settings.tag_name)
if not is_success:
return False
- # == Big Query tables related settings (Common for both server and client) ==
-
# Create a unique id for this run (Note: Using timestamp instead of UUID to
# make it easier to deduce the date/time of the run just by looking at the run
# run id. This is useful in debugging when looking at records in Biq query)
run_id = datetime.datetime.now().strftime('%Y_%m_%d_%H_%M_%S')
- dataset_id = 'stress_test_%s' % run_id
- summary_table_id = 'summary'
- qps_table_id = 'qps'
- bq_settings = BigQuerySettings(run_id, dataset_id, summary_table_id,
- qps_table_id)
-
- bq_helper = BigQueryHelper(run_id, '', '', gcp_project_id, dataset_id,
- summary_table_id, qps_table_id)
+ dataset_id = '%s_%s' % (_DATASET_ID_PREFIX, run_id)
+
+ # Big Query settings (common for both Stress Server and Client)
+ bq_settings = BigQuerySettings(run_id, dataset_id, _SUMMARY_TABLE_ID,
+ _QPS_TABLE_ID)
+
+ bq_helper = BigQueryHelper(run_id, '', '', args.project_id, dataset_id,
+ _SUMMARY_TABLE_ID, _QPS_TABLE_ID)
bq_helper.initialize()
- is_success = _launch_server_and_client(bq_settings, gcp_project_id, tag_name,
- num_client_instances)
- if not is_success:
- return False
- start_time = datetime.datetime.now()
- end_time = start_time + datetime.timedelta(seconds=total_duration_secs)
-
- while True:
- if datetime.datetime.now() > end_time:
- print 'Test was run for %d seconds' % total_duration_secs
- break
-
- # Check if either stress server or clients have failed
- if bq_helper.check_if_any_tests_failed():
- is_success = False
- print 'Some tests failed.'
- break
- # Things seem to be running fine. Wait until next poll time to check the
- # status
- print 'Sleeping for %d seconds..' % poll_interval_secs
- time.sleep(poll_interval_secs)
-
- # Print BiqQuery tables
- bq_helper.print_summary_records()
- bq_helper.print_qps_records()
-
- _delete_server_and_client(num_client_instances)
+ try:
+ is_success = _launch_server_and_client(gke_settings, stress_server_settings,
+ stress_client_settings, bq_settings,
+ test_settings.kubernetes_proxy_port)
+ if not is_success:
+ return False
+
+ start_time = datetime.datetime.now()
+ end_time = start_time + datetime.timedelta(
+ seconds=test_settings.test_duration_secs)
+ print 'Running the test until %s' % end_time.isoformat()
+
+ while True:
+ if datetime.datetime.now() > end_time:
+ print 'Test was run for %d seconds' % test_settings.test_duration_secs
+ break
+
+ # Check if either stress server or clients have failed
+ if bq_helper.check_if_any_tests_failed():
+ is_success = False
+ print 'Some tests failed.'
+ break
+
+ # Things seem to be running fine. Wait until next poll time to check the
+ # status
+ print 'Sleeping for %d seconds..' % test_settings.test_poll_interval_secs
+ time.sleep(test_settings.test_poll_interval_secs)
+
+ # Print BiqQuery tables
+ bq_helper.print_summary_records()
+ bq_helper.print_qps_records()
+
+ finally:
+ # If is_success is False at this point, it means that the stress tests were
+ # started successfully but failed while running the tests. In this case we
+ # do should not delete the pods (since they contain all the failure
+ # information)
+ if is_success:
+ _delete_server_and_client(stress_server_settings, stress_client_settings,
+ test_settings.kubernetes_proxy_port)
+
return is_success
+argp = argparse.ArgumentParser(
+ description='Launch stress tests in GKE',
+ formatter_class=argparse.ArgumentDefaultsHelpFormatter)
+argp.add_argument('--project_id',
+ required=True,
+ help='The Google Cloud Platform Project Id')
+argp.add_argument('--num_clients',
+ default=1,
+ type=int,
+ help='Number of client instances to start')
+argp.add_argument('--docker_image_name',
+ default=_DEFAULT_DOCKER_IMAGE_NAME,
+ help='The name of the docker image containing stress client '
+ 'and stress servers')
+argp.add_argument('--build_docker_image',
+ dest='build_docker_image',
+ action='store_true',
+ help='Build a docker image and push to Google Container '
+ 'Registry')
+argp.add_argument('--do_not_build_docker_image',
+ dest='build_docker_image',
+ action='store_false',
+ help='Do not build and push docker image to Google Container '
+ 'Registry')
+argp.set_defaults(build_docker_image=True)
+
+argp.add_argument('--test_poll_interval_secs',
+ default=_DEFAULT_TEST_POLL_INTERVAL_SECS,
+ type=int,
+ help='How frequently should this script should monitor the '
+ 'health of stress clients and servers running in the GKE '
+ 'cluster')
+argp.add_argument('--test_duration_secs',
+ default=_DEFAULT_TEST_DURATION_SECS,
+ type=int,
+ help='How long should this test be run')
+argp.add_argument('--kubernetes_proxy_port',
+ default=_DEFAULT_KUBERNETES_PROXY_PORT,
+ type=int,
+ help='The port on which the kubernetes proxy (on localhost)'
+ ' is started')
+argp.add_argument('--stress_server_port',
+ default=_DEFAULT_STRESS_SERVER_PORT,
+ type=int,
+ help='The port on which the stress server (in GKE '
+ 'containers) listens')
+argp.add_argument('--stress_client_metrics_port',
+ default=_DEFAULT_METRICS_PORT,
+ type=int,
+ help='The port on which the stress clients (in GKE '
+ 'containers) expose metrics')
+argp.add_argument('--stress_client_poll_interval_secs',
+ default=_DEFAULT_STRESS_CLIENT_POLL_INTERVAL_SECS,
+ type=int,
+ help='How frequently should the stress client wrapper script'
+ ' running inside GKE should monitor health of the actual '
+ ' stress client process and upload the metrics to BigQuery')
+argp.add_argument('--stress_client_metrics_collection_interval_secs',
+ default=_DEFAULT_METRICS_COLLECTION_INTERVAL_SECS,
+ type=int,
+ help='How frequently should metrics be collected in-memory on'
+ ' the stress clients (running inside GKE containers). Note '
+ 'that this is NOT the same as the upload-to-BigQuery '
+ 'frequency. The metrics upload frequency is controlled by the'
+ ' --stress_client_poll_interval_secs flag')
+argp.add_argument('--stress_client_num_channels_per_server',
+ default=_DEFAULT_NUM_CHANNELS_PER_SERVER,
+ type=int,
+ help='The number of channels created to each server from a '
+ 'stress client')
+argp.add_argument('--stress_client_num_stubs_per_channel',
+ default=_DEFAULT_NUM_STUBS_PER_CHANNEL,
+ type=int,
+ help='The number of stubs created per channel. This number '
+ 'indicates the max number of RPCs that can be made in '
+ 'parallel on each channel at any given time')
+argp.add_argument('--stress_client_test_cases',
+ default=_DEFAULT_TEST_CASES_STR,
+ help='List of test cases (with weights) to be executed by the'
+ ' stress test client. The list is in the following format:\n'
+ ' <testcase_1:w_1,<test_case2:w_2>..<testcase_n:w_n>\n'
+ ' (Note: The weights do not have to add up to 100)')
+
if __name__ == '__main__':
- image_name = 'grpc_stress_test'
- gcp_project_id = 'sree-gce'
- tag_name = 'gcr.io/%s/%s' % (gcp_project_id, image_name)
- num_client_instances = 3
- poll_interval_secs = 10
- test_duration_secs = 150
- run_test(True, gcp_project_id, image_name, tag_name, num_client_instances,
- poll_interval_secs, test_duration_secs)
+ args = argp.parse_args()
+
+ test_settings = TestSettings(
+ args.build_docker_image, args.test_poll_interval_secs,
+ args.test_duration_secs, args.kubernetes_proxy_port)
+
+ gke_settings = GkeSettings(args.project_id, args.docker_image_name)
+
+ stress_server_settings = StressServerSettings(_SERVER_POD_NAME,
+ args.stress_server_port)
+ stress_client_settings = StressClientSettings(
+ args.num_clients, _CLIENT_POD_NAME_PREFIX, _SERVER_POD_NAME,
+ args.stress_server_port, args.stress_client_metrics_port,
+ args.stress_client_metrics_collection_interval_secs,
+ args.stress_client_poll_interval_secs,
+ args.stress_client_num_channels_per_server,
+ args.stress_client_num_stubs_per_channel, args.stress_client_test_cases)
+
+ run_test_main(test_settings, gke_settings, stress_server_settings,
+ stress_client_settings)