HDDS-13037. Let container create command support STANDALONE , RATIS and EC containers

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

@@ -179,10 +179,13 @@ ContainerDataProto readContainer(long containerID)
    * @return ContainerInfo
    * @throws IOException - in case of error.
    */
+  @Deprecated
   ContainerWithPipeline createContainer(HddsProtos.ReplicationType type,
       HddsProtos.ReplicationFactor replicationFactor,
       String owner) throws IOException;
 
+  ContainerWithPipeline createContainer(ReplicationConfig replicationConfig, String owner) throws IOException;
+
   /**
    * Gets the list of underReplicated and unClosed containers on a decommissioning node.
    *

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

@@ -81,6 +81,8 @@ ContainerWithPipeline allocateContainer(
       HddsProtos.ReplicationFactor factor, String owner)
       throws IOException;
 
+  ContainerWithPipeline allocateContainer(ReplicationConfig replicationConfig, String owner) throws IOException;
+
   /**
    * Ask SCM the location of the container. SCM responds with a group of
    * nodes where this container and its replicas are located.

hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/scm/protocolPB/StorageContainerLocationProtocolClientSideTranslatorPB.java

@@ -42,6 +42,7 @@
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos.DeletedBlocksTransactionInfo;
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos.GetScmInfoResponseProto;
+import org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationFactor;
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos.TransferLeadershipRequestProto;
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos.UpgradeFinalizationStatus;
 import org.apache.hadoop.hdds.protocol.proto.StorageContainerLocationProtocolProtos;
@@ -222,13 +223,33 @@ private ScmContainerLocationResponse submitRpcRequest(
   public ContainerWithPipeline allocateContainer(
       HddsProtos.ReplicationType type, HddsProtos.ReplicationFactor factor,
       String owner) throws IOException {
+    ReplicationConfig replicationConfig =
+        ReplicationConfig.fromProtoTypeAndFactor(type, factor);
+    return allocateContainer(replicationConfig, owner);
+  }
 
-    ContainerRequestProto request = ContainerRequestProto.newBuilder()
-        .setTraceID(TracingUtil.exportCurrentSpan())
-        .setReplicationFactor(factor)
-        .setReplicationType(type)
-        .setOwner(owner)
-        .build();
+  @Override
+  public ContainerWithPipeline allocateContainer(
+      ReplicationConfig replicationConfig, String owner) throws IOException {
+
+    ContainerRequestProto request;
+    if (replicationConfig.getReplicationType() == HddsProtos.ReplicationType.EC) {
+      HddsProtos.ECReplicationConfig ecProto = ((ECReplicationConfig) replicationConfig).toProto();
+      request = ContainerRequestProto.newBuilder()
+          .setTraceID(TracingUtil.exportCurrentSpan())
+          .setEcReplicationConfig(ecProto)
+          .setReplicationFactor(ReplicationFactor.ONE)// Set for backward compatibility, ignored for EC.
+          .setReplicationType(replicationConfig.getReplicationType())
+          .setOwner(owner)
+          .build();
+    } else {
+      request = ContainerRequestProto.newBuilder()
+          .setTraceID(TracingUtil.exportCurrentSpan())
+          .setReplicationFactor(ReplicationFactor.valueOf(replicationConfig.getReplication()))
+          .setReplicationType(replicationConfig.getReplicationType())
+          .setOwner(owner)
+          .build();
+    }
 
     ContainerResponseProto response =
         submitRequest(Type.AllocateContainer,
@@ -239,8 +260,7 @@ public ContainerWithPipeline allocateContainer(
       throw new IOException(response.hasErrorMessage() ?
           response.getErrorMessage() : "Allocate container failed.");
     }
-    return ContainerWithPipeline.fromProtobuf(
-        response.getContainerWithPipeline());
+    return ContainerWithPipeline.fromProtobuf(response.getContainerWithPipeline());
   }
 
   @Override

hadoop-hdds/interface-admin/src/main/proto/ScmAdminProtocol.proto

@@ -203,10 +203,11 @@ enum Type {
 */
 message ContainerRequestProto {
   // Ozone only support replication of either 1 or 3.
-  required ReplicationFactor replicationFactor = 2;
+  optional ReplicationFactor replicationFactor = 2;
   required ReplicationType  replicationType = 3;
   required string owner = 4;
   optional string traceID = 5;
+  optional ECReplicationConfig ecReplicationConfig = 6;
 }
 
 /**

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

@@ -194,6 +194,9 @@ public ContainerInfo allocateContainer(
     if (pipelines.isEmpty()) {
       try {
         pipeline = pipelineManager.createPipeline(replicationConfig);
+        if (replicationConfig.getReplicationType() == HddsProtos.ReplicationType.EC) {
+          pipelineManager.openPipeline(pipeline.getId());
+        }
         pipelineManager.waitPipelineReady(pipeline.getId(), 0);
       } catch (IOException e) {
         scmContainerManagerMetrics.incNumFailureCreateContainers();

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

@@ -753,9 +753,15 @@ public GetContainerReplicasResponseProto getContainerReplicas(
 
   public ContainerResponseProto allocateContainer(ContainerRequestProto request,
       int clientVersion) throws IOException {
-    ContainerWithPipeline cp = impl
-        .allocateContainer(request.getReplicationType(),
-            request.getReplicationFactor(), request.getOwner());
+    ContainerWithPipeline cp;
+    if (request.getReplicationType() == HddsProtos.ReplicationType.EC) {
+      cp = impl
+          .allocateContainer(new ECReplicationConfig(request.getEcReplicationConfig()), request.getOwner());
+    } else {
+      cp = impl
+          .allocateContainer(request.getReplicationType(), 
+              request.getReplicationFactor(), request.getOwner());
+    }
     return ContainerResponseProto.newBuilder()
         .setContainerWithPipeline(cp.getProtobuf(clientVersion))
         .setErrorCode(ContainerResponseProto.Error.success)

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

@@ -230,10 +230,16 @@ public void join() throws InterruptedException {
   public ContainerWithPipeline allocateContainer(HddsProtos.ReplicationType
       replicationType, HddsProtos.ReplicationFactor factor,
       String owner) throws IOException {
+    ReplicationConfig replicationConfig =
+        ReplicationConfig.fromProtoTypeAndFactor(replicationType, factor);
+    return allocateContainer(replicationConfig, owner);
+  }
 
+  @Override
+  public ContainerWithPipeline allocateContainer(ReplicationConfig replicationConfig, String owner) throws IOException {
     Map<String, String> auditMap = Maps.newHashMap();
-    auditMap.put("replicationType", String.valueOf(replicationType));
-    auditMap.put("factor", String.valueOf(factor));
+    auditMap.put("replicationType", String.valueOf(replicationConfig.getReplicationType()));
+    auditMap.put("replication", String.valueOf(replicationConfig.getReplication()));
     auditMap.put("owner", String.valueOf(owner));
 
     try {
@@ -244,7 +250,7 @@ public ContainerWithPipeline allocateContainer(HddsProtos.ReplicationType
       getScm().checkAdminAccess(getRemoteUser(), false);
       final ContainerInfo container = scm.getContainerManager()
           .allocateContainer(
-              ReplicationConfig.fromProtoTypeAndFactor(replicationType, factor),
+              replicationConfig,
               owner);
       final Pipeline pipeline = scm.getPipelineManager()
           .getPipeline(container.getPipelineID());

hadoop-ozone/cli-admin/src/main/java/org/apache/hadoop/hdds/scm/cli/ContainerOperationClient.java

@@ -221,6 +221,35 @@ public ContainerWithPipeline createContainer(HddsProtos.ReplicationType type,
     }
   }
 
+  @Override
+  public ContainerWithPipeline createContainer(ReplicationConfig replicationConfig, String owner) throws IOException {
+    XceiverClientSpi client = null;
+    XceiverClientManager clientManager = getXceiverClientManager();
+    try {
+      ContainerWithPipeline containerWithPipeline;
+      if (replicationConfig.getReplicationType() == HddsProtos.ReplicationType.EC) {
+        containerWithPipeline =
+            storageContainerLocationClient.allocateContainer(replicationConfig,
+                owner);
+      } else {
+        containerWithPipeline =
+            storageContainerLocationClient.allocateContainer(replicationConfig.getReplicationType(),
+                HddsProtos.ReplicationFactor.valueOf(replicationConfig.getReplication()), 
+                owner);
+      }
+      Pipeline pipeline = containerWithPipeline.getPipeline();
+      // connect to pipeline leader and allocate container on leader datanode.
+      client = clientManager.acquireClient(pipeline);
+      createContainer(client,
+          containerWithPipeline.getContainerInfo().getContainerID());
+      return containerWithPipeline;
+    } finally {
+      if (client != null) {
+        clientManager.releaseClient(client, false);
+      }
+    }
+  }
+
   @Override
   public Map<String, List<ContainerID>> getContainersOnDecomNode(DatanodeDetails dn) throws IOException {
     return storageContainerLocationClient.getContainersOnDecomNode(dn);

hadoop-ozone/cli-admin/src/main/java/org/apache/hadoop/hdds/scm/cli/container/CreateSubcommand.java

@@ -19,9 +19,14 @@
 
 import java.io.IOException;
 import org.apache.hadoop.hdds.cli.HddsVersionProvider;
+import org.apache.hadoop.hdds.client.ReplicationConfig;
+import org.apache.hadoop.hdds.conf.OzoneConfiguration;
+import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
 import org.apache.hadoop.hdds.scm.cli.ScmSubcommand;
 import org.apache.hadoop.hdds.scm.client.ScmClient;
 import org.apache.hadoop.hdds.scm.container.common.helpers.ContainerWithPipeline;
+import org.apache.hadoop.ozone.shell.ShellReplicationOptions;
+import picocli.CommandLine;
 import picocli.CommandLine.Command;
 import picocli.CommandLine.Option;
 
@@ -30,7 +35,8 @@
  */
 @Command(
     name = "create",
-    description = "Create container",
+    description = "Create container. If no replication config provided, " +
+        "defaults to STAND_ALONE with replication factor ONE.",
     mixinStandardHelpOptions = true,
     versionProvider = HddsVersionProvider.class)
 public class CreateSubcommand extends ScmSubcommand {
@@ -39,10 +45,19 @@ public class CreateSubcommand extends ScmSubcommand {
       names = { "-o", "--owner"})
   private String owner;
 
+  @CommandLine.Mixin
+  private ShellReplicationOptions containerReplicationOptions;
+
   @Override
   public void execute(ScmClient scmClient) throws IOException {
-    ContainerWithPipeline container = scmClient.createContainer(owner);
-    System.out.printf("Container %s is created.%n",
-        container.getContainerInfo().getContainerID());
+    ReplicationConfig replicationConfig = containerReplicationOptions.fromParamsOrConfig(new OzoneConfiguration());
+    if (replicationConfig == null) {
+      // if replication options not provided via command then by default STAND_ALONE container will be created.
+      replicationConfig = ReplicationConfig.fromProtoTypeAndFactor(HddsProtos.ReplicationType.STAND_ALONE, 
+          HddsProtos.ReplicationFactor.ONE);
+    }
+    ContainerWithPipeline container = scmClient.createContainer(replicationConfig, owner);
+    System.out.printf("Container %s is created with replication config %s.%n",
+        container.getContainerInfo().getContainerID(), replicationConfig);
   }
 }

hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/TestAllocateContainer.java

@@ -20,7 +20,9 @@
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 
+import org.apache.hadoop.hdds.client.ECReplicationConfig;
 import org.apache.hadoop.hdds.conf.OzoneConfiguration;
+import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
 import org.apache.hadoop.hdds.scm.container.common.helpers.ContainerWithPipeline;
 import org.apache.hadoop.hdds.scm.protocolPB.StorageContainerLocationProtocolClientSideTranslatorPB;
 import org.apache.hadoop.hdds.utils.IOUtils;
@@ -73,4 +75,25 @@ public void testAllocateNull() {
             SCMTestUtils.getReplicationType(conf),
             SCMTestUtils.getReplicationFactor(conf), null));
   }
+
+  @Test
+  public void testAllocateRatis() throws Exception {
+    ContainerWithPipeline container =
+        storageContainerLocationClient.allocateContainer(HddsProtos.ReplicationType.RATIS, 
+            HddsProtos.ReplicationFactor.THREE, OzoneConsts.OZONE);
+
+    assertNotNull(container);
+    assertNotNull(container.getPipeline().getFirstNode());
+  }
+
+  @Test
+  public void testAllocateEC() throws Exception {
+    ECReplicationConfig ecReplicationConfig = new ECReplicationConfig("RS-3-2-1024k");
+
+    ContainerWithPipeline container =
+        storageContainerLocationClient.allocateContainer(ecReplicationConfig, OzoneConsts.OZONE);
+
+    assertNotNull(container);
+    assertNotNull(container.getPipeline().getFirstNode());
+  }
 }

hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/TestContainerOperations.java

@@ -34,6 +34,8 @@
 import org.apache.commons.lang3.RandomStringUtils;
 import org.apache.commons.lang3.RandomUtils;
 import org.apache.hadoop.hdds.client.BlockID;
+import org.apache.hadoop.hdds.client.ECReplicationConfig;
+import org.apache.hadoop.hdds.client.ReplicationConfig;
 import org.apache.hadoop.hdds.conf.OzoneConfiguration;
 import org.apache.hadoop.hdds.protocol.DatanodeDetails;
 import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
@@ -265,4 +267,25 @@ public void testNodeOperationalStates() throws Exception {
       nm.setNodeOperationalState(node, originalState);
     }
   }
+
+  @Test
+  public void testCreateRatis() throws Exception {
+    ContainerWithPipeline container = storageClient.createContainer(
+        ReplicationConfig.fromProtoTypeAndFactor(HddsProtos.ReplicationType.RATIS, HddsProtos.ReplicationFactor.THREE),
+        OzoneConsts.OZONE);
+
+    assertEquals(container.getContainerInfo().getContainerID(),
+        storageClient.getContainer(container.getContainerInfo().getContainerID())
+            .getContainerID());
+  }
+
+  @Test
+  public void testCreateEC() throws Exception {
+    ECReplicationConfig ecConfig = new ECReplicationConfig("RS-3-2-1024k");
+    ContainerWithPipeline container = storageClient.createContainer(ecConfig, OzoneConsts.OZONE);
+
+    assertEquals(container.getContainerInfo().getContainerID(),
+        storageClient.getContainer(container.getContainerInfo().getContainerID())
+            .getContainerID());
+  }
 }