HDFS-15683. Allow configuring DISK/ARCHIVE capacity for individual volumes. (apache#2625)

Author: LeonGao91

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSConfigKeys.java

@@ -1557,6 +1557,12 @@ public class DFSConfigKeys extends CommonConfigurationKeys {
   public static final double
       DFS_DATANODE_RESERVE_FOR_ARCHIVE_DEFAULT_PERCENTAGE_DEFAULT = 0.0;
 
+  public static final String
+      DFS_DATANODE_SAME_DISK_TIERING_CAPACITY_RATIO_PERCENTAGE =
+      "dfs.datanode.same-disk-tiering.capacity-ratio.percentage";
+  public static final String
+      DFS_DATANODE_SAME_DISK_TIERING_CAPACITY_RATIO_PERCENTAGE_DEFAULT = "";
+
   // dfs.client.retry confs are moved to HdfsClientConfigKeys.Retry
   @Deprecated
   public static final String  DFS_CLIENT_RETRY_POLICY_ENABLED_KEY

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataNode.java

@@ -20,6 +20,8 @@
 
 import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_DATANODE_ADDRESS_DEFAULT;
 import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_DATANODE_ADDRESS_KEY;
+import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_DATANODE_ALLOW_SAME_DISK_TIERING;
+import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_DATANODE_ALLOW_SAME_DISK_TIERING_DEFAULT;
 import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_DATANODE_DATA_DIR_KEY;
 import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_DATANODE_DIRECTORYSCAN_INTERVAL_DEFAULT;
 import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_DATANODE_DIRECTORYSCAN_INTERVAL_KEY;
@@ -739,9 +741,51 @@ ChangedVolumes parseChangedVolumes(String newVolumes) throws IOException {
       }
     }
 
+    validateVolumesWithSameDiskTiering(results);
+
     return results;
   }
 
+  /**
+   * Check conflict with same disk tiering feature
+   * and throws exception.
+   *
+   * TODO: We can add feature to
+   *   allow refreshing volume with capacity ratio,
+   *   and solve the case of replacing volume on same mount.
+   */
+  private void validateVolumesWithSameDiskTiering(ChangedVolumes
+      changedVolumes) throws IOException {
+    if (dnConf.getConf().getBoolean(DFS_DATANODE_ALLOW_SAME_DISK_TIERING,
+        DFS_DATANODE_ALLOW_SAME_DISK_TIERING_DEFAULT)
+        && data.getMountVolumeMap() != null) {
+      // Check if mount already exist.
+      for (StorageLocation location : changedVolumes.newLocations) {
+        if (StorageType.allowSameDiskTiering(location.getStorageType())) {
+          File dir = new File(location.getUri());
+          // Get the first parent dir that exists to check disk mount point.
+          while (!dir.exists()) {
+            dir = dir.getParentFile();
+            if (dir == null) {
+              throw new IOException("Invalid path: "
+                  + location + ": directory does not exist");
+            }
+          }
+          DF df = new DF(dir, dnConf.getConf());
+          String mount = df.getMount();
+          if (data.getMountVolumeMap().hasMount(mount)) {
+            String errMsg = "Disk mount " + mount
+                + " already has volume, when trying to add "
+                + location + ". Please try removing mounts first"
+                + " or restart datanode.";
+            LOG.error(errMsg);
+            throw new IOException(errMsg);
+          }
+        }
+      }
+    }
+  }
+
   /**
    * Attempts to reload data volumes with new configuration.
    * @param newVolumes a comma separated string that specifies the data volumes.

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/StorageLocation.java

@@ -18,6 +18,8 @@
 
 package org.apache.hadoop.hdfs.server.datanode;
 
+import java.util.HashMap;
+import java.util.Map;
 import java.util.regex.Pattern;
 
 import java.io.File;
@@ -58,7 +60,16 @@ public class StorageLocation
   /** Regular expression that describes a storage uri with a storage type.
    *  e.g. [Disk]/storages/storage1/
    */
