vaijosh commented on code in PR #3081: URL: https://github.com/apache/hugegraph/pull/3081#discussion_r3586740119
########## 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); Review Comment: Thanks @VGalaxies. I have addressed it in commit #32e407e -- 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]
