HDDS-12596. OM fs snapshot max limit is not enforced

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/exceptions/OMException.java

@@ -271,6 +271,8 @@ public enum ResultCodes {
     TOO_MANY_BUCKETS,
     KEY_UNDER_LEASE_RECOVERY,
     KEY_ALREADY_CLOSED,
-    KEY_UNDER_LEASE_SOFT_LIMIT_PERIOD
+    KEY_UNDER_LEASE_SOFT_LIMIT_PERIOD,
+
+    TOO_MANY_SNAPSHOTS,
   }
 }

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

@@ -345,6 +345,29 @@ public void testKeyAndSnapshotDeletionService() throws IOException, InterruptedE
     checkSnapshotIsPurgedFromDB(omFollower, tableKey);
   }
 
+  @Test
+  public void testSnapshotInFlightCount() throws Exception {
+    // snapshot inflight count should be reset to 0 when leader changes
+
+    // first do some snapshot creations
+    String snapshotName1 = UUID.randomUUID().toString();
+    store.createSnapshot(volumeName, bucketName, snapshotName1);
+
+    // then shutdown the leader
+    OzoneManager omLeader = cluster.getOMLeader();
+    cluster.shutdownOzoneManager(omLeader);
+
+    // wait for the new leader to be elected
+    cluster.waitForLeaderOM();
+
+    // check the inflight count on the new leader is 0
+    OzoneManager newLeader = cluster.getOMLeader();
+    assertEquals(0, newLeader.getOmSnapshotManager().getInFlightSnapshotCount());
+
+    // restart the previous shutdowned node
+    cluster.restartOzoneManager(omLeader, true);
+  }
+
   private void createSnapshot(String volName, String buckName, String snapName) throws IOException {
     store.createSnapshot(volName, buckName, snapName);
 

hadoop-ozone/interface-client/src/main/proto/OmClientProtocol.proto

@@ -564,6 +564,7 @@ enum Status {
     KEY_ALREADY_CLOSED = 96;
     KEY_UNDER_LEASE_SOFT_LIMIT_PERIOD = 97;
 
+    TOO_MANY_SNAPSHOTS = 98;
 }
 
 /**

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

@@ -24,6 +24,8 @@
 import static org.apache.hadoop.ozone.OzoneConsts.OM_SNAPSHOT_CHECKPOINT_DIR;
 import static org.apache.hadoop.ozone.OzoneConsts.OM_SNAPSHOT_DIFF_DB_NAME;
 import static org.apache.hadoop.ozone.OzoneConsts.OM_SNAPSHOT_INDICATOR;
+import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_FS_SNAPSHOT_MAX_LIMIT;
+import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_FS_SNAPSHOT_MAX_LIMIT_DEFAULT;
 import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_SNAPSHOT_CACHE_CLEANUP_SERVICE_RUN_INTERVAL;
 import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_SNAPSHOT_CACHE_CLEANUP_SERVICE_RUN_INTERVAL_DEFAULT;
 import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_SNAPSHOT_CACHE_MAX_SIZE;
@@ -63,6 +65,8 @@
 import java.util.Set;
 import java.util.UUID;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.stream.Collectors;
 import org.apache.commons.lang3.tuple.Pair;
 import org.apache.hadoop.hdds.StringUtils;
@@ -173,6 +177,9 @@ public final class OmSnapshotManager implements AutoCloseable {
   // Soft limit of the snapshot cache size.
   private final int softCacheSize;
 
+  private int fsSnapshotMaxLimit;
+  private final AtomicInteger inFlightSnapshotCount = new AtomicInteger(0);
+
   public OmSnapshotManager(OzoneManager ozoneManager) {
 
     boolean isFilesystemSnapshotEnabled =
@@ -247,6 +254,9 @@ public OmSnapshotManager(OzoneManager ozoneManager) {
     this.softCacheSize = ozoneManager.getConfiguration().getInt(
         OZONE_OM_SNAPSHOT_CACHE_MAX_SIZE,
         OZONE_OM_SNAPSHOT_CACHE_MAX_SIZE_DEFAULT);
+
+    fsSnapshotMaxLimit = ozoneManager.getConfiguration().getInt(OZONE_OM_FS_SNAPSHOT_MAX_LIMIT,
+        OZONE_OM_FS_SNAPSHOT_MAX_LIMIT_DEFAULT);
 
     CacheLoader<UUID, OmSnapshot> loader = createCacheLoader();
 
@@ -860,6 +870,42 @@ private void validateSnapshotsExistAndActive(final String volumeName,
     checkSnapshotActive(toSnapInfo, false);
   }
 
+  public void snapshotLimitCheck() throws IOException, OMException {
+    OmMetadataManagerImpl omMetadataManager = (OmMetadataManagerImpl) ozoneManager.getMetadataManager();
+    SnapshotChainManager snapshotChainManager = omMetadataManager.getSnapshotChainManager();
+    int currentSnapshotNum = snapshotChainManager.getGlobalSnapshotChain().size();
+
+    AtomicReference<OMException> exceptionRef = new AtomicReference<>(null);
+    inFlightSnapshotCount.updateAndGet(count -> {
+      if (currentSnapshotNum + count >= fsSnapshotMaxLimit) {
+        exceptionRef.set(new OMException(
+            String.format("Snapshot limit of %d reached. Cannot create more snapshots. " +
+                "Current snapshots: %d, In-flight creations: %d",
+                fsSnapshotMaxLimit, currentSnapshotNum, count) +
+                " If you already deleted some snapshots, " +
+                "please wait for the background service to complete the cleanup.",
+            OMException.ResultCodes.TOO_MANY_SNAPSHOTS));
+        return count;
+      }
+      return count + 1;
+    });
+    if (exceptionRef.get() != null) {
+      throw exceptionRef.get();
+    }
+  }
+
+  public void decrementInFlightSnapshotCount() {
+    inFlightSnapshotCount.decrementAndGet();
+  }
+
+  public void resetInFlightSnapshotCount() {
+    inFlightSnapshotCount.set(0);
+  }
+
+  public int getInFlightSnapshotCount() {
+    return inFlightSnapshotCount.get();
+  }
+
   private int getIndexFromToken(final String token) throws IOException {
     if (isBlank(token)) {
       return 0;

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

@@ -156,6 +156,11 @@ public SnapshotInfo getLatestSnapshot() {
     return snapshotInfo;
   }
 
+  @Override
+  public void notifyLeaderReady() {
+    ozoneManager.getOmSnapshotManager().resetInFlightSnapshotCount();
+  }
+
   @Override
   public void notifyLeaderChanged(RaftGroupMemberId groupMemberId,
                                   RaftPeerId newLeaderId) {

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

@@ -104,6 +104,8 @@ public OMSnapshotCreateRequest(OMRequest omRequest) {
   @RequireSnapshotFeatureState(true)
   public OMRequest preExecute(OzoneManager ozoneManager) throws IOException {
     final OMRequest omRequest = super.preExecute(ozoneManager);
+    // verify snapshot limit
+    ozoneManager.getOmSnapshotManager().snapshotLimitCheck();
     // Verify name
     OmUtils.validateSnapshotName(snapshotName);
     // Updating the volumeName & bucketName in case the bucket is a linked bucket. We need to do this before a
@@ -188,7 +190,7 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut
       //  pre-replicated key size counter in OmBucketInfo.
       snapshotInfo.setReferencedSize(estimateBucketDataSize(omBucketInfo));
 
-      addSnapshotInfoToSnapshotChainAndCache(omMetadataManager, context.getIndex());
+      addSnapshotInfoToSnapshotChainAndCache(ozoneManager, omMetadataManager, context.getIndex());
 
       omResponse.setCreateSnapshotResponse(
           CreateSnapshotResponse.newBuilder()
@@ -250,6 +252,7 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut
    * it was removed at T-5.
    */
   private void addSnapshotInfoToSnapshotChainAndCache(
+      OzoneManager ozoneManager,
       OmMetadataManagerImpl omMetadataManager,
       long transactionLogIndex
   ) throws IOException {
@@ -288,6 +291,8 @@ private void addSnapshotInfoToSnapshotChainAndCache(
         removeSnapshotInfoFromSnapshotChainManager(snapshotChainManager,
             snapshotInfo);
         throw new IOException(exception.getMessage(), exception);
+      } finally {
+        ozoneManager.getOmSnapshotManager().decrementInFlightSnapshotCount();
       }
     }
   }

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