-  private static final Pattern regex = Pattern.compile("^\\[(\\w*)\\](.+)$");
+  private static final Pattern STORAGE_LOCATION_REGEX =
+      Pattern.compile("^\\[(\\w*)\\](.+)$");
+
+  /** Regular expression for the capacity ratio of a storage volume (uri).
+   *  This is useful when configuring multiple
+   *  storage types on same disk mount (same-disk-tiering).
+   *  e.g. [0.3]/disk1/archive/
+   */
+  private static final Pattern CAPACITY_RATIO_REGEX =
+      Pattern.compile("^\\[([0-9.]*)\\](.+)$");
 
   private StorageLocation(StorageType storageType, URI uri) {
     this.storageType = storageType;
@@ -127,7 +138,7 @@ public boolean matchesStorageDirectory(StorageDirectory sd,
    */
   public static StorageLocation parse(String rawLocation)
       throws IOException, SecurityException {
-    Matcher matcher = regex.matcher(rawLocation);
+    Matcher matcher = STORAGE_LOCATION_REGEX.matcher(rawLocation);
     StorageType storageType = StorageType.DEFAULT;
     String location = rawLocation;
 
@@ -144,6 +155,44 @@ public static StorageLocation parse(String rawLocation)
     return new StorageLocation(storageType, new Path(location).toUri());
   }
 
+  /**
+   * Attempt to parse the storage capacity ratio and related volume directory
+   * out of the capacity ratio config string.
+   *
+   * @param capacityRatioConf Config string of the capacity ratio
+   * @return Map of URI of the volume and capacity ratio.
+   * @throws SecurityException when format is incorrect or ratio is not
+   *         between 0 - 1.
+   */
+  public static Map<URI, Double> parseCapacityRatio(String capacityRatioConf)
+      throws SecurityException {
+    Map<URI, Double> result = new HashMap<>();
+    capacityRatioConf = capacityRatioConf.replaceAll("\\s", "");
+    if (capacityRatioConf.isEmpty()) {
+      return result;
+    }
+    String[] capacityRatios = capacityRatioConf.split(",");
+    for (String ratio : capacityRatios) {
+      Matcher matcher = CAPACITY_RATIO_REGEX.matcher(ratio);
+      if (matcher.matches()) {
+        String capacityString = matcher.group(1).trim();
+        String location = matcher.group(2).trim();
+        double capacityRatio = Double.parseDouble(capacityString);
+        if (capacityRatio > 1 || capacityRatio < 0) {
+          throw new IllegalArgumentException("Capacity ratio" + capacityRatio
+              + " is not between 0 to 1: " + ratio);
+        }
+        result.put(new Path(location).toUri(), capacityRatio);
+      } else {
+        throw new IllegalArgumentException(
+            "Capacity ratio config is not with correct format: "
+                + capacityRatioConf
+        );
+      }
+    }
+    return result;
+  }
+
   @Override
   public String toString() {
     return "[" + storageType + "]" + baseURI.normalize();

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/FsDatasetSpi.java

@@ -35,6 +35,7 @@
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.StorageType;
+import org.apache.hadoop.hdfs.server.datanode.fsdataset.impl.MountVolumeMap;
 import org.apache.hadoop.util.AutoCloseableLock;
 import org.apache.hadoop.hdfs.DFSConfigKeys;
 import org.apache.hadoop.hdfs.protocol.Block;
@@ -680,4 +681,11 @@ ReplicaInfo moveBlockAcrossVolumes(final ExtendedBlock block,
    * @throws IOException
    */
   Set<? extends Replica> deepCopyReplica(String bpid) throws IOException;
+
+  /**
+   * Get relationship between disk mount and FsVolume.
+   * @return Disk mount and FsVolume relationship.
+   * @throws IOException
+   */
+  MountVolumeMap getMountVolumeMap() throws IOException;
 }

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/FsDatasetImpl.java

@@ -193,10 +193,6 @@ public FsVolumeImpl getVolume(final ExtendedBlock b) {
     }
   }
 
-  MountVolumeMap getMountVolumeMap() {
-    return volumes.getMountVolumeMap();
-  }
-
   @Override // FsDatasetSpi
   public Block getStoredBlock(String bpid, long blkid)
       throws IOException {
@@ -249,7 +245,7 @@ public LengthInputStream getMetaDataInputStream(ExtendedBlock b)
     }
     return info.getMetadataInputStream(0);
   }
-    
+
   final DataNode datanode;
   private final DataNodeMetrics dataNodeMetrics;
   final DataStorage dataStorage;
@@ -3524,7 +3520,12 @@ public boolean getPinning(ExtendedBlock block) throws IOException {
     ReplicaInfo r = getBlockReplica(block);
     return r.getPinning(localFS);
   }
-  
+
+  @Override
+  public MountVolumeMap getMountVolumeMap() {
+    return volumes.getMountVolumeMap();
+  }
+
   @Override
   public boolean isDeletingBlock(String bpid, long blockId) {
     synchronized(deletingBlock) {

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/FsVolumeList.java

@@ -18,6 +18,7 @@
 package org.apache.hadoop.hdfs.server.datanode.fsdataset.impl;
 
 import java.io.IOException;
+import java.net.URI;
 import java.nio.channels.ClosedChannelException;
 import java.util.ArrayList;
 import java.util.Collection;
@@ -65,6 +66,7 @@ class FsVolumeList {
 
   private final boolean enableSameDiskTiering;
   private final MountVolumeMap mountVolumeMap;
+  private Map<URI, Double> capacityRatioMap;
 
   FsVolumeList(List<VolumeFailureInfo> initialVolumeFailureInfos,
       BlockScanner blockScanner,
@@ -82,6 +84,7 @@ class FsVolumeList {
         DFSConfigKeys.DFS_DATANODE_ALLOW_SAME_DISK_TIERING,
         DFSConfigKeys.DFS_DATANODE_ALLOW_SAME_DISK_TIERING_DEFAULT);
     mountVolumeMap = new MountVolumeMap(config);
+    initializeCapacityRatio(config);
   }
 
   MountVolumeMap getMountVolumeMap() {
@@ -135,6 +138,20 @@ FsVolumeReference getVolumeByMount(StorageType storageType,
     return null;
   }
 
+  private void initializeCapacityRatio(Configuration config) {
+    if (capacityRatioMap == null) {
+      String capacityRatioConfig = config.get(
+          DFSConfigKeys
+              .DFS_DATANODE_SAME_DISK_TIERING_CAPACITY_RATIO_PERCENTAGE,
+          DFSConfigKeys
+              .DFS_DATANODE_SAME_DISK_TIERING_CAPACITY_RATIO_PERCENTAGE_DEFAULT
+      );
+
+      this.capacityRatioMap = StorageLocation
+          .parseCapacityRatio(capacityRatioConfig);
+    }
+  }
+
   /** 
    * Get next volume.
    *
@@ -325,11 +342,15 @@ public String toString() {
    *
    * @param ref       a reference to the new FsVolumeImpl instance.
    */
-  void addVolume(FsVolumeReference ref) {
+  void addVolume(FsVolumeReference ref) throws IOException {
     FsVolumeImpl volume = (FsVolumeImpl) ref.getVolume();
     volumes.add(volume);
     if (isSameDiskTieringApplied(volume)) {
       mountVolumeMap.addVolume(volume);
+      URI uri = volume.getStorageLocation().getUri();
+      if (capacityRatioMap.containsKey(uri)) {
+        mountVolumeMap.setCapacityRatio(volume, capacityRatioMap.get(uri));
+      }
     }
     if (blockScanner != null) {
       blockScanner.addVolumeScanner(ref);

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/MountVolumeInfo.java

@@ -24,21 +24,24 @@
 import org.apache.hadoop.hdfs.server.datanode.fsdataset.FsVolumeReference;
 
 import java.nio.channels.ClosedChannelException;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentMap;
+import java.util.EnumMap;
+import java.util.Map;
 
 /**
  * MountVolumeInfo is a wrapper of
  * detailed volume information for MountVolumeMap.
  */
 @InterfaceAudience.Private
 class MountVolumeInfo {
-  private final ConcurrentMap<StorageType, FsVolumeImpl>
+  private final EnumMap<StorageType, FsVolumeImpl>
       storageTypeVolumeMap;
+  private final EnumMap<StorageType, Double>
+      capacityRatioMap;
   private double reservedForArchiveDefault;
 
   MountVolumeInfo(Configuration conf) {
-    storageTypeVolumeMap = new ConcurrentHashMap<>();
+    storageTypeVolumeMap = new EnumMap<>(StorageType.class);
+    capacityRatioMap = new EnumMap<>(StorageType.class);
     reservedForArchiveDefault = conf.getDouble(
         DFSConfigKeys.DFS_DATANODE_RESERVE_FOR_ARCHIVE_DEFAULT_PERCENTAGE,
         DFSConfigKeys
@@ -71,12 +74,22 @@ FsVolumeReference getVolumeRef(StorageType storageType) {
 
   /**
    * Return configured capacity ratio.
-   * If the volume is the only one on the mount,
-   * return 1 to avoid unnecessary allocation.
-   *
-   * TODO: We should support customized capacity ratio for volumes.
    */
   double getCapacityRatio(StorageType storageType) {
+    // If capacity ratio is set, return the val.
+    if (capacityRatioMap.containsKey(storageType)) {
+      return capacityRatioMap.get(storageType);
+    }
+    // If capacity ratio is set for counterpart,
+    // use the rest of capacity of the mount for it.
+    if (!capacityRatioMap.isEmpty()) {
+      double leftOver = 1;
+      for (Map.Entry<StorageType, Double> e : capacityRatioMap.entrySet()) {
+        leftOver -= e.getValue();
+      }
+      return leftOver;
+    }
+    // Use reservedForArchiveDefault by default.
     if (storageTypeVolumeMap.containsKey(storageType)
         && storageTypeVolumeMap.size() > 1) {
       if (storageType == StorageType.ARCHIVE) {
@@ -102,9 +115,28 @@ boolean addVolume(FsVolumeImpl volume) {
     return true;
   }
 
-
   void removeVolume(FsVolumeImpl target) {
     storageTypeVolumeMap.remove(target.getStorageType());
+    capacityRatioMap.remove(target.getStorageType());
+  }
+
+  /**
+   * Set customize capacity ratio for a storage type.
+   * Return false if the value is too big.
+   */
+  boolean setCapacityRatio(StorageType storageType,
+      double capacityRatio) {
+    double leftover = 1;
+    for (Map.Entry<StorageType, Double> e : capacityRatioMap.entrySet()) {
+      if (e.getKey() != storageType) {
+        leftover -= e.getValue();
+      }
+    }
+    if (leftover < capacityRatio) {
+      return false;
+    }
+    capacityRatioMap.put(storageType, capacityRatio);
+    return true;
   }
 
   int size() {

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/MountVolumeMap.java

@@ -22,6 +22,7 @@
 import org.apache.hadoop.fs.StorageType;
 import org.apache.hadoop.hdfs.server.datanode.fsdataset.FsVolumeReference;
 
+import java.io.IOException;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 
@@ -34,7 +35,7 @@
  * we don't configure multiple volumes with same storage type on one mount.
  */
 @InterfaceAudience.Private
-class MountVolumeMap {
+public class MountVolumeMap {
   private final ConcurrentMap<String, MountVolumeInfo>
       mountVolumeMapping;
   private final Configuration conf;
@@ -89,4 +90,24 @@ void removeVolume(FsVolumeImpl target) {
       }
     }
   }
+
+  void setCapacityRatio(FsVolumeImpl target, double capacityRatio)
+      throws IOException {
+    String mount = target.getMount();
+    if (!mount.isEmpty()) {
+      MountVolumeInfo info = mountVolumeMapping.get(mount);
+      if (!info.setCapacityRatio(
+          target.getStorageType(), capacityRatio)) {
+        throw new IOException(
+            "Not enough capacity ratio left on mount: "
+                + mount + ", for " + target + ": capacity ratio: "
+                + capacityRatio + ". Sum of the capacity"
+                + " ratio of on same disk mount should be <= 1");
+      }
+    }
+  }
+
+  public boolean hasMount(String mount) {
+    return mountVolumeMapping.containsKey(mount);
+  }
 }