grpc/tools/run_tests/run_grpclb_interop_tests.py

#!/usr/bin/env python3
# Copyright 2015 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 interop (cross-language) tests in parallel."""

from __future__ import print_function

import argparse
import atexit
import itertools
import json
import multiprocessing
import os
import re
import subprocess
import sys
import tempfile
import time
import traceback
import uuid

import six

import python_utils.dockerjob as dockerjob
import python_utils.jobset as jobset
import python_utils.report_utils as report_utils

# Docker doesn't clean up after itself, so we do it on exit.
atexit.register(lambda: subprocess.call(["stty", "echo"]))

ROOT = os.path.abspath(os.path.join(os.path.dirname(sys.argv[0]), "../.."))
os.chdir(ROOT)

_FALLBACK_SERVER_PORT = 443
_BALANCER_SERVER_PORT = 12000
_BACKEND_SERVER_PORT = 8080

_TEST_TIMEOUT = 30

_FAKE_SERVERS_SAFENAME = "fake_servers"

# Use a name that's verified by the test certs
_SERVICE_NAME = "server.test.google.fr"


class CXXLanguage:
    def __init__(self):
        self.client_cwd = "/var/local/git/grpc"
        self.safename = "cxx"

    def client_cmd(self, args):
        return ["bins/opt/interop_client"] + args

    def global_env(self):
        # 1) Set c-ares as the resolver, to
        #    enable grpclb.
        # 2) Turn on verbose logging.
        # 3) Set the ROOTS_PATH env variable
        #    to the test CA in order for
        #    GoogleDefaultCredentials to be
        #    able to use the test CA.
        return {
            "GRPC_DNS_RESOLVER": "ares",
            "GRPC_VERBOSITY": "DEBUG",
            "GRPC_TRACE": "client_channel,glb",
            "GRPC_DEFAULT_SSL_ROOTS_FILE_PATH": (
                "/var/local/git/grpc/src/core/tsi/test_creds/ca.pem"
            ),
        }

    def __str__(self):
        return "c++"


class JavaLanguage:
    def __init__(self):
        self.client_cwd = "/var/local/git/grpc-java"
        self.safename = str(self)

    def client_cmd(self, args):
        # Take necessary steps to import our test CA into
        # the set of test CA's that the Java runtime of the
        # docker container will pick up, so that
        # Java GoogleDefaultCreds can use it.
        pem_to_der_cmd = (
            "openssl x509 -outform der "
            "-in /external_mount/src/core/tsi/test_creds/ca.pem "
            "-out /tmp/test_ca.der"
        )
        keystore_import_cmd = (
            "keytool -import "
            "-keystore /usr/lib/jvm/java-8-oracle/jre/lib/security/cacerts "
            "-file /tmp/test_ca.der "
            "-deststorepass changeit "
            "-noprompt"
        )
        return [
            "bash",
            "-c",
            (
                "{pem_to_der_cmd} && "
                "{keystore_import_cmd} && "
                "./run-test-client.sh {java_client_args}"
            ).format(
                pem_to_der_cmd=pem_to_der_cmd,
                keystore_import_cmd=keystore_import_cmd,
                java_client_args=" ".join(args),
            ),
        ]

    def global_env(self):
        # 1) Enable grpclb
        # 2) Enable verbose logging
        return {
            "JAVA_OPTS": (
                "-Dio.grpc.internal.DnsNameResolverProvider.enable_grpclb=true "
                "-Djava.util.logging.config.file=/var/local/grpc_java_logging/logconf.txt"
            )
        }

    def __str__(self):
        return "java"


class GoLanguage:
    def __init__(self):
        self.client_cwd = "/go/src/google.golang.org/grpc/interop/client"
        self.safename = str(self)

    def client_cmd(self, args):
        # Copy the test CA file into the path that
        # the Go runtime in the docker container will use, so
        # that Go's GoogleDefaultCredentials can use it.
        # See https://golang.org/src/crypto/x509/root_linux.go.
        return [
            "bash",
            "-c",
            (
                "cp /external_mount/src/core/tsi/test_creds/ca.pem "
                "/etc/ssl/certs/ca-certificates.crt && "
                "/go/bin/client {go_client_args}"
            ).format(go_client_args=" ".join(args)),
        ]

    def global_env(self):
        return {
            "GRPC_GO_LOG_VERBOSITY_LEVEL": "3",
            "GRPC_GO_LOG_SEVERITY_LEVEL": "INFO",
        }

    def __str__(self):
        return "go"


