mirror of
https://github.com/apache/impala.git
synced 2025-12-19 09:58:28 -05:00
daaf73a7c2
This patch implements 'wait_for_finished_timeout', 'wait_for_admission_control', and 'get_admission_result' for ImpylaHS2Client. This patch also changes the behavior of ImpylaHS2Connection to produce several extra cursors aside from self.__cursor for 'execute' call that supplies user argument and each 'execute_async' to make issuing multiple concurrent queries possible. Note that each HS2 cursor opens its own HS2 Session. Therefore, this patch breaks the assumption that an ImpylaHS2Connection is always under a single HS2 Session (see HIVE-11402 and HIVE-14247 on why concurrent query with shared HS2 Session is problematic). However, they do share the same query options stored at self.__query_options. In practice, most Impala tests do not care about running concurrent queries under a single HS2 session but only require them to use the same query options. The following additions are new for both BeeswaxConnection and ImpylaHS2Connection: - Add method 'log_client' for convenience. - Implement consistent query state mapping and checking through several accessor methods. - Add methods 'wait_for_impala_state' and 'wait_for_any_impala_state' that use 'get_impala_exec_state' method internally. - Add 'fetch_profile_after_close' parameter to 'close_query' method. If True, 'close_query' will return the query profile after closing the query. - Add 'discard_results' parameter for 'fetch' method. This can save time parsing results if the test does not care about the result. Reuse existing op_handle_to_query_id and add new session_handle_to_session_id to parse HS2 TOperationHandle.operationId.guid and TSessionHandle.sessionId.guid respectively. To show that ImpylaHS2Connection is on par with BeeswaxConnection, this patch refactors test_admission_controller.py to test using HS2 protocol by default. Test that does raw HS2 RPC (require capabilities from HS2TestSuite) is separated out into a new TestAdmissionControllerRawHS2 class and stays using beeswax protocol by default. All calls to copy.copy is replaced with copy.deepcopy for safety. Testing: - Pass exhaustive tests. Change-Id: I9ac07732424c16338e060c9392100b54337f11b8 Reviewed-on: http://gerrit.cloudera.org:8080/22362 Reviewed-by: Impala Public Jenkins <impala-public-jenkins@cloudera.com> Tested-by: Impala Public Jenkins <impala-public-jenkins@cloudera.com>
90 lines
3.4 KiB
Python
90 lines
3.4 KiB
Python
# 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.
|
|
#
|
|
# Thrift utility functions
|
|
from __future__ import absolute_import, division, print_function
|
|
import getpass
|
|
import sasl
|
|
import struct
|
|
|
|
from thrift.transport.TSocket import TSocket
|
|
from thrift.transport.TTransport import TBufferedTransport
|
|
from thrift_sasl import TSaslClientTransport
|
|
|
|
|
|
def create_transport(host, port, service, transport_type="buffered", user=None,
|
|
password=None, use_ssl=False, ssl_cert=None):
|
|
"""
|
|
Create a new Thrift Transport based on the requested type.
|
|
Supported transport types:
|
|
- buffered, returns simple buffered transport
|
|
- plain_sasl, return a SASL transport with the PLAIN mechanism
|
|
- kerberos, return a SASL transport with the GSSAPI mechanism
|
|
|
|
If use_ssl is True, the connection will use SSL, optionally using the file at ssl_cert
|
|
as the CA cert.
|
|
"""
|
|
port = int(port)
|
|
if use_ssl:
|
|
from thrift.transport import TSSLSocket
|
|
if ssl_cert is None:
|
|
sock = TSSLSocket.TSSLSocket(host, port, validate=False)
|
|
else:
|
|
sock = TSSLSocket.TSSLSocket(host, port, validate=True, ca_certs=ssl_cert)
|
|
# Set allowed SSL / TLS protocols to a permissive set to connect to any Impala server.
|
|
import ssl
|
|
sock.SSL_VERSION = ssl.PROTOCOL_SSLv23
|
|
else:
|
|
sock = TSocket(host, port)
|
|
if transport_type.lower() == "buffered":
|
|
return TBufferedTransport(sock)
|
|
|
|
# Set defaults for LDAP connections
|
|
if transport_type.lower() == "plain_sasl":
|
|
if user is None: user = getpass.getuser()
|
|
if password is None: password = ""
|
|
|
|
# Initializes a sasl client
|
|
def sasl_factory():
|
|
sasl_client = sasl.Client()
|
|
sasl_client.setAttr("host", host)
|
|
sasl_client.setAttr("service", service)
|
|
if transport_type.lower() == "plain_sasl":
|
|
sasl_client.setAttr("username", user)
|
|
sasl_client.setAttr("password", password)
|
|
sasl_client.init()
|
|
return sasl_client
|
|
if transport_type.lower() == "plain_sasl":
|
|
return TSaslClientTransport(sasl_factory, "PLAIN", sock)
|
|
else:
|
|
# GSSASPI is the underlying mechanism used by kerberos to authenticate.
|
|
return TSaslClientTransport(sasl_factory, "GSSAPI", sock)
|
|
|
|
|
|
def op_handle_to_query_id(t_op_handle):
|
|
if t_op_handle is None or t_op_handle.operationId is None:
|
|
return None
|
|
# This should use the same logic as in ImpalaServer::THandleIdentifierToTUniqueId().
|
|
return "%016x:%016x" % struct.unpack("QQ", t_op_handle.operationId.guid)
|
|
|
|
|
|
def session_handle_to_session_id(t_session_op_handle):
|
|
if t_session_op_handle is None or t_session_op_handle.sessionId is None:
|
|
return None
|
|
# This should use the same logic as in ImpalaServer::THandleIdentifierToTUniqueId().
|
|
return "%016x:%016x" % struct.unpack("QQ", t_session_op_handle.sessionId.guid)
|