_LOGGER = logging.getLogger(__name__)
COULD_NOT_REMOVE = 'Could not remove catalog'
def deploy_files(filenames, local_dir, remote_dir, user_group, mode=0600):
def gather_catalogs(local_config_dir, allow_overwrite=False):
def validate(filenames):
def add(name=None):
def remove(name):
def remove_file(path): ALL_CONFIG = [CONFIG_PROPERTIES, LOG_PROPERTIES, JVM_CONFIG, NODE_PROPERTIES]
_LOGGER = logging.getLogger(__name__)
def deploy(rolename=None):
def gather_config_directory():
def deploy_config_directory(tarfile):
def configuration_fetch(file_name, config_destination, should_warn=True):
def configuration_show(file_name, should_warn=True):
def show(config_type=None): _LOGGER = logging.getLogger(__name__)
def install(local_path):
def check_if_valid_rpm(local_path):
def deploy_install(local_path):
def deploy_upgrade(local_path):
def deploy_action(local_path, rpm_action):
def deploy(local_path=None):
def _rpm_install(package_path):
def _nodeps_rpm_option():
def rpm_install(rpm_name):
def _rpm_path(rpm_filename):
def rpm_upgrade(rpm_name):
def _rpm_upgrade(package_name):
def uninstall(rpm_name):
def rpm_uninstall(package_name):
def is_rpm_installed(package_name):
def _rpm_uninstall(package_name): class PrestoClient:
    def __init__(self, server, user, coordinator_config=None):
        # immutable stuff
        self.server = server
        self.user = user
        if (coordinator_config is None):
            coordinator_config = PrestoConfig.coordinator_config()
        self.coordinator_config = coordinator_config
        self.port = PrestoClient._get_configured_port(self.coordinator_config)

        # mutable stuff
        self.ca_file_path = ""
        self.keystore_data = ""
        self.rows = []
        self.next_uri = ''
        self.response_from_server = {}

    @staticmethod
    def _remove_silently(path):
        try:
            os.remove(path)
        except:
            pass

    def close(self):
        PrestoClient._remove_silently(self.ca_file_path)

    def _clear_old_results(self):
        if self.rows:
            self.rows = []

        if self.next_uri:
            self.next_uri = ''

        if self.response_from_server:
            self.response_from_server = {}

    def run_sql(self, sql, schema="default", catalog="hive"):
        """
        Execute a query connecting to Presto server using passed parameters.

        Args:
            sql: SQL query to be executed
            schema: Presto schema to be used while executing query
                (default=default)
            catalog: Catalog to be used by the server

        Returns:
            list of rows or None if client was unable to connect to Presto
        """
        status = self._execute_query(sql, schema, catalog)
        if status:
            return self._get_rows()
        else:
            return None

    def _execute_query(self, sql, schema, catalog):
        if not sql:
            raise InvalidArgumentError("SQL query missing")

        if not self.server:
            raise InvalidArgumentError("Server IP missing")

        if not self.user:
            raise InvalidArgumentError("Username missing")

        self._clear_old_results()

        headers = {"X-Presto-Catalog": catalog,
                   "X-Presto-Schema": schema,
                   "X-Presto-User": self.user,
                   "X-Presto-Source": "presto-admin"}
        answer = ''
        try:
            _LOGGER.info("Connecting to server at: " + self.server +
                         ":" + str(self.port) + " as user " + self.user +
                         " to execute query " + sql)
            conn = self._get_connection()
            self._add_auth_headers(headers)
            conn.request("POST", "/v1/statement", sql, headers)
            response = conn.getresponse()

            if response.status != 200:
                conn.close()
                _LOGGER.error("Connection error: " +
                              str(response.status) + " " + response.reason)
                return False

            answer = response.read()
            conn.close()

            self.response_from_server = json.loads(answer)
            _LOGGER.info("Query executed successfully: %s" % (sql))
            return True
        except (HTTPException, socket.error) as e:
            _LOGGER.error("Error connecting to presto server at: " +
                          self.server + ":" + str(self.port) + ' ' + e.message)
            return False
        except ValueError as e:
            _LOGGER.error('Error connecting to Presto server: ' + e.message +
                          ' error from server: ' + answer)
            raise e

    def _get_response_from(self, uri):
        """
        Sends a GET request to the Presto server at the specified next_uri
        and updates the response

        Remove the scheme and host/port from the uri; the connection itself
        has that information.
        """
        parts = list(urlparse.urlsplit(uri))
        parts[0] = None
        parts[1] = None
        location = urlparse.urlunsplit(parts)
        conn = self._get_connection()
        headers = {"X-Presto-User": self.user}
        self._add_auth_headers(headers)
        conn.request("GET", location, headers=headers)
        response = conn.getresponse()

        if response.status != 200:
            conn.close()
            _LOGGER.error("Error making GET request to %s: %s %s" %
                          (uri, response.status, response.reason))
            return False

        answer = response.read()
        conn.close()

        self.response_from_server = json.loads(answer)
        _LOGGER.info("GET request successful for uri: " + uri)
        return True

    def _build_results_from_response(self):
        """
        Build result from the response

        The reponse_from_server may contain up to 3 uri's.
        1. link to fetch the next packet of data ('nextUri')
        2. TODO: information about the query execution ('infoUri')
        3. TODO: cancel the query ('partialCancelUri').
        """
        if NEXT_URI_RESP in self.response_from_server:
            self.next_uri = self.response_from_server[NEXT_URI_RESP]
        else:
            self.next_uri = ""

        if DATA_RESP in self.response_from_server:
            if self.rows:
                self.rows.extend(self.response_from_server[DATA_RESP])
            else:
                self.rows = self.response_from_server[DATA_RESP]

    def _get_rows(self, num_of_rows=NUM_ROWS):
        """
        Get the rows returned from the query.

        The client sends GET requests to the server using the 'nextUri'
        from the previous response until the servers response does not
        contain anymore 'nextUri's.  When there is no 'nextUri' the query is
        finished

        Note that this can only be called once and does not page through
        the results.

        Parameters:
            num_of_rows: to be retrieved. 1000 by default
        """
        if num_of_rows == 0:
            return []

        self._build_results_from_response()

        if not self._get_next_uri():
            return []

        while self._get_next_uri():
            if not self._get_response_from(self._get_next_uri()):
                return []
            if (len(self.rows) <= num_of_rows):
                self._build_results_from_response()
        return self.rows

    def _get_next_uri(self):
        return self.next_uri

    def _get_connection(self):
        if self.coordinator_config.use_https():
            return self._get_https_connection()
        else:
            return HTTPConnection(self.server, self.port, False, URL_TIMEOUT_MS)

    @staticmethod
    def _get_configured_port(coordinator_config):
        if coordinator_config.use_https():
            return coordinator_config.get_https_port()
        else:
            return coordinator_config.get_http_port()

    def _get_https_connection(self):
        ca_file_path = self._get_pem()
        result = HTTPSCaCertConnection(
                self.server, self.port, None, None, ca_file_path, False, URL_TIMEOUT_MS)
        return result

    def _fetch_keystore_data(self):
        if not self.keystore_data:
            remote_keystore_path = self.coordinator_config.get_client_keystore_path()
            keystore_data = StringIO()
            get(remote_keystore_path, keystore_data, use_sudo=True)
            keystore_data.seek(0)
            self.keystore_data = keystore_data.getvalue()
        return self.keystore_data

    def _pem_string(self, der_bytes, type):
        result = "-----BEGIN %s-----\n" % type
        result += "\r\n".join(
            textwrap.wrap(base64.b64encode(der_bytes).decode('ascii'), 64))
        result += "\n-----END %s-----\n" % type
        return result

    def _write_pem_file(self, directory, der_bytes_list, type):
        prefix = os.path.join(directory,
                              '%s-' % type.lower().replace(' ', '-'))
        fd, pem_path = mkstemp('.pem', prefix)
        # https://www.digicert.com/ssl-support/pem-ssl-creation.htm
        with open(pem_path, 'w') as pem_file:
            for der_bytes in der_bytes_list:
                pem_file.write(self._pem_string(der_bytes, type))
        os.close(fd)
        return pem_path

    def _get_pem(self):
        keystore_data = self._fetch_keystore_data()

        keystore = jks.KeyStore.loads(
                keystore_data,
                self.coordinator_config.get_client_keystore_password())

        if len(keystore.private_keys.items()) == 1:
            _, private_key = keystore.private_keys.items()[0]
        else:
            private_key = self._get_private_key(keystore)
        if not self.ca_file_path:
            """
            Each member of the cert chain is a tuple (cert_type, cert_data)
            We only need to write the data out to the .PEM file.

            This usage is shown in the example in the README.md on github:
            https://github.com/kurtbrose/pyjks
            """
            self.ca_file_path = self._write_pem_file(
                    get_coordinator_directory(),
                    [cert[1] for cert in private_key.cert_chain], 'CERTIFICATE')

        return self.ca_file_path

    def _get_private_key(self, keystore):
        all_keys = ", ".join(keystore.private_keys.keys())
        try:
            alias = env.conf[CERTIFICATE_ALIAS]
        except KeyError:
            error('Multiple keys found in %s. Set %s in %s. Available aliases are %s' %
                  (self.coordinator_config.get_client_keystore_path(),
                   CERTIFICATE_ALIAS, get_topology_path(), all_keys))

        try:
            return keystore.private_keys[alias]
        except KeyError:
            error('No alias %s found in %s. Available aliases are %s' %
                  (alias, self.coordinator_config.get_client_keystore_path(),
                   all_keys))

    def _add_auth_headers(self, headers):
        if self.coordinator_config.use_ldap():
            if self.coordinator_config.use_ldap():
                auth_headers = self._create_auth_headers(
                    self.coordinator_config.get_ldap_user(),
                    self.coordinator_config.get_ldap_password())
                headers.update(auth_headers)
                _LOGGER.info("Using LDAP = %s" % self.coordinator_config.use_ldap())

    @staticmethod
    def _create_auth_headers(user, password):
        if not user:
            error('LDAP user (taken from %s in %s on the coordinator) cannot be null or empty' %
                  (LDAP_CLIENT_USER_KEY, os.path.join(REMOTE_CONF_DIR, CONFIG_PROPERTIES)))
            return {}
        if not password:
            error('LDAP password (taken from %s in %s on the coordinator) cannot be null or empty' %
                  (LDAP_CLIENT_PASSWORD_KEY, os.path.join(REMOTE_CONF_DIR, CONFIG_PROPERTIES)))
            return {}

        if ':' in user:
            error("LDAP user cannot contain ':': %s" % user)

        # base64 encode the username and password
        auth = base64.encodestring('%s:%s' % (user, password)).replace('\n', '')
        return {'Authorization': 'Basic %s' % auth} class StandaloneConfig(BaseConfig):
    def __init__(self):
        super(StandaloneConfig, self).__init__(get_topology_path(), _TOPOLOGY_CONFIG)

    @overrides
    def read_conf(self):
        conf = self._get_conf_from_file()
        config.fill_defaults(conf, DEFAULT_PROPERTIES)
        validate(conf)
        return conf

    def _get_conf_from_file(self):
        return config.get_conf_from_json_file(self.config_path)

    @overrides
    def is_config_loaded(self):
        return STANDALONE_CONFIG_LOADED in env and env[STANDALONE_CONFIG_LOADED]

    @overrides
    def set_config_loaded(self):
        env[STANDALONE_CONFIG_LOADED] = True

    @overrides
    def set_env_from_conf(self, conf):
        self.config = conf
        env.user = conf['username']
        env.port = conf['port']
        try:
            env.java8_home = conf['java8_home']
        except KeyError:
            env.java8_home = None
        env.roledefs['coordinator'] = [conf['coordinator']]
        env.roledefs['worker'] = conf['workers']
        env.roledefs['all'] = self._dedup_list(util.get_coordinator_role() +
                                               util.get_worker_role())

        env.hosts = env.roledefs['all'][:]
        env.conf = conf

    @staticmethod
    def _dedup_list(host_list):
        deduped_list = []
        for item in host_list:
            if item not in deduped_list:
                deduped_list.append(item)
        return deduped_list LOGGING_CONFIG_FILE_NAME = 'presto-admin-logging.ini'
