Zeppelin-540: Notebook versioning control, commit from frontend notebook menu

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

@@ -99,12 +99,17 @@ public static enum OP {
 
     ANGULAR_OBJECT_UPDATE,  // [s-c] add/update angular object
     ANGULAR_OBJECT_REMOVE,  // [s-c] add angular object del
-
+    
     ANGULAR_OBJECT_UPDATED,  // [c-s] angular object value updated,
 
     LIST_CONFIGURATIONS, // [c-s] ask all key/value pairs of configurations
-    CONFIGURATIONS_INFO // [s-c] all key/value pairs of configurations
+    CONFIGURATIONS_INFO, // [s-c] all key/value pairs of configurations
                   // @param settings serialized Map<String, String> object
+
+    CHECKPOINT_NOTEBOOK     // [c-s] checkpoint notebook to storage repository
+                            // @param noteId
+                            // @param checkpointName
+
   }
 
   public OP op;

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

@@ -171,6 +171,9 @@ public void onMessage(NotebookSocket conn, String msg) {
           case LIST_CONFIGURATIONS:
             sendAllConfigurations(conn, notebook);
             break;
+          case CHECKPOINT_NOTEBOOK:
+            checkpointNotebook(conn, notebook, messagereceived);
+            break;
           default:
             broadcastNoteList();
             break;
@@ -736,6 +739,13 @@ public boolean apply(String key) {
         .put("configurations", configurations)));
   }
 