@@ -32,6 +32,7 @@
 import static org.apache.hadoop.ozone.om.snapshot.OmSnapshotUtils.getINode;
 import static org.apache.hadoop.ozone.om.snapshot.OmSnapshotUtils.truncateFileName;
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
@@ -54,6 +55,7 @@
 import java.util.Arrays;
 import java.util.HashMap;
 import java.util.HashSet;
+import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
@@ -76,6 +78,7 @@
 import org.apache.ozone.test.GenericTestUtils;
 import org.apache.ozone.test.GenericTestUtils.LogCapturer;
 import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -90,6 +93,9 @@
 class TestOmSnapshotManager {
 
   private OzoneManager om;
+  private SnapshotChainManager snapshotChainManager;
+  private OmMetadataManagerImpl omMetadataManager;
+  private OmSnapshotManager omSnapshotManager;
   private static final String CANDIDATE_DIR_NAME = OM_DB_NAME +
       SNAPSHOT_CANDIDATE_DIR;
   private File leaderDir;
@@ -114,16 +120,37 @@ void init(@TempDir File tempDir) throws Exception {
         OMConfigKeys.OZONE_OM_SNAPSHOT_CACHE_MAX_SIZE, 1);
     configuration.setBoolean(
         OMConfigKeys.OZONE_OM_SNAPSHOT_ROCKSDB_METRICS_ENABLED, false);
+    // Allow 2 fs snapshots
+    configuration.setInt(
+        OMConfigKeys.OZONE_OM_FS_SNAPSHOT_MAX_LIMIT, 2);
 
     OmTestManagers omTestManagers = new OmTestManagers(configuration);
     om = omTestManagers.getOzoneManager();
+    omMetadataManager = (OmMetadataManagerImpl) om.getMetadataManager();
+    omSnapshotManager = om.getOmSnapshotManager();
+    snapshotChainManager = omMetadataManager.getSnapshotChainManager();
   }
 
   @AfterAll
   void stop() {
     om.stop();
   }
 
