This is an automated email from the ASF dual-hosted git repository.
sarvekshayr pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ozone.git
The following commit(s) were added to refs/heads/master by this push:
new 4250410bc29 HDDS-16173. Add cluster utilization analysis for container
balancer CLI commands (#11037)
4250410bc29 is described below
commit 4250410bc291ea4ca26d207fb2065b5493fb1ec7
Author: sravani <[email protected]>
AuthorDate: Thu Sep 17 12:57:20 2026 +0530
HDDS-16173. Add cluster utilization analysis for container balancer CLI
commands (#11037)
---
.../balancer/ContainerBalancerClusterAnalyzer.java | 272 +++++++++++++++++++
.../balancer/ContainerBalancerClusterSnapshot.java | 250 +++++++++++++++++
.../TestContainerBalancerClusterAnalyzer.java | 296 +++++++++++++++++++++
.../scm/container/balancer/ContainerBalancer.java | 10 +-
.../container/balancer/ContainerBalancerTask.java | 47 +---
.../TestContainerBalancerDatanodeNodeLimit.java | 88 ++++++
6 files changed, 922 insertions(+), 41 deletions(-)
diff --git
a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/ContainerBalancerClusterAnalyzer.java
b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/ContainerBalancerClusterAnalyzer.java
new file mode 100644
index 00000000000..56a5675b18d
--- /dev/null
+++
b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/ContainerBalancerClusterAnalyzer.java
@@ -0,0 +1,272 @@
+/*
+ * 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.container.balancer;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Set;
+import org.apache.hadoop.hdds.protocol.DatanodeDetails;
+import org.apache.hadoop.hdds.protocol.proto.HddsProtos.DatanodeUsageInfoProto;
+import
org.apache.hadoop.hdds.scm.container.balancer.ContainerBalancerClusterSnapshot.NodeUtilization;
+
+/**
+ * Classifies datanode usage protos using the same rules as
+ * {@code ContainerBalancerTask.initializeIteration()}.
+ */
+public final class ContainerBalancerClusterAnalyzer {
+
+ private static final int TOP_NODE_LIMIT = 5;
+
+ private ContainerBalancerClusterAnalyzer() {
+ }
+
+ /**
+ * Core shared formula: (totalCapacity - totalRemaining) / totalCapacity.
+ */
+ public static double calculateAvgUtilization(long totalCapacity, long
totalRemaining) {
+ if (totalCapacity == 0) {
+ return 0;
+ }
+ return (totalCapacity - totalRemaining) / (double) totalCapacity;
+ }
+
+ /**
+ * Cluster average from datanode usage protos.
+ */
+ public static double calculateAvgUtilization(List<DatanodeUsageInfoProto>
nodes) {
+ if (nodes.isEmpty()) {
+ return 0;
+ }
+ long totalCapacity = 0;
+ long totalRemaining = 0;
+ for (DatanodeUsageInfoProto node : nodes) {
+ totalCapacity += node.getCapacity();
+ totalRemaining += node.getRemaining();
+ }
+ return calculateAvgUtilization(totalCapacity, totalRemaining);
+ }
+
+ /**
+ * Returns true if utilization is above the upper limit (source candidate).
+ */
+ public static boolean isOverUtilized(double utilization, double upperLimit) {
+ return Double.compare(utilization, upperLimit) > 0;
+ }
+
+ /**
+ * Returns true if utilization is below the lower limit (target candidate).
+ */
+ public static boolean isUnderUtilized(double utilization, double lowerLimit)
{
+ return Double.compare(utilization, lowerLimit) < 0;
+ }
+
+ /**
+ * Bytes above the upper limit on an over-utilized node.
+ */
+ public static long overUtilizedBytes(long capacity, double utilization,
double upperLimit) {
+ return ratioToBytes(capacity, utilization) - ratioToBytes(capacity,
upperLimit);
+ }
+
+ /**
+ * Bytes below the lower limit on an under-utilized node.
+ */
+ public static long underUtilizedBytes(long capacity, double utilization,
double lowerLimit) {
+ return ratioToBytes(capacity, lowerLimit) - ratioToBytes(capacity,
utilization);
+ }
+
+ /**
+ * Upper utilization band: cluster average plus threshold.
+ * Nodes above this are candidates to be source nodes.
+ */
+ public static double computeUpperLimit(double clusterAvgUtilization, double
thresholdRatio) {
+ return clusterAvgUtilization + thresholdRatio;
+ }
+
+ /**
+ * Lower utilization band: cluster average minus threshold.
+ * Nodes below this are candidates to be target nodes.
+ */
+ public static double computeLowerLimit(double clusterAvgUtilization, double
thresholdRatio) {
+ return clusterAvgUtilization - thresholdRatio;
+ }
+
+ /**
+ * Builds a cluster snapshot after applying include/exclude filters.
+ *
+ * @param nodes all nodes from getDatanodeUsageInfo (typically healthy
IN_SERVICE)
+ * @param thresholdRatio threshold as ratio, e.g. 0.10 for 10%
+ * @param includeNodes empty = all included; non-empty = allow-list
+ * @param excludeNodes nodes to skip
+ */
+ public static ContainerBalancerClusterSnapshot analyze(
+ List<DatanodeUsageInfoProto> nodes,
+ double thresholdRatio,
+ Set<String> includeNodes,
+ Set<String> excludeNodes) {
+ List<DatanodeUsageInfoProto> eligible = filterEligibleNodes(nodes,
includeNodes, excludeNodes);
+
+ if (eligible.isEmpty()) {
+ return emptySnapshot(thresholdRatio);
+ }
+
+ double clusterAvgUtilization = calculateAvgUtilization(eligible);
+ double upperLimit = computeUpperLimit(clusterAvgUtilization,
thresholdRatio);
+ double lowerLimit = computeLowerLimit(clusterAvgUtilization,
thresholdRatio);
+
+ List<NodeUtilization> sources = new ArrayList<>();
+ List<NodeUtilization> targets = new ArrayList<>();
+ double maxUtilization = Double.NEGATIVE_INFINITY;
+ double minUtilization = Double.POSITIVE_INFINITY;
+ long clusterCapacityBytes = 0;
+ long totalOverUtilizedBytes = 0;
+ long totalUnderUtilizedBytes = 0;
+
+ for (DatanodeUsageInfoProto node : eligible) {
+ long capacity = node.getCapacity();
+ clusterCapacityBytes += capacity;
+ double utilization = calculateNodeUtilization(node);
+
+ maxUtilization = Math.max(maxUtilization, utilization);
+ minUtilization = Math.min(minUtilization, utilization);
+
+ String hostname = getDisplayHostname(node);
+ if (isOverUtilized(utilization, upperLimit)) {
+ totalOverUtilizedBytes += overUtilizedBytes(capacity, utilization,
upperLimit);
+ sources.add(new NodeUtilization(hostname, utilization));
+ } else if (isUnderUtilized(utilization, lowerLimit)) {
+ totalUnderUtilizedBytes += underUtilizedBytes(capacity, utilization,
lowerLimit);
+ targets.add(new NodeUtilization(hostname, utilization));
+ }
+ }
+
+
sources.sort(Comparator.comparingDouble(NodeUtilization::getUtilization).reversed());
+ targets.sort(Comparator.comparingDouble(NodeUtilization::getUtilization));
+
+ double imbalance = maxUtilization - minUtilization;
+
+ return ContainerBalancerClusterSnapshot.newBuilder()
+ .setTotalEligibleDatanodes(eligible.size())
+ .setClusterAvgUtilization(clusterAvgUtilization)
+ .setClusterCapacityBytes(clusterCapacityBytes)
+ .setMaxUtilization(maxUtilization)
+ .setMinUtilization(minUtilization)
+ .setUpperLimit(upperLimit)
+ .setLowerLimit(lowerLimit)
+ .setSourceCount(sources.size())
+ .setTargetCount(targets.size())
+ .setTotalOverUtilizedBytes(totalOverUtilizedBytes)
+ .setTotalUnderUtilizedBytes(totalUnderUtilizedBytes)
+ .setImbalance(imbalance)
+ .setTopSourceNodes(topNodes(sources))
+ .setBottomTargetNodes(topNodes(targets))
+ .build();
+ }
+
+ /**
+ * Function that returns whether a datanode should be excluded from cluster
analysis.
+ * A datanode is excluded if it appears in excludeNodes set, or if
+ * includeNodes is non-empty and the datanode is not listed there
+ * by hostname or IP address.
+ */
+ public static boolean shouldExcludeDatanode(DatanodeDetails datanode,
+ Set<String> excludeNodes,
Set<String> includeNodes) {
+ if (excludeNodes.contains(datanode.getHostName()) ||
+ excludeNodes.contains(datanode.getIpAddress())) {
+ return true;
+ } else if (!includeNodes.isEmpty()) {
+ return !includeNodes.contains(datanode.getHostName()) &&
+ !includeNodes.contains(datanode.getIpAddress());
+ }
+ return false;
+ }
+
+ static double calculateNodeUtilization(DatanodeUsageInfoProto node) {
+ long capacity = node.getCapacity();
+ if (capacity == 0) {
+ return 0;
+ }
+ return (capacity - node.getRemaining()) / (double) capacity;
+ }
+
+ private static List<DatanodeUsageInfoProto> filterEligibleNodes(
+ List<DatanodeUsageInfoProto> nodes,
+ Set<String> includeNodes,
+ Set<String> excludeNodes) {
+ List<DatanodeUsageInfoProto> eligible = new ArrayList<>();
+ for (DatanodeUsageInfoProto node : nodes) {
+ if (!node.hasNode()) {
+ continue;
+ }
+ DatanodeDetails datanode =
DatanodeDetails.getFromProtoBuf(node.getNode());
+ if (!shouldExcludeDatanode(datanode, excludeNodes, includeNodes)) {
+ eligible.add(node);
+ }
+ }
+ return eligible;
+ }
+
+ /**
+ * Calculates the number of used bytes given capacity and utilization ratio.
+ *
+ * @param nodeCapacity capacity of the node.
+ * @param utilizationRatio used space by capacity ratio of the node.
+ * @return number of bytes
+ */
+ public static long ratioToBytes(long nodeCapacity, double utilizationRatio) {
+ return (long) (nodeCapacity * utilizationRatio);
+ }
+
+ private static String getDisplayHostname(DatanodeUsageInfoProto node) {
+ DatanodeDetails datanode = DatanodeDetails.getFromProtoBuf(node.getNode());
+ String hostname = datanode.getHostName();
+ if (hostname != null && !hostname.isEmpty()) {
+ return hostname;
+ }
+ return datanode.getIpAddress();
+ }
+
+ private static List<NodeUtilization> topNodes(List<NodeUtilization> nodes) {
+ int limit = Math.min(TOP_NODE_LIMIT, nodes.size());
+ List<NodeUtilization> result = new ArrayList<>(limit);
+ for (int i = 0; i < limit; i++) {
+ result.add(nodes.get(i));
+ }
+ return result;
+ }
+
+ private static ContainerBalancerClusterSnapshot emptySnapshot(double
thresholdRatio) {
+ return ContainerBalancerClusterSnapshot.newBuilder()
+ .setTotalEligibleDatanodes(0)
+ .setClusterAvgUtilization(0)
+ .setClusterCapacityBytes(0)
+ .setMaxUtilization(0)
+ .setMinUtilization(0)
+ .setUpperLimit(computeUpperLimit(0, thresholdRatio))
+ .setLowerLimit(computeLowerLimit(0, thresholdRatio))
+ .setSourceCount(0)
+ .setTargetCount(0)
+ .setTotalOverUtilizedBytes(0)
+ .setTotalUnderUtilizedBytes(0)
+ .setImbalance(0)
+ .setTopSourceNodes(Collections.emptyList())
+ .setBottomTargetNodes(Collections.emptyList())
+ .build();
+ }
+}
diff --git
a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/ContainerBalancerClusterSnapshot.java
b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/ContainerBalancerClusterSnapshot.java
new file mode 100644
index 00000000000..7c74c16763e
--- /dev/null
+++
b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/ContainerBalancerClusterSnapshot.java
@@ -0,0 +1,250 @@
+/*
+ * 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.container.balancer;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+
+/**
+ * Point-in-time cluster imbalance view built from datanode usage protos.
+ */
+public final class ContainerBalancerClusterSnapshot {
+
+ private final int totalEligibleDatanodes;
+ private final double clusterAvgUtilization;
+ private final long clusterCapacityBytes;
+ private final double maxUtilization;
+ private final double minUtilization;
+ private final double upperLimit;
+ private final double lowerLimit;
+ private final int sourceCount;
+ private final int targetCount;
+ private final long totalOverUtilizedBytes;
+ private final long totalUnderUtilizedBytes;
+
+ // bytesToMove has same value as totalOverUtilizedBytes. The analyzer sets it
+ // once via setTotalOverUtilizedBytes(), bytesToMove is assigned here for
assessment output.
+ private final long bytesToMove;
+ private final double imbalance;
+ private final List<NodeUtilization> topSourceNodes;
+ private final List<NodeUtilization> bottomTargetNodes;
+
+ private ContainerBalancerClusterSnapshot(Builder b) {
+ this.totalEligibleDatanodes = b.totalEligibleDatanodes;
+ this.clusterAvgUtilization = b.clusterAvgUtilization;
+ this.clusterCapacityBytes = b.clusterCapacityBytes;
+ this.maxUtilization = b.maxUtilization;
+ this.minUtilization = b.minUtilization;
+ this.upperLimit = b.upperLimit;
+ this.lowerLimit = b.lowerLimit;
+ this.sourceCount = b.sourceCount;
+ this.targetCount = b.targetCount;
+ this.totalOverUtilizedBytes = b.totalOverUtilizedBytes;
+ this.totalUnderUtilizedBytes = b.totalUnderUtilizedBytes;
+ this.bytesToMove = b.totalOverUtilizedBytes;
+ this.imbalance = b.imbalance;
+ this.topSourceNodes = Collections.unmodifiableList(
+ Objects.requireNonNull(b.topSourceNodes));
+ this.bottomTargetNodes = Collections.unmodifiableList(
+ Objects.requireNonNull(b.bottomTargetNodes));
+ }
+
+ public int getTotalEligibleDatanodes() {
+ return totalEligibleDatanodes;
+ }
+
+ public double getClusterAvgUtilization() {
+ return clusterAvgUtilization;
+ }
+
+ public long getClusterCapacityBytes() {
+ return clusterCapacityBytes;
+ }
+
+ public double getMaxUtilization() {
+ return maxUtilization;
+ }
+
+ public double getMinUtilization() {
+ return minUtilization;
+ }
+
+ public double getUpperLimit() {
+ return upperLimit;
+ }
+
+ public double getLowerLimit() {
+ return lowerLimit;
+ }
+
+ public int getSourceCount() {
+ return sourceCount;
+ }
+
+ public int getTargetCount() {
+ return targetCount;
+ }
+
+ public long getTotalOverUtilizedBytes() {
+ return totalOverUtilizedBytes;
+ }
+
+ public long getTotalUnderUtilizedBytes() {
+ return totalUnderUtilizedBytes;
+ }
+
+ public long getBytesToMove() {
+ return bytesToMove;
+ }
+
+ public double getImbalance() {
+ return imbalance;
+ }
+
+ public List<NodeUtilization> getTopSourceNodes() {
+ return topSourceNodes;
+ }
+
+ public List<NodeUtilization> getBottomTargetNodes() {
+ return bottomTargetNodes;
+ }
+
+ public static Builder newBuilder() {
+ return new Builder();
+ }
+
+ /**
+ * Builder class for building ContainerBalancerClusterSnapshot.
+ */
+ public static final class Builder {
+ private int totalEligibleDatanodes;
+ private double clusterAvgUtilization;
+ private long clusterCapacityBytes;
+ private double maxUtilization;
+ private double minUtilization;
+ private double upperLimit;
+ private double lowerLimit;
+ private int sourceCount;
+ private int targetCount;
+ private long totalOverUtilizedBytes;
+ private long totalUnderUtilizedBytes;
+ private double imbalance;
+ private List<NodeUtilization> topSourceNodes = Collections.emptyList();
+ private List<NodeUtilization> bottomTargetNodes = Collections.emptyList();
+
+ private Builder() {
+ }
+
+ public Builder setTotalEligibleDatanodes(int totalEligibleDatanodes) {
+ this.totalEligibleDatanodes = totalEligibleDatanodes;
+ return this;
+ }
+
+ public Builder setClusterAvgUtilization(double clusterAvgUtilization) {
+ this.clusterAvgUtilization = clusterAvgUtilization;
+ return this;
+ }
+
+ public Builder setClusterCapacityBytes(long clusterCapacityBytes) {
+ this.clusterCapacityBytes = clusterCapacityBytes;
+ return this;
+ }
+
+ public Builder setMaxUtilization(double maxUtilization) {
+ this.maxUtilization = maxUtilization;
+ return this;
+ }
+
+ public Builder setMinUtilization(double minUtilization) {
+ this.minUtilization = minUtilization;
+ return this;
+ }
+
+ public Builder setUpperLimit(double upperLimit) {
+ this.upperLimit = upperLimit;
+ return this;
+ }
+
+ public Builder setLowerLimit(double lowerLimit) {
+ this.lowerLimit = lowerLimit;
+ return this;
+ }
+
+ public Builder setSourceCount(int sourceCount) {
+ this.sourceCount = sourceCount;
+ return this;
+ }
+
+ public Builder setTargetCount(int targetCount) {
+ this.targetCount = targetCount;
+ return this;
+ }
+
+ public Builder setTotalOverUtilizedBytes(long totalOverUtilizedBytes) {
+ this.totalOverUtilizedBytes = totalOverUtilizedBytes;
+ return this;
+ }
+
+ public Builder setTotalUnderUtilizedBytes(long totalUnderUtilizedBytes) {
+ this.totalUnderUtilizedBytes = totalUnderUtilizedBytes;
+ return this;
+ }
+
+ public Builder setImbalance(double imbalance) {
+ this.imbalance = imbalance;
+ return this;
+ }
+
+ public Builder setTopSourceNodes(List<NodeUtilization> topSourceNodes) {
+ this.topSourceNodes = topSourceNodes;
+ return this;
+ }
+
+ public Builder setBottomTargetNodes(List<NodeUtilization>
bottomTargetNodes) {
+ this.bottomTargetNodes = bottomTargetNodes;
+ return this;
+ }
+
+ public ContainerBalancerClusterSnapshot build() {
+ return new ContainerBalancerClusterSnapshot(this);
+ }
+ }
+
+ /**
+ * Hostname and utilization ratio of a single datanode, retained on the
snapshot
+ * for per-node display in CLI output (e.g., assessment).
+ */
+ public static final class NodeUtilization {
+ private final String hostname;
+ private final double utilization;
+
+ public NodeUtilization(String hostname, double utilization) {
+ this.hostname = hostname;
+ this.utilization = utilization;
+ }
+
+ public String getHostname() {
+ return hostname;
+ }
+
+ public double getUtilization() {
+ return utilization;
+ }
+ }
+}
diff --git
a/hadoop-hdds/common/src/test/java/org/apache/hadoop/hdds/scm/container/balancer/TestContainerBalancerClusterAnalyzer.java
b/hadoop-hdds/common/src/test/java/org/apache/hadoop/hdds/scm/container/balancer/TestContainerBalancerClusterAnalyzer.java
new file mode 100644
index 00000000000..004d78feecc
--- /dev/null
+++
b/hadoop-hdds/common/src/test/java/org/apache/hadoop/hdds/scm/container/balancer/TestContainerBalancerClusterAnalyzer.java
@@ -0,0 +1,296 @@
+/*
+ * 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.container.balancer;
+
+import static org.apache.hadoop.ozone.ClientVersion.DEFAULT_VERSION;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.UUID;
+import java.util.stream.Collectors;
+import org.apache.hadoop.hdds.protocol.DatanodeDetails;
+import org.apache.hadoop.hdds.protocol.proto.HddsProtos.DatanodeUsageInfoProto;
+import
org.apache.hadoop.hdds.scm.container.balancer.ContainerBalancerClusterSnapshot.NodeUtilization;
+import org.junit.jupiter.api.Test;
+
+/** Tests for {@link ContainerBalancerClusterAnalyzer}. */
+public final class TestContainerBalancerClusterAnalyzer {
+
+ private static final double THRESHOLD = 0.10;
+
+ @Test
+ void testCalculateAvgUtilizationFromTotals() {
+ assertEquals(0.70d, ContainerBalancerClusterAnalyzer
+ .calculateAvgUtilization(200, 60), 0.0001);
+ assertEquals(0, ContainerBalancerClusterAnalyzer
+ .calculateAvgUtilization(0, 0));
+ }
+
+ @Test
+ void testBalancedClusterHasNoSourcesOrTargets() {
+ List<DatanodeUsageInfoProto> nodes = Arrays.asList(
+ proto("dn-1", 100, 30),
+ proto("dn-2", 100, 30));
+
+ ContainerBalancerClusterSnapshot snapshot =
+ ContainerBalancerClusterAnalyzer.analyze(nodes, THRESHOLD,
+ Collections.emptySet(), Collections.emptySet());
+
+ assertEquals(2, snapshot.getTotalEligibleDatanodes());
+ assertEquals(0.70d, snapshot.getClusterAvgUtilization(), 0.0001);
+ assertEquals(0, snapshot.getSourceCount());
+ assertEquals(0, snapshot.getTargetCount());
+ assertEquals(0, snapshot.getBytesToMove());
+ assertTrue(snapshot.getTopSourceNodes().isEmpty());
+ assertTrue(snapshot.getBottomTargetNodes().isEmpty());
+ }
+
+ @Test
+ void testOverAndUnderUtilizedClassification() {
+ List<DatanodeUsageInfoProto> nodes = Arrays.asList(
+ proto("source-1", 100, 10),
+ proto("target-1", 100, 50));
+
+ ContainerBalancerClusterSnapshot snapshot =
+ ContainerBalancerClusterAnalyzer.analyze(nodes, THRESHOLD,
+ Collections.emptySet(), Collections.emptySet());
+
+ assertEquals(1, snapshot.getSourceCount());
+ assertEquals(1, snapshot.getTargetCount());
+ assertEquals(0.90d, snapshot.getMaxUtilization(), 0.0001);
+ assertEquals(0.50d, snapshot.getMinUtilization(), 0.0001);
+ assertEquals(0.40d, snapshot.getImbalance(), 0.0001);
+ assertEquals(10, snapshot.getTotalOverUtilizedBytes());
+ assertEquals(10, snapshot.getTotalUnderUtilizedBytes());
+ assertEquals(snapshot.getTotalOverUtilizedBytes(),
snapshot.getBytesToMove());
+ assertEquals(Collections.singletonList("source-1"),
+ snapshot.getTopSourceNodes().stream()
+ .map(NodeUtilization::getHostname).collect(Collectors.toList()));
+ assertEquals(Collections.singletonList("target-1"),
+ snapshot.getBottomTargetNodes().stream()
+ .map(NodeUtilization::getHostname).collect(Collectors.toList()));
+ }
+
+ @Test
+ void testIncludeExcludeFilters() {
+ List<DatanodeUsageInfoProto> nodes = Arrays.asList(
+ proto("keep-me", 100, 10), // 90%
+ proto("also-keep", 100, 50), // 50%
+ proto("drop-me", 100, 30)); // filtered out
+
+ Set<String> include = new HashSet<>(Arrays.asList("keep-me", "also-keep"));
+
+ ContainerBalancerClusterSnapshot snapshot =
+ ContainerBalancerClusterAnalyzer.analyze(nodes, THRESHOLD, include,
+ Collections.emptySet());
+
+ assertEquals(2, snapshot.getTotalEligibleDatanodes());
+ assertEquals(1, snapshot.getSourceCount()); // keep-me
+ assertEquals(1, snapshot.getTargetCount()); // also-keep
+ }
+
+ @Test
+ void testEmptyAfterFilterReturnsEmptySnapshot() {
+ List<DatanodeUsageInfoProto> nodes = Collections.singletonList(
+ proto("excluded", 100, 10));
+
+ ContainerBalancerClusterSnapshot snapshot =
+ ContainerBalancerClusterAnalyzer.analyze(nodes, THRESHOLD,
+ Collections.emptySet(), Collections.singleton("excluded"));
+
+ assertEquals(0, snapshot.getTotalEligibleDatanodes());
+ assertEquals(0, snapshot.getClusterCapacityBytes());
+ assertEquals(0, snapshot.getBytesToMove());
+ }
+
+ @Test
+ void testZeroCapacityNodeTreatedAsZeroUtilization() {
+ List<DatanodeUsageInfoProto> nodes = Arrays.asList(
+ proto("zero-cap", 0, 0),
+ proto("normal", 100, 30));
+
+ ContainerBalancerClusterSnapshot snapshot =
+ ContainerBalancerClusterAnalyzer.analyze(nodes, THRESHOLD,
+ Collections.emptySet(), Collections.emptySet());
+
+ assertEquals(2, snapshot.getTotalEligibleDatanodes());
+ assertEquals(0, snapshot.getMinUtilization(), 0.0001);
+ }
+
+ @Test
+ void testProtoWithoutNodeIsSkipped() {
+ DatanodeUsageInfoProto noNode = DatanodeUsageInfoProto.newBuilder()
+ .setCapacity(100)
+ .setRemaining(10)
+ .setUsed(90)
+ .build();
+
+ ContainerBalancerClusterSnapshot snapshot =
+ ContainerBalancerClusterAnalyzer.analyze(
+ Collections.singletonList(noNode), THRESHOLD,
+ Collections.emptySet(), Collections.emptySet());
+
+ assertEquals(0, snapshot.getTotalEligibleDatanodes());
+ }
+
+ @Test
+ void testEmptyInputListReturnsEmptySnapshot() {
+ ContainerBalancerClusterSnapshot snapshot =
+ ContainerBalancerClusterAnalyzer.analyze(
+ Collections.emptyList(), THRESHOLD,
+ Collections.emptySet(), Collections.emptySet());
+
+ assertEquals(0, snapshot.getTotalEligibleDatanodes());
+ assertEquals(0, snapshot.getClusterAvgUtilization(), 0.0001);
+ assertEquals(0, snapshot.getBytesToMove());
+ }
+
+ @Test
+ void testExcludeByIpAddress() {
+ DatanodeDetails datanode = DatanodeDetails.newBuilder()
+ .setHostName("dn-host")
+ .setIpAddress("10.0.0.5")
+ .setUuid(UUID.randomUUID())
+ .build();
+ DatanodeUsageInfoProto node = DatanodeUsageInfoProto.newBuilder()
+ .setNode(datanode.toProto(DEFAULT_VERSION.toProtoValue()))
+ .setCapacity(100)
+ .setRemaining(10)
+ .setUsed(90)
+ .build();
+
+ ContainerBalancerClusterSnapshot snapshot =
+ ContainerBalancerClusterAnalyzer.analyze(
+ Collections.singletonList(node), THRESHOLD,
+ Collections.emptySet(), Collections.singleton("10.0.0.5"));
+
+ assertEquals(0, snapshot.getTotalEligibleDatanodes());
+ }
+
+ @Test
+ void testExcludeWinsOverInclude() {
+ List<DatanodeUsageInfoProto> nodes = Collections.singletonList(
+ proto("both-lists", 100, 10));
+
+ Set<String> include = Collections.singleton("both-lists");
+ Set<String> exclude = Collections.singleton("both-lists");
+
+ ContainerBalancerClusterSnapshot snapshot =
+ ContainerBalancerClusterAnalyzer.analyze(nodes, THRESHOLD, include,
exclude);
+
+ assertEquals(0, snapshot.getTotalEligibleDatanodes());
+ }
+
+ @Test
+ void testTopFiveSourcesAndBottomFiveTargets() {
+ List<DatanodeUsageInfoProto> nodes = new ArrayList<>();
+ for (int i = 0; i < 7; i++) {
+ nodes.add(proto(String.format("source-%02d", i), 100, 10));
+ }
+ for (int i = 0; i < 7; i++) {
+ nodes.add(proto(String.format("target-%02d", i), 100, 50));
+ }
+
+ ContainerBalancerClusterSnapshot snapshot =
+ ContainerBalancerClusterAnalyzer.analyze(nodes, THRESHOLD,
+ Collections.emptySet(), Collections.emptySet());
+
+ assertEquals(7, snapshot.getSourceCount());
+ assertEquals(7, snapshot.getTargetCount());
+ assertEquals(5, snapshot.getTopSourceNodes().size());
+ assertEquals(5, snapshot.getBottomTargetNodes().size());
+ Set<String> expectedSources = new HashSet<>();
+ for (int i = 0; i < 7; i++) {
+ expectedSources.add(String.format("source-%02d", i));
+ }
+ Set<String> expectedTargets = new HashSet<>();
+ for (int i = 0; i < 7; i++) {
+ expectedTargets.add(String.format("target-%02d", i));
+ }
+ assertTrue(expectedSources.containsAll(
+ snapshot.getTopSourceNodes().stream()
+ .map(NodeUtilization::getHostname).collect(Collectors.toList())));
+ assertTrue(expectedTargets.containsAll(
+ snapshot.getBottomTargetNodes().stream()
+ .map(NodeUtilization::getHostname).collect(Collectors.toList())));
+ }
+
+ @Test
+ void testTopSourceNodesCarryUtilization() {
+ List<DatanodeUsageInfoProto> nodes = Arrays.asList(
+ proto("source-hot", 100, 5), // 95% used
+ proto("source-warm", 100, 15), // 85% used
+ proto("balanced", 100, 50),
+ proto("target-cool", 100, 90) // 10% used
+ );
+
+ ContainerBalancerClusterSnapshot snapshot =
+ ContainerBalancerClusterAnalyzer.analyze(nodes, THRESHOLD,
+ Collections.emptySet(), Collections.emptySet());
+
+ List<NodeUtilization> top = snapshot.getTopSourceNodes();
+ assertEquals("source-hot", top.get(0).getHostname());
+ assertEquals(0.95, top.get(0).getUtilization(), 1e-9);
+ assertEquals("source-warm", top.get(1).getHostname());
+ assertEquals(0.85, top.get(1).getUtilization(), 1e-9);
+
+ List<NodeUtilization> bottom = snapshot.getBottomTargetNodes();
+ assertEquals("target-cool", bottom.get(0).getHostname());
+ assertEquals(0.10, bottom.get(0).getUtilization(), 1e-9);
+ }
+
+ @Test
+ void testShouldExcludeDatanodeDirectly() {
+ DatanodeDetails datanode = DatanodeDetails.newBuilder()
+ .setHostName("dn-1")
+ .setIpAddress("10.0.0.1")
+ .setUuid(UUID.randomUUID())
+ .build();
+
+ assertTrue(ContainerBalancerClusterAnalyzer.shouldExcludeDatanode(
+ datanode, Collections.singleton("dn-1"), Collections.emptySet()));
+ assertTrue(ContainerBalancerClusterAnalyzer.shouldExcludeDatanode(
+ datanode, Collections.singleton("10.0.0.1"), Collections.emptySet()));
+ assertTrue(ContainerBalancerClusterAnalyzer.shouldExcludeDatanode(
+ datanode, Collections.emptySet(), Collections.singleton("other")));
+ assertFalse(ContainerBalancerClusterAnalyzer.shouldExcludeDatanode(
+ datanode, Collections.emptySet(), Collections.singleton("dn-1")));
+ }
+
+ private static DatanodeUsageInfoProto proto(String hostname, long capacity,
long remaining) {
+ DatanodeDetails datanode = DatanodeDetails.newBuilder()
+ .setHostName(hostname)
+ .setIpAddress("127.0.0.1")
+ .setUuid(UUID.randomUUID())
+ .build();
+ long used = capacity - remaining;
+ return DatanodeUsageInfoProto.newBuilder()
+ .setNode(datanode.toProto(DEFAULT_VERSION.toProtoValue()))
+ .setCapacity(capacity)
+ .setRemaining(remaining)
+ .setUsed(used)
+ .build();
+ }
+}
diff --git
a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/ContainerBalancer.java
b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/ContainerBalancer.java
index 159551fe4ca..14aa7efdc38 100644
---
a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/ContainerBalancer.java
+++
b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/ContainerBalancer.java
@@ -660,14 +660,8 @@ private int
countEligibleDatanodes(ContainerBalancerConfiguration conf) {
static boolean shouldExcludeDatanode(DatanodeDetails datanode,
Set<String> excludeNodes, Set<String> includeNodes) {
- if (excludeNodes.contains(datanode.getHostName()) ||
- excludeNodes.contains(datanode.getIpAddress())) {
- return true;
- } else if (!includeNodes.isEmpty()) {
- return !includeNodes.contains(datanode.getHostName()) &&
- !includeNodes.contains(datanode.getIpAddress());
- }
- return false;
+ return ContainerBalancerClusterAnalyzer.shouldExcludeDatanode(
+ datanode, excludeNodes, includeNodes);
}
public ContainerBalancerMetrics getMetrics() {
diff --git
a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/ContainerBalancerTask.java
b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/ContainerBalancerTask.java
index 358d696d5b8..4dff0265445 100644
---
a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/ContainerBalancerTask.java
+++
b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/ContainerBalancerTask.java
@@ -563,9 +563,11 @@ private boolean initializeIteration() {
double threshold = config.getThresholdAsRatio();
// over utilized nodes have utilization(that is, used / capacity) greater
than upper limit
- this.upperLimit = clusterAvgUtilisation + threshold;
+ this.upperLimit = ContainerBalancerClusterAnalyzer.computeUpperLimit(
+ clusterAvgUtilisation, threshold);
// under utilized nodes have utilization(that is, used / capacity) less
than lower limit
- this.lowerLimit = clusterAvgUtilisation - threshold;
+ this.lowerLimit = ContainerBalancerClusterAnalyzer.computeLowerLimit(
+ clusterAvgUtilisation, threshold);
if (LOG.isDebugEnabled()) {
LOG.debug("Lower limit for utilization is {} and Upper limit for
utilization is {}", lowerLimit, upperLimit);
@@ -587,28 +589,18 @@ private boolean initializeIteration() {
datanodeUsageInfo.getScmNodeStat().getRemaining().get(),
utilization);
}
- if (Double.compare(utilization, upperLimit) > 0) {
+ if (ContainerBalancerClusterAnalyzer.isOverUtilized(utilization,
upperLimit)) {
overUtilizedNodes.add(datanodeUsageInfo);
metrics.incrementNumDatanodesUnbalanced(1);
-
- // amount of bytes greater than upper limit in this node
- long overUtilizedBytes = ratioToBytes(
- datanodeUsageInfo.getScmNodeStat().getCapacity().get(),
- utilization) - ratioToBytes(
- datanodeUsageInfo.getScmNodeStat().getCapacity().get(),
- upperLimit);
- totalOverUtilizedBytes += overUtilizedBytes;
- } else if (Double.compare(utilization, lowerLimit) < 0) {
+ totalOverUtilizedBytes +=
+
ContainerBalancerClusterAnalyzer.overUtilizedBytes(datanodeUsageInfo.getScmNodeStat().getCapacity().get(),
+ utilization, upperLimit);
+ } else if (ContainerBalancerClusterAnalyzer.isUnderUtilized(utilization,
lowerLimit)) {
underUtilizedNodes.add(datanodeUsageInfo);
metrics.incrementNumDatanodesUnbalanced(1);
-
- // amount of bytes lesser than lower limit in this node
- long underUtilizedBytes = ratioToBytes(
- datanodeUsageInfo.getScmNodeStat().getCapacity().get(),
- lowerLimit) - ratioToBytes(
- datanodeUsageInfo.getScmNodeStat().getCapacity().get(),
- utilization);
- totalUnderUtilizedBytes += underUtilizedBytes;
+ totalUnderUtilizedBytes +=
+
ContainerBalancerClusterAnalyzer.underUtilizedBytes(datanodeUsageInfo.getScmNodeStat().getCapacity().get(),
+ utilization, lowerLimit);
}
}
metrics.incrementDataSizeUnbalancedGB(
@@ -1128,17 +1120,6 @@ private void updateTargetsAndSelectionCriteria(
selectedSources.add(source);
}
- /**
- * Calculates the number of used bytes given capacity and utilization ratio.
- *
- * @param nodeCapacity capacity of the node.
- * @param utilizationRatio used space by capacity ratio of the node.
- * @return number of bytes
- */
- private long ratioToBytes(Long nodeCapacity, double utilizationRatio) {
- return (long) (nodeCapacity * utilizationRatio);
- }
-
/**
* Calculates the average utilization for the specified nodes.
* Utilization is (capacity - remaining) divided by capacity.
@@ -1160,8 +1141,8 @@ public static double
calculateAvgUtilization(List<DatanodeUsageInfo> nodes) {
}
long clusterCapacity = aggregatedStats.getCapacity().get();
long clusterRemaining = aggregatedStats.getRemaining().get();
-
- return (clusterCapacity - clusterRemaining) / (double) clusterCapacity;
+ return ContainerBalancerClusterAnalyzer.calculateAvgUtilization(
+ clusterCapacity, clusterRemaining);
}
/**
diff --git
a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/balancer/TestContainerBalancerDatanodeNodeLimit.java
b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/balancer/TestContainerBalancerDatanodeNodeLimit.java
index f37eaab9654..1b4f420cf1a 100644
---
a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/balancer/TestContainerBalancerDatanodeNodeLimit.java
+++
b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/balancer/TestContainerBalancerDatanodeNodeLimit.java
@@ -35,6 +35,7 @@
import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Comparator;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
@@ -46,6 +47,7 @@
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.hdds.protocol.DatanodeDetails;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
+import org.apache.hadoop.hdds.protocol.proto.HddsProtos.DatanodeUsageInfoProto;
import
org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos;
import org.apache.hadoop.hdds.scm.ContainerPlacementStatus;
import org.apache.hadoop.hdds.scm.container.ContainerID;
@@ -53,9 +55,11 @@
import org.apache.hadoop.hdds.scm.container.ContainerNotFoundException;
import org.apache.hadoop.hdds.scm.container.ContainerReplica;
import org.apache.hadoop.hdds.scm.container.ContainerReplicaNotFoundException;
+import
org.apache.hadoop.hdds.scm.container.balancer.ContainerBalancerClusterSnapshot.NodeUtilization;
import org.apache.hadoop.hdds.scm.node.DatanodeUsageInfo;
import org.apache.hadoop.hdds.scm.node.NodeStatus;
import org.apache.hadoop.hdds.scm.node.states.NodeNotFoundException;
+import org.apache.hadoop.ozone.ClientVersion;
import org.apache.hadoop.ozone.OzoneConsts;
import org.apache.ozone.test.GenericTestUtils;
import org.junit.jupiter.api.BeforeAll;
@@ -96,6 +100,90 @@ private static Stream<Arguments> createMockedSCMs() {
Arguments.of(getMockedSCM(30)));
}
+ @ParameterizedTest(name = "MockedSCM #{index}: {0}")
+ @MethodSource("createMockedSCMs")
+ void analyzerMatchesTaskAfterInitializeIteration(@Nonnull MockedSCM
mockedSCM) {
+ ContainerBalancerConfiguration config =
+ new ContainerBalancerConfigBuilder(mockedSCM.getNodeCount()).build();
+
+ ContainerBalancerTask task = mockedSCM.startBalancerTask(config);
+
+ List<DatanodeUsageInfo> eligible =
mockedSCM.getNodeManager().getMostOrLeastUsedDatanodes(true)
+ .stream()
+ .filter(n -> !ContainerBalancerClusterAnalyzer.shouldExcludeDatanode(
+ n.getDatanodeDetails(), config.getExcludeNodes(),
config.getIncludeNodes()))
+ .collect(Collectors.toList());
+ List<DatanodeUsageInfoProto> protos = eligible.stream()
+ .map(n -> n.toProto(ClientVersion.DEFAULT_VERSION.toProtoValue()))
+ .collect(Collectors.toList());
+
+ ContainerBalancerClusterSnapshot snapshot =
ContainerBalancerClusterAnalyzer.analyze(
+ protos, config.getThresholdAsRatio(), config.getIncludeNodes(),
config.getExcludeNodes());
+
+ double taskAvg = ContainerBalancerTask.calculateAvgUtilization(eligible);
+ double upperLimit = ContainerBalancerClusterAnalyzer.computeUpperLimit(
+ taskAvg, config.getThresholdAsRatio());
+ double lowerLimit = ContainerBalancerClusterAnalyzer.computeLowerLimit(
+ taskAvg, config.getThresholdAsRatio());
+
+ assertEquals(eligible.size(), snapshot.getTotalEligibleDatanodes());
+ assertEquals(task.getOverUtilizedNodes().size(),
snapshot.getSourceCount());
+ assertEquals(task.getUnderUtilizedNodes().size(),
snapshot.getTargetCount());
+ assertEquals(taskAvg, snapshot.getClusterAvgUtilization(), 0.0001);
+ assertEquals(upperLimit, snapshot.getUpperLimit(), 0.0001);
+ assertEquals(lowerLimit, snapshot.getLowerLimit(), 0.0001);
+
+ long expectedOverBytes = task.getOverUtilizedNodes().stream()
+ .mapToLong(n -> {
+ long capacity = n.getScmNodeStat().getCapacity().get();
+ double utilization = n.calculateUtilization();
+ return ContainerBalancerClusterAnalyzer.overUtilizedBytes(
+ capacity, utilization, upperLimit);
+ })
+ .sum();
+ long expectedUnderBytes = task.getUnderUtilizedNodes().stream()
+ .mapToLong(n -> {
+ long capacity = n.getScmNodeStat().getCapacity().get();
+ double utilization = n.calculateUtilization();
+ return ContainerBalancerClusterAnalyzer.underUtilizedBytes(
+ capacity, utilization, lowerLimit);
+ })
+ .sum();
+ assertEquals(expectedOverBytes, snapshot.getTotalOverUtilizedBytes());
+ assertEquals(expectedUnderBytes, snapshot.getTotalUnderUtilizedBytes());
+ assertEquals(expectedOverBytes, snapshot.getBytesToMove());
+ assertEquals(
+ Math.max(snapshot.getTotalOverUtilizedBytes(),
snapshot.getTotalUnderUtilizedBytes()) / OzoneConsts.GB,
+ task.getMetrics().getDataSizeUnbalancedGB());
+
+ List<NodeUtilization> expectedTopSources =
task.getOverUtilizedNodes().stream()
+ .sorted(Comparator.comparingDouble((DatanodeUsageInfo n) ->
n.calculateUtilization()).reversed())
+ .limit(5)
+ .map(n -> new NodeUtilization(n.getDatanodeDetails().getHostName(),
+ n.calculateUtilization()))
+ .collect(Collectors.toList());
+ List<NodeUtilization> expectedBottomTargets =
task.getUnderUtilizedNodes().stream()
+ .sorted(Comparator.comparingDouble((DatanodeUsageInfo n) ->
n.calculateUtilization()))
+ .limit(5)
+ .map(n -> new NodeUtilization(n.getDatanodeDetails().getHostName(),
+ n.calculateUtilization()))
+ .collect(Collectors.toList());
+
+ List<NodeUtilization> actualTopSources = snapshot.getTopSourceNodes();
+ assertEquals(expectedTopSources.size(), actualTopSources.size());
+ for (int i = 0; i < expectedTopSources.size(); i++) {
+ assertEquals(expectedTopSources.get(i).getHostname(),
actualTopSources.get(i).getHostname());
+ assertEquals(expectedTopSources.get(i).getUtilization(),
actualTopSources.get(i).getUtilization(), 0.0001);
+ }
+
+ List<NodeUtilization> actualBottomTargets =
snapshot.getBottomTargetNodes();
+ assertEquals(expectedBottomTargets.size(), actualBottomTargets.size());
+ for (int i = 0; i < expectedBottomTargets.size(); i++) {
+ assertEquals(expectedBottomTargets.get(i).getHostname(),
actualBottomTargets.get(i).getHostname());
+ assertEquals(expectedBottomTargets.get(i).getUtilization(),
actualBottomTargets.get(i).getUtilization(), 0.0001);
+ }
+ }
+
@ParameterizedTest(name = "MockedSCM #{index}: {0}")
@MethodSource("createMockedSCMs")
public void containerBalancerShouldObeyMaxDatanodesToInvolveLimit(@Nonnull
MockedSCM mockedSCM) {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
