Let snapshot saved disk same as rocksdb data path

hugegraph-api/pom.xml

@@ -171,7 +171,7 @@
                             </addDefaultSpecificationEntries>
                         </manifest>
                         <manifestEntries>
-                            <Implementation-Version>0.59.0.0</Implementation-Version>
+                            <Implementation-Version>0.60.0.0</Implementation-Version>
                         </manifestEntries>
                     </archive>
                 </configuration>

hugegraph-api/src/main/java/com/baidu/hugegraph/api/profile/GraphsAPI.java

@@ -138,6 +138,34 @@ public void clear(@Context GraphManager manager,
         g.truncateBackend();
     }
 
+    @PUT
+    @Timed
+    @Path("{name}/snapshot_create")
+    @Produces(APPLICATION_JSON_WITH_CHARSET)
+    @RolesAllowed({"admin", "$owner=$name"})
+    public Object createSnapshot(@Context GraphManager manager,
+                                 @PathParam("name") String name) {
+        LOG.debug("Create snapshot for graph '{}'", name);
+
+        HugeGraph g = graph(manager, name);
+        g.createSnapshot();
+        return ImmutableMap.of(name, "succeed");
+    }
+
+    @PUT
+    @Timed
+    @Path("{name}/snapshot_resume")
+    @Produces(APPLICATION_JSON_WITH_CHARSET)
+    @RolesAllowed({"admin", "$owner=$name"})
+    public Object resumeSnapshot(@Context GraphManager manager,
+                                 @PathParam("name") String name) {
+        LOG.debug("Resume snapshot for graph '{}'", name);
+
+        HugeGraph g = graph(manager, name);
+        g.resumeSnapshot();
+        return ImmutableMap.of(name, "succeed");
+    }
+
     @PUT
     @Timed
     @Path("{name}/mode")

hugegraph-api/src/main/java/com/baidu/hugegraph/auth/HugeGraphAuthProxy.java

@@ -700,6 +700,18 @@ public void truncateBackend() {
         }
     }
 
+    @Override
+    public void createSnapshot() {
+        this.verifyPermission(HugePermission.WRITE, ResourceType.STATUS);
+        this.hugegraph.createSnapshot();
+    }
+
+    @Override
+    public void resumeSnapshot() {
+        this.verifyPermission(HugePermission.WRITE, ResourceType.STATUS);
+        this.hugegraph.resumeSnapshot();
+    }
+
     private void verifyAdminPermission() {
         verifyPermission(HugePermission.ANY, ResourceType.ROOT);
     }

hugegraph-api/src/main/java/com/baidu/hugegraph/version/ApiVersion.java