+  @AfterEach
+  void cleanup() throws IOException {
+    Table<String, SnapshotInfo> snapshotInfoTable = omMetadataManager.getSnapshotInfoTable();
+
+    Iterator<UUID> iter = snapshotChainManager.iterator(true);
+    while (iter.hasNext()) {
+      UUID snapshotId = iter.next();
+      String snapshotInfoKey = snapshotChainManager.getTableKey(snapshotId);
+      SnapshotInfo snapshotInfo = snapshotInfoTable.get(snapshotInfoKey);
+      snapshotChainManager.deleteSnapshot(snapshotInfo);
+      snapshotInfoTable.delete(snapshotInfoKey);
+    }
+    omSnapshotManager.invalidateCache();
+  }
+
   @Test
   public void testSnapshotFeatureFlagSafetyCheck() throws IOException {
     // Verify that the snapshot feature config safety check method
@@ -151,11 +178,11 @@ public void testCloseOnEviction() throws IOException,
     Table<String, OmBucketInfo> bucketTable = mock(Table.class);
     Table<String, SnapshotInfo> snapshotInfoTable = mock(Table.class);
     HddsWhiteboxTestUtils.setInternalState(
-        om.getMetadataManager(), VOLUME_TABLE, volumeTable);
+        omMetadataManager, VOLUME_TABLE, volumeTable);
     HddsWhiteboxTestUtils.setInternalState(
-        om.getMetadataManager(), BUCKET_TABLE, bucketTable);
+        omMetadataManager, BUCKET_TABLE, bucketTable);
     HddsWhiteboxTestUtils.setInternalState(
-        om.getMetadataManager(), SNAPSHOT_INFO_TABLE, snapshotInfoTable);
+        omMetadataManager, SNAPSHOT_INFO_TABLE, snapshotInfoTable);
 
     final String volumeName = UUID.randomUUID().toString();
     final String dbVolumeKey = om.getMetadataManager().getVolumeKey(volumeName);
@@ -185,16 +212,15 @@ public void testCloseOnEviction() throws IOException,
     when(snapshotInfoTable.get(first.getTableKey())).thenReturn(first);
     when(snapshotInfoTable.get(second.getTableKey())).thenReturn(second);
 
-    ((OmMetadataManagerImpl) om.getMetadataManager()).getSnapshotChainManager().addSnapshot(first);
-    ((OmMetadataManagerImpl) om.getMetadataManager()).getSnapshotChainManager().addSnapshot(second);
+    snapshotChainManager.addSnapshot(first);
+    snapshotChainManager.addSnapshot(second);
     RDBBatchOperation rdbBatchOperation = new RDBBatchOperation();
     // create the first snapshot checkpoint
     OmSnapshotManager.createOmSnapshotCheckpoint(om.getMetadataManager(),
         first, rdbBatchOperation);
     om.getMetadataManager().getStore().commitBatchOperation(rdbBatchOperation);
 
     // retrieve it and setup store mock
