HDDS-4926. Support start/stop for container balancer via command line

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

@@ -305,6 +305,24 @@ Map<String, Pair<Boolean, String>> getSafeModeRuleStatuses()
    */
   boolean getReplicationManagerStatus() throws IOException;
 
+  /**
+   * Start ContainerBalancer.
+   */
+  boolean startContainerBalancer(double threshold, int idleiterations,
+      int maxDatanodesToBalance, long maxSizeToMove) throws IOException;
+
+  /**
+   * Stop ContainerBalancer.
+   */
+  void stopContainerBalancer() throws IOException;
+
+  /**
+   * Returns ContainerBalancer status.
+   *
+   * @return True if ContainerBalancer is running, false otherwise.
+   */
+  boolean getContainerBalancerStatus() throws IOException;
+
   /**
    * returns the list of ratis peer roles. Currently only include peer address.
    */

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

@@ -286,6 +286,24 @@ Map<String, Pair<Boolean, String>> getSafeModeRuleStatuses()
    */
   boolean getReplicationManagerStatus() throws IOException;
 
+  /**
+   * Start ContainerBalancer.
+   */
+  boolean startContainerBalancer(double threshold, int idleiterations,
+      int maxDatanodesToBalance, long maxSizeToMove) throws IOException;
+
+  /**
+   * Stop ContainerBalancer.
+   */
+  void stopContainerBalancer() throws IOException;
+
+  /**
+   * Returns ContainerBalancer status.
+   *
+   * @return True if ContainerBalancer is running, false otherwise.
+   */
+  boolean getContainerBalancerStatus() throws IOException;
+
   /**
    * Get Datanode usage information by ip or uuid.
    *

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

@@ -44,6 +44,9 @@ public enum SCMAction implements AuditAction {
   START_REPLICATION_MANAGER,
   STOP_REPLICATION_MANAGER,
   GET_REPLICATION_MANAGER_STATUS,
+  START_CONTAINER_BALANCER,
+  STOP_CONTAINER_BALANCER,
+  GET_CONTAINER_BALANCER_STATUS,
   GET_CONTAINER_WITH_PIPELINE_BATCH,
   ADD_SCM;
 

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

@@ -57,6 +57,8 @@
 import org.apache.hadoop.hdds.protocol.proto.StorageContainerLocationProtocolProtos.RecommissionNodesResponseProto;
 import org.apache.hadoop.hdds.protocol.proto.StorageContainerLocationProtocolProtos.ReplicationManagerStatusRequestProto;
 import org.apache.hadoop.hdds.protocol.proto.StorageContainerLocationProtocolProtos.ReplicationManagerStatusResponseProto;
+import org.apache.hadoop.hdds.protocol.proto.StorageContainerLocationProtocolProtos.ContainerBalancerStatusRequestProto;
+import org.apache.hadoop.hdds.protocol.proto.StorageContainerLocationProtocolProtos.ContainerBalancerStatusResponseProto;
 import org.apache.hadoop.hdds.protocol.proto.StorageContainerLocationProtocolProtos.SCMCloseContainerRequestProto;
 import org.apache.hadoop.hdds.protocol.proto.StorageContainerLocationProtocolProtos.SCMDeleteContainerRequestProto;
 import org.apache.hadoop.hdds.protocol.proto.StorageContainerLocationProtocolProtos.SCMListContainerRequestProto;
@@ -69,6 +71,9 @@
 import org.apache.hadoop.hdds.protocol.proto.StorageContainerLocationProtocolProtos.StartMaintenanceNodesResponseProto;
 import org.apache.hadoop.hdds.protocol.proto.StorageContainerLocationProtocolProtos.StartReplicationManagerRequestProto;
 import org.apache.hadoop.hdds.protocol.proto.StorageContainerLocationProtocolProtos.StopReplicationManagerRequestProto;
+import org.apache.hadoop.hdds.protocol.proto.StorageContainerLocationProtocolProtos.StartContainerBalancerRequestProto;
+import org.apache.hadoop.hdds.protocol.proto.StorageContainerLocationProtocolProtos.StartContainerBalancerResponseProto;
+import org.apache.hadoop.hdds.protocol.proto.StorageContainerLocationProtocolProtos.StopContainerBalancerRequestProto;
 import org.apache.hadoop.hdds.protocol.proto.StorageContainerLocationProtocolProtos.Type;
 import org.apache.hadoop.hdds.scm.DatanodeAdminError;
 import org.apache.hadoop.hdds.scm.ScmInfo;
@@ -712,6 +717,45 @@ public boolean getReplicationManagerStatus() throws IOException {
 
   }
 
+  @Override
+  public boolean startContainerBalancer(double threshold, int idleiterations,
+         int maxDatanodesToBalance, long maxSizeToMove) throws IOException {
+    StartContainerBalancerRequestProto request =
+        StartContainerBalancerRequestProto.newBuilder()
+            .setIdleiterations(idleiterations)
+            .setMaxDatanodesToBalance(maxDatanodesToBalance)
+            .setMaxSizeToMove(maxSizeToMove)
+            .setThreshold(threshold).build();
+    StartContainerBalancerResponseProto response =
+        submitRequest(Type.StartContainerBalancer,
+            builder -> builder.setStartContainerBalancerRequest(request))
+            .getStartContainerBalancerResponse();
+    return response.getStart();
+  }
+
+  @Override
+  public void stopContainerBalancer() throws IOException {
+
+    StopContainerBalancerRequestProto request =
+        StopContainerBalancerRequestProto.getDefaultInstance();
+    submitRequest(Type.StopContainerBalancer,
+        builder -> builder.setStopContainerBalancerRequest(request));
+
+  }
+
+  @Override
+  public boolean getContainerBalancerStatus() throws IOException {
+
+    ContainerBalancerStatusRequestProto request =
+        ContainerBalancerStatusRequestProto.getDefaultInstance();
+    ContainerBalancerStatusResponseProto response =
+        submitRequest(Type.GetContainerBalancerStatus,
+            builder -> builder.setContainerBalancerStatusRequest(request))
+            .getContainerBalancerStatusResponse();
+    return response.getIsRunning();
+
+  }
+
   /**
    * Builds request for datanode usage information and receives response.
    *

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

@@ -69,6 +69,9 @@ message ScmContainerLocationRequest {
   optional DatanodeUsageInfoRequestProto DatanodeUsageInfoRequest = 30;
   optional GetExistContainerWithPipelinesInBatchRequestProto getExistContainerWithPipelinesInBatchRequest = 31;
   optional GetContainerTokenRequestProto containerTokenRequest = 32;
+  optional StartContainerBalancerRequestProto startContainerBalancerRequest = 33;
+  optional StopContainerBalancerRequestProto stopContainerBalancerRequest = 34;
+  optional ContainerBalancerStatusRequestProto containerBalancerStatusRequest = 35;
 }
 
 message ScmContainerLocationResponse {
@@ -109,6 +112,9 @@ message ScmContainerLocationResponse {
   optional DatanodeUsageInfoResponseProto DatanodeUsageInfoResponse = 30;
   optional GetExistContainerWithPipelinesInBatchResponseProto getExistContainerWithPipelinesInBatchResponse = 31;
   optional GetContainerTokenResponseProto containerTokenResponse = 32;
+  optional StartContainerBalancerResponseProto startContainerBalancerResponse = 33;
+  optional StopContainerBalancerResponseProto stopContainerBalancerResponse = 34;
+  optional ContainerBalancerStatusResponseProto containerBalancerStatusResponse = 35;
 
   enum Status {
     OK = 1;
@@ -147,6 +153,9 @@ enum Type {
   DatanodeUsageInfo = 25;
   GetExistContainerWithPipelinesInBatch = 26;
   GetContainerToken = 27;
+  StartContainerBalancer = 28;
+  StopContainerBalancer = 29;
+  GetContainerBalancerStatus = 30;
 }
 
 /**
@@ -445,6 +454,33 @@ message GetContainerTokenResponseProto {
   required TokenProto token = 1;
 }
 
+message StartContainerBalancerRequestProto {
+  optional string traceID = 1;
+  required double threshold = 2;
+  required int32 idleiterations = 3;
+  required int32 maxDatanodesToBalance = 4;
+  required int64 maxSizeToMove = 5;
+}
+
+message StartContainerBalancerResponseProto {
+  required bool start = 1;
+}
+
+message StopContainerBalancerRequestProto {
+  optional string traceID = 1;
+}
+
+message StopContainerBalancerResponseProto {
+}
+
+message ContainerBalancerStatusRequestProto {
+  optional string traceID = 1;
+}
+
+message ContainerBalancerStatusResponseProto {
+  required bool isRunning = 1;
+}
+
 /**
  * Protocol used from an HDFS node to StorageContainerManager.  See the request
  * and response messages for details of the RPC calls.

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

@@ -52,6 +52,7 @@ public class ContainerBalancer {
   private double threshold;
   private int maxDatanodesToBalance;
   private long maxSizeToMove;
+  private int idleIteration;
   private List<DatanodeUsageInfo> unBalancedNodes;
   private List<DatanodeUsageInfo> overUtilizedNodes;
   private List<DatanodeUsageInfo> underUtilizedNodes;
@@ -63,6 +64,7 @@ public class ContainerBalancer {
   private long clusterRemaining;
   private double clusterAvgUtilisation;
   private final AtomicBoolean balancerRunning = new AtomicBoolean(false);
+  private Thread currentBalancingThread;
 
   /**
    * Constructs ContainerBalancer with the specified arguments. Initializes
@@ -97,7 +99,6 @@ public ContainerBalancer(
     this.unBalancedNodes = new ArrayList<>();
     this.withinThresholdUtilizedNodes = new ArrayList<>();
   }
-
   /**
    * Starts ContainerBalancer. Current implementation is incomplete.
    *
@@ -111,27 +112,42 @@ public boolean start(ContainerBalancerConfiguration balancerConfiguration) {
 
     ozoneConfiguration = new OzoneConfiguration();
     this.config = balancerConfiguration;
+    this.idleIteration = config.getIdleIteration();
     this.threshold = config.getThreshold();
     this.maxDatanodesToBalance = config.getMaxDatanodesToBalance();
     this.maxSizeToMove = config.getMaxSizeToMove();
     this.unBalancedNodes = new ArrayList<>();
-
     LOG.info("Starting Container Balancer...{}", this);
-    balance();
+    //we should start a new balancer thread async
+    //and response to cli as soon as possible
+
+
+    //this is a temporary implementation
+    //modify this later
+    currentBalancingThread = new Thread(() -> balance());
+    currentBalancingThread.start();
+    ////////////////////////
+
+
     return true;
   }
 
   /**
    * Balances the cluster.
    */
   private void balance() {
-    initializeIteration();
-
-    // unBalancedNodes is not cleared since the next iteration uses this
-    // iteration's unBalancedNodes to find out how many nodes were balanced
-    overUtilizedNodes.clear();
-    underUtilizedNodes.clear();
-    withinThresholdUtilizedNodes.clear();
+    for (int i = 0; i < idleIteration; i++) {
+      if (!initializeIteration()) {
+        //balancer should be stopped immediately
+        break;
+      }
+      // unBalancedNodes is not cleared since the next iteration uses this
+      // iteration's unBalancedNodes to find out how many nodes were balanced
+      overUtilizedNodes.clear();
+      underUtilizedNodes.clear();
+      withinThresholdUtilizedNodes.clear();
+    }
+    balancerRunning.compareAndSet(true, false);
   }
 
   /**
@@ -152,7 +168,6 @@ private boolean initializeIteration() {
     if (datanodeUsageInfos.isEmpty()) {
       LOG.info("Container Balancer could not retrieve nodes from Node " +
           "Manager.");
-      stop();
       return false;
     }
 
@@ -221,15 +236,21 @@ private boolean initializeIteration() {
         maxDatanodesToBalance) {
       LOG.info("Approaching Max Datanodes To Balance limit in Container " +
           "Balancer. Stopping Balancer.");
-      stop();
       return false;
     } else {
       unBalancedNodes.addAll(overUtilizedNodes);
       unBalancedNodes.addAll(underUtilizedNodes);
 
+      //for now, we just sleep to simulate the execution of balancer
+      //this if for acceptance test now. modify this later when balancer
+      //if fully completed
+      try {
+        Thread.sleep(50);
+      } catch (InterruptedException e) {}
+      /////////////////////////////
+
       if (unBalancedNodes.isEmpty()) {
         LOG.info("Did not find any unbalanced Datanodes.");
-        stop();
         return false;
       } else {
         LOG.info("Container Balancer has identified Datanodes that need to be" +
@@ -323,8 +344,22 @@ public static double calculateUtilization(
    * Stops ContainerBalancer.
    */
   public void stop() {
-    balancerRunning.set(false);
-    LOG.info("Container Balancer stopped.");
+    //we should stop the balancer thread gracefully
+    if(!balancerRunning.get()) {
+      LOG.info("Container Balancer is not running.");
+      return;
+    }
+
+
+    //this is a temporary implementation
+    //modify this later
+    if (currentBalancingThread.isAlive()) {
+      currentBalancingThread.stop();
+    }
+    ///////////////////////////
+
+    balancerRunning.compareAndSet(true, false);
+    LOG.info("Container Balancer stopped successfully.");
   }
 
   public void setNodeManager(NodeManager nodeManager) {

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

@@ -56,6 +56,11 @@ public final class ContainerBalancerConfiguration {
           "by Container Balancer.")
   private long maxSizeToMove = 10 * OzoneConsts.GB;
 
+  @Config(key = "idle.iterations", type = ConfigType.INT,
+      defaultValue = "10", tags = {ConfigTag.BALANCER},
+      description = "The idle iteration count of Container Balancer")
+  private int idleIterations = 10;
+
   /**
    * Gets the threshold value for Container Balancer.
    *
@@ -78,6 +83,28 @@ public void setThreshold(double threshold) {
     this.threshold = String.valueOf(threshold);
   }
 
+  /**
+   * Gets the idle iteration value for Container Balancer.
+   *
+   * @return a idle iteration count larger than 0
+   */
+  public int getIdleIteration() {
+    return idleIterations;
+  }
+
+  /**
+   * Sets the idle iteration value for Container Balancer.
+   *
+   * @param count a idle iteration count larger than 0
+   */
+  public void setIdleIteration(int count) {
+    if (count < 1) {
+      throw new IllegalArgumentException(
+          "Idle iteration count must be larger than 0.");
+    }
+    this.idleIterations = count;
+  }
+
   /**
    * Gets the value of maximum number of datanodes that will be balanced by
    * Container Balancer.