remote_data_load.py

#!/usr/bin/env python
# Copyright 2015 Cloudera Inc.
#
# 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.
#
#
# This is a setup script that will downloaded a test warehouse snapshot and
# deploy it on a remote, CM-managed cluster. Once the data is loaded, it is
# possible to run a subset of the Impala core / exhaustive tests on the
# remote cluster.
#
#   * This script should be executed from a machine that has the Impala
#     development environment set up.
#
#   * The cluster needs to be configured appropriately:
#     - The following services need to be installed:
#       HDFS, YARN, HIVE, IMPALA, MAPREDUCE, KUDU, HBASE, ZOOKEEPER
#     - GPL Extras parcel needs to be installed
#     - Metastore DB SERDE properties PARAM_VALUE needs to be altered to
#       allow for wide tables (See HIVE-1364.)
#     - The hive-warehouse path needs to be /test-warehouse
#
# Usage: remote_data_load.py [options] cm_host
#
#  Options:
#    -h, --help           show this help message and exit
#    --cm_user=CM_USER    Cloudera Manager admin user
#    --cm_pass=CM_PASS    Cloudera Manager admin user password
#    --no-load            Do not try to load the snapshot
#    --test               Run end-to-end tests against cluster.
#    --gateway=GATEWAY    Gateway host to upload the data from. If not set, uses
#                         the CM host as gateway.
#    --ssh_user=SSH_USER  System user on the remote machine with passwordless SSH
#                         configured.
#
import fnmatch
import glob
import logging
import os
import sh
import shutil
import sys
import time

from cm_api.api_client import ApiResource
from functools import wraps
from optparse import OptionParser
from sh import ssh
from tempfile import mkdtemp
from urllib import quote as urlquote


REQUIRED_SERVICES = ['HBASE',
                     'HDFS',
                     'HIVE',
                     'IMPALA',
                     'KUDU',
                     'MAPREDUCE',
                     'YARN',
                     'ZOOKEEPER']

# TODO: It's not currently possible to get the version from the cluster.
# It would be nice to generate this dynamically.
# (v14 happens to be the version that ships with CDH 5.9.x)
CM_API_VERSION = 'v14'

# Impala's data loading and test framework assumes this Hive Warehouse Directory.
# Making this configurable would be an invasive change, and therefore, we prefer to
# re-configure the Hive service via the CM API before loading data and running tests.
HIVE_WAREHOUSE_DIR = "/test-warehouse"

logger = logging.getLogger("remote_data_load")
logger.setLevel(logging.DEBUG)

# Goes to the file
fh = logging.FileHandler("remote_data_load.log")
fh.setLevel(logging.DEBUG)

# Goes to stdout
ch = logging.StreamHandler()
ch.setLevel(logging.INFO)

# create formatter and add it to the handlers
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
fh.setFormatter(formatter)
ch.setFormatter(formatter)

# add the handlers to the logger
logger.addHandler(fh)
logger.addHandler(ch)


def timing(func):
    """
    A decorator for timing how much time a function takes.

    We can modify this later to do something more intelligent than just logging.
    """
    @wraps(func)
    def wrap(*args, **kwargs):
        t1 = time.time()
        result = func(*args, **kwargs)
        t2 = time.time()

        output = 'TIME: {name}() took: {t:.4f} seconds'
        logger.info(output.format(name=func.__name__, t=(t2-t1)))
        return result
    return wrap


def tee(line):
    """Output wrapper function used by sh to send the stdout or stderr to the
    module's logger."""
    logger.debug(line.strip())


