Storage tiering. Add creation of the typed pipelines

Author: greenwich

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

@@ -445,6 +445,11 @@ public final class ScmConfigKeys {
   public static final String OZONE_SCM_PIPELINE_SCRUB_INTERVAL_DEFAULT =
       "150s";
 
+  public static final String OZONE_SCM_PIPELINE_CREATION_STORAGE_TYPE_AWARE =
+      "ozone.scm.pipeline.creation.storage-type-aware.enabled";
+  public static final boolean
+      OZONE_SCM_PIPELINE_CREATION_STORAGE_TYPE_AWARE_DEFAULT = false;
+
   // Allow SCM to auto create factor ONE ratis pipeline.
   public static final String OZONE_SCM_PIPELINE_AUTO_CREATE_FACTOR_ONE =
       "ozone.scm.pipeline.creation.auto.factor.one";

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

@@ -25,11 +25,14 @@
 import static org.apache.hadoop.hdds.scm.ha.SCMService.Event.PRE_CHECK_COMPLETED;
 import static org.apache.hadoop.hdds.scm.ha.SCMService.Event.UNHEALTHY_TO_HEALTHY_NODE_HANDLER_TRIGGERED;
 
+import com.google.common.annotations.VisibleForTesting;
 import com.google.common.util.concurrent.ThreadFactoryBuilder;
 import java.io.IOException;
 import java.time.Clock;
+import java.util.AbstractMap;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Map;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.locks.Lock;
@@ -40,6 +43,7 @@
 import org.apache.hadoop.hdds.client.ReplicationConfig;
 import org.apache.hadoop.hdds.client.StandaloneReplicationConfig;
 import org.apache.hadoop.hdds.conf.ConfigurationSource;
+import org.apache.hadoop.hdds.protocol.StorageType;
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationFactor;
 import org.apache.hadoop.hdds.scm.ScmConfigKeys;
@@ -88,6 +92,7 @@ public class BackgroundPipelineCreator implements SCMService {
   private final AtomicBoolean running = new AtomicBoolean(false);
   private final long intervalInMillis;
   private final Clock clock;
+  private final boolean storageTypeAwareCreation;
 
   BackgroundPipelineCreator(PipelineManager pipelineManager,
       ConfigurationSource conf, SCMContext scmContext, Clock clock) {
@@ -110,6 +115,10 @@ public class BackgroundPipelineCreator implements SCMService {
         ScmConfigKeys.OZONE_SCM_PIPELINE_CREATION_INTERVAL_DEFAULT,
         TimeUnit.MILLISECONDS);
 
+    this.storageTypeAwareCreation = conf.getBoolean(
+        ScmConfigKeys.OZONE_SCM_PIPELINE_CREATION_STORAGE_TYPE_AWARE,
+        ScmConfigKeys.OZONE_SCM_PIPELINE_CREATION_STORAGE_TYPE_AWARE_DEFAULT);
+
     threadName = scmContext.threadNamePrefix() + THREAD_NAME;
   }
 
@@ -203,7 +212,8 @@ private boolean skipCreation(ReplicationConfig replicationConfig,
     return true;
   }
 
-  private void createPipelines() throws RuntimeException {
+  @VisibleForTesting
+  void createPipelines() throws RuntimeException {
     // TODO: #CLUTIL Different replication factor may need to be supported
     HddsProtos.ReplicationType type = HddsProtos.ReplicationType.valueOf(
         conf.get(OzoneConfigKeys.OZONE_REPLICATION_TYPE,
@@ -212,8 +222,7 @@ private void createPipelines() throws RuntimeException {
         ScmConfigKeys.OZONE_SCM_PIPELINE_AUTO_CREATE_FACTOR_ONE,
         ScmConfigKeys.OZONE_SCM_PIPELINE_AUTO_CREATE_FACTOR_ONE_DEFAULT);
 
-    List<ReplicationConfig> list =
-        new ArrayList<>();
+    List<ReplicationConfig> replicationConfigs = new ArrayList<>();
     for (HddsProtos.ReplicationFactor factor : HddsProtos.ReplicationFactor
         .values()) {
       if (factor == ReplicationFactor.ZERO) {
@@ -233,10 +242,20 @@ private void createPipelines() throws RuntimeException {
         // Skip this iteration for creating pipeline
         continue;
       }
-      list.add(replicationConfig);
+      replicationConfigs.add(replicationConfig);
+    }
+
+    if (storageTypeAwareCreation) {
+      createTypedPipelines(replicationConfigs);
+    } else {
+      createUntypedPipelines(replicationConfigs);
     }
 
-    LoopingIterator it = new LoopingIterator(list);
+    LOG.debug("BackgroundPipelineCreator createPipelines finished.");
+  }
+
+  private void createUntypedPipelines(List<ReplicationConfig> configs) {
+    LoopingIterator it = new LoopingIterator(configs);
     while (it.hasNext()) {
       ReplicationConfig replicationConfig =
           (ReplicationConfig) it.next();
@@ -251,8 +270,46 @@ private void createPipelines() throws RuntimeException {
         it.remove();
       }
     }
+  }
 
-    LOG.debug("BackgroundPipelineCreator createPipelines finished.");
+  private void createTypedPipelines(List<ReplicationConfig> configs) {
+    // Build (ReplicationConfig, StorageType) pairs: for each config,
+    // one null entry (untyped) plus one per concrete StorageType.
+    StorageType[] storageTypes = {
+        StorageType.SSD, StorageType.DISK, StorageType.ARCHIVE
+    };
+    List<Map.Entry<ReplicationConfig, StorageType>> pairs = new ArrayList<>();
+    for (ReplicationConfig config : configs) {
+      pairs.add(new AbstractMap.SimpleEntry<>(config, null));
+      for (StorageType st : storageTypes) {
+        pairs.add(new AbstractMap.SimpleEntry<>(config, st));
+      }
+    }
+
+    LoopingIterator it = new LoopingIterator(pairs);
+    while (it.hasNext()) {
+      @SuppressWarnings("unchecked")
+      Map.Entry<ReplicationConfig, StorageType> entry =
+          (Map.Entry<ReplicationConfig, StorageType>) it.next();
+
+      try {
+        Pipeline pipeline;
+        if (entry.getValue() == null) {
+          pipeline = pipelineManager.createPipeline(entry.getKey());
+        } else {
+          pipeline = pipelineManager.createPipeline(
+              entry.getKey(), entry.getValue());
+        }
+        LOG.info("Created new pipeline {} with StorageType {}",
+            pipeline, entry.getValue());
+      } catch (IOException ioe) {
+        it.remove();
+      } catch (Throwable t) {
+        LOG.error("Error while creating pipelines for StorageType "
+            + entry.getValue(), t);
+        it.remove();
+      }
+    }
   }
 
   @Override

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

@@ -25,6 +25,7 @@
 import java.util.Set;
 import org.apache.hadoop.hdds.client.ReplicationConfig;
 import org.apache.hadoop.hdds.protocol.DatanodeDetails;
+import org.apache.hadoop.hdds.protocol.StorageType;
 import org.apache.hadoop.hdds.scm.container.ContainerID;
 import org.apache.hadoop.hdds.scm.container.ContainerReplica;
 import org.apache.hadoop.hdds.utils.db.CodecException;
@@ -39,6 +40,11 @@ public interface PipelineManager extends Closeable, PipelineManagerMXBean {
   Pipeline createPipeline(ReplicationConfig replicationConfig)
       throws IOException;
 
+  default Pipeline createPipeline(ReplicationConfig replicationConfig,
+      StorageType storageType) throws IOException {
+    return createPipeline(replicationConfig);
+  }
+
   Pipeline createPipeline(ReplicationConfig replicationConfig,
                           List<DatanodeDetails> excludedNodes,
                           List<DatanodeDetails> favoredNodes)

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

@@ -30,6 +30,7 @@
 import java.util.Map;
 import java.util.NavigableSet;
 import java.util.Set;
+import java.util.UUID;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
@@ -41,6 +42,7 @@
 import org.apache.hadoop.hdds.client.StandaloneReplicationConfig;
 import org.apache.hadoop.hdds.conf.ConfigurationSource;
 import org.apache.hadoop.hdds.protocol.DatanodeDetails;
+import org.apache.hadoop.hdds.protocol.StorageType;
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationFactor;
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationType;
@@ -56,6 +58,7 @@
 import org.apache.hadoop.hdds.scm.ha.SCMServiceManager;
 import org.apache.hadoop.hdds.scm.node.DatanodeInfo;
 import org.apache.hadoop.hdds.scm.node.NodeManager;
+import org.apache.hadoop.hdds.scm.node.NodeStatus;
 import org.apache.hadoop.hdds.scm.server.upgrade.FinalizationManager;
 import org.apache.hadoop.hdds.server.events.EventPublisher;
 import org.apache.hadoop.hdds.utils.db.CodecException;
@@ -268,6 +271,33 @@ public Pipeline createPipeline(ReplicationConfig replicationConfig,
     }
   }
 
+  @Override
+  public Pipeline createPipeline(ReplicationConfig replicationConfig,
+      StorageType storageType) throws IOException {
+    if (storageType == null) {
+      return createPipeline(replicationConfig);
+    }
+    // Compute excluded nodes: all healthy nodes that do NOT have the
+    // requested StorageType.
+    List<DatanodeDetails> allHealthy =
+        nodeManager.getNodes(NodeStatus.inServiceHealthy());
+    Set<UUID> qualifiedNodeIds =
+        PipelineStorageTypeFilter.getNodesWithStorageType(
+            nodeManager, storageType);
+
+    if (qualifiedNodeIds.isEmpty()) {
+      throw new IOException("No healthy nodes with StorageType "
+          + storageType + " available for pipeline creation");
+    }
+
+    List<DatanodeDetails> excludedNodes = allHealthy.stream()
+        .filter(dn -> !qualifiedNodeIds.contains(dn.getUuid()))
+        .collect(Collectors.toList());
+
+    return createPipeline(replicationConfig, excludedNodes,
+        Collections.emptyList());
+  }
+
   private void checkIfPipelineCreationIsAllowed(
       ReplicationConfig replicationConfig) throws IOException {
     if (!isPipelineCreationAllowed() && !factorOne(replicationConfig)) {

hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestBackgroundPipelineCreatorStorageType.java

@@ -0,0 +1,103 @@
+/*
+ * 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 static org.mockito.Mockito.any;
+import static org.mockito.Mockito.atLeastOnce;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.io.IOException;
+import java.time.Instant;
+import java.time.ZoneOffset;
+import org.apache.hadoop.hdds.client.ReplicationConfig;
+import org.apache.hadoop.hdds.conf.OzoneConfiguration;
+import org.apache.hadoop.hdds.protocol.StorageType;
+import org.apache.hadoop.hdds.scm.ScmConfigKeys;
+import org.apache.hadoop.hdds.scm.ha.SCMContext;
+import org.apache.ozone.test.TestClock;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Tests for storage-type-aware pipeline creation in
+ * BackgroundPipelineCreator.
+ */
+public class TestBackgroundPipelineCreatorStorageType {
+
+  @Test
+  public void testStorageTypeAwareDisabled() throws IOException {
+    OzoneConfiguration conf = new OzoneConfiguration();
+    conf.setBoolean(
+        ScmConfigKeys.OZONE_SCM_PIPELINE_CREATION_STORAGE_TYPE_AWARE,
+        false);
+
+    PipelineManager pipelineManager = mock(PipelineManager.class);
+    when(pipelineManager.createPipeline(any(ReplicationConfig.class)))
+        .thenThrow(new IOException("exhausted"));
+
+    SCMContext scmContext = SCMContext.emptyContext();
+
+    TestClock clock = new TestClock(Instant.now(), ZoneOffset.UTC);
+    BackgroundPipelineCreator creator =
+        new BackgroundPipelineCreator(pipelineManager, conf, scmContext,
+            clock);
+
+    creator.createPipelines();
+
+    // Untyped createPipeline(ReplicationConfig) should have been called.
+    verify(pipelineManager, atLeastOnce())
+        .createPipeline(any(ReplicationConfig.class));
+    // Typed createPipeline(ReplicationConfig, StorageType) should NOT
+    // have been called.
+    verify(pipelineManager, never())
+        .createPipeline(any(ReplicationConfig.class),
+            any(StorageType.class));
+  }
+
+  @Test
+  public void testStorageTypeAwareEnabled() throws IOException {
+    OzoneConfiguration conf = new OzoneConfiguration();
+    conf.setBoolean(
+        ScmConfigKeys.OZONE_SCM_PIPELINE_CREATION_STORAGE_TYPE_AWARE,
+        true);
+
+    PipelineManager pipelineManager = mock(PipelineManager.class);
+    when(pipelineManager.createPipeline(any(ReplicationConfig.class)))
+        .thenThrow(new IOException("exhausted"));
+    when(pipelineManager.createPipeline(any(ReplicationConfig.class),
+        any(StorageType.class)))
+        .thenThrow(new IOException("exhausted"));
+
+    SCMContext scmContext = SCMContext.emptyContext();
+
+    TestClock clock = new TestClock(Instant.now(), ZoneOffset.UTC);
+    BackgroundPipelineCreator creator =
+        new BackgroundPipelineCreator(pipelineManager, conf, scmContext,
+            clock);
+
+    creator.createPipelines();
+
+    // When storage-type-aware is enabled, the typed method should be called
+    // for SSD, DISK, and ARCHIVE.
+    verify(pipelineManager, atLeastOnce())
+        .createPipeline(any(ReplicationConfig.class),
+            any(StorageType.class));
+  }
+}

hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestPipelineManagerImpl.java

@@ -71,6 +71,7 @@
 import org.apache.hadoop.hdds.protocol.DatanodeDetails;
 import org.apache.hadoop.hdds.protocol.DatanodeID;
 import org.apache.hadoop.hdds.protocol.MockDatanodeDetails;
+import org.apache.hadoop.hdds.protocol.StorageType;
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationFactor;
 import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos;
@@ -920,6 +921,34 @@ public void testWaitForAllocatedPipeline() throws IOException {
     pipelineManager.close();
   }
 
+  @Test
+  public void testCreatePipelineWithStorageType() throws Exception {
+    PipelineManagerImpl pipelineManager = createPipelineManager(true);
+
+    // MockNodeManager creates storage reports with DISK type by default.
+    // DISK-typed pipeline should succeed.
+    Pipeline diskPipeline = pipelineManager.createPipeline(
+        RatisReplicationConfig.getInstance(ReplicationFactor.THREE),
+        StorageType.DISK);
+    assertNotNull(diskPipeline);
+    assertEquals(3, diskPipeline.getNodes().size());
+
+    // SSD-typed pipeline should fail since no nodes have SSD storage.
+    assertThrows(IOException.class,
+        () -> pipelineManager.createPipeline(
+            RatisReplicationConfig.getInstance(ReplicationFactor.THREE),
+            StorageType.SSD));
+
+    // null StorageType should fall through to untyped creation.
+    Pipeline untypedPipeline = pipelineManager.createPipeline(
+        RatisReplicationConfig.getInstance(ReplicationFactor.THREE),
+        (StorageType) null);
+    assertNotNull(untypedPipeline);
+    assertEquals(3, untypedPipeline.getNodes().size());
+
+    pipelineManager.close();
+  }
+
   public void testCreatePipelineForRead() throws IOException {
     PipelineManager pipelineManager = createPipelineManager(true);
     List<DatanodeDetails> dns = nodeManager

hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestKeyManagerImpl.java

@@ -84,6 +84,7 @@
 import org.apache.hadoop.hdds.client.StandaloneReplicationConfig;
 import org.apache.hadoop.hdds.conf.OzoneConfiguration;
 import org.apache.hadoop.hdds.protocol.DatanodeDetails;
+import org.apache.hadoop.hdds.protocol.StorageType;
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationFactor;
 import org.apache.hadoop.hdds.scm.HddsTestUtils;
 import org.apache.hadoop.hdds.scm.HddsWhiteboxTestUtils;
@@ -227,7 +228,8 @@ public static void setUp() throws Exception {
             any(ReplicationConfig.class),
             anyString(),
             any(ExcludeList.class),
-            anyString())).thenThrow(
+            anyString(),
+            any(StorageType.class))).thenThrow(
                 new SCMException("SafeModePrecheck failed for allocateBlock",
             ResultCodes.SAFE_MODE_EXCEPTION));
     createVolume(VOLUME_NAME);