+  private void checkpointNotebook(NotebookSocket conn, Notebook notebook,
+      Message fromMessage) throws IOException {
+    String noteId = (String) fromMessage.get("noteId");
+    String commitMessage = (String) fromMessage.get("commitMessage");
+    notebook.checkpointNote(noteId, commitMessage);
+  }
+
   /**
    * This callback is for the paragraph that runs on ZeppelinServer
    * @param noteId

zeppelin-web/src/app/notebook/notebook-actionBar.html

@@ -61,6 +61,32 @@ <h3>
               tooltip-placement="bottom" tooltip="Export the notebook">
         <i class="fa fa-download"></i>
       </button>
+        <ul class="dropdown-menu" role="menu" style="width:250px">
+          <li>
+            <div class="cron-preset-container">
+              <div>
+                <input type="text"
+                       dropdown-input
+                       placeholder="commit message"
+                       id="note.checkpoint.message"
+                       ng-model="note.checkpoint.message"/>
+                <button type="button"
+                  class="btn btn-default btn-xs"
+                  ng-hide="viewOnly"
+                  ng-click="checkpointNotebook(note.checkpoint.message)"
+                  tooltip-placement="bottom" tooltip="Commit the notebook">Commit
+                </button>
+              </div>
+            </div>
+          </li>
+        </ul>
+        <button type="button"
+                class="btn btn-default btn-xs dropdown-toggle"
+                ng-hide="viewOnly"
+                data-toggle="dropdown"
+                tooltip-placement="bottom" tooltip="Version control">
+          <i class="fa fa-file-code-o"></i>
+        </button>
     </span>
 
 <!-- put the delete action by itself for your protection -->

zeppelin-web/src/app/notebook/notebook.controller.js

@@ -154,6 +154,21 @@ angular.module('zeppelinWebApp').controller('NotebookCtrl',
     });
   };
 
+  // checkpoint/commit notebook
+  $scope.checkpointNotebook = function(commitMessage) {
+    BootstrapDialog.confirm({
+      closable: true,
+      title: '',
+      message: 'Commit notebook to current repository?',
+      callback: function(result) {
+        if (result) {
+          websocketMsgSrv.checkpointNotebook($routeParams.noteId, commitMessage);
+        }
+      }
+    });
+    document.getElementById('note.checkpoint.message').value='';
+  };
+
   $scope.runNote = function() {
     BootstrapDialog.confirm({
       closable: true,

zeppelin-web/src/components/websocketEvents/websocketMsg.service.js

@@ -128,6 +128,16 @@ angular.module('zeppelinWebApp').service('websocketMsgSrv', function($rootScope,
       });
     },
 
+    checkpointNotebook: function(noteId, commitMessage) {
+      websocketEvents.sendNewEvent({
+        op: 'CHECKPOINT_NOTEBOOK',
+        data: {
+          noteId: noteId,
+          commitMessage: commitMessage
+        }
+      });
+    },
+
     isConnected: function(){
       return websocketEvents.isConnected();
     }

zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/Notebook.java

@@ -307,6 +307,10 @@ public void removeNote(String id) {
     }
   }
 
+  public void checkpointNote(String noteId, String checkpointMessage) throws IOException {
+    notebookRepo.checkpoint(noteId, checkpointMessage);
+  }
+
   @SuppressWarnings("rawtypes")
   private Note loadNoteFromRepo(String id) {
     Note note = null;

zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/repo/GitNotebookRepo.java

@@ -61,28 +61,34 @@ public GitNotebookRepo(ZeppelinConfiguration conf) throws IOException {
       localRepo.create();
     }
     git = new Git(localRepo);
-    maybeAddAndCommit(".");
+    checkpoint(".", "initialization commit");
   }
 
   @Override
   public synchronized void save(Note note) throws IOException {
     super.save(note);
-    maybeAddAndCommit(note.getId());
   }
 
-  private void maybeAddAndCommit(String pattern) {
+  /* implemented as git add+commit
+   * @param pattern is the noteId
+   * @param commitMessage is a commit message (checkpoint name)
+   * (non-Javadoc)
+   * @see org.apache.zeppelin.notebook.repo.VFSNotebookRepo#checkpoint(String, String)
+   */
+  @Override
+  public void checkpoint(String pattern, String commitMessage) {
     try {
       List<DiffEntry> gitDiff = git.diff().call();
       if (!gitDiff.isEmpty()) {
         LOG.debug("Changes found for pattern '{}': {}", pattern, gitDiff);
         DirCache added = git.add().addFilepattern(pattern).call();
         LOG.debug("{} changes are about to be commited", added.getEntryCount());
-        git.commit().setMessage("Updated " + pattern).call();
+        git.commit().setMessage(commitMessage).call();
       } else {
         LOG.debug("No changes found {}", pattern);
       }
     } catch (GitAPIException e) {
-      LOG.error("Faild to add+comit {} to Git", pattern, e);
+      LOG.error("Failed to add+comit {} to Git", pattern, e);
     }
   }
 
@@ -100,7 +106,7 @@ public List<Rev> history(String noteId) {
       Iterable<RevCommit> logs = git.log().addPath(noteId).call();
       for (RevCommit log: logs) {
         history.add(new Rev(log.getName(), log.getCommitTime()));
-        LOG.debug(" - ({},{})", log.getName(), log.getCommitTime());
+        LOG.debug(" - ({},{},{})", log.getName(), log.getCommitTime(), log.getFullMessage());
       }
     } catch (GitAPIException e) {
       LOG.error("Failed to get logs for {}", noteId, e);

zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/repo/NotebookRepo.java

@@ -36,4 +36,9 @@ public interface NotebookRepo {
    * Release any underlying resources
    */
   public void close();
+
+  /**
+   * chekpoint (versioning) for notebooks (optional)
+   */
+  public void checkpoint(String noteId, String checkPointName) throws IOException;
 }

zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/repo/NotebookRepoSync.java

@@ -224,7 +224,8 @@ int getMaxRepoNum() {
 
   NotebookRepo getRepo(int repoIndex) throws IOException {
     if (repoIndex < 0 || repoIndex >= getRepoCount()) {
-      throw new IOException("Storage repo index is out of range");
+      throw new IOException("Requested storage index " + repoIndex
+          + " isn't initialized," + " repository count is " + getRepoCount());
     }
     return repos.get(repoIndex);
   }
@@ -351,4 +352,27 @@ public void close() {
     }
   }
 
+  //checkpoint to all available storages
+  @Override
+  public void checkpoint(String noteId, String checkPointName) throws IOException {
+    int repoCount = getRepoCount();
+    int errorCount = 0;
+    String errorMessage = "";
+    for (int i = 0; i < Math.min(repoCount, getMaxRepoNum()); i++) {
+      try {
+        getRepo(i).checkpoint(noteId, checkPointName);
+      } catch (IOException e) {
+        LOG.warn("Couldn't checkpoint in {} storage with index {} for note {}", 
+          getRepo(i).getClass().toString(), i, noteId);
+        errorMessage += "Error on storage class " + getRepo(i).getClass().toString() + 
+          " with index " + i + " : " + e.getMessage() + "\n";
+        errorCount++;
+      }
+    }
+    // throw exception if failed to commit for all initialized repos
+    if (errorCount == Math.min(repoCount, getMaxRepoNum())) {
+      throw new IOException(errorMessage);
+    }
+
+  }
 }

zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/repo/S3NotebookRepo.java

@@ -189,4 +189,10 @@ public void remove(String noteId) throws IOException {
   public void close() {
     //no-op
   }
+
+  @Override
+  public void checkpoint(String noteId, String checkPointName) throws IOException {
+    // no-op
+    LOG.info("Checkpoint feature isn't suported in {}", this.getClass().toString());
+  }
 }

zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/repo/VFSNotebookRepo.java

@@ -244,4 +244,10 @@ public void close() {
     //no-op    
   }
 
+  @Override
+  public void checkpoint(String noteId, String checkPointName) throws IOException {
+    // no-op
+    logger.info("Checkpoint feature isn't suported in {}", this.getClass().toString());
+  }
+
 }

zeppelin-zengine/src/test/java/org/apache/zeppelin/notebook/repo/GitNotebookRepoTest.java

@@ -22,12 +22,16 @@
 import java.io.File;
 import java.io.IOException;
 import java.util.List;
+import java.util.Map;
 
 import org.apache.commons.io.FileUtils;
 import org.apache.zeppelin.conf.ZeppelinConfiguration;
 import org.apache.zeppelin.conf.ZeppelinConfiguration.ConfVars;
 import org.apache.zeppelin.interpreter.mock.MockInterpreter1;
 import org.apache.zeppelin.interpreter.mock.MockInterpreter2;
+import org.apache.zeppelin.notebook.Note;
+import org.apache.zeppelin.notebook.NoteInfo;
+import org.apache.zeppelin.notebook.Paragraph;
 import org.apache.zeppelin.notebook.repo.NotebookRepoVersioned.Rev;
 import org.eclipse.jgit.api.Git;
 import org.eclipse.jgit.api.errors.GitAPIException;
@@ -112,4 +116,39 @@ public void showNotebookHistory() throws GitAPIException, IOException {
     assertThat(testNotebookHistory).isNotEmpty();
   }
 
+  @Test
+  public void addCheckpoint() throws IOException {
+    // initial checks
+    notebookRepo = new GitNotebookRepo(conf);
+    assertThat(notebookRepo.list()).isNotEmpty();
+    assertThat(containsNote(notebookRepo.list(), TEST_NOTE_ID)).isTrue();
+    List<Rev> notebookHistoryBefore = notebookRepo.history(TEST_NOTE_ID);
+    assertThat(notebookHistoryBefore).isNotEmpty();
+    int initialCount = notebookHistoryBefore.size();
+
+    // add changes to note
+    Note note = notebookRepo.get(TEST_NOTE_ID);
+    Paragraph p = note.addParagraph();
+    Map<String, Object> config = p.getConfig();
+    config.put("enabled", true);
+    p.setConfig(config);
+    p.setText("%md checkpoint test text");
+
+    // save and checkpoint note
+    notebookRepo.save(note);
+    notebookRepo.checkpoint(TEST_NOTE_ID, "test commit-mesage");
+
+    // see if commit is added
+    List<Rev> notebookHistoryAfter = notebookRepo.history(TEST_NOTE_ID);
+    assertThat(notebookHistoryAfter.size()).isEqualTo(initialCount + 1);
+  }
+
+  private boolean containsNote(List<NoteInfo> notes, String noteId) {
+    for (NoteInfo note: notes) {
+      if (note.getId().equals(noteId)) {
+        return true;
+      }
+    }
+    return false;
+  }
 }

