HDDS-10699. Refactor ContainerBalancerTask and TestContainerBalancerTask (#6537)

Author: Montura

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

@@ -75,7 +75,6 @@ public class ContainerBalancerTask implements Runnable {
   private OzoneConfiguration ozoneConfiguration;
   private ContainerBalancer containerBalancer;
   private final SCMContext scmContext;
-  private double threshold;
   private int totalNodesInCluster;
   private double maxDatanodesRatioToInvolvePerIteration;
   private long maxSizeToMovePerIteration;
@@ -84,17 +83,13 @@ public class ContainerBalancerTask implements Runnable {
   // count actual size moved in bytes
   private long sizeActuallyMovedInLatestIteration;
   private int iterations;
-  private List<DatanodeUsageInfo> unBalancedNodes;
-  private List<DatanodeUsageInfo> overUtilizedNodes;
-  private List<DatanodeUsageInfo> underUtilizedNodes;
+  private final List<DatanodeUsageInfo> overUtilizedNodes;
+  private final List<DatanodeUsageInfo> underUtilizedNodes;
   private List<DatanodeUsageInfo> withinThresholdUtilizedNodes;
   private Set<String> excludeNodes;
   private Set<String> includeNodes;
   private ContainerBalancerConfiguration config;
   private ContainerBalancerMetrics metrics;
-  private long clusterCapacity;
-  private long clusterRemaining;
-  private double clusterAvgUtilisation;
   private PlacementPolicyValidateProxy placementPolicyValidateProxy;
   private NetworkTopology networkTopology;
   private double upperLimit;
@@ -150,7 +145,6 @@ public ContainerBalancerTask(StorageContainerManager scm,
     this.overUtilizedNodes = new ArrayList<>();
     this.underUtilizedNodes = new ArrayList<>();
     this.withinThresholdUtilizedNodes = new ArrayList<>();
-    this.unBalancedNodes = new ArrayList<>();
     this.placementPolicyValidateProxy = scm.getPlacementPolicyValidateProxy();
     this.networkTopology = scm.getClusterMap();
     this.nextIterationIndex = nextIterationIndex;
@@ -348,7 +342,6 @@ private boolean initializeIteration() {
       return false;
     }
 
-    this.threshold = config.getThresholdAsRatio();
     this.maxDatanodesRatioToInvolvePerIteration =
         config.getMaxDatanodesRatioToInvolvePerIteration();
     this.maxSizeToMovePerIteration = config.getMaxSizeToMovePerIteration();
@@ -368,22 +361,19 @@ private boolean initializeIteration() {
 
     this.totalNodesInCluster = datanodeUsageInfos.size();
 
-    clusterAvgUtilisation = calculateAvgUtilization(datanodeUsageInfos);
+    double clusterAvgUtilisation = calculateAvgUtilization(datanodeUsageInfos);
     if (LOG.isDebugEnabled()) {
-      LOG.debug("Average utilization of the cluster is {}",
-          clusterAvgUtilisation);
+      LOG.debug("Average utilization of the cluster is {}", clusterAvgUtilisation);
     }
 
-    // over utilized nodes have utilization(that is, used / capacity) greater
-    // than upper limit
+    double threshold = config.getThresholdAsRatio();
+    // over utilized nodes have utilization(that is, used / capacity) greater than upper limit
     this.upperLimit = clusterAvgUtilisation + threshold;
-    // under utilized nodes have utilization(that is, used / capacity) less
-    // than lower limit
+    // under utilized nodes have utilization(that is, used / capacity) less than lower limit
     this.lowerLimit = clusterAvgUtilisation - threshold;
 
     if (LOG.isDebugEnabled()) {
-      LOG.debug("Lower limit for utilization is {} and Upper limit for " +
-          "utilization is {}", lowerLimit, upperLimit);
+      LOG.debug("Lower limit for utilization is {} and Upper limit for utilization is {}", lowerLimit, upperLimit);
     }
 
     long totalOverUtilizedBytes = 0L, totalUnderUtilizedBytes = 0L;
@@ -433,12 +423,7 @@ private boolean initializeIteration() {
             OzoneConsts.GB);
     Collections.reverse(underUtilizedNodes);
 
-    unBalancedNodes = new ArrayList<>(
-        overUtilizedNodes.size() + underUtilizedNodes.size());
-    unBalancedNodes.addAll(overUtilizedNodes);
-    unBalancedNodes.addAll(underUtilizedNodes);
-
-    if (unBalancedNodes.isEmpty()) {
+    if (overUtilizedNodes.isEmpty() && underUtilizedNodes.isEmpty()) {
       LOG.info("Did not find any unbalanced Datanodes.");
       return false;
     }
@@ -487,7 +472,7 @@ private IterationResult doIteration() {
     findTargetStrategy.reInitialize(potentialTargets, config, upperLimit);
     findSourceStrategy.reInitialize(getPotentialSources(), config, lowerLimit);
 
-    moveSelectionToFutureMap = new HashMap<>(unBalancedNodes.size());
+    moveSelectionToFutureMap = new HashMap<>(underUtilizedNodes.size() + overUtilizedNodes.size());
     boolean isMoveGeneratedInThisIteration = false;
     iterationResult = IterationResult.ITERATION_COMPLETED;
     boolean canAdaptWhenNearingLimits = true;
@@ -965,8 +950,8 @@ private long ratioToBytes(Long nodeCapacity, double utilizationRatio) {
    * @return Average utilization value
    */
   @VisibleForTesting
-  double calculateAvgUtilization(List<DatanodeUsageInfo> nodes) {
-    if (nodes.size() == 0) {
+  public static double calculateAvgUtilization(List<DatanodeUsageInfo> nodes) {
+    if (nodes.isEmpty()) {
       LOG.warn("No nodes to calculate average utilization for in " +
           "ContainerBalancer.");
       return 0;
@@ -976,8 +961,8 @@ private long ratioToBytes(Long nodeCapacity, double utilizationRatio) {
     for (DatanodeUsageInfo node : nodes) {
       aggregatedStats.add(node.getScmNodeStat());
     }
-    clusterCapacity = aggregatedStats.getCapacity().get();
-    clusterRemaining = aggregatedStats.getRemaining().get();
+    long clusterCapacity = aggregatedStats.getCapacity().get();
+    long clusterRemaining = aggregatedStats.getRemaining().get();
 
     return (clusterCapacity - clusterRemaining) / (double) clusterCapacity;
   }
@@ -1060,11 +1045,8 @@ private void incSizeSelectedForMoving(DatanodeDetails source,
    */
   private void resetState() {
     moveManager.resetState();
-    this.clusterCapacity = 0L;
-    this.clusterRemaining = 0L;
     this.overUtilizedNodes.clear();
     this.underUtilizedNodes.clear();
-    this.unBalancedNodes.clear();
     this.containerToSourceMap.clear();
     this.containerToTargetMap.clear();
     this.selectedSources.clear();
@@ -1090,15 +1072,14 @@ private boolean isBalancerRunning() {
     return taskStatus == Status.RUNNING;
   }
 
-  /**
-   * Gets the list of unBalanced nodes, that is, the over and under utilized
-   * nodes in the cluster.
-   *
-   * @return List of DatanodeUsageInfo containing unBalanced nodes.
-   */
   @VisibleForTesting
-  List<DatanodeUsageInfo> getUnBalancedNodes() {
-    return unBalancedNodes;
+  public List<DatanodeUsageInfo> getOverUtilizedNodes() {
+    return overUtilizedNodes;
+  }
+
+  @VisibleForTesting
+  public List<DatanodeUsageInfo> getUnderUtilizedNodes() {
+    return underUtilizedNodes;
   }
 
   /**

hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/balancer/MockedSCM.java

@@ -6,9 +6,9 @@
  * 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
- * <p>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p>
+ *
+ *     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.
@@ -66,28 +66,32 @@ public final class MockedSCM {
   private final StorageContainerManager scm;
   private final TestableCluster cluster;
   private final MockNodeManager mockNodeManager;
-  private MockedReplicationManager mockedReplicaManager;
-  private MoveManager moveManager;
-  private ContainerManager containerManager;
-
+  private final MockedReplicationManager mockedReplicaManager;
+  private final MoveManager moveManager;
+  private final ContainerManager containerManager;
   private MockedPlacementPolicies mockedPlacementPolicies;
 
   public MockedSCM(@Nonnull TestableCluster testableCluster) {
     scm = mock(StorageContainerManager.class);
     cluster = testableCluster;
     mockNodeManager = new MockNodeManager(cluster.getDatanodeToContainersMap());
+    try {
+      moveManager = mockMoveManager();
+      containerManager = mockContainerManager(cluster);
+      mockedReplicaManager = MockedReplicationManager.doMock();
+    } catch (NodeNotFoundException | ContainerReplicaNotFoundException | ContainerNotFoundException |
+             TimeoutException e
+    ) {
+      throw new RuntimeException("Can't create MockedSCM instance: ", e);
+    }
   }
 
-  public void init(@Nonnull ContainerBalancerConfiguration balancerConfig) {
-    init(balancerConfig, new OzoneConfiguration());
-  }
-
-  public void init(@Nonnull ContainerBalancerConfiguration balancerConfig, @Nonnull OzoneConfiguration ozoneCfg) {
+  private void init(@Nonnull ContainerBalancerConfiguration balancerConfig, @Nonnull OzoneConfiguration ozoneCfg) {
     ozoneCfg.setFromObject(balancerConfig);
     try {
       doMock(balancerConfig, ozoneCfg);
     } catch (IOException | NodeNotFoundException | TimeoutException e) {
-      throw new RuntimeException("Can't initialize TestOzoneHDDS: ", e);
+      throw new RuntimeException("Can't create MockedSCM instance: ", e);
     }
   }
 
@@ -96,9 +100,6 @@ public void init(@Nonnull ContainerBalancerConfiguration balancerConfig, @Nonnul
    */
   private void doMock(@Nonnull ContainerBalancerConfiguration cfg, @Nonnull OzoneConfiguration ozoneCfg)
       throws IOException, NodeNotFoundException, TimeoutException {
-    containerManager = mockContainerManager(cluster);
-    mockedReplicaManager = MockedReplicationManager.doMock();
-    moveManager = mockMoveManager();
     StatefulServiceStateManager stateManager = MockedServiceStateManager.doMock();
     SCMServiceManager scmServiceManager = mockSCMServiceManger();
 
@@ -137,7 +138,7 @@ public String toString() {
   }
 
   public @Nonnull ContainerBalancerTask startBalancerTask(@Nonnull ContainerBalancerConfiguration config) {
-    init(config);
+    init(config, new OzoneConfiguration());
     return startBalancerTask(new ContainerBalancer(scm), config);
   }