| # Licensed to the Apache Software Foundation (ASF) under one |
| # or more contributor license agreements. See the NOTICE file |
| # distributed with this work for additional information |
| # regarding copyright ownership. The ASF licenses this file |
| # to you 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. |
| # |
| # Basic object model of a Impala Services (impalad + statestored). Provides a way to |
| # programatically interact with the services and perform operations such as querying |
| # the debug webpage, getting metric values, or creating client connections. |
| |
| from __future__ import absolute_import, division, print_function |
| from collections import defaultdict |
| from datetime import datetime |
| import json |
| import logging |
| import os |
| import re |
| import socket |
| import subprocess |
| from time import sleep, time |
| |
| import requests |
| from thrift.transport.TSocket import TSocket |
| from thrift.transport.TTransport import TBufferedTransport |
| |
| from tests.common.impala_connection import create_connection, create_ldap_connection |
| from tests.common.network import to_host_port, CERT_TO_CA_MAP |
| from tests.common.test_vector import BEESWAX, HS2, HS2_HTTP |
| |
| LOG = logging.getLogger('impala_service') |
| LOG.setLevel(level=logging.DEBUG) |
| |
| WEBSERVER_USERNAME = os.environ.get('IMPALA_WEBSERVER_USERNAME', None) |
| WEBSERVER_PASSWORD = os.environ.get('IMPALA_WEBSERVER_PASSWORD', None) |
| |
| |
| # Base class for all Impala services |
| # TODO: Refactor the retry/timeout logic into a common place. |
| class BaseImpalaService(object): |
| def __init__(self, hostname, webserver_interface, webserver_port, |
| webserver_certificate_file, ssl_client_ca_certificate_file): |
| self.hostname = hostname |
| self.webserver_interface = webserver_interface |
| if webserver_interface == "": |
| # If no webserver interface was specified, just use the hostname. |
| self.webserver_interface = hostname |
| self.webserver_port = webserver_port |
| self.webserver_certificate_file = webserver_certificate_file |
| self.ssl_client_ca_certificate_file = ssl_client_ca_certificate_file |
| self.webserver_username_password = None |
| if WEBSERVER_USERNAME is not None and WEBSERVER_PASSWORD is not None: |
| self.webserver_username_password = (WEBSERVER_USERNAME, WEBSERVER_PASSWORD) |
| |
| def open_debug_webpage(self, page_name, timeout=10, interval=1): |
| start_time = time() |
| |
| while (time() - start_time < timeout): |
| try: |
| protocol = "http" |
| if self.webserver_certificate_file != "": |
| protocol = "https" |
| host_port = to_host_port(self.webserver_interface, self.webserver_port) |
| url = "%s://%s/%s" % (protocol, host_port, page_name) |
| cert = self.webserver_certificate_file |
| # Instead of cert use its CA cert if available. |
| file_part = cert.split("/")[-1] |
| if file_part in CERT_TO_CA_MAP: |
| cert = cert.replace(file_part, CERT_TO_CA_MAP[file_part]) |
| return requests.get(url, verify=cert, auth=self.webserver_username_password) |
| except Exception as e: |
| LOG.info("Debug webpage not yet available: %s", str(e)) |
| sleep(interval) |
| assert 0, 'Debug webpage did not become available in expected time.' |
| |
| def read_debug_webpage(self, page_name, timeout=10, interval=1): |
| return self.open_debug_webpage(page_name, timeout=timeout, interval=interval).text |
| |
| def get_debug_webpage_json(self, page_name, timeout=10, interval=1): |
| """Returns the json for the given Impala debug webpage, eg. '/queries'""" |
| return json.loads(self.read_debug_webpage( |
| page_name + "?json", timeout=timeout, interval=interval)) |
| |
| def dump_debug_webpage_json(self, page_name, filename): |
| """Dumps the json for a given Impalad debug webpage to the specified file. |
| Prints the JSON with indenting to be somewhat human readable.""" |
| debug_json = self.get_debug_webpage_json(page_name) |
| with open(filename, "w") as json_out: |
| json.dump(debug_json, json_out, indent=2) |
| |
| def get_metric_value(self, metric_name, default_value=None): |
| """Returns the value of the the given metric name from the Impala debug webpage""" |
| return self.get_metric_values([metric_name], [default_value])[0] |
| |
| def get_metric_values(self, metric_names, default_values=None): |
| """Returns the value of the given metrics from the Impala debug webpage. If |
| default_values is provided and a metric is not present, the default value |
| is returned instead.""" |
| if default_values is None: |
| default_values = [None for m in metric_names] |
| assert len(metric_names) == len(default_values) |
| metrics = self.get_debug_webpage_json('jsonmetrics') |
| return [metrics.get(metric_name, default_value) |
| for metric_name, default_value in zip(metric_names, default_values)] |
| |
| def get_flag_current_value(self, flag): |
| """Returns the value of the the given flag name from the Impala /varz debug webpage. |
| If the flag does not exist it returns None.""" |
| varz = self.get_debug_webpage_json("varz") |
| for var in varz.get("flags"): |
| if var["name"] == flag: |
| return var["current"] |
| return None |
| |
| def get_flag_current_values(self): |
| """Returns the value of all flags from the Impala /varz debug webpage.""" |
| flags = dict() |
| varz = self.get_debug_webpage_json("varz") |
| for var in varz.get("flags"): |
| flags[var['name']] = var['current'] |
| return flags |
| |
| def wait_for_metric_value(self, metric_name, expected_value, timeout=10, interval=1, |
| allow_greater=False): |
| start_time = time() |
| total_wait = 0 |
| while (total_wait < timeout): |
| LOG.info("Getting metric: %s from %s:%s" % |
| (metric_name, self.hostname, self.webserver_port)) |
| value = None |
| try: |
| value = self.get_metric_value(metric_name) |
| except Exception as e: |
| LOG.error(e) |
| |
| # if allow_greater is True we wait until the metric value becomes >= the expected |
| # value. |
| if (value == expected_value) or (allow_greater and value >= expected_value): |
| LOG.info("Metric '{0}' has reached desired value: {1}. total_wait: {2}s".format( |
| metric_name, value, total_wait)) |
| return value |
| else: |
| LOG.info("Waiting for metric value '{0}'{1}{2}. Current value: {3}. " |
| "total_wait: {4}s".format( |
| metric_name, ('>=' if allow_greater else '='), expected_value, value, |
| total_wait)) |
| LOG.info("Sleeping %ds before next retry." % interval) |
| sleep(interval) |
| total_wait = time() - start_time |
| |
| LOG.info("Metric {0} did not reach value {1} in {2}s. Actual value was '{3}'. " |
| "total_wait: {4}s. Failing...".format( |
| metric_name, expected_value, timeout, value, total_wait)) |
| self.__metric_timeout_assert(metric_name, expected_value, timeout, value) |
| |
| def wait_for_page_ready(self, page_name, timeout=10, interval=1): |
| start_time = time() |
| total_wait = 0 |
| while (total_wait < timeout): |
| response = self.open_debug_webpage(page_name) |
| total_wait = time() - start_time |
| if response.status_code == requests.codes.not_found: |
| LOG.info( |
| "Debug webpage {}:{}/{} not yet available. Sleeping {}s before next retry" |
| .format(self.webserver_interface, self.webserver_port, page_name, |
| interval)) |
| sleep(interval) |
| continue |
| LOG.info( |
| "Debug webpage {}:{}/{} is ready. total_wait: {}s" |
| .format(self.webserver_interface, self.webserver_port, page_name, total_wait)) |
| return |
| |
| def __request_minidump(self, pid): |
| """ |
| Impala processes (impalad, catalogd, statestored) have a signal handler for |
| SIGUSR1 that dumps a minidump. This sends a SIGUSR1 to the specified pid |
| to trigger a minidump. Note that this will not work for processes that don't |
| implement a handler for SIGUSR1. This does not wait for the minidump to be |
| generated, so callers will need to consider this. |
| """ |
| cmd = ["kill", "-SIGUSR1", pid] |
| subprocess.check_output(cmd) |
| |
| def __metric_timeout_assert(self, metric_name, expected_value, timeout, actual_value): |
| """ |
| Helper function to dump diagnostic information for debugging a metric timeout and |
| then assert. |
| """ |
| impala_home = os.environ["IMPALA_HOME"] |
| log_dir = os.environ["IMPALA_LOGS_DIR"] |
| |
| # Create a diagnostic directory to contain all the relevent information |
| datetime_string = datetime.now().strftime("%Y%m%d_%H:%M:%S") |
| diag_dir = os.path.join(log_dir, "metric_timeout_diags_{0}".format(datetime_string)) |
| if not os.path.exists(diag_dir): |
| os.makedirs(diag_dir) |
| |
| # Read debug pages from the Web UI and dump them to files in the diagnostic |
| # directory |
| json_dir = os.path.join(diag_dir, "json") |
| if not os.path.exists(json_dir): |
| os.makedirs(json_dir) |
| json_diag_string = "Dumping debug webpages in JSON format...\n" |
| debug_pages = ["memz", "metrics", "queries", "sessions", "threadz", "rpcz"] |
| for debug_page in debug_pages: |
| json_filename = os.path.join(json_dir, "{0}.json".format(debug_page)) |
| self.dump_debug_webpage_json(debug_page, json_filename) |
| json_filename_rewritten = json_filename.replace(impala_home, "$IMPALA_HOME") |
| json_diag_string += \ |
| "Dumped {0} JSON to {1}\n".format(debug_page, json_filename_rewritten) |
| |
| # Requests a minidump for each running impalad/catalogd. The minidump will be |
| # written to the processes's minidump_path. For simplicity, we leave it there, |
| # as it will be preserved along with everything else in the log directory. |
| impalad_pids = subprocess.check_output(["pgrep", "impalad"], |
| universal_newlines=True).split("\n")[:-1] |
| catalogd_pids = subprocess.check_output(["pgrep", "catalogd"], |
| universal_newlines=True).split("\n")[:-1] |
| minidump_diag_string = "Dumping minidumps for impalads/catalogds...\n" |
| for pid in impalad_pids: |
| self.__request_minidump(pid) |
| minidump_diag_string += "Dumped minidump for Impalad PID {0}\n".format(pid) |
| for pid in catalogd_pids: |
| self.__request_minidump(pid) |
| minidump_diag_string += "Dumped minidump for Catalogd PID {0}\n".format(pid) |
| |
| # This is not a critical path, so sleep 30 seconds to be sure that the minidumps |
| # have been written. |
| sleep(30) |
| |
| # Now, fire the assert with the information about the dumps. This provides |
| # information about where to find the json files and other diagnostics. |
| # This is in the logs directory, so it should be packed up along with everything |
| # else for automated jobs. |
| assert_string = \ |
| "Metric {0} did not reach value {1} in {2}s. Actual value was '{3}'.\n" \ |
| .format(metric_name, expected_value, timeout, actual_value) |
| assert_string += json_diag_string |
| assert_string += minidump_diag_string |
| assert 0, assert_string |
| |
| def get_catalog_object_dump(self, object_type, object_name): |
| """ Gets the web-page for the given 'object_type' and 'object_name'.""" |
| return self.read_debug_webpage('catalog_object?object_type=%s&object_name=%s' % |
| (object_type, object_name)) |
| |
| def get_catalog_objects(self, excludes=['_impala_builtins']): |
| """ Returns a dictionary containing all catalog objects. Each entry's key is the fully |
| qualified object name and the value is a tuple of the form (type, version). |
| Does not return databases listed in the 'excludes' list.""" |
| catalog = self.get_debug_webpage_json('catalog') |
| objects = {} |
| for db_desc in catalog["databases"]: |
| db_name = db_desc["name"] |
| if db_name in excludes: |
| continue |
| db = self.get_catalog_object_dump('DATABASE', db_name) |
| objects[db_name] = ('DATABASE', self.extract_catalog_object_version(db)) |
| for table_desc in db_desc["tables"]: |
| table_name = table_desc["fqtn"] |
| table = self.get_catalog_object_dump('TABLE', table_name) |
| objects[table_name] = ('TABLE', self.extract_catalog_object_version(table)) |
| return objects |
| |
| def extract_catalog_object_version(self, thrift_txt): |
| """ Extracts and returns the version of the catalog object's 'thrift_txt' |
| representation.""" |
| result = re.search(r'catalog_version \(i64\) = (\d+)', thrift_txt) |
| assert result, 'Unable to find catalog version in object: ' + thrift_txt |
| return int(result.group(1)) |
| |
| |
| # Allows for interacting with an Impalad instance to perform operations such as creating |
| # new connections or accessing the debug webpage. |
| class ImpaladService(BaseImpalaService): |
| def __init__(self, hostname, webserver_interface="", external_interface="", |
| webserver_port=25000, beeswax_port=21000, krpc_port=27000, hs2_port=21050, |
| hs2_http_port=28000, webserver_certificate_file="", |
| ssl_client_ca_certificate_file=""): |
| super(ImpaladService, self).__init__( |
| hostname, webserver_interface, webserver_port, webserver_certificate_file, |
| ssl_client_ca_certificate_file) |
| self.external_interface = external_interface if external_interface else hostname |
| self.beeswax_port = beeswax_port |
| self.krpc_port = krpc_port |
| self.hs2_port = hs2_port |
| self.hs2_http_port = hs2_http_port |
| |
| def get_num_known_live_executors(self, timeout=30, interval=1, |
| include_shutting_down=True): |
| return self.get_num_known_live_backends(timeout=timeout, |
| interval=interval, |
| include_shutting_down=include_shutting_down, |
| only_executors=True) |
| |
| def get_num_known_live_backends(self, timeout=30, interval=1, |
| include_shutting_down=True, only_coordinators=False, only_executors=False): |
| LOG.debug("Getting num_known_live_backends from %s:%s" % |
| (self.hostname, self.webserver_port)) |
| result = self.get_debug_webpage_json('backends', timeout, interval) |
| count = 0 |
| for backend in result['backends']: |
| if backend['is_quiescing'] and not include_shutting_down: |
| continue |
| if only_coordinators and not backend['is_coordinator']: |
| continue |
| if only_executors and not backend['is_executor']: |
| continue |
| count += 1 |
| return count |
| |
| def get_executor_groups(self, timeout=30, interval=1): |
| """Returns a mapping from executor group name to a list of all KRPC endpoints of a |
| group's executors.""" |
| LOG.debug("Getting executor groups from %s:%s" % |
| (self.hostname, self.webserver_port)) |
| result = self.get_debug_webpage_json('backends', timeout, interval) |
| groups = defaultdict(list) |
| for backend in result['backends']: |
| groups[backend['executor_groups']].append(backend['krpc_address']) |
| return groups |
| |
| def get_queries_json(self, timeout=30, interval=1): |
| """Return the full JSON from the /queries page.""" |
| return json.loads( |
| self.read_debug_webpage('queries?json', timeout=timeout, interval=interval)) |
| |
| def get_query_locations(self): |
| # Returns a dictionary of the format <host_address, num_of_queries_running_there> |
| result = self.get_queries_json() |
| if result['query_locations'] is not None: |
| return dict([(loc["location"], loc["count"]) for loc in result['query_locations']]) |
| return None |
| |
| def get_in_flight_queries(self, timeout=30, interval=1): |
| """Returns the number of in flight queries.""" |
| return self.get_queries_json(timeout=timeout, interval=interval)['in_flight_queries'] |
| |
| def get_completed_queries(self, timeout=30, interval=1): |
| """Returns the number of completed queries.""" |
| result = self.get_debug_webpage_json('queries', timeout, interval) |
| return result['completed_queries'] |
| |
| def _get_pool_counter(self, pool_name, counter_name, timeout=30, interval=1): |
| """Returns the value of the field 'counter_name' in pool 'pool_name' or 0 if the pool |
| doesn't exist.""" |
| result = self.get_debug_webpage_json('admission', timeout, interval) |
| pools = result['resource_pools'] |
| for pool in pools: |
| if pool['pool_name'] == pool_name: |
| return pool[counter_name] |
| return 0 |
| |
| def get_num_queued_queries(self, pool_name, timeout=30, interval=1): |
| """Returns the number of queued queries in pool 'pool_name' or 0 if the pool doesn't |
| exist.""" |
| return self._get_pool_counter(pool_name, 'agg_num_queued', timeout, interval) |
| |
| def get_total_admitted_queries(self, pool_name, timeout=30, interval=1): |
| """Returns the total number of queries that have been admitted to pool 'pool_name' or |
| 0 if the pool doesn't exist.""" |
| return self._get_pool_counter(pool_name, 'total_admitted', timeout, interval) |
| |
| def get_num_running_queries(self, pool_name, timeout=30, interval=1): |
| """Returns the number of queries currently running in pool 'pool_name' or 0 if the |
| pool doesn't exist.""" |
| return self._get_pool_counter(pool_name, 'agg_num_running', timeout, interval) |
| |
| def get_num_in_flight_queries(self, timeout=30, interval=1): |
| LOG.info("Getting num_in_flight_queries from %s:%s" % |
| (self.hostname, self.webserver_port)) |
| result = self.read_debug_webpage('inflight_query_ids?raw', timeout, interval) |
| return None if result is None else len( |
| [line for line in result.split('\n') if line]) |
| |
| def wait_for_num_in_flight_queries(self, expected_val, timeout=10): |
| """Waits for the number of in-flight queries to reach a certain value""" |
| start_time = time() |
| while (time() - start_time < timeout): |
| num_in_flight_queries = self.get_num_in_flight_queries() |
| if num_in_flight_queries == expected_val: return True |
| sleep(1) |
| LOG.info("The number of in flight queries: %s, expected: %s" % |
| (num_in_flight_queries, expected_val)) |
| return False |
| |
| def wait_for_num_known_live_backends(self, expected_value, timeout=30, interval=1, |
| include_shutting_down=True, early_abort_fn=lambda: False): |
| """Poll the debug web server until the number of backends known by this service |
| reaches 'expected_value'. 'early_abort_fn' is called periodically and can |
| throw an exception if polling should be aborted early.""" |
| start_time = time() |
| while (time() - start_time < timeout): |
| early_abort_fn() |
| value = None |
| try: |
| value = self.get_num_known_live_backends(timeout=1, interval=interval, |
| include_shutting_down=include_shutting_down) |
| except Exception as e: |
| LOG.error(e) |
| if value == expected_value: |
| LOG.info("num_known_live_backends has reached value: %s" % value) |
| return value |
| else: |
| LOG.info("Waiting for num_known_live_backends=%s. Current value: %s" % |
| (expected_value, value)) |
| sleep(interval) |
| assert 0, 'num_known_live_backends did not reach expected value in time' |
| |
| def read_query_profile_page(self, query_id, timeout=10, interval=1): |
| """Fetches the raw contents of the query's runtime profile webpage. |
| Fails an assertion if Impala's webserver is unavailable or the query's |
| profile page doesn't exist.""" |
| return self.read_debug_webpage( |
| "query_profile?query_id=%s&raw" % (query_id), timeout=timeout, interval=interval) |
| |
| def get_query_status(self, query_id, timeout=10, interval=1): |
| """Gets the 'Query Status' section of the query's runtime profile.""" |
| page = self.read_query_profile_page(query_id, timeout=timeout, interval=interval) |
| status_line =\ |
| next((x for x in page.split('\n') if re.search('Query Status:', x)), None) |
| return status_line.split('Query Status:')[1].strip() |
| |
| def wait_for_query_state(self, client, query_handle, target_state, |
| timeout=10, interval=1): |
| """Keeps polling for the query's state using client in the given interval until |
| the query's state reaches the target state or the given timeout has been reached.""" |
| start_time = time() |
| while (time() - start_time < timeout): |
| try: |
| query_state = client.get_state(query_handle) |
| except Exception as e: |
| LOG.error("Exception while getting state of query {0}\n{1}".format( |
| client.handle_id(query_handle), str(e))) |
| if query_state == target_state: |
| return |
| sleep(interval) |
| assert target_state == query_state, \ |
| 'Query {0} did not reach query state in time target={1} actual={2}'.format( |
| client.handle_id(query_handle), target_state, query_state) |
| return |
| |
| def wait_for_query_status(self, query_id, expected_content, timeout=30, interval=1): |
| """Polls for the query's status in the query profile web page to contain the |
| specified content. Returns False if the timeout was reached before a successful |
| match, True otherwise.""" |
| start_time = time() |
| query_status = "" |
| while (time() - start_time < timeout): |
| try: |
| query_status = self.get_query_status( |
| query_id, timeout=timeout, interval=interval) |
| if query_status is None: |
| raise Exception("Could not find 'Query Status' section in profile of " |
| "query with id {}".format(query_id)) |
| except Exception: |
| pass |
| if expected_content in query_status: |
| return True |
| sleep(interval) |
| return False |
| |
| def use_ssl_for_clients(self): |
| return self.ssl_client_ca_certificate_file != "" |
| |
| def is_port_open(self, host, port): |
| try: |
| sock = socket.create_connection((host, port), timeout=1) |
| sock.close() |
| return True |
| except Exception: |
| return False |
| |
| def webserver_port_is_open(self): |
| return self.is_port_open(self.webserver_interface, self.webserver_port) |
| |
| def create_beeswax_client(self, use_kerberos=False): |
| """Creates a new beeswax client connection to the impalad. |
| DEPRECATED: Use create_hs2_client() instead.""" |
| client = create_connection(to_host_port(self.external_interface, self.beeswax_port), |
| use_kerberos, BEESWAX, use_ssl=self.use_ssl_for_clients()) |
| client.connect() |
| return client |
| |
| def beeswax_port_is_open(self): |
| """Test if the beeswax port is open. Does not need to authenticate.""" |
| # Check if the port is open first to avoid chatty logging of Thrift connection. |
| if not self.is_port_open(self.external_interface, self.beeswax_port): return False |
| |
| try: |
| # The beeswax client will connect successfully even if not authenticated. |
| client = self.create_beeswax_client() |
| client.close() |
| return True |
| except Exception as e: |
| return False |
| |
| def create_ldap_beeswax_client(self, user, password, use_ssl=False): |
| client = create_ldap_connection(to_host_port(self.hostname, self.beeswax_port), |
| user=user, password=password, use_ssl=use_ssl) |
| client.connect() |
| return client |
| |
| def create_hs2_client(self, user=None): |
| """Creates a new HS2 client connection to the impalad""" |
| client = create_connection('%s:%d' % (self.external_interface, self.hs2_port), |
| protocol=HS2, user=user, |
| use_ssl=self.use_ssl_for_clients()) |
| client.connect() |
| return client |
| |
| def hs2_port_is_open(self): |
| """Test if the HS2 port is open. Does not need to authenticate.""" |
| # Check if the port is open first to avoid chatty logging of Thrift connection. |
| if not self.is_port_open(self.external_interface, self.hs2_port): return False |
| |
| # Impyla will try to authenticate as part of connecting, so preserve previous logic |
| # that uses the HS2 thrift code directly. |
| try: |
| sock = TSocket(self.external_interface, self.hs2_port) |
| transport = TBufferedTransport(sock) |
| transport.open() |
| transport.close() |
| return True |
| except Exception as e: |
| LOG.info(e) |
| return False |
| |
| def hs2_http_port_is_open(self): |
| # Only check if the port is open, do not create Thrift transport. |
| return self.is_port_open(self.external_interface, self.hs2_http_port) |
| |
| def create_client(self, protocol): |
| """Creates a new client connection for given protocol to this impalad""" |
| port = self.hs2_port |
| if protocol == HS2_HTTP: |
| port = self.hs2_http_port |
| if protocol == BEESWAX: |
| LOG.warning('beeswax protocol is deprecated.') |
| port = self.beeswax_port |
| client = create_connection(to_host_port(self.external_interface, port), |
| protocol=protocol) |
| client.connect() |
| return client |
| |
| def create_client_from_vector(self, vector): |
| """A shorthand for create_client with test vector as input. |
| Vector must have 'protocol' and 'exec_option' dimension. |
| Return a client of specified 'protocol' and with cofiguration 'exec_option' set.""" |
| client = self.create_client(protocol=vector.get_value('protocol')) |
| client.set_configuration(vector.get_exec_option_dict()) |
| return client |
| |
| |
| # Allows for interacting with the StateStore service to perform operations such as |
| # accessing the debug webpage. |
| class StateStoredService(BaseImpalaService): |
| def __init__(self, hostname, webserver_interface, webserver_port, |
| webserver_certificate_file, ssl_client_ca_certificate_file, service_port): |
| super(StateStoredService, self).__init__( |
| hostname, webserver_interface, webserver_port, webserver_certificate_file, |
| ssl_client_ca_certificate_file) |
| self.service_port = service_port |
| |
| def wait_for_live_subscribers(self, num_subscribers, timeout=15, interval=1): |
| self.wait_for_metric_value('statestore.live-backends', num_subscribers, |
| timeout=timeout, interval=interval) |
| |
| def get_statestore_service_port(self): |
| return self.service_port |
| |
| |
| # Allows for interacting with the Catalog service to perform operations such as |
| # accessing the debug webpage. |
| class CatalogdService(BaseImpalaService): |
| def __init__(self, hostname, webserver_interface, webserver_port, |
| webserver_certificate_file, ssl_client_ca_certificate_file, service_port): |
| super(CatalogdService, self).__init__( |
| hostname, webserver_interface, webserver_port, webserver_certificate_file, |
| ssl_client_ca_certificate_file) |
| self.service_port = service_port |
| |
| def get_catalog_version(self, timeout=10, interval=1): |
| """ Gets catalogd's latest catalog version. Retry for 'timeout' |
| seconds, sleeping 'interval' seconds between tries. If the |
| version cannot be obtained, this method fails.""" |
| start_time = time() |
| while (time() - start_time < timeout): |
| try: |
| info = self.get_debug_webpage_json('catalog') |
| if "version" in info: return info['version'] |
| except Exception: |
| LOG.info('Catalogd version not yet available.') |
| sleep(interval) |
| assert False, 'Catalog version not ready in expected time.' |
| |
| def get_catalog_service_port(self): |
| return self.service_port |
| |
| def verify_table_metadata_loaded(self, db, table, expect_loaded=True, timeout_s=10): |
| page_url = "table_metrics?json&name=%s.%s" % (db, table) |
| start_time = time() |
| while (time() - start_time < timeout_s): |
| response = self.open_debug_webpage(page_url) |
| assert response.status_code == requests.codes.ok |
| response_json = json.loads(response.text) |
| assert "table_metrics" in response_json |
| table_metrics = response_json["table_metrics"] |
| if expect_loaded: |
| if "Table not yet loaded" not in table_metrics: |
| return |
| LOG.info("Table {0}.{1} is not loaded".format(db, table)) |
| else: |
| if "Table not yet loaded" in table_metrics: |
| return |
| LOG.info("Table {0}.{1} is loaded".format(db, table)) |
| sleep(1) |
| assert False, "Timeout waiting table {0}.{1} to be {2}".format( |
| db, table, "loaded" if expect_loaded else "not loaded") |
| |
| class AdmissiondService(BaseImpalaService): |
| def __init__(self, hostname, webserver_interface, webserver_port, |
| webserver_certificate_file, ssl_client_ca_certificate_file): |
| super(AdmissiondService, self).__init__( |
| hostname, webserver_interface, webserver_port, webserver_certificate_file, |
| ssl_client_ca_certificate_file) |
