[ZEPPELIN-1210] Run interpreter per user

zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/InterpreterGroup.java

@@ -33,7 +33,7 @@
  * and InterpreterGroup will have reference to these all interpreters.
  *
  * Remember, list of interpreters are dedicated to a note.
- * (when InterpreterOption.perNoteSession==true)
+ * (when InterpreterOption.session==true)
  * So InterpreterGroup internally manages map of [noteId, list of interpreters]
  *
  * A InterpreterGroup runs on interpreter process.
@@ -203,6 +203,14 @@ public void destroy(String noteId) {
     LOGGER.info("Destroy interpreter group " + getId() + " for note " + noteId);
     List<Interpreter> intpForNote = this.get(noteId);
     destroy(intpForNote);
+
+    if (remoteInterpreterProcess != null) {
+      remoteInterpreterProcess.dereference();
+      if (remoteInterpreterProcess.referenceCount() <= 0) {
+        remoteInterpreterProcess = null;
+        allInterpreterGroups.remove(id);
+      }
+    }
   }
 
 
@@ -222,6 +230,7 @@ public void destroy() {
       while (remoteInterpreterProcess.referenceCount() > 0) {
         remoteInterpreterProcess.dereference();
       }
+      remoteInterpreterProcess = null;
     }
 
     allInterpreterGroups.remove(id);

zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreter.java

