Split logs by test case and run client separately for each test

tools/internal_ci/linux/grpc_xds.cfg

@@ -21,3 +21,9 @@ env_vars {
   key: "BAZEL_SCRIPT"
   value: "tools/internal_ci/linux/grpc_xds_bazel_test_in_docker.sh"
 }
+action {
+  define_artifacts {
+    regex: "**/*sponge_log.*"
+    regex: "github/grpc/reports/**"
+  }
+}

tools/run_tests/python_utils/report_utils.py

@@ -31,9 +31,9 @@ def _filter_msg(msg, output_format):
     if output_format in ['XML', 'HTML']:
         # keep whitespaces but remove formfeed and vertical tab characters
         # that make XML report unparsable.
-        filtered_msg = filter(
-            lambda x: x in string.printable and x != '\f' and x != '\v',
-            msg.decode('UTF-8', 'ignore'))
+        filtered_msg = ''.join(
+            filter(lambda x: x in string.printable and x != '\f' and x != '\v',
+                   msg.decode('UTF-8', 'ignore')))
         if output_format == 'HTML':
             filtered_msg = filtered_msg.replace('"', '"')
         return filtered_msg

tools/run_tests/run_xds_tests.py

@@ -29,6 +29,9 @@ import time
 
 from oauth2client.client import GoogleCredentials
 
+import python_utils.jobset as jobset
+import python_utils.report_utils as report_utils
+
 from src.proto.grpc.testing import messages_pb2
 from src.proto.grpc.testing import test_pb2_grpc
 
@@ -38,6 +41,26 @@ formatter = logging.Formatter(fmt='%(asctime)s: %(levelname)-8s %(message)s')
 console_handler.setFormatter(formatter)
 logger.addHandler(console_handler)
 
+_TEST_CASES = [
+    'backends_restart',
+    'change_backend_service',
+    'new_instance_group_receives_traffic',
+    'ping_pong',
+    'remove_instance_group',
+    'round_robin',
+    'secondary_locality_gets_no_requests_on_partial_primary_failure',
+    'secondary_locality_gets_requests_on_primary_failure',
+]
+
+
+def parse_test_cases(arg):
+    if arg == 'all':
+        return _TEST_CASES
+    test_cases = arg.split(',')
+    if all([test_case in _TEST_CASES for test_case in test_cases]):
+        return test_cases
+    raise Exception('Failed to parse test cases %s' % arg)
+
 
 def parse_port_range(port_arg):
     try:
