You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by xy...@apache.org on 2024/06/19 09:40:56 UTC
(pulsar) branch master updated: [improve] [pip] PIP-358: let resource weight work for OverloadShedder, LeastLongTermMessageRate, ModularLoadManagerImpl. (#22888)
This is an automated email from the ASF dual-hosted git repository.
xyz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new 5dc030431a6 [improve] [pip] PIP-358: let resource weight work for OverloadShedder, LeastLongTermMessageRate, ModularLoadManagerImpl. (#22888)
5dc030431a6 is described below
commit 5dc030431a60b49e81d577cd06a1ae63dbee0293
Author: Wenzhi Feng <52...@users.noreply.github.com>
AuthorDate: Wed Jun 19 17:40:50 2024 +0800
[improve] [pip] PIP-358: let resource weight work for OverloadShedder, LeastLongTermMessageRate, ModularLoadManagerImpl. (#22888)
---
conf/broker.conf | 5 ---
conf/standalone.conf | 5 ---
.../loadbalance/impl/LeastLongTermMessageRate.java | 6 +++-
.../loadbalance/impl/ModularLoadManagerImpl.java | 28 +++++++++++-----
.../broker/loadbalance/impl/OverloadShedder.java | 9 +++--
.../ModularLoadManagerStrategyTest.java | 3 ++
.../impl/ModularLoadManagerImplTest.java | 4 +++
.../loadbalance/impl/OverloadShedderTest.java | 38 ++++++++++++++++++++++
8 files changed, 76 insertions(+), 22 deletions(-)
diff --git a/conf/broker.conf b/conf/broker.conf
index 8fd266d609c..5c5d8d42817 100644
--- a/conf/broker.conf
+++ b/conf/broker.conf
@@ -1394,19 +1394,15 @@ loadBalancerMsgThroughputMultiplierDifferenceShedderThreshold=4
loadBalancerHistoryResourcePercentage=0.9
# The BandWidthIn usage weight when calculating new resource usage.
-# It only takes effect in the ThresholdShedder strategy.
loadBalancerBandwidthInResourceWeight=1.0
# The BandWidthOut usage weight when calculating new resource usage.
-# It only takes effect in the ThresholdShedder strategy.
loadBalancerBandwidthOutResourceWeight=1.0
# The CPU usage weight when calculating new resource usage.
-# It only takes effect in the ThresholdShedder strategy.
loadBalancerCPUResourceWeight=1.0
# The direct memory usage weight when calculating new resource usage.
-# It only takes effect in the ThresholdShedder strategy.
# Direct memory usage cannot accurately reflect the machine's load,
# and it is not recommended to use it to score the machine's load.
loadBalancerDirectMemoryResourceWeight=0
@@ -1814,7 +1810,6 @@ strictBookieAffinityEnabled=false
# These settings are left here for compatibility
# The heap memory usage weight when calculating new resource usage.
-# It only takes effect in the ThresholdShedder strategy.
# Deprecated: Memory is no longer used as a load balancing item
loadBalancerMemoryResourceWeight=0
diff --git a/conf/standalone.conf b/conf/standalone.conf
index 6b261ce11c6..635b31ac38d 100644
--- a/conf/standalone.conf
+++ b/conf/standalone.conf
@@ -911,23 +911,18 @@ loadBalancerBrokerThresholdShedderPercentage=10
loadBalancerHistoryResourcePercentage=0.9
# The BandWidthIn usage weight when calculating new resource usage.
-# It only takes effect in the ThresholdShedder strategy.
loadBalancerBandwidthInResourceWeight=1.0
# The BandWidthOut usage weight when calculating new resource usage.
-# It only takes effect in the ThresholdShedder strategy.
loadBalancerBandwidthOutResourceWeight=1.0
# The CPU usage weight when calculating new resource usage.
-# It only takes effect in the ThresholdShedder strategy.
loadBalancerCPUResourceWeight=1.0
# The heap memory usage weight when calculating new resource usage.
-# It only takes effect in the ThresholdShedder strategy.
loadBalancerMemoryResourceWeight=0
# The direct memory usage weight when calculating new resource usage.
-# It only takes effect in the ThresholdShedder strategy.
loadBalancerDirectMemoryResourceWeight=0
# Bundle unload minimum throughput threshold (MB), avoiding bundle unload frequently.
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/LeastLongTermMessageRate.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/LeastLongTermMessageRate.java
index fe161467338..f51ca797f0e 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/LeastLongTermMessageRate.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/LeastLongTermMessageRate.java
@@ -52,7 +52,11 @@ public class LeastLongTermMessageRate implements ModularLoadManagerStrategy {
// Any broker at (or above) the overload threshold will have a score of POSITIVE_INFINITY.
private static double getScore(final BrokerData brokerData, final ServiceConfiguration conf) {
final double overloadThreshold = conf.getLoadBalancerBrokerOverloadedThresholdPercentage() / 100.0;
- final double maxUsage = brokerData.getLocalData().getMaxResourceUsage();
+ final double maxUsage = brokerData.getLocalData().getMaxResourceUsageWithWeight(
+ conf.getLoadBalancerCPUResourceWeight(),
+ conf.getLoadBalancerDirectMemoryResourceWeight(),
+ conf.getLoadBalancerBandwidthInResourceWeight(),
+ conf.getLoadBalancerBandwidthOutResourceWeight());
if (maxUsage > overloadThreshold) {
log.warn("Broker {} is overloaded: max usage={}", brokerData.getLocalData().getWebServiceUrl(), maxUsage);
return Double.POSITIVE_INFINITY;
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java
index e1259e97aa3..3af372607cb 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java
@@ -419,6 +419,14 @@ public class ModularLoadManagerImpl implements ModularLoadManager {
return 100 * Math.abs((oldValue - newValue) / oldValue);
}
+ private double getMaxResourceUsageWithWeight(LocalBrokerData localBrokerData, ServiceConfiguration conf) {
+ return localBrokerData.getMaxResourceUsageWithWeight(
+ conf.getLoadBalancerCPUResourceWeight(),
+ conf.getLoadBalancerDirectMemoryResourceWeight(),
+ conf.getLoadBalancerBandwidthInResourceWeight(),
+ conf.getLoadBalancerBandwidthOutResourceWeight());
+ }
+
// Determine if the broker data requires an update by delegating to the update condition.
private boolean needBrokerDataUpdate() {
final long updateMaxIntervalMillis = TimeUnit.MINUTES
@@ -431,14 +439,15 @@ public class ModularLoadManagerImpl implements ModularLoadManager {
// Always update after surpassing the maximum interval.
return true;
}
- final double maxChange = Math
- .max(100.0 * (Math.abs(lastData.getMaxResourceUsage() - localData.getMaxResourceUsage())),
- Math.max(percentChange(lastData.getMsgRateIn() + lastData.getMsgRateOut(),
- localData.getMsgRateIn() + localData.getMsgRateOut()),
- Math.max(
- percentChange(lastData.getMsgThroughputIn() + lastData.getMsgThroughputOut(),
- localData.getMsgThroughputIn() + localData.getMsgThroughputOut()),
- percentChange(lastData.getNumBundles(), localData.getNumBundles()))));
+ final double maxChange = LocalBrokerData.max(
+ percentChange(lastData.getMsgRateIn() + lastData.getMsgRateOut(),
+ localData.getMsgRateIn() + localData.getMsgRateOut()),
+ percentChange(lastData.getMsgThroughputIn() + lastData.getMsgThroughputOut(),
+ localData.getMsgThroughputIn() + localData.getMsgThroughputOut()),
+ percentChange(lastData.getNumBundles(), localData.getNumBundles()),
+ 100.0 * Math.abs(getMaxResourceUsageWithWeight(lastData, conf)
+ - getMaxResourceUsageWithWeight(localData, conf))
+ );
if (maxChange > conf.getLoadBalancerReportUpdateThresholdPercentage()) {
log.info("Writing local data to metadata store because maximum change {}% exceeded threshold {}%; "
+ "time since last report written is {} seconds", maxChange,
@@ -927,7 +936,8 @@ public class ModularLoadManagerImpl implements ModularLoadManager {
}
final double overloadThreshold = conf.getLoadBalancerBrokerOverloadedThresholdPercentage() / 100.0;
- final double maxUsage = loadData.getBrokerData().get(broker.get()).getLocalData().getMaxResourceUsage();
+ final double maxUsage = getMaxResourceUsageWithWeight(
+ loadData.getBrokerData().get(broker.get()).getLocalData(), conf);
if (maxUsage > overloadThreshold) {
// All brokers that were in the filtered list were overloaded, so check if there is a better broker
LoadManagerShared.applyNamespacePolicies(serviceUnit, policies, brokerCandidateCache,
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/OverloadShedder.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/OverloadShedder.java
index a4eb5077224..fb31548227b 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/OverloadShedder.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/OverloadShedder.java
@@ -36,7 +36,8 @@ import org.slf4j.LoggerFactory;
/**
* Load shedding strategy which will attempt to shed exactly one bundle on brokers which are overloaded, that is, whose
* maximum system resource usage exceeds loadBalancerBrokerOverloadedThresholdPercentage. To see which resources are
- * considered when determining the maximum system resource, see {@link LocalBrokerData#getMaxResourceUsage()}. A bundle
+ * considered when determining the maximum system resource, see
+ * {@link LocalBrokerData#getMaxResourceUsageWithWeight(double, double, double, double)}. A bundle
* is recommended for unloading off that broker if and only if the following conditions hold: The broker has at
* least two bundles assigned and the broker has at least one bundle that has not been unloaded recently according to
* LoadBalancerSheddingGracePeriodMinutes. The unloaded bundle will be the most expensive bundle in terms of message
@@ -71,7 +72,11 @@ public class OverloadShedder implements LoadSheddingStrategy {
loadData.getBrokerData().forEach((broker, brokerData) -> {
final LocalBrokerData localData = brokerData.getLocalData();
- final double currentUsage = localData.getMaxResourceUsage();
+ final double currentUsage = localData.getMaxResourceUsageWithWeight(
+ conf.getLoadBalancerCPUResourceWeight(),
+ conf.getLoadBalancerDirectMemoryResourceWeight(),
+ conf.getLoadBalancerBandwidthInResourceWeight(),
+ conf.getLoadBalancerBandwidthOutResourceWeight());
if (currentUsage < overloadThreshold) {
if (log.isDebugEnabled()) {
log.debug("[{}] Broker is not overloaded, ignoring at this point ({})", broker,
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/ModularLoadManagerStrategyTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/ModularLoadManagerStrategyTest.java
index 2b6bfb742eb..f5bd0f46a5e 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/ModularLoadManagerStrategyTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/ModularLoadManagerStrategyTest.java
@@ -68,6 +68,9 @@ public class ModularLoadManagerStrategyTest {
assertEquals(strategy.selectBroker(brokerDataMap.keySet(), bundleData, loadData, conf), Optional.of("2"));
brokerData2.getLocalData().setCpu(new ResourceUsage(90, 100));
assertEquals(strategy.selectBroker(brokerDataMap.keySet(), bundleData, loadData, conf), Optional.of("3"));
+ // disable considering cpu usage to avoid broker2 being overloaded.
+ conf.setLoadBalancerCPUResourceWeight(0);
+ assertEquals(strategy.selectBroker(brokerDataMap.keySet(), bundleData, loadData, conf), Optional.of("2"));
}
// Test that least resource usage with weight works correctly.
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImplTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImplTest.java
index 6ae491c55b8..68bef8b241c 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImplTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImplTest.java
@@ -684,6 +684,10 @@ public class ModularLoadManagerImplTest {
currentData.setCpu(new ResourceUsage(206, 1000));
assert (needUpdate.get());
+ // set the resource weight of cpu to 0, so that it should not trigger an update
+ conf.setLoadBalancerCPUResourceWeight(0);
+ assert (!needUpdate.get());
+
lastData.setCpu(new ResourceUsage());
currentData.setCpu(new ResourceUsage());
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/impl/OverloadShedderTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/impl/OverloadShedderTest.java
index 1f0962f2e44..a05c5d4a9e0 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/impl/OverloadShedderTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/impl/OverloadShedderTest.java
@@ -149,6 +149,44 @@ public class OverloadShedderTest {
assertEquals(bundlesToUnload.get("broker-1"), List.of("bundle-10", "bundle-9"));
}
+ @Test
+ public void testBrokerWithResourceWeight() {
+ int numBundles = 10;
+ LoadData loadData = new LoadData();
+ LocalBrokerData broker1 = new LocalBrokerData();
+
+ double brokerThroghput = 0;
+ for (int i = 1; i <= numBundles; i++) {
+ broker1.getBundles().add("bundle-" + i);
+ BundleData bundle = new BundleData();
+ TimeAverageMessageData db = new TimeAverageMessageData();
+ double throughput = i * 1024 * 1024;
+ db.setMsgThroughputIn(throughput);
+ db.setMsgThroughputOut(throughput);
+ bundle.setShortTermData(db);
+ loadData.getBundleData().put("bundle-" + i, bundle);
+ brokerThroghput += throughput;
+ }
+ broker1.setMsgThroughputIn(brokerThroghput);
+ broker1.setMsgThroughputOut(brokerThroghput);
+
+ loadData.getBrokerData().put("broker-1", new BrokerData(broker1));
+
+ // set bandwidth usage to 99.9% so that it is considered overloaded
+ broker1.setBandwidthIn(new ResourceUsage(999, 1000));
+ broker1.setBandwidthOut(new ResourceUsage(999, 1000));
+ assertFalse(os.findBundlesForUnloading(loadData, conf).isEmpty());
+
+ // set bandwidth resource weight to 0 so that it is not considered overloaded
+ conf.setLoadBalancerBandwidthInResourceWeight(0);
+ conf.setLoadBalancerBandwidthOutResourceWeight(0);
+ assertTrue(os.findBundlesForUnloading(loadData, conf).isEmpty());
+
+ // set bandwidth resource weight back to 1, or it will affect other tests
+ conf.setLoadBalancerBandwidthInResourceWeight(1);
+ conf.setLoadBalancerBandwidthOutResourceWeight(1);
+ }
+
@Test
public void testFilterRecentlyUnloaded() {
int numBundles = 10;