@@ -298,6 +298,7 @@ public InterpreterResult interpret(String st, InterpreterContext context) {
     if (logger.isDebugEnabled()) {
       logger.debug("st:\n{}", st);
     }
+
     FormType form = getFormType();
     RemoteInterpreterProcess interpreterProcess = getInterpreterProcess();
     Client client = null;

zeppelin-server/src/main/java/org/apache/zeppelin/rest/NotebookRestApi.java

@@ -177,7 +177,7 @@ public Response putNotePermissions(@PathParam("noteId") String noteId, String re
   public Response bind(@PathParam("noteId") String noteId, String req) throws IOException {
     List<String> settingIdList = gson.fromJson(req, new TypeToken<List<String>>() {
     }.getType());
-    notebook.bindInterpretersToNote(noteId, settingIdList);
+    notebook.bindInterpretersToNote(SecurityUtils.getPrincipal(), noteId, settingIdList);
     return new JsonResponse<>(Status.OK).build();
   }
 
@@ -458,7 +458,7 @@ public Response deleteParagraph(@PathParam("notebookId") String notebookId,
     }
 
     AuthenticationInfo subject = new AuthenticationInfo(SecurityUtils.getPrincipal());
-    note.removeParagraph(paragraphId);
+    note.removeParagraph(SecurityUtils.getPrincipal(), paragraphId);
     note.persist(subject);
     notebookServer.broadcastNote(note);
 
@@ -599,6 +599,11 @@ public Response runParagraph(@PathParam("notebookId") String notebookId,
     // handle params if presented
     handleParagraphParams(message, note, paragraph);
 
+    AuthenticationInfo subject = new AuthenticationInfo(SecurityUtils.getPrincipal());
+
+    paragraph.setAuthenticationInfo(subject);
+    note.persist(subject);
+
     note.run(paragraph.getId());
     return new JsonResponse<>(Status.OK).build();
   }

zeppelin-server/src/main/java/org/apache/zeppelin/server/ZeppelinServer.java

@@ -89,7 +89,7 @@ public ZeppelinServer() throws Exception {
     this.heliumApplicationFactory = new HeliumApplicationFactory();
     this.schedulerFactory = new SchedulerFactory();
     this.replFactory = new InterpreterFactory(conf, notebookWsServer,
-        notebookWsServer, heliumApplicationFactory, depResolver);
+        notebookWsServer, heliumApplicationFactory, depResolver, SecurityUtils.isAuthenticated());
     this.notebookRepo = new NotebookRepoSync(conf);
     this.notebookIndex = new LuceneSearch();
     this.notebookAuthorization = NotebookAuthorization.init(conf);

zeppelin-server/src/main/java/org/apache/zeppelin/socket/NotebookServer.java

@@ -134,7 +134,7 @@ public void onMessage(NotebookSocket conn, String msg) {
       if (LOG.isTraceEnabled()) {
         LOG.trace("RECEIVE MSG = " + messagereceived);
       }
-      
+
       String ticket = TicketContainer.instance.getTicket(messagereceived.principal);
       if (ticket != null && !ticket.equals(messagereceived.ticket)){
         /* not to pollute logs, log instead of exception */
@@ -472,7 +472,8 @@ public void saveInterpreterBindings(NotebookSocket conn, Message fromMessage) {
       List<String> settingIdList = gson.fromJson(String.valueOf(
           fromMessage.data.get("selectedSettingIds")), new TypeToken<ArrayList<String>>() {
           }.getType());
-      notebook().bindInterpretersToNote(noteId, settingIdList);
+      AuthenticationInfo subject = new AuthenticationInfo(fromMessage.principal);
+      notebook().bindInterpretersToNote(subject.getUser(), noteId, settingIdList);
       broadcastInterpreterBindings(noteId,
           InterpreterBindingUtils.getInterpreterBindings(notebook(), noteId));
     } catch (Exception e) {
@@ -600,6 +601,8 @@ private void sendNote(NotebookSocket conn, HashSet<String> userAndRoles, Noteboo
       return;
     }
 
+    String user = fromMessage.principal;
+
     Note note = notebook.getNote(noteId);
     NotebookAuthorization notebookAuthorization = notebook.getNotebookAuthorization();
     if (note != null) {
@@ -610,7 +613,7 @@ private void sendNote(NotebookSocket conn, HashSet<String> userAndRoles, Noteboo
       }
       addConnectionToNote(note.getId(), conn);
       conn.send(serializeMessage(new Message(OP.NOTE).put("note", note)));
-      sendAllAngularObjects(note, conn);
+      sendAllAngularObjects(note, user, conn);
     } else {
       conn.send(serializeMessage(new Message(OP.NOTE).put("note", null)));
     }
@@ -619,6 +622,7 @@ private void sendNote(NotebookSocket conn, HashSet<String> userAndRoles, Noteboo
   private void sendHomeNote(NotebookSocket conn, HashSet<String> userAndRoles,
                             Notebook notebook, Message fromMessage) throws IOException {
     String noteId = notebook.getConf().getString(ConfVars.ZEPPELIN_NOTEBOOK_HOMESCREEN);
+    String user = fromMessage.principal;
 
     Note note = null;
     if (noteId != null) {
@@ -634,7 +638,7 @@ private void sendHomeNote(NotebookSocket conn, HashSet<String> userAndRoles,
       }
       addConnectionToNote(note.getId(), conn);
       conn.send(serializeMessage(new Message(OP.NOTE).put("note", note)));
-      sendAllAngularObjects(note, conn);
+      sendAllAngularObjects(note, user, conn);
     } else {
       removeConnectionFromAllNote(conn);
       conn.send(serializeMessage(new Message(OP.NOTE).put("note", null)));
@@ -786,6 +790,8 @@ protected Note importNote(NotebookSocket conn, HashSet<String> userAndRoles,
       AuthenticationInfo subject = null;
       if (fromMessage.principal != null) {
         subject = new AuthenticationInfo(fromMessage.principal);
+      } else {
+        subject = new AuthenticationInfo("anonymous");
       }
       note = notebook.importNote(noteJson, noteName, subject);
       note.persist(subject);
@@ -804,7 +810,7 @@ private void removeParagraph(NotebookSocket conn, HashSet<String> userAndRoles,
     String noteId = getOpenNoteId(conn);
     final Note note = notebook.getNote(noteId);
     NotebookAuthorization notebookAuthorization = notebook.getNotebookAuthorization();
-    AuthenticationInfo subject = new AuthenticationInfo(SecurityUtils.getPrincipal());
+    AuthenticationInfo subject = new AuthenticationInfo(fromMessage.principal);
     if (!notebookAuthorization.isWriter(noteId, userAndRoles)) {
       permissionError(conn, "write", fromMessage.principal,
           userAndRoles, notebookAuthorization.getWriters(noteId));
@@ -813,7 +819,7 @@ private void removeParagraph(NotebookSocket conn, HashSet<String> userAndRoles,
 
     /** We dont want to remove the last paragraph */
     if (!note.isLastParagraph(paragraphId)) {
-      note.removeParagraph(paragraphId);
+      note.removeParagraph(subject.getUser(), paragraphId);
       note.persist(subject);
       broadcastNote(note);
     }
@@ -869,6 +875,7 @@ private void angularObjectUpdated(NotebookSocket conn, HashSet<String> userAndRo
     String interpreterGroupId = (String) fromMessage.get("interpreterGroupId");
     String varName = (String) fromMessage.get("name");
     Object varValue = fromMessage.get("value");
+    String user = fromMessage.principal;
     AngularObject ao = null;
     boolean global = false;
     // propagate change to (Remote) AngularObjectRegistry
@@ -877,12 +884,12 @@ private void angularObjectUpdated(NotebookSocket conn, HashSet<String> userAndRo
       List<InterpreterSetting> settings = notebook.getInterpreterFactory()
           .getInterpreterSettings(note.getId());
       for (InterpreterSetting setting : settings) {
-        if (setting.getInterpreterGroup(note.getId()) == null) {
+        if (setting.getInterpreterGroup(user, note.getId()) == null) {
           continue;
         }
-        if (interpreterGroupId.equals(setting.getInterpreterGroup(note.getId()).getId())) {
+        if (interpreterGroupId.equals(setting.getInterpreterGroup(user, note.getId()).getId())) {
           AngularObjectRegistry angularObjectRegistry = setting
-              .getInterpreterGroup(note.getId()).getAngularObjectRegistry();
+              .getInterpreterGroup(user, note.getId()).getAngularObjectRegistry();
 
           // first trying to get local registry
           ao = angularObjectRegistry.get(varName, noteId, paragraphId);
@@ -919,12 +926,12 @@ private void angularObjectUpdated(NotebookSocket conn, HashSet<String> userAndRo
         List<InterpreterSetting> settings = notebook.getInterpreterFactory()
             .getInterpreterSettings(note.getId());
         for (InterpreterSetting setting : settings) {
-          if (setting.getInterpreterGroup(n.getId()) == null) {
+          if (setting.getInterpreterGroup(user, n.getId()) == null) {
             continue;
           }
-          if (interpreterGroupId.equals(setting.getInterpreterGroup(n.getId()).getId())) {
+          if (interpreterGroupId.equals(setting.getInterpreterGroup(user, n.getId()).getId())) {
             AngularObjectRegistry angularObjectRegistry = setting
-                .getInterpreterGroup(n.getId()).getAngularObjectRegistry();
+                .getInterpreterGroup(user, n.getId()).getAngularObjectRegistry();
             this.broadcastExcept(
                 n.getId(),
                 new Message(OP.ANGULAR_OBJECT_UPDATE).put("angularObject", ao)
@@ -1110,7 +1117,7 @@ private void moveParagraph(NotebookSocket conn, HashSet<String> userAndRoles, No
     String noteId = getOpenNoteId(conn);
     final Note note = notebook.getNote(noteId);
     NotebookAuthorization notebookAuthorization = notebook.getNotebookAuthorization();
-    AuthenticationInfo subject = new AuthenticationInfo(SecurityUtils.getPrincipal());
+    AuthenticationInfo subject = new AuthenticationInfo(fromMessage.principal);
     if (!notebookAuthorization.isWriter(noteId, userAndRoles)) {
       permissionError(conn, "write", fromMessage.principal,
           userAndRoles, notebookAuthorization.getWriters(noteId));
@@ -1129,7 +1136,7 @@ private void insertParagraph(NotebookSocket conn, HashSet<String> userAndRoles,
     String noteId = getOpenNoteId(conn);
     final Note note = notebook.getNote(noteId);
     NotebookAuthorization notebookAuthorization = notebook.getNotebookAuthorization();
-    AuthenticationInfo subject = new AuthenticationInfo(SecurityUtils.getPrincipal());
+    AuthenticationInfo subject = new AuthenticationInfo(fromMessage.principal);
     if (!notebookAuthorization.isWriter(noteId, userAndRoles)) {
       permissionError(conn, "write", fromMessage.principal,
           userAndRoles, notebookAuthorization.getWriters(noteId));
@@ -1181,14 +1188,9 @@ private void runParagraph(NotebookSocket conn, HashSet<String> userAndRoles, Not
     String text = (String) fromMessage.get("paragraph");
     p.setText(text);
     p.setTitle((String) fromMessage.get("title"));
-    if (!fromMessage.principal.equals("anonymous")) {
-      AuthenticationInfo authenticationInfo = new AuthenticationInfo(fromMessage.principal,
-          fromMessage.ticket);
-      p.setAuthenticationInfo(authenticationInfo);
-
-    } else {
-      p.setAuthenticationInfo(new AuthenticationInfo());
-    }
+    AuthenticationInfo authenticationInfo =
+        new AuthenticationInfo(fromMessage.principal, fromMessage.ticket);
+    p.setAuthenticationInfo(authenticationInfo);
 
     Map<String, Object> params = (Map<String, Object>) fromMessage
        .get("params");
@@ -1513,7 +1515,7 @@ public void afterStatusChange(Job job, Status before, Status after) {
         LOG.info("Job {} is finished", job.getId());
         try {
           //TODO(khalid): may change interface for JobListener and pass subject from interpreter
-          note.persist(null);
+          note.persist(job instanceof Paragraph ? ((Paragraph) job).getAuthenticationInfo() : null);
         } catch (IOException e) {
           LOG.error(e.toString(), e);
         }
@@ -1569,25 +1571,24 @@ public NotebookEventListener getNotebookInformationListener() {
     return new NotebookInformationListener(this);
   }
 
-  private void sendAllAngularObjects(Note note, NotebookSocket conn) throws IOException {
+  private void sendAllAngularObjects(Note note, String user, NotebookSocket conn)
+      throws IOException {
     List<InterpreterSetting> settings =
         notebook().getInterpreterFactory().getInterpreterSettings(note.getId());
     if (settings == null || settings.size() == 0) {
       return;
     }
 
     for (InterpreterSetting intpSetting : settings) {
-      AngularObjectRegistry registry = intpSetting.getInterpreterGroup(note.getId())
-          .getAngularObjectRegistry();
+      AngularObjectRegistry registry =
+          intpSetting.getInterpreterGroup(user, note.getId()).getAngularObjectRegistry();
       List<AngularObject> objects = registry.getAllWithGlobal(note.getId());
       for (AngularObject object : objects) {
-        conn.send(serializeMessage(new Message(OP.ANGULAR_OBJECT_UPDATE)
-            .put("angularObject", object)
-            .put("interpreterGroupId",
-                intpSetting.getInterpreterGroup(note.getId()).getId())
-            .put("noteId", note.getId())
-            .put("paragraphId", object.getParagraphId())
-        ));
+        conn.send(serializeMessage(
+            new Message(OP.ANGULAR_OBJECT_UPDATE).put("angularObject", object)
+                .put("interpreterGroupId",
+                    intpSetting.getInterpreterGroup(user, note.getId()).getId())
+                .put("noteId", note.getId()).put("paragraphId", object.getParagraphId())));
       }
     }
   }
@@ -1654,9 +1655,10 @@ private void getEditorSetting(NotebookSocket conn, Message fromMessage)
     String paragraphId = (String) fromMessage.get("paragraphId");
     String replName = (String) fromMessage.get("magic");
     String noteId = getOpenNoteId(conn);
+    String user = fromMessage.principal;
     Message resp = new Message(OP.EDITOR_SETTING);
     resp.put("paragraphId", paragraphId);
-    resp.put("editor", notebook().getInterpreterFactory().getEditorSetting(noteId, replName));
+    resp.put("editor", notebook().getInterpreterFactory().getEditorSetting(user, noteId, replName));
     conn.send(serializeMessage(resp));
     return;
   }

zeppelin-server/src/main/java/org/apache/zeppelin/utils/SecurityUtils.java

@@ -119,4 +119,10 @@ public static HashSet<String> getRoles() {
     return roles;
   }
 
+  /**
+   * Checked if shiro enabled or not
+   */
+  public static boolean isAuthenticated() {
+    return org.apache.shiro.SecurityUtils.getSubject().isAuthenticated();
+  }
 }

zeppelin-server/src/main/resources/log4j.properties

@@ -0,0 +1,25 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# 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.
+#
+
+log4j.rootLogger = INFO, stdout
+
+log4j.appender.stdout = org.apache.log4j.ConsoleAppender
+log4j.appender.stdout.layout = org.apache.log4j.PatternLayout
+log4j.appender.stdout.layout.ConversionPattern=%5p [%d] ({%t} %F[%M]:%L) - %m%n
+
+log4j.additivity.org.apache.zeppelin.interpreter = false
+log4j.logger.org.apache.zeppelin.interpreter = DEBUG, stdout

zeppelin-server/src/main/resources/shiro.ini

@@ -18,14 +18,19 @@
 [users]
 # List of users with their password allowed to access Zeppelin.
 # To use a different strategy (LDAP / Database / ...) check the shiro doc at http://shiro.apache.org/configuration.html#Configuration-INISections
-admin = password
+admin = password, admin
+user1 = user1, role1
 
 
 [urls]
-
 # anon means the access is anonymous.
 # authcBasic means Basic Auth Security
 # To enfore security, comment the line below and uncomment the next one
 /** = anon
-#/** = authcBasic
+#/** = authc
 
+[roles]
+role1 = *
+role2 = *
+role3 = *
+admin = *