SaketaChalamchala commented on code in PR #11094: URL: https://github.com/apache/ozone/pull/11094#discussion_r3898984104
########## hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/snapshot/TestOmSnapshotDefragSpaceSavings.java: ########## @@ -0,0 +1,708 @@ +/* + * 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 + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * 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.snapshot; + +import static org.apache.hadoop.hdds.client.ReplicationFactor.ONE; +import static org.apache.hadoop.hdds.client.ReplicationType.RATIS; +import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_REPLICATION; +import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_REPLICATION_TYPE; +import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_SNAPSHOT_DELETING_SERVICE_INTERVAL; +import static org.apache.hadoop.ozone.OzoneConsts.OM_KEY_PREFIX; +import static org.apache.hadoop.ozone.OzoneConsts.ROCKSDB_SST_SUFFIX; +import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_FILESYSTEM_SNAPSHOT_ENABLED_KEY; +import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_SNAPSHOT_DEFRAG_SERVICE_INTERVAL; +import static org.apache.hadoop.ozone.om.OMConfigKeys.SNAPSHOT_DEFRAG_LIMIT_PER_TASK; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.junit.jupiter.api.Assumptions.assumeTrue; + +import java.io.File; +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.HashSet; +import java.util.List; +import java.util.Set; +import java.util.UUID; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.stream.Collectors; +import java.util.stream.Stream; +import org.apache.hadoop.hdds.client.DefaultReplicationConfig; +import org.apache.hadoop.hdds.client.ReplicationConfig; +import org.apache.hadoop.hdds.conf.OzoneConfiguration; +import org.apache.hadoop.hdds.utils.IOUtils; +import org.apache.hadoop.hdds.utils.db.DBStore; +import org.apache.hadoop.hdds.utils.db.ManagedRawSSTFileReader; +import org.apache.hadoop.ozone.DataTestUtil; +import org.apache.hadoop.ozone.MiniOzoneCluster; +import org.apache.hadoop.ozone.client.ObjectStore; +import org.apache.hadoop.ozone.client.OzoneBucket; +import org.apache.hadoop.ozone.client.OzoneClient; +import org.apache.hadoop.ozone.om.OMMetadataManager; +import org.apache.hadoop.ozone.om.OmSnapshotInternalMetrics; +import org.apache.hadoop.ozone.om.OmSnapshotManager; +import org.apache.hadoop.ozone.om.OzoneManager; +import org.apache.hadoop.ozone.om.helpers.BucketLayout; +import org.apache.hadoop.ozone.om.helpers.SnapshotInfo; +import org.apache.ozone.test.GenericTestUtils; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.TestInstance; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.EnumSource; + +/** + * HDDS-13218: integration tests that snapshot defrag reduces checkpoint disk footprint. + * + * <p>Uses inode-aware sizing (matching {@link OMSnapshotDirectoryMetrics}) so hardlinked SST files + * are not double-counted across snapshot checkpoint directories. Version-0 checkpoints hardlink to + * AOS SST files, so their on-disk byte totals are not comparable to materialized post-defrag + * checkpoints. Savings are validated by cross-snapshot SST reference reduction in the chain. + * + * <p>Covers a three-snapshot chain with AOS compactions and insert/overwrite/delete churn on OBS + * and FSO buckets, full-then-incremental defrag paths, footprint checks after deleting the + * middle snapshot and running a follow-up defrag on the remaining youngest snapshot, isolated + * full defrag on a single snapshot, and idempotent repeated defrag on an already-defragged chain. + * + * <p>Uses one mini-cluster for the whole class (1 datanode, replication factor one) because + * assertions inspect OM checkpoint directories only. Shared snapshot-defrag helpers with + * {@link TestOmSnapshotCheckpointDbContent} may be consolidated in a follow-up under HDDS-13003. + */ +@TestInstance(TestInstance.Lifecycle.PER_CLASS) +public class TestOmSnapshotDefragSpaceSavings { + + private static final byte[] TEST_KEY_CONTENT = new byte[] {0x61, 0x62, 0x63}; + private static final byte[] OVERWRITE_KEY_CONTENT = new byte[] {0x64, 0x65, 0x66}; + private static final int INITIAL_KEY_COUNT = 100; + private static final int OVERWRITE_KEY_COUNT = 50; + private static final int DELETE_KEY_COUNT = 25; + private static final int NEW_KEYS_PER_SNAPSHOT = 10; + private static final int CHECKPOINT_WAIT_MS = 120_000; + private static final int PURGE_WAIT_MS = 180_000; + private static final int DEFRAG_WAIT_MS = 600_000; + private static final int KEY_DELETE_WAIT_MS = 60_000; + private static final long FOOTPRINT_TOLERANCE_BYTES = 8192; + private static final long REPEATED_DEFRAG_FOOTPRINT_TOLERANCE_BYTES = 16_384; + private static final DefaultReplicationConfig REPLICATION_CONFIG_ONE = + new DefaultReplicationConfig( + ReplicationConfig.fromTypeAndFactor(RATIS, ONE)); + + private MiniOzoneCluster cluster; + private OzoneConfiguration conf; + private OzoneClient client; + private ObjectStore store; + + @BeforeAll + void initCluster() throws Exception { + startCluster(); + } + + private void startCluster() throws Exception { + assumeTrue(ManagedRawSSTFileReader.tryLoadLibrary(), + "Snapshot defrag requires rocks-tools native library"); + + conf = new OzoneConfiguration(); + conf.setBoolean(OZONE_FILESYSTEM_SNAPSHOT_ENABLED_KEY, true); + // Keep background defrag idle during the test; manual triggerSnapshotDefrag() still requires + // the service to be initialized (interval must be > 0). + conf.setTimeDuration(OZONE_SNAPSHOT_DEFRAG_SERVICE_INTERVAL, 2, TimeUnit.HOURS); + conf.setInt(SNAPSHOT_DEFRAG_LIMIT_PER_TASK, 10); + conf.setTimeDuration(OZONE_SNAPSHOT_DELETING_SERVICE_INTERVAL, 1, TimeUnit.SECONDS); + conf.set(OZONE_REPLICATION, ONE.name()); + conf.set(OZONE_REPLICATION_TYPE, RATIS.name()); + + cluster = MiniOzoneCluster.newBuilder(conf).setNumDatanodes(1).build(); + cluster.waitForClusterToBeReady(); + cluster.waitForPipelineTobeReady( + org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationFactor.ONE, 60_000); + client = cluster.newClient(); + store = client.getObjectStore(); + resumeBackgroundServices(); + } + + @AfterAll + void shutdownCluster() { + IOUtils.closeQuietly(client, cluster); + } + + private void resumeBackgroundServices() { + OzoneManager om = cluster.getOzoneManager(); + om.getKeyManager().getDeletingService().resume(); + om.getKeyManager().getDirDeletingService().resume(); + om.getKeyManager().getSnapshotDeletingService().resume(); + } + + /** + * Three-snapshot chain with churn on OBS and FSO: defrag should reduce aggregate checkpoint + * footprint on both layouts. Runs as separate parameterized invocations on the shared + * mini-cluster (no cluster restart between layouts). The OBS invocation also verifies one full + * defrag and two incremental defrags. + */ + @ParameterizedTest(name = "layout={0}") + @EnumSource(value = BucketLayout.class, names = {"OBJECT_STORE", "FILE_SYSTEM_OPTIMIZED"}) + public void testSnapshotDefragReducesCheckpointFootprintWithChurn(BucketLayout layout) + throws Exception { + runChurnFootprintScenario(layout); + } + + /** + * After an initial defrag pass, deleting the middle snapshot and defragging again should not + * increase the youngest snapshot footprint and should shrink the remaining chain footprint. + */ + @Test + public void testObsSnapshotDefragReducesFootprintAfterMiddleSnapshotPurge() throws Exception { + SnapshotChainSetup setup = createSnapshotChainWithChurn(BucketLayout.OBJECT_STORE); + triggerDefragUntilDone(setup.snapshots); + + SnapshotInfo s2 = setup.snapshots.get(1); + SnapshotInfo s3 = setup.snapshots.get(2); + int s3VersionAfterFirstDefrag = readSnapshotVersion(s3); + CheckpointFootprint s3FootprintAfterFirstDefrag = measureActiveAggregateCheckpointFootprint( + Arrays.asList(s3)); + CheckpointFootprint aggregateAfterFirstDefrag = + measureActiveAggregateCheckpointFootprint(setup.snapshots); + + store.deleteSnapshot(setup.volumeName, setup.bucketName, s2.getName()); + waitForSnapshotPurged(s2); + s3 = loadSnapshotInfo(setup.volumeName, setup.bucketName, s3.getName()); + + triggerDefragUntilVersionIncreases(s3, s3VersionAfterFirstDefrag); + + CheckpointFootprint s3FootprintAfterSecondDefrag = measureActiveAggregateCheckpointFootprint( + Arrays.asList(s3)); + assertTrue( + s3FootprintAfterSecondDefrag.getTotalBytes() + <= s3FootprintAfterFirstDefrag.getTotalBytes() + FOOTPRINT_TOLERANCE_BYTES, + () -> String.format( + "Expected S3 footprint not to grow materially after purge re-defrag: first=%d bytes, " + + "second=%d bytes", + s3FootprintAfterFirstDefrag.getTotalBytes(), s3FootprintAfterSecondDefrag.getTotalBytes())); + + SnapshotInfo s1 = setup.snapshots.get(0); + CheckpointFootprint aggregateAfterSecondDefrag = measureActiveAggregateCheckpointFootprint( + Arrays.asList(s1, s3)); + assertTrue(aggregateAfterSecondDefrag.getTotalBytes() < aggregateAfterFirstDefrag.getTotalBytes(), + () -> String.format( + "Expected remaining chain footprint to shrink after S2 purge: before=%d bytes, after=%d bytes", + aggregateAfterFirstDefrag.getTotalBytes(), aggregateAfterSecondDefrag.getTotalBytes())); + } + + /** + * A lone OBS snapshot should run through the full defrag path, materialize a defragged checkpoint, + * and remove the version-0 directory. Byte savings for a single snapshot are validated on a chain + * in {@link #testSnapshotDefragReducesCheckpointFootprintWithChurn()}. + */ + @Test + public void testObsSingleSnapshotFullDefragReducesCheckpointFootprint() throws Exception { + SnapshotInfo snapshotInfo = createSingleSnapshotWithChurn(BucketLayout.OBJECT_STORE); + List<SnapshotInfo> snapshots = Arrays.asList(snapshotInfo); + + OmSnapshotInternalMetrics metrics = cluster.getOzoneManager().getOmSnapshotIntMetrics(); + long fullDefragBefore = metrics.getNumSnapshotFullDefrag(); + + triggerDefragUntilDone(snapshots); + + assertEquals(1, readSnapshotVersion(snapshotInfo), + "Single snapshot should be at defrag version 1"); + assertNull(snapshotInfo.getPathPreviousSnapshotId(), + "Single snapshot should use the full defrag path"); + assertTrue(metrics.getNumSnapshotFullDefrag() >= fullDefragBefore + 1, + "Expected a full defrag for the lone snapshot"); + assertVersionZeroCheckpointRemoved(snapshotInfo); + assertTrue(isSnapshotDefragComplete(snapshotInfo), + "Single snapshot should be defrag-complete after defrag"); + } + + /** + * Manually re-triggering defrag on an already-defragged three-snapshot chain should not bump + * snapshot local-data versions or SST reference counts; any transient checkpoint growth should + * stay within a small tolerance. + */ + @Test + public void testObsRepeatedDefragDoesNotIncreaseCheckpointFootprint() throws Exception { + List<SnapshotInfo> snapshots = createSnapshotChainWithChurn(BucketLayout.OBJECT_STORE).snapshots; + triggerDefragUntilDone(snapshots); + + CheckpointFootprint footprintAfterFirstDefrag = + measureActiveAggregateCheckpointFootprint(snapshots); + int s1Version = readSnapshotVersion(snapshots.get(0)); + int s2Version = readSnapshotVersion(snapshots.get(1)); + int s3Version = readSnapshotVersion(snapshots.get(2)); + + cluster.getOzoneManager().triggerSnapshotDefrag(false); + waitForDefragCondition("already-defragged chain to settle after re-trigger", + () -> areAllSnapshotsDefragComplete(snapshots)); + + CheckpointFootprint footprintAfterSecondDefrag = + measureActiveAggregateCheckpointFootprint(snapshots); + assertTrue( + footprintAfterSecondDefrag.getTotalBytes() + <= footprintAfterFirstDefrag.getTotalBytes() + REPEATED_DEFRAG_FOOTPRINT_TOLERANCE_BYTES, + () -> String.format( + "Repeated defrag should not materially increase checkpoint bytes: before=%d, after=%d", + footprintAfterFirstDefrag.getTotalBytes(), footprintAfterSecondDefrag.getTotalBytes())); + assertEquals(footprintAfterFirstDefrag.getSstFileCount(), + footprintAfterSecondDefrag.getSstFileCount(), + "Repeated defrag should not increase SST file count"); + assertEquals(s1Version, readSnapshotVersion(snapshots.get(0)), + "Repeated defrag should not bump S1 version"); + assertEquals(s2Version, readSnapshotVersion(snapshots.get(1)), + "Repeated defrag should not bump S2 version"); + assertEquals(s3Version, readSnapshotVersion(snapshots.get(2)), + "Repeated defrag should not bump S3 version"); + } + + private void runChurnFootprintScenario(BucketLayout layout) throws Exception { + List<SnapshotInfo> snapshots = createSnapshotChainWithChurn(layout).snapshots; + CheckpointFootprint duplicateInclusiveBefore = + measureDuplicateInclusiveAggregateFootprint(snapshots, 0); + CheckpointFootprint dedupedBefore = measureAggregateCheckpointFootprint(snapshots, 0); + + OmSnapshotInternalMetrics metrics = cluster.getOzoneManager().getOmSnapshotIntMetrics(); + long fullDefragBefore = metrics.getNumSnapshotFullDefrag(); + long incDefragBefore = metrics.getNumSnapshotIncDefrag(); + + triggerDefragUntilDone(snapshots); + + assertDefragReducedChainFootprint(layout, snapshots, duplicateInclusiveBefore, dedupedBefore); + if (layout == BucketLayout.OBJECT_STORE) { + assertTrue(metrics.getNumSnapshotFullDefrag() >= fullDefragBefore + 1, + "Expected at least one full defrag for the chain head snapshot"); + assertTrue(metrics.getNumSnapshotIncDefrag() >= incDefragBefore + 2, + "Expected incremental defrag for the second and third snapshots"); + assertNull(snapshots.get(0).getPathPreviousSnapshotId(), + "Chain head snapshot should use the full defrag path"); Review Comment: The assertion here is just checking that S1 is at the top of the chain and doesn't really check if it uses a full defrag. If it's difficult to map whether a snapshot uses full/incremental defrag then this assertion can be skipped. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
