| #!/usr/bin/env python3 |
| # Copyright 2016 gRPC authors. |
| # |
| # Licensed under the Apache License, Version 2.0 (the "License"); |
| # you may not use this file except in compliance with the License. |
| # You may obtain a copy of the License at |
| # |
| # http://www.apache.org/licenses/LICENSE-2.0 |
| # |
| # Unless required by applicable law or agreed to in writing, software |
| # distributed under the License is distributed on an "AS IS" BASIS, |
| # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| # See the License for the specific language governing permissions and |
| # limitations under the License. |
| """Run performance tests locally or remotely.""" |
| |
| from __future__ import print_function |
| |
| import argparse |
| import collections |
| import itertools |
| import json |
| import os |
| import re |
| import shlex |
| import sys |
| import time |
| |
| import six |
| |
| import performance.scenario_config as scenario_config |
| import python_utils.jobset as jobset |
| import python_utils.report_utils as report_utils |
| |
| _ROOT = os.path.abspath(os.path.join(os.path.dirname(sys.argv[0]), "../..")) |
| os.chdir(_ROOT) |
| |
| _REMOTE_HOST_USERNAME = "jenkins" |
| |
| _SCENARIO_TIMEOUT = 3 * 60 |
| _WORKER_TIMEOUT = 3 * 60 |
| _NETPERF_TIMEOUT = 60 |
| _QUIT_WORKER_TIMEOUT = 2 * 60 |
| |
| |
| class QpsWorkerJob: |
| """Encapsulates a qps worker server job.""" |
| |
| def __init__(self, spec, language, host_and_port, perf_file_base_name=None): |
| self._spec = spec |
| self.language = language |
| self.host_and_port = host_and_port |
| self._job = None |
| self.perf_file_base_name = perf_file_base_name |
| |
| def start(self): |
| self._job = jobset.Job( |
| self._spec, newline_on_success=True, travis=True, add_env={} |
| ) |
| |
| def is_running(self): |
| """Polls a job and returns True if given job is still running.""" |
| return self._job and self._job.state() == jobset._RUNNING |
| |
| def kill(self): |
| if self._job: |
| self._job.kill() |
| self._job = None |
| |
| |
| def create_qpsworker_job( |
| language, shortname=None, port=10000, remote_host=None, perf_cmd=None |
| ): |
| cmdline = language.worker_cmdline() + ["--driver_port=%s" % port] |
| |
| if remote_host: |
| host_and_port = "%s:%s" % (remote_host, port) |
| else: |
| host_and_port = "localhost:%s" % port |
| |
| perf_file_base_name = None |
| if perf_cmd: |
| perf_file_base_name = "%s-%s" % (host_and_port, shortname) |
| # specify -o output file so perf.data gets collected when worker stopped |
| cmdline = ( |
| perf_cmd + ["-o", "%s-perf.data" % perf_file_base_name] + cmdline |
| ) |
| |
| worker_timeout = _WORKER_TIMEOUT |
| if remote_host: |
| user_at_host = "%s@%s" % (_REMOTE_HOST_USERNAME, remote_host) |
| ssh_cmd = ["ssh"] |
| cmdline = ["timeout", "%s" % (worker_timeout + 30)] + cmdline |
| ssh_cmd.extend( |
| [ |
| str(user_at_host), |
| "cd ~/performance_workspace/grpc/ && %s" % " ".join(cmdline), |
| ] |
| ) |
| cmdline = ssh_cmd |
| |
| jobspec = jobset.JobSpec( |
| cmdline=cmdline, |
| shortname=shortname, |
| timeout_seconds=worker_timeout, # workers get restarted after each scenario |
| verbose_success=True, |
| ) |
| return QpsWorkerJob(jobspec, language, host_and_port, perf_file_base_name) |
| |
| |
| def create_scenario_jobspec( |
| scenario_json, |
| workers, |
| remote_host=None, |
| bq_result_table=None, |
| server_cpu_load=0, |
| ): |
| """Runs one scenario using QPS driver.""" |
| # setting QPS_WORKERS env variable here makes sure it works with SSH too. |
| cmd = 'QPS_WORKERS="%s" ' % ",".join(workers) |
| if bq_result_table: |
| cmd += 'BQ_RESULT_TABLE="%s" ' % bq_result_table |
| cmd += "tools/run_tests/performance/run_qps_driver.sh " |
| cmd += "--scenarios_json=%s " % shlex.quote( |
| json.dumps({"scenarios": [scenario_json]}) |
| ) |
| cmd += "--scenario_result_file=scenario_result.json " |
| if server_cpu_load != 0: |
| cmd += ( |
| "--search_param=offered_load --initial_search_value=1000" |
| " --targeted_cpu_load=%d --stride=500 --error_tolerance=0.01" |
| % server_cpu_load |
| ) |
| if remote_host: |
| user_at_host = "%s@%s" % (_REMOTE_HOST_USERNAME, remote_host) |
| cmd = 'ssh %s "cd ~/performance_workspace/grpc/ && "%s' % ( |
| user_at_host, |
| shlex.quote(cmd), |
| ) |
| |
| return jobset.JobSpec( |
| cmdline=[cmd], |
| shortname="%s" % scenario_json["name"], |
| timeout_seconds=_SCENARIO_TIMEOUT, |
| shell=True, |
| verbose_success=True, |
| ) |
| |
| |
| def create_quit_jobspec(workers, remote_host=None): |
| """Runs quit using QPS driver.""" |
| # setting QPS_WORKERS env variable here makes sure it works with SSH too. |
| cmd = 'QPS_WORKERS="%s" cmake/build/qps_json_driver --quit' % ",".join( |
| w.host_and_port for w in workers |
| ) |
| if remote_host: |
| user_at_host = "%s@%s" % (_REMOTE_HOST_USERNAME, remote_host) |
| cmd = 'ssh %s "cd ~/performance_workspace/grpc/ && "%s' % ( |
| user_at_host, |
| shlex.quote(cmd), |
| ) |
| |
| return jobset.JobSpec( |
| cmdline=[cmd], |
| shortname="shutdown_workers", |
| timeout_seconds=_QUIT_WORKER_TIMEOUT, |
| shell=True, |
| verbose_success=True, |
| ) |
| |
| |
| def create_netperf_jobspec( |
| server_host="localhost", client_host=None, bq_result_table=None |
| ): |
| """Runs netperf benchmark.""" |
| cmd = 'NETPERF_SERVER_HOST="%s" ' % server_host |
| if bq_result_table: |
| cmd += 'BQ_RESULT_TABLE="%s" ' % bq_result_table |
| if client_host: |
| # If netperf is running remotely, the env variables populated by Jenkins |
| # won't be available on the client, but we need them for uploading results |
| # to BigQuery. |
| jenkins_job_name = os.getenv("KOKORO_JOB_NAME") |
| if jenkins_job_name: |
| cmd += 'KOKORO_JOB_NAME="%s" ' % jenkins_job_name |
| jenkins_build_number = os.getenv("KOKORO_BUILD_NUMBER") |
| if jenkins_build_number: |
| cmd += 'KOKORO_BUILD_NUMBER="%s" ' % jenkins_build_number |
| |
| cmd += "tools/run_tests/performance/run_netperf.sh" |
| if client_host: |
| user_at_host = "%s@%s" % (_REMOTE_HOST_USERNAME, client_host) |
| cmd = 'ssh %s "cd ~/performance_workspace/grpc/ && "%s' % ( |
| user_at_host, |
| shlex.quote(cmd), |
| ) |
| |
| return jobset.JobSpec( |
| cmdline=[cmd], |
| shortname="netperf", |
| timeout_seconds=_NETPERF_TIMEOUT, |
| shell=True, |
| verbose_success=True, |
| ) |
| |
| |
| def archive_repo(languages): |
| """Archives local version of repo including submodules.""" |
| cmdline = ["tar", "-cf", "../grpc.tar", "../grpc/"] |
| if "java" in languages: |
| cmdline.append("../grpc-java") |
| if "go" in languages: |
| cmdline.append("../grpc-go") |
| if "node" in languages: |
| cmdline.append("../grpc-node") |
| |
| archive_job = jobset.JobSpec( |
| cmdline=cmdline, shortname="archive_repo", timeout_seconds=3 * 60 |
| ) |
| |
| jobset.message("START", "Archiving local repository.", do_newline=True) |
| num_failures, _ = jobset.run( |
| [archive_job], newline_on_success=True, maxjobs=1 |
| ) |
| if num_failures == 0: |
| jobset.message( |
| "SUCCESS", |
| "Archive with local repository created successfully.", |
| do_newline=True, |
| ) |
| else: |
| jobset.message( |
| "FAILED", "Failed to archive local repository.", do_newline=True |
| ) |
| sys.exit(1) |
| |
| |
| def prepare_remote_hosts(hosts, prepare_local=False): |
| """Prepares remote hosts (and maybe prepare localhost as well).""" |
| prepare_timeout = 10 * 60 |
| prepare_jobs = [] |
| for host in hosts: |
| user_at_host = "%s@%s" % (_REMOTE_HOST_USERNAME, host) |
| prepare_jobs.append( |
| jobset.JobSpec( |
| cmdline=["tools/run_tests/performance/remote_host_prepare.sh"], |
| shortname="remote_host_prepare.%s" % host, |
| environ={"USER_AT_HOST": user_at_host}, |
| timeout_seconds=prepare_timeout, |
| ) |
| ) |
| if prepare_local: |
| # Prepare localhost as well |
| prepare_jobs.append( |
| jobset.JobSpec( |
| cmdline=["tools/run_tests/performance/kill_workers.sh"], |
| shortname="local_prepare", |
| timeout_seconds=prepare_timeout, |
| ) |
| ) |
| jobset.message("START", "Preparing hosts.", do_newline=True) |
| num_failures, _ = jobset.run( |
| prepare_jobs, newline_on_success=True, maxjobs=10 |
| ) |
| if num_failures == 0: |
| jobset.message( |
| "SUCCESS", "Prepare step completed successfully.", do_newline=True |
| ) |
| else: |
| jobset.message( |
| "FAILED", "Failed to prepare remote hosts.", do_newline=True |
| ) |
| sys.exit(1) |
| |
| |
| def build_on_remote_hosts( |
| hosts, languages=list(scenario_config.LANGUAGES.keys()), build_local=False |
| ): |
| """Builds performance worker on remote hosts (and maybe also locally).""" |
| build_timeout = 45 * 60 |
| # Kokoro VMs (which are local only) do not have caching, so they need more time to build |
| local_build_timeout = 60 * 60 |
| build_jobs = [] |
| for host in hosts: |
| user_at_host = "%s@%s" % (_REMOTE_HOST_USERNAME, host) |
| build_jobs.append( |
| jobset.JobSpec( |
| cmdline=["tools/run_tests/performance/remote_host_build.sh"] |
| + languages, |
| shortname="remote_host_build.%s" % host, |
| environ={"USER_AT_HOST": user_at_host, "CONFIG": "opt"}, |
| timeout_seconds=build_timeout, |
| ) |
| ) |
| if build_local: |
| # start port server locally |
| build_jobs.append( |
| jobset.JobSpec( |
| cmdline=["python", "tools/run_tests/start_port_server.py"], |
| shortname="local_start_port_server", |
| timeout_seconds=2 * 60, |
| ) |
| ) |
| # Build locally as well |
| build_jobs.append( |
| jobset.JobSpec( |
| cmdline=["tools/run_tests/performance/build_performance.sh"] |
| + languages, |
| shortname="local_build", |
| environ={"CONFIG": "opt"}, |
| timeout_seconds=local_build_timeout, |
| ) |
| ) |
| jobset.message("START", "Building.", do_newline=True) |
| num_failures, _ = jobset.run( |
| build_jobs, newline_on_success=True, maxjobs=10 |
| ) |
| if num_failures == 0: |
| jobset.message("SUCCESS", "Built successfully.", do_newline=True) |
| else: |
| jobset.message("FAILED", "Build failed.", do_newline=True) |
| sys.exit(1) |
| |
| |
| def create_qpsworkers(languages, worker_hosts, perf_cmd=None): |
| """Creates QPS workers (but does not start them).""" |
| if not worker_hosts: |
| # run two workers locally (for each language) |
| workers = [(None, 10000), (None, 10010)] |
| elif len(worker_hosts) == 1: |
| # run two workers on the remote host (for each language) |
| workers = [(worker_hosts[0], 10000), (worker_hosts[0], 10010)] |
| else: |
| # run one worker per each remote host (for each language) |
| workers = [(worker_host, 10000) for worker_host in worker_hosts] |
| |
| return [ |
| create_qpsworker_job( |
| language, |
| shortname="qps_worker_%s_%s" % (language, worker_idx), |
| port=worker[1] + language.worker_port_offset(), |
| remote_host=worker[0], |
| perf_cmd=perf_cmd, |
| ) |
| for language in languages |
| for worker_idx, worker in enumerate(workers) |
| ] |
| |
| |
| def perf_report_processor_job( |
| worker_host, perf_base_name, output_filename, flame_graph_reports |
| ): |
| print("Creating perf report collection job for %s" % worker_host) |
| cmd = "" |
| if worker_host != "localhost": |
| user_at_host = "%s@%s" % (_REMOTE_HOST_USERNAME, worker_host) |
| cmd = ( |
| "USER_AT_HOST=%s OUTPUT_FILENAME=%s OUTPUT_DIR=%s PERF_BASE_NAME=%s" |
| " tools/run_tests/performance/process_remote_perf_flamegraphs.sh" |
| % ( |
| user_at_host, |
| output_filename, |
| flame_graph_reports, |
| perf_base_name, |
| ) |
| ) |
| else: |
| cmd = ( |
| "OUTPUT_FILENAME=%s OUTPUT_DIR=%s PERF_BASE_NAME=%s" |
| " tools/run_tests/performance/process_local_perf_flamegraphs.sh" |
| % (output_filename, flame_graph_reports, perf_base_name) |
| ) |
| |
| return jobset.JobSpec( |
| cmdline=cmd, |
| timeout_seconds=3 * 60, |
| shell=True, |
| verbose_success=True, |
| shortname="process perf report", |
| ) |
| |
| |
| Scenario = collections.namedtuple("Scenario", "jobspec workers name") |
| |
| |
| def create_scenarios( |
| languages, |
| workers_by_lang, |
| remote_host=None, |
| regex=".*", |
| category="all", |
| bq_result_table=None, |
| netperf=False, |
| netperf_hosts=[], |
| server_cpu_load=0, |
| ): |
| """Create jobspecs for scenarios to run.""" |
| all_workers = [ |
| worker |
| for workers in list(workers_by_lang.values()) |
| for worker in workers |
| ] |
| scenarios = [] |
| _NO_WORKERS = [] |
| |
| if netperf: |
| if not netperf_hosts: |
| netperf_server = "localhost" |
| netperf_client = None |
| elif len(netperf_hosts) == 1: |
| netperf_server = netperf_hosts[0] |
| netperf_client = netperf_hosts[0] |
| else: |
| netperf_server = netperf_hosts[0] |
| netperf_client = netperf_hosts[1] |
| scenarios.append( |
| Scenario( |
| create_netperf_jobspec( |
| server_host=netperf_server, |
| client_host=netperf_client, |
| bq_result_table=bq_result_table, |
| ), |
| _NO_WORKERS, |
| "netperf", |
| ) |
| ) |
| |
| for language in languages: |
| for scenario_json in language.scenarios(): |
| if re.search(regex, scenario_json["name"]): |
| categories = scenario_json.get( |
| "CATEGORIES", ["scalable", "smoketest"] |
| ) |
| if category in categories or category == "all": |
| workers = workers_by_lang[str(language)][:] |
| # 'SERVER_LANGUAGE' is an indicator for this script to pick |
| # a server in different language. |
| custom_server_lang = scenario_json.get( |
| "SERVER_LANGUAGE", None |
| ) |
| custom_client_lang = scenario_json.get( |
| "CLIENT_LANGUAGE", None |
| ) |
| scenario_json = scenario_config.remove_nonproto_fields( |
| scenario_json |
| ) |
| if custom_server_lang and custom_client_lang: |
| raise Exception( |
| "Cannot set both custom CLIENT_LANGUAGE and" |
| " SERVER_LANGUAGEin the same scenario" |
| ) |
| if custom_server_lang: |
| if not workers_by_lang.get(custom_server_lang, []): |
| print( |
| "Warning: Skipping scenario %s as" |
| % scenario_json["name"] |
| ) |
| print( |
| "SERVER_LANGUAGE is set to %s yet the language" |
| " has not been selected with -l" |
| % custom_server_lang |
| ) |
| continue |
| for idx in range(0, scenario_json["num_servers"]): |
| # replace first X workers by workers of a different language |
| workers[idx] = workers_by_lang[custom_server_lang][ |
| idx |
| ] |
| if custom_client_lang: |
| if not workers_by_lang.get(custom_client_lang, []): |
| print( |
| "Warning: Skipping scenario %s as" |
| % scenario_json["name"] |
| ) |
| print( |
| "CLIENT_LANGUAGE is set to %s yet the language" |
| " has not been selected with -l" |
| % custom_client_lang |
| ) |
| continue |
| for idx in range( |
| scenario_json["num_servers"], len(workers) |
| ): |
| # replace all client workers by workers of a different language, |
| # leave num_server workers as they are server workers. |
| workers[idx] = workers_by_lang[custom_client_lang][ |
| idx |
| ] |
| scenario = Scenario( |
| create_scenario_jobspec( |
| scenario_json, |
| [w.host_and_port for w in workers], |
| remote_host=remote_host, |
| bq_result_table=bq_result_table, |
| server_cpu_load=server_cpu_load, |
| ), |
| workers, |
| scenario_json["name"], |
| ) |
| scenarios.append(scenario) |
| |
| return scenarios |
| |
| |
| def finish_qps_workers(jobs, qpsworker_jobs): |
| """Waits for given jobs to finish and eventually kills them.""" |
| retries = 0 |
| num_killed = 0 |
| while any(job.is_running() for job in jobs): |
| for job in qpsworker_jobs: |
| if job.is_running(): |
| print('QPS worker "%s" is still running.' % job.host_and_port) |
| if retries > 10: |
| print("Killing all QPS workers.") |
| for job in jobs: |
| job.kill() |
| num_killed += 1 |
| retries += 1 |
| time.sleep(3) |
| print("All QPS workers finished.") |
| return num_killed |
| |
| |
| profile_output_files = [] |
| |
| |
| # Collect perf text reports and flamegraphs if perf_cmd was used |
| # Note the base names of perf text reports are used when creating and processing |
| # perf data. The scenario name uniqifies the output name in the final |
| # perf reports directory. |
| # Alos, the perf profiles need to be fetched and processed after each scenario |
| # in order to avoid clobbering the output files. |
| def run_collect_perf_profile_jobs( |
| hosts_and_base_names, scenario_name, flame_graph_reports |
| ): |
| perf_report_jobs = [] |
| global profile_output_files |
| for host_and_port in hosts_and_base_names: |
| perf_base_name = hosts_and_base_names[host_and_port] |
| output_filename = "%s-%s" % (scenario_name, perf_base_name) |
| # from the base filename, create .svg output filename |
| host = host_and_port.split(":")[0] |
| profile_output_files.append("%s.svg" % output_filename) |
| perf_report_jobs.append( |
| perf_report_processor_job( |
| host, perf_base_name, output_filename, flame_graph_reports |
| ) |
| ) |
| |
| jobset.message( |
| "START", "Collecting perf reports from qps workers", do_newline=True |
| ) |
| failures, _ = jobset.run( |
| perf_report_jobs, newline_on_success=True, maxjobs=1 |
| ) |
| jobset.message( |
| "SUCCESS", "Collecting perf reports from qps workers", do_newline=True |
| ) |
| return failures |
| |
| |
| def main(): |
| argp = argparse.ArgumentParser(description="Run performance tests.") |
| argp.add_argument( |
| "-l", |
| "--language", |
| choices=["all"] + sorted(scenario_config.LANGUAGES.keys()), |
| nargs="+", |
| required=True, |
| help="Languages to benchmark.", |
| ) |
| argp.add_argument( |
| "--remote_driver_host", |
| default=None, |
| help=( |
| "Run QPS driver on given host. By default, QPS driver is run" |
| " locally." |
| ), |
| ) |
| argp.add_argument( |
| "--remote_worker_host", |
| nargs="+", |
| default=[], |
| help="Worker hosts where to start QPS workers.", |
| ) |
| argp.add_argument( |
| "--dry_run", |
| default=False, |
| action="store_const", |
| const=True, |
| help="Just list scenarios to be run, but don't run them.", |
| ) |
| argp.add_argument( |
| "-r", |
| "--regex", |
| default=".*", |
| type=str, |
| help="Regex to select scenarios to run.", |
| ) |
| argp.add_argument( |
| "--bq_result_table", |
| default=None, |
| type=str, |
| help='Bigquery "dataset.table" to upload results to.', |
| ) |
| argp.add_argument( |
| "--category", |
| choices=["smoketest", "all", "scalable", "sweep"], |
| default="all", |
| help="Select a category of tests to run.", |
| ) |
| argp.add_argument( |
| "--netperf", |
| default=False, |
| action="store_const", |
| const=True, |
| help="Run netperf benchmark as one of the scenarios.", |
| ) |
| argp.add_argument( |
| "--server_cpu_load", |
| default=0, |
| type=int, |
| help=( |
| "Select a targeted server cpu load to run. 0 means ignore this flag" |
| ), |
| ) |
| argp.add_argument( |
| "-x", |
| "--xml_report", |
| default="report.xml", |
| type=str, |
| help="Name of XML report file to generate.", |
| ) |
| argp.add_argument( |
| "--perf_args", |
| help=( |
| 'Example usage: "--perf_args=record -F 99 -g". ' |
| "Wrap QPS workers in a perf command " |
| "with the arguments to perf specified here. " |
| '".svg" flame graph profiles will be ' |
| "created for each Qps Worker on each scenario. " |
| 'Files will output to "<repo_root>/<args.flame_graph_reports>" ' |
| "directory. Output files from running the worker " |
| "under perf are saved in the repo root where its ran. " |
| 'Note that the perf "-g" flag is necessary for ' |
| "flame graphs generation to work (assuming the binary " |
| "being profiled uses frame pointers, check out " |
| '"--call-graph dwarf" option using libunwind otherwise.) ' |
| 'Also note that the entire "--perf_args=<arg(s)>" must ' |
| "be wrapped in quotes as in the example usage. " |
| 'If the "--perg_args" is unspecified, "perf" will ' |
| "not be used at all. " |
| "See http://www.brendangregg.com/perf.html " |
| "for more general perf examples." |
| ), |
| ) |
| argp.add_argument( |
| "--skip_generate_flamegraphs", |
| default=False, |
| action="store_const", |
| const=True, |
| help=( |
| "Turn flame graph generation off. " |
| 'May be useful if "perf_args" arguments do not make sense for ' |
| 'generating flamegraphs (e.g., "--perf_args=stat ...")' |
| ), |
| ) |
| argp.add_argument( |
| "-f", |
| "--flame_graph_reports", |
| default="perf_reports", |
| type=str, |
| help=( |
| "Name of directory to output flame graph profiles to, if any are" |
| " created." |
| ), |
| ) |
| argp.add_argument( |
| "-u", |
| "--remote_host_username", |
| default="", |
| type=str, |
| help='Use a username that isn\'t "Jenkins" to SSH into remote workers.', |
| ) |
| |
| args = argp.parse_args() |
| |
| global _REMOTE_HOST_USERNAME |
| if args.remote_host_username: |
| _REMOTE_HOST_USERNAME = args.remote_host_username |
| |
| languages = set( |
| scenario_config.LANGUAGES[l] |
| for l in itertools.chain.from_iterable( |
| six.iterkeys(scenario_config.LANGUAGES) if x == "all" else [x] |
| for x in args.language |
| ) |
| ) |
| |
| # Put together set of remote hosts where to run and build |
| remote_hosts = set() |
| if args.remote_worker_host: |
| for host in args.remote_worker_host: |
| remote_hosts.add(host) |
| if args.remote_driver_host: |
| remote_hosts.add(args.remote_driver_host) |
| |
| if not args.dry_run: |
| if remote_hosts: |
| archive_repo(languages=[str(l) for l in languages]) |
| prepare_remote_hosts(remote_hosts, prepare_local=True) |
| else: |
| prepare_remote_hosts([], prepare_local=True) |
| |
| build_local = False |
| if not args.remote_driver_host: |
| build_local = True |
| if not args.dry_run: |
| build_on_remote_hosts( |
| remote_hosts, |
| languages=[str(l) for l in languages], |
| build_local=build_local, |
| ) |
| |
| perf_cmd = None |
| if args.perf_args: |
| print("Running workers under perf profiler") |
| # Expect /usr/bin/perf to be installed here, as is usual |
| perf_cmd = ["/usr/bin/perf"] |
| perf_cmd.extend(re.split("\s+", args.perf_args)) |
| |
| qpsworker_jobs = create_qpsworkers( |
| languages, args.remote_worker_host, perf_cmd=perf_cmd |
| ) |
| |
| # get list of worker addresses for each language. |
| workers_by_lang = dict([(str(language), []) for language in languages]) |
| for job in qpsworker_jobs: |
| workers_by_lang[str(job.language)].append(job) |
| |
| scenarios = create_scenarios( |
| languages, |
| workers_by_lang=workers_by_lang, |
| remote_host=args.remote_driver_host, |
| regex=args.regex, |
| category=args.category, |
| bq_result_table=args.bq_result_table, |
| netperf=args.netperf, |
| netperf_hosts=args.remote_worker_host, |
| server_cpu_load=args.server_cpu_load, |
| ) |
| |
| if not scenarios: |
| raise Exception("No scenarios to run") |
| |
| total_scenario_failures = 0 |
| qps_workers_killed = 0 |
| merged_resultset = {} |
| perf_report_failures = 0 |
| |
| for scenario in scenarios: |
| if args.dry_run: |
| print(scenario.name) |
| else: |
| scenario_failures = 0 |
| try: |
| for worker in scenario.workers: |
| worker.start() |
| jobs = [scenario.jobspec] |
| if scenario.workers: |
| # TODO(jtattermusch): ideally the "quit" job won't show up |
| # in the report |
| jobs.append( |
| create_quit_jobspec( |
| scenario.workers, |
| remote_host=args.remote_driver_host, |
| ) |
| ) |
| scenario_failures, resultset = jobset.run( |
| jobs, newline_on_success=True, maxjobs=1 |
| ) |
| total_scenario_failures += scenario_failures |
| merged_resultset = dict( |
| itertools.chain( |
| six.iteritems(merged_resultset), |
| six.iteritems(resultset), |
| ) |
| ) |
| finally: |
| # Consider qps workers that need to be killed as failures |
| qps_workers_killed += finish_qps_workers( |
| scenario.workers, qpsworker_jobs |
| ) |
| |
| if ( |
| perf_cmd |
| and scenario_failures == 0 |
| and not args.skip_generate_flamegraphs |
| ): |
| workers_and_base_names = {} |
| for worker in scenario.workers: |
| if not worker.perf_file_base_name: |
| raise Exception( |
| "using perf buf perf report filename is unspecified" |
| ) |
| workers_and_base_names[ |
| worker.host_and_port |
| ] = worker.perf_file_base_name |
| perf_report_failures += run_collect_perf_profile_jobs( |
| workers_and_base_names, |
| scenario.name, |
| args.flame_graph_reports, |
| ) |
| |
| # Still write the index.html even if some scenarios failed. |
| # 'profile_output_files' will only have names for scenarios that passed |
| if perf_cmd and not args.skip_generate_flamegraphs: |
| # write the index fil to the output dir, with all profiles from all scenarios/workers |
| report_utils.render_perf_profiling_results( |
| "%s/index.html" % args.flame_graph_reports, profile_output_files |
| ) |
| |
| report_utils.render_junit_xml_report( |
| merged_resultset, |
| args.xml_report, |
| suite_name="benchmarks", |
| multi_target=True, |
| ) |
| |
| if total_scenario_failures > 0 or qps_workers_killed > 0: |
| print( |
| "%s scenarios failed and %s qps worker jobs killed" |
| % (total_scenario_failures, qps_workers_killed) |
| ) |
| sys.exit(1) |
| |
| if perf_report_failures > 0: |
| print("%s perf profile collection jobs failed" % perf_report_failures) |
| sys.exit(1) |
| |
| |
| if __name__ == "__main__": |
| main() |
