You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@zeppelin.apache.org by pr...@apache.org on 2018/03/02 03:13:18 UTC
zeppelin git commit: [ZEPPELIN-3271] Option for disabling scheduler
Repository: zeppelin
Updated Branches:
refs/heads/master 65b797c22 -> 2c322d72b
[ZEPPELIN-3271] Option for disabling scheduler
### What is this PR for?
Zeppelin server should have an option to enable/disable cron scheduler from Zeppelin config.
### What type of PR is it?
[Bug Fix]
### What is the Jira issue?
* [ZEPPELIN-3271](https://issues.apache.org/jira/browse/ZEPPELIN-3271)
### How should this be tested?
* On adding below mentioned property in `zeppelin-site.xml`, this feature should get disabled from both web-socket and REST-APIs.
```
<property>
<name>zeppelin.notebook.cron.enable</name>
<value>false</value>
<description>Notebook enable cron scheduler feature</description>
</property>
```
### Questions:
* Does the licenses files need update? N/A
* Is there breaking changes for older versions? Yes
* Does this needs documentation? Yes
Author: Prabhjyot Singh <pr...@gmail.com>
Closes #2821 from prabhjyotsingh/ZEPPELIN-3271 and squashes the following commits:
93e9dc6 [Prabhjyot Singh] cron disable by default
359e687 [Prabhjyot Singh] add test-case for Cron disabled
eaec944 [Prabhjyot Singh] refactor isCronSupported
6421439 [Prabhjyot Singh] setCronSupported property for import, clone and create new
48488cf [Prabhjyot Singh] option for 'zeppelin.notebook.cron.folders'
9491d3d [Prabhjyot Singh] log error before throwing ForbiddenException
d25ebcc [Prabhjyot Singh] ZEPPELIN-3271: Option for disabling scheduler
Project: http://git-wip-us.apache.org/repos/asf/zeppelin/repo
Commit: http://git-wip-us.apache.org/repos/asf/zeppelin/commit/2c322d72
Tree: http://git-wip-us.apache.org/repos/asf/zeppelin/tree/2c322d72
Diff: http://git-wip-us.apache.org/repos/asf/zeppelin/diff/2c322d72
Branch: refs/heads/master
Commit: 2c322d72b662e703710a30abc4460f9aee2a29bb
Parents: 65b797c
Author: Prabhjyot Singh <pr...@gmail.com>
Authored: Thu Mar 1 15:28:08 2018 +0530
Committer: Prabhjyot Singh <pr...@gmail.com>
Committed: Fri Mar 2 08:41:55 2018 +0530
----------------------------------------------------------------------
conf/zeppelin-site.xml.template | 11 ++
docs/usage/other_features/cron_scheduler.md | 8 ++
.../zeppelin/conf/ZeppelinConfiguration.java | 12 +-
.../apache/zeppelin/rest/NotebookRestApi.java | 38 ++++--
.../apache/zeppelin/socket/NotebookServer.java | 9 +-
.../zeppelin/rest/ZeppelinRestApiTest.java | 63 ++++++++--
.../src/app/notebook/notebook-actionBar.html | 2 +-
.../java/org/apache/zeppelin/notebook/Note.java | 33 +++--
.../org/apache/zeppelin/notebook/Notebook.java | 53 ++++++--
.../apache/zeppelin/notebook/NotebookTest.java | 121 ++++++++++++++++---
10 files changed, 285 insertions(+), 65 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/zeppelin/blob/2c322d72/conf/zeppelin-site.xml.template
----------------------------------------------------------------------
diff --git a/conf/zeppelin-site.xml.template b/conf/zeppelin-site.xml.template
index 9774f0d..d39b19c 100755
--- a/conf/zeppelin-site.xml.template
+++ b/conf/zeppelin-site.xml.template
@@ -540,5 +540,16 @@
<value>origin</value>
<description>Git repository remote</description>
</property>
+
+<property>
+ <name>zeppelin.notebook.cron.enable</name>
+ <value>false</value>
+ <description>Notebook enable cron scheduler feature</description>
+</property>
+<property>
+ <name>zeppelin.notebook.cron.folders</name>
+ <value></value>
+ <description>Notebook cron folders</description>
+</property>
-->
</configuration>
http://git-wip-us.apache.org/repos/asf/zeppelin/blob/2c322d72/docs/usage/other_features/cron_scheduler.md
----------------------------------------------------------------------
diff --git a/docs/usage/other_features/cron_scheduler.md b/docs/usage/other_features/cron_scheduler.md
index e8a9975..c7fc284 100644
--- a/docs/usage/other_features/cron_scheduler.md
+++ b/docs/usage/other_features/cron_scheduler.md
@@ -50,3 +50,11 @@ You can set the cron executing user by filling in this form and press the enter
When this checkbox is set to "on", the interpreters which are binded to the notebook are stopped automatically after the cron execution. This feature is useful if you want to release the interpreter resources after the cron execution.
> **Note**: A cron execution is skipped if one of the paragraphs is in a state of `RUNNING` or `PENDING` no matter whether it is executed automatically (i.e. by the cron scheduler) or manually by a user opening this notebook.
+
+### Enable cron
+
+Set property **zeppelin.notebook.cron.enable** to **true** in `$ZEPPELIN_HOME/conf/zeppelin-site.xml` to enable Cron feature.
+
+### Run cron selectively on folders
+
+In `$ZEPPELIN_HOME/conf/zeppelin-site.xml` make sure the property **zeppelin.notebook.cron.enable** is set to **true**, and then set property **zeppelin.notebook.cron.folders** to the desired folder as comma-separated values, e.g. `*yst*, Sys?em, System`. This property accepts wildcard and joker.
http://git-wip-us.apache.org/repos/asf/zeppelin/blob/2c322d72/zeppelin-interpreter/src/main/java/org/apache/zeppelin/conf/ZeppelinConfiguration.java
----------------------------------------------------------------------
diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/conf/ZeppelinConfiguration.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/conf/ZeppelinConfiguration.java
index 5beb2c7..e0ebfa2 100644
--- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/conf/ZeppelinConfiguration.java
+++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/conf/ZeppelinConfiguration.java
@@ -581,6 +581,14 @@ public class ZeppelinConfiguration extends XMLConfiguration {
return getString(ConfVars.ZEPPELIN_NOTEBOOK_GIT_REMOTE_ORIGIN);
}
+ public Boolean isZeppelinNotebookCronEnable() {
+ return getBoolean(ConfVars.ZEPPELIN_NOTEBOOK_CRON_ENABLE);
+ }
+
+ public String getZeppelinNotebookCronFolders() {
+ return getString(ConfVars.ZEPPELIN_NOTEBOOK_CRON_FOLDERS);
+ }
+
public Map<String, String> dumpConfigurations(ZeppelinConfiguration conf,
ConfigurationKeyPredicate predicate) {
Map<String, String> configurations = new HashMap<>();
@@ -771,7 +779,9 @@ public class ZeppelinConfiguration extends XMLConfiguration {
ZEPPELIN_NOTEBOOK_GIT_REMOTE_URL("zeppelin.notebook.git.remote.url", ""),
ZEPPELIN_NOTEBOOK_GIT_REMOTE_USERNAME("zeppelin.notebook.git.remote.username", "token"),
ZEPPELIN_NOTEBOOK_GIT_REMOTE_ACCESS_TOKEN("zeppelin.notebook.git.remote.access-token", ""),
- ZEPPELIN_NOTEBOOK_GIT_REMOTE_ORIGIN("zeppelin.notebook.git.remote.origin", "origin");
+ ZEPPELIN_NOTEBOOK_GIT_REMOTE_ORIGIN("zeppelin.notebook.git.remote.origin", "origin"),
+ ZEPPELIN_NOTEBOOK_CRON_ENABLE("zeppelin.notebook.cron.enable", false),
+ ZEPPELIN_NOTEBOOK_CRON_FOLDERS("zeppelin.notebook.cron.folders", null);
private String varName;
@SuppressWarnings("rawtypes")
http://git-wip-us.apache.org/repos/asf/zeppelin/blob/2c322d72/zeppelin-server/src/main/java/org/apache/zeppelin/rest/NotebookRestApi.java
----------------------------------------------------------------------
diff --git a/zeppelin-server/src/main/java/org/apache/zeppelin/rest/NotebookRestApi.java b/zeppelin-server/src/main/java/org/apache/zeppelin/rest/NotebookRestApi.java
index 2042c4c..8bfaef5 100644
--- a/zeppelin-server/src/main/java/org/apache/zeppelin/rest/NotebookRestApi.java
+++ b/zeppelin-server/src/main/java/org/apache/zeppelin/rest/NotebookRestApi.java
@@ -33,6 +33,7 @@ import javax.ws.rs.core.Response.Status;
import org.apache.commons.lang3.StringUtils;
import org.apache.zeppelin.annotation.ZeppelinApi;
+import org.apache.zeppelin.conf.ZeppelinConfiguration;
import org.apache.zeppelin.interpreter.InterpreterResult;
import org.apache.zeppelin.notebook.Note;
import org.apache.zeppelin.notebook.Notebook;
@@ -139,7 +140,7 @@ public class NotebookRestApi {
throw new ForbiddenException(errorMsg);
}
}
-
+
/**
* Check if the current user is either Owner or Writer for the given note.
*/
@@ -151,7 +152,7 @@ public class NotebookRestApi {
throw new ForbiddenException(errorMsg);
}
}
-
+
/**
* Check if the current user can access (at least he have to be reader) the given note.
*/
@@ -175,19 +176,26 @@ public class NotebookRestApi {
throw new ForbiddenException(errorMsg);
}
}
-
+
private void checkIfNoteIsNotNull(Note note) {
if (note == null) {
throw new NotFoundException("note not found");
}
}
-
+
+ private void checkIfNoteSupportsCron(Note note) {
+ if (!note.isCronSupported(notebook.getConf())) {
+ LOG.error("Cron is not enabled from Zeppelin server");
+ throw new ForbiddenException("Cron is not enabled from Zeppelin server");
+ }
+ }
+
private void checkIfParagraphIsNotNull(Paragraph paragraph) {
if (paragraph == null) {
throw new NotFoundException("paragraph not found");
}
}
-
+
/**
* set note authorization information
*/
@@ -205,7 +213,7 @@ public class NotebookRestApi {
checkIfUserIsAnon(getBlockNotAuthenticatedUserErrorMsg());
checkIfUserIsOwner(noteId,
ownerPermissionError(userAndRoles, notebookAuthorization.getOwners(noteId)));
-
+
HashMap<String, HashSet<String>> permMap =
gson.fromJson(req, new TypeToken<HashMap<String, HashSet<String>>>() {}.getType());
Note note = notebook.getNote(noteId);
@@ -380,6 +388,7 @@ public class NotebookRestApi {
note.setName(noteName);
note.persist(subject);
+ note.setCronSupported(notebook.getConf());
notebookServer.broadcastNote(note);
notebookServer.broadcastNoteList(subject, SecurityUtils.getRoles());
return new JsonResponse<>(Status.OK, "", note.getId()).build();
@@ -715,7 +724,7 @@ public class NotebookRestApi {
@GET
@Path("job/{noteId}/{paragraphId}")
@ZeppelinApi
- public Response getNoteParagraphJobStatus(@PathParam("noteId") String noteId,
+ public Response getNoteParagraphJobStatus(@PathParam("noteId") String noteId,
@PathParam("paragraphId") String paragraphId)
throws IOException, IllegalArgumentException {
LOG.info("get note paragraph job status.");
@@ -853,6 +862,7 @@ public class NotebookRestApi {
Note note = notebook.getNote(noteId);
checkIfNoteIsNotNull(note);
checkIfUserCanRun(noteId, "Insufficient privileges you cannot set a cron job for this note");
+ checkIfNoteSupportsCron(note);
if (!CronExpression.isValidExpression(request.getCronString())) {
return new JsonResponse<>(Status.BAD_REQUEST, "wrong cron expressions.").build();
@@ -879,11 +889,12 @@ public class NotebookRestApi {
public Response removeCronJob(@PathParam("noteId") String noteId)
throws IOException, IllegalArgumentException {
LOG.info("Remove cron job note {}", noteId);
-
+
Note note = notebook.getNote(noteId);
checkIfNoteIsNotNull(note);
checkIfUserIsOwner(noteId,
"Insufficient privileges you cannot remove this cron job from this note");
+ checkIfNoteSupportsCron(note);
Map<String, Object> config = note.getConfig();
config.put("cron", null);
@@ -910,6 +921,7 @@ public class NotebookRestApi {
Note note = notebook.getNote(noteId);
checkIfNoteIsNotNull(note);
checkIfUserCanRead(noteId, "Insufficient privileges you cannot get cron information");
+ checkIfNoteSupportsCron(note);
return new JsonResponse<>(Status.OK, note.getConfig().get("cron")).build();
}
@@ -1015,22 +1027,22 @@ public class NotebookRestApi {
checkIfParagraphIsNotNull(p);
p.setTitle(request.getTitle());
p.setText(request.getText());
- Map< String, Object > config = request.getConfig();
- if ( config != null && !config.isEmpty()) {
+ Map<String, Object> config = request.getConfig();
+ if (config != null && !config.isEmpty()) {
configureParagraph(p, config, user);
}
}
- private void configureParagraph(Paragraph p, Map< String, Object> newConfig, String user)
+ private void configureParagraph(Paragraph p, Map<String, Object> newConfig, String user)
throws IOException {
LOG.info("Configure Paragraph for user {}", user);
if (newConfig == null || newConfig.isEmpty()) {
LOG.warn("{} is trying to update paragraph {} of note {} with empty config",
- user, p.getId(), p.getNote().getId());
+ user, p.getId(), p.getNote().getId());
throw new BadRequestException("paragraph config cannot be empty");
}
Map<String, Object> origConfig = p.getConfig();
- for ( final Map.Entry<String, Object> entry : newConfig.entrySet()){
+ for (final Map.Entry<String, Object> entry : newConfig.entrySet()) {
origConfig.put(entry.getKey(), entry.getValue());
}
http://git-wip-us.apache.org/repos/asf/zeppelin/blob/2c322d72/zeppelin-server/src/main/java/org/apache/zeppelin/socket/NotebookServer.java
----------------------------------------------------------------------
diff --git a/zeppelin-server/src/main/java/org/apache/zeppelin/socket/NotebookServer.java b/zeppelin-server/src/main/java/org/apache/zeppelin/socket/NotebookServer.java
index d1cf9e5..113dfd6 100644
--- a/zeppelin-server/src/main/java/org/apache/zeppelin/socket/NotebookServer.java
+++ b/zeppelin-server/src/main/java/org/apache/zeppelin/socket/NotebookServer.java
@@ -869,7 +869,7 @@ public class NotebookServer extends WebSocketServlet
}
private void updateNote(NotebookSocket conn, HashSet<String> userAndRoles, Notebook notebook,
- Message fromMessage) throws SchedulerException, IOException {
+ Message fromMessage) throws IOException {
String noteId = (String) fromMessage.get("id");
String name = (String) fromMessage.get("name");
Map<String, Object> config = (Map<String, Object>) fromMessage.get("config");
@@ -887,6 +887,11 @@ public class NotebookServer extends WebSocketServlet
Note note = notebook.getNote(noteId);
if (note != null) {
+ if (!(Boolean) note.getConfig().get("isZeppelinNotebookCronEnable")) {
+ if (config.get("cron") != null) {
+ config.remove("cron");
+ }
+ }
boolean cronUpdated = isCronUpdated(config, note.getConfig());
note.setName(name);
note.setConfig(config);
@@ -950,6 +955,7 @@ public class NotebookServer extends WebSocketServlet
Note note = notebook.getNote(noteId);
if (note != null) {
note.setName(name);
+ note.setCronSupported(notebook.getConf());
AuthenticationInfo subject = new AuthenticationInfo(fromMessage.principal);
note.persist(subject);
@@ -1040,6 +1046,7 @@ public class NotebookServer extends WebSocketServlet
noteName = "Note " + note.getId();
}
note.setName(noteName);
+ note.setCronSupported(notebook.getConf());
}
note.persist(subject);
http://git-wip-us.apache.org/repos/asf/zeppelin/blob/2c322d72/zeppelin-server/src/test/java/org/apache/zeppelin/rest/ZeppelinRestApiTest.java
----------------------------------------------------------------------
diff --git a/zeppelin-server/src/test/java/org/apache/zeppelin/rest/ZeppelinRestApiTest.java b/zeppelin-server/src/test/java/org/apache/zeppelin/rest/ZeppelinRestApiTest.java
index 7e1a28a..da68087 100644
--- a/zeppelin-server/src/test/java/org/apache/zeppelin/rest/ZeppelinRestApiTest.java
+++ b/zeppelin-server/src/test/java/org/apache/zeppelin/rest/ZeppelinRestApiTest.java
@@ -29,6 +29,7 @@ import org.apache.commons.httpclient.methods.GetMethod;
import org.apache.commons.httpclient.methods.PostMethod;
import org.apache.commons.httpclient.methods.PutMethod;
import org.apache.commons.lang3.StringUtils;
+import org.apache.zeppelin.conf.ZeppelinConfiguration.ConfVars;
import org.apache.zeppelin.notebook.Note;
import org.apache.zeppelin.notebook.Paragraph;
import org.apache.zeppelin.server.ZeppelinServer;
@@ -375,11 +376,11 @@ public class ZeppelinRestApiTest extends AbstractTestRestApi {
assertNotNull("can't create new note", note);
note.setName("note for run test");
Paragraph paragraph = note.addNewParagraph(AuthenticationInfo.ANONYMOUS);
-
+
Map config = paragraph.getConfig();
config.put("enabled", true);
paragraph.setConfig(config);
-
+
paragraph.setText("%md This is test paragraph.");
note.persist(anonymous);
String noteId = note.getId();
@@ -394,7 +395,7 @@ public class ZeppelinRestApiTest extends AbstractTestRestApi {
break;
}
}
-
+
// Call Run note jobs REST API
PostMethod postNoteJobs = httpPost("/notebook/job/" + noteId, "");
assertThat("test note jobs run:", postNoteJobs, isAllowed());
@@ -403,21 +404,21 @@ public class ZeppelinRestApiTest extends AbstractTestRestApi {
// Call Stop note jobs REST API
DeleteMethod deleteNoteJobs = httpDelete("/notebook/job/" + noteId);
assertThat("test note stop:", deleteNoteJobs, isAllowed());
- deleteNoteJobs.releaseConnection();
+ deleteNoteJobs.releaseConnection();
Thread.sleep(1000);
-
+
// Call Run paragraph REST API
PostMethod postParagraph = httpPost("/notebook/job/" + noteId + "/" + paragraph.getId(), "");
assertThat("test paragraph run:", postParagraph, isAllowed());
- postParagraph.releaseConnection();
+ postParagraph.releaseConnection();
Thread.sleep(1000);
-
+
// Call Stop paragraph REST API
DeleteMethod deleteParagraph = httpDelete("/notebook/job/" + noteId + "/" + paragraph.getId());
assertThat("test paragraph stop:", deleteParagraph, isAllowed());
- deleteParagraph.releaseConnection();
+ deleteParagraph.releaseConnection();
Thread.sleep(1000);
-
+
//cleanup
ZeppelinServer.notebook.removeNote(note.getId(), anonymous);
}
@@ -514,7 +515,7 @@ public class ZeppelinRestApiTest extends AbstractTestRestApi {
note.setName("note for run test");
Paragraph paragraph = note.addNewParagraph(AuthenticationInfo.ANONYMOUS);
paragraph.setText("%md This is test paragraph.");
-
+
Map config = paragraph.getConfig();
config.put("enabled", true);
paragraph.setConfig(config);
@@ -526,20 +527,20 @@ public class ZeppelinRestApiTest extends AbstractTestRestApi {
PostMethod postCron = httpPost("/notebook/cron/notexistnote", jsonRequest);
assertThat("", postCron, isNotFound());
postCron.releaseConnection();
-
+
// right cron expression.
postCron = httpPost("/notebook/cron/" + note.getId(), jsonRequest);
assertThat("", postCron, isAllowed());
postCron.releaseConnection();
Thread.sleep(1000);
-
+
// wrong cron expression.
jsonRequest = "{\"cron\":\"a * * * * ?\" }";
postCron = httpPost("/notebook/cron/" + note.getId(), jsonRequest);
assertThat("", postCron, isBadRequest());
postCron.releaseConnection();
Thread.sleep(1000);
-
+
// remove cron job.
DeleteMethod deleteCron = httpDelete("/notebook/cron/" + note.getId());
assertThat("", deleteCron, isAllowed());
@@ -548,6 +549,42 @@ public class ZeppelinRestApiTest extends AbstractTestRestApi {
}
@Test
+ public void testCronDisable() throws InterruptedException, IOException{
+ // create a note and a paragraph
+ System.setProperty(ConfVars.ZEPPELIN_NOTEBOOK_CRON_ENABLE.getVarName(), "false");
+ Note note = ZeppelinServer.notebook.createNote(anonymous);
+
+ note.setName("note for run test");
+ Paragraph paragraph = note.addNewParagraph(AuthenticationInfo.ANONYMOUS);
+ paragraph.setText("%md This is test paragraph.");
+
+ Map config = paragraph.getConfig();
+ config.put("enabled", true);
+ paragraph.setConfig(config);
+
+ note.runAll(AuthenticationInfo.ANONYMOUS, false);
+
+ String jsonRequest = "{\"cron\":\"* * * * * ?\" }";
+ // right cron expression.
+ PostMethod postCron = httpPost("/notebook/cron/" + note.getId(), jsonRequest);
+ assertThat("", postCron, isForbidden());
+ postCron.releaseConnection();
+
+ System.setProperty(ConfVars.ZEPPELIN_NOTEBOOK_CRON_ENABLE.getVarName(), "true");
+ System.setProperty(ConfVars.ZEPPELIN_NOTEBOOK_CRON_FOLDERS.getVarName(), "System/*");
+
+ note.setName("System/test2");
+ note.runAll(AuthenticationInfo.ANONYMOUS, false);
+ postCron = httpPost("/notebook/cron/" + note.getId(), jsonRequest);
+ assertThat("", postCron, isAllowed());
+ postCron.releaseConnection();
+ Thread.sleep(1000);
+
+ System.clearProperty(ConfVars.ZEPPELIN_NOTEBOOK_CRON_FOLDERS.getVarName());
+ ZeppelinServer.notebook.removeNote(note.getId(), anonymous);
+ }
+
+ @Test
public void testRegressionZEPPELIN_527() throws IOException {
Note note = ZeppelinServer.notebook.createNote(anonymous);
http://git-wip-us.apache.org/repos/asf/zeppelin/blob/2c322d72/zeppelin-web/src/app/notebook/notebook-actionBar.html
----------------------------------------------------------------------
diff --git a/zeppelin-web/src/app/notebook/notebook-actionBar.html b/zeppelin-web/src/app/notebook/notebook-actionBar.html
index f8ff830..b4add93 100644
--- a/zeppelin-web/src/app/notebook/notebook-actionBar.html
+++ b/zeppelin-web/src/app/notebook/notebook-actionBar.html
@@ -249,7 +249,7 @@ limitations under the License.
</span>
<span ng-hide="viewOnly">
- <div class="labelBtn btn-group">
+ <div class="labelBtn btn-group" ng-if="note.config.isZeppelinNotebookCronEnable">
<div class="btn btn-default btn-xs dropdown-toggle"
type="button"
data-toggle="dropdown"
http://git-wip-us.apache.org/repos/asf/zeppelin/blob/2c322d72/zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/Note.java
----------------------------------------------------------------------
diff --git a/zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/Note.java b/zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/Note.java
index 47aa858..495f670 100644
--- a/zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/Note.java
+++ b/zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/Note.java
@@ -19,6 +19,10 @@ package org.apache.zeppelin.notebook;
import static java.lang.String.format;
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+import com.google.gson.Gson;
+import com.google.gson.GsonBuilder;
import java.io.IOException;
import java.util.Date;
import java.util.HashMap;
@@ -30,7 +34,6 @@ import java.util.Map;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
-
import org.apache.commons.lang.StringUtils;
import org.apache.zeppelin.common.JsonSerializable;
import org.apache.zeppelin.completer.CompletionType;
@@ -47,9 +50,9 @@ import org.apache.zeppelin.interpreter.InterpreterSetting;
import org.apache.zeppelin.interpreter.InterpreterSettingManager;
import org.apache.zeppelin.interpreter.remote.RemoteAngularObjectRegistry;
import org.apache.zeppelin.interpreter.thrift.InterpreterCompletion;
+import org.apache.zeppelin.notebook.repo.NotebookRepo;
import org.apache.zeppelin.notebook.repo.NotebookRepoSync;
import org.apache.zeppelin.notebook.repo.NotebookRepoWithVersionControl;
-import org.apache.zeppelin.notebook.repo.NotebookRepo;
import org.apache.zeppelin.notebook.utility.IdHashes;
import org.apache.zeppelin.scheduler.Job;
import org.apache.zeppelin.scheduler.Job.Status;
@@ -59,11 +62,6 @@ import org.apache.zeppelin.user.Credentials;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import com.google.common.annotations.VisibleForTesting;
-import com.google.common.base.Preconditions;
-import com.google.gson.Gson;
-import com.google.gson.GsonBuilder;
-
/**
* Binded interpreters for a note
*/
@@ -314,6 +312,27 @@ public class Note implements ParagraphJobListener, JsonSerializable {
}
}
+ public Boolean isCronSupported(ZeppelinConfiguration config) {
+ if (config.isZeppelinNotebookCronEnable()) {
+ config.getZeppelinNotebookCronFolders();
+ if (config.getZeppelinNotebookCronFolders() == null) {
+ return true;
+ } else {
+ for (String folder : config.getZeppelinNotebookCronFolders().split(",")) {
+ folder = folder.replaceAll("\\*", "\\.*").replaceAll("\\?", "\\.");
+ if (getName().matches(folder)) {
+ return true;
+ }
+ }
+ }
+ }
+ return false;
+ }
+
+ public void setCronSupported(ZeppelinConfiguration config) {
+ getConfig().put("isZeppelinNotebookCronEnable", isCronSupported(config));
+ }
+
public void setIndex(SearchService index) {
this.index = index;
}
http://git-wip-us.apache.org/repos/asf/zeppelin/blob/2c322d72/zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/Notebook.java
----------------------------------------------------------------------
diff --git a/zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/Notebook.java b/zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/Notebook.java
index 633072c..a9168a3 100644
--- a/zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/Notebook.java
+++ b/zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/Notebook.java
@@ -21,30 +21,51 @@ import com.google.common.base.Preconditions;
import com.google.common.base.Predicate;
import com.google.common.collect.FluentIterable;
import com.google.common.collect.Sets;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
import org.apache.zeppelin.conf.ZeppelinConfiguration;
import org.apache.zeppelin.conf.ZeppelinConfiguration.ConfVars;
import org.apache.zeppelin.display.AngularObject;
import org.apache.zeppelin.display.AngularObjectRegistry;
-import org.apache.zeppelin.interpreter.*;
+import org.apache.zeppelin.interpreter.InterpreterException;
+import org.apache.zeppelin.interpreter.InterpreterFactory;
+import org.apache.zeppelin.interpreter.InterpreterGroup;
+import org.apache.zeppelin.interpreter.InterpreterResult;
+import org.apache.zeppelin.interpreter.InterpreterSetting;
+import org.apache.zeppelin.interpreter.InterpreterSettingManager;
import org.apache.zeppelin.interpreter.remote.RemoteAngularObjectRegistry;
import org.apache.zeppelin.notebook.repo.NotebookRepo;
+import org.apache.zeppelin.notebook.repo.NotebookRepoSync;
import org.apache.zeppelin.notebook.repo.NotebookRepoWithVersionControl;
import org.apache.zeppelin.notebook.repo.NotebookRepoWithVersionControl.Revision;
-import org.apache.zeppelin.notebook.repo.NotebookRepoSync;
import org.apache.zeppelin.scheduler.Job;
import org.apache.zeppelin.scheduler.SchedulerFactory;
import org.apache.zeppelin.search.SearchService;
import org.apache.zeppelin.user.AuthenticationInfo;
import org.apache.zeppelin.user.Credentials;
-import org.quartz.*;
+import org.quartz.CronScheduleBuilder;
+import org.quartz.CronTrigger;
+import org.quartz.JobBuilder;
+import org.quartz.JobDetail;
+import org.quartz.JobExecutionContext;
+import org.quartz.JobExecutionException;
+import org.quartz.JobKey;
+import org.quartz.SchedulerException;
+import org.quartz.TriggerBuilder;
import org.quartz.impl.StdSchedulerFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.io.IOException;
-import java.util.*;
-import java.util.concurrent.TimeUnit;
-
/**
* Collection of Notes.
*/
@@ -182,10 +203,12 @@ public class Notebook implements NoteEventListener {
Note oldNote = Note.fromJson(sourceJson);
convertFromSingleResultToMultipleResultsFormat(oldNote);
newNote = createNote(subject);
- if (noteName != null)
+ if (noteName != null) {
newNote.setName(noteName);
- else
+ } else {
newNote.setName(oldNote.getName());
+ }
+ newNote.setCronSupported(getConf());
List<Paragraph> paragraphs = oldNote.getParagraphs();
for (Paragraph p : paragraphs) {
newNote.addCloneParagraph(p);
@@ -222,6 +245,7 @@ public class Notebook implements NoteEventListener {
} else {
newNote.setName("Note " + newNote.getId());
}
+ newNote.setCronSupported(getConf());
// Copy the interpreter bindings
List<String> boundInterpreterSettingsIds = getBindedInterpreterSettingsIds(sourceNote.getId());
bindInterpretersToNote(subject.getUser(), newNote.getId(), boundInterpreterSettingsIds);
@@ -498,6 +522,7 @@ public class Notebook implements NoteEventListener {
note.setJobListenerFactory(jobListenerFactory);
note.setNotebookRepo(notebookRepo);
note.setRevisionSupported(notebookRepo);
+ note.setCronSupported(getConf());
Map<String, SnapshotAngularObject> angularObjectSnapshot = new HashMap<>();
@@ -899,6 +924,11 @@ public class Notebook implements NoteEventListener {
return;
}
+ if (!note.isCronSupported(notebook.getConf())) {
+ logger.warn("execution of the cron job is skipped cron is not enabled from Zeppelin server");
+ return;
+ }
+
note.runAll();
boolean releaseResource = false;
@@ -941,6 +971,11 @@ public class Notebook implements NoteEventListener {
return;
}
+ if (!note.isCronSupported(getConf())) {
+ logger.warn("execution of the cron job is skipped cron is not enabled from Zeppelin server");
+ return;
+ }
+
String cronExpr = (String) note.getConfig().get("cron");
if (cronExpr == null || cronExpr.trim().length() == 0) {
return;
http://git-wip-us.apache.org/repos/asf/zeppelin/blob/2c322d72/zeppelin-zengine/src/test/java/org/apache/zeppelin/notebook/NotebookTest.java
----------------------------------------------------------------------
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 83c0932..02490ea 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
@@ -86,6 +86,7 @@ public class NotebookTest extends AbstractInterpreterTest implements JobListener
public void setUp() throws Exception {
System.setProperty(ConfVars.ZEPPELIN_NOTEBOOK_PUBLIC.getVarName(), "true");
System.setProperty(ConfVars.ZEPPELIN_INTERPRETER_GROUP_ORDER.getVarName(), "mock1,mock2");
+ System.setProperty(ConfVars.ZEPPELIN_NOTEBOOK_CRON_ENABLE.getVarName(), "true");
super.setUp();
schedulerFactory = SchedulerFactory.singleton();
@@ -226,7 +227,7 @@ public class NotebookTest extends AbstractInterpreterTest implements JobListener
fail("Subject is non-emtpy anonymous, shouldn't fail");
}
}
-
+
@Test
public void testPersist() throws IOException, SchedulerException, RepositoryException {
Note note = notebook.createNote(anonymous);
@@ -434,13 +435,93 @@ public class NotebookTest extends AbstractInterpreterTest implements JobListener
notebook.refreshCron(note.getId());
}
+ @Test
+ public void testScheduleDisabled() throws InterruptedException, IOException {
+
+ System.setProperty(ConfVars.ZEPPELIN_NOTEBOOK_CRON_ENABLE.getVarName(), "false");
+ try {
+ final int timeout = 10;
+ final String everySecondCron = "* * * * * ?";
+ final CountDownLatch jobsToExecuteCount = new CountDownLatch(5);
+ final Note note = notebook.createNote(anonymous);
+
+ executeNewParagraphByCron(note, everySecondCron);
+ afterStatusChangedListener = new StatusChangedListener() {
+ @Override
+ public void onStatusChanged(Job job, Status before, Status after) {
+ if (after == Status.FINISHED) {
+ jobsToExecuteCount.countDown();
+ }
+ }
+ };
+
+ //This job should not run because "ZEPPELIN_NOTEBOOK_CRON_ENABLE" is set to false
+ assertFalse(jobsToExecuteCount.await(timeout, TimeUnit.SECONDS));
+
+ terminateScheduledNote(note);
+ afterStatusChangedListener = null;
+ } finally {
+ System.setProperty(ConfVars.ZEPPELIN_NOTEBOOK_CRON_ENABLE.getVarName(), "true");
+ }
+ }
+
+ @Test
+ public void testScheduleDisabledWithName() throws InterruptedException, IOException {
+
+ System.setProperty(ConfVars.ZEPPELIN_NOTEBOOK_CRON_FOLDERS.getVarName(), "System/*");
+ try {
+ final int timeout = 10;
+ final String everySecondCron = "* * * * * ?";
+ final CountDownLatch jobsToExecuteCount = new CountDownLatch(5);
+ final Note note = notebook.createNote(anonymous);
+
+ executeNewParagraphByCron(note, everySecondCron);
+ afterStatusChangedListener = new StatusChangedListener() {
+ @Override
+ public void onStatusChanged(Job job, Status before, Status after) {
+ if (after == Status.FINISHED) {
+ jobsToExecuteCount.countDown();
+ }
+ }
+ };
+
+ //This job should not run because it's name does not matches "ZEPPELIN_NOTEBOOK_CRON_FOLDERS"
+ assertFalse(jobsToExecuteCount.await(timeout, TimeUnit.SECONDS));
+
+ terminateScheduledNote(note);
+ afterStatusChangedListener = null;
+
+ final Note noteNameSystem = notebook.createNote(anonymous);
+ noteNameSystem.setName("System/test1");
+ final CountDownLatch jobsToExecuteCountNameSystem = new CountDownLatch(5);
+
+ executeNewParagraphByCron(noteNameSystem, everySecondCron);
+ afterStatusChangedListener = new StatusChangedListener() {
+ @Override
+ public void onStatusChanged(Job job, Status before, Status after) {
+ if (after == Status.FINISHED) {
+ jobsToExecuteCountNameSystem.countDown();
+ }
+ }
+ };
+
+ //This job should run because it's name contains "System/"
+ assertTrue(jobsToExecuteCountNameSystem.await(timeout, TimeUnit.SECONDS));
+
+ terminateScheduledNote(noteNameSystem);
+ afterStatusChangedListener = null;
+ } finally {
+ System.clearProperty(ConfVars.ZEPPELIN_NOTEBOOK_CRON_FOLDERS.getVarName());
+ }
+ }
+
private void terminateScheduledNote(Note note) {
note.getConfig().remove("cron");
notebook.refreshCron(note.getId());
notebook.removeNote(note.getId(), anonymous);
}
-
+
@Test
public void testAutoRestartInterpreterAfterSchedule() throws InterruptedException, IOException{
// create a note and a paragraph
@@ -859,9 +940,9 @@ public class NotebookTest extends AbstractInterpreterTest implements JobListener
// set admin roles for both user1 and user2
notebookAuthorization.setRoles(user1, roles);
notebookAuthorization.setRoles(user2, roles);
-
+
Note note = notebook.createNote(new AuthenticationInfo(user1));
-
+
// check that user1 is owner, reader, runner and writer
assertEquals(notebookAuthorization.isOwner(note.getId(),
Sets.newHashSet(user1)), true);
@@ -871,7 +952,7 @@ public class NotebookTest extends AbstractInterpreterTest implements JobListener
Sets.newHashSet(user2)), true);
assertEquals(notebookAuthorization.isWriter(note.getId(),
Sets.newHashSet(user1)), true);
-
+
// since user1 and user2 both have admin role, user2 will be reader and writer as well
assertEquals(notebookAuthorization.isOwner(note.getId(),
Sets.newHashSet(user2)), false);
@@ -881,14 +962,14 @@ public class NotebookTest extends AbstractInterpreterTest implements JobListener
Sets.newHashSet(user2)), true);
assertEquals(notebookAuthorization.isWriter(note.getId(),
Sets.newHashSet(user2)), true);
-
+
// check that user1 has note listed in his workbench
Set<String> user1AndRoles = notebookAuthorization.getRoles(user1);
user1AndRoles.add(user1);
List<Note> user1Notes = notebook.getAllNotes(user1AndRoles);
assertEquals(user1Notes.size(), 1);
assertEquals(user1Notes.get(0).getId(), note.getId());
-
+
// check that user2 has note listed in his workbench because of admin role
Set<String> user2AndRoles = notebookAuthorization.getRoles(user2);
user2AndRoles.add(user2);
@@ -896,7 +977,7 @@ public class NotebookTest extends AbstractInterpreterTest implements JobListener
assertEquals(user2Notes.size(), 1);
assertEquals(user2Notes.get(0).getId(), note.getId());
}
-
+
@Test
public void testAbortParagraphStatusOnInterpreterRestart() throws InterruptedException,
IOException, InterpreterException {
@@ -1225,7 +1306,7 @@ public class NotebookTest extends AbstractInterpreterTest implements JobListener
notes2 = notebook.getAllNotes(user2);
assertEquals(notes1.size(), 1);
assertEquals(notes2.size(), 1);
-
+
notebook.getNotebookAuthorization().setReaders(note.getId(), Sets.newHashSet("user1"));
//note is public since writers empty
notes1 = notebook.getAllNotes(user1);
@@ -1238,7 +1319,7 @@ public class NotebookTest extends AbstractInterpreterTest implements JobListener
notes2 = notebook.getAllNotes(user2);
assertEquals(notes1.size(), 1);
assertEquals(notes2.size(), 1);
-
+
notebook.getNotebookAuthorization().setWriters(note.getId(), Sets.newHashSet("user1"));
notes1 = notebook.getAllNotes(user1);
notes2 = notebook.getAllNotes(user2);
@@ -1250,19 +1331,19 @@ public class NotebookTest extends AbstractInterpreterTest implements JobListener
public void testPublicPrivateNewNote() throws IOException, SchedulerException {
HashSet<String> user1 = Sets.newHashSet("user1");
HashSet<String> user2 = Sets.newHashSet("user2");
-
+
// case of public note
assertTrue(conf.isNotebookPublic());
assertTrue(notebookAuthorization.isPublic());
-
+
List<Note> notes1 = notebook.getAllNotes(user1);
List<Note> notes2 = notebook.getAllNotes(user2);
assertEquals(notes1.size(), 0);
assertEquals(notes2.size(), 0);
-
+
// user1 creates note
Note notePublic = notebook.createNote(new AuthenticationInfo("user1"));
-
+
// both users have note
notes1 = notebook.getAllNotes(user1);
notes2 = notebook.getAllNotes(user2);
@@ -1270,7 +1351,7 @@ public class NotebookTest extends AbstractInterpreterTest implements JobListener
assertEquals(notes2.size(), 1);
assertEquals(notes1.get(0).getId(), notePublic.getId());
assertEquals(notes2.get(0).getId(), notePublic.getId());
-
+
// user1 is only owner
assertEquals(notebookAuthorization.getOwners(notePublic.getId()).size(), 1);
assertEquals(notebookAuthorization.getReaders(notePublic.getId()).size(), 0);
@@ -1283,13 +1364,13 @@ public class NotebookTest extends AbstractInterpreterTest implements JobListener
assertFalse(conf2.isNotebookPublic());
// notebook authorization reads from conf, so no need to re-initilize
assertFalse(notebookAuthorization.isPublic());
-
+
// check that still 1 note per user
notes1 = notebook.getAllNotes(user1);
notes2 = notebook.getAllNotes(user2);
assertEquals(notes1.size(), 1);
assertEquals(notes2.size(), 1);
-
+
// create private note
Note notePrivate = notebook.createNote(new AuthenticationInfo("user1"));
@@ -1299,18 +1380,18 @@ public class NotebookTest extends AbstractInterpreterTest implements JobListener
assertEquals(notes1.size(), 2);
assertEquals(notes2.size(), 1);
assertEquals(true, notes1.contains(notePrivate));
-
+
// user1 have all rights
assertEquals(notebookAuthorization.getOwners(notePrivate.getId()).size(), 1);
assertEquals(notebookAuthorization.getReaders(notePrivate.getId()).size(), 1);
assertEquals(notebookAuthorization.getRunners(notePrivate.getId()).size(), 1);
assertEquals(notebookAuthorization.getWriters(notePrivate.getId()).size(), 1);
-
+
//set back public to true
System.setProperty(ConfVars.ZEPPELIN_NOTEBOOK_PUBLIC.getVarName(), "true");
ZeppelinConfiguration.create();
}
-
+
private void delete(File file){
if(file.isFile()) file.delete();
else if(file.isDirectory()){