Pārlūkot izejas kodu

[spark] Add support for paging through batch logfiles

Erick Tryzelaar 10 gadi atpakaļ
vecāks
revīzija
d0a0f49

+ 1 - 1
apps/spark/java/livy-server/src/main/scala/com/cloudera/hue/livy/server/batch/BatchServlet.scala

@@ -141,7 +141,7 @@ private object Serializers {
   def getLogs(batch: Batch, fromOpt: Option[Int], sizeOpt: Option[Int]): JValue = {
     val lines = batch.lines
 
-    val size = sizeOpt.getOrElse(10)
+    val size = sizeOpt.getOrElse(100)
     var from = fromOpt.getOrElse(-1)
     if (from < 0) {
       from = math.min(0, lines.length - size)

+ 7 - 1
apps/spark/src/spark/api.py

@@ -133,8 +133,14 @@ def get_logs(request):
   notebook = json.loads(request.POST.get('notebook', '{}'))
   snippet = json.loads(request.POST.get('snippet', '{}'))
 
+  startFrom = request.POST.get('from')
+  startFrom = int(startFrom) if startFrom else None
+
+  size = request.POST.get('size')
+  size = int(size) if size else None
+
   db = get_api(request.user, snippet)
-  response['logs'] = db.get_log(snippet)
+  response['logs'] = db.get_log(snippet, startFrom=startFrom, size=size)
   response['progress'] = db._progress(snippet, response['logs']) if snippet['status'] != 'available' else 100
   response['job_urls'] = [{
       'name': job,

+ 12 - 3
apps/spark/src/spark/job_server_api.py

@@ -116,9 +116,18 @@ class JobServerApi(object):
     response = self._root.get('batches/%s/state' % uuid)
     return response['state']
 
-  def get_batch_log(self, uuid):
-    response = self._root.get('batches/%s/log' % uuid)
-    return response['log']
+  def get_batch_log(self, uuid, startFrom=None, size=None):
+    params = {}
+
+    if startFrom is not None:
+      params['from'] = startFrom
+
+    if size is not None:
+      params['size'] = size
+
+    response = self._root.get('batches/%s/log' % uuid, params=params)
+
+    return '\n'.join(response['log'])
 
   def delete_batch(self, uuid):
     return self._root.delete('batches/%s' % uuid)

+ 5 - 6
apps/spark/src/spark/models.py

@@ -215,11 +215,11 @@ class HS2Api():
     return {'status': 'canceled'}
 
   @query_error_handler
-  def get_log(self, snippet):
+  def get_log(self, snippet, startFrom=None, size=None):
     db = self._get_db(snippet)
 
     handle = self._get_handle(snippet)
-    return db.get_log(handle)
+    return db.get_log(handle, start_over=startFrom == 0)
 
   def download(self, notebook, snippet, format):
     try:
@@ -398,7 +398,7 @@ class SparkApi():
 
     return {'status': 'canceled'}
 
-  def get_log(self, snippet):
+  def get_log(self, snippet, startFrom=0, size=None):
     return 'Not available'
 
   def _progress(self, snippet, logs):
@@ -452,11 +452,10 @@ class SparkBatchApi():
         'status': state,
     }
 
-  def get_log(self, snippet):
+  def get_log(self, snippet, startFrom=0, size=None):
     api = get_spark_api(self.user)
 
-    log = api.get_batch_log(snippet['result']['handle']['id'])
-    return '\n'.join(log)
+    return api.get_batch_log(snippet['result']['handle']['id'], startFrom=startFrom, size=size)
 
   def _progress(self, snippet, logs):
     return 50

+ 13 - 2
apps/spark/src/spark/static/spark/js/spark.ko.js

@@ -68,6 +68,7 @@ var Result = function (snippet, result) {
   self.data = ko.observableArray(typeof result.data != "undefined" && result.data != null ? result.data : []);
   self.data.extend({ rateLimit: 50 });
   self.logs = ko.observable('');
+  self.logLines = 0;
   self.errors = ko.observable('');
   self.hasSomeResults = ko.computed(function () {
     return self.hasResultset() && self.data().length > 0; // status() == 'available'
@@ -487,10 +488,20 @@ var Snippet = function (vm, notebook, snippet) {
   self.getLogs = function () {
     $.post("/spark/api/get_logs", {
       notebook: ko.mapping.toJSON(notebook.getContext()),
-      snippet: ko.mapping.toJSON(self.getContext())
+      snippet: ko.mapping.toJSON(self.getContext()),
+      from: self.result.logLines
     }, function (data) {
       if (data.status == 0) {
-        self.result.logs(data.logs); // Way to append?
+        if (data.logs.length > 0) {
+          var logs = data.logs.split("\n");
+          self.result.logLines += logs.length;
+          var oldLogs = self.result.logs();
+          if (oldLogs === "") {
+            self.result.logs(data.logs);
+          } else {
+            self.result.logs(oldLogs + "\n" + data.logs);
+          }
+        }
         self.progress(data.progress);
       } else {
         self._ajax_error(data);