class RemoteDataLoad(object):
    """This is an implementation of the process to load a test-warehouse snapshot on
    a remote CM managed cluster. This script assumes that the warehouse snapshot was
    already downloaded and was either passed in as a parameter, or can be found by
    either inspecting the SNAPSHOT_DIR environment variable, or based on the WORKSPACE
    environment variable on a Jenkins build slave.

    The reason for the additional setup code is that in the local development
    environment it is assumed that $USER is HDFS superuser, which is not the case for
    remote deloyments.
    """

    def __init__(self, cm_host, options):
        logger.info("Starting remote data load...")
        self.options = options
        self.cm_host = cm_host

        # Gateway host can be used if the CM host is not configured as a Hadoop gateway
        self.gateway = options.gateway if options.gateway else cm_host
        self.impala_home = os.environ["IMPALA_HOME"]
        self.api = ApiResource(self.cm_host, username=options.cm_user,
                               password=options.cm_pass)

        # The API returns a list of clusters managed by the CM host. We're assuming
        # that this CM host was set up for the purpose of Impala testing on one
        # cluster, so the list should only have one value.
        self.cluster = self.api.get_all_clusters()[0]
        self.services = self.get_services()

        self.config = self.get_service_client_configurations()
        logger.info("Retrieved service configuration")
        logger.info(str(self.config))
        self.prepare()
        logger.info("IMPALA_HOME: {0}".format(self.impala_home))

    def get_hostname_for_ref(self, host_ref):
        """Translate the HostRef instance into the hostname."""
        return self.api.get_host(host_ref.hostId).hostname

    @staticmethod
    def get_or_default(config):
        return config.value if config.value else config.default

    def get_services(self):
        """Confirm that all services are running, and return service dict."""
        services = dict((s.type, s) for s in self.cluster.get_all_services())

        if set(REQUIRED_SERVICES) != set(services.keys()):
            missing_services = set(REQUIRED_SERVICES) - set(services.keys())
            logger.error("Services not installed: {0}".format(list(missing_services)))
            raise RuntimeError("Cluster not ready.")

        if not all(services[s].serviceState == 'STARTED' for s in services):
            stopped = [s for s in services if services[s].serviceState != "STARTED"]
            logger.error("Not all services started: {0}".format(stopped))
            raise RuntimeError("Cluster not ready.")

        return services

    @timing
    def download_client_config(self, cluster, service):
        """Download the client configuration zip for a particular cluster and service.

        Since cm_api does not provide a way to download the archive we build the URL
        manually and download the file. Once it downloaded the file the archive is
        extracted and its content is copied to the Hadoop configuration directories
        defined by Impala.
        """
        logger.info("Downloading client configuration for {0}".format(service.name))
        url = "http://{0}:7180/api/{1}/clusters/{2}/services/{3}/clientConfig".format(
            self.cm_host, CM_API_VERSION, urlquote(cluster.name), urlquote(service.name))
        path = mkdtemp()
        sh.curl(url, o=os.path.join(path, "clientConfig.zip"), _out=tee, _err=tee)
        current = os.getcwd()
        os.chdir(path)
        sh.unzip("clientConfig.zip")
        for root, _, file_names in os.walk("."):
            for filename in fnmatch.filter(file_names, "*.xml"):
                src = os.path.join(root, filename)
                dst = os.path.join(self.impala_home, "fe", "src", "test", "resources")
                logger.debug("Copying {0} to {1}".format(src, dst))
                shutil.copy(src, dst)
        os.chdir(current)

    # TODO: this may be available in tests/comparison/cluster.py
    def set_hive_warehouse_dir(self, cluster, service):
        logger.info("Setting the Hive Warehouse Dir")
        for service in self.api.get_all_clusters()[0].get_all_services():
            logger.info(service)
            if service.type == "HIVE":
              hive_config = { "hive_warehouse_directory" : HIVE_WAREHOUSE_DIR }
              service.update_config(hive_config)

    # TODO: This functionality should be more generally available to other infrastructure
    # code, rather than being quarantined in this script. See IMPALA-4367.
    @timing
    def get_service_client_configurations(self):
        """Download the client configurations necessary to upload data to the remote
        cluster. Unfortunately, the CM API does not allow downloading it so we have to
        iterate over the services and download the config for all of them.

        In addition, returns an options dictionary with settings required for data loading
        like the HS2 server, Impala hosts, Name node etc.

        Returns:
            A client-configuration dictionary, e.g.:

            {
                'hive_warehouse_directory': '/test-warehouse',
                'hs2': 'impala-test-cluster-1.gce.cloudera.com:10000',
                'impalad': ['impala-test-cluster-4.gce.cloudera.com:21000',
                            'impala-test-cluster-2.gce.cloudera.com:21000',
                            'impala-test-cluster-3.gce.cloudera.com:21000'],
                'metastore': 'impala-test-cluster-1.gce.cloudera.com:9083',
                'namenode': 'impala-test-cluster-1.gce.cloudera.com',
                'namenode_http': 'impala-test-cluster-1.gce.cloudera.com:20101',
                'kudu_master': 'impala-test-cluster-1.gce.cloudera.com'
            }
        """
        # Iterate overs services and find the information we need
        result = {}
        for service_type, service in self.services.iteritems():
            if service_type == "IMPALA":
                roles = service.get_roles_by_type("IMPALAD")
                impalads = []
                for r in roles:
                    rc_config = r.get_config("full")
                    hostname = self.get_hostname_for_ref(r.hostRef)
                    hs2_port = self.get_or_default(rc_config["beeswax_port"])
                    impalads.append("{0}:{1}".format(hostname, hs2_port))
                    result["impalad"] = impalads
            elif service_type == "HBASE":
                self.download_client_config(self.cluster, service)
            elif service_type == "HDFS":
                self.download_client_config(self.cluster, service)
                role = service.get_roles_by_type("NAMENODE")
                config = role[0].get_config("full")
                namenode = self.get_hostname_for_ref(role[0].hostRef)
                result["namenode"] = namenode
                result["namenode_http"] = "{0}:{1}".format(
                    namenode,
                    self.get_or_default(config["dfs_http_port"])
                )
            elif service_type == "HIVE":
                self.set_hive_warehouse_dir(self.cluster, service)
                self.download_client_config(self.cluster, service)
                hs2 = service.get_roles_by_type("HIVESERVER2")[0]
                rc_config = hs2.get_config("full")
                result["hive_warehouse_directory"] = self.get_or_default(
                    service.get_config("full")[0]["hive_warehouse_directory"])
                hostname = self.get_hostname_for_ref(hs2.hostRef)
                result["hs2"] = "{0}:{1}".format(hostname, self.get_or_default(
                    rc_config["hs2_thrift_address_port"]))

                # Get Metastore information
                ms = service.get_roles_by_type("HIVEMETASTORE")[0]
                rc_config = ms.get_config("full")
                result["metastore"] = "{0}:{1}".format(
                    self.get_hostname_for_ref(ms.hostRef),
                    self.get_or_default(rc_config["hive_metastore_port"])
                )
            elif service_type == "KUDU":
                # Service KUDU does not require a client configuration
                result["kudu_master"] = self.cm_host

        return result

    # TODO: This functionality should be more generally available to other infrastructure
    # code, rather than being quarantined in this script. See IMPALA-4367.
    @staticmethod
    def find_snapshot_file(snapshot_dir):
        """Given snapshot_directory, walks the directory tree until it finds a file
        matching the test-warehouse archive pattern."""
        for root, _, file_names in os.walk(snapshot_dir):
            for filename in fnmatch.filter(file_names, "test-warehouse-*-SNAPSHOT.tar.gz"):
                logger.info("Found Snapshot file {0}".format(filename))
                return os.path.join(root, filename)

    @timing
    def prepare(self):
        """Populate the environment of the process with the necessary values.

        In addition, it creates helper objects to run shell and SSH processes.
        """
        # Populate environment with required variables
        os.environ["HS2_HOST_PORT"] = self.config["hs2"]
        os.environ["HDFS_NN"] = self.config["namenode"]
        os.environ["IMPALAD"] = self.config["impalad"][0]
        os.environ["REMOTE_LOAD"] = "1"
        os.environ["HADOOP_USER_NAME"] = "hdfs"
        os.environ["TEST_WAREHOUSE_DIR"] = self.config["hive_warehouse_directory"]
        os.environ["KUDU_MASTER"] = self.config["kudu_master"]

        if self.options.snapshot_file is None:
            if "SNAPSHOT_DIR" in os.environ:
                snapshot_dir = os.environ["SNAPSHOT_DIR"]
            else:
                snapshot_dir = "{0}/testdata/test-warehouse-SNAPSHOT".format(
                    os.getenv("WORKSPACE"))
            if not os.path.isdir(snapshot_dir):
                err_msg = 'Snapshot directory "{0}" is not a valid directory'
                logger.error(err_msg.format(snapshot_dir))
                raise OSError("Could not find test-warehouse snapshot file.")

            logger.info("Snapshot directory: {0}".format(snapshot_dir))
            self.snapshot_file = self.find_snapshot_file(snapshot_dir)
        else:
            self.snapshot_file = self.options.snapshot_file

        # Prepare shortcuts for connecting to remote services
        self.gtw_ssh = ssh.bake("{0}@{1}".format(self.options.ssh_user, self.gateway),
                                "-oStrictHostKeyChecking=no",
                                "-oUserKnownHostsFile=/dev/null",
                                t=True, _out=tee, _err=tee)

        self.beeline = sh.beeline.bake(silent=False, outputformat="csv2", n="impala",
                                       u="jdbc:hive2://{0}/default".format(
                                           self.config["hs2"]))

        self.load_test_warehouse = sh.Command(
            "{0}/testdata/bin/load-test-warehouse-snapshot.sh".format(
                self.impala_home)).bake(
            _out=tee, _err=tee)

        self.create_load_data = sh.Command(
            "{0}/testdata/bin/create-load-data.sh".format(self.impala_home))

        self.main_impalad = self.config["impalad"][0]
        self.impala_shell = sh.Command("impala-shell.sh").bake(i=self.main_impalad,
                                                               _out=tee, _err=tee)

        self.python = sh.Command("impala-python").bake(u=True)
        self.compute_stats = sh.Command(
            "{0}/testdata/bin/compute-table-stats.sh".format(self.impala_home)).bake(
            _out=tee, _err=tee)

    @timing
    def load(self):
        """This method performs the actual data load. First it removes any known artifacts
        from the remote location. Next it drops potentially existing database from the
        Hive Metastore. Now, it invokes the load-test-warehouse-snapshot.sh and
        create-load-data.sh scripts with the appropriate parameters. The most important
        paramters are implicitly passed to the scripts as environment variables pointing
        to the remote HDFS, Hive and Impala.
        """
        exploration_strategy = self.options.exploration_strategy

        logger.info("Removing other databases")
        dblist = self.beeline(e="show databases;", _err=tee).stdout
        database_list = dblist.split()[1:]  # The first element is the header string
        for db in database_list:
            if db.strip() != "default":
                logger.debug("Dropping database %s", db)
                self.impala_shell(q="drop database if exists {0} cascade;".format(db))

        logger.info("Invalidating metadata in Impala")
        self.impala_shell(q="invalidate metadata;")

        logger.info("Removing previous remote {0}".format(
            self.config["hive_warehouse_directory"]))
        r = sh.hdfs.dfs("-rm", "-r", "-f", "{0}".format(
            self.config["hive_warehouse_directory"]))

        logger.info("Expunging HDFS trash")
        r = sh.hdfs.dfs("-expunge")

        logger.info("Uploading test warehouse snapshot")
        self.load_test_warehouse(self.snapshot_file)

        # TODO: We need to confirm that if we change any permissions, that we don't
        # affect any running tests. See IMPALA-4375.
        logger.info("Changing warehouse ownership")
        r = sh.hdfs.dfs("-chown", "-R", "impala:hdfs", "{0}".format(
            self.config["hive_warehouse_directory"]))
        sh.hdfs.dfs("-chmod", "-R", "g+rwx", "{0}".format(
            self.config["hive_warehouse_directory"]))
        sh.hdfs.dfs("-chmod", "1777", "{0}".format(
            self.config["hive_warehouse_directory"]))

        logger.info("Calling create_load_data.sh")
        # The $USER variable is used in the create-load-data.sh script for beeline
        # impersonation.
        new_env = os.environ.copy()
        new_env["LOGNAME"] = "impala"
        new_env["USER"] = "impala"
        new_env["USERNAME"] = "impala"

        # Regardless of whether we are in fact skipping the snapshot load or not,
        # we nonetheless always pass -skip_snapshot_load to create-load-data.sh.
        # This is because we have already loaded the snapshot earlier in this
        # script, so we don't want create-load-data.sh to invoke
        # load-test-warehouse-snapshot.sh again.
        #
        # It would actually be nice to be able to skip the snapshot load, but
        # because of the existing messiness of create-load-data.sh, we can't.
        # This invocation...
        #
        #    $ create-load-data.sh -skip_snapshot_load -exploration_strategy core
        #
        # ...results in this error:
        #
        #    Creating /test-warehouse HDFS directory \
        #    (logging to create-test-warehouse-dir.log)... FAILED
        #    'hadoop fs -mkdir /test-warehouse' failed. Tail of log:
        #    Log for command 'hadoop fs -mkdir /test-warehouse'
        #    mkdir: `/test-warehouse': File exists
        #
        # Similarly, even though we might pass in "core" as the exploration strategy,
        # because we aren't loading a metadata snapshot (i.e., -skip_metadata_load is
        # false), an exhaustive dataload will always be done. This again is the result
        # of logic in create-load-data.sh, which itself ignores the value passed in
        # for -exploration_strategy.
        #
        # See IMPALA-4399: "create-load-data.sh has bitrotted to some extent, and needs
        #                   to be cleaned up"
        create_load_data_args = ["-skip_snapshot_load", "-cm_host", self.cm_host,
                                 "-snapshot_file", self.snapshot_file,
                                 "-exploration_strategy", exploration_strategy]

        self.create_load_data(*create_load_data_args, _env=new_env, _out=tee, _err=tee)

        sh.hdfs.dfs("-chown", "-R", "impala:hdfs", "{0}".format(
            self.config["hive_warehouse_directory"]))

        logger.info("Re-load HBase data")
        # Manually load the HBase data last.
        self.python("{0}/bin/load-data.py".format(self.impala_home),
                    "--hive_warehouse_dir={0}".format(
                        self.config["hive_warehouse_directory"]),
                    "--table_formats=hbase/none",
                    "--hive_hs2_hostport={0}".format(self.config["hs2"]),
                    "--hdfs_namenode={0}".format(self.config["namenode"]),
                    "--exploration_strategy={0}".format(exploration_strategy),
                    workloads="functional-query",
                    force=True,
                    impalad=self.main_impalad,
                    _env=new_env,
                    _out=tee,
                    _err=tee)

        self.compute_stats()
        logger.info("Load data finished")

    # TODO: Should this be refactored out of this script? It has nothing to do with
    # data loading per se. If tests rely on the environment on the client being set
    # a certain way -- as in the prepare() method -- we may need to find another way
    # to deal with that. See IMPALA-4376.
    @timing
    def test(self):
        """Execute Impala's end-to-end tests against a remote cluster. All configuration
        paramters are picked from the cluster configuration that was fetched via the
        CM API."""

        # TODO: Running tests via runtest.py is currently not working against a remote
        # cluster (although running directly via py.test seems to work.) This method
        # may be refactored out of this file under IMPALA-4376, so for the time being,
        # raise a NotImplementedError.
        raise NotImplementedError

        # Overwrite the username to match the service user on the remote system and deal
        # with the assumption that in the local development environment the current user
        # is HDFS superuser as well.
        new_env = os.environ.copy()
        new_env["LOGNAME"] = "impala"
        new_env["USER"] = "impala"
        new_env["USERNAME"] = "impala"

        strategy = self.options.exploration_strategy
        logger.info("Running tests with exploration strategy {0}".format(strategy))
        run_tests = sh.Command("{0}/tests/run-tests.py".format(self.impala_home))
        run_tests("--skip_local_tests",
                  "--exploration_strategy={0}".format(strategy),
                  "--workload_exploration_strategy=functional-query:{0}".format(strategy),
                  "--namenode_http_address={0}".format(self.config["namenode_http"]),
                  "--hive_server2={0}".format(self.config["hs2"]),
                  "--metastore_server={0}".format(self.config["metastore"]),
                  "query_test",
                  maxfail=10,
                  impalad=",".join(self.config["impalad"]),
                  _env=new_env,
                  _out=tee,
                  _err=tee)