LOGGING_CONFIG_FILE_DIRECTORIES = [
    os.path.join(prestoadmin.main_dir, 'prestoadmin')
]
LOG_DIR_ENV_VARIABLE = 'PRESTO_ADMIN_LOG_DIR'
CONFIG_DIR_ENV_VARIABLE = 'PRESTO_ADMIN_CONFIG_DIR'
LOCAL_CONF_DIR = '.prestoadmin'
DEFAULT_LOCAL_CONF_DIR = os.path.join(os.path.expanduser('~'), LOCAL_CONF_DIR)
TOPOLOGY_CONFIG_FILE = 'config.json'
COORDINATOR_DIR_NAME = 'coordinator'
WORKERS_DIR_NAME = 'workers'
CATALOG_DIR_NAME = 'catalog'
REMOTE_CONF_DIR = '/etc/presto'
REMOTE_CATALOG_DIR = os.path.join(REMOTE_CONF_DIR, 'catalog')
REMOTE_PACKAGES_PATH = '/opt/prestoadmin/packages'
DEFAULT_PRESTO_SERVER_LOG_FILE = '/var/log/presto/server.log'
DEFAULT_PRESTO_LAUNCHER_LOG_FILE = '/var/log/presto/launcher.log'
REMOTE_PLUGIN_DIR = '/usr/lib/presto/lib/plugin'
REMOTE_COPY_DIR = '/tmp'
CONFIG_PROPERTIES = "config.properties"
LOG_PROPERTIES = "log.properties"
JVM_CONFIG = "jvm.config"
NODE_PROPERTIES = "node.properties" def requires_config(config_class):
    def wrap(func):
        config_instance = config_class()
        func.pa_config_callback = config_instance.get_config

        @wraps(func)
        def wrapper(*args, **kwargs):
            if not config_instance.is_config_loaded():
                raise ConfigurationError('Required config not loaded at task '
                                         'execution time.')
            return func(*args, **kwargs)
        return wrapper
    return wrap class ConfigFileNotFoundError(ConfigurationError):
    def __init__(self, message='', config_path=''):
        super(ConfigFileNotFoundError, self).__init__(message)
        self.config_path = config_path class ConfigurationError(ExceptionWithCause):
    pass def get_host_list():
    return [host for host in env.hosts if host not in env.exclude_hosts] def get_coordinator_role():
    return env.roledefs['coordinator'] def get_catalog_directory():
    return os.path.join(get_config_directory(), CATALOG_DIR_NAME) def lookup_port(host):
    """
    Get the http port from config.properties http-server.http.port property
    if available.
    If the property is missing return default port 8080.
    If the file is missing or cannot parse the port number,
    throw ConfigurationError
    :param host:
    :return:
    """
    port = lookup_in_config('http-server.http.port', GENERAL_CONFIG_FILE, host)
    if not port:
        _LOGGER.info('Could not find property http-server.http.port.'
                     'Defaulting to 8080.')
        return 8080
    try:
        port = port.split('=', 1)[1]
        port = prestoadmin.util.validators.validate_port(port)
        _LOGGER.info('Looked up port ' + str(port) + ' on host ' +
                     host)
        return port
    except ConfigurationError as e:
        raise ConfigurationError(e.message +
                                 ' for property '
                                 'http-server.http.port on host ' +
                                 host + '.') def lookup_server_log_file(host):
    try:
        return lookup_string_config('node.server-log-file', NODE_CONFIG_FILE,
                                    host, DEFAULT_PRESTO_SERVER_LOG_FILE)
    except:
        return DEFAULT_PRESTO_SERVER_LOG_FILE def lookup_launcher_log_file(host):
    try:
        return lookup_string_config('node.launcher-log-file', NODE_CONFIG_FILE,
                                    host, DEFAULT_PRESTO_LAUNCHER_LOG_FILE)
    except:
        return DEFAULT_PRESTO_LAUNCHER_LOG_FILE def lookup_string_config(config_value, config_file, host, default=''):
    value = lookup_in_config(config_value, config_file, host)
    if value:
        return value.split('=', 1)[1]
    else:
        return default class VersionRange(object):
    """
    Represents a range of version numbers [min_version, max_version).
    The interval is right-open so that you can construct a numerically
    continuous list of versions like so:
    l = [VersionRange((0, 0), (0, 5)), VersionRange((0, 5), (1, 0))]
    and for all versions v where 0.0 <= v < 1.0 is contained in exactly one
    VersionRange in l.

    Continuity between version ranges can be checked using is_continuous.

    VersionRanges understand how to check if a Teradata version is contained
    in a VersionRange, but do no special handling to accomodate Teradata
    versions in their internal min_version and max_version members. I.e.,
    creating a VersionRange with a Teradata version will work, but __contains__
    will not work correctly. We don't currently need this, and hope not to.

    Note that the right-open interval representation of a version range does
    NOT allow the creation of a VersionRange that contains exactly one version.

    Note that empty intervals cannot be constructed as the serve no useful
    purpose. Specifically, we assert that min_version < max_version in the
    constructor.
    """

    def __init__(self, min_version, max_version, versioned_thing=None):
        # not pythonic, but bare ints screw things up.
        assert isinstance(min_version, tuple)
        assert isinstance(max_version, tuple)
        l = max(len(min_version), len(max_version))
        min_pad = VersionRange.pad_tuple(min_version, l, 0)
        max_pad = VersionRange.pad_tuple(max_version, l, 0)
        assert min_pad < max_pad
        self.min_version = min_version
        self.max_version = max_version
        self.versioned_thing = versioned_thing

    def __str__(self):
        return '[%s, %s) -> %s' % (
            '.'.join([str(c) for c in self.min_version]),
            '.'.join([str(c) for c in self.max_version]),
            self.versioned_thing)

    @staticmethod
    def strip_td_suffix(version):
        new_version = ()
        for component in version:
            if TD_VERSION.match(str(component)):
                new_last = component[:-1]
                new_version += (int(new_last),)
            elif component is not 't':
                new_version += (int(component),)

        return new_version

    @staticmethod
    def pad_tuple(tup, length, pad):
        assert len(tup) <= length
        result = list(tup)
        while len(result) < length:
            result.append(pad)
        return tuple(result)

    def zero_pad(self, other):
        """
        Pad out min_version, max_version, and other with zeroes to the length
        of the longest of the three. This allows subsequent comparisons to work
        as expected when tuples are of unequal length.
        Returns a tuple of tuples padded out to the same length
        """
        l = max(len(self.min_version), len(self.max_version), len(other))
        return (self.pad_tuple(self.min_version, l, 0),
                self.pad_tuple(self.max_version, l, 0),
                self.pad_tuple(other, l, 0))

    def __contains__(self, other):
        other = self.strip_td_suffix(other)
        other = tuple([int(component) for component in other])

        min_pad, max_pad, o_pad = self.zero_pad(other)
        return min_pad <= o_pad and o_pad < max_pad

    def is_continuous(self, next):
        min_pad, max_pad, next_min_pad = self.zero_pad(next.min_version)
        return max_pad == next_min_pad class VersionRangeList(object):
    """
    A VersionRangeList is a list of continuous, non-overlapping VersionRanges.
    This is guaranteed by calling VersionRange.is_continuous on all pairs of
    VersionRanges vr[i], vr[i + 1] in the list, which ensures that the list is
    both sorted in order of ascending version and that the interval
    [vr[0].min_version, vr[n].max_version) has no discontinuities.
    """

    def __init__(self, *range_list):
        if len(range_list) >= 2:
            for i in range(0, len(range_list) - 1):
                assert range_list[i].is_continuous(range_list[i + 1])

        self.range_list = range_list

    def __str__(self):
        return '\n'.join([str(vr) for vr in self.range_list])

    def for_version(self, version):
        for range in self.range_list:
            if version in range:
                return range.versioned_thing
        raise KeyError(version) def split_version(version_string):
    # We split on '.' and '-' because ancient tagged versions had the tag
    # delimited by a '-'
    return re.split('\.|-', version_string.strip()) def strip_tag(version):
    """
    Strip any parts of the version that are not numeric components or t's
    We leave the 't' on numeric components if it's present.
    ['1', '2', 'THREE'] -> (1, 2)
    ['1', 'TWO', '3'] -> (1, 3)
    ['0', '115t', 'SNAPSHOT'] -> (0, '115t')
    ['ZERO', '123t'] -> (123t)
    ['0', '148', 't'] => (0, 148, 't')
    ['0', '148', 't', 0, 1] => (0, 148, 't', 0, 1)
    ['0', '148', 't', 0, 1, 'SNAPSHOT'] => (0, 148, 't', 0, 1)
    ['0', '162', 'SNAPSHOT', 't', 'SNAPSHOT'] => (0, 162, 't')

    This checks the components of the version from least to most significant.

    :param version: something that can be sliced
    :return: a tuple containing only integer components or the letter t
    """

    result = list(version[:])
    result = [get_int_or_t(x) for x in result if is_int_or_t(x)]
    return tuple(result)
