HDDS-15250. PipelineProvider support create pipeline by StorageTier

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/client/StorageTier.java

@@ -17,9 +17,6 @@
 
 package org.apache.hadoop.hdds.client;
 
-import static org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationFactor.ONE;
-import static org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationFactor.THREE;
-
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.EnumMap;
@@ -40,21 +37,23 @@ public enum StorageTier {
 
   private final String tierName;
   private final List<StorageType> storageTypes;
-  private final boolean uniformStorageType;
-  private static final Map<StorageTier, Map<ReplicationConfig, List<StorageType>>>
+  private final boolean isUniform;
+  private static final Map<StorageTier, Map<Integer, List<StorageType>>>
       CACHE = new EnumMap<>(StorageTier.class);
+  public static final int MAX_NODE_COUNT = 20;
+  private static StorageTier defaultTier = DISK;
 
   StorageTier(String tierName) {
     this.tierName = tierName;
     this.storageTypes = Collections.emptyList();
-    this.uniformStorageType = true;
+    this.isUniform = true;
   }
 
   // Constructor for uniform storage tiers
   StorageTier(String tierName, StorageType uniformStorageType) {
     this.tierName = tierName;
     this.storageTypes = Collections.singletonList(uniformStorageType);
-    this.uniformStorageType = true;
+    this.isUniform = true;
   }
 
   // Constructor for non-uniform storage tiers
@@ -68,27 +67,24 @@ public enum StorageTier {
           " StorageType were provided.");
     }
     this.storageTypes = Arrays.asList(storageTypes);
