HDDS-11284. refactor quota repair non-blocking while upgrade

hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/db/DBStore.java

@@ -167,6 +167,14 @@ <KEY, VALUE> void move(KEY sourceKey, KEY destKey, VALUE value,
    */
   DBCheckpoint getCheckpoint(boolean flush) throws IOException;
 
+  /**
+   * Get current snapshot of DB store as an artifact stored on
+   * the local filesystem with different parent path.
+   * @return An object that encapsulates the checkpoint information along with
+   * location.
+   */
+  DBCheckpoint getCheckpoint(String parentDir, boolean flush) throws IOException;
+
   /**
    * Get DB Store location.
    * @return DB file location.

hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/db/RDBStore.java

@@ -338,6 +338,14 @@ public DBCheckpoint getCheckpoint(boolean flush) throws IOException {
     return checkPointManager.createCheckpoint(checkpointsParentDir);
   }
 
+  @Override
+  public DBCheckpoint getCheckpoint(String parentPath, boolean flush) throws IOException {
+    if (flush) {
+      this.flushDB();
+    }
+    return checkPointManager.createCheckpoint(parentPath, null);
+  }
+
   public DBCheckpoint getSnapshot(String name) throws IOException {
     this.flushLog(true);
     return checkPointManager.createCheckpoint(snapshotsParentDir, name);

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/OmUtils.java

@@ -327,6 +327,7 @@ public static boolean isReadOnly(
     case SetTimes:
     case AbortExpiredMultiPartUploads:
     case SetSnapshotProperty:
+    case QuotaRepair:
     case UnknownCommand:
       return false;
     case EchoRPC:

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

@@ -147,8 +147,8 @@ enum Type {
   ListStatusLight = 129;
   GetSnapshotInfo = 130;
   RenameSnapshot = 131;
-
   ListOpenFiles = 132;
+  QuotaRepair = 133;
 }
 
 enum SafeMode {
@@ -285,8 +285,8 @@ message OMRequest {
   optional SetSnapshotPropertyRequest       SetSnapshotPropertyRequest     = 127;
   optional SnapshotInfoRequest              SnapshotInfoRequest            = 128;
   optional RenameSnapshotRequest            RenameSnapshotRequest          = 129;
-
   optional ListOpenFilesRequest             ListOpenFilesRequest           = 130;
+  optional QuotaRepairRequest               QuotaRepairRequest             = 131;
 }
 
 message OMResponse {
@@ -410,8 +410,8 @@ message OMResponse {
   optional SnapshotInfoResponse              SnapshotInfoResponse          = 130;
   optional OMLockDetailsProto                omLockDetails                 = 131;
   optional RenameSnapshotResponse            RenameSnapshotResponse        = 132;
-
   optional ListOpenFilesResponse             ListOpenFilesResponse         = 133;
+  optional QuotaRepairResponse            QuotaRepairResponse        = 134;
 }
 
 enum Status {
@@ -2187,6 +2187,21 @@ message SetSafeModeResponse {
   optional bool response = 1;
 }
 
+message QuotaRepairRequest {
+    repeated BucketQuotaCount bucketCount = 1;
+    required bool supportVolumeOldQuota = 2 [default=false];
+}
+message BucketQuotaCount {
+    required string volName = 1;
+    required string bucketName = 2;
+    required int64 diffUsedBytes = 3;
+    required int64 diffUsedNamespace = 4;
+    required bool supportOldQuota = 5 [default=false];
+}
+
+message QuotaRepairResponse {
+}
+
 message OMLockDetailsProto {
   optional bool isLockAcquired = 1;
   optional uint64 waitLockNanos    = 2;

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

@@ -84,6 +84,7 @@
 import org.apache.hadoop.ozone.om.request.upgrade.OMFinalizeUpgradeRequest;
 import org.apache.hadoop.ozone.om.request.upgrade.OMPrepareRequest;
 import org.apache.hadoop.ozone.om.request.util.OMEchoRPCWriteRequest;
+import org.apache.hadoop.ozone.om.request.volume.OMQuotaRepairRequest;
 import org.apache.hadoop.ozone.om.request.volume.OMVolumeCreateRequest;
 import org.apache.hadoop.ozone.om.request.volume.OMVolumeDeleteRequest;
 import org.apache.hadoop.ozone.om.request.volume.OMVolumeSetOwnerRequest;
@@ -331,6 +332,8 @@ public static OMClientRequest createClientRequest(OMRequest omRequest,
       return new OMEchoRPCWriteRequest(omRequest);
     case AbortExpiredMultiPartUploads:
       return new S3ExpiredMultipartUploadsAbortRequest(omRequest);
+    case QuotaRepair:
+      return new OMQuotaRepairRequest(omRequest);
     default:
       throw new OMException("Unrecognized write command type request "
           + cmdType, OMException.ResultCodes.INVALID_REQUEST);

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/volume/OMQuotaRepairRequest.java

@@ -0,0 +1,191 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.ozone.om.request.volume;
+
+import com.google.common.base.Preconditions;
+import java.io.IOException;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.hadoop.hdds.utils.db.Table;
+import org.apache.hadoop.hdds.utils.db.TableIterator;
+import org.apache.hadoop.hdds.utils.db.cache.CacheKey;
+import org.apache.hadoop.hdds.utils.db.cache.CacheValue;
+import org.apache.hadoop.ozone.om.OMMetadataManager;
+import org.apache.hadoop.ozone.om.OzoneManager;
+import org.apache.hadoop.ozone.om.exceptions.OMException;
+import org.apache.hadoop.ozone.om.helpers.OmBucketInfo;
+import org.apache.hadoop.ozone.om.helpers.OmVolumeArgs;
+import org.apache.hadoop.ozone.om.request.OMClientRequest;
+import org.apache.hadoop.ozone.om.request.util.OmResponseUtil;
+import org.apache.hadoop.ozone.om.response.OMClientResponse;
+import org.apache.hadoop.ozone.om.response.volume.OMQuotaRepairResponse;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMRequest;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.ratis.server.protocol.TermIndex;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static org.apache.hadoop.ozone.OzoneConsts.OLD_QUOTA_DEFAULT;
+import static org.apache.hadoop.ozone.OzoneConsts.QUOTA_RESET;
+import static org.apache.hadoop.ozone.om.lock.OzoneManagerLock.Resource.BUCKET_LOCK;
+import static org.apache.hadoop.ozone.om.lock.OzoneManagerLock.Resource.VOLUME_LOCK;
+
+/**
+ * Handle OMQuotaRepairRequest Request.
+ */
+public class OMQuotaRepairRequest extends OMClientRequest {
+  private static final Logger LOG =
+      LoggerFactory.getLogger(OMQuotaRepairRequest.class);
+
+  public OMQuotaRepairRequest(OMRequest omRequest) {
+    super(omRequest);
+  }
+
+  @Override
+  public OMRequest preExecute(OzoneManager ozoneManager) throws IOException {
+    UserGroupInformation ugi = createUGIForApi();
+    if (ozoneManager.getAclsEnabled() && !ozoneManager.isAdmin(ugi)) {
+      throw new OMException("Access denied for user " + ugi + ". Admin privilege is required for quota repair.",
+          OMException.ResultCodes.ACCESS_DENIED);
+    }
+    return super.preExecute(ozoneManager);
+  }
+
+  @Override
+  @SuppressWarnings("methodlength")
+  public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, TermIndex termIndex) {
+    final long transactionLogIndex = termIndex.getIndex();
+    OzoneManagerProtocolProtos.QuotaRepairRequest quotaRepairRequest =
+        getOmRequest().getQuotaRepairRequest();
+    Preconditions.checkNotNull(quotaRepairRequest);
+
+    OMMetadataManager omMetadataManager = ozoneManager.getMetadataManager();
+    OzoneManagerProtocolProtos.OMResponse.Builder omResponse = OmResponseUtil.getOMResponseBuilder(getOmRequest());
+    Map<Pair<String, String>, OmBucketInfo> bucketMap = new HashMap<>();
+    OMClientResponse omClientResponse = null;
+    try {
+      for (int i = 0; i < quotaRepairRequest.getBucketCountCount(); ++i) {
+        OzoneManagerProtocolProtos.BucketQuotaCount bucketCountInfo = quotaRepairRequest.getBucketCount(i);
+        updateBucketInfo(omMetadataManager, bucketCountInfo, transactionLogIndex, bucketMap);
+      }
+      Map<String, OmVolumeArgs> volUpdateMap;
+      if (quotaRepairRequest.getSupportVolumeOldQuota()) {
+        volUpdateMap = updateOldVolumeQuotaSupport(omMetadataManager, transactionLogIndex);
+      } else {
+        volUpdateMap = Collections.emptyMap();
+      }
+      omResponse.setQuotaRepairResponse(
+          OzoneManagerProtocolProtos.QuotaRepairResponse.newBuilder().build());
+      omClientResponse = new OMQuotaRepairResponse(omResponse.build(), volUpdateMap, bucketMap);
+    } catch (IOException ex) {
+      LOG.error("failed to update repair count", ex);
+      omClientResponse = new OMQuotaRepairResponse(createErrorOMResponse(omResponse, ex));
+    } finally {
+      if (omClientResponse != null) {
+        omClientResponse.setOmLockDetails(getOmLockDetails());
+      }
+    }
+
+    return omClientResponse;
+  }
+
+  private void updateBucketInfo(
+      OMMetadataManager omMetadataManager, OzoneManagerProtocolProtos.BucketQuotaCount bucketCountInfo,
+      long transactionLogIndex, Map<Pair<String, String>, OmBucketInfo> bucketMap) throws IOException {
+    // acquire lock.
+    mergeOmLockDetails(omMetadataManager.getLock().acquireWriteLock(
+        BUCKET_LOCK, bucketCountInfo.getVolName(), bucketCountInfo.getBucketName()));
+    boolean acquiredBucketLock = getOmLockDetails().isLockAcquired();
+    try {
+      String bucketKey = omMetadataManager.getBucketKey(bucketCountInfo.getVolName(),
+          bucketCountInfo.getBucketName());
+      OmBucketInfo bucketInfo = omMetadataManager.getBucketTable().get(bucketKey);
+      if (null == bucketInfo) {
+        // bucket might be deleted when running repair count parallel
+        return;
+      }
+      bucketInfo.incrUsedBytes(bucketCountInfo.getDiffUsedBytes());
+      bucketInfo.incrUsedNamespace(bucketCountInfo.getDiffUsedNamespace());
+      if (bucketCountInfo.getSupportOldQuota()) {
+        OmBucketInfo.Builder builder = bucketInfo.toBuilder();
+        if (bucketInfo.getQuotaInBytes() == OLD_QUOTA_DEFAULT) {
+          builder.setQuotaInBytes(QUOTA_RESET);
+        }
+        if (bucketInfo.getQuotaInNamespace() == OLD_QUOTA_DEFAULT) {
+          builder.setQuotaInNamespace(QUOTA_RESET);
+        }
+        bucketInfo = builder.build();
+      }
+
+      omMetadataManager.getBucketTable().addCacheEntry(
+          new CacheKey<>(bucketKey), CacheValue.get(transactionLogIndex, bucketInfo));
+      bucketMap.put(Pair.of(bucketCountInfo.getVolName(), bucketCountInfo.getBucketName()), bucketInfo);
+    } finally {
+      if (acquiredBucketLock) {
+        mergeOmLockDetails(omMetadataManager.getLock()
+            .releaseWriteLock(BUCKET_LOCK, bucketCountInfo.getVolName(), bucketCountInfo.getBucketName()));
+      }
+    }
+  }
+
+  private Map<String, OmVolumeArgs> updateOldVolumeQuotaSupport(
+      OMMetadataManager metadataManager, long transactionLogIndex) throws IOException {
+    LOG.info("Starting volume quota support update");
+    Map<String, OmVolumeArgs> volUpdateMap = new HashMap<>();
+    try (TableIterator<String, ? extends Table.KeyValue<String, OmVolumeArgs>>
+             iterator = metadataManager.getVolumeTable().iterator()) {
+      while (iterator.hasNext()) {
+        Table.KeyValue<String, OmVolumeArgs> entry = iterator.next();
+        OmVolumeArgs omVolumeArgs = entry.getValue();
+        if (!(omVolumeArgs.getQuotaInBytes() == OLD_QUOTA_DEFAULT
+            || omVolumeArgs.getQuotaInNamespace() == OLD_QUOTA_DEFAULT)) {
+          continue;
+        }
+        mergeOmLockDetails(metadataManager.getLock().acquireWriteLock(
+            VOLUME_LOCK, omVolumeArgs.getVolume()));
+        boolean acquiredVolumeLock = getOmLockDetails().isLockAcquired();
+        try {
+          boolean isQuotaReset = false;
+          if (omVolumeArgs.getQuotaInBytes() == OLD_QUOTA_DEFAULT) {
+            omVolumeArgs.setQuotaInBytes(QUOTA_RESET);
+            isQuotaReset = true;
+          }
+          if (omVolumeArgs.getQuotaInNamespace() == OLD_QUOTA_DEFAULT) {
+            omVolumeArgs.setQuotaInNamespace(QUOTA_RESET);
+            isQuotaReset = true;
+          }
+          if (isQuotaReset) {
+            metadataManager.getVolumeTable().addCacheEntry(
+                new CacheKey<>(entry.getKey()), CacheValue.get(transactionLogIndex, omVolumeArgs));
+            volUpdateMap.put(entry.getKey(), omVolumeArgs);
+          }
+        } finally {
+          if (acquiredVolumeLock) {
+            mergeOmLockDetails(metadataManager.getLock().releaseWriteLock(VOLUME_LOCK, omVolumeArgs.getVolume()));
+          }
+        }
+      }
+    }
+    LOG.info("Completed volume quota support update for volume count {}", volUpdateMap.size());
+    return volUpdateMap;
+  }
+}

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/response/volume/OMQuotaRepairResponse.java

