HDDS-7652. Volume Quota not enforced during write when bucket quota is not set

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/bucket/OMBucketSetPropertyRequest.java

@@ -305,10 +305,16 @@ public boolean checkQuotaBytesValid(OMMetadataManager metadataManager,
     List<OmBucketInfo> bucketList = metadataManager.listBuckets(
         omVolumeArgs.getVolume(), null, null, Integer.MAX_VALUE);
     for (OmBucketInfo bucketInfo : bucketList) {
+      if (omBucketArgs.getBucketName().equals(bucketInfo.getBucketName())) {
+        continue;
+      }
       long nextQuotaInBytes = bucketInfo.getQuotaInBytes();
-      if (nextQuotaInBytes > OzoneConsts.QUOTA_RESET &&
-          !omBucketArgs.getBucketName().equals(bucketInfo.getBucketName())) {
+      if (nextQuotaInBytes > OzoneConsts.QUOTA_RESET) {
         totalBucketQuota += nextQuotaInBytes;
+      } else {
+        // consider used space for bucket where quota is not set
+        // This quota will be part of volume quota
+        totalBucketQuota += bucketInfo.getUsedBytes();
       }
     }
 

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/file/OMFileCreateRequest.java

@@ -277,7 +277,8 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager,
       long preAllocatedSpace = newLocationList.size()
           * ozoneManager.getScmBlockSize()
           * omKeyInfo.getReplicationConfig().getRequiredNodes();
-      checkBucketQuotaInBytes(omBucketInfo, preAllocatedSpace);
+      checkBucketQuotaInBytes(omMetadataManager, omBucketInfo,
+          preAllocatedSpace);
       numMissingParents = missingParentInfos.size();
       checkBucketQuotaInNamespace(omBucketInfo, numMissingParents + 1L);
       omBucketInfo.incrUsedNamespace(numMissingParents);

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/file/OMFileCreateRequestWithFSO.java

@@ -199,7 +199,8 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager,
       long preAllocatedSpace =
           newLocationList.size() * ozoneManager.getScmBlockSize() * repConfig
               .getRequiredNodes();
-      checkBucketQuotaInBytes(omBucketInfo, preAllocatedSpace);
+      checkBucketQuotaInBytes(omMetadataManager, omBucketInfo,
+          preAllocatedSpace);
       checkBucketQuotaInNamespace(omBucketInfo, numKeysCreated + 1L);
       omBucketInfo.incrUsedNamespace(numKeysCreated);
 

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMAllocateBlockRequest.java

@@ -222,7 +222,8 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager,
       long totalAllocatedSpace = QuotaUtil.getReplicatedSize(
           preAllocatedKeySize, repConfig) + QuotaUtil.getReplicatedSize(
           hadAllocatedKeySize, repConfig);
-      checkBucketQuotaInBytes(omBucketInfo, totalAllocatedSpace);
+      checkBucketQuotaInBytes(omMetadataManager, omBucketInfo,
+          totalAllocatedSpace);
       // Append new block
       openKeyInfo.appendNewBlocks(newLocationList, false);
 

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMAllocateBlockRequestWithFSO.java

@@ -152,7 +152,8 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager,
       long totalAllocatedSpace = QuotaUtil.getReplicatedSize(
           preAllocatedKeySize, repConfig) + QuotaUtil.getReplicatedSize(
           hadAllocatedKeySize, repConfig);
-      checkBucketQuotaInBytes(omBucketInfo, totalAllocatedSpace);
+      checkBucketQuotaInBytes(omMetadataManager, omBucketInfo,
+          totalAllocatedSpace);
       // Append new block
       openKeyInfo.appendNewBlocks(newLocationList, false);
 

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyCommitRequest.java

@@ -235,10 +235,12 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager,
         oldKeyVersionsToDelete = getOldVersionsToCleanUp(dbOzoneKey,
             keyToDelete, omMetadataManager,
             trxnLogIndex, ozoneManager.isRatisEnabled());
-        checkBucketQuotaInBytes(omBucketInfo, correctedSpace);
+        checkBucketQuotaInBytes(omMetadataManager, omBucketInfo,
+            correctedSpace);
       } else {
         checkBucketQuotaInNamespace(omBucketInfo, 1L);
-        checkBucketQuotaInBytes(omBucketInfo, correctedSpace);
+        checkBucketQuotaInBytes(omMetadataManager, omBucketInfo,
+            correctedSpace);
         omBucketInfo.incrUsedNamespace(1L);
       }
 

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyCommitRequestWithFSO.java

@@ -173,10 +173,12 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager,
         oldKeyVersionsToDelete = getOldVersionsToCleanUp(dbFileKey,
             keyToDelete, omMetadataManager,
             trxnLogIndex, ozoneManager.isRatisEnabled());
-        checkBucketQuotaInBytes(omBucketInfo, correctedSpace);
+        checkBucketQuotaInBytes(omMetadataManager, omBucketInfo,
+            correctedSpace);
       } else {
         checkBucketQuotaInNamespace(omBucketInfo, 1L);
-        checkBucketQuotaInBytes(omBucketInfo, correctedSpace);
+        checkBucketQuotaInBytes(omMetadataManager, omBucketInfo,
+            correctedSpace);
         omBucketInfo.incrUsedNamespace(1L);
       }
 

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyCreateRequest.java

