HDDS-13003. [POC] Snapshot Defragmentation to reduce storage footprint

hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/OzoneConfigKeys.java

@@ -277,6 +277,12 @@ public final class OzoneConfigKeys {
       OZONE_SNAPSHOT_SST_FILTERING_SERVICE_TIMEOUT_DEFAULT = "300s";
       // 300s for default
 
+  public static final String OZONE_SNAPSHOT_DEFRAG_SERVICE_TIMEOUT =
+      "ozone.snapshot.defrag.service.timeout";
+  public static final String
+      OZONE_SNAPSHOT_DEFRAG_SERVICE_TIMEOUT_DEFAULT = "300s";
+      // 300s for default
+
   public static final String OZONE_SNAPSHOT_DELETING_SERVICE_INTERVAL =
       "ozone.snapshot.deleting.service.interval";
   public static final String

hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/OzoneConsts.java

@@ -208,10 +208,11 @@ public final class OzoneConsts {
   public static final String OM_SLD_VERSION = "version";
   public static final String OM_SLD_CHECKSUM = "checksum";
   public static final String OM_SLD_IS_SST_FILTERED = "isSSTFiltered";
-  public static final String OM_SLD_UNCOMPACTED_SST_FILE_LIST = "uncompactedSSTFileList";
-  public static final String OM_SLD_LAST_COMPACTION_TIME = "lastCompactionTime";
-  public static final String OM_SLD_NEEDS_COMPACTION = "needsCompaction";
-  public static final String OM_SLD_COMPACTED_SST_FILE_LIST = "compactedSSTFileList";
+  public static final String OM_SLD_NOT_DEFRAGGED_SST_FILE_LIST = "notDefraggedSSTFileList";
+  public static final String OM_SLD_LAST_DEFRAG_TIME = "lastDefragTime";
+  public static final String OM_SLD_NEEDS_DEFRAGMENTATION = "needsDefragmentation";
+  public static final String OM_SLD_DEFRAGGED_SST_FILE_LIST = "defraggedSSTFileList";
+  public static final String OM_SLD_DB_CHECKPOINT_DIR = "dbCheckpointDir";
 
   // YAML fields for .container files
   public static final String CONTAINER_ID = "containerID";
@@ -509,6 +510,7 @@ public final class OzoneConsts {
   public static final String OM_SNAPSHOT_DIR = "db.snapshots";
   public static final String OM_SNAPSHOT_CHECKPOINT_DIR = OM_SNAPSHOT_DIR
       + OM_KEY_PREFIX + "checkpointState";
+  public static final String OM_SNAPSHOT_CHECKPOINT_DEFRAGED_DIR = "checkpointStateDefragged";
   public static final String OM_SNAPSHOT_DIFF_DIR = OM_SNAPSHOT_DIR
       + OM_KEY_PREFIX + "diffState";
 

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

@@ -216,6 +216,9 @@ public RDBStore build() throws RocksDatabaseException {
       ManagedWriteOptions writeOptions = new ManagedWriteOptions();
       writeOptions.setSync(rocksDBConfiguration.getSyncOption());
 
+      // Disable WAL to reduce disk write, for testing only
+      writeOptions.setDisableWAL(true);
+
       File dbFile = getDBFile();
       if (!dbFile.getParentFile().exists()) {
         throw new RocksDatabaseException("The DB destination directory should exist.");
@@ -399,6 +402,9 @@ private ManagedDBOptions getDefaultDBOptions(
     // RocksDB log settings.
     dbOptions.setMaxLogFileSize(rocksDBConfiguration.getMaxLogFileSize());
     dbOptions.setKeepLogFileNum(rocksDBConfiguration.getKeepLogFileNum());
+    // Disable WAL archival (for more accurate RocksDB size measurements when there aren't too many keys)
+    rocksDBConfiguration.setWalTTL(0);
+    rocksDBConfiguration.setWalSizeLimit(0);
 
     // Apply WAL settings.
     dbOptions.setWalTtlSeconds(rocksDBConfiguration.getWalTTL());

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

@@ -28,15 +28,15 @@
 /**
  * DumpFileWriter using rocksdb sst files.
  */
-class RDBSstFileWriter implements Closeable {
+public class RDBSstFileWriter implements Closeable {
 
   private ManagedSstFileWriter sstFileWriter;
   private File sstFile;
   private AtomicLong keyCounter;
   private ManagedOptions emptyOption = new ManagedOptions();
   private final ManagedEnvOptions emptyEnvOptions = new ManagedEnvOptions();
 
-  RDBSstFileWriter(File externalFile) throws RocksDatabaseException {
+  public RDBSstFileWriter(File externalFile) throws RocksDatabaseException {
     this.sstFileWriter = new ManagedSstFileWriter(emptyEnvOptions, emptyOption);
     this.keyCounter = new AtomicLong(0);
     this.sstFile = externalFile;
@@ -60,6 +60,17 @@ public void put(byte[] key, byte[] value) throws RocksDatabaseException {
     }
   }
 
+  public void delete(byte[] key) throws RocksDatabaseException {
+    try {
+      sstFileWriter.delete(key);
+      keyCounter.incrementAndGet();
+    } catch (RocksDBException e) {
+      closeOnFailure();
+      throw new RocksDatabaseException("Failed to delete key (length=" + key.length
+          + "), sstFile=" + sstFile.getAbsolutePath(), e);
+    }
+  }
+
   @Override
   public void close() throws RocksDatabaseException {
     if (sstFileWriter != null) {

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

@@ -139,6 +139,7 @@ public class RDBStore implements DBStore {
         checkpointsParentDir = checkpointsParentDirPath.toString();
         Files.createDirectories(checkpointsParentDirPath);
 
+        // TODO: scm, dn, etc. don't need this. Only om does
         Path snapshotsParentDirPath =
             Paths.get(dbLocation.getParent(), OM_SNAPSHOT_CHECKPOINT_DIR);
         snapshotsParentDir = snapshotsParentDirPath.toString();

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

@@ -237,7 +237,7 @@ public boolean isClosed() {
    *
    * @see ManagedCheckpoint
    */
-  final class RocksCheckpoint implements Closeable {
+  public final class RocksCheckpoint implements Closeable {
     private final ManagedCheckpoint checkpoint;
 
     private RocksCheckpoint() {
@@ -599,7 +599,7 @@ public List<LiveFileMetaData> getLiveFilesMetaData() throws RocksDatabaseExcepti
     }
   }
 
-  RocksCheckpoint createCheckpoint() {
+  public RocksCheckpoint createCheckpoint() {
     return new RocksCheckpoint();
   }
 
@@ -650,7 +650,7 @@ public Collection<ColumnFamily> getExtraColumnFamilies() {
     return Collections.unmodifiableCollection(columnFamilies.values());
   }
 
-  byte[] get(ColumnFamily family, byte[] key) throws RocksDatabaseException {
+  public byte[] get(ColumnFamily family, byte[] key) throws RocksDatabaseException {
     try (UncheckedAutoCloseable ignored = acquire()) {
       return db.get().get(family.getHandle(), key);
     } catch (RocksDBException e) {

hadoop-hdds/rocks-native/pom.xml

@@ -326,10 +326,6 @@
             <configuration>
               <includes>
                 <include>**/*.class</include>
-                <include>**/lib*.dylib</include>
-                <include>**/lib*.so</include>
-                <include>**/lib*.jnilib</include>
-                <include>**/lib*.dll</include>
               </includes>
             </configuration>
           </plugin>

hadoop-hdds/rocks-native/src/main/java/org/apache/hadoop/hdds/utils/NativeLibraryLoader.java

@@ -125,7 +125,8 @@ public synchronized boolean loadLibrary(final String libraryName, final List<Str
         System.loadLibrary(libraryName);
         loaded = true;
       } catch (Throwable e) {
-
+        LOG.debug("Failed to load library {} using System.loadLibrary, " +
+            "trying to load from jar", libraryName, e);
       }
       if (!loaded) {
         Pair<Optional<File>, List<File>> files = copyResourceFromJarToTemp(libraryName, dependentFiles);

hadoop-ozone/cli-admin/src/main/java/org/apache/hadoop/ozone/admin/om/OMAdmin.java

@@ -59,7 +59,8 @@
         UpdateRangerSubcommand.class,
         TransferOmLeaderSubCommand.class,
         FetchKeySubCommand.class,
-        LeaseSubCommand.class
+        LeaseSubCommand.class,
+        TriggerSnapshotDefragSubCommand.class
     })
 @MetaInfServices(AdminSubcommand.class)
 public class OMAdmin implements AdminSubcommand {

hadoop-ozone/cli-admin/src/main/java/org/apache/hadoop/ozone/admin/om/TriggerSnapshotDefragSubCommand.java

@@ -0,0 +1,89 @@
+/*
+ * 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.admin.om;
+
+import java.util.concurrent.Callable;
+import org.apache.hadoop.hdds.cli.HddsVersionProvider;
+import org.apache.hadoop.ozone.om.protocol.OzoneManagerProtocol;
+import picocli.CommandLine;
+
+/**
+ * Handler of ozone admin om triggerSnapshotDefrag command.
+ */
+@CommandLine.Command(
+    name = "triggerSnapshotDefrag",
+    description = "Triggers the Snapshot Defragmentation Service to run " +
+        "immediately. This command manually initiates the snapshot " +
+        "defragmentation process which compacts snapshot data and removes " +
+        "fragmentation to improve storage efficiency.",
+    mixinStandardHelpOptions = true,
+    versionProvider = HddsVersionProvider.class
+)
+public class TriggerSnapshotDefragSubCommand implements Callable<Void> {
+
+  @CommandLine.ParentCommand
+  private OMAdmin parent;
+
+  @CommandLine.Option(
+      names = {"-id", "--service-id"},
+      description = "Ozone Manager Service ID"
+  )
+  private String omServiceId;
+
+  @CommandLine.Option(
+      names = {"-host", "--service-host"},
+      description = "Ozone Manager Host"
+  )
+  private String omHost;
+
+  @CommandLine.Option(
+      names = {"--no-wait"},
+      description = "Do not wait for the defragmentation task to complete. " +
+          "The command will return immediately after triggering the task.",
+      defaultValue = "false"
+  )
+  private boolean noWait;
+
+  @Override
+  public Void call() throws Exception {
+    boolean forceHA = false;
+
+    try (OzoneManagerProtocol client = parent.createOmClient(omServiceId, omHost, forceHA)) {
+      System.out.println("Triggering Snapshot Defragmentation Service...");
+      boolean result = client.triggerSnapshotDefrag(noWait);
+
+      if (noWait) {
+        System.out.println("Snapshot defragmentation task has been triggered " +
+            "successfully and is running in the background.");
+      } else {
+        if (result) {
+          System.out.println("Snapshot defragmentation completed successfully.");
+        } else {
+          System.out.println("Snapshot defragmentation task failed or was " +
+              "interrupted.");
+        }
+      }
+    } catch (Exception e) {
+      System.err.println("Failed to trigger snapshot defragmentation: " +
+          e.getMessage());
+      throw e;
+    }
+
+    return null;
+  }
+}

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

@@ -267,6 +267,7 @@ public static boolean isReadOnly(
       // write to OM DB. And therefore it doesn't need a OMClientRequest.
       // Although indirectly the Ranger sync service task could invoke write
       // operation SetRangerServiceVersion.
+    case SnapshotDefrag:
     case GetKeyInfo:
     case SnapshotDiff:
     case CancelSnapshotDiff:

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

@@ -426,11 +426,21 @@ public final class OMConfigKeys {
       "ozone.snapshot.deleting.limit.per.task";
   public static final int SNAPSHOT_DELETING_LIMIT_PER_TASK_DEFAULT = 10;
 
+  // Snapshot defragmentation service configuration
+  public static final String SNAPSHOT_DEFRAG_LIMIT_PER_TASK =
+      "ozone.snapshot.defrag.limit.per.task";
+  public static final int SNAPSHOT_DEFRAG_LIMIT_PER_TASK_DEFAULT = 1;
+
   public static final String OZONE_SNAPSHOT_SST_FILTERING_SERVICE_INTERVAL =
       "ozone.snapshot.filtering.service.interval";
   public static final String
       OZONE_SNAPSHOT_SST_FILTERING_SERVICE_INTERVAL_DEFAULT = "60s";
 
+  public static final String OZONE_SNAPSHOT_DEFRAG_SERVICE_INTERVAL =
+      "ozone.snapshot.defrag.service.interval";
+  public static final String
+      OZONE_SNAPSHOT_DEFRAG_SERVICE_INTERVAL_DEFAULT = "60s";
+
   public static final String
       OZONE_SNAPSHOT_CHECKPOINT_DIR_CREATION_POLL_TIMEOUT =
       "ozone.om.snapshot.checkpoint.dir.creation.poll.timeout";
@@ -558,13 +568,12 @@ public final class OMConfigKeys {
 
   public static final String OZONE_OM_SNAPSHOT_DIFF_REPORT_MAX_PAGE_SIZE
       = "ozone.om.snapshot.diff.max.page.size";
+  public static final int OZONE_OM_SNAPSHOT_DIFF_REPORT_MAX_PAGE_SIZE_DEFAULT = 1000;
 
   public static final String OZONE_OM_SNAPSHOT_DB_MAX_OPEN_FILES
       = "ozone.om.snapshot.db.max.open.files";
   public static final int OZONE_OM_SNAPSHOT_DB_MAX_OPEN_FILES_DEFAULT
       = 100;
-  public static final int OZONE_OM_SNAPSHOT_DIFF_REPORT_MAX_PAGE_SIZE_DEFAULT
-      = 1000;
 
   public static final String OZONE_OM_SNAPSHOT_DIFF_THREAD_POOL_SIZE
       = "ozone.om.snapshot.diff.thread.pool.size";

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/QuotaUtil.java

@@ -19,6 +19,7 @@
 
 import static org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationType.EC;
 import static org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationType.RATIS;
+import static org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationType.STAND_ALONE;
 
 import org.apache.hadoop.hdds.client.ECReplicationConfig;
 import org.apache.hadoop.hdds.client.RatisReplicationConfig;
@@ -59,6 +60,9 @@ public static long getReplicatedSize(
           fullStripes * rc.getParity() * rc.getEcChunkSize()
               + partialFirstChunk * rc.getParity();
       return dataSize + replicationOverhead;
+    } else if (repConfig.getReplicationType() == STAND_ALONE) {
+      // For replication type STAND_ALONE, the replicated size is the same as data size.
+      return dataSize;
     } else {
       LOG.warn("Unknown replication type '{}'. Returning original data size.",
           repConfig.getReplicationType());

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/protocol/OzoneManagerProtocol.java

@@ -459,6 +459,18 @@ ListOpenFilesResult listOpenFiles(String path, int maxKeys, String contToken)
    */
   boolean triggerRangerBGSync(boolean noWait) throws IOException;
 
+  /**
+   * Triggers Snapshot Defragmentation Service immediately.
+   *
+   * Requires Ozone administrator privilege.
+   *
+   * @param noWait set to true if client won't wait for the result.
+   * @return true if noWait is true or when task completed successfully,
+   *         false otherwise.
+   * @throws IOException OMException (e.g. PERMISSION_DENIED)
+   */
+  boolean triggerSnapshotDefrag(boolean noWait) throws IOException;
+
   /**
    * Initiate metadata upgrade finalization.
    * This method when called, initiates finalization of Ozone Manager metadata

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/protocolPB/OzoneManagerProtocolClientSideTranslatorPB.java

@@ -219,6 +219,8 @@
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.SetTimesRequest;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.SetVolumePropertyRequest;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.SetVolumePropertyResponse;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.SnapshotDefragRequest;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.SnapshotDefragResponse;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.SnapshotInfoRequest;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.TenantAssignAdminRequest;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.TenantAssignUserAccessIdRequest;
@@ -1926,6 +1928,22 @@ public boolean triggerRangerBGSync(boolean noWait) throws IOException {
     return resp.getRunSuccess();
   }
 
+  @Override
+  public boolean triggerSnapshotDefrag(boolean noWait) throws IOException {
+    SnapshotDefragRequest req = SnapshotDefragRequest.newBuilder()
+        .setNoWait(noWait)
+        .build();
+
+    OMRequest omRequest = createOMRequest(Type.SnapshotDefrag)
+        .setSnapshotDefragRequest(req)
+        .build();
+
+    SnapshotDefragResponse resp = handleError(submitRequest(omRequest))
+        .getSnapshotDefragResponse();
+
+    return resp.getRunSuccess();
+  }
+
   @Override
   public StatusAndMessages finalizeUpgrade(String upgradeClientID)
       throws IOException {

hadoop-ozone/dist/pom.xml

@@ -171,6 +171,25 @@
           </execution>
         </executions>
       </plugin>
+      <plugin>
+        <groupId>org.apache.maven.plugins</groupId>
+        <artifactId>maven-antrun-plugin</artifactId>
+        <executions>
+          <execution>
+            <goals>
+              <goal>run</goal>
+            </goals>
+            <phase>prepare-package</phase>
+            <configuration>
+              <target>
+                <copy toDir="${project.build.directory}/ozone-${ozone.version}/share/ozone/lib">
+                  <fileset dir="${project.basedir}/../../hadoop-hdds/rocks-native/target/classes" includes="**/lib*.*" />
+                </copy>
+              </target>
+            </configuration>
+          </execution>
+        </executions>
+      </plugin>
       <plugin>
         <groupId>org.apache.maven.plugins</groupId>
         <artifactId>maven-resources-plugin</artifactId>