@@ -109,10 +109,11 @@ public final class ApiVersion {
      * [0.58] Issue-1173: Supports customized kout/kneighbor,
      *        multi-node-shortest-path, jaccard-similar and template-paths
      * [0.59] Issue-1333: Support graph read mode for olap property
+     * [0.60] Issue-1392: Support create and resume snapshot
      */
 
     // The second parameter of Version.of() is for IDE running without JAR
-    public static final Version VERSION = Version.of(ApiVersion.class, "0.59");
+    public static final Version VERSION = Version.of(ApiVersion.class, "0.60");
 
     public static final void check() {
         // Check version of hugegraph-core. Firstly do check from version 0.3

hugegraph-core/src/main/java/com/baidu/hugegraph/HugeGraph.java

@@ -153,6 +153,9 @@ public interface HugeGraph extends Graph {
     public void clearBackend();
     public void truncateBackend();
 
+    public void createSnapshot();
+    public void resumeSnapshot();
+
     @Override
     public HugeFeatures features();
 

hugegraph-core/src/main/java/com/baidu/hugegraph/StandardHugeGraph.java

@@ -360,14 +360,36 @@ public void truncateBackend() {
              * When restarting, load the snapshot first and then read backend,
              * will not encounter such an intermediate state.
              */
-            this.storeProvider.writeSnapshot();
+            this.storeProvider.createSnapshot();
         } finally {
             LockUtil.unlock(this.name, LockUtil.GRAPH_LOCK);
         }
 
         LOG.info("Graph '{}' has been truncated", this.name);
     }
 
+    @Override
+    public void createSnapshot() {
+        LockUtil.lock(this.name, LockUtil.GRAPH_LOCK);
+        try {
+            this.storeProvider.createSnapshot();
+        } finally {
+            LockUtil.unlock(this.name, LockUtil.GRAPH_LOCK);
+        }
+        LOG.info("Graph '{}' has created snapshot", this.name);
+    }
+
+    @Override
+    public void resumeSnapshot() {
+        LockUtil.lock(this.name, LockUtil.GRAPH_LOCK);
+        try {
+            this.storeProvider.resumeSnapshot();
+        } finally {
+            LockUtil.unlock(this.name, LockUtil.GRAPH_LOCK);
+        }
+        LOG.info("Graph '{}' has resumed from snapshot", this.name);
+    }
+
     private SchemaTransaction openSchemaTransaction() throws HugeException {
         this.checkGraphNotClosed();
         try {
@@ -1424,7 +1446,7 @@ public void invalid2(HugeType type, Object[] ids) {
 
         @Override
         public void clear(HugeType type) {
-            this.hub.notify(Events.CACHE, Cache.ACTION_CLEAR, type, null);
+            this.hub.notify(Events.CACHE, Cache.ACTION_CLEAR, type);
         }
 
         @Override

hugegraph-core/src/main/java/com/baidu/hugegraph/backend/store/AbstractBackendStoreProvider.java

@@ -19,14 +19,20 @@
 
 package com.baidu.hugegraph.backend.store;
 
+import java.io.File;
+import java.io.IOException;
+import java.util.HashSet;
 import java.util.Map;
+import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.Future;
 
+import org.apache.commons.io.FileUtils;
 import org.slf4j.Logger;
 
 import com.baidu.hugegraph.HugeGraph;
 import com.baidu.hugegraph.backend.BackendException;
+import com.baidu.hugegraph.backend.store.raft.StoreSnapshotFile;
 import com.baidu.hugegraph.event.EventHub;
 import com.baidu.hugegraph.event.EventListener;
 import com.baidu.hugegraph.util.E;
@@ -151,13 +157,32 @@ public void initSystemInfo(HugeGraph graph) {
     }
 
     @Override
-    public void writeSnapshot() {
-        // TODO: to be implemented
+    public void createSnapshot() {
+        String snapshotPrefix = StoreSnapshotFile.SNAPSHOT_DIR;
+        for (BackendStore store : this.stores.values()) {
+            if (store.features().supportsSnapshot()) {
+                store.createSnapshot(snapshotPrefix);
+            }
+        }
     }
 
     @Override
-    public void readSnapshot() {
-        // TODO: to be implemented
+    public void resumeSnapshot() {
+        String snapshotPrefix = StoreSnapshotFile.SNAPSHOT_DIR;
+        Set<String> snapshotDirs = new HashSet<>();
+        for (BackendStore store : this.stores.values()) {
+            if (store.features().supportsSnapshot()) {
+                snapshotDirs.addAll(store.resumeSnapshot(snapshotPrefix));
+            }
+        }
+        // Delete all snapshot parent directories
+        for (String snapshotDir : snapshotDirs) {
+            try {
+                FileUtils.deleteDirectory(new File(snapshotDir));
+            } catch (IOException e) {
+                LOG.warn("Failed to delete snapshot directory {}", snapshotDir);
+            }
+        }
     }
 
     @Override

hugegraph-core/src/main/java/com/baidu/hugegraph/backend/store/BackendFeatures.java

@@ -29,6 +29,10 @@ public default boolean supportsSharedStorage() {
         return true;
     }
 
+    public default boolean supportsSnapshot() {
+        return false;
+    }
+
     public boolean supportsScanToken();
 
     public boolean supportsScanKeyPrefix();

hugegraph-core/src/main/java/com/baidu/hugegraph/backend/store/BackendStore.java

@@ -20,6 +20,7 @@
 package com.baidu.hugegraph.backend.store;
 
 import java.util.Iterator;
+import java.util.Set;
 
 import com.baidu.hugegraph.backend.id.Id;
 import com.baidu.hugegraph.backend.id.IdGenerator;
@@ -116,12 +117,12 @@ public default void setCounterLowest(HugeType type, long lowest) {
     // Get current counter for a specific type
     public long getCounter(HugeType type);
 
-    public default void writeSnapshot(String snapshotPath) {
-        throw new UnsupportedOperationException("writeSnapshot");
+    public default Set<String> createSnapshot(String snapshotDir) {
+        throw new UnsupportedOperationException("createSnapshot");
     }
 
-    public default void readSnapshot(String snapshotPath) {
-        throw new UnsupportedOperationException("readSnapshot");
+    public default Set<String> resumeSnapshot(String snapshotDir) {
+        throw new UnsupportedOperationException("resumeSnapshot");
     }
 
     static enum TxState {

hugegraph-core/src/main/java/com/baidu/hugegraph/backend/store/BackendStoreProvider.java

@@ -54,9 +54,9 @@ public interface BackendStoreProvider {
 
     public void initSystemInfo(HugeGraph graph);
 
-    public void writeSnapshot();
+    public void createSnapshot();
 
-    public void readSnapshot();
+    public void resumeSnapshot();
 
     public void listen(EventListener listener);
 

hugegraph-core/src/main/java/com/baidu/hugegraph/backend/store/raft/RaftBackendStoreProvider.java

@@ -199,7 +199,7 @@ public void initSystemInfo(HugeGraph graph) {
     }
 
     @Override
-    public void writeSnapshot() {
+    public void createSnapshot() {
         StoreCommand command = new StoreCommand(StoreType.ALL,
                                                 StoreAction.SNAPSHOT, null);
         StoreClosure closure = new StoreClosure(command);
@@ -208,8 +208,9 @@ public void writeSnapshot() {
     }
 
     @Override
-    public void readSnapshot() {
-        // How to read snapshot by jraft explicity?
+    public void resumeSnapshot() {
+        // Jraft doesn't expose API to load snapshot
+        throw new UnsupportedOperationException("resumeSnapshot");
     }
 
     @Override

hugegraph-core/src/main/java/com/baidu/hugegraph/backend/store/raft/RaftSharedContext.java

@@ -22,6 +22,7 @@
 import java.io.File;
 import java.io.IOException;
 import java.nio.file.Paths;
+import java.util.List;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.LinkedBlockingQueue;
@@ -43,6 +44,7 @@
 import com.baidu.hugegraph.HugeException;
 import com.baidu.hugegraph.HugeGraphParams;
 import com.baidu.hugegraph.backend.cache.Cache;
+import com.baidu.hugegraph.backend.id.Id;
 import com.baidu.hugegraph.backend.store.BackendStore;
 import com.baidu.hugegraph.backend.store.raft.rpc.ListPeersProcessor;
 import com.baidu.hugegraph.backend.store.raft.rpc.RaftRequests.StoreType;
@@ -259,7 +261,7 @@ public void clearCache() {
         this.notifyCache(Cache.ACTION_CLEAR, HugeType.VERTEX, null);
     }
 
-    protected void notifyCache(String action, HugeType type, Object id) {
+    protected void notifyCache(String action, HugeType type, List<Id> ids) {
         EventHub eventHub;
         if (type.isGraph()) {
             eventHub = this.params.graphEventHub();
@@ -270,7 +272,15 @@ protected void notifyCache(String action, HugeType type, Object id) {
         }
         try {
             // How to avoid update cache from server info
-            eventHub.notify(Events.CACHE, action, type, id);
+            if (ids == null) {
+                eventHub.call(Events.CACHE, action, type);
+            } else {
+                if (ids.size() == 1) {
+                    eventHub.call(Events.CACHE, action, type, ids.get(0));
+                } else {
+                    eventHub.call(Events.CACHE, action, type, ids.toArray());
+                }
+            }
         } catch (RejectedExecutionException e) {
             LOG.warn("Can't update cache due to EventHub is too busy");
         }