@@ -305,7 +305,8 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager,
           * ozoneManager.getScmBlockSize()
           * replicationConfig.getRequiredNodes();
       // check bucket and volume quota
-      checkBucketQuotaInBytes(omBucketInfo, preAllocatedSpace);
+      checkBucketQuotaInBytes(omMetadataManager, omBucketInfo,
+          preAllocatedSpace);
       checkBucketQuotaInNamespace(omBucketInfo, numMissingParents + 1L);
       omBucketInfo.incrUsedNamespace(numMissingParents);
 

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyCreateRequestWithFSO.java

@@ -185,7 +185,8 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager,
       long preAllocatedSpace =
           newLocationList.size() * ozoneManager.getScmBlockSize() * repConfig
               .getRequiredNodes();
-      checkBucketQuotaInBytes(omBucketInfo, preAllocatedSpace);
+      checkBucketQuotaInBytes(omMetadataManager, omBucketInfo,
+          preAllocatedSpace);
       checkBucketQuotaInNamespace(omBucketInfo, numKeysCreated + 1L);
       omBucketInfo.incrUsedNamespace(numKeysCreated);
 

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyRequest.java

@@ -520,11 +520,13 @@ protected FileEncryptionInfo getFileEncryptionInfo(KeyArgs keyArgs) {
 
   /**
    * Check bucket quota in bytes.
+   * @paran metadataManager
    * @param omBucketInfo
    * @param allocateSize
    * @throws IOException
    */
-  protected void checkBucketQuotaInBytes(OmBucketInfo omBucketInfo,
+  protected void checkBucketQuotaInBytes(
+      OMMetadataManager metadataManager, OmBucketInfo omBucketInfo,
       long allocateSize) throws IOException {
     if (omBucketInfo.getQuotaInBytes() > OzoneConsts.QUOTA_RESET) {
       long usedBytes = omBucketInfo.getUsedBytes();
@@ -536,6 +538,40 @@ protected void checkBucketQuotaInBytes(OmBucketInfo omBucketInfo,
             + allocateSize) + " Bytes.",
             OMException.ResultCodes.QUOTA_EXCEEDED);
       }
+    } else {
+      checkVolumeQuotaInBytes(metadataManager, omBucketInfo.getVolumeName(),
+          allocateSize);
+    }
+  }
+
+  private void checkVolumeQuotaInBytes(
+      OMMetadataManager metadataManager, String volumeName,
+      long allocateSize) throws IOException {
+    String volumeKey = metadataManager.getVolumeKey(volumeName);
+    OmVolumeArgs omVolumeArgs = metadataManager.getVolumeTable()
+        .get(volumeKey);
+    if (omVolumeArgs.getQuotaInBytes() <= OzoneConsts.QUOTA_RESET) {
+      return;
+    }
+
+    long totalBucketQuota = 0L;
+    List<OmBucketInfo> bucketList = metadataManager.listBuckets(
+        volumeName, null, null, Integer.MAX_VALUE);
+    for (OmBucketInfo bucketInfo : bucketList) {
+      long nextQuotaInBytes = bucketInfo.getQuotaInBytes();
+      if (nextQuotaInBytes > OzoneConsts.QUOTA_RESET) {
+        totalBucketQuota += nextQuotaInBytes;
+      } else {
+        totalBucketQuota += bucketInfo.getUsedBytes();
+      }
+    }
+
+    if (omVolumeArgs.getQuotaInBytes() - totalBucketQuota < allocateSize) {
+      throw new OMException("The DiskSpace quota of volume:"
+          + volumeName + " exceeded: quotaInBytes: "
+          + omVolumeArgs.getQuotaInBytes() + " Bytes but diskspace consumed: "
+          + (totalBucketQuota + allocateSize) + " Bytes.",
+          OMException.ResultCodes.QUOTA_EXCEEDED);
     }
   }
 

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/s3/multipart/S3MultipartUploadCommitPartRequest.java

@@ -234,7 +234,8 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager,
             OmKeyInfo.getFromProtobuf(oldPartKeyInfo.getPartKeyInfo());
         correctedSpace -= partKeyToBeDeleted.getReplicatedSize();
       }
