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,14 @@ 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

@@ -18,6 +18,8 @@
 package org.apache.hadoop.hdds.scm.protocolPB;
 
 import static org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationType.EC;
+import static org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationType.RATIS;
+import static org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationType.STAND_ALONE;
 import static org.apache.hadoop.hdds.protocol.proto.StorageContainerLocationProtocolProtos.SCMCloseContainerResponseProto.Status.CONTAINER_ALREADY_CLOSED;
 import static org.apache.hadoop.hdds.protocol.proto.StorageContainerLocationProtocolProtos.SCMCloseContainerResponseProto.Status.CONTAINER_ALREADY_CLOSING;
 
@@ -42,6 +44,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;
@@ -243,6 +246,51 @@ public ContainerWithPipeline allocateContainer(
         response.getContainerWithPipeline());
   }
 
+  @Override
+  public ContainerWithPipeline allocateContainer(
+      ReplicationConfig replicationConfig, String owner) throws IOException {
+
+    if (replicationConfig.getReplicationType() == RATIS || replicationConfig.getReplicationType() == STAND_ALONE) {
+      ContainerRequestProto request = ContainerRequestProto.newBuilder()
+          .setTraceID(TracingUtil.exportCurrentSpan())
+          .setReplicationFactor(ReplicationFactor.valueOf(replicationConfig.getReplication()))
+          .setReplicationType(replicationConfig.getReplicationType())
+          .setOwner(owner)
+          .build();
+
+      ContainerResponseProto response =
+          submitRequest(Type.AllocateContainer,
+              builder -> builder.setContainerRequest(request))
+              .getContainerResponse();
+      //TODO should be migrated to use the top level status structure.
+      if (response.getErrorCode() != ContainerResponseProto.Error.success) {
+        throw new IOException(response.hasErrorMessage() ?
+            response.getErrorMessage() : "Allocate container failed.");
+      }
+      return ContainerWithPipeline.fromProtobuf(
+          response.getContainerWithPipeline());
+    } else {
+      StorageContainerLocationProtocolProtos.ECContainerRequestProto.Builder requestBuilder =
+          StorageContainerLocationProtocolProtos.ECContainerRequestProto.newBuilder()
+              .setTraceID(TracingUtil.exportCurrentSpan())
+              .setOwner(owner);
+
+      ECReplicationConfig ecConfig = (ECReplicationConfig) replicationConfig;
+      requestBuilder.setEcReplicationConfig(ecConfig.toProto());
+
+      StorageContainerLocationProtocolProtos.ECContainerResponseProto response = submitRequest(
+          Type.AllocateContainer,
+          builder -> builder.setEcContainerRequest(requestBuilder.build()))
+          .getEcContainerResponse();
+
+      if (response.getErrorCode() != StorageContainerLocationProtocolProtos.ECContainerResponseProto.Error.success) {
+        throw new IOException(response.hasErrorMessage() ?
+            response.getErrorMessage() : "Allocate container failed.");
+      }
+      return ContainerWithPipeline.fromProtobuf(response.getContainerWithPipeline());
+    }
+  }
+
   @Override
   public ContainerInfo getContainer(long containerID) throws IOException {
     Preconditions.checkState(containerID >= 0,

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

@@ -85,6 +85,12 @@ message ScmContainerLocationRequest {
   optional GetContainersOnDecomNodeRequestProto getContainersOnDecomNodeRequest = 46;
   optional GetMetricsRequestProto getMetricsRequest = 47;
   optional ContainerBalancerStatusInfoRequestProto containerBalancerStatusInfoRequest = 48;
+  optional ECContainerRequestProto ecContainerRequest = 49; 
+  optional GetEcContainerRequestProto getEcContainerRequest = 50;
+  optional GetEcContainerWithPipelineRequestProto getEcContainerWithPipelineRequest = 51;
+  optional GetEcContainerReplicasRequestProto getEcContainerReplicasRequest = 52;
+  optional GetEcContainerWithPipelineBatchRequestProto getEcContainerWithPipelineBatchRequest = 53;
+  optional GetExistEcContainerWithPipelinesInBatchRequestProto getExistEcContainerWithPipelinesInBatchRequest = 54;
 }
 
 message ScmContainerLocationResponse {
@@ -141,6 +147,12 @@ message ScmContainerLocationResponse {
   optional GetContainersOnDecomNodeResponseProto getContainersOnDecomNodeResponse = 46;
   optional GetMetricsResponseProto getMetricsResponse = 47;
   optional ContainerBalancerStatusInfoResponseProto containerBalancerStatusInfoResponse = 48;
+  optional ECContainerResponseProto ecContainerResponse = 49; 
+  optional GetEcContainerResponseProto getEcContainerResponse = 50;
+  optional GetEcContainerWithPipelineResponseProto getEcContainerWithPipelineResponse = 51;
+  optional GetEcContainerReplicasResponseProto getEcContainerReplicasResponse = 52;
+  optional GetExistEcContainerWithPipelinesInBatchResponseProto getExistEcContainerWithPipelinesInBatchResponse = 53;
+  optional GetEcContainerWithPipelineBatchResponseProto getEcContainerWithPipelineBatchResponse = 54;
 
   enum Status {
     OK = 1;
@@ -209,6 +221,72 @@ message ContainerRequestProto {
   optional string traceID = 5;
 }
 
+message ECContainerRequestProto {
+  required ECReplicationConfig ecReplicationConfig = 1;
+  required string owner = 2;
+  optional string traceID = 3;
+}
+
+message ECContainerResponseProto {
+  enum Error {
+    success = 1;
+    errorContainerAlreadyExists = 2;
+    errorContainerMissing = 3;
+    scmNotLeader = 4;
+  }
+  required Error errorCode = 1;
+  required ContainerWithPipeline containerWithPipeline = 2;
+  optional string errorMessage = 3;
+}
+
+message GetEcContainerRequestProto {
+  required int64 containerID = 1;
+  optional string traceID = 2;
+
+}
+
+message GetEcContainerResponseProto {
+  required ContainerInfoProto containerInfo = 1;
+}
+
+message GetEcContainerWithPipelineRequestProto {
+  required int64 containerID = 1;
+  optional string traceID = 2;
+
+}
+
+message GetEcContainerWithPipelineResponseProto {
+  required ContainerWithPipeline containerWithPipeline = 1;
+}
+
+message GetEcContainerReplicasRequestProto {
+  required int64 containerID = 1;
+  optional string traceID = 2;
+}
+
+message GetEcContainerReplicasResponseProto {
+  repeated SCMContainerReplicaProto containerReplica = 1;
+}
+
+message GetEcContainerWithPipelineBatchRequestProto {
+  repeated int64 containerIDs = 1;
+  optional string traceID = 2;
+}
+
+message GetExistEcContainerWithPipelinesInBatchResponseProto {
+  repeated int64 containerIDs = 1;
+  optional string traceID = 2;
+}
+
+message GetExistEcContainerWithPipelinesInBatchRequestProto {
+  repeated int64 containerIDs = 1;
+  optional string traceID = 2;
+}
+
+message GetEcContainerWithPipelineBatchResponseProto {
+  repeated ContainerWithPipeline containerWithPipelines = 1;
+}
+
 /**
  * Reply from SCM indicating that the container.
  */

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

@@ -411,12 +411,21 @@ public ScmContainerLocationResponse processRequest(
     try {
       switch (request.getCmdType()) {
       case AllocateContainer:
-        return ScmContainerLocationResponse.newBuilder()
-            .setCmdType(request.getCmdType())
-            .setStatus(Status.OK)
-            .setContainerResponse(allocateContainer(
-                request.getContainerRequest(), request.getVersion()))
-            .build();
+        ScmContainerLocationResponse.Builder responseBuilder =
+            ScmContainerLocationResponse.newBuilder()
+                .setCmdType(request.getCmdType())
+                .setStatus(Status.OK);
+
+        if (request.hasContainerRequest()) {
+          responseBuilder.setContainerResponse(
+              allocateContainer(request.getContainerRequest(), request.getVersion()));
+        } else if (request.hasEcContainerRequest()) {
+          responseBuilder.setEcContainerResponse(
+              allocateEcContainer(request.getEcContainerRequest(), request.getVersion()));
+        } else {
+          throw new IOException("Invalid request");
+        }
+        return responseBuilder.build();
       case GetContainer:
         return ScmContainerLocationResponse.newBuilder()
             .setCmdType(request.getCmdType())
@@ -763,6 +772,17 @@ public ContainerResponseProto allocateContainer(ContainerRequestProto request,
 
   }
 
+  public StorageContainerLocationProtocolProtos.ECContainerResponseProto allocateEcContainer(
+      StorageContainerLocationProtocolProtos.ECContainerRequestProto request,
+      int clientVersion) throws IOException {
+    ContainerWithPipeline cp = impl
+        .allocateContainer(new ECReplicationConfig(request.getEcReplicationConfig()), request.getOwner());
+    return StorageContainerLocationProtocolProtos.ECContainerResponseProto.newBuilder()
+        .setContainerWithPipeline(cp.getProtobuf(clientVersion))
+        .setErrorCode(StorageContainerLocationProtocolProtos.ECContainerResponseProto.Error.success)
+        .build();
+  }
+
   public GetContainerResponseProto getContainer(
       GetContainerRequestProto request) throws IOException {
     ContainerInfo container = impl.getContainer(request.getContainerID());

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

@@ -261,6 +261,41 @@ public ContainerWithPipeline allocateContainer(HddsProtos.ReplicationType
     }
   }
 
+  @Override
+  public ContainerWithPipeline allocateContainer(ReplicationConfig replicationConfig, String owner) throws IOException {
+    Map<String, String> auditMap = Maps.newHashMap();
+    auditMap.put("replicationType", String.valueOf(replicationConfig.getReplicationType()));
+    if (replicationConfig.getReplicationType() == HddsProtos.ReplicationType.RATIS ||
+        replicationConfig.getReplicationType() == HddsProtos.ReplicationType.STAND_ALONE) {
+      auditMap.put("factor", String.valueOf(replicationConfig.getReplication()));
+    }
+    auditMap.put("owner", String.valueOf(owner));
+
+    try {
+      if (scm.getScmContext().isInSafeMode()) {
+        throw new SCMException("SafeModePrecheck failed for allocateContainer",
+            ResultCodes.SAFE_MODE_EXCEPTION);
+      }
+      getScm().checkAdminAccess(getRemoteUser(), false);
+      final ContainerInfo container = scm.getContainerManager()
+          .allocateContainer(
+              replicationConfig,
+              owner);
+      final Pipeline pipeline = scm.getPipelineManager()
+          .getPipeline(container.getPipelineID());
+      ContainerWithPipeline cp = new ContainerWithPipeline(container, pipeline);
+      AUDIT.logWriteSuccess(buildAuditMessageForSuccess(
+          SCMAction.ALLOCATE_CONTAINER, auditMap)
+      );
+      return cp;
+    } catch (Exception ex) {
+      AUDIT.logWriteFailure(buildAuditMessageForFailure(
+          SCMAction.ALLOCATE_CONTAINER, auditMap, ex)
+      );
+      throw ex;
+    }
+  }
+
   @Override
   public ContainerInfo getContainer(long containerID) throws IOException {
     Map<String, String> auditMap = Maps.newHashMap();

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

@@ -221,6 +221,33 @@ public ContainerWithPipeline createContainer(HddsProtos.ReplicationType type,
     }
   }
 
+  @Override
+  public ContainerWithPipeline createContainer(ReplicationConfig replicationConfig,
+                                                String owner) throws IOException {
+    XceiverClientSpi client = null;
+    XceiverClientManager clientManager = getXceiverClientManager();
+    try {
+      if (replicationConfig == null) {
+        replicationConfig = ReplicationConfig.fromProtoTypeAndFactor(replicationType, replicationFactor);
+      }
+      // allocate container on SCM.
+      ContainerWithPipeline containerWithPipeline =
+          storageContainerLocationClient.allocateContainer(replicationConfig,
+              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/ContainerReplicationOptions.java

@@ -0,0 +1,42 @@
+/*
+ * 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.cli.container;
+
+import org.apache.hadoop.ozone.shell.ReplicationOptions;
+import picocli.CommandLine;
+
+/** 
+ * CLI options for specifying container replication type and replication definition.
+ */
+
+public class ContainerReplicationOptions extends ReplicationOptions {
+
+  @CommandLine.Option(names = {"-t", "--type", "--replication-type"},
+      description = TYPE_DESCRIPTION)
+  @Override
+  public void setType(String type) {
+    super.setType(type);
+  }
+
+  @CommandLine.Option(names = {"--replication", "-r"},
+      description = REPLICATION_DESCRIPTION)
+  @Override
+  public void setReplication(String replication) {
+    super.setReplication(replication);
+  }
+}

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

@@ -19,9 +19,12 @@
 
 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.scm.cli.ScmSubcommand;
 import org.apache.hadoop.hdds.scm.client.ScmClient;
 import org.apache.hadoop.hdds.scm.container.common.helpers.ContainerWithPipeline;
+import picocli.CommandLine;
 import picocli.CommandLine.Command;
 import picocli.CommandLine.Option;
 
@@ -39,9 +42,13 @@ public class CreateSubcommand extends ScmSubcommand {
       names = { "-o", "--owner"})
   private String owner;
 
+  @CommandLine.Mixin
+  private ContainerReplicationOptions containerReplicationOptions;
+
   @Override
   public void execute(ScmClient scmClient) throws IOException {
-    ContainerWithPipeline container = scmClient.createContainer(owner);
+    ReplicationConfig replicationConfig = containerReplicationOptions.fromParamsOrConfig(new OzoneConfiguration());
+    ContainerWithPipeline container = scmClient.createContainer(replicationConfig, owner);
     System.out.printf("Container %s is created.%n",
         container.getContainerInfo().getContainerID());
   }