HDDS-3727. Volume space: check quotaUsageInBytes when write key.

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

@@ -229,7 +229,9 @@ public enum ResultCodes {
 
     NOT_SUPPORTED_OPERATION,
 
-    PARTIAL_RENAME
+    PARTIAL_RENAME,
+
+    QUOTA_EXCEEDED
 
   }
 }

hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/ozone/TestRootedOzoneFileSystem.java

@@ -818,11 +818,11 @@ public void testTempMount() throws Exception {
     VolumeArgs volumeArgs = new VolumeArgs.Builder()
         .setAcls(Collections.singletonList(aclWorldAccess))
         .setQuotaInCounts(1000)
-        .setQuotaInBytes("1MB").build();
+        .setQuotaInBytes("1TB").build();
     // Sanity check
     Assert.assertNull(volumeArgs.getOwner());
     Assert.assertNull(volumeArgs.getAdmin());
-    Assert.assertEquals("1MB", volumeArgs.getQuotaInBytes());
+    Assert.assertEquals("1TB", volumeArgs.getQuotaInBytes());
     Assert.assertEquals(1000, volumeArgs.getQuotaInCounts());
     Assert.assertEquals(0, volumeArgs.getMetadata().size());
     Assert.assertEquals(1, volumeArgs.getAcls().size());

hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestOzoneRpcClientAbstract.java

@@ -707,9 +707,71 @@ public void testPutKey() throws IOException {
     }
   }
 
+  @Test
+  public void testCheckUsedBytesQuota() throws IOException {
+    String volumeName = UUID.randomUUID().toString();
+    String bucketName = UUID.randomUUID().toString();
+    OzoneVolume volume = null;
+
+    String value = "sample value";
+    int blockSize = (int) ozoneManager.getConfiguration().getStorageSize(
+        OZONE_SCM_BLOCK_SIZE, OZONE_SCM_BLOCK_SIZE_DEFAULT, StorageUnit.BYTES);
+    int valueLength = value.getBytes().length;
+    int countException = 0;
+
+    store.createVolume(volumeName);
+    volume = store.getVolume(volumeName);
+    // Set quota In Bytes for a smaller value
+    store.getVolume(volumeName).setQuota(
+        OzoneQuota.parseQuota("1 Bytes", 100));
+    volume.createBucket(bucketName);
+    OzoneBucket bucket = volume.getBucket(bucketName);
+
+    // Test write key.
+    // The remaining quota does not satisfy a block size, so the write fails.
+    try {
+      writeKey(bucket, UUID.randomUUID().toString(), ONE, value, valueLength);
+    } catch (IOException ex) {
+      countException++;
+      GenericTestUtils.assertExceptionContains("QUOTA_EXCEEDED", ex);
+    }
+    // Write failed, volume usedBytes should be 0
+    Assert.assertEquals(0L, store.getVolume(volumeName).getUsedBytes());
+
+    // Test write file.
+    // The remaining quota does not satisfy a block size, so the write fails.
+    try {
+      writeFile(bucket, UUID.randomUUID().toString(), ONE, value, 0);
+    } catch (IOException ex) {
+      countException++;
+      GenericTestUtils.assertExceptionContains("QUOTA_EXCEEDED", ex);
+    }
+    // Write failed, volume usedBytes should be 0
+    Assert.assertEquals(0L, store.getVolume(volumeName).getUsedBytes());
+
+    // Write large key, test allocateBlock fails.
+    store.getVolume(volumeName).setQuota(
+        OzoneQuota.parseQuota(blockSize + "Bytes", 100));
+    try {
+      OzoneOutputStream out = bucket.createKey(UUID.randomUUID().toString(),
+          valueLength, STAND_ALONE, ONE, new HashMap<>());
+      for(int i=0; i <= blockSize/value.length(); i++) {
+        out.write(value.getBytes());
+      }
+      out.close();
+    } catch (IOException ex) {
+      countException++;
+      GenericTestUtils.assertExceptionContains("QUOTA_EXCEEDED", ex);
+    }
+    // AllocateBlock failed, volume usedBytes should be 0
+    Assert.assertEquals(0L, store.getVolume(volumeName).getUsedBytes());
+
+    Assert.assertEquals(3, countException);
+  }
+
   @Test
   @SuppressWarnings("methodlength")
