From adabfbb82621da73c251bce67bb18bdce8117553 Mon Sep 17 00:00:00 2001 From: Mina Lee Date: Fri, 1 Jan 2016 11:04:33 -0800 Subject: [PATCH 1/3] [ZEPPELIN-546] Pass DependencyResolver to InterpreterFactory --- .../org/apache/zeppelin/server/ZeppelinServer.java | 5 ++++- .../zeppelin/interpreter/InterpreterFactory.java | 12 +++++++++--- 2 files changed, 13 insertions(+), 4 deletions(-) diff --git a/zeppelin-server/src/main/java/org/apache/zeppelin/server/ZeppelinServer.java b/zeppelin-server/src/main/java/org/apache/zeppelin/server/ZeppelinServer.java index 7ad2b713037..9e7a97cf862 100644 --- a/zeppelin-server/src/main/java/org/apache/zeppelin/server/ZeppelinServer.java +++ b/zeppelin-server/src/main/java/org/apache/zeppelin/server/ZeppelinServer.java @@ -30,6 +30,7 @@ import org.apache.cxf.jaxrs.servlet.CXFNonSpringJaxrsServlet; import org.apache.zeppelin.conf.ZeppelinConfiguration; import org.apache.zeppelin.conf.ZeppelinConfiguration.ConfVars; +import org.apache.zeppelin.dep.DependencyResolver; import org.apache.zeppelin.interpreter.InterpreterFactory; import org.apache.zeppelin.notebook.Notebook; import org.apache.zeppelin.notebook.repo.NotebookRepo; @@ -73,12 +74,14 @@ public class ZeppelinServer extends Application { private InterpreterFactory replFactory; private NotebookRepo notebookRepo; private SearchService notebookIndex; + private DependencyResolver depResolver; public ZeppelinServer() throws Exception { ZeppelinConfiguration conf = ZeppelinConfiguration.create(); + this.depResolver = new DependencyResolver(conf.getString(ConfVars.ZEPPELIN_DEP_LOCALREPO)); this.schedulerFactory = new SchedulerFactory(); - this.replFactory = new InterpreterFactory(conf, notebookWsServer); + this.replFactory = new InterpreterFactory(conf, notebookWsServer, depResolver); this.notebookRepo = new NotebookRepoSync(conf); this.notebookIndex = new LuceneSearch(); diff --git a/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/InterpreterFactory.java b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/InterpreterFactory.java index fc8cc04fafb..4ff0cc3ad59 100644 --- a/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/InterpreterFactory.java +++ b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/InterpreterFactory.java @@ -23,6 +23,7 @@ import org.apache.commons.lang.NullArgumentException; import org.apache.zeppelin.conf.ZeppelinConfiguration; import org.apache.zeppelin.conf.ZeppelinConfiguration.ConfVars; +import org.apache.zeppelin.dep.DependencyResolver; import org.apache.zeppelin.display.AngularObjectRegistry; import org.apache.zeppelin.display.AngularObjectRegistryListener; import org.apache.zeppelin.interpreter.Interpreter.RegisteredInterpreter; @@ -65,19 +66,24 @@ public class InterpreterFactory { AngularObjectRegistryListener angularObjectRegistryListener; + DependencyResolver depResolver; + public InterpreterFactory(ZeppelinConfiguration conf, - AngularObjectRegistryListener angularObjectRegistryListener) + AngularObjectRegistryListener angularObjectRegistryListener, + DependencyResolver depResolver) throws InterpreterException, IOException { - this(conf, new InterpreterOption(true), angularObjectRegistryListener); + this(conf, new InterpreterOption(true), angularObjectRegistryListener, depResolver); } public InterpreterFactory(ZeppelinConfiguration conf, InterpreterOption defaultOption, - AngularObjectRegistryListener angularObjectRegistryListener) + AngularObjectRegistryListener angularObjectRegistryListener, + DependencyResolver depResolver) throws InterpreterException, IOException { this.conf = conf; this.defaultOption = defaultOption; this.angularObjectRegistryListener = angularObjectRegistryListener; + this.depResolver = depResolver; String replsConf = conf.getString(ConfVars.ZEPPELIN_INTERPRETERS); interpreterClassList = replsConf.split(","); From 0588c34b19c46e175ddcc9062f1f5ffd09f3a0e5 Mon Sep 17 00:00:00 2001 From: Mina Lee Date: Fri, 1 Jan 2016 11:05:05 -0800 Subject: [PATCH 2/3] [ZEPPELIN-546] Fix tests --- .../spark/dep/DependencyResolverTest.java | 20 +++--- .../interpreter/InterpreterFactoryTest.java | 62 +++++++++---------- .../notebook/NoteInterpreterLoaderTest.java | 2 +- .../zeppelin/notebook/NotebookTest.java | 4 +- .../notebook/repo/NotebookRepoSyncTest.java | 2 +- .../notebook/repo/VFSNotebookRepoTest.java | 2 +- 6 files changed, 46 insertions(+), 46 deletions(-) diff --git a/spark/src/test/java/org/apache/zeppelin/spark/dep/DependencyResolverTest.java b/spark/src/test/java/org/apache/zeppelin/spark/dep/DependencyResolverTest.java index e41de60976b..5cbba662d56 100644 --- a/spark/src/test/java/org/apache/zeppelin/spark/dep/DependencyResolverTest.java +++ b/spark/src/test/java/org/apache/zeppelin/spark/dep/DependencyResolverTest.java @@ -19,7 +19,7 @@ import static org.junit.Assert.assertEquals; -import org.apache.zeppelin.spark.dep.DependencyResolver; +import org.apache.zeppelin.spark.dep.SparkDependencyResolver; import org.junit.Test; public class DependencyResolverTest { @@ -30,23 +30,23 @@ public void testInferScalaVersion() { String scalaVersion = version[0] + "." + version[1]; assertEquals("groupId:artifactId:version", - DependencyResolver.inferScalaVersion("groupId:artifactId:version")); + SparkDependencyResolver.inferScalaVersion("groupId:artifactId:version")); assertEquals("groupId:artifactId_" + scalaVersion + ":version", - DependencyResolver.inferScalaVersion("groupId::artifactId:version")); + SparkDependencyResolver.inferScalaVersion("groupId::artifactId:version")); assertEquals("groupId:artifactId:version::test", - DependencyResolver.inferScalaVersion("groupId:artifactId:version::test")); + SparkDependencyResolver.inferScalaVersion("groupId:artifactId:version::test")); assertEquals("*", - DependencyResolver.inferScalaVersion("*")); + SparkDependencyResolver.inferScalaVersion("*")); assertEquals("groupId:*", - DependencyResolver.inferScalaVersion("groupId:*")); + SparkDependencyResolver.inferScalaVersion("groupId:*")); assertEquals("groupId:artifactId*", - DependencyResolver.inferScalaVersion("groupId:artifactId*")); + SparkDependencyResolver.inferScalaVersion("groupId:artifactId*")); assertEquals("groupId:artifactId_" + scalaVersion, - DependencyResolver.inferScalaVersion("groupId::artifactId")); + SparkDependencyResolver.inferScalaVersion("groupId::artifactId")); assertEquals("groupId:artifactId_" + scalaVersion + "*", - DependencyResolver.inferScalaVersion("groupId::artifactId*")); + SparkDependencyResolver.inferScalaVersion("groupId::artifactId*")); assertEquals("groupId:artifactId_" + scalaVersion + ":*", - DependencyResolver.inferScalaVersion("groupId::artifactId:*")); + SparkDependencyResolver.inferScalaVersion("groupId::artifactId:*")); } } diff --git a/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/InterpreterFactoryTest.java b/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/InterpreterFactoryTest.java index 6a69b83b2e9..abd0e3bf11b 100644 --- a/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/InterpreterFactoryTest.java +++ b/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/InterpreterFactoryTest.java @@ -38,32 +38,32 @@ public class InterpreterFactoryTest { - private InterpreterFactory factory; + private InterpreterFactory factory; private File tmpDir; private ZeppelinConfiguration conf; private InterpreterContext context; @Before - public void setUp() throws Exception { + public void setUp() throws Exception { tmpDir = new File(System.getProperty("java.io.tmpdir")+"/ZeppelinLTest_"+System.currentTimeMillis()); tmpDir.mkdirs(); new File(tmpDir, "conf").mkdirs(); MockInterpreter1.register("mock1", "org.apache.zeppelin.interpreter.mock.MockInterpreter1"); - MockInterpreter2.register("mock2", "org.apache.zeppelin.interpreter.mock.MockInterpreter2"); + MockInterpreter2.register("mock2", "org.apache.zeppelin.interpreter.mock.MockInterpreter2"); - System.setProperty(ConfVars.ZEPPELIN_HOME.getVarName(), tmpDir.getAbsolutePath()); - System.setProperty(ConfVars.ZEPPELIN_INTERPRETERS.getVarName(), "org.apache.zeppelin.interpreter.mock.MockInterpreter1,org.apache.zeppelin.interpreter.mock.MockInterpreter2"); - conf = new ZeppelinConfiguration(); - factory = new InterpreterFactory(conf, new InterpreterOption(false), null); - context = new InterpreterContext("note", "id", "title", "text", null, null, null, null); + System.setProperty(ConfVars.ZEPPELIN_HOME.getVarName(), tmpDir.getAbsolutePath()); + System.setProperty(ConfVars.ZEPPELIN_INTERPRETERS.getVarName(), "org.apache.zeppelin.interpreter.mock.MockInterpreter1,org.apache.zeppelin.interpreter.mock.MockInterpreter2"); + conf = new ZeppelinConfiguration(); + factory = new InterpreterFactory(conf, new InterpreterOption(false), null, null); + context = new InterpreterContext("note", "id", "title", "text", null, null, null, null); - } + } - @After - public void tearDown() throws Exception { - delete(tmpDir); - } + @After + public void tearDown() throws Exception { + delete(tmpDir); + } private void delete(File file){ if(file.isFile()) file.delete(); @@ -78,24 +78,24 @@ else if(file.isDirectory()){ } } - @Test - public void testBasic() { - List all = factory.getDefaultInterpreterSettingList(); + @Test + public void testBasic() { + List all = factory.getDefaultInterpreterSettingList(); - // get interpreter - Interpreter repl1 = factory.get(all.get(0)).getInterpreterGroup().getFirst(); - assertFalse(((LazyOpenInterpreter) repl1).isOpen()); - repl1.interpret("repl1", context); - assertTrue(((LazyOpenInterpreter) repl1).isOpen()); + // get interpreter + Interpreter repl1 = factory.get(all.get(0)).getInterpreterGroup().getFirst(); + assertFalse(((LazyOpenInterpreter) repl1).isOpen()); + repl1.interpret("repl1", context); + assertTrue(((LazyOpenInterpreter) repl1).isOpen()); - // try to get unavailable interpreter - assertNull(factory.get("unknown")); + // try to get unavailable interpreter + assertNull(factory.get("unknown")); - // restart interpreter - factory.restart(all.get(0)); - repl1 = factory.get(all.get(0)).getInterpreterGroup().getFirst(); - assertFalse(((LazyOpenInterpreter) repl1).isOpen()); - } + // restart interpreter + factory.restart(all.get(0)); + repl1 = factory.get(all.get(0)).getInterpreterGroup().getFirst(); + assertFalse(((LazyOpenInterpreter) repl1).isOpen()); + } @Test public void testFactoryDefaultList() throws IOException { @@ -119,8 +119,8 @@ public void testExceptions() throws IOException { try { factory.add("a mock", "mock2", null, new Properties()); } catch(NullArgumentException e) { - assertEquals("Test null option" , e.getMessage(),new NullArgumentException("option").getMessage()); - } + assertEquals("Test null option" , e.getMessage(),new NullArgumentException("option").getMessage()); + } try { factory.add("a mock" , "mock2" , new InterpreterOption(false),null); } catch (NullArgumentException e){ @@ -140,7 +140,7 @@ public void testSaveLoad() throws InterpreterException, IOException { factory.add("newsetting", "mock1", new InterpreterOption(false), new Properties()); assertEquals(3, factory.get().size()); - InterpreterFactory factory2 = new InterpreterFactory(conf, null); + InterpreterFactory factory2 = new InterpreterFactory(conf, null, null); assertEquals(3, factory2.get().size()); } } diff --git a/zeppelin-zengine/src/test/java/org/apache/zeppelin/notebook/NoteInterpreterLoaderTest.java b/zeppelin-zengine/src/test/java/org/apache/zeppelin/notebook/NoteInterpreterLoaderTest.java index 2e1f5e3e5b2..a0455eb7ccd 100644 --- a/zeppelin-zengine/src/test/java/org/apache/zeppelin/notebook/NoteInterpreterLoaderTest.java +++ b/zeppelin-zengine/src/test/java/org/apache/zeppelin/notebook/NoteInterpreterLoaderTest.java @@ -58,7 +58,7 @@ public void setUp() throws Exception { MockInterpreter11.register("mock11", "group1", "org.apache.zeppelin.interpreter.mock.MockInterpreter11"); MockInterpreter2.register("mock2", "group2", "org.apache.zeppelin.interpreter.mock.MockInterpreter2"); - factory = new InterpreterFactory(conf, new InterpreterOption(false), null); + factory = new InterpreterFactory(conf, new InterpreterOption(false), null, null); } @After diff --git a/zeppelin-zengine/src/test/java/org/apache/zeppelin/notebook/NotebookTest.java b/zeppelin-zengine/src/test/java/org/apache/zeppelin/notebook/NotebookTest.java index 917ea6b7d7e..34f7a1bb108 100644 --- a/zeppelin-zengine/src/test/java/org/apache/zeppelin/notebook/NotebookTest.java +++ b/zeppelin-zengine/src/test/java/org/apache/zeppelin/notebook/NotebookTest.java @@ -85,7 +85,7 @@ public void setUp() throws Exception { MockInterpreter1.register("mock1", "org.apache.zeppelin.interpreter.mock.MockInterpreter1"); MockInterpreter2.register("mock2", "org.apache.zeppelin.interpreter.mock.MockInterpreter2"); - factory = new InterpreterFactory(conf, new InterpreterOption(false), null); + factory = new InterpreterFactory(conf, new InterpreterOption(false), null, null); SearchService search = mock(SearchService.class); notebookRepo = new VFSNotebookRepo(conf); @@ -172,7 +172,7 @@ public void testPersist() throws IOException, SchedulerException{ note.persist(); Notebook notebook2 = new Notebook( - conf, notebookRepo, schedulerFactory, new InterpreterFactory(conf, null), this, null); + conf, notebookRepo, schedulerFactory, new InterpreterFactory(conf, null, null), this, null); assertEquals(1, notebook2.getAllNotes().size()); } diff --git a/zeppelin-zengine/src/test/java/org/apache/zeppelin/notebook/repo/NotebookRepoSyncTest.java b/zeppelin-zengine/src/test/java/org/apache/zeppelin/notebook/repo/NotebookRepoSyncTest.java index 4e9e1800b2c..60b3ba35566 100644 --- a/zeppelin-zengine/src/test/java/org/apache/zeppelin/notebook/repo/NotebookRepoSyncTest.java +++ b/zeppelin-zengine/src/test/java/org/apache/zeppelin/notebook/repo/NotebookRepoSyncTest.java @@ -87,7 +87,7 @@ public void setUp() throws Exception { MockInterpreter1.register("mock1", "org.apache.zeppelin.interpreter.mock.MockInterpreter1"); MockInterpreter2.register("mock2", "org.apache.zeppelin.interpreter.mock.MockInterpreter2"); - factory = new InterpreterFactory(conf, new InterpreterOption(false), null); + factory = new InterpreterFactory(conf, new InterpreterOption(false), null, null); SearchService search = mock(SearchService.class); notebookRepoSync = new NotebookRepoSync(conf); diff --git a/zeppelin-zengine/src/test/java/org/apache/zeppelin/notebook/repo/VFSNotebookRepoTest.java b/zeppelin-zengine/src/test/java/org/apache/zeppelin/notebook/repo/VFSNotebookRepoTest.java index 65be61b5bc8..cff086dc843 100644 --- a/zeppelin-zengine/src/test/java/org/apache/zeppelin/notebook/repo/VFSNotebookRepoTest.java +++ b/zeppelin-zengine/src/test/java/org/apache/zeppelin/notebook/repo/VFSNotebookRepoTest.java @@ -76,7 +76,7 @@ public void setUp() throws Exception { MockInterpreter1.register("mock1", "org.apache.zeppelin.interpreter.mock.MockInterpreter1"); this.schedulerFactory = new SchedulerFactory(); - factory = new InterpreterFactory(conf, new InterpreterOption(false), null); + factory = new InterpreterFactory(conf, new InterpreterOption(false), null, null); SearchService search = mock(SearchService.class); notebookRepo = new VFSNotebookRepo(conf); From 9f73df3fd977f827883b61144ab39fea4adbe8fc Mon Sep 17 00:00:00 2001 From: Estail7s Date: Tue, 12 Jan 2016 20:48:40 -0800 Subject: [PATCH 3/3] ZEPPELIN-598 Dynamic Interpreter Loader --- .../zeppelin/conf/ZeppelinConfiguration.java | 1 + .../interpreter/InterpreterFactory.java | 99 ++++++++++++++++++- 2 files changed, 96 insertions(+), 4 deletions(-) diff --git a/zeppelin-zengine/src/main/java/org/apache/zeppelin/conf/ZeppelinConfiguration.java b/zeppelin-zengine/src/main/java/org/apache/zeppelin/conf/ZeppelinConfiguration.java index edcf5133667..27ccfab66f9 100755 --- a/zeppelin-zengine/src/main/java/org/apache/zeppelin/conf/ZeppelinConfiguration.java +++ b/zeppelin-zengine/src/main/java/org/apache/zeppelin/conf/ZeppelinConfiguration.java @@ -428,6 +428,7 @@ public static enum ConfVars { // Decide when new note is created, interpreter settings will be binded automatically or not. ZEPPELIN_NOTEBOOK_AUTO_INTERPRETER_BINDING("zeppelin.notebook.autoInterpreterBinding", true), ZEPPELIN_CONF_DIR("zeppelin.conf.dir", "conf"), + ZEPPELIN_INTERPRETER_REPO_DIR("zeppelin.user.interpreter.dir", "interpreter"), ZEPPELIN_DEP_LOCALREPO("zeppelin.dep.localrepo", "local-repo"), // Allows a way to specify a ',' separated list of allowed origins for rest and websockets // i.e. http://localhost:8080 diff --git a/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/InterpreterFactory.java b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/InterpreterFactory.java index 4ff0cc3ad59..f76de3d202d 100644 --- a/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/InterpreterFactory.java +++ b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/InterpreterFactory.java @@ -53,7 +53,7 @@ public class InterpreterFactory { .synchronizedMap(new HashMap()); private ZeppelinConfiguration conf; - String[] interpreterClassList; + List interpreterClassList; private Map interpreterSettings = new HashMap(); @@ -75,7 +75,6 @@ public InterpreterFactory(ZeppelinConfiguration conf, this(conf, new InterpreterOption(true), angularObjectRegistryListener, depResolver); } - public InterpreterFactory(ZeppelinConfiguration conf, InterpreterOption defaultOption, AngularObjectRegistryListener angularObjectRegistryListener, DependencyResolver depResolver) @@ -85,8 +84,10 @@ public InterpreterFactory(ZeppelinConfiguration conf, InterpreterOption defaultO this.angularObjectRegistryListener = angularObjectRegistryListener; this.depResolver = depResolver; String replsConf = conf.getString(ConfVars.ZEPPELIN_INTERPRETERS); - interpreterClassList = replsConf.split(","); - + interpreterClassList = new ArrayList(); + for (String className : replsConf.split(",")) { + interpreterClassList.add(className); + } GsonBuilder builder = new GsonBuilder(); builder.setPrettyPrinting(); builder.registerTypeAdapter(Interpreter.class, new InterpreterSerializer()); @@ -95,6 +96,96 @@ public InterpreterFactory(ZeppelinConfiguration conf, InterpreterOption defaultO init(); } + public boolean loadDynamicInterpreter(String intpGroupName, String intpName, String artifact, + String intpClassName) { + return loadDynamicInterpreter(intpGroupName, intpName, artifact, intpClassName, null, false); + } + + public boolean loadDynamicInterpreter(String intpGroupName, String intpName, String artifact, + String intpClassName, String repositoryUrl, + boolean isSnapShotRepo) { + String[] artifactItem = artifact.split(":"); + String interpreterDesPath = conf.getString(ConfVars.ZEPPELIN_INTERPRETER_REPO_DIR); + interpreterDesPath += "/" + intpGroupName + "/" + intpName + "/"; + + String interpreterLoadPath = conf.getRelativeDir(ConfVars.ZEPPELIN_INTERPRETER_REPO_DIR); + interpreterLoadPath += "/" + intpGroupName + "/" + intpName; + + if (artifactItem.length <= 0) { + logger.error("invalid artifact [" + artifact + "]"); + return false; + } + + try { + if (repositoryUrl != null) { + depResolver.addRepo("dyInterpreterRepo", repositoryUrl, isSnapShotRepo); + } + logger.info("interpreter- path {}", interpreterLoadPath); + depResolver.load(artifact, interpreterDesPath); + setDynamicInterpreter(intpClassName, interpreterLoadPath); + } catch (Exception e) { + logger.error(e.getMessage()); + return false; + } + return true; + } + + public boolean unloadDynamicInterpreter(String intpGorupName, String intpName) { + try { + remove(intpName); + } catch (Exception e) { + logger.error(e.getMessage()); + return false; + } + return true; + } + + protected void setDynamicInterpreter(String interpreterClassName, String fileDirPath) + throws InterpreterException, IOException { + logger.info("loadDynamicInterpreter !!"); + + ClassLoader oldcl = Thread.currentThread().getContextClassLoader(); + interpreterClassList.add(interpreterClassName); + // Load classes + File interpreterDir = new File(fileDirPath); + + if (interpreterDir != null) { + URL[] urls = null; + try { + urls = recursiveBuildLibList(interpreterDir); + } catch (MalformedURLException e1) { + logger.error("Can't load jars ", e1); + } + URLClassLoader ccl = new URLClassLoader(urls, oldcl); + + logger.info("load dynamic interpreter [" + interpreterClassName + "]"); + + try { + Class.forName(interpreterClassName, true, ccl); + Set keys = Interpreter.registeredInterpreters.keySet(); + for (String intName : keys) { + if (interpreterClassName.equals( + Interpreter.registeredInterpreters.get(intName).getClassName())) { + Interpreter.registeredInterpreters.get(intName).setPath(fileDirPath); + logger.info("Interpreter " + intName + " found. class=" + fileDirPath); + cleanCl.put(fileDirPath, ccl); + } + } + } catch (ClassNotFoundException e) { + logger.error("Load error {}", e); + } + } + + for (String settingId : interpreterSettings.keySet()) { + InterpreterSetting setting = interpreterSettings.get(settingId); + logger.info("Interpreter setting group {} : id={}, name={}", + setting.getGroup(), settingId, setting.getName()); + for (Interpreter interpreter : setting.getInterpreterGroup()) { + logger.info(" className = {}", interpreter.getClassName()); + } + } + } + private void init() throws InterpreterException, IOException { ClassLoader oldcl = Thread.currentThread().getContextClassLoader();