import cgi
import logging
import re
import sys
import urllib2
import urlparse
import util.filesystem
from contextlib import closing
from fabric.api import task, sudo, env
from fabric.context_managers import settings, hide
from fabric.decorators import runs_once, with_settings, parallel
from fabric.operations import run, os
from fabric.tasks import execute
from fabric.utils import warn, error, abort
from retrying import retry, RetryError
from prestoadmin import catalog
from prestoadmin import configure_cmds
from prestoadmin import package
from prestoadmin.prestoclient import PrestoClient
from prestoadmin.standalone.config import StandaloneConfig
from prestoadmin.util import constants
from prestoadmin.util.base_config import requires_config
from prestoadmin.util.exception import ConfigFileNotFoundError, ConfigurationError
from prestoadmin.util.fabricapi import get_host_list, get_coordinator_role
from prestoadmin.util.local_config_util import get_catalog_directory
from prestoadmin.util.remote_config_util import lookup_port, \
    lookup_server_log_file, lookup_launcher_log_file, lookup_string_config
from prestoadmin.util.version_util import VersionRange, VersionRangeList, \
    split_version, strip_tag
        {'http://node1/statement': [presto-main:0.97-SNAPSHOT, True]}
    """
    output = {}
    for row in node_info_row:
        if row:
            output[row[0]] = [row[1], state_transform(row[2])]

    _LOGGER.info('Node info: %s ', output)
    return output


def get_catalog_info_from(client):
    """
    Returns installed catalogs

    Parameters:
        client - client that executes the query

    Returns:
        comma delimited catalogs eg: tpch, hive, system
    """
    syscatalog = []
    catalog_info = execute_catalog_info_sql(client)
    for conn_info in catalog_info:
        if conn_info:
            syscatalog.append(conn_info[0])
    return ', '.join(syscatalog)


def is_server_up(status):
    if status:
        return 'Running'
    else:
        return 'Not Running'


def get_roles_for(host):
    roles = []
    for role in ['coordinator', 'worker']:
        if host in env.roledefs[role]:
            roles.append(role)
    return roles


def print_node_info(node_status, catalog_status):
    for k in node_status:
        print('\tNode URI(http): ' + str(k) +
              '\n\tPresto Version: ' + str(node_status[k][0]) +
              '\n\tNode status:    ' + str(node_status[k][1]))
        if catalog_status:
            print('\tCatalogs:     ' + catalog_status)


def get_ext_ip_of_node(client):
    node_properties_file = os.path.join(constants.REMOTE_CONF_DIR,
                                        'node.properties')
    with settings(hide('stdout')):
        node_uuid = sudo('sed -n s/^node.id=//p ' + node_properties_file)
    external_ip_row = execute_external_ip_sql(client, node_uuid)
    external_ip = ''
    if len(external_ip_row) > 1:
        warn_more_than_one_ip = 'More than one external ip found for ' + env.host + \
                                '. There could be multiple nodes associated with the same node.id'
        _LOGGER.debug(warn_more_than_one_ip)
        warn(warn_more_than_one_ip)
        return external_ip
    for row in external_ip_row:
        if row:
            external_ip = row[0]
    if not external_ip:
        _LOGGER.debug('Cannot get external IP for ' + env.host)
        external_ip = 'Unknown'
    return external_ip


def print_status_header(external_ip, server_status, host):
    print('Server Status:')
    print('\t%s(IP: %s, Roles: %s): %s' % (host, external_ip,
                                           ', '.join(get_roles_for(host)),
                                           is_server_up(server_status)))


@parallel
def collect_node_information():
    with closing(PrestoClient(get_coordinator_role()[0], env.user)) as client:
        with settings(hide('warnings')):
            error_message = check_presto_version()
        if error_message:
            external_ip = 'Unknown'
            is_running = False
        else:
            with settings(hide('warnings', 'aborts', 'stdout')):
                try:
                    external_ip = get_ext_ip_of_node(client)
                except:
                    external_ip = 'Unknown'
                try:
                    is_running = service('status')
                except:
                    is_running = False
        return external_ip, is_running, error_message


def get_status_from_coordinator():
    with closing(PrestoClient(get_coordinator_role()[0], env.user)) as client:
        try:
            coordinator_status = client.run_sql(SYSTEM_RUNTIME_NODES)
            catalog_status = get_catalog_info_from(client)
        except BaseException as e:
            # Just log errors that come from a missing port or anything else; if
            # we can't connect to the coordinator, we just want to print out a
            # minimal status anyway.
            _LOGGER.warn(e.message)
            coordinator_status = []
            catalog_status = []

        with settings(hide('running')):
            node_information = execute(collect_node_information,
                                       hosts=get_host_list())


        for host in get_host_list():