Browse Source

HUE-9374 [impala] Apply proxy endpoint to server URL

Ying Chen 5 years ago
parent
commit
2d6b75190f

+ 19 - 10
apps/impala/src/impala/conf.py

@@ -46,25 +46,32 @@ SERVER_PORT = Config(
   default=21050,
   default=21050,
   type=int)
   type=int)
 
 
+PROXY_ENDPOINT = Config(
+  key="proxy_endpoint",
+  help=_t("Endpoint of the Impala Proxy Server, "
+          "for example: '/endpoint'. Note that SERVER_PORT will be used when set."),
+  type=str,
+  default="")
+
 COORDINATOR_URL = Config(
 COORDINATOR_URL = Config(
   key="coordinator_url",
   key="coordinator_url",
   help=_t("URL of the Impala Coordinator Server."),
   help=_t("URL of the Impala Coordinator Server."),
   type=str,
   type=str,
   default="")
   default="")
 
 
-IMPALA_PRINCIPAL=Config(
+IMPALA_PRINCIPAL = Config(
   key='impala_principal',
   key='impala_principal',
   help=_t("Kerberos principal name for Impala. Typically 'impala/hostname.foo.com'."),
   help=_t("Kerberos principal name for Impala. Typically 'impala/hostname.foo.com'."),
   type=str,
   type=str,
   default="impala/%s" % socket.getfqdn())
   default="impala/%s" % socket.getfqdn())
 
 
-IMPERSONATION_ENABLED=Config(
+IMPERSONATION_ENABLED = Config(
   key='impersonation_enabled',
   key='impersonation_enabled',
   help=_t("Turn on/off impersonation mechanism when talking to Impala."),
   help=_t("Turn on/off impersonation mechanism when talking to Impala."),
   type=coerce_bool,
   type=coerce_bool,
   dynamic_default=is_impersonation_enabled)
   dynamic_default=is_impersonation_enabled)
 
 
