瀏覽代碼

HUE-8617 [impala] Support multi cluster in invalidate metadata

Romain Rigaux 7 年之前
父節點
當前提交
26a2d82

+ 14 - 8
apps/beeswax/src/beeswax/server/dbms.py

@@ -79,14 +79,7 @@ def get(user, query_server=None, cluster=None):
 def get_query_server_config(name='beeswax', server=None, cluster=None):
   LOG.debug("Query cluster %s: %s" % (name, cluster))
 
-  if cluster and cluster.get('id') != CLUSTER_ID.get():
-    if 'altus:dataware:k8s' in cluster['id']:
-      compute_end_point = cluster['compute_end_point'][0] if type(cluster['compute_end_point']) == list else cluster['compute_end_point'] # TODO getting list from left assist
-      cluster_config = {'server_host': compute_end_point, 'name': cluster['name']} # TODO get port too
-    else:
-      cluster_config = Cluster(user=None).get_config(cluster['id']) # Direct cluster
-  else:
-    cluster_config = None
+  cluster_config = get_cluster_config(cluster)
 
   if name == 'impala':
     from impala.dbms import get_query_server_config as impala_query_server_config
@@ -126,6 +119,19 @@ def get_query_server_config(name='beeswax', server=None, cluster=None):
   return query_server
 
 
+def get_cluster_config(cluster=None):
+  if cluster and cluster.get('id') != CLUSTER_ID.get():
+    if 'altus:dataware:k8s' in cluster['id']:
+      compute_end_point = cluster['compute_end_point'][0] if type(cluster['compute_end_point']) == list else cluster['compute_end_point'] # TODO getting list from left assist
+      cluster_config = {'server_host': compute_end_point, 'name': cluster['name']} # TODO get port too
+    else:
+      cluster_config = Cluster(user=None).get_config(cluster['id']) # Direct cluster
+  else:
+    cluster_config = None
+
+  return cluster_config
+
+
 class QueryServerException(Exception):
   # Ideally the query handle will be stored here too.
 

+ 8 - 6
apps/beeswax/src/beeswax/server/hive_server2_lib.py

@@ -919,12 +919,14 @@ class HiveServerClient:
     req = TExecuteStatementReq(statement=statement.encode('utf-8'), confOverlay=confOverlay, runAsync=True)
     (res, session) = self.call_return_result_and_session(self._client.ExecuteStatement, req, with_multiple_session=with_multiple_session)
 
-    return HiveServerQueryHandle(secret=res.operationHandle.operationId.secret,
-                                 guid=res.operationHandle.operationId.guid,
-                                 operation_type=res.operationHandle.operationType,
-                                 has_result_set=res.operationHandle.hasResultSet,
-                                 modified_row_count=res.operationHandle.modifiedRowCount,
-                                 session_guid=session.guid)
+    return HiveServerQueryHandle(
+        secret=res.operationHandle.operationId.secret,
+        guid=res.operationHandle.operationId.guid,
+        operation_type=res.operationHandle.operationType,
+        has_result_set=res.operationHandle.hasResultSet,
+        modified_row_count=res.operationHandle.modifiedRowCount,
+        session_guid=session.guid
+    )
 
 
   def fetch_data(self, operation_handle, orientation=TFetchOrientation.FETCH_NEXT, max_rows=1000):

+ 8 - 7
apps/impala/src/impala/api.py

@@ -26,6 +26,7 @@ from django.views.decorators.http import require_POST
 from desktop.lib.django_util import JsonResponse
 
 from beeswax.api import error_handler
+from beeswax.server.dbms import get_cluster_config
 from beeswax.models import Session
 from beeswax.server import dbms as beeswax_dbms
 from beeswax.views import authorized_get_query_history
@@ -39,17 +40,17 @@ LOG = logging.getLogger(__name__)
 @error_handler
 def invalidate(request):
   cluster = json.loads(request.POST.get('cluster', '{}'))
+  database = request.POST.get('database', None)
+  table = request.POST.get('table', None)
+  flush_all = request.POST.get('flush_all', 'false').lower() == 'true'
 
-  # TODO: create cluster_config for get_query_server_config
-  query_server = dbms.get_query_server_config()
-  db = beeswax_dbms.get(request.user, query_server=query_server, cluster=cluster)
+  cluster_config = get_cluster_config(cluster)
+  query_server = dbms.get_query_server_config(cluster_config=cluster_config)
+  db = beeswax_dbms.get(request.user, query_server=query_server)
 
   response = {'status': 0, 'message': ''}
 
