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

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

@@ -479,6 +479,7 @@ public final class OzoneManager extends ServiceRuntimeInfoImpl
   private final BucketUtilizationMetrics bucketUtilizationMetrics;
 
   private boolean fsSnapshotEnabled;
+  private int fsSnapshotMaxLimit;
 
   private String omHostName;
 
@@ -988,6 +989,9 @@ private void instantiateServices(boolean withNewSnapshot) throws IOException {
     fsSnapshotEnabled = configuration.getBoolean(
         OMConfigKeys.OZONE_FILESYSTEM_SNAPSHOT_ENABLED_KEY,
         OMConfigKeys.OZONE_FILESYSTEM_SNAPSHOT_ENABLED_DEFAULT);
+    fsSnapshotMaxLimit = configuration.getInt(
+        OMConfigKeys.OZONE_OM_FS_SNAPSHOT_MAX_LIMIT,
+        OMConfigKeys.OZONE_OM_FS_SNAPSHOT_MAX_LIMIT_DEFAULT);
     omSnapshotManager = new OmSnapshotManager(this);
 
     // Snapshot metrics
@@ -4365,6 +4369,15 @@ public boolean isFilesystemSnapshotEnabled() {
     return fsSnapshotEnabled;
   }
 
+  /**
+   * Get the maximum number of Ozone filesystem snapshots allowed.
+   *
+   * @return the maximum number of Ozone filesystem snapshots allowed.
+   */
+  public int getFsSnapshotMaxLimit() {
+    return fsSnapshotMaxLimit;
+  }
+
   /**
    * Get DB updates since a specific sequence number.
    *

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

@@ -165,6 +165,14 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut
         throw new OMException("Snapshot already exists", FILE_ALREADY_EXISTS);
       }
 
+      // Check snapshot limit
+      int maxSnapshots = ozoneManager.getFsSnapshotMaxLimit();
+      if (omMetrics.getNumSnapshotActive() >= maxSnapshots) {
+        throw new OMException(
+            String.format("Snapshot limit of %d reached. Cannot create more snapshots.", maxSnapshots),
+            OMException.ResultCodes.INVALID_SNAPSHOT_ERROR);
+      }
+
       // Note down RDB latest transaction sequence number, which is used
       // as snapshot generation in the Differ.
       final long dbLatestSequenceNumber =

hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/key/TestOMKeyRequest.java

@@ -17,16 +17,17 @@
 
 package org.apache.hadoop.ozone.om.request.key;
 
+import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_FS_SNAPSHOT_MAX_LIMIT_DEFAULT;
 import static org.apache.hadoop.ozone.om.request.OMRequestTestUtils.setupReplicationConfigValidation;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertNull;
 import static org.junit.jupiter.api.Assertions.assertTrue;
-import static org.mockito.Mockito.any;
-import static org.mockito.Mockito.anyInt;
-import static org.mockito.Mockito.anyLong;
-import static org.mockito.Mockito.anyString;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.Mockito.doNothing;
 import static org.mockito.Mockito.framework;
 import static org.mockito.Mockito.mock;
@@ -153,6 +154,7 @@ public void setup() throws Exception {
     when(ozoneManager.getDeletionMetrics()).thenReturn(delMetrics);
     when(ozoneManager.getMetadataManager()).thenReturn(omMetadataManager);
     when(ozoneManager.getConfiguration()).thenReturn(ozoneConfiguration);
+    when(ozoneManager.getFsSnapshotMaxLimit()).thenReturn(OZONE_OM_FS_SNAPSHOT_MAX_LIMIT_DEFAULT);
     OMLayoutVersionManager lvm = mock(OMLayoutVersionManager.class);
     when(lvm.isAllowed(anyString())).thenReturn(true);
     when(ozoneManager.getVersionManager()).thenReturn(lvm);

hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/snapshot/TestOMSnapshotCreateRequest.java

@@ -28,7 +28,7 @@
 import static org.junit.jupiter.api.Assertions.assertNull;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
-import static org.mockito.Mockito.any;
+import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.when;
 
 import java.io.IOException;
@@ -264,6 +264,39 @@ public void testEntryExists() throws Exception {
     assertEquals(2, getOmMetrics().getNumSnapshotCreates());
   }
 
+  @Test
+  public void testSnapshotLimit() throws Exception {
+    when(getOzoneManager().isAdmin(any())).thenReturn(true);
+
+    when(getOzoneManager().getFsSnapshotMaxLimit()).thenReturn(1);
+
+    String key1 = getTableKey(getVolumeName(), getBucketName(), snapshotName1);
+
+    OMRequest omRequest =
+        createSnapshotRequest(getVolumeName(), getBucketName(), snapshotName1);
+    OMSnapshotCreateRequest omSnapshotCreateRequest = doPreExecute(omRequest);
+
+    assertNull(getOmMetadataManager().getSnapshotInfoTable().get(key1));
+    omSnapshotCreateRequest.validateAndUpdateCache(getOzoneManager(), 1);
+
+    assertNotNull(getOmMetadataManager().getSnapshotInfoTable().get(key1));
+
+    // Should fail as snapshot limit is 1
+    omRequest = createSnapshotRequest(getVolumeName(), getBucketName(), snapshotName2);
+    omSnapshotCreateRequest = doPreExecute(omRequest);
+    OMClientResponse omClientResponse =
+        omSnapshotCreateRequest.validateAndUpdateCache(getOzoneManager(), 2);
+
+    OMResponse omResponse = omClientResponse.getOMResponse();
+    assertNotNull(omResponse.getCreateSnapshotResponse());
+    assertEquals(OzoneManagerProtocolProtos.Status.INVALID_SNAPSHOT_ERROR,
+        omResponse.getStatus());
+
+    assertEquals(1, getOmMetrics().getNumSnapshotCreateFails());
+    assertEquals(1, getOmMetrics().getNumSnapshotActive());
+    assertEquals(2, getOmMetrics().getNumSnapshotCreates());
+  }
+
   private void renameKey(String fromKey, String toKey, long offset)
       throws IOException {
     OmKeyInfo toKeyInfo = addKey(toKey, offset + 1L);

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

@@ -17,6 +17,7 @@
 
 package org.apache.hadoop.ozone.om.snapshot;
 
+import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_FS_SNAPSHOT_MAX_LIMIT_DEFAULT;
 import static org.apache.hadoop.ozone.om.request.OMRequestTestUtils.createOmKeyInfo;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -167,6 +168,7 @@ public void baseSetup() throws Exception {
         omMetadataManager);
     omSnapshotManager = new OmSnapshotManager(ozoneManager);
     when(ozoneManager.getOmSnapshotManager()).thenReturn(omSnapshotManager);
+    when(ozoneManager.getFsSnapshotMaxLimit()).thenReturn(OZONE_OM_FS_SNAPSHOT_MAX_LIMIT_DEFAULT);
   }
 
   @AfterEach