-QUERYCACHE_ROWS=Config(
+QUERYCACHE_ROWS = Config(
   key='querycache_rows',
   key='querycache_rows',
   help=_t("Number of initial rows of a resultset to ask Impala to cache in order to"
   help=_t("Number of initial rows of a resultset to ask Impala to cache in order to"
           " support re-fetching them for downloading them."
           " support re-fetching them for downloading them."
@@ -120,31 +127,31 @@ SSL = ConfigSection(
   key='ssl',
   key='ssl',
   help=_t('SSL configuration for the server.'),
   help=_t('SSL configuration for the server.'),
   members=dict(
   members=dict(
-    ENABLED = Config(
+    ENABLED=Config(
       key="enabled",
       key="enabled",
       help=_t("SSL communication enabled for this server."),
       help=_t("SSL communication enabled for this server."),
       type=coerce_bool,
       type=coerce_bool,
       default=False
       default=False
     ),
     ),
-    CACERTS = Config(
+    CACERTS=Config(
       key="cacerts",
       key="cacerts",
       help=_t("Path to Certificate Authority certificates."),
       help=_t("Path to Certificate Authority certificates."),
       type=str,
       type=str,
       dynamic_default=default_ssl_cacerts,
       dynamic_default=default_ssl_cacerts,
     ),
     ),
-    KEY = Config(
+    KEY=Config(
       key="key",
       key="key",
       help=_t("Path to the private key file, e.g. /etc/hue/key.pem"),
       help=_t("Path to the private key file, e.g. /etc/hue/key.pem"),
       type=str,
       type=str,
       default=None
       default=None
     ),
     ),
-    CERT = Config(
+    CERT=Config(
       key="cert",
       key="cert",
       help=_t("Path to the public certificate file, e.g. /etc/hue/cert.pem"),
       help=_t("Path to the public certificate file, e.g. /etc/hue/cert.pem"),
       type=str,
       type=str,
       default=None
       default=None
     ),
     ),
-    VALIDATE = Config(
+    VALIDATE=Config(
       key="validate",
       key="validate",
       help=_t("Choose whether Hue should validate certificates received from the server."),
       help=_t("Choose whether Hue should validate certificates received from the server."),
       type=coerce_bool,
       type=coerce_bool,
@@ -216,7 +223,8 @@ DAEMON_API_PASSWORD = Config(
 
 
 DAEMON_API_PASSWORD_SCRIPT = Config(
 DAEMON_API_PASSWORD_SCRIPT = Config(
   key="daemon_api_password_script",
   key="daemon_api_password_script",
-  help=_t("Execute this script to produce the Impala Daemon Password. This will be used when `daemon_api_password` is not set."),
+  help=_t("Execute this script to produce the Impala Daemon Password. "
+          "This will be used when `daemon_api_password` is not set."),
   private=True,
   private=True,
   type=coerce_password_from_script,
   type=coerce_password_from_script,
   default=None
   default=None
@@ -231,7 +239,8 @@ DAEMON_API_USERNAME = Config(
 
 
 DAEMON_API_AUTH_SCHEME = Config(
 DAEMON_API_AUTH_SCHEME = Config(
   key="daemon_api_auth_scheme",
   key="daemon_api_auth_scheme",
-  help=_t("The authentication scheme to use with 'daemon_api_username' and 'daemon_api_password' when authenticating to the Impala Daemon UI, either 'digest' (default) or 'basic'."),
+  help=_t("The authentication scheme to use with 'daemon_api_username' and 'daemon_api_password' "
+          "when authenticating to the Impala Daemon UI, either 'digest' (default) or 'basic'."),
   private=False,
   private=False,
   default="digest"
   default="digest"
 )
 )

+ 4 - 3
apps/impala/src/impala/dbms.py

@@ -40,17 +40,18 @@ def get_query_server_config(connector=None):
   if connector and has_connectors():
   if connector and has_connectors():
     query_server = get_query_server_config_via_connector(connector)
     query_server = get_query_server_config_via_connector(connector)
   else:
   else:
-    server_port = get_hs2_http_port() if conf.USE_THRIFT_HTTP.get() else conf.SERVER_PORT.get()
+    server_port = get_hs2_http_port() if conf.USE_THRIFT_HTTP.get() and not conf.PROXY_ENDPOINT.get() else conf.SERVER_PORT.get()
     query_server = {
     query_server = {
         'server_name': 'impala',
         'server_name': 'impala',
         'dialect': 'impala',
         'dialect': 'impala',
         'server_host': conf.SERVER_HOST.get(),
         'server_host': conf.SERVER_HOST.get(),
         'server_port': server_port,
         'server_port': server_port,
         'principal': conf.IMPALA_PRINCIPAL.get(),
         'principal': conf.IMPALA_PRINCIPAL.get(),
-        'http_url': '%(protocol)s://%(host)s:%(port)s' % {
+        'http_url': '%(protocol)s://%(host)s:%(port)s%(cli_endpoint)s' % {
             'protocol': 'https' if conf.SSL.ENABLED.get() else 'http',
             'protocol': 'https' if conf.SSL.ENABLED.get() else 'http',
             'host': conf.SERVER_HOST.get(),
             'host': conf.SERVER_HOST.get(),
-            'port': server_port
+            'port': server_port,
+            'cli_endpoint': conf.PROXY_ENDPOINT.get()
           },
           },
         'impersonation_enabled': conf.IMPERSONATION_ENABLED.get(),
         'impersonation_enabled': conf.IMPERSONATION_ENABLED.get(),
         'querycache_rows': conf.QUERYCACHE_ROWS.get(),
         'querycache_rows': conf.QUERYCACHE_ROWS.get(),

+ 2 - 2
apps/impala/src/impala/impala_flags.py

@@ -31,7 +31,7 @@ _AUTHORIZED_PROXY_USER_CONFIG = '-authorized_proxy_user_config'
 _PRINCIPAL = '-principal'
 _PRINCIPAL = '-principal'
 _DEFAULT_QUERY_OPTIONS = '-default_query_options'
 _DEFAULT_QUERY_OPTIONS = '-default_query_options'
 _DEFAULT_TRANSACTIONAL_TYPE = 'default_transactional_type'
 _DEFAULT_TRANSACTIONAL_TYPE = 'default_transactional_type'
-_DEFAULT_HS2_HTTP_PORT='-hs2_http_port'
+_DEFAULT_HS2_HTTP_PORT = '-hs2_http_port'
 
 
 
 
 def reset():
 def reset():
@@ -104,7 +104,7 @@ def is_webserver_spnego_enabled():
   return get_conf().get(_WEBSERVER_REQUIRE_SPNEGO)
   return get_conf().get(_WEBSERVER_REQUIRE_SPNEGO)
 
 
 def get_hs2_http_port():
 def get_hs2_http_port():
-  return get_conf().get(_DEFAULT_HS2_HTTP_PORT, 26000)
+  return get_conf().get(_DEFAULT_HS2_HTTP_PORT, 28000)
 
 
 
 
 def _parse_impala_flags():
 def _parse_impala_flags():

+ 46 - 22
apps/impala/src/impala/tests.py

@@ -159,10 +159,10 @@ class TestImpalaIntegration(object):
     ]
     ]
 
 
     for query in queries:
     for query in queries:
-       resp = _make_query(cls.client, query, database='default', local=False, server_name='impala')
-       resp = wait_for_query_to_finish(cls.client, resp, max=180.0)
-       content = json.loads(resp.content)
-       assert_true(content['status'] == 0, resp.content)
+      resp = _make_query(cls.client, query, database='default', local=False, server_name='impala')
+      resp = wait_for_query_to_finish(cls.client, resp, max=180.0)
+      content = json.loads(resp.content)
+      assert_true(content['status'] == 0, resp.content)
 
 
     queries = ["""
     queries = ["""
       CREATE TABLE tweets (row_num INTEGER, id_str STRING, text STRING) STORED AS PARQUET;
       CREATE TABLE tweets (row_num INTEGER, id_str STRING, text STRING) STORED AS PARQUET;
@@ -179,10 +179,10 @@ class TestImpalaIntegration(object):
     """]
     """]
 
 
     for query in queries:
     for query in queries:
-       resp = _make_query(cls.client, query, database=cls.DATABASE, local=False, server_name='impala')
-       resp = wait_for_query_to_finish(cls.client, resp, max=180.0)
-       content = json.loads(resp.content)
-       assert_true(content['status'] == 0, resp.content)
+      resp = _make_query(cls.client, query, database=cls.DATABASE, local=False, server_name='impala')
+      resp = wait_for_query_to_finish(cls.client, resp, max=180.0)
+      content = json.loads(resp.content)
+      assert_true(content['status'] == 0, resp.content)
 
 
 
 
   @classmethod
   @classmethod
@@ -227,7 +227,8 @@ class TestImpalaIntegration(object):
 
 
     # Check that we multiple fetches get all the result set
     # Check that we multiple fetches get all the result set
     while len(results) < 5:
     while len(results) < 5:
-      content = fetch_query_result_data(self.client, response, n=len(results), server_name='impala') # We get less than 5 results most of the time, so increase offset
+      # We get less than 5 results most of the time, so increase offset
+      content = fetch_query_result_data(self.client, response, n=len(results), server_name='impala')
       results += content['results']
       results += content['results']
 
 
     assert_equal([1, 2, 3, 4, 5], [col[0] for col in results])
     assert_equal([1, 2, 3, 4, 5], [col[0] for col in results])
@@ -347,7 +348,8 @@ class TestImpalaIntegration(object):
 
 
     impala_tables, beeswax_tables = get_impala_beeswax_tables()
     impala_tables, beeswax_tables = get_impala_beeswax_tables()
     assert_equal(impala_tables, beeswax_tables,
     assert_equal(impala_tables, beeswax_tables,
-      "\ntest_invalidate_tables: `%s`\nImpala Tables: %s\nBeeswax Tables: %s" % (self.DATABASE, ','.join(impala_tables), ','.join(beeswax_tables)))
+      "\ntest_invalidate_tables: `%s`\nImpala Tables: %s\nBeeswax Tables: %s"
+      % (self.DATABASE, ','.join(impala_tables), ','.join(beeswax_tables)))
 
 
     hql = """
     hql = """
       CREATE TABLE new_table (a INT);
       CREATE TABLE new_table (a INT);
@@ -364,7 +366,8 @@ class TestImpalaIntegration(object):
     impala_tables, beeswax_tables = get_impala_beeswax_tables()
     impala_tables, beeswax_tables = get_impala_beeswax_tables()
     # Invalidate picks up new table
     # Invalidate picks up new table
     assert_equal(impala_tables, beeswax_tables,
     assert_equal(impala_tables, beeswax_tables,
-      "\ntest_invalidate_tables: `%s`\nImpala Tables: %s\nBeeswax Tables: %s" % (self.DATABASE, ','.join(impala_tables), ','.join(beeswax_tables)))
+      "\ntest_invalidate_tables: `%s`\nImpala Tables: %s\nBeeswax Tables: %s"
+      % (self.DATABASE, ','.join(impala_tables), ','.join(beeswax_tables)))
 
 
 
 
   def test_refresh_table(self):
   def test_refresh_table(self):
@@ -378,7 +381,8 @@ class TestImpalaIntegration(object):
 
 
     impala_columns, beeswax_columns = get_impala_beeswax_columns()
     impala_columns, beeswax_columns = get_impala_beeswax_columns()
     assert_equal(impala_columns, beeswax_columns,
     assert_equal(impala_columns, beeswax_columns,
-      "\ntest_refresh_table: `%s`.`%s`\nImpala Columns: %s\nBeeswax Columns: %s" % (self.DATABASE, 'tweets', ','.join(impala_columns), ','.join(beeswax_columns)))
+      "\ntest_refresh_table: `%s`.`%s`\nImpala Columns: %s\nBeeswax Columns: %s"
+      % (self.DATABASE, 'tweets', ','.join(impala_columns), ','.join(beeswax_columns)))
 
 
     hql = """
     hql = """
       ALTER TABLE tweets ADD COLUMNS (new_column INT);
       ALTER TABLE tweets ADD COLUMNS (new_column INT);
@@ -395,7 +399,8 @@ class TestImpalaIntegration(object):
     impala_columns, beeswax_columns = get_impala_beeswax_columns()
     impala_columns, beeswax_columns = get_impala_beeswax_columns()
     # Invalidate picks up new column
     # Invalidate picks up new column
     assert_equal(impala_columns, beeswax_columns,
     assert_equal(impala_columns, beeswax_columns,
-      "\ntest_refresh_table: `%s`.`%s`\nImpala Columns: %s\nBeeswax Columns: %s" % (self.DATABASE, 'tweets', ','.join(impala_columns), ','.join(beeswax_columns)))
+      "\ntest_refresh_table: `%s`.`%s`\nImpala Columns: %s\nBeeswax Columns: %s"
+      % (self.DATABASE, 'tweets', ','.join(impala_columns), ','.join(beeswax_columns)))
 
 
 
 
   def test_get_exec_summary(self):
   def test_get_exec_summary(self):
@@ -442,17 +447,17 @@ class TestImpalaIntegration(object):
 
 
 # Could be refactored with SavedQuery.create_empty()
 # Could be refactored with SavedQuery.create_empty()
 def create_saved_query(app_name, owner):
 def create_saved_query(app_name, owner):
-    query_type = SavedQuery.TYPES_MAPPING[app_name]
-    design = SavedQuery(owner=owner, type=query_type)
-    design.name = 'create_saved_query'
-    design.desc = ''
-    design.data = hql_query('show $tables', database='db1').dumps()
-    design.is_auto = False
-    design.save()
+  query_type = SavedQuery.TYPES_MAPPING[app_name]
+  design = SavedQuery(owner=owner, type=query_type)
+  design.name = 'create_saved_query'
+  design.desc = ''
+  design.data = hql_query('show $tables', database='db1').dumps()
+  design.is_auto = False
+  design.save()
 
 
-    Document.objects.link(design, owner=design.owner, extra=design.type, name=design.name, description=design.desc)
+  Document.objects.link(design, owner=design.owner, extra=design.type, name=design.name, description=design.desc)
 
 
-    return design
+  return design
 
 
 
 
 def test_ssl_cacerts():
 def test_ssl_cacerts():
@@ -521,6 +526,25 @@ def test_thrift_over_http_config():
         reset()
         reset()
 
 
 
 
+def test_thrift_over_http_config_with_proxy_endpoint():
+  resets = [
+      conf.SERVER_HOST.set_for_testing('impala_proxy'),
+      conf.SERVER_PORT.set_for_testing(36000),
+      conf.USE_THRIFT_HTTP.set_for_testing(True),
+      conf.PROXY_ENDPOINT.set_for_testing('/endpoint')
+  ]
+  with patch('impala.dbms.get_hs2_http_port') as get_hs2_http_port:
+    get_hs2_http_port.return_value = 30000
+    try:
+      query_server = get_query_server_config(name='impala')
+      assert_equal(query_server['server_port'], 36000)
+      assert_equal(query_server['transport_mode'], 'http')
+      assert_equal(query_server['http_url'], 'http://impala_proxy:36000/endpoint')
+    finally:
+      for reset in resets:
+        reset()
+
+
 class TestImpalaDbms(object):
 class TestImpalaDbms(object):
 
 
   def test_get_impala_nested_select(self):
   def test_get_impala_nested_select(self):

+ 4 - 0
desktop/conf.dist/hue.ini

@@ -1334,6 +1334,10 @@
   # Port of the Impala Server
   # Port of the Impala Server
   ## server_port=21050
   ## server_port=21050
 
 
+  # Endpoint of the Impala Proxy Server, for example: '/endpoint'.
+  # Note that SERVER_PORT will be used when set.
+  ## proxy_endpoint=
+
   # URL of the Impala Coordinator Server.
   # URL of the Impala Coordinator Server.
   ## coordinator_url=localhost:25000
   ## coordinator_url=localhost:25000
 
 

+ 4 - 0
desktop/conf/pseudo-distributed.ini.tmpl

@@ -1319,6 +1319,10 @@
   # Port of the Impala Server
   # Port of the Impala Server
   ## server_port=21050
   ## server_port=21050
 
 
+  # Endpoint of the Impala Proxy Server, for example: '/endpoint'
+  # Note that SERVER_PORT will be used when set.
+  ## proxy_endpoint=
+
   # URL of the Impala Coordinator Server.
   # URL of the Impala Coordinator Server.
   ## coordinator_url=localhost:25000
   ## coordinator_url=localhost:25000