-
Notifications
You must be signed in to change notification settings - Fork 1.2k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
* (fix) multi-group snapshot with memoryStore * (fix) multi-group snapshot with rocksDBStore * (fix) add multi sst snapshot benchmark * (fix) add meta check on load snapshot * (fix) add meta check on load snapshot * (fix) add timer metric with memory snapshot * (fix) typo * (fix) log * (bugfix) fencing token isolation * (bugfix) rename local variable * (bugfix) fencing token isolation by region * (fix) typo * (fix) region info visibility * (feat) add comment * (fix) delete existing data on rocksdb restart, because raft will play them back * (fix) delete existing data on rocksdb restart, because raft will play them back * (fix) add restart test #86 * (fix) add restart test #86 * (fix) update restart test * (fix) add error log on splitting fail * (fix) requires * (fix) remove useless code * (fix) code refactoring with snapshot * (fix) fix bad name * (fix) typo
- Loading branch information
1 parent
09fb200
commit 11c7d60
Showing
31 changed files
with
1,405 additions
and
622 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
130 changes: 130 additions & 0 deletions
130
...kv-core/src/main/java/com/alipay/sofa/jraft/rhea/storage/AbstractKVStoreSnapshotFile.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,130 @@ | ||
| /* | ||
| * 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 com.alipay.sofa.jraft.rhea.storage; | ||
|
|
||
| import java.io.FileOutputStream; | ||
| import java.io.IOException; | ||
| import java.nio.file.Paths; | ||
| import java.util.concurrent.ExecutorService; | ||
| import java.util.zip.ZipOutputStream; | ||
|
|
||
| import org.slf4j.Logger; | ||
| import org.slf4j.LoggerFactory; | ||
|
|
||
| import com.alipay.sofa.jraft.Closure; | ||
| import com.alipay.sofa.jraft.Status; | ||
| import com.alipay.sofa.jraft.error.RaftError; | ||
| import com.alipay.sofa.jraft.rhea.metadata.Region; | ||
| import com.alipay.sofa.jraft.rhea.serialization.Serializer; | ||
| import com.alipay.sofa.jraft.rhea.serialization.Serializers; | ||
| import com.alipay.sofa.jraft.rhea.util.StackTraceUtil; | ||
| import com.alipay.sofa.jraft.rhea.util.ZipUtil; | ||
| import com.alipay.sofa.jraft.storage.snapshot.SnapshotReader; | ||
| import com.alipay.sofa.jraft.storage.snapshot.SnapshotWriter; | ||
| import com.google.protobuf.ByteString; | ||
|
|
||
| import static com.alipay.sofa.jraft.entity.LocalFileMetaOutter.LocalFileMeta; | ||
|
|
||
| /** | ||
| * @author jiachun.fjc | ||
| */ | ||
| public abstract class AbstractKVStoreSnapshotFile implements KVStoreSnapshotFile { | ||
|
|
||
| private static final Logger LOG = LoggerFactory.getLogger(AbstractKVStoreSnapshotFile.class); | ||
|
|
||
| private static final String SNAPSHOT_DIR = "kv"; | ||
| private static final String SNAPSHOT_ARCHIVE = "kv.zip"; | ||
|
|
||
| protected final Serializer serializer = Serializers.getDefault(); | ||
|
|
||
| @Override | ||
| public void save(final SnapshotWriter writer, final Closure done, final Region region, | ||
| final ExecutorService executor) { | ||
| final String writerPath = writer.getPath(); | ||
| final String snapshotPath = Paths.get(writerPath, SNAPSHOT_DIR).toString(); | ||
| try { | ||
| final LocalFileMeta meta = doSnapshotSave(snapshotPath, region); | ||
| executor.execute(() -> compressSnapshot(writer, meta, done)); | ||
| } catch (final Throwable t) { | ||
| LOG.error("Fail to save snapshot, path={}, file list={}, {}.", writerPath, writer.listFiles(), | ||
| StackTraceUtil.stackTrace(t)); | ||
| done.run(new Status(RaftError.EIO, "Fail to save snapshot at %s, error is %s", writerPath, | ||
| t.getMessage())); | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public boolean load(final SnapshotReader reader, final Region region) { | ||
| final LocalFileMeta meta = (LocalFileMeta) reader.getFileMeta(SNAPSHOT_ARCHIVE); | ||
| final String readerPath = reader.getPath(); | ||
| if (meta == null) { | ||
| LOG.error("Can't find kv snapshot file, path={}.", readerPath); | ||
| return false; | ||
| } | ||
| final String snapshotPath = Paths.get(readerPath, SNAPSHOT_DIR).toString(); | ||
| try { | ||
| decompressSnapshot(readerPath); | ||
| doSnapshotLoad(snapshotPath, meta, region); | ||
| return true; | ||
| } catch (final Throwable t) { | ||
| LOG.error("Fail to load snapshot, path={}, file list={}, {}.", readerPath, reader.listFiles(), | ||
| StackTraceUtil.stackTrace(t)); | ||
| return false; | ||
| } | ||
| } | ||
|
|
||
| abstract LocalFileMeta doSnapshotSave(final String snapshotPath, final Region region) throws Exception; | ||
|
|
||
| abstract void doSnapshotLoad(final String snapshotPath, final LocalFileMeta meta, final Region region) | ||
| throws Exception; | ||
|
|
||
| protected void compressSnapshot(final SnapshotWriter writer, final LocalFileMeta meta, final Closure done) { | ||
| final String writerPath = writer.getPath(); | ||
| final String outputFile = Paths.get(writerPath, SNAPSHOT_ARCHIVE).toString(); | ||
| try { | ||
| try (final ZipOutputStream out = new ZipOutputStream(new FileOutputStream(outputFile))) { | ||
| ZipUtil.compressDirectoryToZipFile(writerPath, SNAPSHOT_DIR, out); | ||
| } | ||
| if (writer.addFile(SNAPSHOT_ARCHIVE, meta)) { | ||
| done.run(Status.OK()); | ||
| } else { | ||
| done.run(new Status(RaftError.EIO, "Fail to add snapshot file: %s", writerPath)); | ||
| } | ||
| } catch (final Throwable t) { | ||
| LOG.error("Fail to compress snapshot, path={}, file list={}, {}.", writerPath, writer.listFiles(), | ||
| StackTraceUtil.stackTrace(t)); | ||
| done.run(new Status(RaftError.EIO, "Fail to compress snapshot at %s, error is %s", writerPath, t | ||
| .getMessage())); | ||
| } | ||
| } | ||
|
|
||
| protected void decompressSnapshot(final String readerPath) throws IOException { | ||
| final String sourceFile = Paths.get(readerPath, SNAPSHOT_ARCHIVE).toString(); | ||
| ZipUtil.unzipFile(sourceFile, readerPath); | ||
| } | ||
|
|
||
| protected <T> T readMetadata(final LocalFileMeta meta, final Class<T> cls) { | ||
| final ByteString userMeta = meta.getUserMeta(); | ||
| return this.serializer.readObject(userMeta.toByteArray(), cls); | ||
| } | ||
|
|
||
| protected <T> LocalFileMeta buildMetadata(final T metadata) { | ||
| return metadata == null ? null : LocalFileMeta.newBuilder() // | ||
| .setUserMeta(ByteString.copyFrom(this.serializer.writeObject(metadata))) // | ||
| .build(); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
50 changes: 50 additions & 0 deletions
50
...akv/rheakv-core/src/main/java/com/alipay/sofa/jraft/rhea/storage/KVStoreSnapshotFile.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,50 @@ | ||
| /* | ||
| * 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 com.alipay.sofa.jraft.rhea.storage; | ||
|
|
||
| import java.util.concurrent.ExecutorService; | ||
|
|
||
| import com.alipay.sofa.jraft.Closure; | ||
| import com.alipay.sofa.jraft.rhea.metadata.Region; | ||
| import com.alipay.sofa.jraft.storage.snapshot.SnapshotReader; | ||
| import com.alipay.sofa.jraft.storage.snapshot.SnapshotWriter; | ||
|
|
||
| /** | ||
| * | ||
| * @author jiachun.fjc | ||
| */ | ||
| public interface KVStoreSnapshotFile { | ||
|
|
||
| /** | ||
| * Save a snapshot for the specified region. | ||
| * | ||
| * @param writer snapshot writer | ||
| * @param done callback | ||
| * @param region the region to save snapshot | ||
| * @param executor the executor to compress snapshot | ||
| */ | ||
| void save(final SnapshotWriter writer, final Closure done, final Region region, final ExecutorService executor); | ||
|
|
||
| /** | ||
| * Load snapshot for the specified region. | ||
| * | ||
| * @param reader snapshot reader | ||
| * @param region the region to load snapshot | ||
| * @return true if load succeed | ||
| */ | ||
| boolean load(final SnapshotReader reader, final Region region); | ||
| } |
44 changes: 44 additions & 0 deletions
44
...akv-core/src/main/java/com/alipay/sofa/jraft/rhea/storage/KVStoreSnapshotFileFactory.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,44 @@ | ||
| /* | ||
| * 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 com.alipay.sofa.jraft.rhea.storage; | ||
|
|
||
| import com.alipay.sofa.jraft.util.Requires; | ||
|
|
||
| /** | ||
| * | ||
| * @author jiachun.fjc | ||
| */ | ||
| public final class KVStoreSnapshotFileFactory { | ||
|
|
||
| public static <T> KVStoreSnapshotFile getKVStoreSnapshotFile(final BaseRawKVStore<T> kvStore) { | ||
| Requires.requireNonNull(kvStore, "kvStore"); | ||
| if (kvStore instanceof RocksRawKVStore) { | ||
| return new RocksKVStoreSnapshotFile((RocksRawKVStore) kvStore); | ||
| } | ||
| if (kvStore instanceof MemoryRawKVStore) { | ||
| return new MemoryKVStoreSnapshotFile((MemoryRawKVStore) kvStore); | ||
| } | ||
| throw reject("fail to find a KVStoreSnapshotFile with " + kvStore.getClass().getName()); | ||
| } | ||
|
|
||
| private static UnsupportedOperationException reject(final String message) { | ||
| return new UnsupportedOperationException(message); | ||
| } | ||
|
|
||
| private KVStoreSnapshotFileFactory() { | ||
| } | ||
| } |
Oops, something went wrong.