-      checkBucketQuotaInBytes(omBucketInfo, correctedSpace);
+      checkBucketQuotaInBytes(omMetadataManager, omBucketInfo,
+          correctedSpace);
       omBucketInfo.incrUsedBytes(correctedSpace);
 
       omResponse.setCommitMultiPartUploadResponse(

hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/bucket/TestOMBucketSetPropertyRequest.java

@@ -214,6 +214,46 @@ public void testValidateAndUpdateCacheWithQuota() throws Exception {
             "should not be greater than volume quota"));
   }
 
+  @Test
+  public void testValidateAndUpdateCacheWithOtherBucketQuotaUsedCrossLimit()
+      throws Exception {
+    String volumeName = UUID.randomUUID().toString();
+    String bucketName = UUID.randomUUID().toString();
+    String testBucketName = bucketName + "1";
+
+    OMRequestTestUtils.addVolumeToDB(
+        volumeName, omMetadataManager, 10 * GB);
+    OMRequestTestUtils.addBucketToDB(volumeName, bucketName,
+        omMetadataManager);
+    OMRequestTestUtils.addBucketToDB(volumeName, testBucketName,
+        omMetadataManager);
+
+    // updated used size of one bucket
+    String bucketKey = omMetadataManager
+        .getBucketKey(volumeName, bucketName);
+    OmBucketInfo omBucketInfo = omMetadataManager.getBucketTable()
+        .getCacheValue(new CacheKey<>(bucketKey)).getCacheValue();
+    omBucketInfo.incrUsedBytes(8 * GB);
+    omMetadataManager.getBucketTable().put(bucketKey, omBucketInfo);
+
+    // create request to set bucket size where other bucket have used quota
+    OMRequest omRequest = createSetBucketPropertyRequest(volumeName,
+        testBucketName, true, 3 * GB);
+    OMBucketSetPropertyRequest omBucketSetPropertyRequest =
+        new OMBucketSetPropertyRequest(omRequest);
+
+    OMClientResponse omClientResponse = omBucketSetPropertyRequest
+        .validateAndUpdateCache(ozoneManager, 1,
+            ozoneManagerDoubleBufferHelper);
+
+    Assert.assertFalse(omClientResponse.getOMResponse().getSuccess());
+    Assert.assertEquals(omClientResponse.getOMResponse().getStatus(),
+        OzoneManagerProtocolProtos.Status.QUOTA_EXCEEDED);
+    Assert.assertTrue(omClientResponse.getOMResponse().getMessage().
+        contains("Total buckets quota in this volume " +
+            "should not be greater than volume quota"));
+  }
+
   @Test
   public void rejectsSettingQuotaOnLink() throws Exception {
     // GIVEN

hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/file/TestOMFileCreateRequest.java

@@ -194,6 +194,29 @@ public void testValidateAndUpdateCacheWithNamespaceQuotaExceeded()
         == OzoneManagerProtocolProtos.Status.QUOTA_EXCEEDED);
   }
 
