HDDS-7935. [Snapshot] LRU Cache entries may get evicted/closed during long running processes

hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/OzoneConfigKeys.java

@@ -606,6 +606,7 @@ public final class OzoneConfigKeys {
       "org.apache.hadoop.ozone.om.TrashPolicyOzone";
 
 
+  // TODO: [SNAPSHOT] Document this in ozone-default.xml
   public static final String OZONE_OM_SNAPSHOT_CACHE_MAX_SIZE =
       "ozone.om.snapshot.cache.max.size";
   public static final int OZONE_OM_SNAPSHOT_CACHE_MAX_SIZE_DEFAULT = 10;

hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestOmSnapshotFileSystem.java

@@ -39,7 +39,6 @@
 import org.apache.hadoop.ozone.client.OzoneVolume;
 import org.apache.hadoop.ozone.client.io.OzoneInputStream;
 import org.apache.hadoop.ozone.client.io.OzoneOutputStream;
-import org.apache.hadoop.ozone.om.exceptions.OMException;
 import org.apache.hadoop.ozone.om.helpers.BucketLayout;
 import org.apache.hadoop.ozone.om.helpers.OmKeyArgs;
 import org.apache.hadoop.ozone.om.helpers.OpenKeySession;
@@ -59,6 +58,7 @@
 import org.slf4j.LoggerFactory;
 
 import java.io.File;
+import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.net.URI;
 import java.nio.ByteBuffer;
@@ -411,15 +411,19 @@ public void testBlockSnapshotFSAccessAfterDeletion() throws Exception {
     deleteSnapshot(snapshotName);
 
     // Can't access keys in snapshot anymore with FS API. Should throw exception
-    final String errorMsg = "no longer active";
-    LambdaTestUtils.intercept(OMException.class, errorMsg,
+    final String errorMsg1 = "no longer active";
+    LambdaTestUtils.intercept(FileNotFoundException.class, errorMsg1,
         () -> o3fs.listStatus(snapshotRoot));
-    LambdaTestUtils.intercept(OMException.class, errorMsg,
+    LambdaTestUtils.intercept(FileNotFoundException.class, errorMsg1,
         () -> o3fs.listStatus(snapshotParent));
 
-    LambdaTestUtils.intercept(OMException.class, errorMsg,
+    // Note: Different error message due to inconsistent FNFE client-side
+    //  handling in BasicOzoneClientAdapterImpl#getFileStatus
+    // TODO: Reconciliation?
+    final String errorMsg2 = "No such file or directory";
+    LambdaTestUtils.intercept(FileNotFoundException.class, errorMsg2,
         () -> o3fs.getFileStatus(snapshotKey1));
-    LambdaTestUtils.intercept(OMException.class, errorMsg,
+    LambdaTestUtils.intercept(FileNotFoundException.class, errorMsg2,
         () -> o3fs.getFileStatus(snapshotKey2));
   }
 

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OmSnapshot.java

@@ -260,6 +260,16 @@ public void close() throws IOException {
     omMetadataManager.getStore().close();
   }
 
+  @Override
+  protected void finalize() throws Throwable {
+    // Close the DB if it hasn't been closed when this OmSnapshot instance is
+    // garbage-collected by the JVM to avoid handle leaks.
+    if (!omMetadataManager.getStore().isClosed()) {
+      omMetadataManager.getStore().close();
+    }
+    super.finalize();
+  }
+
   @VisibleForTesting
   public OMMetadataManager getMetadataManager() {
     return omMetadataManager;

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OmSnapshotManager.java

@@ -150,6 +150,9 @@ public final class OmSnapshotManager implements AutoCloseable {
   private final CodecRegistry codecRegistry;
   private final SnapshotDiffCleanupService snapshotDiffCleanupService;
 
+  // Soft limit of the snapshot cache size.
+  private final int softCacheSize;
+
   public OmSnapshotManager(OzoneManager ozoneManager) {
     this.options = new ManagedDBOptions();
     this.options.setCreateIfMissing(true);
@@ -194,29 +197,38 @@ public OmSnapshotManager(OzoneManager ozoneManager) {
         .getStore()
         .getRocksDBCheckpointDiffer();
 
-    // size of lru cache
-    int cacheSize = ozoneManager.getConfiguration().getInt(
+    // snapshot cache size, soft-limit
+    this.softCacheSize = ozoneManager.getConfiguration().getInt(
         OzoneConfigKeys.OZONE_OM_SNAPSHOT_CACHE_MAX_SIZE,
         OzoneConfigKeys.OZONE_OM_SNAPSHOT_CACHE_MAX_SIZE_DEFAULT);
 
     CacheLoader<String, OmSnapshot> loader = createCacheLoader();
 
-    RemovalListener<String, OmSnapshot> removalListener
-        = notification -> {
-          try {
-            // close snapshot's rocksdb on eviction
-            LOG.debug("Closing snapshot: {}", notification.getKey());
-            // TODO: [SNAPSHOT] HDDS-7935.Close only when refcount reaches zero?
-            notification.getValue().close();
-          } catch (IOException e) {
-            LOG.error("Failed to close snapshot: {} {}",
-                notification.getKey(), e);
-          }
-        };
+    RemovalListener<String, OmSnapshot> removalListener = notification -> {
+      try {
+        final String snapshotTableKey = notification.getKey();
+        final OmSnapshot omSnapshot = notification.getValue();
+        if (omSnapshot != null) {
+          // close snapshot's rocksdb on eviction
+          LOG.debug("Closing OmSnapshot '{}'", snapshotTableKey);
+          omSnapshot.close();
+        } else {
+          // Assuming the value becomes null when soft ref is GC'ed by JVM.
+          LOG.debug("OmSnapshot '{}' was already garbage collected by JVM.",
+              snapshotTableKey);
+        }
+      } catch (IOException e) {
+        LOG.error("Failed to close snapshot: {}", notification.getKey(), e);
+      }
+    };
 
     // init LRU cache
-    snapshotCache = CacheBuilder.newBuilder()
-        .maximumSize(cacheSize)
+    this.snapshotCache = CacheBuilder.newBuilder()
+        // Indicating OmSnapshot instances are softly referenced from the cache.
+        // If no thread is holding a strong reference to an OmSnapshot instance
+        // (e.g. SnapDiff), the instance could be garbage collected by JVM at
+        // its discretion.
+        .softValues()
         .removalListener(removalListener)
         .build(loader);
 
@@ -241,6 +253,30 @@ public OmSnapshotManager(OzoneManager ozoneManager) {
     this.snapshotDiffCleanupService.start();
   }
 
+  /**
+   * Throws OMException FILE_NOT_FOUND if snapshot is not in active status.
+   * @param snapshotTableKey snapshot table key
+   */
+  public void checkSnapshotActive(String snapshotTableKey) throws IOException {
+    checkSnapshotActive(getSnapshotInfo(snapshotTableKey));
+  }
+
+  private void checkSnapshotActive(SnapshotInfo snapInfo) throws OMException {
+
+    if (!snapInfo.getSnapshotStatus().equals(SnapshotStatus.SNAPSHOT_ACTIVE)) {
+      if (isCalledFromSnapshotDeletingService()) {
+        LOG.debug("Permitting {} to load snapshot {} even in status: {}",
+            SnapshotDeletingService.class.getSimpleName(),
+            snapInfo.getTableKey(),
+            snapInfo.getSnapshotStatus());
+      } else {
+        throw new OMException("Unable to load snapshot. " +
+            "Snapshot with table key '" + snapInfo.getTableKey() +
+            "' is no longer active", FILE_NOT_FOUND);
+      }
+    }
+  }
+
   private CacheLoader<String, OmSnapshot> createCacheLoader() {
     return new CacheLoader<String, OmSnapshot>() {
       @Override
@@ -249,31 +285,12 @@ private CacheLoader<String, OmSnapshot> createCacheLoader() {
       @Nonnull
       public OmSnapshot load(@Nonnull String snapshotTableKey)
           throws IOException {
-        SnapshotInfo snapshotInfo;
         // see if the snapshot exists
-        snapshotInfo = getSnapshotInfo(snapshotTableKey);
+        SnapshotInfo snapshotInfo = getSnapshotInfo(snapshotTableKey);
 
         // Block snapshot from loading when it is no longer active e.g. DELETED,
         // unless this is called from SnapshotDeletingService.
-        // TODO: [SNAPSHOT] However, snapshotCache.get() from other requests
-        //  (not from SDS) would be able to piggyback off of this because
-        //  snapshot still in cache won't trigger loader again.
-        //  This needs proper addressal in e.g. HDDS-7935
-        //  by introducing another cache just for SDS.
-        //  While the snapshotCache would host ACTIVE snapshots only.
-        if (!snapshotInfo.getSnapshotStatus().equals(
-            SnapshotStatus.SNAPSHOT_ACTIVE)) {
-          if (isCalledFromSnapshotDeletingService()) {
-            LOG.debug("Permitting {} to load snapshot {} in status: {}",
-                SnapshotDeletingService.class.getSimpleName(),
-                snapshotInfo.getTableKey(),
-                snapshotInfo.getSnapshotStatus());
-          } else {
-            throw new OMException("Unable to load snapshot. " +
-                "Snapshot with table key '" + snapshotTableKey +
-                "' is no longer active", FILE_NOT_FOUND);
-          }
-        }
+        checkSnapshotActive(snapshotInfo);
 
         CacheValue<SnapshotInfo> cacheValue =
             ozoneManager.getMetadataManager().getSnapshotInfoTable()
@@ -292,7 +309,7 @@ public OmSnapshot load(@Nonnull String snapshotTableKey)
           snapshotMetadataManager = new OmMetadataManagerImpl(conf,
               snapshotInfo.getCheckpointDirName(), isSnapshotInCache);
         } catch (IOException e) {
-          LOG.error("Failed to retrieve snapshot: {}, {}", snapshotTableKey, e);
+          LOG.error("Failed to retrieve snapshot: {}", snapshotTableKey, e);
           throw e;
         }
 
@@ -433,7 +450,7 @@ private static void deleteKeysInSnapshotScopeFromDTableInternal(
 
     try (TableIterator<String,
         ? extends Table.KeyValue<String, RepeatedOmKeyInfo>>
-        keyIter = omMetadataManager.getDeletedTable().iterator()) {
+             keyIter = omMetadataManager.getDeletedTable().iterator()) {
 
       keyIter.seek(beginKey);
       // Continue only when there are entries of snapshot (bucket) scope
@@ -500,6 +517,15 @@ public IOmMetadataReader checkForSnapshot(String volumeName,
       String snapshotTableKey = SnapshotInfo.getTableKey(volumeName,
           bucketName, snapshotName);
 
+      // Block FS API reads when snapshot is not active.
+      checkSnapshotActive(snapshotTableKey);
+
+      // Warn if actual cache size exceeds the soft limit already.
+      if (snapshotCache.size() > softCacheSize) {
+        LOG.warn("Snapshot cache size ({}) exceeds configured soft-limit ({}).",
+            snapshotCache.size(), softCacheSize);
+      }
+
       // retrieve the snapshot from the cache
       try {
         return snapshotCache.get(snapshotTableKey);
@@ -521,7 +547,7 @@ public static String getSnapshotPrefix(String snapshotName) {
   }
 
   public static String getSnapshotPath(OzoneConfiguration conf,
-                                     SnapshotInfo snapshotInfo) {
+                                       SnapshotInfo snapshotInfo) {
     return OMStorage.getOmDbDir(conf) +
         OM_KEY_PREFIX + OM_SNAPSHOT_CHECKPOINT_DIR + OM_KEY_PREFIX +
         OM_DB_NAME + snapshotInfo.getCheckpointDirName();
@@ -558,10 +584,14 @@ public SnapshotDiffResponse getSnapshotDiffReport(final String volume,
     final String fsKey = SnapshotInfo.getTableKey(volume, bucket, fromSnapshot);
     final String tsKey = SnapshotInfo.getTableKey(volume, bucket, toSnapshot);
     try {
+      // Block SnapDiff if either one of the snapshot is not active.
+      checkSnapshotActive(fsKey);
+      checkSnapshotActive(tsKey);
+
       final OmSnapshot fs = snapshotCache.get(fsKey);
       final OmSnapshot ts = snapshotCache.get(tsKey);
       return snapshotDiffManager.getSnapshotDiffReport(volume, bucket, fs, ts,
-              fsInfo, tsInfo, index, pageSize, forceFullDiff);
+          fsInfo, tsInfo, index, pageSize, forceFullDiff);
     } catch (ExecutionException e) {
       throw new IOException(e.getCause());
     }

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/snapshot/OMSnapshotDeleteRequest.java

@@ -184,10 +184,7 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager,
       omClientResponse = new OMSnapshotDeleteResponse(
           omResponse.build(), tableKey, snapshotInfo);
 
-      // Evict the snapshot entry from cache, and close the snapshot DB
-      // Nothing happens if the key doesn't exist in cache (snapshot not loaded)
-      ozoneManager.getOmSnapshotManager().getSnapshotCache()
-          .invalidate(tableKey);
+      // No longer need to invalidate the entry in the snapshot cache here.
 
     } catch (IOException ex) {
       exception = ex;

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/snapshot/SnapshotDiffManager.java

@@ -1055,6 +1055,11 @@ private synchronized void loadJobOnStartUp(final String jobKey,
 
     String fsKey = SnapshotInfo.getTableKey(volume, bucket, fromSnapshot);
     String tsKey = SnapshotInfo.getTableKey(volume, bucket, toSnapshot);
+
+    // Block SnapDiff if either one of the snapshots is not active
+    ozoneManager.getOmSnapshotManager().checkSnapshotActive(fsKey);
+    ozoneManager.getOmSnapshotManager().checkSnapshotActive(tsKey);
+
     try {
       submitSnapDiffJob(jobKey, jobId, volume, bucket, snapshotCache.get(fsKey),
           snapshotCache.get(tsKey), fsInfo, tsInfo, forceFullDiff);

hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/TestOmSnapshotManager.java

@@ -123,6 +123,11 @@ public void testCloseOnEviction() throws IOException {
         .checkForSnapshot(second.getVolumeName(),
         second.getBucketName(), getSnapshotPrefix(second.getName()));
 
+    // As a workaround, invalidate all cache entries in order to trigger
+    // instances close in this test case, since JVM GC most likely would not
+    // have triggered and closed the instances yet at this point.
+    omSnapshotManager.getSnapshotCache().invalidateAll();
+
     // confirm store was closed
     verify(firstSnapshotStore, timeout(3000).times(1)).close();
   }