浏览代码

HUE-2387 [beeswax] Support getting logs using FetchResults() api

Peter Slawski 11 年之前
父节点
当前提交
6a02467

+ 13 - 1
apps/beeswax/gen-py/TCLIService/ttypes.py

@@ -5124,6 +5124,7 @@ class TFetchResultsReq(object):
    - operationHandle
    - orientation
    - maxRows
+   - fetchType
   """
 
   thrift_spec = (
@@ -5131,12 +5132,14 @@ class TFetchResultsReq(object):
     (1, TType.STRUCT, 'operationHandle', (TOperationHandle, TOperationHandle.thrift_spec), None, ), # 1
     (2, TType.I32, 'orientation', None,     0, ), # 2
     (3, TType.I64, 'maxRows', None, None, ), # 3
+    (4, TType.I16, 'fetchType', None, 0, ), # 4
   )
 
-  def __init__(self, operationHandle=None, orientation=thrift_spec[2][4], maxRows=None,):
+  def __init__(self, operationHandle=None, orientation=thrift_spec[2][4], maxRows=None, fetchType=thrift_spec[4][4],):
     self.operationHandle = operationHandle
     self.orientation = orientation
     self.maxRows = maxRows
+    self.fetchType = fetchType
 
   def read(self, iprot):
     if iprot.__class__ == TBinaryProtocol.TBinaryProtocolAccelerated and isinstance(iprot.trans, TTransport.CReadableTransport) and self.thrift_spec is not None and fastbinary is not None:
@@ -5163,6 +5166,11 @@ class TFetchResultsReq(object):
           self.maxRows = iprot.readI64();
         else:
           iprot.skip(ftype)
+      elif fid == 4:
+        if ftype == TType.I16:
+          self.fetchType = iprot.readI16();
+        else:
+          iprot.skip(ftype)
       else:
         iprot.skip(ftype)
       iprot.readFieldEnd()
@@ -5185,6 +5193,10 @@ class TFetchResultsReq(object):
       oprot.writeFieldBegin('maxRows', TType.I64, 3)
       oprot.writeI64(self.maxRows)
       oprot.writeFieldEnd()
+    if self.fetchType is not None:
+      oprot.writeFieldBegin('fetchType', TType.I16, 4)
+      oprot.writeI16(self.fetchType)
+      oprot.writeFieldEnd()
     oprot.writeFieldStop()
     oprot.writeStructEnd()
 

+ 2 - 1
apps/beeswax/src/beeswax/api.py

@@ -172,7 +172,8 @@ def watch_query_refresh_json(request, id):
     handle, state = _get_query_handle_and_state(query_history)
 
   try:
-    log = db.get_log(handle)
+    start_over = request.POST.get('log-start-over') == 'true'
+    log = db.get_log(handle, start_over=start_over)
   except Exception, ex:
     log = str(ex)
 

+ 8 - 0
apps/beeswax/src/beeswax/conf.py

@@ -60,6 +60,14 @@ SERVER_CONN_TIMEOUT = Config(
   type=int,
   help=_t('Timeout in seconds for Thrift calls.'))
 
+USE_GET_LOG_API = Config(
+  key='use_get_log_api',
+  default=True,
+  type=coerce_bool,
+  help=_t('Choose whether Hue uses the GetLog() thrift call to retrieve Hive logs.'
+          'If false, Hue will use the FetchResults() thrift call instead.')
+)
+
 BROWSE_PARTITIONED_TABLE_LIMIT = Config(
   key='browse_partitioned_table_limit',
   default=250,

+ 2 - 2
apps/beeswax/src/beeswax/server/dbms.py

@@ -338,8 +338,8 @@ class HiveServer2Dbms(object):
     return self.client.use(query)
 
 
-  def get_log(self, query_handle):
-    return self.client.get_log(query_handle)
+  def get_log(self, query_handle, start_over=True):
+    return self.client.get_log(query_handle, start_over)
 
 
   def get_state(self, handle):

+ 21 - 2
apps/beeswax/src/beeswax/server/hive_server2_lib.py

@@ -14,6 +14,7 @@
 # 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.
+from itertools import imap
 
 import logging
 import re
@@ -590,6 +591,15 @@ class HiveServerClient:
     return res, schema
 
 
+  def fetch_log(self, operation_handle, orientation=TFetchOrientation.FETCH_NEXT, max_rows=1000):
+    req = TFetchResultsReq(operationHandle=operation_handle, orientation=orientation,
+                           maxRows=max_rows, fetchType=1)
+    res = self.call(self._client.FetchResults, req)
+
+    lines = imap(lambda r: r.colVals[0].stringVal.value, res.results.rows)
+    return '\n'.join(lines)
+
+
   def get_operation_status(self, operation_handle):
     req = TGetOperationStatusReq(operationHandle=operation_handle)
     return self.call(self._client.GetOperationStatus, req)
@@ -759,9 +769,18 @@ class HiveServerClientCompatible(object):
     return 'Does not exist in HS2'
 
 
-  def get_log(self, handle):
+  def get_log(self, handle, start_over=True):
     operationHandle = handle.get_rpc_handle()
-    return self._client.get_log(operationHandle)
+
+    if beeswax_conf.USE_GET_LOG_API.get() or self.query_server['server_name'] == 'impala':
+      return self._client.get_log(operationHandle)
+    else:
+      if start_over:
+        orientation = TFetchOrientation.FETCH_FIRST
+      else:
+        orientation = TFetchOrientation.FETCH_NEXT
+
+      return self._client.fetch_log(operationHandle, orientation=orientation, max_rows=-1)
 
 
   def get_databases(self):

+ 5 - 0
apps/beeswax/src/beeswax/templates/execute.mako

@@ -2637,6 +2637,11 @@ function setupCodeMirrorSubscription() {
 
 // Knockout
 viewModel = new BeeswaxViewModel("${app_name}");
+
+% if not beeswax_conf.USE_GET_LOG_API.get() and app_name != 'impala':
+  viewModel.shouldAppendLogs = true;
+% endif
+
 % if query_history:
   loadQueryHistory(${query_history.id});
 % elif design.id:

+ 15 - 3
apps/beeswax/static/js/beeswax.vm.js

@@ -560,7 +560,8 @@ function BeeswaxViewModel(server) {
   self.watchQuery = function() {
     var data = {
       'query-query': self.design.query.value(),
-      'query-database': self.database()
+      'query-database': self.database(),
+      'log-start-over': self.design.watch.logs().length == 0
     };
     $.extend(data, self.getSettingsFormData());
     $.extend(data, self.getFileResourcesFormData());
@@ -589,6 +590,17 @@ function BeeswaxViewModel(server) {
     $.ajax(request);
   };
 
+  self.shouldAppendLogs = false;
+
+  self.applyLogs = function(log) {
+    var lines = log.split("\n")
+
+    if (self.shouldAppendLogs) {
+      lines = self.design.watch.logs().concat(lines);
+    }
+    self.design.watch.logs(lines);
+  };
+
   self.watchQueryLoop = function(fn) {
     var TIMEOUT = 100;
     var timer = null;
@@ -607,7 +619,7 @@ function BeeswaxViewModel(server) {
           clearTimeout(timer);
           self.design.isRunning(false);
           if (data.log) {
-            self.design.watch.logs(data.log.split("\n"));
+            self.applyLogs(data.log)
             // scroll logs
             self.design.watch.jobUrls(data.jobUrls);
           }
@@ -623,7 +635,7 @@ function BeeswaxViewModel(server) {
         } else {
           self.design.statement(data.statement); // In case new no result statement executed
           if (data.log) {
-            self.design.watch.logs(data.log.split("\n"));
+            self.applyLogs(data.log)
             // scroll logs
             self.design.watch.jobUrls(data.jobUrls);
           }

+ 3 - 0
apps/beeswax/thrift/TCLIService.thrift

@@ -1002,6 +1002,9 @@ struct TFetchResultsReq {
   // Max number of rows that should be returned in
   // the rowset.
   3: required i64 maxRows
+
+  // The type of a fetch results request. 0 represents Query output. 1 represents Log
+  4: optional i16 fetchType = 0
 }
 
 struct TFetchResultsResp {

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

@@ -724,6 +724,10 @@
   # Timeout in seconds for thrift calls to Hive service
   ## server_conn_timeout=120
 
+  # Choose whether Hue uses the GetLog() thrift call to retrieve Hive logs.
+  # If false, Hue will use the FetchResults() thrift call instead.
+  ## use_get_log_api=true
+
   # Set a LIMIT clause when browsing a partitioned table.
   # A positive value will be set as the LIMIT. If 0 or negative, do not set any limit.
   ## browse_partitioned_table_limit=250

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

@@ -731,6 +731,10 @@
   # Timeout in seconds for thrift calls to Hive service
   ## server_conn_timeout=120
 
+  # Choose whether Hue uses the GetLog() thrift call to retrieve Hive logs.
+  # If false, Hue will use the FetchResults() thrift call instead.
+  ## use_get_log_api=true
+
   # Set a LIMIT clause when browsing a partitioned table.
   # A positive value will be set as the LIMIT. If 0 or negative, do not set any limit.
   ## browse_partitioned_table_limit=250