_LANGUAGES = {
    "c++": CXXLanguage(),
    "go": GoLanguage(),
    "java": JavaLanguage(),
}


def docker_run_cmdline(cmdline, image, docker_args, cwd, environ=None):
    """Wraps given cmdline array to create 'docker run' cmdline from it."""
    # turn environ into -e docker args
    docker_cmdline = "docker run -i --rm=true".split()
    if environ:
        for k, v in list(environ.items()):
            docker_cmdline += ["-e", "%s=%s" % (k, v)]
    return docker_cmdline + ["-w", cwd] + docker_args + [image] + cmdline


def _job_kill_handler(job):
    assert job._spec.container_name
    dockerjob.docker_kill(job._spec.container_name)


def transport_security_to_args(transport_security):
    args = []
    if transport_security == "tls":
        args += ["--use_tls=true"]
    elif transport_security == "alts":
        args += ["--use_tls=false", "--use_alts=true"]
    elif transport_security == "insecure":
        args += ["--use_tls=false"]
    elif transport_security == "google_default_credentials":
        args += ["--custom_credentials_type=google_default_credentials"]
    else:
        print("Invalid transport security option.")
        sys.exit(1)
    return args


def lb_client_interop_jobspec(
    language, dns_server_ip, docker_image, transport_security="tls"
):
    """Runs a gRPC client under test in a docker container"""
    interop_only_options = [
        "--server_host=%s" % _SERVICE_NAME,
        "--server_port=%d" % _FALLBACK_SERVER_PORT,
    ] + transport_security_to_args(transport_security)
    # Don't set the server host override in any client;
    # Go and Java default to no override.
    # We're using a DNS server so there's no need.
    if language.safename == "c++":
        interop_only_options += ['--server_host_override=""']
    # Don't set --use_test_ca; we're configuring
    # clients to use test CA's via alternate means.
    interop_only_options += ["--use_test_ca=false"]
    client_args = language.client_cmd(interop_only_options)
    container_name = dockerjob.random_name(
        "lb_interop_client_%s" % language.safename
    )
    docker_cmdline = docker_run_cmdline(
        client_args,
        environ=language.global_env(),
        image=docker_image,
        cwd=language.client_cwd,
        docker_args=[
            "--dns=%s" % dns_server_ip,
            "--net=host",
            "--name=%s" % container_name,
            "-v",
            "{grpc_grpc_root_dir}:/external_mount:ro".format(
                grpc_grpc_root_dir=ROOT
            ),
        ],
    )
    jobset.message(
        "IDLE",
        "docker_cmdline:\b|%s|" % " ".join(docker_cmdline),
        do_newline=True,
    )
    test_job = jobset.JobSpec(
        cmdline=docker_cmdline,
        shortname="lb_interop_client:%s" % language,
        timeout_seconds=_TEST_TIMEOUT,
        kill_handler=_job_kill_handler,
    )
    test_job.container_name = container_name
    return test_job


def fallback_server_jobspec(transport_security, shortname):
    """Create jobspec for running a fallback server"""
    cmdline = [
        "bin/server",
        "--port=%d" % _FALLBACK_SERVER_PORT,
    ] + transport_security_to_args(transport_security)
    return grpc_server_in_docker_jobspec(
        server_cmdline=cmdline, shortname=shortname
    )


def backend_server_jobspec(transport_security, shortname):
    """Create jobspec for running a backend server"""
    cmdline = [
        "bin/server",
        "--port=%d" % _BACKEND_SERVER_PORT,
    ] + transport_security_to_args(transport_security)
    return grpc_server_in_docker_jobspec(
        server_cmdline=cmdline, shortname=shortname
    )


def grpclb_jobspec(transport_security, short_stream, backend_addrs, shortname):
    """Create jobspec for running a balancer server"""
    cmdline = [
        "bin/fake_grpclb",
        "--backend_addrs=%s" % ",".join(backend_addrs),
        "--port=%d" % _BALANCER_SERVER_PORT,
        "--short_stream=%s" % short_stream,
        "--service_name=%s" % _SERVICE_NAME,
    ] + transport_security_to_args(transport_security)
    return grpc_server_in_docker_jobspec(
        server_cmdline=cmdline, shortname=shortname
    )


