Merge remote-tracking branch 'origin/HDDS-5713' into master-new

Author: ChenSammi

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/HddsConfigKeys.java

@@ -332,6 +332,9 @@ public final class HddsConfigKeys {
   public static final String OZONE_SECURITY_RECONFIGURE_PROTOCOL_ACL =
       "ozone.security.reconfigure.protocol.acl";
 
+  public static final String HDDS_SECURITY_CLIENT_DATANODE_DISK_BALANCER_PROTOCOL_ACL =
+      "hdds.security.client.datanode.disk.balancer.protocol.acl";
+
   // Determines if the Container Chunk Manager will write user data to disk
   // Set to false only for specific performance tests
   public static final String HDDS_CONTAINER_PERSISTDATA =
@@ -398,6 +401,10 @@ public final class HddsConfigKeys {
   public static final String OZONE_DATANODE_IO_METRICS_PERCENTILES_INTERVALS_SECONDS_KEY =
       "ozone.volume.io.percentiles.intervals.seconds";
 
+  public static final String HDDS_DATANODE_DISK_BALANCER_ENABLED_KEY =
+      "hdds.datanode.disk.balancer.enabled";
+  public static final boolean HDDS_DATANODE_DISK_BALANCER_ENABLED_DEFAULT = false;
+
   public static final String HDDS_DATANODE_DNS_INTERFACE_KEY =
       "hdds.datanode.dns.interface";
   public static final String HDDS_DATANODE_DNS_NAMESERVER_KEY =

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/cli/ItemsFromStdin.java

@@ -32,7 +32,7 @@ public abstract class ItemsFromStdin implements Iterable<String> {
   protected static final String FORMAT_DESCRIPTION =
       ": one or more, separated by spaces. To read from stdin, specify '-' and supply one item per line.";
 
-  private List<String> items;
+  private List<String> items = new ArrayList<>();
 
   protected void setItems(List<String> arguments) {
     items = readItemsFromStdinIfNeeded(arguments);

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

@@ -199,6 +199,9 @@ public final class OzoneConsts {
    */
   public static final String OZONE_SCM_DATANODE_ID_FILE_DEFAULT = "datanode.id";
 
+  public static final String
+      OZONE_SCM_DATANODE_DISK_BALANCER_INFO_FILE_DEFAULT = "diskBalancer.info";
+
   /**
    * The ServiceListJSONServlet context attribute where OzoneManager
    * instance gets stored.

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

@@ -208,6 +208,15 @@
       The value should be between 0-1. Such as 0.1 which means 10% of volume space will be reserved.
     </description>
   </property>
+  <property>
+    <name>hdds.datanode.disk.balancer.enabled</name>
+    <value>false</value>
+    <tag>OZONE, DATANODE, DISKBALANCER</tag>
+    <description>If this property is set to true, then the Disk Balancer
+      feature is enabled on Datanodes, and users can use
+      this service. By default, this is disabled.
+    </description>
+  </property>
   <property>
     <name>hdds.datanode.volume.choosing.policy</name>
     <value>org.apache.hadoop.ozone.container.common.volume.CapacityVolumeChoosingPolicy</value>
@@ -371,7 +380,6 @@
       defined with postfix (ns,ms,s,m,h,d)</description>
   </property>
 
-
   <property>
     <name>hdds.prometheus.endpoint.enabled</name>
     <value>true</value>
@@ -784,6 +792,16 @@
       Then maximum 500000/(100/8) = 40000 blocks will be sent to each DN in every interval.
     </description>
   </property>
+  <property>
+    <name>ozone.scm.block.deletion.per.dn.distribution.factor</name>
+    <value>8</value>
+    <tag>OZONE, SCM</tag>
+    <description>
+      Factor with which number of delete blocks sent to each datanode in every interval.
+      If total number of DNs are 100 and hdds.scm.block.deletion.per-interval.max is 500000
+      Then maximum 500000/(100/8) = 40000 blocks will be sent to each DN in every interval.
+    </description>
+  </property>
   <property>
     <name>ozone.scm.block.size</name>
     <value>256MB</value>
@@ -2974,7 +2992,14 @@
       Comma separated list of users and groups allowed to access reconfigure protocol.
     </description>
   </property>
-
+  <property>
+    <name>hdds.security.client.datanode.disk.balancer.protocol.acl</name>
+    <value>*</value>
+    <tag>SECURITY</tag>
+    <description>
+      Comma separated list of users and groups allowed to access disk balancer protocol.
+    </description>
+  </property>
   <property>
     <name>hdds.datanode.http.auth.kerberos.principal</name>
     <value>HTTP/_HOST@REALM</value>

hadoop-hdds/config/src/main/java/org/apache/hadoop/hdds/conf/ConfigTag.java

@@ -53,5 +53,6 @@ public enum ConfigTag {
   TOKEN,
   UPGRADE,
   X509,
+  DISKBALANCER,
   CRYPTO_COMPLIANCE
 }

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/HddsDatanodeClientProtocolServer.java

@@ -33,11 +33,16 @@
 import org.apache.hadoop.hdds.conf.OzoneConfiguration;
 import org.apache.hadoop.hdds.conf.ReconfigurationHandler;
 import org.apache.hadoop.hdds.protocol.DatanodeDetails;
+import org.apache.hadoop.hdds.protocol.DiskBalancerProtocol;
+import org.apache.hadoop.hdds.protocol.proto.DiskBalancerProtocolProtos;
 import org.apache.hadoop.hdds.protocol.proto.ReconfigureProtocolProtos;
+import org.apache.hadoop.hdds.protocolPB.DiskBalancerProtocolPB;
+import org.apache.hadoop.hdds.protocolPB.DiskBalancerProtocolServerSideTranslatorPB;
 import org.apache.hadoop.hdds.protocolPB.ReconfigureProtocolDatanodePB;
 import org.apache.hadoop.hdds.protocolPB.ReconfigureProtocolServerSideTranslatorPB;
 import org.apache.hadoop.hdds.server.ServerUtils;
 import org.apache.hadoop.hdds.server.ServiceRuntimeInfoImpl;
+import org.apache.hadoop.hdds.utils.HddsServerUtil;
 import org.apache.hadoop.hdds.utils.VersionInfo;
 import org.apache.hadoop.ipc_.ProtobufRpcEngine;
 import org.apache.hadoop.ipc_.RPC;
@@ -56,12 +61,13 @@ public class HddsDatanodeClientProtocolServer extends ServiceRuntimeInfoImpl {
 
   protected HddsDatanodeClientProtocolServer(
       DatanodeDetails datanodeDetails, OzoneConfiguration conf,
-      VersionInfo versionInfo, ReconfigurationHandler reconfigurationHandler
+      VersionInfo versionInfo, ReconfigurationHandler reconfigurationHandler,
+      DiskBalancerProtocol diskBalancerProtocol
   ) throws IOException {
     super(versionInfo);
     this.conf = conf;
 
-    rpcServer = getRpcServer(conf, reconfigurationHandler);
+    rpcServer = getRpcServer(conf, reconfigurationHandler, diskBalancerProtocol);
     clientRpcAddress = ServerUtils.updateRPCListenAddress(this.conf,
         HDDS_DATANODE_CLIENT_ADDRESS_KEY,
         HddsUtils.getDatanodeRpcAddress(conf), rpcServer);
@@ -96,12 +102,15 @@ public void join() throws InterruptedException {
    * running then returns the same.
    */
   private RPC.Server getRpcServer(OzoneConfiguration configuration,
-      ReconfigurationHandler reconfigurationHandler)
+      ReconfigurationHandler reconfigurationHandler,
+      DiskBalancerProtocol diskBalancerProtocol)
       throws IOException {
     InetSocketAddress rpcAddress = HddsUtils.getDatanodeRpcAddress(conf);
-    // Add reconfigureProtocolService.
+    // Set protocol engines for all protocols before creating the server.
     RPC.setProtocolEngine(
         configuration, ReconfigureProtocolDatanodePB.class, ProtobufRpcEngine.class);
+    RPC.setProtocolEngine(
+        configuration, DiskBalancerProtocolPB.class, ProtobufRpcEngine.class);
 
     final int handlerCount = conf.getInt(HDDS_DATANODE_HANDLER_COUNT_KEY,
         HDDS_DATANODE_HANDLER_COUNT_DEFAULT);
@@ -113,8 +122,17 @@ private RPC.Server getRpcServer(OzoneConfiguration configuration,
         .ReconfigureProtocolService.newReflectiveBlockingService(
             reconfigureServerProtocol);
 
-    return startRpcServer(configuration, rpcAddress,
+    RPC.Server server = startRpcServer(configuration, rpcAddress,
         ReconfigureProtocolDatanodePB.class, reconfigureService, handlerCount, readThreads);
+    if (diskBalancerProtocol != null) {
+      DiskBalancerProtocolServerSideTranslatorPB diskBalancerTranslator =
+          new DiskBalancerProtocolServerSideTranslatorPB(diskBalancerProtocol);
+      BlockingService diskBalancerService = DiskBalancerProtocolProtos
+          .DiskBalancerProtocolService.newReflectiveBlockingService(diskBalancerTranslator);
+      HddsServerUtil.addPBProtocol(configuration, DiskBalancerProtocolPB.class,
+          diskBalancerService, server);
+    }
+    return server;
   }
 
   /**

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/HddsDatanodeService.java

@@ -66,6 +66,7 @@
 import org.apache.hadoop.hdds.conf.ReconfigurationHandler;
 import org.apache.hadoop.hdds.protocol.DatanodeDetails;
 import org.apache.hadoop.hdds.protocol.DatanodeID;
+import org.apache.hadoop.hdds.protocol.DiskBalancerProtocol;
 import org.apache.hadoop.hdds.protocol.SecretKeyProtocol;
 import org.apache.hadoop.hdds.protocolPB.SCMSecurityProtocolClientSideTranslatorPB;
 import org.apache.hadoop.hdds.security.SecurityConfig;
@@ -91,6 +92,7 @@
 import org.apache.hadoop.ozone.container.common.volume.HddsVolume;
 import org.apache.hadoop.ozone.container.common.volume.MutableVolumeSet;
 import org.apache.hadoop.ozone.container.common.volume.StorageVolume;
+import org.apache.hadoop.ozone.container.diskbalancer.DiskBalancerProtocolServer;
 import org.apache.hadoop.ozone.util.OzoneNetUtils;
 import org.apache.hadoop.ozone.util.ShutdownHookManager;
 import org.apache.hadoop.security.SecurityUtil;
@@ -342,9 +344,12 @@ public String getNamespace() {
         LOG.error("HttpServer failed to start.", ex);
       }
 
+      DiskBalancerProtocol diskBalancerProtocol =
+          new DiskBalancerProtocolServer(datanodeStateMachine,
+              this::checkAdminPrivilege);
       clientProtocolServer = new HddsDatanodeClientProtocolServer(
           datanodeDetails, conf, HddsVersionInfo.HDDS_VERSION_INFO,
-          reconfigurationHandler);
+          reconfigurationHandler, diskBalancerProtocol);
 
       int clientRpcport = clientProtocolServer.getClientRpcAddress().getPort();
       serviceRuntimeInfo.setClientRpcPort(String.valueOf(clientRpcport));

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/HddsPolicyProvider.java

@@ -17,13 +17,15 @@
 
 package org.apache.hadoop.ozone;
 
+import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_SECURITY_CLIENT_DATANODE_DISK_BALANCER_PROTOCOL_ACL;
 import static org.apache.hadoop.hdds.HddsConfigKeys.OZONE_SECURITY_RECONFIGURE_PROTOCOL_ACL;
 
-import java.util.Collections;
+import java.util.Arrays;
 import java.util.List;
 import java.util.function.Supplier;
 import org.apache.hadoop.hdds.annotation.InterfaceAudience.Private;
 import org.apache.hadoop.hdds.annotation.InterfaceStability.Unstable;
+import org.apache.hadoop.hdds.protocol.DiskBalancerProtocol;
 import org.apache.hadoop.hdds.protocol.ReconfigureProtocol;
 import org.apache.hadoop.security.authorize.PolicyProvider;
 import org.apache.hadoop.security.authorize.Service;
@@ -40,10 +42,13 @@ public final class HddsPolicyProvider extends PolicyProvider {
       MemoizedSupplier.valueOf(HddsPolicyProvider::new);
 
   private static final List<Service> DN_SERVICES =
-      Collections.singletonList(
+      Arrays.asList(
           new Service(
               OZONE_SECURITY_RECONFIGURE_PROTOCOL_ACL,
-              ReconfigureProtocol.class)
+              ReconfigureProtocol.class),
+          new Service(
+              HDDS_SECURITY_CLIENT_DATANODE_DISK_BALANCER_PROTOCOL_ACL,
+              DiskBalancerProtocol.class)
       );
 
   private HddsPolicyProvider() {

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/impl/ContainerSet.java

@@ -235,6 +235,29 @@ private void updateContainerIdTable(long containerId, ContainerData containerDat
     }
   }
 
+  /**
+   * Update Container to container map.
+   * @param container container to be added
+   * @return If container is added to containerMap returns true, otherwise
+   * false
+   */
+  public Container updateContainer(Container<?> container) throws
+      StorageContainerException {
+    Objects.requireNonNull(container, "container cannot be null");
+
+    long containerId = container.getContainerData().getContainerID();
+    if (!containerMap.containsKey(containerId)) {
+      LOG.error("Container doesn't exists with container Id {}", containerId);
+      throw new StorageContainerException("Container doesn't exist with " +
+          "container Id " + containerId,
+          ContainerProtos.Result.CONTAINER_NOT_FOUND);
+    } else {
+      LOG.debug("Container with container Id {} is updated to containerMap",
+          containerId);
+      return containerMap.put(containerId, container);
+    }
+  }
+
   /**
    * Returns the Container with specified containerId.
    * @param containerId ID of the container to get

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/interfaces/Container.java

@@ -21,6 +21,7 @@
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
+import java.nio.file.Path;
 import java.time.Instant;
 import java.util.Map;
 import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
@@ -195,6 +196,11 @@ ContainerReplicaProto getContainerReport()
   DataScanResult scanData(DataTransferThrottler throttler, Canceler canceler)
       throws InterruptedException;
 
+  /**
+   * Copy all the data of the container to the destination path.
+   */
+  void copyContainerDirectory(Path destPath) throws IOException;
+
   /** Acquire read lock. */
   void readLock();