-
Notifications
You must be signed in to change notification settings - Fork 4
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- introduce a task-based archiver - the new archiver is currently part of the persistence service - updates to reading logic to accomodate proper zip entry naming - currently we rely on gap between latest live written and what will be zipped for proper operation due to lack of certain infrastructure Signed-off-by: Atanas Atanasov <[email protected]>
- Loading branch information
Showing
17 changed files
with
342 additions
and
366 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
115 changes: 115 additions & 0 deletions
115
...va/com/hedera/block/server/persistence/storage/archive/AsyncBlockAsLocalFileArchiver.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,115 @@ | ||
| // SPDX-License-Identifier: Apache-2.0 | ||
| package com.hedera.block.server.persistence.storage.archive; | ||
|
|
||
| import com.hedera.block.common.utils.FileUtilities; | ||
| import com.hedera.block.common.utils.Preconditions; | ||
| import com.hedera.block.server.persistence.storage.PersistenceStorageConfig; | ||
| import com.hedera.block.server.persistence.storage.path.BlockPathResolver; | ||
| import edu.umd.cs.findbugs.annotations.NonNull; | ||
| import java.io.File; | ||
| import java.io.IOException; | ||
| import java.lang.System.Logger.Level; | ||
| import java.nio.file.Files; | ||
| import java.nio.file.Path; | ||
| import java.util.Comparator; | ||
| import java.util.List; | ||
| import java.util.Objects; | ||
| import java.util.stream.Stream; | ||
| import java.util.zip.ZipEntry; | ||
| import java.util.zip.ZipOutputStream; | ||
|
|
||
| /** | ||
| * TODO: add documentation | ||
| */ | ||
| public final class AsyncBlockAsLocalFileArchiver implements AsyncLocalBlockArchiver { | ||
| private static final System.Logger LOGGER = System.getLogger(AsyncBlockAsLocalFileArchiver.class.getName()); | ||
| private final BlockPathResolver pathResolver; | ||
| private final long blockNumberThreshold; | ||
|
|
||
| AsyncBlockAsLocalFileArchiver( | ||
| final long blockNumberThreshold, | ||
| @NonNull final PersistenceStorageConfig config, | ||
| @NonNull final BlockPathResolver pathResolver) { | ||
| this.blockNumberThreshold = blockNumberThreshold; | ||
| this.pathResolver = Objects.requireNonNull(pathResolver); | ||
| Preconditions.requireWhole(blockNumberThreshold); | ||
| final int archiveGroupSize = config.archiveBatchSize(); | ||
| if (blockNumberThreshold % archiveGroupSize != 0) { // @todo(517) require divisible exactly by | ||
| throw new IllegalArgumentException("Block number must be divisible by " + archiveGroupSize); | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public void run() { | ||
| try { | ||
| doArchive(); | ||
| } catch (final IOException e) { | ||
| // todo return a result instead of exception | ||
| throw new RuntimeException(e); | ||
| } | ||
| } | ||
|
|
||
| private void doArchive() throws IOException { | ||
| final long upperBound = blockNumberThreshold - 1; | ||
| final Path rootToArchive = pathResolver.resolveRawPathToArchiveParentUnderLive(upperBound); | ||
| LOGGER.log(Level.DEBUG, "Archiving Block Files under [%s]".formatted(rootToArchive)); | ||
| final List<Path> pathsToArchive; | ||
| try (final Stream<Path> tree = Files.walk(rootToArchive)) { | ||
| pathsToArchive = tree.sorted(Comparator.reverseOrder()) | ||
| .filter(Files::isRegularFile) | ||
| .toList(); | ||
| } | ||
| if (!pathsToArchive.isEmpty()) { | ||
| final Path zipFilePath = archiveInZip(upperBound, pathsToArchive, rootToArchive); | ||
| createSymlink(rootToArchive, zipFilePath); | ||
| deleteLive(rootToArchive); | ||
| } | ||
| } | ||
|
|
||
| private Path archiveInZip(final long upperBound, final List<Path> pathsToArchive, final Path rootToArchive) | ||
| throws IOException { | ||
| final Path zipFilePath = pathResolver.resolveRawPathToArchiveParentUnderArchive(upperBound); | ||
| if (!Files.exists(zipFilePath)) { | ||
| // @todo(517) should we assume something if the zip file already exists? If yes, what and how to | ||
| // handle? | ||
| FileUtilities.createFile(zipFilePath); | ||
| } | ||
| LOGGER.log(Level.DEBUG, "Target Zip Path [%s]".formatted(zipFilePath)); | ||
| try (final ZipOutputStream out = new ZipOutputStream(Files.newOutputStream(zipFilePath))) { | ||
| for (int i = 0; i < pathsToArchive.size(); i++) { | ||
| final Path pathToArchive = pathsToArchive.get(i); | ||
| final String relativizedEntryName = | ||
| rootToArchive.relativize(pathToArchive).toString(); | ||
| final ZipEntry zipEntry = new ZipEntry(relativizedEntryName); | ||
| LOGGER.log(Level.TRACE, "Adding Zip Entry [%s] to zip file [%s]".formatted(zipEntry, zipFilePath)); | ||
| out.putNextEntry(zipEntry); | ||
| Files.copy(pathToArchive, out); | ||
| out.closeEntry(); | ||
| LOGGER.log( | ||
| Level.TRACE, | ||
| "Zip Entry [%s] successfully added to zip file [%s]".formatted(zipEntry, zipFilePath)); | ||
| } | ||
| } | ||
| LOGGER.log(Level.DEBUG, "Zip File [%s] successfully created".formatted(zipFilePath)); | ||
| return zipFilePath; | ||
| } | ||
|
|
||
| private static void createSymlink(final Path rootToArchive, final Path zipFilePath) throws IOException { | ||
| // we need to create a symlink to the zip file we just created so readers can find it | ||
| final Path liveSymlink = FileUtilities.appendExtension(rootToArchive, ".zip"); | ||
| Files.createSymbolicLink(liveSymlink, zipFilePath); | ||
| LOGGER.log(Level.DEBUG, "Symlink [%s <-> %s] created".formatted(liveSymlink, zipFilePath)); | ||
| } | ||
|
|
||
| private void deleteLive(final Path rootToArchive) throws IOException { | ||
| // we need to move the live dir that we just archived so readers will no longer be able | ||
| // to find it, hence they will fall back to search for the symlink we just made as well | ||
| // in the meantime, while readers get data from the symlink, we can safely delete the | ||
| // live dir | ||
| final Path movedToDelete = FileUtilities.appendExtension(rootToArchive, "del"); | ||
| Files.move(rootToArchive, movedToDelete); | ||
| try (Stream<Path> paths = Files.walk(movedToDelete)) { | ||
| paths.sorted(Comparator.reverseOrder()).map(Path::toFile).forEach(File::delete); | ||
| } | ||
| } | ||
| } |
26 changes: 26 additions & 0 deletions
26
...ava/com/hedera/block/server/persistence/storage/archive/AsyncBlockAsLocalFileFactory.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,26 @@ | ||
| // SPDX-License-Identifier: Apache-2.0 | ||
| package com.hedera.block.server.persistence.storage.archive; | ||
|
|
||
| import com.hedera.block.server.persistence.storage.PersistenceStorageConfig; | ||
| import com.hedera.block.server.persistence.storage.path.BlockPathResolver; | ||
| import edu.umd.cs.findbugs.annotations.NonNull; | ||
| import java.util.Objects; | ||
|
|
||
| /** | ||
| * TODO: add documentation | ||
| */ | ||
| public class AsyncBlockAsLocalFileFactory implements AsyncLocalBlockArchiverFactory { | ||
| private final PersistenceStorageConfig config; | ||
| private final BlockPathResolver pathResolver; | ||
|
|
||
| public AsyncBlockAsLocalFileFactory( | ||
| @NonNull final PersistenceStorageConfig config, @NonNull final BlockPathResolver pathResolver) { | ||
| this.config = Objects.requireNonNull(config); | ||
| this.pathResolver = Objects.requireNonNull(pathResolver); | ||
| } | ||
|
|
||
| @Override | ||
| public AsyncLocalBlockArchiver create(final long blockNumber) { | ||
| return new AsyncBlockAsLocalFileArchiver(blockNumber, config, pathResolver); | ||
| } | ||
| } |
7 changes: 7 additions & 0 deletions
7
...ain/java/com/hedera/block/server/persistence/storage/archive/AsyncLocalBlockArchiver.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,7 @@ | ||
| // SPDX-License-Identifier: Apache-2.0 | ||
| package com.hedera.block.server.persistence.storage.archive; | ||
|
|
||
| /** | ||
| * TODO: add documentation | ||
| */ | ||
| public interface AsyncLocalBlockArchiver extends Runnable {} |
9 changes: 9 additions & 0 deletions
9
...a/com/hedera/block/server/persistence/storage/archive/AsyncLocalBlockArchiverFactory.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,9 @@ | ||
| // SPDX-License-Identifier: Apache-2.0 | ||
| package com.hedera.block.server.persistence.storage.archive; | ||
|
|
||
| /** | ||
| * TODO: add documentation | ||
| */ | ||
| public interface AsyncLocalBlockArchiverFactory { | ||
| AsyncLocalBlockArchiver create(final long blockNumber); | ||
| } |
9 changes: 0 additions & 9 deletions
9
server/src/main/java/com/hedera/block/server/persistence/storage/archive/BlockArchiver.java
This file was deleted.
Oops, something went wrong.
Oops, something went wrong.