Browse Source

[spark] Restart a Spark session when it has expired

Romain Rigaux 11 years ago
parent
commit
56896085b2
2 changed files with 52 additions and 29 deletions
  1. 18 11
      apps/spark/src/spark/api.py
  2. 34 18
      apps/spark/static/js/spark.vm.js

+ 18 - 11
apps/spark/src/spark/api.py

@@ -32,7 +32,7 @@ from spark.models import get_api, Notebook
 LOG = logging.getLogger(__name__)
 LOG = logging.getLogger(__name__)
 
 
 
 
-@json_error_handler
+
 def create_session(request):
 def create_session(request):
   response = {'status': -1}
   response = {'status': -1}
 
 
@@ -43,12 +43,11 @@ def create_session(request):
     response['session'] = get_api(request.user, snippet).create_session(lang=snippet['type'])
     response['session'] = get_api(request.user, snippet).create_session(lang=snippet['type'])
     response['status'] = 0
     response['status'] = 0
   except Exception, e:
   except Exception, e:
-    raise PopupException(e, title=_('Error while accessing query server'))
     response['error'] = force_unicode(str(e))
     response['error'] = force_unicode(str(e))
 
 
   return HttpResponse(json.dumps(response), mimetype="application/json")
   return HttpResponse(json.dumps(response), mimetype="application/json")
 
 
-@json_error_handler
+
 def execute(request):
 def execute(request):
   response = {'status': -1}
   response = {'status': -1}
 
 
@@ -59,13 +58,15 @@ def execute(request):
     response['handle'] = get_api(request.user, snippet).execute(notebook, snippet)
     response['handle'] = get_api(request.user, snippet).execute(notebook, snippet)
     response['status'] = 0
     response['status'] = 0
   except Exception, e:
   except Exception, e:
-    raise PopupException(e, title=_('Error while accessing query server'))
-    response['error'] = force_unicode(str(e))
+    message = force_unicode(str(e))
+    if 'session not found' in message:
+      response['status'] = -2
+    else:
+      response['error'] = force_unicode(str(e))
 
 
   return HttpResponse(json.dumps(response), mimetype="application/json")
   return HttpResponse(json.dumps(response), mimetype="application/json")
 
 
 
 
-@json_error_handler
 def check_status(request):
 def check_status(request):
   response = {'status': -1}
   response = {'status': -1}
 
 
@@ -76,13 +77,15 @@ def check_status(request):
     response['query_status'] = get_api(request.user, snippet).check_status(notebook, snippet)
     response['query_status'] = get_api(request.user, snippet).check_status(notebook, snippet)
     response['status'] = 0
     response['status'] = 0
   except Exception, e:
   except Exception, e:
-    raise PopupException(e, title=_('Error while accessing query server'))
-    response['error'] = force_unicode(str(e))
+    message = force_unicode(str(e))
+    if 'session not found' in message:
+      response['status'] = -2
+    else:
+      response['error'] = force_unicode(str(e))
 
 
   return HttpResponse(json.dumps(response), mimetype="application/json")
   return HttpResponse(json.dumps(response), mimetype="application/json")
 
 
 
 
-@json_error_handler
 def fetch_result(request):
 def fetch_result(request):
   response = {'status': -1}
   response = {'status': -1}
 
 
@@ -93,8 +96,12 @@ def fetch_result(request):
     response['result'] = get_api(request.user, snippet).fetch_result(notebook, snippet)
     response['result'] = get_api(request.user, snippet).fetch_result(notebook, snippet)
     response['status'] = 0
     response['status'] = 0
   except Exception, e:
   except Exception, e:
-    raise PopupException(e, title=_('Error while accessing query server'))
-    response['error'] = force_unicode(str(e))
+    message = force_unicode(str(e))
+    if 'session not found' in message:
+      response['status'] = -2
+    else:
+      response['error'] = force_unicode(str(e))
+
 
 
   return HttpResponse(json.dumps(response), mimetype="application/json")
   return HttpResponse(json.dumps(response), mimetype="application/json")
 
 

+ 34 - 18
apps/spark/static/js/spark.vm.js