@@ -0,0 +1,72 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.ozone.om.response.volume;
+
+import jakarta.annotation.Nonnull;
+import java.io.IOException;
+import java.util.Map;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.hadoop.hdds.utils.db.BatchOperation;
+import org.apache.hadoop.ozone.om.OMMetadataManager;
+import org.apache.hadoop.ozone.om.helpers.OmBucketInfo;
+import org.apache.hadoop.ozone.om.helpers.OmVolumeArgs;
+import org.apache.hadoop.ozone.om.request.volume.OMQuotaRepairRequest;
+import org.apache.hadoop.ozone.om.response.CleanupTableInfo;
+import org.apache.hadoop.ozone.om.response.OMClientResponse;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMResponse;
+
+import static org.apache.hadoop.ozone.om.OmMetadataManagerImpl.BUCKET_TABLE;
+import static org.apache.hadoop.ozone.om.OmMetadataManagerImpl.VOLUME_TABLE;
+
+/**
+ * Response for {@link OMQuotaRepairRequest} request.
+ */
+@CleanupTableInfo(cleanupTables = {VOLUME_TABLE, BUCKET_TABLE})
+public class OMQuotaRepairResponse extends OMClientResponse {
+  private Map<String, OmVolumeArgs> volumeArgsMap;
+  private Map<Pair<String, String>, OmBucketInfo> volBucketInfoMap;
+
+  /**
+   * for quota failure response update.
+   */
+  public OMQuotaRepairResponse(@Nonnull OMResponse omResponse) {
+    super(omResponse);
+  }
+
+  public OMQuotaRepairResponse(
+      @Nonnull OMResponse omResponse, Map<String, OmVolumeArgs> volumeArgsMap,
+      Map<Pair<String, String>, OmBucketInfo> volBucketInfoMap) {
+    super(omResponse);
+    this.volBucketInfoMap = volBucketInfoMap;
+    this.volumeArgsMap = volumeArgsMap;
+  }
+
+  @Override
+  public void addToDBBatch(OMMetadataManager metadataManager,
+      BatchOperation batchOp) throws IOException {
+    for (OmBucketInfo omBucketInfo : volBucketInfoMap.values()) {
+      metadataManager.getBucketTable().putWithBatch(batchOp,
+          metadataManager.getBucketKey(omBucketInfo.getVolumeName(),
+              omBucketInfo.getBucketName()), omBucketInfo);
+    }
+    for (OmVolumeArgs volArgs : volumeArgsMap.values()) {
+      metadataManager.getVolumeTable().putWithBatch(batchOp, volArgs.getVolume(), volArgs);
+    }
+  }
+}