-  public void testVolumeAndBucketUsedBytes() throws IOException {
+  public void testVolumeUsedBytes() throws IOException {
     String volumeName = UUID.randomUUID().toString();
     String bucketName = UUID.randomUUID().toString();
     OzoneVolume volume = null;

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

@@ -314,6 +314,8 @@ enum Status {
 
     PARTIAL_RENAME = 65;
 
+    QUOTA_EXCEEDED = 66;
+
 }
 
 /**

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

@@ -277,6 +277,16 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager,
           .collect(Collectors.toList());
       omKeyInfo.appendNewBlocks(newLocationList, false);
 
+      omVolumeArgs = getVolumeInfo(omMetadataManager, volumeName);
+      omBucketInfo = getBucketInfo(omMetadataManager, volumeName, bucketName);
+      // check volume quota
+      long preAllocatedSpace = newLocationList.size()
+          * ozoneManager.getScmBlockSize()
+          * omKeyInfo.getFactor().getNumber();
+      if (omVolumeArgs.getQuotaInBytes() > OzoneConsts.QUOTA_RESET) {
+        checkVolumeQuotaInBytes(omVolumeArgs, preAllocatedSpace);
+      }
+
       // Add to cache entry can be done outside of lock for this openKey.
       // Even if bucket gets deleted, when commitKey we shall identify if
       // bucket gets deleted.
@@ -290,13 +300,7 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager,
           bucketName, Optional.absent(), Optional.of(missingParentInfos),
           trxnLogIndex);
 
-      long scmBlockSize = ozoneManager.getScmBlockSize();
-      omVolumeArgs = getVolumeInfo(omMetadataManager, volumeName);
-      omBucketInfo = getBucketInfo(omMetadataManager, volumeName, bucketName);
-
       // update usedBytes atomically.
-      long preAllocatedSpace = newLocationList.size() * scmBlockSize
-          * omKeyInfo.getFactor().getNumber();
       omVolumeArgs.getUsedBytes().add(preAllocatedSpace);
       omBucketInfo.getUsedBytes().add(preAllocatedSpace);
 

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

@@ -164,7 +164,7 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager,
         getOmRequest());
     OMClientResponse omClientResponse = null;
 
-    OmKeyInfo openKeyInfo;
+    OmKeyInfo openKeyInfo = null;
     IOException exception = null;
     OmVolumeArgs omVolumeArgs = null;
     OmBucketInfo omBucketInfo = null;
@@ -192,9 +192,19 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager,
             KEY_NOT_FOUND);
       }
 
-      // Append new block
+
       List<OmKeyLocationInfo> newLocationList = Collections.singletonList(
           OmKeyLocationInfo.getFromProtobuf(blockLocation));
+      omVolumeArgs = getVolumeInfo(omMetadataManager, volumeName);
+      omBucketInfo = getBucketInfo(omMetadataManager, volumeName, bucketName);
+      // check volume quota
+      long preAllocatedSpace = newLocationList.size()
+          * ozoneManager.getScmBlockSize()
+          * openKeyInfo.getFactor().getNumber();
+      if (omVolumeArgs.getQuotaInBytes() > OzoneConsts.QUOTA_RESET) {
+        checkVolumeQuotaInBytes(omVolumeArgs, preAllocatedSpace);
+      }
+      // Append new block
       openKeyInfo.appendNewBlocks(newLocationList, false);
 
       // Set modification time.
@@ -208,12 +218,7 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager,
           new CacheKey<>(openKeyName),
           new CacheValue<>(Optional.of(openKeyInfo), trxnLogIndex));
 
-      long scmBlockSize = ozoneManager.getScmBlockSize();
-      omVolumeArgs = getVolumeInfo(omMetadataManager, volumeName);
-      omBucketInfo = getBucketInfo(omMetadataManager, volumeName, bucketName);
       // update usedBytes atomically.
-      long preAllocatedSpace = newLocationList.size() * scmBlockSize
-          * openKeyInfo.getFactor().getNumber();
       omVolumeArgs.getUsedBytes().add(preAllocatedSpace);
       omBucketInfo.getUsedBytes().add(preAllocatedSpace);
 
@@ -227,6 +232,18 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager,
     } catch (IOException ex) {
       omMetrics.incNumBlockAllocateCallFails();
       exception = ex;
+      if (exception.toString().contains(
+          OMException.ResultCodes.QUOTA_EXCEEDED.toString())) {
+        long keyAllocatedSpace = openKeyInfo.getLatestVersionLocations()
+            .getLocationListCount() * ozoneManager.getScmBlockSize()
+            * openKeyInfo.getFactor().getNumber();
+        // Update usedBytes atomically. ErrorOMResponse does not persist the DB,
+        // so we update the cache first. The next time another key is written,
+        // volume Args will be persisted to the DB.
+        // TODO: There is a delay in updating DB in this way, and if necessary
+        //  we can modify the ErrorOMResponse to avoid it.
+        omVolumeArgs.getUsedBytes().add(-keyAllocatedSpace);
+      }
       omClientResponse = new OMAllocateBlockResponse(createErrorOMResponse(
           omResponse, exception));
       LOG.error("Allocate Block failed. Volume:{}, Bucket:{}, OpenKey:{}. " +
@@ -239,8 +256,7 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager,
     auditLog(auditLogger, buildAuditMessage(OMAction.ALLOCATE_BLOCK, auditMap,
         exception, getOmRequest().getUserInfo()));
 
-
-
     return omClientResponse;
   }
+
 }

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

@@ -29,6 +29,7 @@
 import com.google.common.base.Preconditions;
 import org.apache.hadoop.ozone.OmUtils;
 import org.apache.hadoop.ozone.OzoneAcl;
+import org.apache.hadoop.ozone.OzoneConsts;
 import org.apache.hadoop.ozone.om.OMConfigKeys;
 import org.apache.hadoop.ozone.om.exceptions.OMException;
 import org.apache.hadoop.ozone.om.helpers.OmVolumeArgs;
@@ -286,30 +287,33 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager,
           .collect(Collectors.toList());
       omKeyInfo.appendNewBlocks(newLocationList, false);
 
-      // Add to cache entry can be done outside of lock for this openKey.
-      // Even if bucket gets deleted, when commitKey we shall identify if
-      // bucket gets deleted.
-      omMetadataManager.getOpenKeyTable().addCacheEntry(
-          new CacheKey<>(dbOpenKeyName),
-          new CacheValue<>(Optional.of(omKeyInfo), trxnLogIndex));
-
-      long scmBlockSize = ozoneManager.getScmBlockSize();
       omVolumeArgs = getVolumeInfo(omMetadataManager, volumeName);
       omBucketInfo = getBucketInfo(omMetadataManager, volumeName, bucketName);
-
       // Here we refer to the implementation of HDFS:
       // If the key size is 600MB, when createKey, keyLocationInfo in
       // keyLocationList is 3, and  the every pre-allocated block length is
       // 256MB. If the number of factor  is 3, the total pre-allocated block
       // ize is 256MB * 3 * 3. We will allocate more 256MB * 3 * 3 - 600mb * 3
       // = 504MB in advance, and we  will subtract this part when we finally
       // commitKey.
-      long preAllocatedSpace = newLocationList.size() * scmBlockSize
+      long preAllocatedSpace = newLocationList.size()
+          * ozoneManager.getScmBlockSize()
           * omKeyInfo.getFactor().getNumber();
+      // check volume quota
+      if (omVolumeArgs.getQuotaInBytes() > OzoneConsts.QUOTA_RESET) {
+        checkVolumeQuotaInBytes(omVolumeArgs, preAllocatedSpace);
+      }
+
+      // Add to cache entry can be done outside of lock for this openKey.
+      // Even if bucket gets deleted, when commitKey we shall identify if
+      // bucket gets deleted.
+      omMetadataManager.getOpenKeyTable().addCacheEntry(
+          new CacheKey<>(dbOpenKeyName),
+          new CacheValue<>(Optional.of(omKeyInfo), trxnLogIndex));
+
       omVolumeArgs.getUsedBytes().add(preAllocatedSpace);
       omBucketInfo.getUsedBytes().add(preAllocatedSpace);
 
-
       // Prepare response
       omResponse.setCreateKeyResponse(CreateKeyResponse.newBuilder()
           .setKeyInfo(omKeyInfo.getProtobuf())

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

@@ -533,6 +533,26 @@ protected FileEncryptionInfo getFileEncryptionInfo(KeyArgs keyArgs) {
     return encryptionInfo;
   }
 
+  /**
+   * Check volume quota in bytes.
+   * @param omVolumeArgs
+   * @param allocateSize
+   * @throws IOException
+   */
+  protected void checkVolumeQuotaInBytes(OmVolumeArgs omVolumeArgs,
+      long allocateSize) throws IOException {
+    long usedBytes = omVolumeArgs.getUsedBytes().sum();
+    long quotaInBytes = omVolumeArgs.getQuotaInBytes();
+    if (quotaInBytes - usedBytes < allocateSize) {
+      throw new OMException("The DiskSpace quota of volume:"
+          + omVolumeArgs.getVolume() + "exceeded: quotaInBytes: "
+          + quotaInBytes + " Bytes but diskspace consumed: " + (usedBytes
+          + allocateSize) + " Bytes.",
+          OMException.ResultCodes.QUOTA_EXCEEDED);
+    }
+
+  }
+
   /**
    * Check directory exists. If exists return true, else false.
    * @param volumeName

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

@@ -283,6 +283,7 @@ public static void addVolumeToDB(String volumeName, String ownerName,
     OmVolumeArgs omVolumeArgs =
         OmVolumeArgs.newBuilder().setCreationTime(Time.now())
             .setVolume(volumeName).setAdminName(ownerName)
+            .setQuotaInBytes(Long.MAX_VALUE)
             .setOwnerName(ownerName).build();
     omMetadataManager.getVolumeTable().put(
         omMetadataManager.getVolumeKey(volumeName), omVolumeArgs);