@@ -58,17 +81,9 @@ argp.add_argument(
 argp.add_argument(
     '--test_case',
     default='ping_pong',
-    choices=[
-        'all',
-        'backends_restart',
-        'change_backend_service',
-        'new_instance_group_receives_traffic',
-        'ping_pong',
-        'remove_instance_group',
-        'round_robin',
-        'secondary_locality_gets_no_requests_on_partial_primary_failure',
-        'secondary_locality_gets_requests_on_primary_failure',
-    ])
+    type=parse_test_cases,
+    help='Comma-separated list of test cases to run, or \'all\' to run every '
+    'test. Available tests: %s' % ' '.join(_TEST_CASES))
 argp.add_argument(
     '--client_cmd',
     default=None,
@@ -183,6 +198,9 @@ _BASE_URL_MAP_NAME = 'test-map'
 _BASE_SERVICE_HOST = 'grpc-test'
 _BASE_TARGET_PROXY_NAME = 'test-target-proxy'
 _BASE_FORWARDING_RULE_NAME = 'test-forwarding-rule'
+_TEST_LOG_BASE_DIR = 'reports'
+_SPONGE_LOG_NAME = 'sponge_log.log'
+_SPONGE_XML_NAME = 'sponge_log.xml'
 
 
 def get_client_stats(num_rpcs, timeout_sec):
@@ -905,8 +923,6 @@ if args.compute_discovery_document:
 else:
     compute = googleapiclient.discovery.build('compute', 'v1')
 
-client_process = None
-
 try:
     gcp = GcpState(compute, args.project_id)
     health_check_name = _BASE_HEALTH_CHECK_NAME + args.gcp_suffix
@@ -1034,56 +1050,75 @@ try:
         server_uri = service_host_name
     else:
         server_uri = service_host_name + ':' + str(gcp.service_port)
-    cmd = args.client_cmd.format(server_uri=server_uri,
-                                 stats_port=args.stats_port,
-                                 qps=args.qps)
-    client_process = start_xds_client(cmd)
-
-    if args.test_case == 'all':
-        test_backends_restart(gcp, backend_service, instance_group)
-        test_change_backend_service(gcp, backend_service, instance_group,
-                                    alternate_backend_service,
-                                    same_zone_instance_group)
-        test_new_instance_group_receives_traffic(gcp, backend_service,
-                                                 instance_group,
-                                                 same_zone_instance_group)
-        test_ping_pong(gcp, backend_service, instance_group)
-        test_remove_instance_group(gcp, backend_service, instance_group,
-                                   same_zone_instance_group)
-        test_round_robin(gcp, backend_service, instance_group)
-        test_secondary_locality_gets_no_requests_on_partial_primary_failure(
-            gcp, backend_service, instance_group, secondary_zone_instance_group)
-        test_secondary_locality_gets_requests_on_primary_failure(
-            gcp, backend_service, instance_group, secondary_zone_instance_group)
-    elif args.test_case == 'backends_restart':
-        test_backends_restart(gcp, backend_service, instance_group)
-    elif args.test_case == 'change_backend_service':
-        test_change_backend_service(gcp, backend_service, instance_group,
-                                    alternate_backend_service,
-                                    same_zone_instance_group)
-    elif args.test_case == 'new_instance_group_receives_traffic':
-        test_new_instance_group_receives_traffic(gcp, backend_service,
-                                                 instance_group,
-                                                 same_zone_instance_group)
-    elif args.test_case == 'ping_pong':
-        test_ping_pong(gcp, backend_service, instance_group)
-    elif args.test_case == 'remove_instance_group':
-        test_remove_instance_group(gcp, backend_service, instance_group,
-                                   same_zone_instance_group)
-    elif args.test_case == 'round_robin':
-        test_round_robin(gcp, backend_service, instance_group)
-    elif args.test_case == 'secondary_locality_gets_no_requests_on_partial_primary_failure':
-        test_secondary_locality_gets_no_requests_on_partial_primary_failure(
-            gcp, backend_service, instance_group, secondary_zone_instance_group)
-    elif args.test_case == 'secondary_locality_gets_requests_on_primary_failure':
-        test_secondary_locality_gets_requests_on_primary_failure(
-            gcp, backend_service, instance_group, secondary_zone_instance_group)
-    else:
-        logger.error('Unknown test case: %s', args.test_case)
-        sys.exit(1)
+    with tempfile.NamedTemporaryFile(delete=False) as bootstrap_file:
+        bootstrap_file.write(
+            _BOOTSTRAP_TEMPLATE.format(
+                node_id=socket.gethostname()).encode('utf-8'))
+        bootstrap_path = bootstrap_file.name
+    client_env = dict(os.environ, GRPC_XDS_BOOTSTRAP=bootstrap_path)
+    client_cmd = shlex.split(
+        args.client_cmd.format(server_uri=server_uri,
+                               stats_port=args.stats_port,
+                               qps=args.qps))
+
+    test_results = {}
+    for test_case in args.test_case:
+        result = jobset.JobResult()
+        log_dir = os.path.join(_TEST_LOG_BASE_DIR, test_case)
+        if not os.path.exists(log_dir):
+            os.makedirs(log_dir)
+        test_log_file = open(os.path.join(log_dir, _SPONGE_LOG_NAME), 'w+')
+        try:
+            client_process = subprocess.Popen(client_cmd,
+                                              env=client_env,
+                                              stderr=subprocess.STDOUT,
+                                              stdout=test_log_file)
+            if test_case == 'backends_restart':
+                test_backends_restart(gcp, backend_service, instance_group)
+            elif test_case == 'change_backend_service':
+                test_change_backend_service(gcp, backend_service,
+                                            instance_group,
+                                            alternate_backend_service,
+                                            same_zone_instance_group)
+            elif test_case == 'new_instance_group_receives_traffic':
+                test_new_instance_group_receives_traffic(
+                    gcp, backend_service, instance_group,
+                    same_zone_instance_group)
+            elif test_case == 'ping_pong':
+                test_ping_pong(gcp, backend_service, instance_group)
+            elif test_case == 'remove_instance_group':
+                test_remove_instance_group(gcp, backend_service, instance_group,
+                                           same_zone_instance_group)
+            elif test_case == 'round_robin':
+                test_round_robin(gcp, backend_service, instance_group)
+            elif test_case == 'secondary_locality_gets_no_requests_on_partial_primary_failure':
+                test_secondary_locality_gets_no_requests_on_partial_primary_failure(
+                    gcp, backend_service, instance_group,
+                    secondary_zone_instance_group)
+            elif test_case == 'secondary_locality_gets_requests_on_primary_failure':
+                test_secondary_locality_gets_requests_on_primary_failure(
+                    gcp, backend_service, instance_group,
+                    secondary_zone_instance_group)
+            else:
+                logger.error('Unknown test case: %s', test_case)
+                sys.exit(1)
+            result.state = 'PASSED'
+            result.returncode = 0
+        except Exception as e:
+            result.state = 'FAILED'
+            result.message = str(e).encode('UTF-8')
+        finally:
+            if client_process:
+                client_process.terminate()
+            test_results[test_case] = [result]
+    if not os.path.exists(_TEST_LOG_BASE_DIR):
+        os.makedirs(_TEST_LOG_BASE_DIR)
+    report_utils.render_junit_xml_report(test_results,
+                                         os.path.join(_TEST_LOG_BASE_DIR,
+                                                      _SPONGE_XML_NAME),
+                                         suite_name='xds_tests',
+                                         multi_target=True)
 finally:
-    if client_process:
-        client_process.terminate()
     if not args.keep_gcp_resources:
         logger.info('Cleaning up GCP resources. This may take some time.')
         clean_up(gcp)