Add scm pipeline filtering

Author: greenwich

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

@@ -17,17 +17,21 @@
 
 package org.apache.hadoop.hdds.scm;
 
+import org.apache.hadoop.hdds.protocol.StorageType;
+
 /**
  * The information of the request of pipeline.
  */
 public final class PipelineRequestInformation {
   private final long size;
+  private final StorageType storageType;
 
   /**
    * Builder for PipelineRequestInformation.
    */
   public static class Builder {
     private long size;
+    private StorageType storageType;
 
     public static Builder getBuilder() {
       return new Builder();
@@ -43,16 +47,26 @@ public Builder setSize(long sz) {
       return this;
     }
 
+    public Builder setStorageType(StorageType st) {
+      this.storageType = st;
+      return this;
+    }
+
     public PipelineRequestInformation build() {
-      return new PipelineRequestInformation(size);
+      return new PipelineRequestInformation(size, storageType);
     }
   }
 
-  private PipelineRequestInformation(long size) {
+  private PipelineRequestInformation(long size, StorageType storageType) {
     this.size = size;
+    this.storageType = storageType;
   }
 
   public long getSize() {
     return size;
   }
+
+  public StorageType getStorageType() {
+    return storageType;
+  }
 }

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

@@ -165,7 +165,7 @@ public AllocatedBlock allocateBlock(final long size,
     }
 
     ContainerInfo containerInfo = writableContainerFactory.getContainer(
-        size, replicationConfig, owner, excludeList);
+        size, replicationConfig, owner, excludeList, storageType);
 
     if (containerInfo != null) {
       return newBlock(containerInfo);

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

@@ -0,0 +1,74 @@
+/*
+ * 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.scm.pipeline;
+
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.UUID;
+import java.util.stream.Collectors;
+import org.apache.hadoop.hdds.protocol.DatanodeDetails;
+import org.apache.hadoop.hdds.protocol.StorageType;
+import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.StorageReportProto;
+import org.apache.hadoop.hdds.scm.node.DatanodeInfo;
+import org.apache.hadoop.hdds.scm.node.NodeManager;
+import org.apache.hadoop.hdds.scm.node.NodeStatus;
+
+/**
+ * Utility for filtering pipelines by StorageType. Builds a set of
+ * qualifying node UUIDs (nodes that have at least one volume of the
+ * requested StorageType), then filters pipelines to those whose every
+ * member is in that set.
+ */
+final class PipelineStorageTypeFilter {
+
+  private PipelineStorageTypeFilter() {
+  }
+
+  static Set<UUID> getNodesWithStorageType(NodeManager nodeManager,
+      StorageType storageType) {
+    Set<UUID> result = new HashSet<>();
+    for (DatanodeDetails dn :
+        nodeManager.getNodes(NodeStatus.inServiceHealthy())) {
+      DatanodeInfo info = nodeManager.getDatanodeInfo(dn);
+      if (info == null) {
+        continue;
+      }
+      for (StorageReportProto report : info.getStorageReports()) {
+        if (StorageType.valueOf(report.getStorageType()) == storageType) {
+          result.add(dn.getUuid());
+          break;
+        }
+      }
+    }
+    return result;
+  }
+
+  static List<Pipeline> filter(List<Pipeline> pipelines,
+      NodeManager nodeManager, StorageType storageType) {
+    if (storageType == null) {
+      return pipelines;
+    }
+    Set<UUID> qualifiedNodes =
+        getNodesWithStorageType(nodeManager, storageType);
+    return pipelines.stream()
+        .filter(p -> p.getNodes().stream()
+            .allMatch(dn -> qualifiedNodes.contains(dn.getUuid())))
+        .collect(Collectors.toList());
+  }
+}

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

@@ -25,6 +25,7 @@
 import org.apache.hadoop.hdds.client.ECReplicationConfig;
 import org.apache.hadoop.hdds.client.ReplicationConfig;
 import org.apache.hadoop.hdds.conf.ConfigurationSource;
+import org.apache.hadoop.hdds.protocol.StorageType;
 import org.apache.hadoop.hdds.scm.container.ContainerInfo;
 import org.apache.hadoop.hdds.scm.container.common.helpers.ExcludeList;
 import org.apache.hadoop.hdds.scm.pipeline.WritableECContainerProvider.WritableECContainerProviderConfig;
@@ -45,7 +46,8 @@ public WritableContainerFactory(StorageContainerManager scm) {
 
     this.ratisProvider = new WritableRatisContainerProvider(
         scm.getPipelineManager(),
-        scm.getContainerManager(), scm.getPipelineChoosePolicy());
+        scm.getContainerManager(), scm.getPipelineChoosePolicy(),
+        scm.getScmNodeManager());
     this.standaloneProvider = ratisProvider;
 
     WritableECContainerProviderConfig ecProviderConfig =
@@ -79,6 +81,25 @@ public ContainerInfo getContainer(final long size,
     }
   }
 
+  public ContainerInfo getContainer(final long size,
+      ReplicationConfig repConfig, String owner, ExcludeList excludeList,
+      StorageType storageType) throws IOException {
+    switch (repConfig.getReplicationType()) {
+    case STAND_ALONE:
+      return standaloneProvider.getContainer(size, repConfig, owner,
+          excludeList, storageType);
+    case RATIS:
+      return ratisProvider.getContainer(size, repConfig, owner,
+          excludeList, storageType);
+    case EC:
+      return ecProvider.getContainer(size, (ECReplicationConfig) repConfig,
+          owner, excludeList, storageType);
+    default:
+      throw new IOException(repConfig.getReplicationType()
+          + " is an invalid replication type");
+    }
+  }
+
   private long getConfiguredContainerSize(ConfigurationSource conf) {
     return (long) conf.getStorageSize(OZONE_SCM_CONTAINER_SIZE,
         OZONE_SCM_CONTAINER_SIZE_DEFAULT, BYTES);

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

@@ -19,6 +19,7 @@
 
 import java.io.IOException;
 import org.apache.hadoop.hdds.client.ReplicationConfig;
+import org.apache.hadoop.hdds.protocol.StorageType;
 import org.apache.hadoop.hdds.scm.container.ContainerInfo;
 import org.apache.hadoop.hdds.scm.container.common.helpers.ExcludeList;
 
@@ -53,4 +54,10 @@ ContainerInfo getContainer(long size, T repConfig,
       String owner, ExcludeList excludeList)
       throws IOException;
 
+  default ContainerInfo getContainer(long size, T repConfig,
+      String owner, ExcludeList excludeList, StorageType storageType)
+      throws IOException {
+    return getContainer(size, repConfig, owner, excludeList);
+  }
+
 }

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

@@ -34,6 +34,7 @@
 import org.apache.hadoop.hdds.conf.PostConstruct;
 import org.apache.hadoop.hdds.conf.ReconfigurableConfig;
 import org.apache.hadoop.hdds.protocol.DatanodeDetails;
+import org.apache.hadoop.hdds.protocol.StorageType;
 import org.apache.hadoop.hdds.scm.PipelineChoosePolicy;
 import org.apache.hadoop.hdds.scm.PipelineRequestInformation;
 import org.apache.hadoop.hdds.scm.container.ContainerID;
@@ -93,6 +94,20 @@ public WritableECContainerProvider(WritableECContainerProviderConfig config,
   public ContainerInfo getContainer(final long size,
       ECReplicationConfig repConfig, String owner, ExcludeList excludeList)
       throws IOException {
+    return getContainerInternal(size, repConfig, owner, excludeList, null);
+  }
+
+  @Override
+  public ContainerInfo getContainer(final long size,
+      ECReplicationConfig repConfig, String owner, ExcludeList excludeList,
+      StorageType storageType) throws IOException {
+    return getContainerInternal(size, repConfig, owner, excludeList,
+        storageType);
+  }
+
+  private ContainerInfo getContainerInternal(final long size,
+      ECReplicationConfig repConfig, String owner, ExcludeList excludeList,
+      StorageType storageType) throws IOException {
     int maximumPipelines = getMaximumPipelines(repConfig);
     int openPipelineCount;
     synchronized (this) {
@@ -115,12 +130,15 @@ public ContainerInfo getContainer(final long size,
     }
     List<Pipeline> existingPipelines = pipelineManager.getPipelines(
         repConfig, Pipeline.PipelineState.OPEN);
+    existingPipelines = PipelineStorageTypeFilter.filter(
+        existingPipelines, nodeManager, storageType);
     final int pipelineCount = existingPipelines.size();
     LOG.debug("Checking existing pipelines: {}", existingPipelines);
 
     PipelineRequestInformation pri =
         PipelineRequestInformation.Builder.getBuilder()
             .setSize(size)
+            .setStorageType(storageType)
             .build();
     while (!existingPipelines.isEmpty()) {
       int pipelineIndex =

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

@@ -22,12 +22,14 @@
 import java.util.List;
 import java.util.stream.Collectors;
 import org.apache.hadoop.hdds.client.ReplicationConfig;
+import org.apache.hadoop.hdds.protocol.StorageType;
 import org.apache.hadoop.hdds.scm.PipelineChoosePolicy;
 import org.apache.hadoop.hdds.scm.PipelineRequestInformation;
 import org.apache.hadoop.hdds.scm.container.ContainerInfo;
 import org.apache.hadoop.hdds.scm.container.ContainerManager;
 import org.apache.hadoop.hdds.scm.container.common.helpers.ExcludeList;
 import org.apache.hadoop.hdds.scm.exceptions.SCMException;
+import org.apache.hadoop.hdds.scm.node.NodeManager;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -43,20 +45,44 @@ public class WritableRatisContainerProvider
   private final PipelineManager pipelineManager;
   private final PipelineChoosePolicy pipelineChoosePolicy;
   private final ContainerManager containerManager;
+  private final NodeManager nodeManager;
 
   public WritableRatisContainerProvider(
       PipelineManager pipelineManager,
       ContainerManager containerManager,
-      PipelineChoosePolicy pipelineChoosePolicy) {
+      PipelineChoosePolicy pipelineChoosePolicy,
+      NodeManager nodeManager) {
     this.pipelineManager = pipelineManager;
     this.containerManager = containerManager;
     this.pipelineChoosePolicy = pipelineChoosePolicy;
+    this.nodeManager = nodeManager;
+  }
+
+  public WritableRatisContainerProvider(
+      PipelineManager pipelineManager,
+      ContainerManager containerManager,
+      PipelineChoosePolicy pipelineChoosePolicy) {
+    this(pipelineManager, containerManager, pipelineChoosePolicy, null);
   }
 
   @Override
   public ContainerInfo getContainer(final long size,
       ReplicationConfig repConfig, String owner, ExcludeList excludeList)
       throws IOException {
+    return getContainerInternal(size, repConfig, owner, excludeList, null);
+  }
+
+  @Override
+  public ContainerInfo getContainer(final long size,
+      ReplicationConfig repConfig, String owner, ExcludeList excludeList,
+      StorageType storageType) throws IOException {
+    return getContainerInternal(size, repConfig, owner, excludeList,
+        storageType);
+  }
+
+  private ContainerInfo getContainerInternal(final long size,
+      ReplicationConfig repConfig, String owner, ExcludeList excludeList,
+      StorageType storageType) throws IOException {
     /*
       Here is the high level logic.
 
@@ -77,10 +103,13 @@ public ContainerInfo getContainer(final long size,
     //in downstream managers.
 
     PipelineRequestInformation req =
-        PipelineRequestInformation.Builder.getBuilder().setSize(size).build();
+        PipelineRequestInformation.Builder.getBuilder()
+            .setSize(size)
+            .setStorageType(storageType)
+            .build();
 
     ContainerInfo containerInfo =
-        getContainer(repConfig, owner, excludeList, req);
+        getContainer(repConfig, owner, excludeList, req, storageType);
     if (containerInfo != null) {
       return containerInfo;
     }
@@ -126,7 +155,8 @@ public ContainerInfo getContainer(final long size,
 
     // If Exception occurred or successful creation of pipeline do one
     // final try to fetch pipelines.
-    containerInfo = getContainer(repConfig, owner, excludeList, req);
+    containerInfo = getContainer(repConfig, owner, excludeList, req,
+        storageType);
     if (containerInfo != null) {
       return containerInfo;
     }
@@ -143,14 +173,19 @@ public ContainerInfo getContainer(final long size,
 
   @Nullable
   private ContainerInfo getContainer(ReplicationConfig repConfig, String owner,
-      ExcludeList excludeList, PipelineRequestInformation req) {
+      ExcludeList excludeList, PipelineRequestInformation req,
+      StorageType storageType) {
     // Acquire pipeline manager lock, to avoid any updates to pipeline
     // while allocate container happens. This is to avoid scenario like
     // mentioned in HDDS-5655.
     pipelineManager.acquireReadLock();
     try {
       List<Pipeline> availablePipelines = findPipelinesByState(repConfig,
           excludeList, Pipeline.PipelineState.OPEN);
+      if (nodeManager != null) {
+        availablePipelines = PipelineStorageTypeFilter.filter(
+            availablePipelines, nodeManager, storageType);
+      }
       return selectContainer(availablePipelines, req, owner, excludeList);
     } finally {
       pipelineManager.releaseReadLock();

hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/block/TestBlockManager.java

@@ -575,6 +575,36 @@ public void testBlockAllocationWithNoAvailablePipelines()
             new ExcludeList(), StorageType.DEFAULT));
   }
 
+  /**
+   * Integration test: verifies the full BlockManagerImpl → WritableContainerFactory
+   * → WritableRatisContainerProvider → PipelineStorageTypeFilter chain.
+   * MockNodeManager reports only DISK volumes. Requesting SSD should fail
+   * because the filter removes all existing pipelines, and newly created
+   * pipelines also contain DISK-only nodes so they too get filtered on
+   * the second attempt.
+   */
+  @Test
+  public void testAllocateBlockWithNonMatchingStorageTypeFails()
+      throws Exception {
+    pipelineManager.createPipeline(replicationConfig);
+    HddsTestUtils.openAllRatisPipelines(pipelineManager);
+
+    // Verify DISK allocation works (baseline — all nodes report DISK)
+    AllocatedBlock diskBlock = blockManager.allocateBlock(
+        DEFAULT_BLOCK_SIZE, replicationConfig, OzoneConsts.OZONE,
+        new ExcludeList(), StorageType.DISK);
+    assertNotNull(diskBlock);
+
+    // SSD allocation should fail: MockNodeManager nodes only have DISK
+    // volumes, so the PipelineStorageTypeFilter removes all pipelines,
+    // pipeline creation adds another DISK-only pipeline which also gets
+    // filtered, resulting in IOException.
+    assertThrows(IOException.class,
+        () -> blockManager.allocateBlock(DEFAULT_BLOCK_SIZE,
+            replicationConfig, OzoneConsts.OZONE,
+            new ExcludeList(), StorageType.SSD));
+  }
+
   private class DatanodeCommandHandler implements
       EventHandler<CommandForDatanode> {