vaijosh commented on code in PR #3081: URL: https://github.com/apache/hugegraph/pull/3081#discussion_r3576434649
########## hugegraph-store/hg-store-node/src/main/java/org/apache/hugegraph/store/node/cloud/CloudStorageEventListener.java: ########## @@ -0,0 +1,445 @@ +/* + * 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 org.apache.hugegraph.store.node.cloud; + +import java.io.File; +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.stream.Stream; + +import org.apache.hugegraph.rocksdb.access.RocksDBFactory; +import org.apache.hugegraph.rocksdb.access.RocksDBFactory.RocksdbChangedListener; +import org.apache.hugegraph.rocksdb.access.RocksDBSession; +import org.apache.hugegraph.store.cloud.CloudStorageProvider; +import org.apache.hugegraph.store.cloud.CloudStorageProviderFactory; + +import lombok.extern.slf4j.Slf4j; + +/** + * {@link RocksdbChangedListener} that bridges RocksDB table-file lifecycle events + * to the active {@link CloudStorageProvider}. + * + * <p>When cloud storage is enabled: + * <ul> + * <li>{@link #onDBCreated} uploads any SST files that already exist in the DB directory + * (e.g. surviving from a previous run) and triggers an async MemTable flush so that + * WAL-recovered or recently-written data is also written to SST files.</li> + * <li>{@link #onTableFileCreated} uploads newly created SST files.</li> + * <li>{@link #onTableFileDeleted} removes the corresponding object from cloud storage.</li> + * </ul> + * + * <h3>Remote key construction</h3> + * The remote key is derived by stripping the {@code dataRoot} prefix from the absolute + * local file path. This keeps the object layout clean and independent of the container + * filesystem layout: + * <pre> + * dataRoot = /hugegraph-store/storage + * filePath = /hugegraph-store/storage/hgstore-metadata/000008.sst + * remoteKey = hgstore-metadata/000008.sst + * (with path-prefix "hugegraph") → hugegraph/hgstore-metadata/000008.sst + * </pre> + * + * This listener is registered with {@link RocksDBFactory} during application startup + * (see {@link org.apache.hugegraph.store.node.AppConfig}). + */ +@Slf4j +public class CloudStorageEventListener implements RocksdbChangedListener { + + /** Absolute, normalised path of the store's data root directory. */ + private final String dataRoot; + + private static final long DEFAULT_READ_MISS_GUARD_WINDOW_MS = 3000L; + + private final boolean startupHydrationEnabled; + private final long readMissGuardWindowMs; + private final Map<String, Long> readMissAttemptTs; + + /** + * Optional retry queue; when non-null, upload failures are submitted here instead + * of just being logged. When null, failures are only logged (no retry). + */ + private final CloudUploadRetryQueue retryQueue; + + /** + * @param dataRoot absolute path of the store's data directory + * (value of {@code app.data-path}, resolved to an absolute path). + */ + public CloudStorageEventListener(String dataRoot) { + this(dataRoot, true, DEFAULT_READ_MISS_GUARD_WINDOW_MS, null); + } + + public CloudStorageEventListener(String dataRoot, + boolean startupHydrationEnabled) { + this(dataRoot, startupHydrationEnabled, DEFAULT_READ_MISS_GUARD_WINDOW_MS, null); + } + + /** + * @param readMissGuardWindowMs guard window in ms for repeated read-miss hydration attempts + * for the same db/table pair (cloud.storage.read-miss-guard-window-ms) + */ + public CloudStorageEventListener(String dataRoot, + boolean startupHydrationEnabled, + long readMissGuardWindowMs) { + this(dataRoot, startupHydrationEnabled, readMissGuardWindowMs, null); + } + + /** + * Full constructor. + * + * @param retryQueue optional {@link CloudUploadRetryQueue}; when non-null, upload failures + * are retried asynchronously and eventually moved to the dead-letter queue. + * Pass {@code null} to disable retries (failures are only logged). + */ + public CloudStorageEventListener(String dataRoot, + boolean startupHydrationEnabled, + long readMissGuardWindowMs, + CloudUploadRetryQueue retryQueue) { + String normalised = Paths.get(dataRoot).toAbsolutePath().normalize().toString(); + // Strip trailing separator so substring arithmetic is consistent. + this.dataRoot = normalised.endsWith(File.separator) + ? normalised.substring(0, normalised.length() - 1) + : normalised; + this.startupHydrationEnabled = startupHydrationEnabled; + this.readMissGuardWindowMs = Math.max(0L, readMissGuardWindowMs); + this.readMissAttemptTs = new ConcurrentHashMap<>(); + this.retryQueue = retryQueue; + } + + // ----------------------------------------------------------------------- + // RocksdbChangedListener + // ----------------------------------------------------------------------- + + @Override + public void onDBOpening(String dbName, String dbPath) { + if (!startupHydrationEnabled) { + return; + } + CloudStorageProvider provider = CloudStorageProviderFactory.getActiveProvider(); + if (provider == null) { + return; + } + preHydrateDbFiles(provider, dbName, dbPath); + } + + /** + * Called when a read returns null in RocksDB. We try to hydrate missing SST files from cloud, + * ingest them into the target CF, then caller retries get(). + */ + @Override + public boolean onReadMiss(RocksDBSession session, String table, byte[] key) { + if (!shouldAttemptReadMissHydration(session.getGraphName(), table)) { + return false; + } + CloudStorageProvider provider = CloudStorageProviderFactory.getActiveProvider(); + if (provider == null) { + return false; + } + List<String> downloaded = downloadMissingSstFiles(provider, + session.getGraphName(), + session.getDbPath()); + if (downloaded.isEmpty()) { + return false; + } + try { + Map<byte[], List<String>> sstByCf = new HashMap<>(); + sstByCf.put(table.getBytes(StandardCharsets.UTF_8), downloaded); + session.ingestSstFile(sstByCf); + log.info("Cloud read-miss hydration succeeded: db={}, table={}, files={}", + session.getGraphName(), table, downloaded.size()); + return true; + } catch (Exception e) { + log.warn("Cloud read-miss hydration failed: db={}, table={}, reason={}", + session.getGraphName(), table, e.getMessage()); + return false; + } + } + + /** + * Called when a new RocksDB instance is opened for the first time. + * + * <p>Uploads any SST files that already exist in {@code dbPath} (e.g. from a previous run) + * and then triggers a MemTable flush so that WAL-recovered data is also written to + * SST files and eventually forwarded here via {@link #onTableFileCreated}. + * + * @param dbName logical name of the graph / partition + * @param dbPath absolute path of the RocksDB directory + */ + @Override + public void onDBCreated(String dbName, String dbPath) { + CloudStorageProvider provider = CloudStorageProviderFactory.getActiveProvider(); + if (provider == null) { + return; + } + uploadExistingSstFiles(provider, dbName, dbPath); + flushDb(dbName); + } + + /** + * Uploads the newly created SST file to the active cloud storage provider. + * + * @param dbName RocksDB instance name (partition id) + * @param cfName column-family name + * @param filePath absolute local path of the new SST file + * @param fileSize file size in bytes (informational) + */ + @Override + public void onTableFileCreated(String dbName, String cfName, + String filePath, long fileSize) { + CloudStorageProvider provider = CloudStorageProviderFactory.getActiveProvider(); + if (provider == null) { + return; + } + String remoteKey = toRelativeKey(filePath); + try { + provider.uploadFile(filePath, remoteKey); + log.debug("Cloud upload success: db={}, cf={}, path={}, size={}", + dbName, cfName, filePath, fileSize); + } catch (Exception e) { + // NOTE: this callback is invoked via RocksDB's JNI event-listener mechanism. + // Any exception thrown here crosses the JNI boundary and is silently swallowed + // by the native layer — it will NOT crash the server and the SST file will NOT + // be retried automatically. Log the failure and submit to the retry queue (if + // configured) so the upload can be retried asynchronously and, after exhausting + // all attempts, moved to the dead-letter queue for later inspection / replay. + log.error("Cloud upload failed (SST file is local-only, may be missing from cloud): " + + "db={}, cf={}, path={}", dbName, cfName, filePath, e); + if (retryQueue != null) { + retryQueue.submit(dbName, cfName, filePath, remoteKey, e); + } + } + } + + /** + * Removes the deleted SST file from the active cloud storage provider. + * + * @param dbName RocksDB instance name (partition id) + * @param cfName column-family name + * @param filePath absolute local path of the deleted SST file + */ + @Override + public void onTableFileDeleted(String dbName, String cfName, String filePath) { + CloudStorageProvider provider = CloudStorageProviderFactory.getActiveProvider(); + if (provider == null) { + return; + } + String remoteKey = toRelativeKey(filePath); + try { + provider.deleteFile(remoteKey); + log.debug("Cloud delete success: db={}, cf={}, path={}", dbName, cfName, filePath); + } catch (Exception e) { + // Non-fatal: log and continue. + log.error("Cloud delete failed: db={}, cf={}, path={}", dbName, cfName, filePath, e); + } + } + + // ----------------------------------------------------------------------- + // Internal helpers + // ----------------------------------------------------------------------- + + /** + * Converts an absolute local file path to a remote key by stripping the data-root prefix. + * + * <pre> + * dataRoot = /hugegraph-store/storage + * filePath = /hugegraph-store/storage/hgstore-metadata/000008.sst + * result = hgstore-metadata/000008.sst + * </pre> + * + * If {@code filePath} does not start with {@code dataRoot} the leading slash is simply + * stripped so the key is still valid (though possibly not ideallyformatted). + */ + String toRelativeKey(String filePath) { + if (filePath.startsWith(dataRoot)) { + String rel = filePath.substring(dataRoot.length()); + // Strip leading separator produced by the substring. + return rel.startsWith("/") || rel.startsWith(File.separator) + ? rel.substring(1) + : rel; + } + // Fallback: strip any leading slash so the key does not start with '/'. + return filePath.startsWith("/") ? filePath.substring(1) : filePath; + } + + /** + * Walks {@code dbPath} and uploads every {@code *.sst} file that is not already + * present in cloud storage. This handles restarts where SST files from a previous + * run were never uploaded (e.g. cloud storage was enabled after the last shutdown). + */ + private void uploadExistingSstFiles(CloudStorageProvider provider, String dbName, + String dbPath) { + Path root = Paths.get(dbPath); + if (!root.toFile().isDirectory()) { + return; + } + try (Stream<Path> paths = Files.walk(root)) { + paths.filter(p -> p.toString().endsWith(".sst")) Review Comment: Thanks @VGalaxies for sharing the feedback.In recent commit I have implemented this. To be frank earlier it was planned but I was working on some additional resiliancy mechanism to make sure that DB is consistent even during Sync failures and cluster crash scenarios. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