-    this.uniformStorageType = false;
+    this.isUniform = false;
   }
 
   static {
-    // Precompute storage type mappings for each replication config
     for (StorageTier tier : StorageTier.values()) {
-      Map<ReplicationConfig, List<StorageType>> tierCache = new HashMap<>();
-      List<ReplicationConfig> replicationConfigs = Arrays.asList(
-          RatisReplicationConfig.getInstance(ONE),
-          RatisReplicationConfig.getInstance(THREE),
-          StandaloneReplicationConfig.getInstance(ONE),
-          StandaloneReplicationConfig.getInstance(THREE)
-      );
-
-      for (ReplicationConfig config : replicationConfigs) {
-        tierCache.put(config, tier.computeStorageTypes(config));
+      Map<Integer, List<StorageType>> tierCache = new HashMap<>();
+      for (int nodeCount = 0; nodeCount <= MAX_NODE_COUNT; nodeCount++) {
+        List<StorageType> storageTypes = tier.computeStorageTypes(nodeCount, tier);
+        tierCache.put(nodeCount, storageTypes);
       }
       CACHE.put(tier, tierCache);
     }
   }
 
+  public static StorageTier getDefaultTier() {
+    return defaultTier;
+  }
+
   public StorageTierProto toProto() {
     switch (this) {
     case SSD:
@@ -121,50 +117,51 @@ public String getTierName() {
     return tierName;
   }
 
-  public boolean isUniformStorageType() {
-    return uniformStorageType;
+  public boolean isUniform() {
+    return isUniform;
   }
 
   /**
    * Computes the list of StorageTypes based on replication configuration.
    *
-   * @param replicationConfig The replication configuration.
+   * @param nodeCount The node count of the storageTier required.
+   * @param storageTier The required StorageTier.
    * @return The list of StorageTypes for the given tier and replication configuration.
    */
-  private List<StorageType> computeStorageTypes(
-      ReplicationConfig replicationConfig) {
-    if (isUniformStorageType()) {
-      int numberOfNodes = replicationConfig.getRequiredNodes();
+  private List<StorageType> computeStorageTypes(int nodeCount, StorageTier storageTier) {
+    if (isUniform()) {
       if (storageTypes.isEmpty()) {
         return Collections.emptyList();
       }
-      return Collections.nCopies(numberOfNodes, storageTypes.get(0));
+      return Collections.nCopies(nodeCount, storageTypes.get(0));
     } else {
       throw new UnsupportedOperationException(
-          "Unsupported not UniformStorage Storage Tier: " + replicationConfig);
+          "Unsupported not uniform StorageTier: " + storageTier);
     }
   }
 
   /**
    * Maps a StorageTier to its corresponding StorageType based on replication type.
    *
-   * @param replicationConfig The replication configuration.
+   * @param nodeCount The node count of the storageTier required.
    * @return The list of StorageTypes corresponding to the given tier and replication configuration.
    * @throws IllegalArgumentException if the replication configuration is not supported.
    */
-  public List<StorageType> getStorageTypes(
-      ReplicationConfig replicationConfig) {
-    Map<ReplicationConfig, List<StorageType>> tierCache = CACHE.get(this);
+  public List<StorageType> getStorageTypes(int nodeCount) {
+    Map<Integer, List<StorageType>> tierCache = CACHE.get(this);
 
     if (tierCache != null) {
-      List<StorageType> cachedStorageType = tierCache.get(replicationConfig);
+      List<StorageType> cachedStorageType = tierCache.get(nodeCount);
       if (cachedStorageType != null) {
         return cachedStorageType;
       }
     }
 
-    throw new IllegalArgumentException("Unsupported ReplicationConfig: " +
-        replicationConfig + " for StorageTier: " + getTierName());
+    throw new IllegalArgumentException("Unsupported node count: " +
+        nodeCount + " for StorageTier: " + getTierName());
   }
 
+  public static void setDefault(StorageTier storageTier) {
+    defaultTier = storageTier;
+  }
 }

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/client/StorageTierUtil.java

@@ -0,0 +1,62 @@
+/*
+ * 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.hdds.client;
+
+import java.util.List;
+import org.apache.hadoop.fs.StorageType;
+import org.apache.hadoop.hdds.scm.exceptions.SCMException;
+
+/**
+ * Utility class for managing StorageTier operations.
+ */
+public final class StorageTierUtil {
+  private StorageTierUtil() {
+  }
+
+  /**
+   * Validates the given StorageTier and throws an exception if it is empty.
+   *
+   * @param storageTier the StorageTier to check
+   * @throws SCMException if the StorageTier is empty
+   */
+  public static void validateNotEmpty(StorageTier storageTier) throws SCMException {
+    if (storageTier.equals(StorageTier.EMPTY)) {
+      throw new SCMException("Cannot create Pipeline for empty tier",
+          SCMException.ResultCodes.CANNOT_CREATE_PIPELINE_FOR_EMPTY_TIER);
+    }
+  }
+
+  /**
+   * Returns the StorageType for uniform StorageTier.
+   *
+   * @param storageTier the StorageTier to get StorageType from
+   * @return The uniform StorageTier corresponding StorageType
+   * @throws SCMException if the StorageTier is non-uniform or the EMPTY StorageTier
+   */
+  public static StorageType getStorageTypeForUniformStorageTier(StorageTier storageTier, ReplicationConfig config)
+      throws SCMException {
+    validateNotEmpty(storageTier);
+    List<StorageType> storageTypes = storageTier.getStorageTypes(config.getRequiredNodes());
+    if (storageTier.isUniform()) {
+      return storageTypes.get(0);
+    } else {
+      throw new SCMException("Unsupported non-uniform storage tier " + storageTier,
+          SCMException.ResultCodes.UNSUPPORTED_NON_UNIFORM_STORAGE_TIER);
+    }
+  }
+}

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/client/StorageTypeUtils.java

@@ -43,6 +43,8 @@ public static HddsProtos.StorageTypeProto getStorageTypeProto(@Nonnull StorageTy
       return HddsProtos.StorageTypeProto.PROVIDED;
     case RAM_DISK:
       return HddsProtos.StorageTypeProto.RAM_DISK;
+    case NVDIMM:
+      return HddsProtos.StorageTypeProto.NVDIMM;
     default:
       throw new IllegalArgumentException("Illegal Storage Type specified");
     }
@@ -61,6 +63,8 @@ public static StorageType getFromProtobuf(@Nonnull HddsProtos.StorageTypeProto p
       return StorageType.PROVIDED;
     case RAM_DISK:
       return StorageType.RAM_DISK;
+    case NVDIMM:
+      return StorageType.NVDIMM;
     default:
       throw new IllegalArgumentException("Illegal Storage Type specified");
     }
@@ -93,6 +97,7 @@ public static int getIDFromProtobuf(@Nonnull HddsProtos.StorageTypeProto proto)
     case ARCHIVE:
     case PROVIDED:
     case RAM_DISK:
+    case NVDIMM:
       return proto.getNumber();
     default:
       throw new IllegalArgumentException("Illegal Storage Type specified");

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/exceptions/SCMException.java

@@ -157,6 +157,8 @@ public enum ResultCodes {
     CA_ROTATION_IN_POST_PROGRESS,
     CONTAINER_ALREADY_CLOSED,
     CONTAINER_ALREADY_CLOSING,
-    UNSUPPORTED_OPERATION
+    UNSUPPORTED_OPERATION,
+    CANNOT_CREATE_PIPELINE_FOR_EMPTY_TIER,
+    UNSUPPORTED_NON_UNIFORM_STORAGE_TIER,
   }
 }

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/impl/StorageLocationReport.java

@@ -20,6 +20,7 @@
 import java.io.IOException;
 import net.jcip.annotations.Immutable;
 import org.apache.hadoop.fs.StorageType;
+import org.apache.hadoop.hdds.client.StorageTypeUtils;
 import org.apache.hadoop.hdds.conf.ConfigurationSource;
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos.StorageTypeProto;
 import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.MetadataStorageReportProto;
@@ -116,32 +117,7 @@ public StorageType getStorageType() {
   }
 
   private StorageTypeProto getStorageTypeProto() throws IllegalArgumentException {
-    return getStorageTypeProto(getStorageType());
-  }
-
-  public static StorageTypeProto getStorageTypeProto(StorageType type)
-      throws IllegalArgumentException {
-    StorageTypeProto storageTypeProto;
-    switch (type) {
-    case SSD:
-      storageTypeProto = StorageTypeProto.SSD;
-      break;
-    case DISK:
-      storageTypeProto = StorageTypeProto.DISK;
-      break;
-    case ARCHIVE:
-      storageTypeProto = StorageTypeProto.ARCHIVE;
-      break;
-    case PROVIDED:
-      storageTypeProto = StorageTypeProto.PROVIDED;
-      break;
-    case RAM_DISK:
-      storageTypeProto = StorageTypeProto.RAM_DISK;
-      break;
-    default:
-      throw new IllegalArgumentException("Illegal Storage Type specified");
-    }
-    return storageTypeProto;
+    return StorageTypeUtils.getStorageTypeProto(getStorageType());
   }
 
   public long getReserved() {
@@ -176,7 +152,7 @@ public static StorageType getStorageType(StorageTypeProto proto) throws
       storageType = StorageType.RAM_DISK;
       break;
     default:
-      throw new IllegalArgumentException("Illegal Storage Type specified");
+      throw new IllegalArgumentException("Illegal Storage Type specified: " + proto);
     }
     return storageType;
   }
@@ -247,7 +223,7 @@ public static StorageLocationReport getFromProtobuf(StorageReportProto report)
       builder.setScmUsed(report.getScmUsed());
     }
     if (report.hasStorageType()) {
-      builder.setStorageType(getStorageType(report.getStorageType()));
+      builder.setStorageType(StorageTypeUtils.getFromProtobuf(report.getStorageType()));
     }
     if (report.hasRemaining()) {
       builder.setRemaining(report.getRemaining());

hadoop-hdds/interface-client/src/main/proto/hdds.proto

@@ -159,6 +159,7 @@ enum StorageTypeProto {
   ARCHIVE = 3;
   RAM_DISK = 4;
   PROVIDED = 5;
+  NVDIMM = 6;
 }
 
 /**

hadoop-hdds/interface-server/src/main/proto/ScmServerProtocol.proto

@@ -143,6 +143,8 @@ enum Status {
   CONTAINER_ALREADY_CLOSED = 45;
   CONTAINER_ALREADY_CLOSING = 46;
   UNSUPPORTED_OPERATION = 47;
+  CANNOT_CREATE_PIPELINE_FOR_EMPTY_TIER = 48;
+  UNSUPPORTED_NON_UNIFORM_STORAGE_TIER = 49;
 }
 
 /**

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/PlacementPolicy.java

@@ -22,42 +22,45 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import org.apache.hadoop.fs.StorageType;
 import org.apache.hadoop.hdds.protocol.DatanodeDetails;
 import org.apache.hadoop.hdds.scm.container.ContainerReplica;
+import org.apache.hadoop.hdds.scm.exceptions.SCMException;
 
 /**
  * A PlacementPolicy support choosing datanodes to build
  * pipelines or containers with specified constraints.
  */
 public interface PlacementPolicy {
-
   default List<DatanodeDetails> chooseDatanodes(
-          List<DatanodeDetails> excludedNodes,
-          List<DatanodeDetails> favoredNodes, int nodesRequired,
-          long metadataSizeRequired, long dataSizeRequired) throws IOException {
+      List<DatanodeDetails> excludedNodes,
+      List<DatanodeDetails> favoredNodes, int nodesRequired,
+      long metadataSizeRequired, long dataSizeRequired,
+      StorageType storageType) throws SCMException {
     return this.chooseDatanodes(Collections.emptyList(), excludedNodes,
-            favoredNodes, nodesRequired, metadataSizeRequired,
-            dataSizeRequired);
+        favoredNodes, nodesRequired, metadataSizeRequired,
+        dataSizeRequired, storageType);
   }
 
   /**
    * Given an initial set of datanodes and the size required,
    * return set of datanodes that satisfy the nodes and size requirement.
    *
-   * @param usedNodes - List of nodes already chosen for pipeline
-   * @param excludedNodes - list of nodes to be excluded.
-   * @param favoredNodes - list of nodes preferred.
-   * @param nodesRequired - number of datanodes required.
-   * @param dataSizeRequired - size required for the container.
+   * @param usedNodes            - List of nodes already chosen for pipeline
+   * @param excludedNodes        - list of nodes to be excluded.
+   * @param favoredNodes         - list of nodes preferred.
+   * @param nodesRequired        - number of datanodes required.
    * @param metadataSizeRequired - size required for Ratis metadata.
+   * @param dataSizeRequired     - size required for the container.
+   * @param storageType          - StorageType required for the container.
    * @return list of datanodes chosen.
    * @throws IOException
    */
   List<DatanodeDetails> chooseDatanodes(List<DatanodeDetails> usedNodes,
           List<DatanodeDetails> excludedNodes,
           List<DatanodeDetails> favoredNodes,
           int nodesRequired, long metadataSizeRequired,
-          long dataSizeRequired) throws IOException;
+          long dataSizeRequired, StorageType storageType) throws SCMException;
 
   /**
    * Given a list of datanode and the number of replicas required, return