HDDS-6541. [Merge rocksdb in datanode] Per-disk DB location management.

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/upgrade/HDDSLayoutFeature.java

@@ -31,7 +31,9 @@ public enum HDDSLayoutFeature implements LayoutFeature {
   INITIAL_VERSION(0, "Initial Layout Version"),
   DATANODE_SCHEMA_V2(1, "Datanode RocksDB Schema Version 2 (with column " +
       "families)"),
-  SCM_HA(2, "Storage Container Manager HA");
+  SCM_HA(2, "Storage Container Manager HA"),
+  DATANODE_SCHEMA_V3(3, "Datanode RocksDB Schema Version 3 (one rocksdb " +
+      "per disk)");
 
   //////////////////////////////  //////////////////////////////
 

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

@@ -328,6 +328,11 @@ public final class OzoneConfigKeys {
   public static final String
       HDDS_DATANODE_METADATA_ROCKSDB_CACHE_SIZE_DEFAULT = "1GB";
 
+  // Specifying the dedicated volumes for per-disk db instances.
+  // For container schema v3 only.
+  public static final String HDDS_DATANODE_CONTAINER_DB_DIR =
+      "hdds.datanode.container.db.dir";
+
   public static final String OZONE_SECURITY_ENABLED_KEY =
       "ozone.security.enabled";
   public static final boolean OZONE_SECURITY_ENABLED_DEFAULT = false;

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

@@ -135,6 +135,7 @@ public final class OzoneConsts {
   public static final String SCM_DB_NAME = "scm.db";
   public static final String OM_DB_BACKUP_PREFIX = "om.db.backup.";
   public static final String SCM_DB_BACKUP_PREFIX = "scm.db.backup.";
+  public static final String CONTAINER_DB_NAME = "container.db";
 
   public static final String STORAGE_DIR_CHUNKS = "chunks";
   public static final String OZONE_DB_CHECKPOINT_REQUEST_FLUSH =

hadoop-hdds/common/src/main/resources/ozone-default.xml

@@ -144,6 +144,20 @@
       tagged explicitly.
     </description>
   </property>
+  <property>
+    <name>hdds.datanode.container.db.dir</name>
+    <value/>
+    <tag>OZONE, CONTAINER, STORAGE, MANAGEMENT</tag>
+    <description>Determines where the per-disk rocksdb instances will be
+      stored. Defaults to empty if not specified, then rocksdb instances
+      are stored on the same disk as HDDS data.
+      The directories should be tagged with corresponding storage types
+      ([SSD]/[DISK]/[ARCHIVE]/[RAM_DISK]) for storage policies. The default
+      storage type will be DISK if the directory does not have a storage type
+      tagged explicitly. Ideally, this should be mapped to a fast disk
+      like an SSD.
+    </description>
+  </property>
   <property>
     <name>hdds.datanode.dir.du.reserved</name>
     <value/>

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/HddsDatanodeService.java

@@ -320,7 +320,7 @@ private void startRatisForTest() throws IOException {
     for (Map.Entry<String, StorageVolume> entry : volumeMap.entrySet()) {
       HddsVolume hddsVolume = (HddsVolume) entry.getValue();
       boolean result = HddsVolumeUtil.checkVolume(hddsVolume, clusterId,
-          clusterId, conf, LOG);
+          clusterId, conf, LOG, null);
       if (!result) {
         volumeSet.failVolume(hddsVolume.getHddsRootDir().getPath());
       }

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/DatanodeConfiguration.java

@@ -45,6 +45,8 @@ public class DatanodeConfiguration {
       "hdds.datanode.failed.data.volumes.tolerated";
   public static final String FAILED_METADATA_VOLUMES_TOLERATED_KEY =
       "hdds.datanode.failed.metadata.volumes.tolerated";
+  public static final String FAILED_DB_VOLUMES_TOLERATED_KEY =
+      "hdds.datanode.failed.db.volumes.tolerated";
   public static final String DISK_CHECK_MIN_GAP_KEY =
       "hdds.datanode.disk.check.min.gap";
   public static final String DISK_CHECK_TIMEOUT_KEY =
@@ -195,6 +197,17 @@ public void setBlockDeletionLimit(int limit) {
   )
   private int failedMetadataVolumesTolerated = FAILED_VOLUMES_TOLERATED_DEFAULT;
 
+  @Config(key = "failed.db.volumes.tolerated",
+      defaultValue = "-1",
+      type = ConfigType.INT,
+      tags = { DATANODE },
+      description = "The number of db volumes that are allowed to fail "
+          + "before a datanode stops offering service. "
+          + "Config this to -1 means unlimited, but we should have "
+          + "at least one good volume left."
+  )
+  private int failedDbVolumesTolerated = FAILED_VOLUMES_TOLERATED_DEFAULT;
+
   @Config(key = "disk.check.min.gap",
       defaultValue = "15m",
       type = ConfigType.TIME,
@@ -277,6 +290,13 @@ public void validate() {
       failedMetadataVolumesTolerated = FAILED_VOLUMES_TOLERATED_DEFAULT;
     }
 
+    if (failedDbVolumesTolerated < -1) {
+      LOG.warn(FAILED_DB_VOLUMES_TOLERATED_KEY +
+              "must be greater than -1 and was set to {}. Defaulting to {}",
+          failedDbVolumesTolerated, FAILED_VOLUMES_TOLERATED_DEFAULT);
+      failedDbVolumesTolerated = FAILED_VOLUMES_TOLERATED_DEFAULT;
+    }
+
     if (diskCheckMinGap < 0) {
       LOG.warn(DISK_CHECK_MIN_GAP_KEY +
               " must be greater than zero and was set to {}. Defaulting to {}",
@@ -325,6 +345,14 @@ public void setFailedMetadataVolumesTolerated(int failedVolumesTolerated) {
     this.failedMetadataVolumesTolerated = failedVolumesTolerated;
   }
 
+  public int getFailedDbVolumesTolerated() {
+    return failedDbVolumesTolerated;
+  }
+
+  public void setFailedDbVolumesTolerated(int failedVolumesTolerated) {
+    this.failedDbVolumesTolerated = failedVolumesTolerated;
+  }
+
   public Duration getDiskCheckMinGap() {
     return Duration.ofMillis(diskCheckMinGap);
   }

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/DatanodeStateMachine.java

@@ -316,8 +316,8 @@ private void startStateMachineThread() throws IOException {
   public void handleFatalVolumeFailures() {
     LOG.error("DatanodeStateMachine Shutdown due to too many bad volumes, "
         + "check " + DatanodeConfiguration.FAILED_DATA_VOLUMES_TOLERATED_KEY
-        + " and "
-        + DatanodeConfiguration.FAILED_METADATA_VOLUMES_TOLERATED_KEY);
+        + " and " + DatanodeConfiguration.FAILED_METADATA_VOLUMES_TOLERATED_KEY
+        + " and " + DatanodeConfiguration.FAILED_DB_VOLUMES_TOLERATED_KEY);
     hddsDatanodeStopService.stopService();
   }
 

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/states/endpoint/VersionEndpointTask.java

@@ -17,6 +17,7 @@
 package org.apache.hadoop.ozone.container.common.states.endpoint;
 
 import java.io.IOException;
+import java.util.List;
 import java.util.Map;
 import java.util.concurrent.Callable;
 
@@ -25,10 +26,13 @@
 import org.apache.hadoop.ozone.OzoneConsts;
 import org.apache.hadoop.ozone.container.common.statemachine.EndpointStateMachine;
 import org.apache.hadoop.ozone.container.common.utils.HddsVolumeUtil;
+import org.apache.hadoop.ozone.container.common.utils.StorageVolumeUtil;
+import org.apache.hadoop.ozone.container.common.volume.DbVolume;
 import org.apache.hadoop.ozone.container.common.volume.HddsVolume;
 import org.apache.hadoop.ozone.container.common.volume.MutableVolumeSet;
 import org.apache.hadoop.ozone.container.common.volume.StorageVolume;
 import org.apache.hadoop.ozone.container.ozoneimpl.OzoneContainer;
+import org.apache.hadoop.ozone.container.upgrade.VersionedDatanodeFeatures;
 import org.apache.hadoop.ozone.protocol.VersionResponse;
 import org.apache.hadoop.util.DiskChecker.DiskOutOfSpaceException;
 
@@ -78,24 +82,27 @@ public EndpointStateMachine.EndPointStates call() throws Exception {
           String scmId = response.getValue(OzoneConsts.SCM_ID);
           String clusterId = response.getValue(OzoneConsts.CLUSTER_ID);
 
+          Preconditions.checkNotNull(scmId,
+              "Reply from SCM: scmId cannot be null");
+          Preconditions.checkNotNull(clusterId,
+              "Reply from SCM: clusterId cannot be null");
+
+          formatDbVolumesIfNeeded(scmId, clusterId);
+
           // Check volumes
           MutableVolumeSet volumeSet = ozoneContainer.getVolumeSet();
           volumeSet.writeLock();
           try {
             Map<String, StorageVolume> volumeMap = volumeSet.getVolumeMap();
 
-            Preconditions.checkNotNull(scmId,
-                "Reply from SCM: scmId cannot be null");
-            Preconditions.checkNotNull(clusterId,
-                "Reply from SCM: clusterId cannot be null");
-
             // If version file does not exist
             // create version file and also set scm ID or cluster ID.
             for (Map.Entry<String, StorageVolume> entry
                 : volumeMap.entrySet()) {
               StorageVolume volume = entry.getValue();
               boolean result = HddsVolumeUtil.checkVolume((HddsVolume) volume,
-                  scmId, clusterId, configuration, LOG);
+                  scmId, clusterId, configuration, LOG,
+                  ozoneContainer.getDbVolumeSet());
               if (!result) {
                 volumeSet.failVolume(volume.getStorageDir().getPath());
               }
@@ -129,4 +136,31 @@ public EndpointStateMachine.EndPointStates call() throws Exception {
     }
     return rpcEndPoint.getState();
   }
+
+  private void formatDbVolumesIfNeeded(String scmId, String clusterId) {
+    if (!VersionedDatanodeFeatures.SchemaV3.chooseSchemaVersion()
+        .equals(OzoneConsts.SCHEMA_V3)) {
+      return;
+    }
+
+    MutableVolumeSet dbVolumeSet = ozoneContainer.getDbVolumeSet();
+
+    if (dbVolumeSet != null) {
+      dbVolumeSet.writeLock();
+      try {
+        List<DbVolume> dbVolumeList = StorageVolumeUtil
+            .getDbVolumesList(dbVolumeSet.getVolumesList());
+
+        dbVolumeList.parallelStream().forEach(dbVolume -> {
+          String id = VersionedDatanodeFeatures.ScmHA
+              .chooseContainerPathID(configuration, scmId, clusterId);
+          if (!dbVolume.format(id)) {
+            dbVolumeSet.failVolume(dbVolume.getStorageDir().getPath());
+          }
+        });
+      } finally {
+        dbVolumeSet.writeUnlock();
+      }
+    }
+  }
 }