def parse_args():
    parser = OptionParser()
    parser.add_option("--snapshot-file", default=None,
                      help="Path to the test-warehouse archive")
    parser.add_option("--cm-user", default="admin", help="Cloudera Manager admin user")
    parser.add_option("--cm-pass", default="admin",
                      help="Cloudera Manager admin user password")
    parser.add_option("--gateway", default=None,
                      help=("Gateway host to upload the data from. If not set, uses the "
                            "CM host as gateway."))
    parser.add_option("--ssh-user", default="jenkins",
                      help=("System user on the remote machine with passwordless "
                            "SSH configured."))
    parser.add_option("--no-load", action="store_false", default=True, dest="load",
                      help="Do not try to load the snapshot")
    parser.add_option("--exploration-strategy", default="core")
    parser.add_option("--test", action="store_true", default=False,
                      help="Run end-to-end tests against cluster")
    parser.set_usage("remote_data_load.py [options] cm_host")

    options, args = parser.parse_args()

    try:
        return options, args[0]  # args[0] is the cm_host
    except IndexError:
        logger.error("You must supply the cm_host.")
        parser.print_usage()
        raise

def main(cm_host, options):
    """
    Load data to a remote cluster (and/or run tests) as specified.

    Args:
        cm_host: FQDN or IP of the CM host machine
        options: an optparse 'options' instance containing RemoteDataLoad
                 values (though any object with the correct .attributes, e.g.
                 a collections.namedtuple instance, would also work)
    """
    rd = RemoteDataLoad(cm_host, options)

    if options.load:
        rd.load()
    if options.test:
        rd.test()


if __name__ == "__main__":
    options, cm_host = parse_args()
    main(cm_host=cm_host, options=options)