zeppelin-zengine/src/test/java/org/apache/zeppelin/notebook/repo/NotebookRepoSyncTest.java

@@ -17,6 +17,7 @@
 
 package org.apache.zeppelin.notebook.repo;
 
+import static com.google.common.truth.Truth.assertThat;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;
 import static org.mockito.Mockito.mock;
@@ -44,6 +45,7 @@
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
+import org.quartz.SchedulerException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -59,6 +61,7 @@ public class NotebookRepoSyncTest implements JobListenerFactory {
   private NotebookRepoSync notebookRepoSync;
   private InterpreterFactory factory;
   private DependencyResolver depResolver;
+  private SearchService search;
   private static final Logger LOG = LoggerFactory.getLogger(NotebookRepoSyncTest.class);
 
   @Before
@@ -90,7 +93,7 @@ public void setUp() throws Exception {
     depResolver = new DependencyResolver(mainZepDir.getAbsolutePath() + "/local-repo");
     factory = new InterpreterFactory(conf, new InterpreterOption(false), null, null, depResolver);
 
-    SearchService search = mock(SearchService.class);
+    search = mock(SearchService.class);
     notebookRepoSync = new NotebookRepoSync(conf);
     notebookSync = new Notebook(conf, notebookRepoSync, schedulerFactory, factory, this, search);
   }
@@ -211,6 +214,44 @@ public void testSyncOnReloadedList() throws IOException {
     assertEquals(1, notebookRepoSync.list(1).size());
   }
 
+  @Test
+  public void testCheckpointOneStorage() throws IOException, SchedulerException {
+    System.setProperty(ConfVars.ZEPPELIN_NOTEBOOK_STORAGE.getVarName(), "org.apache.zeppelin.notebook.repo.GitNotebookRepo");
+    ZeppelinConfiguration vConf = ZeppelinConfiguration.create();
+
+    NotebookRepoSync vRepoSync = new NotebookRepoSync(vConf);
+    Notebook vNotebookSync = new Notebook(vConf, vRepoSync, schedulerFactory, factory, this, search);
+
+    // one git versioned storage initialized
+    assertThat(vRepoSync.getRepoCount()).isEqualTo(1);
+    assertThat(vRepoSync.getRepo(0)).isInstanceOf(GitNotebookRepo.class);
+
+    GitNotebookRepo gitRepo = (GitNotebookRepo) vRepoSync.getRepo(0);
+
+    // no notes
+    assertThat(vRepoSync.list().size()).isEqualTo(0);
+    // create note
+    Note note = vNotebookSync.createNote();
+    assertThat(vRepoSync.list().size()).isEqualTo(1);
+
+    String noteId = vRepoSync.list().get(0).getId();
+    // first checkpoint
+    vRepoSync.checkpoint(noteId, "checkpoint message");
+    int vCount = gitRepo.history(noteId).size();
+    assertThat(vCount).isEqualTo(1);
+
+    Paragraph p = note.addParagraph();
+    Map<String, Object> config = p.getConfig();
+    config.put("enabled", true);
+    p.setConfig(config);
+    p.setText("%md checkpoint test");
+
+    // save and checkpoint again
+    vRepoSync.save(note);
+    vRepoSync.checkpoint(noteId, "checkpoint message 2");
+    assertThat(gitRepo.history(noteId).size()).isEqualTo(vCount + 1);
+  }
+
   static void delete(File file){
     if(file.isFile()) file.delete();
       else if(file.isDirectory()){