From 319865a28358b42c876526bb51d373e64ad4dfea Mon Sep 17 00:00:00 2001 From: Chi-Hsuan Huang Date: Fri, 14 Aug 2026 00:21:31 +0800 Subject: [PATCH 1/3] HDDS-16127. Fix quota repair usedBytes for versioned buckets --- .../ozone/om/service/QuotaRepairTask.java | 32 ++- .../ozone/om/service/TestQuotaRepairTask.java | 219 ++++++++++++++++++ 2 files changed, 250 insertions(+), 1 deletion(-) diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/service/QuotaRepairTask.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/service/QuotaRepairTask.java index 2805f84c8372..0d5cd8c658a9 100644 --- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/service/QuotaRepairTask.java +++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/service/QuotaRepairTask.java @@ -66,6 +66,9 @@ import org.apache.hadoop.ozone.om.helpers.BucketLayout; import org.apache.hadoop.ozone.om.helpers.OmBucketInfo; import org.apache.hadoop.ozone.om.helpers.OmKeyInfo; +import org.apache.hadoop.ozone.om.helpers.OmKeyLocationInfo; +import org.apache.hadoop.ozone.om.helpers.OmKeyLocationInfoGroup; +import org.apache.hadoop.ozone.om.helpers.QuotaUtil; import org.apache.hadoop.ozone.om.helpers.RepeatedOmKeyInfo; import org.apache.hadoop.ozone.om.helpers.SnapshotInfo; import org.apache.hadoop.ozone.om.ratis.utils.OzoneManagerRatisUtils; @@ -648,9 +651,36 @@ private static void extractCount( if (haveValue) { VALUE value = kv.getValue(); if (value instanceof OmKeyInfo) { - usage.incrSpace(((OmKeyInfo) value).getReplicatedSize()); + OmKeyInfo keyInfo = (OmKeyInfo) value; + if (keyInfo.getKeyLocationVersions().size() > 1) { + usage.incrSpace(getRetainedVersionsSize(keyInfo)); + } else { + usage.incrSpace(keyInfo.getReplicatedSize()); + } + } + } + } + + /** + * Size of every version a key retains. {@code dataSize} covers the latest version only, while versioning keeps + * the blocks of all of them, so the recount has to walk the version groups. + * + *

Each version is converted once, as {@link org.apache.hadoop.ozone.om.request.key.OMKeyCommitRequest} charges + * it at commit time. {@code OMKeyRequest#sumBlockLengths} is not reused here because it converts every block + * separately, which for EC adds parity per block and would make repair report more than the bucket was charged. + */ + private static long getRetainedVersionsSize(OmKeyInfo keyInfo) { + long replicatedSize = 0; + for (OmKeyLocationInfoGroup group : keyInfo.getKeyLocationVersions()) { + long versionSize = 0; + for (List locations : group.getLocationLists()) { + for (OmKeyLocationInfo location : locations) { + versionSize += location.getLength(); + } } + replicatedSize += QuotaUtil.getReplicatedSize(versionSize, keyInfo.getReplicationConfig()); } + return replicatedSize; } private static synchronized void updateCountToBucketInfo( diff --git a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/service/TestQuotaRepairTask.java b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/service/TestQuotaRepairTask.java index a956fb3cb216..fdd35cb158a0 100644 --- a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/service/TestQuotaRepairTask.java +++ b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/service/TestQuotaRepairTask.java @@ -29,20 +29,31 @@ import static org.mockito.Mockito.when; import java.io.IOException; +import java.util.ArrayList; +import java.util.List; +import java.util.UUID; import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicReference; +import java.util.stream.Collectors; +import org.apache.hadoop.hdds.client.BlockID; +import org.apache.hadoop.hdds.client.ECReplicationConfig; import org.apache.hadoop.hdds.client.RatisReplicationConfig; +import org.apache.hadoop.hdds.client.ReplicationConfig; +import org.apache.hadoop.hdds.protocol.proto.HddsProtos; import org.apache.hadoop.hdds.utils.db.BatchOperation; import org.apache.hadoop.hdds.utils.db.cache.CacheKey; import org.apache.hadoop.hdds.utils.db.cache.CacheValue; +import org.apache.hadoop.ozone.ClientVersion; import org.apache.hadoop.ozone.om.helpers.BucketLayout; import org.apache.hadoop.ozone.om.helpers.OmBucketInfo; import org.apache.hadoop.ozone.om.helpers.OmKeyInfo; +import org.apache.hadoop.ozone.om.helpers.OmKeyLocationInfo; import org.apache.hadoop.ozone.om.helpers.OmVolumeArgs; import org.apache.hadoop.ozone.om.helpers.RepeatedOmKeyInfo; import org.apache.hadoop.ozone.om.ratis.OzoneManagerRatisServer; import org.apache.hadoop.ozone.om.request.OMRequestTestUtils; +import org.apache.hadoop.ozone.om.request.key.OMKeyCommitRequest; import org.apache.hadoop.ozone.om.request.key.OMKeyRequestTests; import org.apache.hadoop.ozone.om.request.volume.OMQuotaRepairRequest; import org.apache.hadoop.ozone.om.response.OMClientResponse; @@ -325,6 +336,214 @@ public void testQuotaRepairSnapshotDbDeletedTableQuota() throws Exception { assertEquals(1, repaired.getSnapshotUsedNamespace()); } + @Test + public void testQuotaRepairVersionedBucketUndercount() throws Exception { + AtomicReference ref = mockRatisSubmit(); + String bucketKey = addVersionedBucket(bucketName); + ReplicationConfig ratisOne = RatisReplicationConfig.getInstance(ONE); + + // v0: 300 bytes, RATIS ONE so replicated size == data size + commitVersion(ratisOne, 0L, 1L, null, 300L); + OmBucketInfo afterV0 = omMetadataManager.getBucketTable().get(bucketKey); + assertEquals(300, afterV0.getUsedBytes()); + assertEquals(1, afterV0.getUsedNamespace()); + + // v1: 600 bytes overwrite, old version stays on disk + String ozoneKey = omMetadataManager.getOzoneKey(volumeName, bucketName, keyName); + OmKeyInfo committedV0 = omMetadataManager.getKeyTable(BucketLayout.OBJECT_STORE).get(ozoneKey); + commitVersion(ratisOne, 1L, 2L, committedV0, 600L); + + OmBucketInfo live = omMetadataManager.getBucketTable().get(bucketKey); + assertEquals(900, live.getUsedBytes(), "live accounting keeps both versions"); + + // key table entry keeps only the latest version's dataSize, but holds both location groups + OmKeyInfo committedV1 = omMetadataManager.getKeyTable(BucketLayout.OBJECT_STORE).get(ozoneKey); + assertEquals(600, committedV1.getDataSize()); + assertEquals(2, committedV1.getKeyLocationVersions().size()); + + OmBucketInfo repaired = runRepair(ref, bucketKey); + assertEquals(900, repaired.getUsedBytes(), "repair must not collapse multi-version keys"); + // usedNamespace counts keys, not versions: the delete path refunds one unit per key + assertEquals(1, repaired.getUsedNamespace(), "a multi-version key uses one namespace unit"); + } + + /** + * EC parity is added per stripe, so converting a version as a whole and converting its blocks one by one give + * different totals. {@link OMKeyCommitRequest} converts each commit as a whole, so the recount has to do the + * same, otherwise repair would move a counter that live accounting had right. + */ + @Test + public void testQuotaRepairVersionedBucketWithECKey() throws Exception { + AtomicReference ref = mockRatisSubmit(); + String bucketKey = addVersionedBucket(bucketName); + // data stripe is 3 * 1024 bytes, so both versions below carry a partial stripe + ReplicationConfig ec = new ECReplicationConfig(3, 2, ECReplicationConfig.EcCodec.RS, 1024); + + // v0: 2048 bytes of data over two blocks, charged as 2048 + 1024 * 2 parity + commitVersion(ec, 0L, 1L, null, 1024L, 1024L); + + // v1: 3072 bytes of data in one block, charged as 3072 + 1024 * 2 parity + String ozoneKey = omMetadataManager.getOzoneKey(volumeName, bucketName, keyName); + OmKeyInfo committedV0 = omMetadataManager.getKeyTable(BucketLayout.OBJECT_STORE).get(ozoneKey); + commitVersion(ec, 1L, 2L, committedV0, 3072L); + + OmBucketInfo live = omMetadataManager.getBucketTable().get(bucketKey); + assertEquals(9216, live.getUsedBytes()); + + // converting v0's two blocks separately would add parity twice and report 11264 instead + OmBucketInfo repaired = runRepair(ref, bucketKey); + assertEquals(live.getUsedBytes(), repaired.getUsedBytes(), + "repair must reproduce what commit charged, not re-derive it per block"); + assertEquals(1, repaired.getUsedNamespace()); + } + + /** + * The FSO key table is keyed by volume and bucket id, so the recount looks the bucket up in a different map + * than the OBS key table does. Build the multi-version entry directly and check the versioning aware branch + * is reached for FSO too. + */ + @Test + public void testQuotaRepairVersionedFsoBucket() throws Exception { + AtomicReference ref = mockRatisSubmit(); + OMRequestTestUtils.addVolumeAndBucketToDB(volumeName, omMetadataManager, + OmBucketInfo.newBuilder() + .setVolumeName(volumeName) + .setBucketName(bucketName) + .setBucketLayout(BucketLayout.FILE_SYSTEM_OPTIMIZED) + .setIsVersionEnabled(true)); + String bucketKey = omMetadataManager.getBucketKey(volumeName, bucketName); + long bucketId = omMetadataManager.getBucketId(volumeName, bucketName); + + String fileName = "file0"; + OmKeyInfo omKeyInfo = OMRequestTestUtils.createOmKeyInfo(volumeName, bucketName, fileName, + RatisReplicationConfig.getInstance(ONE)) + .setObjectID(bucketId + 1) + .setParentObjectID(bucketId) + .setUpdateID(1L) + .build(); + omKeyInfo.setKeyName(fileName); + omKeyInfo.appendNewBlocks(blockLocations(0L, 300L), false); + omKeyInfo.addNewVersion(blockLocations(1L, 600L), false, true); + omKeyInfo.setDataSize(600); + OMRequestTestUtils.addFileToKeyTable(false, false, fileName, omKeyInfo, -1, 1L, omMetadataManager); + + // start from zeroed counters so the assertion below is the recount itself, not a delta + zeroOutBucketUsedBytes(volumeName, bucketName, 1L); + OmBucketInfo zeroed = omMetadataManager.getBucketTable().get(bucketKey); + assertEquals(0, zeroed.getUsedBytes()); + + OmBucketInfo repaired = runRepair(ref, bucketKey); + assertEquals(900, repaired.getUsedBytes(), "FSO recount must reach the versioning aware branch"); + assertEquals(1, repaired.getUsedNamespace()); + } + + private static List blockLocations(long versionNum, long... blockLengths) { + List locations = new ArrayList<>(blockLengths.length); + for (int i = 0; i < blockLengths.length; i++) { + locations.add(new OmKeyLocationInfo.Builder() + .setBlockID(new BlockID(CONTAINER_ID + versionNum, LOCAL_ID + versionNum * 100 + i)) + .setLength(blockLengths[i]) + .setOffset(0) + .setCreateVersion(versionNum) + .build()); + } + return locations; + } + + private AtomicReference mockRatisSubmit() throws Exception { + OzoneManagerProtocolProtos.OMResponse respMock = mock(OzoneManagerProtocolProtos.OMResponse.class); + when(respMock.getSuccess()).thenReturn(true); + OzoneManagerRatisServer ratisServerMock = mock(OzoneManagerRatisServer.class); + AtomicReference ref = new AtomicReference<>(); + doAnswer(invocation -> { + ref.set(invocation.getArgument(0, OzoneManagerProtocolProtos.OMRequest.class)); + return respMock; + }).when(ratisServerMock).submitRequest(any(), any(), anyLong()); + when(ozoneManager.getOmRatisServer()).thenReturn(ratisServerMock); + return ref; + } + + private String addVersionedBucket(String bucket) throws Exception { + OMRequestTestUtils.addVolumeAndBucketToDB(volumeName, omMetadataManager, + OmBucketInfo.newBuilder() + .setVolumeName(volumeName) + .setBucketName(bucket) + .setBucketLayout(BucketLayout.OBJECT_STORE) + .setIsVersionEnabled(true)); + return omMetadataManager.getBucketKey(volumeName, bucket); + } + + private OmBucketInfo runRepair(AtomicReference ref, + String bucketKey) throws Exception { + QuotaRepairTask quotaRepairTask = new QuotaRepairTask(ozoneManager); + assertTrue(awaitRepair(quotaRepairTask.repair())); + + OMQuotaRepairRequest omQuotaRepairRequest = new OMQuotaRepairRequest(ref.get()); + OMClientResponse omClientResponse = omQuotaRepairRequest.validateAndUpdateCache(ozoneManager, 3); + BatchOperation batchOperation = omMetadataManager.getStore().initBatchOperation(); + ((OMQuotaRepairResponse) omClientResponse).addToDBBatch(omMetadataManager, batchOperation); + omMetadataManager.getStore().commitBatchOperation(batchOperation); + + return omMetadataManager.getBucketTable().get(bucketKey); + } + + /** + * Drive one real {@link OMKeyCommitRequest} on the versioning-enabled bucket. When {@code previous} is set, + * the open key is built the way {@code OMKeyRequest.prepareFileInfo} builds it for an overwrite: old location + * groups kept, dataSize accumulated. + */ + private void commitVersion(ReplicationConfig repConfig, long versionNum, long trxnLogIndex, + OmKeyInfo previous, long... blockLengths) throws Exception { + long writerClientId = clientID + versionNum; + long size = 0; + for (long blockLength : blockLengths) { + size += blockLength; + } + List locations = blockLocations(versionNum, blockLengths); + + if (previous == null) { + OMRequestTestUtils.addKeyToTable(true, false, volumeName, bucketName, keyName, writerClientId, + repConfig, trxnLogIndex, omMetadataManager, locations, versionNum); + } else { + OmKeyInfo openKeyInfo = previous.copyObject(); + openKeyInfo.addNewVersion(locations, false, true); + openKeyInfo.setDataSize(previous.getDataSize() + size); + OMRequestTestUtils.addKeyToTable(true, false, openKeyInfo, writerClientId, trxnLogIndex, omMetadataManager); + } + + OzoneManagerProtocolProtos.KeyArgs.Builder keyArgs = OzoneManagerProtocolProtos.KeyArgs.newBuilder() + .setVolumeName(volumeName) + .setBucketName(bucketName) + .setKeyName(keyName) + .setDataSize(size) + .setType(repConfig.getReplicationType()) + .addAllKeyLocations(locations.stream() + .map(l -> l.getProtobuf(false, ClientVersion.CURRENT_VERSION)) + .collect(Collectors.toList())); + if (repConfig.getReplicationType() == HddsProtos.ReplicationType.EC) { + keyArgs.setEcReplicationConfig(((ECReplicationConfig) repConfig).toProto()); + } else { + keyArgs.setFactor(ReplicationConfig.getLegacyFactor(repConfig)); + } + OzoneManagerProtocolProtos.OMRequest omRequest = OzoneManagerProtocolProtos.OMRequest.newBuilder() + .setCmdType(OzoneManagerProtocolProtos.Type.CommitKey) + .setCommitKeyRequest(OzoneManagerProtocolProtos.CommitKeyRequest.newBuilder() + .setKeyArgs(keyArgs.build()) + .setClientID(writerClientId) + .build()) + .setClientId(UUID.randomUUID().toString()) + .build(); + + OMKeyCommitRequest commitRequest = new OMKeyCommitRequest(omRequest, BucketLayout.OBJECT_STORE); + OMClientResponse response = new OMKeyCommitRequest(commitRequest.preExecute(ozoneManager), + BucketLayout.OBJECT_STORE).validateAndUpdateCache(ozoneManager, trxnLogIndex); + assertEquals(OzoneManagerProtocolProtos.Status.OK, response.getOMResponse().getStatus()); + + BatchOperation batch = omMetadataManager.getStore().initBatchOperation(); + response.checkAndUpdateDB(omMetadataManager, batch); + omMetadataManager.getStore().commitBatchOperation(batch); + } + private void zeroOutBucketUsedBytes(String volumeName, String bucketName, long trxnLogIndex) throws IOException { From 97a9df2ece3035259a734336da2d40853e5e1032 Mon Sep 17 00:00:00 2001 From: Chi-Hsuan Huang Date: Fri, 14 Aug 2026 00:45:47 +0800 Subject: [PATCH 2/3] HDDS-16127. Count only the blocks each version created in quota repair --- .../ozone/om/service/QuotaRepairTask.java | 13 +++--- .../ozone/om/service/TestQuotaRepairTask.java | 45 ++++++++++++++++--- 2 files changed, 47 insertions(+), 11 deletions(-) diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/service/QuotaRepairTask.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/service/QuotaRepairTask.java index 0d5cd8c658a9..ae98c4e51f50 100644 --- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/service/QuotaRepairTask.java +++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/service/QuotaRepairTask.java @@ -665,18 +665,19 @@ private static void extractCount( * Size of every version a key retains. {@code dataSize} covers the latest version only, while versioning keeps * the blocks of all of them, so the recount has to walk the version groups. * + *

Only the blocks a group created itself count: before HDDS-5472 a group also copied the earlier groups' blocks, + * and those keys were never migrated. + * *

Each version is converted once, as {@link org.apache.hadoop.ozone.om.request.key.OMKeyCommitRequest} charges - * it at commit time. {@code OMKeyRequest#sumBlockLengths} is not reused here because it converts every block - * separately, which for EC adds parity per block and would make repair report more than the bucket was charged. + * it at commit time. {@code OMKeyRequest#sumBlockLengths} is not reused because it converts every block separately, + * which for EC rounds a partial stripe up per block and reports more than the bucket was charged. */ private static long getRetainedVersionsSize(OmKeyInfo keyInfo) { long replicatedSize = 0; for (OmKeyLocationInfoGroup group : keyInfo.getKeyLocationVersions()) { long versionSize = 0; - for (List locations : group.getLocationLists()) { - for (OmKeyLocationInfo location : locations) { - versionSize += location.getLength(); - } + for (OmKeyLocationInfo location : group.getBlocksLatestVersionOnly()) { + versionSize += location.getLength(); } replicatedSize += QuotaUtil.getReplicatedSize(versionSize, keyInfo.getReplicationConfig()); } diff --git a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/service/TestQuotaRepairTask.java b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/service/TestQuotaRepairTask.java index fdd35cb158a0..128e849c7492 100644 --- a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/service/TestQuotaRepairTask.java +++ b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/service/TestQuotaRepairTask.java @@ -30,7 +30,10 @@ import java.io.IOException; import java.util.ArrayList; +import java.util.Arrays; +import java.util.HashMap; import java.util.List; +import java.util.Map; import java.util.UUID; import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeUnit; @@ -49,6 +52,7 @@ import org.apache.hadoop.ozone.om.helpers.OmBucketInfo; import org.apache.hadoop.ozone.om.helpers.OmKeyInfo; import org.apache.hadoop.ozone.om.helpers.OmKeyLocationInfo; +import org.apache.hadoop.ozone.om.helpers.OmKeyLocationInfoGroup; import org.apache.hadoop.ozone.om.helpers.OmVolumeArgs; import org.apache.hadoop.ozone.om.helpers.RepeatedOmKeyInfo; import org.apache.hadoop.ozone.om.ratis.OzoneManagerRatisServer; @@ -368,9 +372,9 @@ public void testQuotaRepairVersionedBucketUndercount() throws Exception { } /** - * EC parity is added per stripe, so converting a version as a whole and converting its blocks one by one give - * different totals. {@link OMKeyCommitRequest} converts each commit as a whole, so the recount has to do the - * same, otherwise repair would move a counter that live accounting had right. + * EC rounds a partial stripe up to a full parity chunk once per conversion, so converting a version as a whole + * and converting its blocks one by one give different totals. {@link OMKeyCommitRequest} converts each commit as + * a whole, so the recount has to do the same, otherwise repair would move a counter that live accounting had right. */ @Test public void testQuotaRepairVersionedBucketWithECKey() throws Exception { @@ -390,13 +394,44 @@ public void testQuotaRepairVersionedBucketWithECKey() throws Exception { OmBucketInfo live = omMetadataManager.getBucketTable().get(bucketKey); assertEquals(9216, live.getUsedBytes()); - // converting v0's two blocks separately would add parity twice and report 11264 instead + // converting v0's two blocks separately would round the partial stripe up twice and report 11264 instead OmBucketInfo repaired = runRepair(ref, bucketKey); - assertEquals(live.getUsedBytes(), repaired.getUsedBytes(), + assertEquals(9216, repaired.getUsedBytes(), "repair must reproduce what commit charged, not re-derive it per block"); assertEquals(1, repaired.getUsedNamespace()); } + /** + * Keys written before HDDS-5472 carry copies of the earlier groups' blocks and were never migrated, so summing + * whole groups would charge v0 twice here. + */ + @Test + public void testQuotaRepairLegacyKeyWithCopiedVersions() throws Exception { + AtomicReference ref = mockRatisSubmit(); + String bucketKey = addVersionedBucket(bucketName); + + List v0 = blockLocations(0L, 300L); + Map> legacyGroup = new HashMap<>(); + legacyGroup.put(0L, new ArrayList<>(v0)); + legacyGroup.put(1L, blockLocations(1L, 600L)); + + OmKeyInfo omKeyInfo = OMRequestTestUtils.createOmKeyInfo(volumeName, bucketName, keyName, + RatisReplicationConfig.getInstance(ONE)) + .setOmKeyLocationInfos(Arrays.asList( + new OmKeyLocationInfoGroup(0L, v0), + new OmKeyLocationInfoGroup(1L, legacyGroup))) + .setDataSize(600) + .setUpdateID(1L) + .build(); + OMRequestTestUtils.addKeyToTable(false, false, omKeyInfo, clientID, 1L, omMetadataManager); + + zeroOutBucketUsedBytes(volumeName, bucketName, 1L); + + OmBucketInfo repaired = runRepair(ref, bucketKey); + assertEquals(900, repaired.getUsedBytes(), "the copy of v0 in group 1 must not be charged again"); + assertEquals(1, repaired.getUsedNamespace()); + } + /** * The FSO key table is keyed by volume and bucket id, so the recount looks the bucket up in a different map * than the OBS key table does. Build the multi-version entry directly and check the versioning aware branch From 5cbd9835d59df17563ebcc2d4908c7424cb8408e Mon Sep 17 00:00:00 2001 From: Chi-Hsuan Huang Date: Fri, 14 Aug 2026 22:08:25 +0800 Subject: [PATCH 3/3] HDDS-16127. Document that OmKeyInfo holds a single key version --- .../hadoop/ozone/om/helpers/OmKeyInfo.java | 5 + .../ozone/om/service/QuotaRepairTask.java | 33 +-- .../ozone/om/service/TestQuotaRepairTask.java | 254 ------------------ 3 files changed, 6 insertions(+), 286 deletions(-) diff --git a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OmKeyInfo.java b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OmKeyInfo.java index ab4da4badd90..bbe19792b789 100644 --- a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OmKeyInfo.java +++ b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OmKeyInfo.java @@ -74,6 +74,11 @@ public final class OmKeyInfo extends WithParentObjectId // name of key client specified private String keyName; private long dataSize; + /** + * Block locations of the key, one group per key version. No write path produces more than one group today, so a + * key holds a single version and {@code dataSize} covers all of its blocks. Object versioning (HDDS-15728) keeps + * each object version in its own OmKeyInfo rather than adding groups here. + */ private List keyLocationVersions; private final long creationTime; private long modificationTime; diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/service/QuotaRepairTask.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/service/QuotaRepairTask.java index ae98c4e51f50..2805f84c8372 100644 --- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/service/QuotaRepairTask.java +++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/service/QuotaRepairTask.java @@ -66,9 +66,6 @@ import org.apache.hadoop.ozone.om.helpers.BucketLayout; import org.apache.hadoop.ozone.om.helpers.OmBucketInfo; import org.apache.hadoop.ozone.om.helpers.OmKeyInfo; -import org.apache.hadoop.ozone.om.helpers.OmKeyLocationInfo; -import org.apache.hadoop.ozone.om.helpers.OmKeyLocationInfoGroup; -import org.apache.hadoop.ozone.om.helpers.QuotaUtil; import org.apache.hadoop.ozone.om.helpers.RepeatedOmKeyInfo; import org.apache.hadoop.ozone.om.helpers.SnapshotInfo; import org.apache.hadoop.ozone.om.ratis.utils.OzoneManagerRatisUtils; @@ -651,37 +648,9 @@ private static void extractCount( if (haveValue) { VALUE value = kv.getValue(); if (value instanceof OmKeyInfo) { - OmKeyInfo keyInfo = (OmKeyInfo) value; - if (keyInfo.getKeyLocationVersions().size() > 1) { - usage.incrSpace(getRetainedVersionsSize(keyInfo)); - } else { - usage.incrSpace(keyInfo.getReplicatedSize()); - } - } - } - } - - /** - * Size of every version a key retains. {@code dataSize} covers the latest version only, while versioning keeps - * the blocks of all of them, so the recount has to walk the version groups. - * - *

Only the blocks a group created itself count: before HDDS-5472 a group also copied the earlier groups' blocks, - * and those keys were never migrated. - * - *

Each version is converted once, as {@link org.apache.hadoop.ozone.om.request.key.OMKeyCommitRequest} charges - * it at commit time. {@code OMKeyRequest#sumBlockLengths} is not reused because it converts every block separately, - * which for EC rounds a partial stripe up per block and reports more than the bucket was charged. - */ - private static long getRetainedVersionsSize(OmKeyInfo keyInfo) { - long replicatedSize = 0; - for (OmKeyLocationInfoGroup group : keyInfo.getKeyLocationVersions()) { - long versionSize = 0; - for (OmKeyLocationInfo location : group.getBlocksLatestVersionOnly()) { - versionSize += location.getLength(); + usage.incrSpace(((OmKeyInfo) value).getReplicatedSize()); } - replicatedSize += QuotaUtil.getReplicatedSize(versionSize, keyInfo.getReplicationConfig()); } - return replicatedSize; } private static synchronized void updateCountToBucketInfo( diff --git a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/service/TestQuotaRepairTask.java b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/service/TestQuotaRepairTask.java index 128e849c7492..a956fb3cb216 100644 --- a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/service/TestQuotaRepairTask.java +++ b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/service/TestQuotaRepairTask.java @@ -29,35 +29,20 @@ import static org.mockito.Mockito.when; import java.io.IOException; -import java.util.ArrayList; -import java.util.Arrays; -import java.util.HashMap; -import java.util.List; -import java.util.Map; -import java.util.UUID; import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicReference; -import java.util.stream.Collectors; -import org.apache.hadoop.hdds.client.BlockID; -import org.apache.hadoop.hdds.client.ECReplicationConfig; import org.apache.hadoop.hdds.client.RatisReplicationConfig; -import org.apache.hadoop.hdds.client.ReplicationConfig; -import org.apache.hadoop.hdds.protocol.proto.HddsProtos; import org.apache.hadoop.hdds.utils.db.BatchOperation; import org.apache.hadoop.hdds.utils.db.cache.CacheKey; import org.apache.hadoop.hdds.utils.db.cache.CacheValue; -import org.apache.hadoop.ozone.ClientVersion; import org.apache.hadoop.ozone.om.helpers.BucketLayout; import org.apache.hadoop.ozone.om.helpers.OmBucketInfo; import org.apache.hadoop.ozone.om.helpers.OmKeyInfo; -import org.apache.hadoop.ozone.om.helpers.OmKeyLocationInfo; -import org.apache.hadoop.ozone.om.helpers.OmKeyLocationInfoGroup; import org.apache.hadoop.ozone.om.helpers.OmVolumeArgs; import org.apache.hadoop.ozone.om.helpers.RepeatedOmKeyInfo; import org.apache.hadoop.ozone.om.ratis.OzoneManagerRatisServer; import org.apache.hadoop.ozone.om.request.OMRequestTestUtils; -import org.apache.hadoop.ozone.om.request.key.OMKeyCommitRequest; import org.apache.hadoop.ozone.om.request.key.OMKeyRequestTests; import org.apache.hadoop.ozone.om.request.volume.OMQuotaRepairRequest; import org.apache.hadoop.ozone.om.response.OMClientResponse; @@ -340,245 +325,6 @@ public void testQuotaRepairSnapshotDbDeletedTableQuota() throws Exception { assertEquals(1, repaired.getSnapshotUsedNamespace()); } - @Test - public void testQuotaRepairVersionedBucketUndercount() throws Exception { - AtomicReference ref = mockRatisSubmit(); - String bucketKey = addVersionedBucket(bucketName); - ReplicationConfig ratisOne = RatisReplicationConfig.getInstance(ONE); - - // v0: 300 bytes, RATIS ONE so replicated size == data size - commitVersion(ratisOne, 0L, 1L, null, 300L); - OmBucketInfo afterV0 = omMetadataManager.getBucketTable().get(bucketKey); - assertEquals(300, afterV0.getUsedBytes()); - assertEquals(1, afterV0.getUsedNamespace()); - - // v1: 600 bytes overwrite, old version stays on disk - String ozoneKey = omMetadataManager.getOzoneKey(volumeName, bucketName, keyName); - OmKeyInfo committedV0 = omMetadataManager.getKeyTable(BucketLayout.OBJECT_STORE).get(ozoneKey); - commitVersion(ratisOne, 1L, 2L, committedV0, 600L); - - OmBucketInfo live = omMetadataManager.getBucketTable().get(bucketKey); - assertEquals(900, live.getUsedBytes(), "live accounting keeps both versions"); - - // key table entry keeps only the latest version's dataSize, but holds both location groups - OmKeyInfo committedV1 = omMetadataManager.getKeyTable(BucketLayout.OBJECT_STORE).get(ozoneKey); - assertEquals(600, committedV1.getDataSize()); - assertEquals(2, committedV1.getKeyLocationVersions().size()); - - OmBucketInfo repaired = runRepair(ref, bucketKey); - assertEquals(900, repaired.getUsedBytes(), "repair must not collapse multi-version keys"); - // usedNamespace counts keys, not versions: the delete path refunds one unit per key - assertEquals(1, repaired.getUsedNamespace(), "a multi-version key uses one namespace unit"); - } - - /** - * EC rounds a partial stripe up to a full parity chunk once per conversion, so converting a version as a whole - * and converting its blocks one by one give different totals. {@link OMKeyCommitRequest} converts each commit as - * a whole, so the recount has to do the same, otherwise repair would move a counter that live accounting had right. - */ - @Test - public void testQuotaRepairVersionedBucketWithECKey() throws Exception { - AtomicReference ref = mockRatisSubmit(); - String bucketKey = addVersionedBucket(bucketName); - // data stripe is 3 * 1024 bytes, so both versions below carry a partial stripe - ReplicationConfig ec = new ECReplicationConfig(3, 2, ECReplicationConfig.EcCodec.RS, 1024); - - // v0: 2048 bytes of data over two blocks, charged as 2048 + 1024 * 2 parity - commitVersion(ec, 0L, 1L, null, 1024L, 1024L); - - // v1: 3072 bytes of data in one block, charged as 3072 + 1024 * 2 parity - String ozoneKey = omMetadataManager.getOzoneKey(volumeName, bucketName, keyName); - OmKeyInfo committedV0 = omMetadataManager.getKeyTable(BucketLayout.OBJECT_STORE).get(ozoneKey); - commitVersion(ec, 1L, 2L, committedV0, 3072L); - - OmBucketInfo live = omMetadataManager.getBucketTable().get(bucketKey); - assertEquals(9216, live.getUsedBytes()); - - // converting v0's two blocks separately would round the partial stripe up twice and report 11264 instead - OmBucketInfo repaired = runRepair(ref, bucketKey); - assertEquals(9216, repaired.getUsedBytes(), - "repair must reproduce what commit charged, not re-derive it per block"); - assertEquals(1, repaired.getUsedNamespace()); - } - - /** - * Keys written before HDDS-5472 carry copies of the earlier groups' blocks and were never migrated, so summing - * whole groups would charge v0 twice here. - */ - @Test - public void testQuotaRepairLegacyKeyWithCopiedVersions() throws Exception { - AtomicReference ref = mockRatisSubmit(); - String bucketKey = addVersionedBucket(bucketName); - - List v0 = blockLocations(0L, 300L); - Map> legacyGroup = new HashMap<>(); - legacyGroup.put(0L, new ArrayList<>(v0)); - legacyGroup.put(1L, blockLocations(1L, 600L)); - - OmKeyInfo omKeyInfo = OMRequestTestUtils.createOmKeyInfo(volumeName, bucketName, keyName, - RatisReplicationConfig.getInstance(ONE)) - .setOmKeyLocationInfos(Arrays.asList( - new OmKeyLocationInfoGroup(0L, v0), - new OmKeyLocationInfoGroup(1L, legacyGroup))) - .setDataSize(600) - .setUpdateID(1L) - .build(); - OMRequestTestUtils.addKeyToTable(false, false, omKeyInfo, clientID, 1L, omMetadataManager); - - zeroOutBucketUsedBytes(volumeName, bucketName, 1L); - - OmBucketInfo repaired = runRepair(ref, bucketKey); - assertEquals(900, repaired.getUsedBytes(), "the copy of v0 in group 1 must not be charged again"); - assertEquals(1, repaired.getUsedNamespace()); - } - - /** - * The FSO key table is keyed by volume and bucket id, so the recount looks the bucket up in a different map - * than the OBS key table does. Build the multi-version entry directly and check the versioning aware branch - * is reached for FSO too. - */ - @Test - public void testQuotaRepairVersionedFsoBucket() throws Exception { - AtomicReference ref = mockRatisSubmit(); - OMRequestTestUtils.addVolumeAndBucketToDB(volumeName, omMetadataManager, - OmBucketInfo.newBuilder() - .setVolumeName(volumeName) - .setBucketName(bucketName) - .setBucketLayout(BucketLayout.FILE_SYSTEM_OPTIMIZED) - .setIsVersionEnabled(true)); - String bucketKey = omMetadataManager.getBucketKey(volumeName, bucketName); - long bucketId = omMetadataManager.getBucketId(volumeName, bucketName); - - String fileName = "file0"; - OmKeyInfo omKeyInfo = OMRequestTestUtils.createOmKeyInfo(volumeName, bucketName, fileName, - RatisReplicationConfig.getInstance(ONE)) - .setObjectID(bucketId + 1) - .setParentObjectID(bucketId) - .setUpdateID(1L) - .build(); - omKeyInfo.setKeyName(fileName); - omKeyInfo.appendNewBlocks(blockLocations(0L, 300L), false); - omKeyInfo.addNewVersion(blockLocations(1L, 600L), false, true); - omKeyInfo.setDataSize(600); - OMRequestTestUtils.addFileToKeyTable(false, false, fileName, omKeyInfo, -1, 1L, omMetadataManager); - - // start from zeroed counters so the assertion below is the recount itself, not a delta - zeroOutBucketUsedBytes(volumeName, bucketName, 1L); - OmBucketInfo zeroed = omMetadataManager.getBucketTable().get(bucketKey); - assertEquals(0, zeroed.getUsedBytes()); - - OmBucketInfo repaired = runRepair(ref, bucketKey); - assertEquals(900, repaired.getUsedBytes(), "FSO recount must reach the versioning aware branch"); - assertEquals(1, repaired.getUsedNamespace()); - } - - private static List blockLocations(long versionNum, long... blockLengths) { - List locations = new ArrayList<>(blockLengths.length); - for (int i = 0; i < blockLengths.length; i++) { - locations.add(new OmKeyLocationInfo.Builder() - .setBlockID(new BlockID(CONTAINER_ID + versionNum, LOCAL_ID + versionNum * 100 + i)) - .setLength(blockLengths[i]) - .setOffset(0) - .setCreateVersion(versionNum) - .build()); - } - return locations; - } - - private AtomicReference mockRatisSubmit() throws Exception { - OzoneManagerProtocolProtos.OMResponse respMock = mock(OzoneManagerProtocolProtos.OMResponse.class); - when(respMock.getSuccess()).thenReturn(true); - OzoneManagerRatisServer ratisServerMock = mock(OzoneManagerRatisServer.class); - AtomicReference ref = new AtomicReference<>(); - doAnswer(invocation -> { - ref.set(invocation.getArgument(0, OzoneManagerProtocolProtos.OMRequest.class)); - return respMock; - }).when(ratisServerMock).submitRequest(any(), any(), anyLong()); - when(ozoneManager.getOmRatisServer()).thenReturn(ratisServerMock); - return ref; - } - - private String addVersionedBucket(String bucket) throws Exception { - OMRequestTestUtils.addVolumeAndBucketToDB(volumeName, omMetadataManager, - OmBucketInfo.newBuilder() - .setVolumeName(volumeName) - .setBucketName(bucket) - .setBucketLayout(BucketLayout.OBJECT_STORE) - .setIsVersionEnabled(true)); - return omMetadataManager.getBucketKey(volumeName, bucket); - } - - private OmBucketInfo runRepair(AtomicReference ref, - String bucketKey) throws Exception { - QuotaRepairTask quotaRepairTask = new QuotaRepairTask(ozoneManager); - assertTrue(awaitRepair(quotaRepairTask.repair())); - - OMQuotaRepairRequest omQuotaRepairRequest = new OMQuotaRepairRequest(ref.get()); - OMClientResponse omClientResponse = omQuotaRepairRequest.validateAndUpdateCache(ozoneManager, 3); - BatchOperation batchOperation = omMetadataManager.getStore().initBatchOperation(); - ((OMQuotaRepairResponse) omClientResponse).addToDBBatch(omMetadataManager, batchOperation); - omMetadataManager.getStore().commitBatchOperation(batchOperation); - - return omMetadataManager.getBucketTable().get(bucketKey); - } - - /** - * Drive one real {@link OMKeyCommitRequest} on the versioning-enabled bucket. When {@code previous} is set, - * the open key is built the way {@code OMKeyRequest.prepareFileInfo} builds it for an overwrite: old location - * groups kept, dataSize accumulated. - */ - private void commitVersion(ReplicationConfig repConfig, long versionNum, long trxnLogIndex, - OmKeyInfo previous, long... blockLengths) throws Exception { - long writerClientId = clientID + versionNum; - long size = 0; - for (long blockLength : blockLengths) { - size += blockLength; - } - List locations = blockLocations(versionNum, blockLengths); - - if (previous == null) { - OMRequestTestUtils.addKeyToTable(true, false, volumeName, bucketName, keyName, writerClientId, - repConfig, trxnLogIndex, omMetadataManager, locations, versionNum); - } else { - OmKeyInfo openKeyInfo = previous.copyObject(); - openKeyInfo.addNewVersion(locations, false, true); - openKeyInfo.setDataSize(previous.getDataSize() + size); - OMRequestTestUtils.addKeyToTable(true, false, openKeyInfo, writerClientId, trxnLogIndex, omMetadataManager); - } - - OzoneManagerProtocolProtos.KeyArgs.Builder keyArgs = OzoneManagerProtocolProtos.KeyArgs.newBuilder() - .setVolumeName(volumeName) - .setBucketName(bucketName) - .setKeyName(keyName) - .setDataSize(size) - .setType(repConfig.getReplicationType()) - .addAllKeyLocations(locations.stream() - .map(l -> l.getProtobuf(false, ClientVersion.CURRENT_VERSION)) - .collect(Collectors.toList())); - if (repConfig.getReplicationType() == HddsProtos.ReplicationType.EC) { - keyArgs.setEcReplicationConfig(((ECReplicationConfig) repConfig).toProto()); - } else { - keyArgs.setFactor(ReplicationConfig.getLegacyFactor(repConfig)); - } - OzoneManagerProtocolProtos.OMRequest omRequest = OzoneManagerProtocolProtos.OMRequest.newBuilder() - .setCmdType(OzoneManagerProtocolProtos.Type.CommitKey) - .setCommitKeyRequest(OzoneManagerProtocolProtos.CommitKeyRequest.newBuilder() - .setKeyArgs(keyArgs.build()) - .setClientID(writerClientId) - .build()) - .setClientId(UUID.randomUUID().toString()) - .build(); - - OMKeyCommitRequest commitRequest = new OMKeyCommitRequest(omRequest, BucketLayout.OBJECT_STORE); - OMClientResponse response = new OMKeyCommitRequest(commitRequest.preExecute(ozoneManager), - BucketLayout.OBJECT_STORE).validateAndUpdateCache(ozoneManager, trxnLogIndex); - assertEquals(OzoneManagerProtocolProtos.Status.OK, response.getOMResponse().getStatus()); - - BatchOperation batch = omMetadataManager.getStore().initBatchOperation(); - response.checkAndUpdateDB(omMetadataManager, batch); - omMetadataManager.getStore().commitBatchOperation(batch); - } - private void zeroOutBucketUsedBytes(String volumeName, String bucketName, long trxnLogIndex) throws IOException {