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

Author: peterxcli

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/snapshot/TestOMSnapshotCreateRequest.java

@@ -18,6 +18,7 @@
 package org.apache.hadoop.ozone.om.request.snapshot;
 
 import static org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationFactor.THREE;
+import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_FS_SNAPSHOT_MAX_LIMIT_DEFAULT;;
 import static org.apache.hadoop.ozone.om.helpers.SnapshotInfo.getFromProtobuf;
 import static org.apache.hadoop.ozone.om.helpers.SnapshotInfo.getTableKey;
 import static org.apache.hadoop.ozone.om.request.OMRequestTestUtils.createSnapshotRequest;
@@ -28,7 +29,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;
@@ -69,6 +70,7 @@ public class TestOMSnapshotCreateRequest extends TestSnapshotRequestAndResponse
   public void setup() throws Exception {
     snapshotName1 = UUID.randomUUID().toString();
     snapshotName2 = UUID.randomUUID().toString();
+    when(getOzoneManager().getFsSnapshotMaxLimit()).thenReturn(OZONE_OM_FS_SNAPSHOT_MAX_LIMIT_DEFAULT);
   }
 
   @ValueSource(strings = {
@@ -264,6 +266,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);