You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@zeppelin.apache.org by GitBox <gi...@apache.org> on 2021/03/09 16:14:05 UTC
[GitHub] [zeppelin] Reamer opened a new pull request #4072: [ZEPPELIN-4983] Download local repo to interpreter nodes
Reamer opened a new pull request #4072:
URL: https://github.com/apache/zeppelin/pull/4072
### What is this PR for?
With this PR we synchronise required libraries to an interpreter node that is not running on the same machine as the Zeppelin server.
The download is done via Thrift, which is already in use as communication between Zeppelin server and interpreter.
### What type of PR is it?
- Feature
### Todos
* [ ] - How this can be tested?
### What is the Jira issue?
* https://issues.apache.org/jira/browse/ZEPPELIN-4983
### How should this be tested?
* CI
### Questions:
* Does the licenses files need update? No
* Is there breaking changes for older versions? No
* Does this needs documentation? No
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [zeppelin] zjffdu commented on a change in pull request #4072: [ZEPPELIN-4983] Download local repo to interpreter nodes
Posted by GitBox <gi...@apache.org>.
zjffdu commented on a change in pull request #4072:
URL: https://github.com/apache/zeppelin/pull/4072#discussion_r602144305
##########
File path: zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/RemoteInterpreterEventServer.java
##########
@@ -557,4 +563,58 @@ public void updateParagraphConfig(String noteId,
LOGGER.error("Fail to updateParagraphConfig", e);
}
}
+
+ @Override
+ public List<LibraryMetadata> getAllLibraryMetadatas(String interpreter) throws TException {
+ if (StringUtils.isBlank(interpreter)) {
+ LOGGER.warn("Interpreter is blank");
+ return Collections.emptyList();
+ }
+ File interpreterLocalRepo = new File(
+ zConf.getAbsoluteDir(ZeppelinConfiguration.ConfVars.ZEPPELIN_DEP_LOCALREPO)
+ + File.separator
+ + interpreter);
+ if (!interpreterLocalRepo.exists()) {
+ LOGGER.warn("Local interpreter repository {} for interpreter {} doesn't exists", interpreterLocalRepo,
+ interpreter);
+ return Collections.emptyList();
+ }
+ if (!interpreterLocalRepo.isDirectory()) {
+ LOGGER.warn("Local interpreter repository {} is no folder", interpreterLocalRepo);
+ return Collections.emptyList();
+ }
+ Collection<File> files = FileUtils.listFiles(interpreterLocalRepo, new String[] { "jar" }, false);
Review comment:
Yeah, I think it should be OK for now.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [zeppelin] Reamer commented on a change in pull request #4072: [ZEPPELIN-4983] Download local repo to interpreter nodes
Posted by GitBox <gi...@apache.org>.
Reamer commented on a change in pull request #4072:
URL: https://github.com/apache/zeppelin/pull/4072#discussion_r602129604
##########
File path: zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterDownloader.java
##########
@@ -0,0 +1,140 @@
+/*
+ * 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.
+ */
+
+package org.apache.zeppelin.interpreter.remote;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.file.Files;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+
+import org.apache.commons.io.FileUtils;
+import org.apache.zeppelin.interpreter.Constants;
+import org.apache.zeppelin.interpreter.thrift.LibraryMetadata;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class RemoteInterpreterDownloader {
+ private static final Logger LOGGER = LoggerFactory.getLogger(RemoteInterpreterDownloader.class);
+
+ private final RemoteInterpreterEventClient client;
+ private final String interpreter;
+ private final File localRepoDir;
+
+ public RemoteInterpreterDownloader(
+ String interpreter,
+ RemoteInterpreterEventClient client,
+ File localRepoDir) {
+ this.client = client;
+ this.interpreter = interpreter;
+ this.localRepoDir = localRepoDir;
+ }
+
+ public static void main(String[] args) {
+ String zeppelinServerHost = null;
+ int port = Constants.ZEPPELIN_INTERPRETER_DEFAUlT_PORT;
+ String interpreter = null;
+ String localRepoPath = null;
+ if (args.length > 0) {
+ zeppelinServerHost = args[0];
+ port = Integer.parseInt(args[1]);
+ interpreter = args[2];
+ localRepoPath = args[3];
+ }
+ RemoteInterpreterEventClient intpEventClient = new RemoteInterpreterEventClient(
+ zeppelinServerHost, port, 3);
+
+ RemoteInterpreterDownloader downloader = new RemoteInterpreterDownloader(interpreter,
+ intpEventClient, new File(localRepoPath));
+ downloader.syncAllLibraries();
+ }
+
+ private void syncAllLibraries() {
+ LOGGER.info("Loading all libraries for interpreter {} to {}", interpreter, localRepoDir);
+ List<LibraryMetadata> metadatas = client.getAllLibraryMetadatas(interpreter);
+ if (!localRepoDir.isDirectory()) {
+ LOGGER.error("{} is no directory", localRepoDir);
+ return;
+ }
+ Set<String> syncedLibraries = new HashSet<>();
+ // Add or update new libraries
+ for (LibraryMetadata metadata : metadatas) {
+ File library = new File(localRepoDir, metadata.getName());
+ addOrUpdateLibrary(library, metadata);
+ syncedLibraries.add(metadata.getName());
+ }
+ // Delete old Jar files
+ for (File file : FileUtils.listFiles(localRepoDir, new String[] { "jar" }, false)) {
+ if (!syncedLibraries.contains(file.getName())) {
+ try {
+ LOGGER.info("Delete {}, because it's not present on the server side", file.toPath());
+ Files.delete(file.toPath());
+ } catch (IOException e) {
+ LOGGER.error("Unable to delete old library {} during sync.", file, e);
+ }
+ }
+ }
+ }
+
+ private void addOrUpdateLibrary(File library, LibraryMetadata metadata) {
+ try {
+ if (library.exists() && library.canRead()) {
+ long localChecksum = FileUtils.checksumCRC32(library);
+ if (localChecksum == metadata.getChecksum()) {
+ // nothing to do if checksum is matching
+ LOGGER.info("Library {} is present ", library.getName());
+ } else {
+ // checksum didn't match
+ Files.delete(library.toPath());
+ downloadLibrary(library, metadata);
+ }
+ } else {
+ downloadLibrary(library, metadata);
+ }
+ } catch (IOException e) {
+ LOGGER.error("Can not add or update library {}", library, e);
+ }
+ }
+
+ private void downloadLibrary(File library, LibraryMetadata metadata) {
+ LOGGER.debug("Trying to download library {} to {}", metadata.getName(), library);
+ try {
+ if (library.createNewFile()) {
+ int tries = 0;
+ while (tries < 3) {
+ ByteBuffer bytes = client.getLibrary(interpreter, metadata.getName());
+ FileUtils.writeByteArrayToFile(library, bytes.array());
Review comment:
I have improved the code, please review it again.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [zeppelin] cuspymd commented on a change in pull request #4072: [ZEPPELIN-4983] Download local repo to interpreter nodes
Posted by GitBox <gi...@apache.org>.
cuspymd commented on a change in pull request #4072:
URL: https://github.com/apache/zeppelin/pull/4072#discussion_r603933999
##########
File path: zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterDownloader.java
##########
@@ -51,22 +50,21 @@ public RemoteInterpreterDownloader(
}
public static void main(String[] args) {
- String zeppelinServerHost = null;
- int port = Constants.ZEPPELIN_INTERPRETER_DEFAUlT_PORT;
- String interpreter = null;
- String localRepoPath = null;
- if (args.length > 0) {
- zeppelinServerHost = args[0];
- port = Integer.parseInt(args[1]);
- interpreter = args[2];
- localRepoPath = args[3];
- }
- RemoteInterpreterEventClient intpEventClient = new RemoteInterpreterEventClient(
- zeppelinServerHost, port, 3);
+ if (args.length == 4) {
+ String zeppelinServerHost = args[0];
+ int port = Integer.parseInt(args[1]);
+ String interpreter = args[2];
+ String localRepoPath = args[3];
+ RemoteInterpreterEventClient intpEventClient = new RemoteInterpreterEventClient(
+ zeppelinServerHost, port, 3);
- RemoteInterpreterDownloader downloader = new RemoteInterpreterDownloader(interpreter,
- intpEventClient, new File(localRepoPath));
- downloader.syncAllLibraries();
+ RemoteInterpreterDownloader downloader = new RemoteInterpreterDownloader(interpreter,
+ intpEventClient, new File(localRepoPath));
+ downloader.syncAllLibraries();
+ } else {
+ LOGGER.error(
+ "Wrong amount of Arguments. Expected: [ZeppelinHostname] [ZeppelinPort] [InterpreterName] [LocalRepoPath]");
+ }
Review comment:
In this case, the script executing this program cannot determine that this program has not been executed normally due to the mismatch of the arguments. Is this intentional?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [zeppelin] zjffdu commented on a change in pull request #4072: [ZEPPELIN-4983] Download local repo to interpreter nodes
Posted by GitBox <gi...@apache.org>.
zjffdu commented on a change in pull request #4072:
URL: https://github.com/apache/zeppelin/pull/4072#discussion_r602051890
##########
File path: zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/RemoteInterpreterEventServer.java
##########
@@ -557,4 +563,58 @@ public void updateParagraphConfig(String noteId,
LOGGER.error("Fail to updateParagraphConfig", e);
}
}
+
+ @Override
+ public List<LibraryMetadata> getAllLibraryMetadatas(String interpreter) throws TException {
+ if (StringUtils.isBlank(interpreter)) {
+ LOGGER.warn("Interpreter is blank");
+ return Collections.emptyList();
+ }
+ File interpreterLocalRepo = new File(
+ zConf.getAbsoluteDir(ZeppelinConfiguration.ConfVars.ZEPPELIN_DEP_LOCALREPO)
+ + File.separator
+ + interpreter);
+ if (!interpreterLocalRepo.exists()) {
+ LOGGER.warn("Local interpreter repository {} for interpreter {} doesn't exists", interpreterLocalRepo,
+ interpreter);
+ return Collections.emptyList();
+ }
+ if (!interpreterLocalRepo.isDirectory()) {
+ LOGGER.warn("Local interpreter repository {} is no folder", interpreterLocalRepo);
+ return Collections.emptyList();
+ }
+ Collection<File> files = FileUtils.listFiles(interpreterLocalRepo, new String[] { "jar" }, false);
Review comment:
Only jar file is transferred ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [zeppelin] Teo920127 edited a comment on pull request #4072: [ZEPPELIN-4983] Download local repo to interpreter nodes
Posted by GitBox <gi...@apache.org>.
Teo920127 edited a comment on pull request #4072:
URL: https://github.com/apache/zeppelin/pull/4072#issuecomment-802489730
In Kubernetes, Can interpreters of the same type (such as Python) be configured to enable different images?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [zeppelin] cuspymd commented on a change in pull request #4072: [ZEPPELIN-4983] Download local repo to interpreter nodes
Posted by GitBox <gi...@apache.org>.
cuspymd commented on a change in pull request #4072:
URL: https://github.com/apache/zeppelin/pull/4072#discussion_r602090937
##########
File path: zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterDownloader.java
##########
@@ -0,0 +1,140 @@
+/*
+ * 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.
+ */
+
+package org.apache.zeppelin.interpreter.remote;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.file.Files;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+
+import org.apache.commons.io.FileUtils;
+import org.apache.zeppelin.interpreter.Constants;
+import org.apache.zeppelin.interpreter.thrift.LibraryMetadata;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class RemoteInterpreterDownloader {
+ private static final Logger LOGGER = LoggerFactory.getLogger(RemoteInterpreterDownloader.class);
+
+ private final RemoteInterpreterEventClient client;
+ private final String interpreter;
+ private final File localRepoDir;
+
+ public RemoteInterpreterDownloader(
+ String interpreter,
+ RemoteInterpreterEventClient client,
+ File localRepoDir) {
+ this.client = client;
+ this.interpreter = interpreter;
+ this.localRepoDir = localRepoDir;
+ }
+
+ public static void main(String[] args) {
+ String zeppelinServerHost = null;
+ int port = Constants.ZEPPELIN_INTERPRETER_DEFAUlT_PORT;
+ String interpreter = null;
+ String localRepoPath = null;
+ if (args.length > 0) {
+ zeppelinServerHost = args[0];
+ port = Integer.parseInt(args[1]);
+ interpreter = args[2];
+ localRepoPath = args[3];
+ }
+ RemoteInterpreterEventClient intpEventClient = new RemoteInterpreterEventClient(
+ zeppelinServerHost, port, 3);
+
+ RemoteInterpreterDownloader downloader = new RemoteInterpreterDownloader(interpreter,
+ intpEventClient, new File(localRepoPath));
+ downloader.syncAllLibraries();
+ }
+
+ private void syncAllLibraries() {
+ LOGGER.info("Loading all libraries for interpreter {} to {}", interpreter, localRepoDir);
+ List<LibraryMetadata> metadatas = client.getAllLibraryMetadatas(interpreter);
+ if (!localRepoDir.isDirectory()) {
+ LOGGER.error("{} is no directory", localRepoDir);
+ return;
+ }
+ Set<String> syncedLibraries = new HashSet<>();
+ // Add or update new libraries
+ for (LibraryMetadata metadata : metadatas) {
+ File library = new File(localRepoDir, metadata.getName());
+ addOrUpdateLibrary(library, metadata);
+ syncedLibraries.add(metadata.getName());
+ }
+ // Delete old Jar files
+ for (File file : FileUtils.listFiles(localRepoDir, new String[] { "jar" }, false)) {
+ if (!syncedLibraries.contains(file.getName())) {
+ try {
+ LOGGER.info("Delete {}, because it's not present on the server side", file.toPath());
+ Files.delete(file.toPath());
+ } catch (IOException e) {
+ LOGGER.error("Unable to delete old library {} during sync.", file, e);
+ }
+ }
+ }
+ }
+
+ private void addOrUpdateLibrary(File library, LibraryMetadata metadata) {
+ try {
+ if (library.exists() && library.canRead()) {
+ long localChecksum = FileUtils.checksumCRC32(library);
+ if (localChecksum == metadata.getChecksum()) {
+ // nothing to do if checksum is matching
+ LOGGER.info("Library {} is present ", library.getName());
+ } else {
+ // checksum didn't match
+ Files.delete(library.toPath());
+ downloadLibrary(library, metadata);
+ }
+ } else {
+ downloadLibrary(library, metadata);
+ }
+ } catch (IOException e) {
+ LOGGER.error("Can not add or update library {}", library, e);
+ }
+ }
+
+ private void downloadLibrary(File library, LibraryMetadata metadata) {
+ LOGGER.debug("Trying to download library {} to {}", metadata.getName(), library);
+ try {
+ if (library.createNewFile()) {
+ int tries = 0;
+ while (tries < 3) {
+ ByteBuffer bytes = client.getLibrary(interpreter, metadata.getName());
+ FileUtils.writeByteArrayToFile(library, bytes.array());
Review comment:
`bytes` seems be nullable. Is it ok without null checking?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [zeppelin] Reamer commented on a change in pull request #4072: [ZEPPELIN-4983] Download local repo to interpreter nodes
Posted by GitBox <gi...@apache.org>.
Reamer commented on a change in pull request #4072:
URL: https://github.com/apache/zeppelin/pull/4072#discussion_r603865473
##########
File path: zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterDownloader.java
##########
@@ -0,0 +1,148 @@
+/*
+ * 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.
+ */
+
+package org.apache.zeppelin.interpreter.remote;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.file.Files;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+
+import org.apache.commons.io.FileUtils;
+import org.apache.zeppelin.interpreter.Constants;
+import org.apache.zeppelin.interpreter.thrift.LibraryMetadata;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class RemoteInterpreterDownloader {
+ private static final Logger LOGGER = LoggerFactory.getLogger(RemoteInterpreterDownloader.class);
+
+ private static final int MAX_LIBRARY_DOWNLOAD_ATTEMPTS = 3;
+
+ private final RemoteInterpreterEventClient client;
+ private final String interpreter;
+ private final File localRepoDir;
+
+ public RemoteInterpreterDownloader(
+ String interpreter,
+ RemoteInterpreterEventClient client,
+ File localRepoDir) {
+ this.client = client;
+ this.interpreter = interpreter;
+ this.localRepoDir = localRepoDir;
+ }
+
+ public static void main(String[] args) {
+ String zeppelinServerHost = null;
+ int port = Constants.ZEPPELIN_INTERPRETER_DEFAUlT_PORT;
+ String interpreter = null;
+ String localRepoPath = null;
+ if (args.length > 0) {
+ zeppelinServerHost = args[0];
+ port = Integer.parseInt(args[1]);
+ interpreter = args[2];
+ localRepoPath = args[3];
+ }
+ RemoteInterpreterEventClient intpEventClient = new RemoteInterpreterEventClient(
+ zeppelinServerHost, port, 3);
+
+ RemoteInterpreterDownloader downloader = new RemoteInterpreterDownloader(interpreter,
+ intpEventClient, new File(localRepoPath));
+ downloader.syncAllLibraries();
+ }
+
+ private void syncAllLibraries() {
+ LOGGER.info("Loading all libraries for interpreter {} to {}", interpreter, localRepoDir);
+ List<LibraryMetadata> metadatas = client.getAllLibraryMetadatas(interpreter);
+ if (!localRepoDir.isDirectory()) {
+ LOGGER.error("{} is no directory", localRepoDir);
+ return;
+ }
+ Set<String> syncedLibraries = new HashSet<>();
+ // Add or update new libraries
+ for (LibraryMetadata metadata : metadatas) {
+ File library = new File(localRepoDir, metadata.getName());
+ addOrUpdateLibrary(library, metadata);
+ syncedLibraries.add(metadata.getName());
+ }
+ // Delete old Jar files
+ for (File file : FileUtils.listFiles(localRepoDir, new String[] { "jar" }, false)) {
+ if (!syncedLibraries.contains(file.getName())) {
+ try {
+ LOGGER.info("Delete {}, because it's not present on the server side", file.toPath());
+ Files.delete(file.toPath());
+ } catch (IOException e) {
+ LOGGER.error("Unable to delete old library {} during sync.", file, e);
+ }
+ }
+ }
+ }
+
+ private void addOrUpdateLibrary(File library, LibraryMetadata metadata) {
+ try {
+ if (library.exists() && library.canRead()) {
+ long localChecksum = FileUtils.checksumCRC32(library);
+ if (localChecksum == metadata.getChecksum()) {
+ // nothing to do if checksum is matching
+ LOGGER.info("Library {} is present ", library.getName());
+ } else {
+ // checksum didn't match
+ Files.delete(library.toPath());
+ downloadLibrary(library, metadata);
+ }
+ } else {
+ downloadLibrary(library, metadata);
+ }
+ } catch (IOException e) {
+ LOGGER.error("Can not add or update library {}", library, e);
+ }
+ }
+
+ private void downloadLibrary(File library, LibraryMetadata metadata) {
+ LOGGER.debug("Trying to download library {} to {}", metadata.getName(), library);
+ try {
+ if (library.createNewFile()) {
+ int attempt = 0;
+ while (attempt < MAX_LIBRARY_DOWNLOAD_ATTEMPTS) {
+ ByteBuffer bytes = client.getLibrary(interpreter, metadata.getName());
+ if (bytes == null) {
+ LOGGER.error("ByteBuffer of library {} is null."
+ + " For a detailed message take a look into Zeppelin-Server log. Attempt {} of {}",
+ metadata.getName(), ++attempt, MAX_LIBRARY_DOWNLOAD_ATTEMPTS);
Review comment:
In principle, this is a network communication. We already handle ThriftExceptions in the `PooledRemoteClient`, so these should not occur.
I think a retry is not bad because IOExceptions can occur in Zeppelin Server. What is your opinion on this?
Btw. The happy path never returns null.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [zeppelin] zjffdu commented on a change in pull request #4072: [ZEPPELIN-4983] Download local repo to interpreter nodes
Posted by GitBox <gi...@apache.org>.
zjffdu commented on a change in pull request #4072:
URL: https://github.com/apache/zeppelin/pull/4072#discussion_r602051890
##########
File path: zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/RemoteInterpreterEventServer.java
##########
@@ -557,4 +563,58 @@ public void updateParagraphConfig(String noteId,
LOGGER.error("Fail to updateParagraphConfig", e);
}
}
+
+ @Override
+ public List<LibraryMetadata> getAllLibraryMetadatas(String interpreter) throws TException {
+ if (StringUtils.isBlank(interpreter)) {
+ LOGGER.warn("Interpreter is blank");
+ return Collections.emptyList();
+ }
+ File interpreterLocalRepo = new File(
+ zConf.getAbsoluteDir(ZeppelinConfiguration.ConfVars.ZEPPELIN_DEP_LOCALREPO)
+ + File.separator
+ + interpreter);
+ if (!interpreterLocalRepo.exists()) {
+ LOGGER.warn("Local interpreter repository {} for interpreter {} doesn't exists", interpreterLocalRepo,
+ interpreter);
+ return Collections.emptyList();
+ }
+ if (!interpreterLocalRepo.isDirectory()) {
+ LOGGER.warn("Local interpreter repository {} is no folder", interpreterLocalRepo);
+ return Collections.emptyList();
+ }
+ Collection<File> files = FileUtils.listFiles(interpreterLocalRepo, new String[] { "jar" }, false);
Review comment:
Only jar is transferred ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [zeppelin] Reamer commented on a change in pull request #4072: [ZEPPELIN-4983] Download local repo to interpreter nodes
Posted by GitBox <gi...@apache.org>.
Reamer commented on a change in pull request #4072:
URL: https://github.com/apache/zeppelin/pull/4072#discussion_r602111388
##########
File path: zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/RemoteInterpreterEventServer.java
##########
@@ -557,4 +563,58 @@ public void updateParagraphConfig(String noteId,
LOGGER.error("Fail to updateParagraphConfig", e);
}
}
+
+ @Override
+ public List<LibraryMetadata> getAllLibraryMetadatas(String interpreter) throws TException {
+ if (StringUtils.isBlank(interpreter)) {
+ LOGGER.warn("Interpreter is blank");
+ return Collections.emptyList();
+ }
+ File interpreterLocalRepo = new File(
+ zConf.getAbsoluteDir(ZeppelinConfiguration.ConfVars.ZEPPELIN_DEP_LOCALREPO)
+ + File.separator
+ + interpreter);
+ if (!interpreterLocalRepo.exists()) {
+ LOGGER.warn("Local interpreter repository {} for interpreter {} doesn't exists", interpreterLocalRepo,
+ interpreter);
+ return Collections.emptyList();
+ }
+ if (!interpreterLocalRepo.isDirectory()) {
+ LOGGER.warn("Local interpreter repository {} is no folder", interpreterLocalRepo);
+ return Collections.emptyList();
+ }
+ Collection<File> files = FileUtils.listFiles(interpreterLocalRepo, new String[] { "jar" }, false);
Review comment:
I can add more extensions, but at the moment I don't know which ones.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [zeppelin] Reamer commented on a change in pull request #4072: [ZEPPELIN-4983] Download local repo to interpreter nodes
Posted by GitBox <gi...@apache.org>.
Reamer commented on a change in pull request #4072:
URL: https://github.com/apache/zeppelin/pull/4072#discussion_r603963761
##########
File path: zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterDownloader.java
##########
@@ -51,22 +50,21 @@ public RemoteInterpreterDownloader(
}
public static void main(String[] args) {
- String zeppelinServerHost = null;
- int port = Constants.ZEPPELIN_INTERPRETER_DEFAUlT_PORT;
- String interpreter = null;
- String localRepoPath = null;
- if (args.length > 0) {
- zeppelinServerHost = args[0];
- port = Integer.parseInt(args[1]);
- interpreter = args[2];
- localRepoPath = args[3];
- }
- RemoteInterpreterEventClient intpEventClient = new RemoteInterpreterEventClient(
- zeppelinServerHost, port, 3);
+ if (args.length == 4) {
+ String zeppelinServerHost = args[0];
+ int port = Integer.parseInt(args[1]);
+ String interpreter = args[2];
+ String localRepoPath = args[3];
+ RemoteInterpreterEventClient intpEventClient = new RemoteInterpreterEventClient(
+ zeppelinServerHost, port, 3);
- RemoteInterpreterDownloader downloader = new RemoteInterpreterDownloader(interpreter,
- intpEventClient, new File(localRepoPath));
- downloader.syncAllLibraries();
+ RemoteInterpreterDownloader downloader = new RemoteInterpreterDownloader(interpreter,
+ intpEventClient, new File(localRepoPath));
+ downloader.syncAllLibraries();
+ } else {
+ LOGGER.error(
+ "Wrong amount of Arguments. Expected: [ZeppelinHostname] [ZeppelinPort] [InterpreterName] [LocalRepoPath]");
+ }
Review comment:
Thanks for your review, an exit code has been added.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [zeppelin] Reamer commented on pull request #4072: [ZEPPELIN-4983] Download local repo to interpreter nodes
Posted by GitBox <gi...@apache.org>.
Reamer commented on pull request #4072:
URL: https://github.com/apache/zeppelin/pull/4072#issuecomment-817859179
I will merge this PR into master and branch-0.9 on Thursday if no further comments are received.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [zeppelin] cuspymd commented on a change in pull request #4072: [ZEPPELIN-4983] Download local repo to interpreter nodes
Posted by GitBox <gi...@apache.org>.
cuspymd commented on a change in pull request #4072:
URL: https://github.com/apache/zeppelin/pull/4072#discussion_r603758497
##########
File path: zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterDownloader.java
##########
@@ -0,0 +1,148 @@
+/*
+ * 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.
+ */
+
+package org.apache.zeppelin.interpreter.remote;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.file.Files;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+
+import org.apache.commons.io.FileUtils;
+import org.apache.zeppelin.interpreter.Constants;
+import org.apache.zeppelin.interpreter.thrift.LibraryMetadata;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class RemoteInterpreterDownloader {
+ private static final Logger LOGGER = LoggerFactory.getLogger(RemoteInterpreterDownloader.class);
+
+ private static final int MAX_LIBRARY_DOWNLOAD_ATTEMPTS = 3;
+
+ private final RemoteInterpreterEventClient client;
+ private final String interpreter;
+ private final File localRepoDir;
+
+ public RemoteInterpreterDownloader(
+ String interpreter,
+ RemoteInterpreterEventClient client,
+ File localRepoDir) {
+ this.client = client;
+ this.interpreter = interpreter;
+ this.localRepoDir = localRepoDir;
+ }
+
+ public static void main(String[] args) {
+ String zeppelinServerHost = null;
+ int port = Constants.ZEPPELIN_INTERPRETER_DEFAUlT_PORT;
+ String interpreter = null;
+ String localRepoPath = null;
+ if (args.length > 0) {
+ zeppelinServerHost = args[0];
+ port = Integer.parseInt(args[1]);
+ interpreter = args[2];
+ localRepoPath = args[3];
+ }
Review comment:
It seems that the argument check is not strict because the location where this file is used is fixed. All four arguments seem be mandatory, Wouldn't it be better to add an `assert` to work only with "`args.length` == 4"?
##########
File path: zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterDownloader.java
##########
@@ -0,0 +1,148 @@
+/*
+ * 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.
+ */
+
+package org.apache.zeppelin.interpreter.remote;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.file.Files;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+
+import org.apache.commons.io.FileUtils;
+import org.apache.zeppelin.interpreter.Constants;
+import org.apache.zeppelin.interpreter.thrift.LibraryMetadata;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class RemoteInterpreterDownloader {
+ private static final Logger LOGGER = LoggerFactory.getLogger(RemoteInterpreterDownloader.class);
+
+ private static final int MAX_LIBRARY_DOWNLOAD_ATTEMPTS = 3;
+
+ private final RemoteInterpreterEventClient client;
+ private final String interpreter;
+ private final File localRepoDir;
+
+ public RemoteInterpreterDownloader(
+ String interpreter,
+ RemoteInterpreterEventClient client,
+ File localRepoDir) {
+ this.client = client;
+ this.interpreter = interpreter;
+ this.localRepoDir = localRepoDir;
+ }
+
+ public static void main(String[] args) {
+ String zeppelinServerHost = null;
+ int port = Constants.ZEPPELIN_INTERPRETER_DEFAUlT_PORT;
+ String interpreter = null;
+ String localRepoPath = null;
+ if (args.length > 0) {
+ zeppelinServerHost = args[0];
+ port = Integer.parseInt(args[1]);
+ interpreter = args[2];
+ localRepoPath = args[3];
+ }
+ RemoteInterpreterEventClient intpEventClient = new RemoteInterpreterEventClient(
+ zeppelinServerHost, port, 3);
+
+ RemoteInterpreterDownloader downloader = new RemoteInterpreterDownloader(interpreter,
+ intpEventClient, new File(localRepoPath));
+ downloader.syncAllLibraries();
+ }
+
+ private void syncAllLibraries() {
+ LOGGER.info("Loading all libraries for interpreter {} to {}", interpreter, localRepoDir);
+ List<LibraryMetadata> metadatas = client.getAllLibraryMetadatas(interpreter);
+ if (!localRepoDir.isDirectory()) {
+ LOGGER.error("{} is no directory", localRepoDir);
+ return;
+ }
+ Set<String> syncedLibraries = new HashSet<>();
+ // Add or update new libraries
+ for (LibraryMetadata metadata : metadatas) {
+ File library = new File(localRepoDir, metadata.getName());
+ addOrUpdateLibrary(library, metadata);
+ syncedLibraries.add(metadata.getName());
+ }
+ // Delete old Jar files
+ for (File file : FileUtils.listFiles(localRepoDir, new String[] { "jar" }, false)) {
+ if (!syncedLibraries.contains(file.getName())) {
+ try {
+ LOGGER.info("Delete {}, because it's not present on the server side", file.toPath());
+ Files.delete(file.toPath());
+ } catch (IOException e) {
+ LOGGER.error("Unable to delete old library {} during sync.", file, e);
+ }
+ }
+ }
+ }
+
+ private void addOrUpdateLibrary(File library, LibraryMetadata metadata) {
+ try {
+ if (library.exists() && library.canRead()) {
+ long localChecksum = FileUtils.checksumCRC32(library);
+ if (localChecksum == metadata.getChecksum()) {
+ // nothing to do if checksum is matching
+ LOGGER.info("Library {} is present ", library.getName());
+ } else {
+ // checksum didn't match
+ Files.delete(library.toPath());
+ downloadLibrary(library, metadata);
+ }
+ } else {
+ downloadLibrary(library, metadata);
+ }
+ } catch (IOException e) {
+ LOGGER.error("Can not add or update library {}", library, e);
+ }
+ }
+
+ private void downloadLibrary(File library, LibraryMetadata metadata) {
+ LOGGER.debug("Trying to download library {} to {}", metadata.getName(), library);
+ try {
+ if (library.createNewFile()) {
+ int attempt = 0;
+ while (attempt < MAX_LIBRARY_DOWNLOAD_ATTEMPTS) {
+ ByteBuffer bytes = client.getLibrary(interpreter, metadata.getName());
+ if (bytes == null) {
+ LOGGER.error("ByteBuffer of library {} is null."
+ + " For a detailed message take a look into Zeppelin-Server log. Attempt {} of {}",
+ metadata.getName(), ++attempt, MAX_LIBRARY_DOWNLOAD_ATTEMPTS);
Review comment:
Do you think the error if the function 'getLibrary()' returns null can be recovered by redoing it?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [zeppelin] asfgit closed pull request #4072: [ZEPPELIN-4983] Download local repo to interpreter nodes
Posted by GitBox <gi...@apache.org>.
asfgit closed pull request #4072:
URL: https://github.com/apache/zeppelin/pull/4072
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [zeppelin] Reamer commented on pull request #4072: [ZEPPELIN-4983] Download local repo to interpreter nodes
Posted by GitBox <gi...@apache.org>.
Reamer commented on pull request #4072:
URL: https://github.com/apache/zeppelin/pull/4072#issuecomment-794113800
How can this PR be tested? I am open to ideas. At the moment I have done some successful manual tests.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [zeppelin] Reamer commented on pull request #4072: [ZEPPELIN-4983] Download local repo to interpreter nodes
Posted by GitBox <gi...@apache.org>.
Reamer commented on pull request #4072:
URL: https://github.com/apache/zeppelin/pull/4072#issuecomment-799153771
Hi @Leemoonsoo,
Thank you very much for your feedback. K8s integration tests are a great idea, but the Github actions don't work on the local machine. I would prefer a solution that also works locally (e.g. with maven). Currently I am quite happy with the Kubernetes mock tests.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [zeppelin] zjffdu commented on a change in pull request #4072: [ZEPPELIN-4983] Download local repo to interpreter nodes
Posted by GitBox <gi...@apache.org>.
zjffdu commented on a change in pull request #4072:
URL: https://github.com/apache/zeppelin/pull/4072#discussion_r602102340
##########
File path: zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/RemoteInterpreterEventServer.java
##########
@@ -557,4 +563,58 @@ public void updateParagraphConfig(String noteId,
LOGGER.error("Fail to updateParagraphConfig", e);
}
}
+
+ @Override
+ public List<LibraryMetadata> getAllLibraryMetadatas(String interpreter) throws TException {
+ if (StringUtils.isBlank(interpreter)) {
+ LOGGER.warn("Interpreter is blank");
+ return Collections.emptyList();
+ }
+ File interpreterLocalRepo = new File(
+ zConf.getAbsoluteDir(ZeppelinConfiguration.ConfVars.ZEPPELIN_DEP_LOCALREPO)
+ + File.separator
+ + interpreter);
+ if (!interpreterLocalRepo.exists()) {
+ LOGGER.warn("Local interpreter repository {} for interpreter {} doesn't exists", interpreterLocalRepo,
+ interpreter);
+ return Collections.emptyList();
+ }
+ if (!interpreterLocalRepo.isDirectory()) {
+ LOGGER.warn("Local interpreter repository {} is no folder", interpreterLocalRepo);
+ return Collections.emptyList();
+ }
+ Collection<File> files = FileUtils.listFiles(interpreterLocalRepo, new String[] { "jar" }, false);
Review comment:
In most of cases it should be OK, Just concern some corner cases that some library may depend on non-jar files.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [zeppelin] Reamer commented on a change in pull request #4072: [ZEPPELIN-4983] Download local repo to interpreter nodes
Posted by GitBox <gi...@apache.org>.
Reamer commented on a change in pull request #4072:
URL: https://github.com/apache/zeppelin/pull/4072#discussion_r603876767
##########
File path: zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterDownloader.java
##########
@@ -0,0 +1,148 @@
+/*
+ * 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.
+ */
+
+package org.apache.zeppelin.interpreter.remote;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.file.Files;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+
+import org.apache.commons.io.FileUtils;
+import org.apache.zeppelin.interpreter.Constants;
+import org.apache.zeppelin.interpreter.thrift.LibraryMetadata;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class RemoteInterpreterDownloader {
+ private static final Logger LOGGER = LoggerFactory.getLogger(RemoteInterpreterDownloader.class);
+
+ private static final int MAX_LIBRARY_DOWNLOAD_ATTEMPTS = 3;
+
+ private final RemoteInterpreterEventClient client;
+ private final String interpreter;
+ private final File localRepoDir;
+
+ public RemoteInterpreterDownloader(
+ String interpreter,
+ RemoteInterpreterEventClient client,
+ File localRepoDir) {
+ this.client = client;
+ this.interpreter = interpreter;
+ this.localRepoDir = localRepoDir;
+ }
+
+ public static void main(String[] args) {
+ String zeppelinServerHost = null;
+ int port = Constants.ZEPPELIN_INTERPRETER_DEFAUlT_PORT;
+ String interpreter = null;
+ String localRepoPath = null;
+ if (args.length > 0) {
+ zeppelinServerHost = args[0];
+ port = Integer.parseInt(args[1]);
+ interpreter = args[2];
+ localRepoPath = args[3];
+ }
Review comment:
You are right. I have changed the implementation to properly evaluate the mandatory arguments.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [zeppelin] Reamer commented on a change in pull request #4072: [ZEPPELIN-4983] Download local repo to interpreter nodes
Posted by GitBox <gi...@apache.org>.
Reamer commented on a change in pull request #4072:
URL: https://github.com/apache/zeppelin/pull/4072#discussion_r602089778
##########
File path: zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/RemoteInterpreterEventServer.java
##########
@@ -557,4 +563,58 @@ public void updateParagraphConfig(String noteId,
LOGGER.error("Fail to updateParagraphConfig", e);
}
}
+
+ @Override
+ public List<LibraryMetadata> getAllLibraryMetadatas(String interpreter) throws TException {
+ if (StringUtils.isBlank(interpreter)) {
+ LOGGER.warn("Interpreter is blank");
+ return Collections.emptyList();
+ }
+ File interpreterLocalRepo = new File(
+ zConf.getAbsoluteDir(ZeppelinConfiguration.ConfVars.ZEPPELIN_DEP_LOCALREPO)
+ + File.separator
+ + interpreter);
+ if (!interpreterLocalRepo.exists()) {
+ LOGGER.warn("Local interpreter repository {} for interpreter {} doesn't exists", interpreterLocalRepo,
+ interpreter);
+ return Collections.emptyList();
+ }
+ if (!interpreterLocalRepo.isDirectory()) {
+ LOGGER.warn("Local interpreter repository {} is no folder", interpreterLocalRepo);
+ return Collections.emptyList();
+ }
+ Collection<File> files = FileUtils.listFiles(interpreterLocalRepo, new String[] { "jar" }, false);
Review comment:
Yes, that is correct, is there anything else to be transferred?
Currently, I only have Jar artefacts in the packages section of the interpreters, which are downloaded with Maven.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [zeppelin] zjffdu commented on a change in pull request #4072: [ZEPPELIN-4983] Download local repo to interpreter nodes
Posted by GitBox <gi...@apache.org>.
zjffdu commented on a change in pull request #4072:
URL: https://github.com/apache/zeppelin/pull/4072#discussion_r591059563
##########
File path: bin/interpreter.sh
##########
@@ -267,6 +276,7 @@ elif [[ "${INTERPRETER_ID}" == "flink" ]]; then
fi
+downloadInterpreterLibraries
Review comment:
This is only necessary when interpreter is not launched in the same host as zeppelin server.
And what if the download fail ?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [zeppelin] Reamer commented on a change in pull request #4072: [ZEPPELIN-4983] Download local repo to interpreter nodes
Posted by GitBox <gi...@apache.org>.
Reamer commented on a change in pull request #4072:
URL: https://github.com/apache/zeppelin/pull/4072#discussion_r591270899
##########
File path: bin/interpreter.sh
##########
@@ -267,6 +276,7 @@ elif [[ "${INTERPRETER_ID}" == "flink" ]]; then
fi
+downloadInterpreterLibraries
Review comment:
> This is only necessary when interpreter is not launched in the same host as zeppelin server.
It is quite difficult to decide whether it is necessary to transfer libraries. If Zeppelin interpreters are started on the same machine (Standalone), all libraries should be present and no files should be transferred.
Checksum check before transfer: https://github.com/apache/zeppelin/pull/4072/files#diff-126fba7fb446f22526016f2d293494f0b0d3b5ca4e44a318c93563570e636b71R101-R104
> And what if the download fail ?
If the transfer fails, we have an RPC (Thrift) problem that also leads to other errors.
This transfer should always run and be used in all setups (Yarn, Docker, K8s, Standalone). Unnecessary code can be removed afterwards in the Yarn and Docker Launcher.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [zeppelin] Reamer commented on pull request #4072: [ZEPPELIN-4983] Download local repo to interpreter nodes
Posted by GitBox <gi...@apache.org>.
Reamer commented on pull request #4072:
URL: https://github.com/apache/zeppelin/pull/4072#issuecomment-806743948
The infrastructure tests have been passed. I would like to merge this into Master. I think we can also merge this into branch-0.9 as it doesn't break anything. What is your opinion @zjffdu and @Leemoonsoo ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [zeppelin] cuspymd commented on a change in pull request #4072: [ZEPPELIN-4983] Download local repo to interpreter nodes
Posted by GitBox <gi...@apache.org>.
cuspymd commented on a change in pull request #4072:
URL: https://github.com/apache/zeppelin/pull/4072#discussion_r603921780
##########
File path: zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterDownloader.java
##########
@@ -0,0 +1,148 @@
+/*
+ * 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.
+ */
+
+package org.apache.zeppelin.interpreter.remote;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.file.Files;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+
+import org.apache.commons.io.FileUtils;
+import org.apache.zeppelin.interpreter.Constants;
+import org.apache.zeppelin.interpreter.thrift.LibraryMetadata;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class RemoteInterpreterDownloader {
+ private static final Logger LOGGER = LoggerFactory.getLogger(RemoteInterpreterDownloader.class);
+
+ private static final int MAX_LIBRARY_DOWNLOAD_ATTEMPTS = 3;
+
+ private final RemoteInterpreterEventClient client;
+ private final String interpreter;
+ private final File localRepoDir;
+
+ public RemoteInterpreterDownloader(
+ String interpreter,
+ RemoteInterpreterEventClient client,
+ File localRepoDir) {
+ this.client = client;
+ this.interpreter = interpreter;
+ this.localRepoDir = localRepoDir;
+ }
+
+ public static void main(String[] args) {
+ String zeppelinServerHost = null;
+ int port = Constants.ZEPPELIN_INTERPRETER_DEFAUlT_PORT;
+ String interpreter = null;
+ String localRepoPath = null;
+ if (args.length > 0) {
+ zeppelinServerHost = args[0];
+ port = Integer.parseInt(args[1]);
+ interpreter = args[2];
+ localRepoPath = args[3];
+ }
+ RemoteInterpreterEventClient intpEventClient = new RemoteInterpreterEventClient(
+ zeppelinServerHost, port, 3);
+
+ RemoteInterpreterDownloader downloader = new RemoteInterpreterDownloader(interpreter,
+ intpEventClient, new File(localRepoPath));
+ downloader.syncAllLibraries();
+ }
+
+ private void syncAllLibraries() {
+ LOGGER.info("Loading all libraries for interpreter {} to {}", interpreter, localRepoDir);
+ List<LibraryMetadata> metadatas = client.getAllLibraryMetadatas(interpreter);
+ if (!localRepoDir.isDirectory()) {
+ LOGGER.error("{} is no directory", localRepoDir);
+ return;
+ }
+ Set<String> syncedLibraries = new HashSet<>();
+ // Add or update new libraries
+ for (LibraryMetadata metadata : metadatas) {
+ File library = new File(localRepoDir, metadata.getName());
+ addOrUpdateLibrary(library, metadata);
+ syncedLibraries.add(metadata.getName());
+ }
+ // Delete old Jar files
+ for (File file : FileUtils.listFiles(localRepoDir, new String[] { "jar" }, false)) {
+ if (!syncedLibraries.contains(file.getName())) {
+ try {
+ LOGGER.info("Delete {}, because it's not present on the server side", file.toPath());
+ Files.delete(file.toPath());
+ } catch (IOException e) {
+ LOGGER.error("Unable to delete old library {} during sync.", file, e);
+ }
+ }
+ }
+ }
+
+ private void addOrUpdateLibrary(File library, LibraryMetadata metadata) {
+ try {
+ if (library.exists() && library.canRead()) {
+ long localChecksum = FileUtils.checksumCRC32(library);
+ if (localChecksum == metadata.getChecksum()) {
+ // nothing to do if checksum is matching
+ LOGGER.info("Library {} is present ", library.getName());
+ } else {
+ // checksum didn't match
+ Files.delete(library.toPath());
+ downloadLibrary(library, metadata);
+ }
+ } else {
+ downloadLibrary(library, metadata);
+ }
+ } catch (IOException e) {
+ LOGGER.error("Can not add or update library {}", library, e);
+ }
+ }
+
+ private void downloadLibrary(File library, LibraryMetadata metadata) {
+ LOGGER.debug("Trying to download library {} to {}", metadata.getName(), library);
+ try {
+ if (library.createNewFile()) {
+ int attempt = 0;
+ while (attempt < MAX_LIBRARY_DOWNLOAD_ATTEMPTS) {
+ ByteBuffer bytes = client.getLibrary(interpreter, metadata.getName());
+ if (bytes == null) {
+ LOGGER.error("ByteBuffer of library {} is null."
+ + " For a detailed message take a look into Zeppelin-Server log. Attempt {} of {}",
+ metadata.getName(), ++attempt, MAX_LIBRARY_DOWNLOAD_ATTEMPTS);
Review comment:
It doesn't matter if you try a few more attempts.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [zeppelin] Reamer commented on pull request #4072: [ZEPPELIN-4983] Download local repo to interpreter nodes
Posted by GitBox <gi...@apache.org>.
Reamer commented on pull request #4072:
URL: https://github.com/apache/zeppelin/pull/4072#issuecomment-802608364
> In Kubernetes, Can interpreters of the same type (such as Python) be configured to enable different images?
At the moment, this is not possible. For a discussion, see https://github.com/apache/zeppelin/pull/3769.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [zeppelin] Leemoonsoo commented on pull request #4072: [ZEPPELIN-4983] Download local repo to interpreter nodes
Posted by GitBox <gi...@apache.org>.
Leemoonsoo commented on pull request #4072:
URL: https://github.com/apache/zeppelin/pull/4072#issuecomment-796954675
> How can this PR be tested? I am open to ideas. At the moment I have done some successful manual tests.
It could be out of the scope of this PR but, do you think it makes sense to have an end-to-end test for Zeppelin on Kubernetes?
Looks like there're ways to run Kubernetes cluster in the Github Action environment (e.g. https://github.com/marketplace/actions/kind-kubernetes-in-docker-action) and It may help reduce things that are not covered by the automated tests, such as,
- Dependency library sync when Zeppelin server and interpreter runs on a different host (this PR)
- Interpreter lifecycle on Kubernetes (creating/destroying pod, service, etc)
- Container image build
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [zeppelin] Teo920127 commented on pull request #4072: [ZEPPELIN-4983] Download local repo to interpreter nodes
Posted by GitBox <gi...@apache.org>.
Teo920127 commented on pull request #4072:
URL: https://github.com/apache/zeppelin/pull/4072#issuecomment-802489730
In Kubernetes, would it be possible to use customized interpreter image for particular interpreters ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [zeppelin] Reamer commented on a change in pull request #4072: [ZEPPELIN-4983] Download local repo to interpreter nodes
Posted by GitBox <gi...@apache.org>.
Reamer commented on a change in pull request #4072:
URL: https://github.com/apache/zeppelin/pull/4072#discussion_r602096654
##########
File path: zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterDownloader.java
##########
@@ -0,0 +1,140 @@
+/*
+ * 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.
+ */
+
+package org.apache.zeppelin.interpreter.remote;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.file.Files;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+
+import org.apache.commons.io.FileUtils;
+import org.apache.zeppelin.interpreter.Constants;
+import org.apache.zeppelin.interpreter.thrift.LibraryMetadata;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class RemoteInterpreterDownloader {
+ private static final Logger LOGGER = LoggerFactory.getLogger(RemoteInterpreterDownloader.class);
+
+ private final RemoteInterpreterEventClient client;
+ private final String interpreter;
+ private final File localRepoDir;
+
+ public RemoteInterpreterDownloader(
+ String interpreter,
+ RemoteInterpreterEventClient client,
+ File localRepoDir) {
+ this.client = client;
+ this.interpreter = interpreter;
+ this.localRepoDir = localRepoDir;
+ }
+
+ public static void main(String[] args) {
+ String zeppelinServerHost = null;
+ int port = Constants.ZEPPELIN_INTERPRETER_DEFAUlT_PORT;
+ String interpreter = null;
+ String localRepoPath = null;
+ if (args.length > 0) {
+ zeppelinServerHost = args[0];
+ port = Integer.parseInt(args[1]);
+ interpreter = args[2];
+ localRepoPath = args[3];
+ }
+ RemoteInterpreterEventClient intpEventClient = new RemoteInterpreterEventClient(
+ zeppelinServerHost, port, 3);
+
+ RemoteInterpreterDownloader downloader = new RemoteInterpreterDownloader(interpreter,
+ intpEventClient, new File(localRepoPath));
+ downloader.syncAllLibraries();
+ }
+
+ private void syncAllLibraries() {
+ LOGGER.info("Loading all libraries for interpreter {} to {}", interpreter, localRepoDir);
+ List<LibraryMetadata> metadatas = client.getAllLibraryMetadatas(interpreter);
+ if (!localRepoDir.isDirectory()) {
+ LOGGER.error("{} is no directory", localRepoDir);
+ return;
+ }
+ Set<String> syncedLibraries = new HashSet<>();
+ // Add or update new libraries
+ for (LibraryMetadata metadata : metadatas) {
+ File library = new File(localRepoDir, metadata.getName());
+ addOrUpdateLibrary(library, metadata);
+ syncedLibraries.add(metadata.getName());
+ }
+ // Delete old Jar files
+ for (File file : FileUtils.listFiles(localRepoDir, new String[] { "jar" }, false)) {
+ if (!syncedLibraries.contains(file.getName())) {
+ try {
+ LOGGER.info("Delete {}, because it's not present on the server side", file.toPath());
+ Files.delete(file.toPath());
+ } catch (IOException e) {
+ LOGGER.error("Unable to delete old library {} during sync.", file, e);
+ }
+ }
+ }
+ }
+
+ private void addOrUpdateLibrary(File library, LibraryMetadata metadata) {
+ try {
+ if (library.exists() && library.canRead()) {
+ long localChecksum = FileUtils.checksumCRC32(library);
+ if (localChecksum == metadata.getChecksum()) {
+ // nothing to do if checksum is matching
+ LOGGER.info("Library {} is present ", library.getName());
+ } else {
+ // checksum didn't match
+ Files.delete(library.toPath());
+ downloadLibrary(library, metadata);
+ }
+ } else {
+ downloadLibrary(library, metadata);
+ }
+ } catch (IOException e) {
+ LOGGER.error("Can not add or update library {}", library, e);
+ }
+ }
+
+ private void downloadLibrary(File library, LibraryMetadata metadata) {
+ LOGGER.debug("Trying to download library {} to {}", metadata.getName(), library);
+ try {
+ if (library.createNewFile()) {
+ int tries = 0;
+ while (tries < 3) {
+ ByteBuffer bytes = client.getLibrary(interpreter, metadata.getName());
+ FileUtils.writeByteArrayToFile(library, bytes.array());
Review comment:
You are right, I will correct this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org