-    OmSnapshotManager omSnapshotManager = om.getOmSnapshotManager();
     OmSnapshot firstSnapshot = omSnapshotManager
         .getActiveSnapshot(first.getVolumeName(), first.getBucketName(), first.getName())
         .get();
@@ -230,6 +256,37 @@ public void testCloseOnEviction() throws IOException,
     }, 100, 30_000);
   }
 
+  @Test
+  public void testValidateSnapshotLimit() throws IOException {
+    Table<String, SnapshotInfo> snapshotInfoTable = mock(Table.class);
+    HddsWhiteboxTestUtils.setInternalState(
+        omMetadataManager, SNAPSHOT_INFO_TABLE, snapshotInfoTable);
+
+    SnapshotInfo first = createSnapshotInfo("vol1", "buck1");
+    SnapshotInfo second = createSnapshotInfo("vol1", "buck1");
+
+    first.setGlobalPreviousSnapshotId(null);
+    first.setPathPreviousSnapshotId(null);
+    second.setGlobalPreviousSnapshotId(first.getSnapshotId());
+    second.setPathPreviousSnapshotId(first.getSnapshotId());
+
+    when(snapshotInfoTable.get(first.getTableKey())).thenReturn(first);
+    when(snapshotInfoTable.get(second.getTableKey())).thenReturn(second);
+
+    snapshotChainManager.addSnapshot(first);
+    assertDoesNotThrow(() -> omSnapshotManager.snapshotLimitCheck());
+    omSnapshotManager.decrementInFlightSnapshotCount();
+
+    snapshotChainManager.addSnapshot(second);
+
+    OMException exception = assertThrows(OMException.class, () -> omSnapshotManager.snapshotLimitCheck());
+    assertEquals(OMException.ResultCodes.TOO_MANY_SNAPSHOTS, exception.getResult());
+
+    snapshotChainManager.deleteSnapshot(second);
+
+    assertDoesNotThrow(() -> omSnapshotManager.snapshotLimitCheck());
+  }
+
   @BeforeEach
   void setupData(@TempDir File testDir) throws IOException {
     // Set up the leader with the following files:
@@ -334,16 +391,12 @@ public void testHardLinkCreation() throws IOException {
   @Test
   public void testGetSnapshotInfo() throws IOException {
     SnapshotInfo s1 = createSnapshotInfo("vol", "buck");
-    UUID latestGlobalSnapId =
-        ((OmMetadataManagerImpl) om.getMetadataManager()).getSnapshotChainManager()
-            .getLatestGlobalSnapshotId();
+    UUID latestGlobalSnapId = snapshotChainManager.getLatestGlobalSnapshotId();
     UUID latestPathSnapId =
-        ((OmMetadataManagerImpl) om.getMetadataManager()).getSnapshotChainManager()
-            .getLatestPathSnapshotId(String.join("/", "vol", "buck"));
+        snapshotChainManager.getLatestPathSnapshotId(String.join("/", "vol", "buck"));
     s1.setPathPreviousSnapshotId(latestPathSnapId);
     s1.setGlobalPreviousSnapshotId(latestGlobalSnapId);
-    ((OmMetadataManagerImpl) om.getMetadataManager()).getSnapshotChainManager()
-        .addSnapshot(s1);
+    snapshotChainManager.addSnapshot(s1);
     OMException ome = assertThrows(OMException.class,
         () -> om.getOmSnapshotManager().getSnapshot(s1.getSnapshotId()));
     assertEquals(OMException.ResultCodes.FILE_NOT_FOUND, ome.getResult());
@@ -352,6 +405,12 @@ public void testGetSnapshotInfo() throws IOException {
     ome = assertThrows(OMException.class,
         () -> om.getOmSnapshotManager().getSnapshot(s2.getSnapshotId()));
     assertEquals(OMException.ResultCodes.FILE_NOT_FOUND, ome.getResult());
+
+    // add to make cleanup work
+    Table<String, SnapshotInfo> snapshotInfoTable = mock(Table.class);
+    HddsWhiteboxTestUtils.setInternalState(
+        omMetadataManager, SNAPSHOT_INFO_TABLE, snapshotInfoTable);
+    when(snapshotInfoTable.get(s1.getTableKey())).thenReturn(s1);
   }
   /*
    * Test that exclude list is generated correctly.