-  database = request.POST.get('database', None)
-  flush_all = request.POST.get('flush_all', 'false').lower() == 'true'
-
-  db.invalidate(database=database, flush_all=flush_all)
+  db.invalidate(database=database, table=table, flush_all=flush_all)
   response['message'] = _('Successfully invalidated metadata')
 
   return JsonResponse(response)

+ 17 - 12
apps/impala/src/impala/dbms.py

@@ -33,16 +33,16 @@ LOG = logging.getLogger(__name__)
 
 def get_query_server_config(cluster_config=None):
   query_server = {
-        'server_name': 'impala' + ('-' + cluster_config.get('name') if cluster_config else ''),
-        'server_host': conf.SERVER_HOST.get() if not cluster_config else cluster_config.get('server_host'),
-        'server_port': conf.SERVER_PORT.get() if not cluster_config else 21050,
-        'principal': conf.IMPALA_PRINCIPAL.get(),
-        'impersonation_enabled': conf.IMPERSONATION_ENABLED.get(),
-        'querycache_rows': conf.QUERYCACHE_ROWS.get(),
-        'QUERY_TIMEOUT_S': conf.QUERY_TIMEOUT_S.get(),
-        'SESSION_TIMEOUT_S': conf.SESSION_TIMEOUT_S.get(),
-        'auth_username': conf.AUTH_USERNAME.get(),
-        'auth_password': conf.AUTH_PASSWORD.get()
+      'server_name': 'impala' + ('-' + cluster_config.get('name') if cluster_config else ''),
+      'server_host': conf.SERVER_HOST.get() if not cluster_config else cluster_config.get('server_host'),
+      'server_port': conf.SERVER_PORT.get() if not cluster_config else 21050,
+      'principal': conf.IMPALA_PRINCIPAL.get(),
+      'impersonation_enabled': conf.IMPERSONATION_ENABLED.get(),
+      'querycache_rows': conf.QUERYCACHE_ROWS.get(),
+      'QUERY_TIMEOUT_S': conf.QUERY_TIMEOUT_S.get(),
+      'SESSION_TIMEOUT_S': conf.SESSION_TIMEOUT_S.get(),
+      'auth_username': conf.AUTH_USERNAME.get(),
+      'auth_password': conf.AUTH_PASSWORD.get()
   }
 
   debug_query_server = query_server.copy()
@@ -87,19 +87,24 @@ class ImpalaDbms(HiveServer2Dbms):
     return 'SELECT histogram(%s) FROM %s' % (select_clause, from_clause)
 
 
-  def invalidate(self, database=None, flush_all=False):
+  def invalidate(self, database=None, table=None, flush_all=False):
     handle = None
+
     try:
       if flush_all or database is None:
         hql = "INVALIDATE METADATA"
         query = hql_query(hql, query_type=QUERY_TYPES[1])
         handle = self.execute_and_wait(query, timeout_sec=10.0)
-      else:
+      elif table is None:
         diff_tables = self._get_different_tables(database)
         for table in diff_tables:
           hql = "INVALIDATE METADATA `%s`.`%s`" % (database, table)
           query = hql_query(hql, query_type=QUERY_TYPES[1])
           handle = self.execute_and_wait(query, timeout_sec=10.0)
+      else:
+        hql = "INVALIDATE METADATA `%s`.`%s`" % (database, table)
+        query = hql_query(hql, query_type=QUERY_TYPES[1])
+        handle = self.execute_and_wait(query, timeout_sec=10.0)
     except QueryServerTimeoutException, e:
       # Allow timeout exceptions to propagate
       raise e

+ 3 - 3
apps/jobbrowser/src/jobbrowser/apis/clusters.py