@@ -24,7 +24,7 @@ var Result = function (snippet, result) {
   self.meta = ko.mapping.fromJS(typeof result.meta != "undefined" && result.meta != null ? result.meta : []);
   self.meta = ko.mapping.fromJS(typeof result.meta != "undefined" && result.meta != null ? result.meta : []);
   self.data = ko.mapping.fromJS(typeof result.data != "undefined" && result.data != null ? result.data : []);
   self.data = ko.mapping.fromJS(typeof result.data != "undefined" && result.data != null ? result.data : []);
   
   
-  if (typeof result.handle!= "undefined" && result.handle != null) {
+  if (typeof result.handle != "undefined" && result.handle != null) {
     $.each(result.handle, function(key, val) {
     $.each(result.handle, function(key, val) {
       self.handle()[key] = val;
       self.handle()[key] = val;
     });
     });
@@ -54,8 +54,11 @@ var Snippet = function (notebook, snippet) {
   self.type = ko.observable(typeof snippet.type != "undefined" && snippet.type != null ? snippet.type : "hive");
   self.type = ko.observable(typeof snippet.type != "undefined" && snippet.type != null ? snippet.type : "hive");
   self.editorMode = ko.observable(TYPE_EDITOR_MAP[self.type()]);
   self.editorMode = ko.observable(TYPE_EDITOR_MAP[self.type()]);
   self.statement = ko.observable(typeof snippet.statement != "undefined" && snippet.statement != null ? snippet.statement : '');
   self.statement = ko.observable(typeof snippet.statement != "undefined" && snippet.statement != null ? snippet.statement : '');
-  self.statement = ko.observable("");
-  self.status = ko.observable("loading");
+  self.status = ko.observable(typeof snippet.status != "undefined" && snippet.status != null ? snippet.status : 'loading');
+  self.variables = ko.computed(function() {
+	return self.statement().match(/\$[^\d'"](\w*)/g);  
+  });
+  self.result = new Result(snippet, snippet.result);
 
 
   self.size = ko.observable(typeof snippet.size != "undefined" && snippet.size != null ? snippet.size : 12).extend({ numeric: 0 });
   self.size = ko.observable(typeof snippet.size != "undefined" && snippet.size != null ? snippet.size : 12).extend({ numeric: 0 });
   self.offset = ko.observable(typeof snippet.offset != "undefined" && snippet.offset != null ? snippet.offset : 0).extend({ numeric: 0 });
   self.offset = ko.observable(typeof snippet.offset != "undefined" && snippet.offset != null ? snippet.offset : 0).extend({ numeric: 0 });
@@ -95,21 +98,17 @@ var Snippet = function (notebook, snippet) {
   self.remove = function (notebook, snippet) {
   self.remove = function (notebook, snippet) {
     notebook.snippets.remove(snippet);
     notebook.snippets.remove(snippet);
   }
   }
-
-
-  self.result = new Result(snippet, snippet.result);
   
   
   // init()
   // init()
   // checkStatus()
   // checkStatus()
 
 
   self.create_session = function() {
   self.create_session = function() {
-
     $.post("/spark/api/create_session", {
     $.post("/spark/api/create_session", {
     	notebook: ko.mapping.toJSON(notebook),
     	notebook: ko.mapping.toJSON(notebook),
         snippet: ko.mapping.toJSON(self)
         snippet: ko.mapping.toJSON(self)
 	  }, function (data) {
 	  }, function (data) {
 	    if (data.status == 0) {
 	    if (data.status == 0) {
-		  notebook.sessions.push(ko.mapping.fromJS(data.session));
+		  notebook.addSession(ko.mapping.fromJS(data.session));
 	      self.status('ready');
 	      self.status('ready');
 	    }
 	    }
 	    else {
 	    else {
@@ -117,9 +116,7 @@ var Snippet = function (notebook, snippet) {
 	    }
 	    }
 	}).fail(function (xhr, textStatus, errorThrown) {
 	}).fail(function (xhr, textStatus, errorThrown) {
       $(document).trigger("error", xhr.responseText);
       $(document).trigger("error", xhr.responseText);
-    });
-    
-    
+    }); 
   };
   };
   
   
   self.execute = function() {
   self.execute = function() {
@@ -140,7 +137,9 @@ var Snippet = function (notebook, snippet) {
 
 
           self.checkStatus();
           self.checkStatus();
         }
         }
-        else {
+        else if (data.status == -2) {
+          self.create_session();          
+        } else {
           $(document).trigger("error", data.message);
           $(document).trigger("error", data.message);
         }
         }
     }).fail(function (xhr, textStatus, errorThrown) {
     }).fail(function (xhr, textStatus, errorThrown) {
@@ -161,8 +160,9 @@ var Snippet = function (notebook, snippet) {
           } else {
           } else {
         	self.fetchResult();
         	self.fetchResult();
           }
           }
-	    }
-	    else {
+	    } else if (data.status == -2) {
+	      self.create_session();  
+	    } else {
 	      $(document).trigger("error", data.message);
 	      $(document).trigger("error", data.message);
 	    }
 	    }
 	}).fail(function (xhr, textStatus, errorThrown) {
 	}).fail(function (xhr, textStatus, errorThrown) {
@@ -181,8 +181,9 @@ var Snippet = function (notebook, snippet) {
 
 
           // move resultsets to n rows
           // move resultsets to n rows
           // check if N rows fetched...
           // check if N rows fetched...
- 	    }
- 	    else {
+ 	    } else if (data.status == -2) {
+ 	      self.create_session();  
+ 	    } else {
  	      $(document).trigger("error", data.message);
  	      $(document).trigger("error", data.message);
  	    }
  	    }
  	}).fail(function (xhr, textStatus, errorThrown) {
  	}).fail(function (xhr, textStatus, errorThrown) {
@@ -209,7 +210,7 @@ var Notebook = function (vm, notebook) {
   self.name = ko.observable(typeof notebook.name != "undefined" && notebook.name != null ? notebook.name : 'My Notebook');
   self.name = ko.observable(typeof notebook.name != "undefined" && notebook.name != null ? notebook.name : 'My Notebook');
   self.snippets = ko.observableArray();
   self.snippets = ko.observableArray();
   self.selectedSnippet = ko.observable('scala');
   self.selectedSnippet = ko.observable('scala');
-  self.availableSnippets = ko.observableArray(['hive', 'scala', 'sql', 'python', 'pig', 'impala']); // presto, mysql, oracle, sqlite, postgres, phoenix
+  self.availableSnippets = ko.observableArray(['hive', 'scala', 'sql', 'python', 'text', 'pig', 'impala']); // presto, mysql, oracle, sqlite, postgres, phoenix
   self.sessions = ko.mapping.fromJS(typeof notebook.sessions != "undefined" && notebook.sessions != null ? notebook.sessions : []); 
   self.sessions = ko.mapping.fromJS(typeof notebook.sessions != "undefined" && notebook.sessions != null ? notebook.sessions : []); 
 
 
   self.getSession = function(session_type) {
   self.getSession = function(session_type) {
@@ -223,6 +224,21 @@ var Notebook = function (vm, notebook) {
     return _s;
     return _s;
   };
   };
   
   
+  self.addSession = function(session) {
+	var toRemove = []
+    $.each(self.sessions(), function (index, s) {
+      if (s.type() == session.type()) {
+    	toRemove.push(s);
+      }
+    });
+	
+	$.each(toRemove, function (index, s) {
+	  self.sessions.remove(s);
+	});
+	
+    self.sessions.push(session);
+  };  
+  
   self.addSnippet = function(snippet) {
   self.addSnippet = function(snippet) {
 	var _snippet = new Snippet(self, snippet);
 	var _snippet = new Snippet(self, snippet);
 	self.snippets.push(_snippet);
 	self.snippets.push(_snippet);
@@ -312,4 +328,4 @@ function logGA(page) {
   if (typeof trackOnGA == 'function') {
   if (typeof trackOnGA == 'function') {
     trackOnGA('editor/' + page);
     trackOnGA('editor/' + page);
   }
   }
-}
+}