def grpc_server_in_docker_jobspec(server_cmdline, shortname):
    container_name = dockerjob.random_name(shortname)
    environ = {
        "GRPC_GO_LOG_VERBOSITY_LEVEL": "3",
        "GRPC_GO_LOG_SEVERITY_LEVEL": "INFO ",
    }
    docker_cmdline = docker_run_cmdline(
        server_cmdline,
        cwd="/go",
        image=docker_images.get(_FAKE_SERVERS_SAFENAME),
        environ=environ,
        docker_args=["--name=%s" % container_name],
    )
    jobset.message(
        "IDLE",
        "docker_cmdline:\b|%s|" % " ".join(docker_cmdline),
        do_newline=True,
    )
    server_job = jobset.JobSpec(
        cmdline=docker_cmdline, shortname=shortname, timeout_seconds=30 * 60
    )
    server_job.container_name = container_name
    return server_job


def dns_server_in_docker_jobspec(
    grpclb_ips,
    fallback_ips,
    shortname,
    cause_no_error_no_data_for_balancer_a_record,
):
    container_name = dockerjob.random_name(shortname)
    run_dns_server_cmdline = [
        "python",
        "test/cpp/naming/utils/run_dns_server_for_lb_interop_tests.py",
        "--grpclb_ips=%s" % ",".join(grpclb_ips),
        "--fallback_ips=%s" % ",".join(fallback_ips),
    ]
    if cause_no_error_no_data_for_balancer_a_record:
        run_dns_server_cmdline.append(
            "--cause_no_error_no_data_for_balancer_a_record"
        )
    docker_cmdline = docker_run_cmdline(
        run_dns_server_cmdline,
        cwd="/var/local/git/grpc",
        image=docker_images.get(_FAKE_SERVERS_SAFENAME),
        docker_args=["--name=%s" % container_name],
    )
    jobset.message(
        "IDLE",
        "docker_cmdline:\b|%s|" % " ".join(docker_cmdline),
        do_newline=True,
    )
    server_job = jobset.JobSpec(
        cmdline=docker_cmdline, shortname=shortname, timeout_seconds=30 * 60
    )
    server_job.container_name = container_name
    return server_job


def build_interop_image_jobspec(lang_safename, basename_prefix="grpc_interop"):
    """Creates jobspec for building interop docker image for a language"""
    tag = "%s_%s:%s" % (basename_prefix, lang_safename, uuid.uuid4())
    env = {
        "INTEROP_IMAGE": tag,
        "BASE_NAME": "%s_%s" % (basename_prefix, lang_safename),
    }
    build_job = jobset.JobSpec(
        cmdline=["tools/run_tests/dockerize/build_interop_image.sh"],
        environ=env,
        shortname="build_docker_%s" % lang_safename,
        timeout_seconds=30 * 60,
    )
    build_job.tag = tag
    return build_job


argp = argparse.ArgumentParser(description="Run interop tests.")
argp.add_argument(
    "-l",
    "--language",
    choices=["all"] + sorted(_LANGUAGES),
    nargs="+",
    default=["all"],
    help="Clients to run.",
)
argp.add_argument("-j", "--jobs", default=multiprocessing.cpu_count(), type=int)
argp.add_argument(
    "-s",
    "--scenarios_file",
    default=None,
    type=str,
    help="File containing test scenarios as JSON configs.",
)
argp.add_argument(
    "-n",
    "--scenario_name",
    default=None,
    type=str,
    help=(
        "Useful for manual runs: specify the name of "
        "the scenario to run from scenarios_file. Run all scenarios if unset."
    ),
)
argp.add_argument(
    "--cxx_image_tag",
    default=None,
    type=str,
    help=(
        "Setting this skips the clients docker image "
        "build step and runs the client from the named "
        "image. Only supports running a one client language."
    ),
)
argp.add_argument(
    "--go_image_tag",
    default=None,
    type=str,
    help=(
        "Setting this skips the clients docker image build "
        "step and runs the client from the named image. Only "
        "supports running a one client language."
    ),
)
argp.add_argument(
    "--java_image_tag",
    default=None,
    type=str,
    help=(
        "Setting this skips the clients docker image build "
        "step and runs the client from the named image. Only "
        "supports running a one client language."
    ),
)
argp.add_argument(
    "--servers_image_tag",
    default=None,
    type=str,
    help=(
        "Setting this skips the fake servers docker image "
        "build step and runs the servers from the named image."
    ),
)
argp.add_argument(
    "--no_skips",
    default=False,
    type=bool,
    nargs="?",
    const=True,
    help=(
        "Useful for manual runs. Setting this overrides test "
        '"skips" configured in test scenarios.'
    ),
)
argp.add_argument(
    "--verbose",
    default=False,
    type=bool,
    nargs="?",
    const=True,
    help="Increase logging.",
)
args = argp.parse_args()