@@ -51,9 +51,9 @@ class ClusterApi(Api):
       u'status': 0,
       u'total': 3,
       u'apps': [
-        {u'status': u'ONLINE', u'name': u'Internal EDH', u'submitted': u'2018-10-04 08:34:39.128886', u'queue': u'group', u'user': u'jo0', u'canWrite': True, u'duration': 0, u'progress': u'1 / 1', u'type': u'GKE 100 nodes 100CPU 20TB', u'id': u'crn:altus:engine:k8s:12a0079b-1591-4ca0-b721-a446bda74e67:cluster:jo0/cbf7bbb1-f956-45e4-a269-d239efbc9996', u'apiStatus': u'RUNNING'},
-        {u'status': u'ONLINE', u'name': u'gke_gcp-eng-dsdw_us-west2-b_impala-demo', u'submitted': u'2018-10-04 08:34:39.128881', u'queue': u'group', u'user': u'r0', u'canWrite': True, u'duration': 0, u'progress': u'1 / 1', u'type': u'GKE 4 nodes 16CPU 64GB', u'id': u'crn:altus:engine:k8s:12a0079b-1591-4ca0-b721-a446bda74e67:cluster:r0/0da5e627-ee33-45c5-9179-cc6b95008d2e', u'apiStatus': u'RUNNING'},
-        {u'status': u'ONLINE', u'name': u'DW-fraud', u'submitted': u'2018-10-04 08:34:39.128881', u'queue': u'group', u'user': u'r0', u'canWrite': True, u'duration': 0, u'progress': u'1 / 1', u'type': u'OpenShift 50 nodes 30CPU 2TB', u'id': u'crn:altus:engine:k8s:12a0079b-1591-4ca0-b721-a446bda74e67:cluster:r0/0da5e627-ee33-45c5-9179-cc6b95008d2e', u'apiStatus': u'RUNNING'},
+        {u'status': u'ONLINE', u'name': u'Internal EDH', u'submitted': u'2018-10-04 08:34:39.128886', u'queue': u'group', u'user': u'jo0', u'canWrite': True, u'duration': 0, u'progress': u'100 / 100', u'type': u'GKE 100 nodes 100CPU 20TB', u'id': u'crn:altus:engine:k8s:12a0079b-1591-4ca0-b721-a446bda74e67:cluster:jo0/cbf7bbb1-f956-45e4-a269-d239efbc9996', u'apiStatus': u'RUNNING'},
+        {u'status': u'ONLINE', u'name': u'gke_gcp-eng-dsdw_us-west2-b_impala-demo', u'submitted': u'2018-10-04 08:34:39.128881', u'queue': u'group', u'user': u'r0', u'canWrite': True, u'duration': 0, u'progress': u'4 / 4', u'type': u'GKE 4 nodes 16CPU 64GB', u'id': u'crn:altus:engine:k8s:12a0079b-1591-4ca0-b721-a446bda74e67:cluster:r0/0da5e627-ee33-45c5-9179-cc6b95008d2e', u'apiStatus': u'RUNNING'},
+        {u'status': u'ONLINE', u'name': u'DW-fraud', u'submitted': u'2018-10-04 08:34:39.128881', u'queue': u'group', u'user': u'r0', u'canWrite': True, u'duration': 0, u'progress': u'50 / 50', u'type': u'OpenShift 50 nodes 30CPU 2TB', u'id': u'crn:altus:engine:k8s:12a0079b-1591-4ca0-b721-a446bda74e67:cluster:r0/0da5e627-ee33-45c5-9179-cc6b95008d2e', u'apiStatus': u'RUNNING'},
       ]
     }
 

+ 6 - 6
apps/jobbrowser/src/jobbrowser/templates/job_browser.mako

@@ -272,6 +272,12 @@ ${ commonheader("Job Browser", "jobbrowser", user, request) | n,unicode }
 
                   <span data-bind="visible: jobs.createClusterShow">
                     <input type="text" data-bind="clearable: jobs.createClusterName, valueUpdate: 'afterkeydown'" class="input-small" placeholder="${_('Name')}">
+                      Environment 
+                      <select>
+                        <option>GCE-west</option>
+                        <option>AWS-finance-secure</option>
+                        <option>OpenShift-prem</option>
+                      </select>
                     <input type="number" data-bind="value: jobs.createClusterWorkers, valueUpdate: 'afterkeydown'" class="input-mini" placeholder="${_('Size')}">
                     <!-- ko if: $root.cluster() && $root.cluster()['type'] != 'altus-engines' -->
                       ${ _('workers') }
@@ -284,12 +290,6 @@ ${ commonheader("Job Browser", "jobbrowser", user, request) | n,unicode }
                     <!-- /ko -->
                     <!-- ko if: $root.cluster() && $root.cluster()['type'] == 'altus-engines' -->
                       ${ _('nodes') }
-                      Environment
-                      <select>
-                        <option>AWS-finance-secure</option>
-                        <option>GCE-east</option>
-                        <option>OpenShift-prem</option>
-                      </select>
                     <!-- /ko -->
 
                     <button class="btn" data-bind="click: jobs.createCluster, enable: jobs.createClusterName().length > 0 && jobs.createClusterWorkers() > 0" title="${ _('Start creation') }">