HDDS-10000. UD-635. Added datacenter info to datanodes, buckets and containers

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

@@ -18,17 +18,22 @@
 
 package org.apache.hadoop.hdds.scm;
 
+import java.util.Collections;
+import java.util.Set;
+
 /**
  * The information of the request of pipeline.
  */
 public final class PipelineRequestInformation {
   private final long size;
+  private final Set<String> datacenters;
 
   /**
    * Builder for PipelineRequestInformation.
    */
   public static class Builder {
     private long size;
+    private Set<String> datacenters = Collections.emptySet();
 
     public static Builder getBuilder() {
       return new Builder();
@@ -44,16 +49,31 @@ public Builder setSize(long sz) {
       return this;
     }
 
+    /**
+     * sets the datacenters.
+     * @param datacenters request datacenters
+     * @return Builder for PipelineRequestInformation
+     */
+    public Builder setDatacenters(Set<String> datacenters) {
+      this.datacenters = datacenters;
+      return this;
+    }
+
     public PipelineRequestInformation build() {
-      return new PipelineRequestInformation(size);
+      return new PipelineRequestInformation(size, datacenters);
     }
   }
 
-  private PipelineRequestInformation(long size) {
+  private PipelineRequestInformation(long size, Set<String> datacenters) {
     this.size = size;
+    this.datacenters = datacenters;
   }
 
   public long getSize() {
     return size;
   }
+
+  public Set<String> getDatacenters() {
+    return datacenters;
+  }
 }

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

@@ -617,6 +617,13 @@ public final class ScmConfigKeys {
       "ozone.scm.ha.dbtransactionbuffer.flush.interval";
   public static final long
       OZONE_SCM_HA_DBTRANSACTIONBUFFER_FLUSH_INTERVAL_DEFAULT = 600 * 1000L;
+
+  public static final String OZONE_SCM_DC_DATANODE_MAPPING_KEY =
+      "ozone.scm.dc.datanode.mapping";
+
+  public static final String OZONE_SCM_DC_DATANODE_MAPPING_DEFAULT =
+      "";
+
   /**
    * Never constructed.
    */

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

@@ -17,24 +17,27 @@
  */
 package org.apache.hadoop.hdds.scm.container;
 
-import java.time.Clock;
-import java.time.Instant;
-import java.util.Comparator;
-
+import com.fasterxml.jackson.annotation.JsonIgnore;
 import com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.commons.lang3.builder.EqualsBuilder;
+import org.apache.commons.lang3.builder.HashCodeBuilder;
 import org.apache.hadoop.hdds.client.ECReplicationConfig;
 import org.apache.hadoop.hdds.client.ReplicationConfig;
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
 import org.apache.hadoop.hdds.scm.pipeline.PipelineID;
 import org.apache.hadoop.hdds.utils.db.Codec;
 import org.apache.hadoop.hdds.utils.db.DelegatedCodec;
 import org.apache.hadoop.hdds.utils.db.Proto2Codec;
+import org.apache.ratis.util.Preconditions;
+
+import java.time.Clock;
+import java.time.Instant;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashSet;
+import java.util.Set;
 
-import com.fasterxml.jackson.annotation.JsonIgnore;
 import static java.lang.Math.max;
-import org.apache.commons.lang3.builder.EqualsBuilder;
-import org.apache.commons.lang3.builder.HashCodeBuilder;
-import org.apache.ratis.util.Preconditions;
 
 /**
  * Class wraps ozone container info.
@@ -89,6 +92,7 @@ public static Codec<ContainerInfo> getCodec() {
   // The sequenceId of a close container cannot change, and all the
   // container replica should have the same sequenceId.
   private long sequenceId;
+  private final Set<String> datacenters;
 
   @SuppressWarnings("parameternumber")
   private ContainerInfo(
@@ -102,7 +106,8 @@ private ContainerInfo(
       long deleteTransactionId,
       long sequenceId,
       ReplicationConfig repConfig,
-      Clock clock) {
+      Clock clock,
+      Set<String> datacenters) {
     this.containerID = ContainerID.valueOf(containerID);
     this.pipelineID = pipelineID;
     this.usedBytes = usedBytes;
@@ -115,6 +120,7 @@ private ContainerInfo(
     this.sequenceId = sequenceId;
     this.replicationConfig = repConfig;
     this.clock = clock;
+    this.datacenters = datacenters;
   }
 
   public static ContainerInfo fromProtobuf(HddsProtos.ContainerInfoProto info) {
@@ -131,13 +137,14 @@ public static ContainerInfo fromProtobuf(HddsProtos.ContainerInfoProto info) {
         .setDeleteTransactionId(info.getDeleteTransactionId())
         .setReplicationConfig(config)
         .setSequenceId(info.getSequenceId())
+        .setDatacenters(new HashSet<>(info.getDatacentersList()))
         .build();
 
     if (info.hasPipelineID()) {
       builder.setPipelineID(PipelineID.getFromProtobuf(info.getPipelineID()));
     }
-    return builder.build();
 
+    return builder.build();
   }
 
   /**
@@ -229,6 +236,10 @@ public long getSequenceId() {
     return sequenceId;
   }
 
+  public Set<String> getDatacenters() {
+    return datacenters;
+  }
+
   public void updateDeleteTransactionId(long transactionId) {
     deleteTransactionId = max(transactionId, deleteTransactionId);
   }
@@ -267,7 +278,8 @@ public HddsProtos.ContainerInfoProto getProtobuf() {
         .setDeleteTransactionId(getDeleteTransactionId())
         .setOwner(getOwner())
         .setSequenceId(getSequenceId())
-        .setReplicationType(getReplicationType());
+        .setReplicationType(getReplicationType())
+        .addAllDatacenters(getDatacenters());
 
     if (replicationConfig instanceof ECReplicationConfig) {
       builder.setEcReplicationConfig(((ECReplicationConfig) replicationConfig)
@@ -282,6 +294,7 @@ public HddsProtos.ContainerInfoProto getProtobuf() {
     if (getPipelineID() != null) {
       builder.setPipelineID(getPipelineID().getProtobuf());
     }
+
     return builder.build();
   }
 
@@ -383,6 +396,7 @@ public static class Builder {
     private long sequenceId;
     private PipelineID pipelineID;
     private ReplicationConfig replicationConfig;
+    private Set<String> datacenters = Collections.emptySet();
 
     public Builder setPipelineID(PipelineID pipelineId) {
       this.pipelineID = pipelineId;
@@ -444,10 +458,15 @@ public Builder setClock(Clock clock) {
       return this;
     }
 
+    public Builder setDatacenters(Set<String> datacenters) {
+      this.datacenters = datacenters;
+      return this;
+    }
+
     public ContainerInfo build() {
       return new ContainerInfo(containerID, state, pipelineID,
           used, keys, stateEnterTime, owner, deleteTransactionId,
-          sequenceId, replicationConfig, clock);
+          sequenceId, replicationConfig, clock, datacenters);
     }
   }
 

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

@@ -139,6 +139,7 @@ public enum ResultCodes {
     CA_ROTATION_IN_PROGRESS,
     CA_ROTATION_IN_POST_PROGRESS,
     CONTAINER_ALREADY_CLOSED,
-    CONTAINER_ALREADY_CLOSING
+    CONTAINER_ALREADY_CLOSING,
+    FAILED_TO_FIND_ENOUGH_NODES_WITHIN_DATACENTER,
   }
 }

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

@@ -24,6 +24,7 @@
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.Iterator;
 import java.util.LinkedHashMap;
 import java.util.LinkedList;
@@ -93,6 +94,8 @@ public static Codec<Pipeline> getCodec() {
 
   private final Instant stateEnterTime;
 
+  private final Set<String> datacenters;
+
   /**
    * The immutable properties of pipeline object is used in
    * ContainerStateManager#getMatchingContainerByPipeline to take a lock on
@@ -118,6 +121,7 @@ private Pipeline(Builder b) {
     replicaIndexes = b.replicaIndexes != null ? ImmutableMap.copyOf(b.replicaIndexes) : ImmutableMap.of();
     creationTimestamp = b.creationTimestamp != null ? b.creationTimestamp : Instant.now();
     stateEnterTime = Instant.now();
+    datacenters = b.datacenters;
   }
 
   /**
@@ -160,6 +164,10 @@ public Instant getStateEnterTime() {
     return stateEnterTime;
   }
 
+  public Set<String> getDatacenters() {
+    return datacenters;
+  }
+
   /**
    * Return the suggested leaderId which has a high priority among DNs of the
    * pipeline.
@@ -379,7 +387,8 @@ public HddsProtos.Pipeline getProtobufMessage(int clientVersion)
         .setLeaderID(leaderId != null ? leaderId.toString() : "")
         .setCreationTimeStamp(creationTimestamp.toEpochMilli())
         .addAllMembers(members)
-        .addAllMemberReplicaIndexes(memberReplicaIndexes);
+        .addAllMemberReplicaIndexes(memberReplicaIndexes)
+        .addAllDatacenters(datacenters);
 
     if (replicationConfig instanceof ECReplicationConfig) {
       builder.setEcReplicationConfig(((ECReplicationConfig) replicationConfig)
@@ -479,7 +488,8 @@ public static Builder toBuilder(HddsProtos.Pipeline pipeline)
         .setLeaderId(leaderId)
         .setSuggestedLeaderId(suggestedLeaderId)
         .setNodeOrder(pipeline.getMemberOrdersList())
-        .setCreateTimestamp(pipeline.getCreationTimeStamp());
+        .setCreateTimestamp(pipeline.getCreationTimeStamp())
+        .setDatacenters(new HashSet<>(pipeline.getDatacentersList()));
   }
 
   public static Pipeline getFromProtobuf(HddsProtos.Pipeline pipeline)
@@ -555,6 +565,7 @@ public static class Builder {
     private Instant creationTimestamp = null;
     private UUID suggestedLeaderId = null;
     private Map<DatanodeDetails, Integer> replicaIndexes;
+    private Set<String> datacenters = new HashSet<>();
 
     public Builder() { }
 
@@ -567,6 +578,7 @@ public Builder(Pipeline pipeline) {
       this.leaderId = pipeline.getLeaderId();
       this.creationTimestamp = pipeline.getCreationTimestamp();
       this.suggestedLeaderId = pipeline.getSuggestedLeaderId();
+      this.datacenters = pipeline.getDatacenters();
       if (nodeStatus != null) {
         replicaIndexes = new HashMap<>();
         for (DatanodeDetails dn : nodeStatus.keySet()) {
@@ -635,6 +647,10 @@ public Builder setSuggestedLeaderId(UUID uuid) {
       return this;
     }
 
+    public Builder setDatacenters(Set<String> datacenters) {
+      this.datacenters = datacenters;
+      return this;
+    }
 
     public Builder setReplicaIndexes(Map<DatanodeDetails, Integer> indexes) {
       this.replicaIndexes = indexes;

hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/OzoneConsts.java

@@ -407,6 +407,9 @@ private OzoneConsts() {
   public static final String GDPR_SECRET = "secret";
   public static final String GDPR_ALGORITHM = "algorithm";
 
+  // DATACENTERS
+  public static final String DATACENTERS = "datacenters";
+
   /**
    * Block key name as illegal characters
    *

hadoop-hdds/common/src/main/resources/ozone-default.xml

@@ -4349,4 +4349,15 @@
       A key of the cache entry is a pair of bucket and the requested key path.
     </description>
   </property>
+
+  <property>
+    <name>ozone.scm.dc.datanode.mapping</name>
+    <value></value>
+    <tag>SCM</tag>
+    <description>
+      This setting maps each DataNode to its data center using a comma-separated list of
+      hostname:ratis_port=*dc_name* pairs. It enables Ozone to associate DataNodes with specific data centers,
+      optimizing storage management.
+    </description>
+  </property>
 </configuration>

hadoop-hdds/common/src/test/java/org/apache/hadoop/hdds/scm/container/TestContainerInfo.java

@@ -26,6 +26,7 @@
 
 import java.time.Duration;
 import java.time.Instant;
+import java.util.Collections;
 
 import static org.apache.hadoop.hdds.protocol.proto.HddsProtos.LifeCycleState.CLOSING;
 import static org.apache.hadoop.hdds.protocol.proto.HddsProtos.LifeCycleState.OPEN;
@@ -116,6 +117,7 @@ public static ContainerInfo.Builder newBuilderForTest() {
         .setContainerID(1234)
         .setPipelineID(PipelineID.randomId())
         .setState(OPEN)
-        .setOwner("scm");
+        .setOwner("scm")
+        .setDatacenters(Collections.emptySet());
   }
 }

hadoop-hdds/common/src/test/java/org/apache/hadoop/hdds/scm/pipeline/MockPipeline.java

@@ -19,6 +19,7 @@
 
 import java.io.IOException;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -72,6 +73,7 @@ public static Pipeline createPipeline(Iterable<DatanodeDetails> ids) {
         .setReplicationConfig(
             StandaloneReplicationConfig.getInstance(ReplicationFactor.ONE))
         .setNodes(dns)
+        .setDatacenters(Collections.emptySet())
         .build();
   }
 

hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/scm/protocol/ScmBlockLocationProtocol.java

@@ -87,7 +87,7 @@ default List<AllocatedBlock> allocateBlock(long size, int numBlocks,
        ReplicationConfig replicationConfig, String owner,
        ExcludeList excludeList) throws IOException {
     return allocateBlock(size, numBlocks, replicationConfig, owner,
-        excludeList, null);
+        excludeList, null, null);
   }
 
   /**
@@ -109,7 +109,7 @@ default List<AllocatedBlock> allocateBlock(long size, int numBlocks,
    */
   List<AllocatedBlock> allocateBlock(long size, int numBlocks,
       ReplicationConfig replicationConfig, String owner,
-      ExcludeList excludeList, String clientMachine) throws IOException;
+      ExcludeList excludeList, String clientMachine, String datacenters) throws IOException;
 
   /**
    * Delete blocks for a set of object keys.

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

@@ -153,7 +153,7 @@ public List<AllocatedBlock> allocateBlock(
       long size, int num,
       ReplicationConfig replicationConfig,
       String owner, ExcludeList excludeList,
-      String clientMachine
+      String clientMachine, String datacenters
   ) throws IOException {
     Preconditions.checkArgument(size > 0, "block size must be greater than 0");
 
@@ -169,6 +169,10 @@ public List<AllocatedBlock> allocateBlock(
       requestBuilder.setClient(clientMachine);
     }
 
+    if (StringUtils.isNotEmpty(datacenters)) {
+      requestBuilder.setDatacenters(datacenters);
+    }
+
     switch (replicationConfig.getReplicationType()) {
     case STAND_ALONE:
       requestBuilder.setFactor(