docker_images = {}

build_jobs = []
if len(args.language) and args.language[0] == "all":
    languages = list(_LANGUAGES.keys())
else:
    languages = args.language
for lang_name in languages:
    l = _LANGUAGES[lang_name]
    # First check if a pre-built image was supplied, and avoid
    # rebuilding the particular docker image if so.
    if lang_name == "c++" and args.cxx_image_tag:
        docker_images[str(l.safename)] = args.cxx_image_tag
    elif lang_name == "go" and args.go_image_tag:
        docker_images[str(l.safename)] = args.go_image_tag
    elif lang_name == "java" and args.java_image_tag:
        docker_images[str(l.safename)] = args.java_image_tag
    else:
        # Build the test client in docker and save the fully
        # built image.
        job = build_interop_image_jobspec(l.safename)
        build_jobs.append(job)
        docker_images[str(l.safename)] = job.tag

# First check if a pre-built image was supplied.
if args.servers_image_tag:
    docker_images[_FAKE_SERVERS_SAFENAME] = args.servers_image_tag
else:
    # Build the test servers in docker and save the fully
    # built image.
    job = build_interop_image_jobspec(
        _FAKE_SERVERS_SAFENAME, basename_prefix="lb_interop"
    )
    build_jobs.append(job)
    docker_images[_FAKE_SERVERS_SAFENAME] = job.tag

if build_jobs:
    jobset.message("START", "Building interop docker images.", do_newline=True)
    print("Jobs to run: \n%s\n" % "\n".join(str(j) for j in build_jobs))
    num_failures, _ = jobset.run(
        build_jobs, newline_on_success=True, maxjobs=args.jobs
    )
    if num_failures == 0:
        jobset.message(
            "SUCCESS", "All docker images built successfully.", do_newline=True
        )
    else:
        jobset.message(
            "FAILED", "Failed to build interop docker images.", do_newline=True
        )
        sys.exit(1)


def wait_until_dns_server_is_up(dns_server_ip):
    """Probes the DNS server until it's running and safe for tests."""
    for i in range(0, 30):
        print("Health check: attempt to connect to DNS server over TCP.")
        tcp_connect_subprocess = subprocess.Popen(
            [
                os.path.join(
                    os.getcwd(), "test/cpp/naming/utils/tcp_connect.py"
                ),
                "--server_host",
                dns_server_ip,
                "--server_port",
                str(53),
                "--timeout",
                str(1),
            ]
        )
        tcp_connect_subprocess.communicate()
        if tcp_connect_subprocess.returncode == 0:
            print(
                "Health check: attempt to make an A-record query to DNS server."
            )
            dns_resolver_subprocess = subprocess.Popen(
                [
                    os.path.join(
                        os.getcwd(), "test/cpp/naming/utils/dns_resolver.py"
                    ),
                    "--qname",
                    (
                        "health-check-local-dns-server-is-alive."
                        "resolver-tests.grpctestingexp"
                    ),
                    "--server_host",
                    dns_server_ip,
                    "--server_port",
                    str(53),
                ],
                stdout=subprocess.PIPE,
            )
            dns_resolver_stdout, _ = dns_resolver_subprocess.communicate()
            if dns_resolver_subprocess.returncode == 0:
                if "123.123.123.123" in dns_resolver_stdout:
                    print(
                        "DNS server is up! "
                        "Successfully reached it over UDP and TCP."
                    )
                    return
        time.sleep(0.1)
    raise Exception(
        "Failed to reach DNS server over TCP and/or UDP. "
        "Exitting without running tests."
    )


def shortname(shortname_prefix, shortname, index):
    return "%s_%s_%d" % (shortname_prefix, shortname, index)