+  @Test
+  public void testValidateAndUpdateCacheWithOnlyVolumeQuotaExceeded()
+      throws Exception {
+    OMRequest omRequest = createFileRequest(volumeName, bucketName, keyName,
+        HddsProtos.ReplicationFactor.ONE, HddsProtos.ReplicationType.RATIS,
+        false, true);
+
+    // add volume and create bucket with quota limit 1
+    OMRequestTestUtils.addVolumeToDB(volumeName, omMetadataManager, 1L);
+    OMRequestTestUtils.addBucketToDB(volumeName, bucketName,
+        omMetadataManager, getBucketLayout());
+
+    OMFileCreateRequest omFileCreateRequest = getOMFileCreateRequest(omRequest);
+    OMRequest modifiedOmRequest = omFileCreateRequest.preExecute(ozoneManager);
+
+    omFileCreateRequest = getOMFileCreateRequest(modifiedOmRequest);
+    OMClientResponse omFileCreateResponse =
+        omFileCreateRequest.validateAndUpdateCache(ozoneManager, 100L,
+            ozoneManagerDoubleBufferHelper);
+    Assert.assertTrue(omFileCreateResponse.getOMResponse().getStatus()
+        == OzoneManagerProtocolProtos.Status.QUOTA_EXCEEDED);
+  }
+
   @Test
   public void testValidateAndUpdateCacheWithVolumeNotFound() throws Exception {
     OMRequest omRequest = createFileRequest(volumeName, bucketName, keyName,

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

@@ -188,6 +188,30 @@ public void testValidateAndUpdateCacheWithKeyNotFound() throws Exception {
 
   }
 
+  @Test
+  public void testValidateAndUpdateCacheOnlyVolumeQuotaExceeded()
+      throws Exception {
+    // Add volume, bucket entries to DB.
+    // add volume and create bucket with quota limit 1
+    OMRequestTestUtils.addVolumeToDB(volumeName, omMetadataManager, 1L);
+    OMRequestTestUtils.addBucketToDB(volumeName, bucketName,
+        omMetadataManager, getBucketLayout());
+    addKeyToOpenKeyTable(volumeName, bucketName);
+
+    OMRequest modifiedOmRequest =
+        doPreExecute(createAllocateBlockRequest());
+
+    OMAllocateBlockRequest omAllocateBlockRequest =
+        getOmAllocateBlockRequest(modifiedOmRequest);
+
+    OMClientResponse omAllocateBlockResponse =
+        omAllocateBlockRequest.validateAndUpdateCache(ozoneManager, 100L,
+            ozoneManagerDoubleBufferHelper);
+
+    Assert.assertTrue(omAllocateBlockResponse.getOMResponse().getStatus()
+        == OzoneManagerProtocolProtos.Status.QUOTA_EXCEEDED);
+  }
+
   /**
    * This method calls preExecute and verify the modified request.
    * @param originalOMRequest

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

@@ -400,6 +400,37 @@ public void testValidateAndUpdateCacheWithBucketQuotaExceeds()
     Assert.assertEquals(0, bucketInfo.getUsedNamespace());
   }
 
+  @Test
+  public void testValidateAndUpdateCacheWithOnlyVolumeQuotaExceeds()
+      throws Exception {
+    OMRequest modifiedOmRequest =
+        doPreExecute(createCommitKeyRequest());
+
+    OMKeyCommitRequest omKeyCommitRequest =
+        getOmKeyCommitRequest(modifiedOmRequest);
+
+    KeyArgs keyArgs = modifiedOmRequest.getCommitKeyRequest().getKeyArgs();
+
+    // Append new blocks
+    List<OmKeyLocationInfo> allocatedLocationList =
+        keyArgs.getKeyLocationsList().stream()
+            .map(OmKeyLocationInfo::getFromProtobuf)
+            .collect(Collectors.toList());
+
+    OMRequestTestUtils.addVolumeToDB(volumeName, omMetadataManager, 1L);
+    OMRequestTestUtils.addBucketToDB(volumeName, bucketName,
+        omMetadataManager, omKeyCommitRequest.getBucketLayout());
+
+    addKeyToOpenKeyTable(allocatedLocationList);
+
+    OMClientResponse omClientResponse =
+        omKeyCommitRequest.validateAndUpdateCache(ozoneManager,
+            100L, ozoneManagerDoubleBufferHelper);
+
+    Assert.assertEquals(OzoneManagerProtocolProtos.Status.QUOTA_EXCEEDED,
+        omClientResponse.getOMResponse().getStatus());
+  }
+
   @Test
   public void testValidateAndUpdateCacheWithKeyNotFound() throws Exception {
 

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

@@ -231,6 +231,29 @@ public void testValidateAndUpdateCacheWithNamespaceQuotaExceeded()
         == OzoneManagerProtocolProtos.Status.QUOTA_EXCEEDED);
   }
 
+  @Test
+  public void testValidateAndUpdateCacheWithOnlyVolumeQuotaExceeded()
+      throws Exception {
+    when(ozoneManager.getOzoneLockProvider()).thenReturn(
+        new OzoneLockProvider(keyPathLockEnabled, enableFileSystemPaths));
+    OMRequest modifiedOmRequest =
+        doPreExecute(createKeyRequest(false, 0, keyName));
+
+    OMKeyCreateRequest omKeyCreateRequest = getOMKeyCreateRequest(
+        modifiedOmRequest, getBucketLayout());
+
+    // create bucket with quota limit 1
+    OMRequestTestUtils.addVolumeToDB(volumeName, omMetadataManager, 1L);
+    OMRequestTestUtils.addBucketToDB(volumeName, bucketName,
+        omMetadataManager, omKeyCreateRequest.getBucketLayout());
+
+    OMClientResponse omKeyCreateResponse =
+        omKeyCreateRequest.validateAndUpdateCache(ozoneManager, 100L,
+            ozoneManagerDoubleBufferHelper);
+    Assert.assertTrue(omKeyCreateResponse.getOMResponse().getStatus()
+        == OzoneManagerProtocolProtos.Status.QUOTA_EXCEEDED);
+  }
+
   private void checkResponse(OMRequest modifiedOmRequest,
       OMClientResponse omKeyCreateResponse, long id, boolean override,
       BucketLayout bucketLayout) throws Exception {