def run_one_scenario(scenario_config):
    jobset.message("START", "Run scenario: %s" % scenario_config["name"])
    server_jobs = {}
    server_addresses = {}
    suppress_server_logs = True
    try:
        backend_addrs = []
        fallback_ips = []
        grpclb_ips = []
        shortname_prefix = scenario_config["name"]
        # Start backends
        for i in range(len(scenario_config["backend_configs"])):
            backend_config = scenario_config["backend_configs"][i]
            backend_shortname = shortname(shortname_prefix, "backend_server", i)
            backend_spec = backend_server_jobspec(
                backend_config["transport_sec"], backend_shortname
            )
            backend_job = dockerjob.DockerJob(backend_spec)
            server_jobs[backend_shortname] = backend_job
            backend_addrs.append(
                "%s:%d" % (backend_job.ip_address(), _BACKEND_SERVER_PORT)
            )
        # Start fallbacks
        for i in range(len(scenario_config["fallback_configs"])):
            fallback_config = scenario_config["fallback_configs"][i]
            fallback_shortname = shortname(
                shortname_prefix, "fallback_server", i
            )
            fallback_spec = fallback_server_jobspec(
                fallback_config["transport_sec"], fallback_shortname
            )
            fallback_job = dockerjob.DockerJob(fallback_spec)
            server_jobs[fallback_shortname] = fallback_job
            fallback_ips.append(fallback_job.ip_address())
        # Start balancers
        for i in range(len(scenario_config["balancer_configs"])):
            balancer_config = scenario_config["balancer_configs"][i]
            grpclb_shortname = shortname(shortname_prefix, "grpclb_server", i)
            grpclb_spec = grpclb_jobspec(
                balancer_config["transport_sec"],
                balancer_config["short_stream"],
                backend_addrs,
                grpclb_shortname,
            )
            grpclb_job = dockerjob.DockerJob(grpclb_spec)
            server_jobs[grpclb_shortname] = grpclb_job
            grpclb_ips.append(grpclb_job.ip_address())
        # Start DNS server
        dns_server_shortname = shortname(shortname_prefix, "dns_server", 0)
        dns_server_spec = dns_server_in_docker_jobspec(
            grpclb_ips,
            fallback_ips,
            dns_server_shortname,
            scenario_config["cause_no_error_no_data_for_balancer_a_record"],
        )
        dns_server_job = dockerjob.DockerJob(dns_server_spec)
        server_jobs[dns_server_shortname] = dns_server_job
        # Get the IP address of the docker container running the DNS server.
        # The DNS server is running on port 53 of that IP address. Note we will
        # point the DNS resolvers of grpc clients under test to our controlled
        # DNS server by effectively modifying the /etc/resolve.conf "nameserver"
        # lists of their docker containers.
        dns_server_ip = dns_server_job.ip_address()
        wait_until_dns_server_is_up(dns_server_ip)
        # Run clients
        jobs = []
        for lang_name in languages:
            # Skip languages that are known to not currently
            # work for this test.
            if not args.no_skips and lang_name in scenario_config.get(
                "skip_langs", []
            ):
                jobset.message(
                    "IDLE",
                    "Skipping scenario: %s for language: %s\n"
                    % (scenario_config["name"], lang_name),
                )
                continue
            lang = _LANGUAGES[lang_name]
            test_job = lb_client_interop_jobspec(
                lang,
                dns_server_ip,
                docker_image=docker_images.get(lang.safename),
                transport_security=scenario_config["transport_sec"],
            )
            jobs.append(test_job)
        jobset.message(
            "IDLE", "Jobs to run: \n%s\n" % "\n".join(str(job) for job in jobs)
        )
        num_failures, resultset = jobset.run(
            jobs, newline_on_success=True, maxjobs=args.jobs
        )
        report_utils.render_junit_xml_report(resultset, "sponge_log.xml")
        if num_failures:
            suppress_server_logs = False
            jobset.message(
                "FAILED",
                "Scenario: %s. Some tests failed" % scenario_config["name"],
                do_newline=True,
            )
        else:
            jobset.message(
                "SUCCESS",
                "Scenario: %s. All tests passed" % scenario_config["name"],
                do_newline=True,
            )
        return num_failures
    finally:
        # Check if servers are still running.
        for server, job in list(server_jobs.items()):
            if not job.is_running():
                print('Server "%s" has exited prematurely.' % server)
        suppress_failure = suppress_server_logs and not args.verbose
        dockerjob.finish_jobs(
            [j for j in six.itervalues(server_jobs)],
            suppress_failure=suppress_failure,
        )


num_failures = 0
with open(args.scenarios_file, "r") as scenarios_input:
    all_scenarios = json.loads(scenarios_input.read())
    for scenario in all_scenarios:
        if args.scenario_name:
            if args.scenario_name != scenario["name"]:
                jobset.message(
                    "IDLE", "Skipping scenario: %s" % scenario["name"]
                )
                continue
        num_failures += run_one_scenario(scenario)
if num_failures == 0:
    sys.exit(0)
else:
    sys.exit(1)