diff --git a/conf/broker.conf b/conf/broker.conf
index b532478e56d883a44e73c690deba289e8b8fba67..5318fea9199e259e6dc9fabb61a1727a9f48c5c4 100644
--- a/conf/broker.conf
+++ b/conf/broker.conf
@@ -212,10 +212,7 @@ loadBalancerReportUpdateThresholdPercentage=10
# maximum interval to update load report
loadBalancerReportUpdateMaxIntervalMinutes=15
-# Path for the script used to retrieve system usage
-loadBalancerHostUsageScriptPath=
-
-# Frequency of sar report to collect
+# Frequency of report to collect
loadBalancerHostUsageCheckIntervalMinutes=1
# Load shedding interval. Broker periodically checks whether some traffic should be offload from
diff --git a/conf/standalone.conf b/conf/standalone.conf
index bc42c9852231cb2f2c5af71b62ffa45ad5307a2a..3bf62d5488576caba33b28e4ecaf7f8b9a1ff6db 100644
--- a/conf/standalone.conf
+++ b/conf/standalone.conf
@@ -185,10 +185,7 @@ loadBalancerReportUpdateThresholdPercentage=10
# maximum interval to update load report
loadBalancerReportUpdateMaxIntervalMinutes=15
-# Path for the script used to retrieve system usage
-loadBalancerHostUsageScriptPath=
-
-# Frequency of sar report to collect
+# Frequency of report to collect
loadBalancerHostUsageCheckIntervalMinutes=1
# Load shedding interval. Broker periodically checks whether some traffic should be offload from
diff --git a/pulsar-broker-common/src/main/java/com/yahoo/pulsar/broker/ServiceConfiguration.java b/pulsar-broker-common/src/main/java/com/yahoo/pulsar/broker/ServiceConfiguration.java
index 80ce45a0743055d7e4d1953416f4db3063dc0735..11c2220101743465265d5733d20c0fca61c60ad1 100644
--- a/pulsar-broker-common/src/main/java/com/yahoo/pulsar/broker/ServiceConfiguration.java
+++ b/pulsar-broker-common/src/main/java/com/yahoo/pulsar/broker/ServiceConfiguration.java
@@ -186,10 +186,7 @@ public class ServiceConfiguration {
private int loadBalancerReportUpdateThresholdPercentage = 10;
// maximum interval to update load report
private int loadBalancerReportUpdateMaxIntervalMinutes = 15;
- // Path for the script used to retrieve system usage
- @FieldContext(required = false)
- private String loadBalancerHostUsageScriptPath;
- // Frequency of sar report to collect
+ // Frequency of report to collect
private int loadBalancerHostUsageCheckIntervalMinutes = 1;
// Load shedding interval. Broker periodically checks whether some traffic
// should be offload from
@@ -715,14 +712,6 @@ public class ServiceConfiguration {
this.loadBalancerReportUpdateMaxIntervalMinutes = loadBalancerReportUpdateMaxIntervalMinutes;
}
- public String getLoadBalancerHostUsageScriptPath() {
- return loadBalancerHostUsageScriptPath;
- }
-
- public void setLoadBalancerHostUsageScriptPath(String loadBalancerHostUsageScriptPath) {
- this.loadBalancerHostUsageScriptPath = loadBalancerHostUsageScriptPath;
- }
-
public int getLoadBalancerHostUsageCheckIntervalMinutes() {
return loadBalancerHostUsageCheckIntervalMinutes;
}
diff --git a/pulsar-broker/src/main/java/com/yahoo/pulsar/broker/PulsarService.java b/pulsar-broker/src/main/java/com/yahoo/pulsar/broker/PulsarService.java
index 01b753e7742207d00030685b119da13d80be8033..3d722789419dcaf356c47bcac07f3eceabbc161a 100644
--- a/pulsar-broker/src/main/java/com/yahoo/pulsar/broker/PulsarService.java
+++ b/pulsar-broker/src/main/java/com/yahoo/pulsar/broker/PulsarService.java
@@ -128,6 +128,7 @@ public class PulsarService implements AutoCloseable {
this.brokerServiceUrlTls = brokerUrlTls(config);
this.config = config;
this.shutdownService = new MessagingServiceShutdownHook(this);
+ loadManagerExecutor = Executors.newSingleThreadScheduledExecutor();
}
/**
@@ -233,11 +234,11 @@ public class PulsarService implements AutoCloseable {
managedLedgerClientFactory = new ManagedLedgerClientFactory(config, getZkClient(),
getBookKeeperClientFactory());
+ this.brokerService = new BrokerService(this);
+
// Start load management service (even if load balancing is disabled)
this.loadManager = new SimpleLoadManagerImpl(this);
- this.brokerService = new BrokerService(this);
-
this.startLoadManagementService();
// needs load management service
@@ -400,11 +401,6 @@ public class PulsarService implements AutoCloseable {
this.loadManager.start();
if (config.isLoadBalancerEnabled()) {
- if (loadManagerExecutor == null) {
- loadManagerExecutor = Executors.newSingleThreadScheduledExecutor();
- ;
- }
-
LOG.info("Starting load balancer");
if (this.loadReportTask == null) {
long loadReportMinInterval = SimpleLoadManagerImpl.LOAD_REPORT_UPDATE_MIMIMUM_INTERVAL;
@@ -533,6 +529,10 @@ public class PulsarService implements AutoCloseable {
return executor;
}
+ public ScheduledExecutorService getLoadManagerExecutor() {
+ return loadManagerExecutor;
+ }
+
public OrderedSafeExecutor getOrderedExecutor() {
return orderedExecutor;
}
diff --git a/pulsar-broker/src/main/java/com/yahoo/pulsar/broker/loadbalance/BrokerHostUsage.java b/pulsar-broker/src/main/java/com/yahoo/pulsar/broker/loadbalance/BrokerHostUsage.java
index 85378ff8170c8fd28e1abc8672fb35e35f05e8d3..530c9a4ec698cae38e71eff89ca56def1486d1c1 100644
--- a/pulsar-broker/src/main/java/com/yahoo/pulsar/broker/loadbalance/BrokerHostUsage.java
+++ b/pulsar-broker/src/main/java/com/yahoo/pulsar/broker/loadbalance/BrokerHostUsage.java
@@ -16,80 +16,18 @@
package com.yahoo.pulsar.broker.loadbalance;
import java.io.IOException;
-import java.io.StringWriter;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.yahoo.pulsar.broker.PulsarService;
-import com.yahoo.pulsar.utils.CmdUtility;
+import com.yahoo.pulsar.common.policies.data.loadbalancer.SystemResourceUsage;
/**
* Class that will return the broker host usage.
*
*
*/
-public class BrokerHostUsage {
- // The interval for host usage check command
- private final int hostUsageCheckInterval;
-
- // Path to the pulsar-broker-host-usage script
- private final String usageScriptPath;
-
- private static final Logger LOG = LoggerFactory.getLogger(BrokerHostUsage.class);
-
- public BrokerHostUsage(PulsarService pulsar) {
- this.usageScriptPath = pulsar.getConfiguration().getLoadBalancerHostUsageScriptPath();
- this.hostUsageCheckInterval = pulsar.getConfiguration().getLoadBalancerHostUsageCheckIntervalMinutes();
- }
-
+public interface BrokerHostUsage {
/**
* Returns the host usage information in the following format -
*
- *
- * {
- * "bandwidthIn" : {
- * "usage" : "100",
- * "limit" : "1000",
- * },
- * "bandwidthOut" : {
- * "usage" : "659",
- * "limit" : "1000",
- * },
- * "memory" : {
- * "usage" : "16.0",
- * "limit" : "16070",
- * }
- * "cpu-utilization" : {
- * "usage" : "160.0"
- * "limit" : "1600"
- * }
- * }
- *
- *
* @return Broker host usage in the json string format
- *
- * @throws IOException
*/
- public String getBrokerHostUsage() throws IOException {
- StringWriter writer = new StringWriter();
- try {
- /**
- * Spawns a python process and runs the usage exporter script. The script return the machine information in
- * the json format.
- */
-
- int exitCode = CmdUtility.exec(writer, usageScriptPath, "--host-usage-check-interval",
- Integer.toString(hostUsageCheckInterval));
- if (exitCode != 0) {
- LOG.warn("Process exited with non-zero exit code - [{}], stderr - [{}] ", exitCode, writer.toString());
- throw new IOException(writer.toString());
- }
- } catch (IOException e) {
- e.printStackTrace();
- LOG.warn("Error running the usage script {}", e.getMessage());
- throw e;
- }
- return writer.toString();
- }
+ SystemResourceUsage getBrokerHostUsage();
}
diff --git a/pulsar-broker/src/main/java/com/yahoo/pulsar/broker/loadbalance/impl/GenericBrokerHostUsageImpl.java b/pulsar-broker/src/main/java/com/yahoo/pulsar/broker/loadbalance/impl/GenericBrokerHostUsageImpl.java
new file mode 100644
index 0000000000000000000000000000000000000000..1837404a2e5f28a60c57c662c32074d81a87fe00
--- /dev/null
+++ b/pulsar-broker/src/main/java/com/yahoo/pulsar/broker/loadbalance/impl/GenericBrokerHostUsageImpl.java
@@ -0,0 +1,96 @@
+/**
+ * Copyright 2016 Yahoo Inc.
+ *
+ * Licensed 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 com.yahoo.pulsar.broker.loadbalance.impl;
+
+import com.sun.management.OperatingSystemMXBean;
+import com.yahoo.pulsar.broker.PulsarService;
+import com.yahoo.pulsar.broker.loadbalance.BrokerHostUsage;
+import com.yahoo.pulsar.common.policies.data.loadbalancer.ResourceUsage;
+import com.yahoo.pulsar.common.policies.data.loadbalancer.SystemResourceUsage;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.lang.management.ManagementFactory;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Class that will return the broker host usage.
+ */
+public class GenericBrokerHostUsageImpl implements BrokerHostUsage {
+ // The interval for host usage check command
+ private static final int CPU_CHECK_MILLIS = 1000;
+ private static final Logger LOG = LoggerFactory.getLogger(GenericBrokerHostUsageImpl.class);
+ private final int hostUsageCheckIntervalMin;
+ private long lastCollection;
+ private double totalCpuLimit;
+ private double cpuUsageSum = 0d;
+ private int cpuUsageCount = 0;
+ private OperatingSystemMXBean systemBean;
+ private SystemResourceUsage usage;
+
+ public GenericBrokerHostUsageImpl(PulsarService pulsar) {
+ this.hostUsageCheckIntervalMin = pulsar.getConfiguration().getLoadBalancerHostUsageCheckIntervalMinutes();
+ this.systemBean = (OperatingSystemMXBean) ManagementFactory.getOperatingSystemMXBean();
+ this.lastCollection = 0L;
+ this.usage = new SystemResourceUsage();
+ this.totalCpuLimit = getTotalCpuLimit();
+ pulsar.getLoadManagerExecutor().scheduleAtFixedRate(this::checkCpuLoad, 0, CPU_CHECK_MILLIS, TimeUnit.MILLISECONDS);
+ pulsar.getLoadManagerExecutor().scheduleAtFixedRate(this::calculateBrokerHostUsage, 0, hostUsageCheckIntervalMin, TimeUnit.MINUTES);
+ }
+
+ @Override
+ public SystemResourceUsage getBrokerHostUsage() {
+ return usage;
+ }
+
+ private void checkCpuLoad() {
+ cpuUsageSum += systemBean.getSystemCpuLoad();
+ cpuUsageCount++;
+ }
+
+ private void calculateBrokerHostUsage() {
+ SystemResourceUsage usage = new SystemResourceUsage();
+ usage.setCpu(getCpuUsage());
+ usage.setMemory(getMemUsage());
+
+ this.usage = usage;
+ }
+
+ private double getTotalCpuLimit() {
+ return (double) (100 * Runtime.getRuntime().availableProcessors());
+ }
+
+ private double getTotalCpuUsage() {
+ double cpuUsage = cpuUsageSum / cpuUsageCount;
+ cpuUsageSum = 0d;
+ cpuUsageCount = 0;
+ return cpuUsage;
+ }
+
+ private ResourceUsage getCpuUsage() {
+ if (cpuUsageCount == 0) {
+ return new ResourceUsage(0, totalCpuLimit);
+ }
+ return new ResourceUsage(getTotalCpuUsage() * totalCpuLimit, totalCpuLimit);
+ }
+
+ private ResourceUsage getMemUsage() {
+ double total = ((double) systemBean.getTotalPhysicalMemorySize()) / (1024 * 1024);
+ double free = ((double) systemBean.getFreePhysicalMemorySize()) / (1024 * 1024);
+ return new ResourceUsage(total - free, total);
+ }
+}
diff --git a/pulsar-broker/src/main/java/com/yahoo/pulsar/broker/loadbalance/impl/LinuxBrokerHostUsageImpl.java b/pulsar-broker/src/main/java/com/yahoo/pulsar/broker/loadbalance/impl/LinuxBrokerHostUsageImpl.java
new file mode 100644
index 0000000000000000000000000000000000000000..fe353da92a10593dd535f8d65c36e8f2a9765d1b
--- /dev/null
+++ b/pulsar-broker/src/main/java/com/yahoo/pulsar/broker/loadbalance/impl/LinuxBrokerHostUsageImpl.java
@@ -0,0 +1,241 @@
+/**
+ * Copyright 2016 Yahoo Inc.
+ *
+ * Licensed 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 com.yahoo.pulsar.broker.loadbalance.impl;
+
+import com.sun.management.OperatingSystemMXBean;
+import com.yahoo.pulsar.broker.PulsarService;
+import com.yahoo.pulsar.broker.loadbalance.BrokerHostUsage;
+import com.yahoo.pulsar.common.policies.data.loadbalancer.ResourceUsage;
+import com.yahoo.pulsar.common.policies.data.loadbalancer.SystemResourceUsage;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.lang.management.ManagementFactory;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+/**
+ * Class that will return the broker host usage.
+ *
+ *
+ */
+public class LinuxBrokerHostUsageImpl implements BrokerHostUsage {
+ // The interval for host usage check command
+ private final int hostUsageCheckInterval;
+ private long lastCollection;
+ private double lastTotalNicUsageTx;
+ private double lastTotalNicUsageRx;
+ private CpuStat lastCpuStat;
+ private OperatingSystemMXBean systemBean;
+ private SystemResourceUsage usage;
+
+ private static final Logger LOG = LoggerFactory.getLogger(LinuxBrokerHostUsageImpl.class);
+
+ public LinuxBrokerHostUsageImpl(PulsarService pulsar) {
+ this.hostUsageCheckInterval = pulsar.getConfiguration().getLoadBalancerHostUsageCheckIntervalMinutes();
+ this.systemBean = (OperatingSystemMXBean) ManagementFactory.getOperatingSystemMXBean();
+ this.lastCollection = 0L;
+ this.usage = new SystemResourceUsage();
+ pulsar.getLoadManagerExecutor().scheduleAtFixedRate(this::calculateBrokerHostUsage, 0, hostUsageCheckInterval, TimeUnit.SECONDS);
+ }
+
+ @Override
+ public SystemResourceUsage getBrokerHostUsage() {
+ return usage;
+ }
+
+ private void calculateBrokerHostUsage() {
+ List nics = getNics();
+ double totalNicLimit = getTotalNicLimitKbps(nics);
+ double totalNicUsageTx = getTotalNicUsageTxKb(nics);
+ double totalNicUsageRx = getTotalNicUsageRxKb(nics);
+ double totalCpuLimit = getTotalCpuLimit();
+ CpuStat cpuStat = getTotalCpuUsage();
+ SystemResourceUsage usage = new SystemResourceUsage();
+ long now = System.currentTimeMillis();
+
+ if (lastCollection == 0L) {
+ usage.setMemory(getMemUsage());
+ usage.setBandwidthIn(new ResourceUsage(0d, totalNicLimit));
+ usage.setBandwidthOut(new ResourceUsage(0d, totalNicLimit));
+ usage.setCpu(new ResourceUsage(0d, totalCpuLimit));
+ } else {
+ double elapsedSeconds = (now - lastCollection) / 1000d;
+ double nicUsageTx = (totalNicUsageTx - lastTotalNicUsageTx) / elapsedSeconds;
+ double nicUsageRx = (totalNicUsageRx - lastTotalNicUsageRx) / elapsedSeconds;
+
+ if (cpuStat != null && lastCpuStat != null) {
+ // we need two non null stats to get a usage report
+ long cpuTimeDiff = cpuStat.getTotalTime() - lastCpuStat.getTotalTime();
+ long cpuUsageDiff = cpuStat.getUsage() - lastCpuStat.getUsage();
+ double cpuUsage = ((double) cpuUsageDiff / (double) cpuTimeDiff) * totalCpuLimit;
+ usage.setCpu(new ResourceUsage(cpuUsage, totalCpuLimit));
+ }
+
+ usage.setMemory(getMemUsage());
+ usage.setBandwidthIn(new ResourceUsage(nicUsageRx, totalNicLimit));
+ usage.setBandwidthOut(new ResourceUsage(nicUsageTx, totalNicLimit));
+ }
+
+ lastTotalNicUsageTx = totalNicUsageTx;
+ lastTotalNicUsageRx = totalNicUsageRx;
+ lastCpuStat = cpuStat;
+ lastCollection = System.currentTimeMillis();
+ this.usage = usage;
+ }
+
+ private double getTotalCpuLimit() {
+ return (double) (100 * Runtime.getRuntime().availableProcessors());
+ }
+
+ /**
+ * Reads first line of /proc/stat to get total cpu usage.
+ *
+ * cpu user nice system idle iowait irq softirq steal guest guest_nice
+ * cpu 317808 128 58637 2503692 7634 0 13472 0 0 0
+ *
+ * Line is split in "words", filtering the first.
+ * The sum of all numbers give the amount of cpu cycles used this far.
+ * Real CPU usage should equal the sum substracting the idle cycles,
+ * this would include iowait, irq and steal.
+ */
+ private CpuStat getTotalCpuUsage() {
+ try {
+ String[] words = Files.lines(Paths.get("/proc/stat"))
+ .findFirst()
+ .get().split("\\s+");
+
+ long total = Arrays.stream(words)
+ .filter(s -> !s.contains("cpu"))
+ .mapToLong(Long::parseLong)
+ .sum();
+
+ long idle = Long.parseLong(words[4]);
+
+ return new CpuStat(total, total - idle);
+ } catch (IOException e) {
+ LOG.error("Failed to read CPU usage from /proc/stat", e);
+ return null;
+ }
+ }
+
+ private ResourceUsage getMemUsage() {
+ double total = ((double) systemBean.getTotalPhysicalMemorySize()) / (1024 * 1024);
+ double free = ((double) systemBean.getFreePhysicalMemorySize()) / (1024 * 1024);
+ return new ResourceUsage(total - free, total);
+ }
+
+ private List getNics() {
+ try {
+ return Files.list(Paths.get("/sys/class/net/"))
+ .filter(this::isPhysicalNic)
+ .map(path -> path.getFileName().toString())
+ .collect(Collectors.toList());
+ } catch (IOException e) {
+ LOG.error("Failed to find NICs", e);
+ return Collections.emptyList();
+ }
+ }
+
+ private boolean isPhysicalNic(Path path) {
+ try {
+ if (!Files.readSymbolicLink(path).toString().contains("/virtual/")) {
+ try {
+ Files.readAllBytes(path.resolve("speed"));
+ return true;
+ } catch (Exception e) {
+ // wireless nics don't report speed, ignore them.
+ return false;
+ }
+ }
+ return false;
+ } catch (IOException e) {
+ LOG.error("Failed to read link target for NIC " + path, e);
+ return false;
+ }
+ }
+
+ private Path getNicSpeedPath(String nic) {
+ return Paths.get(String.format("/sys/class/net/%s/speed", nic));
+ }
+
+ private double getTotalNicLimitKbps(List nics) {
+ // Nic speed is in Mbits/s, return kbits/s
+ return nics.stream().mapToDouble(s -> {
+ try {
+ return Double.parseDouble(new String(Files.readAllBytes(getNicSpeedPath(s))));
+ } catch (IOException e) {
+ LOG.error("Failed to read speed for nic " + s, e);
+ return 0d;
+ }
+ }).sum() * 1024;
+ }
+
+ private Path getNicTxPath(String nic) {
+ return Paths.get(String.format("/sys/class/net/%s/statistics/tx_bytes", nic));
+ }
+
+ private Path getNicRxPath(String nic) {
+ return Paths.get(String.format("/sys/class/net/%s/statistics/rx_bytes", nic));
+ }
+
+ private double getTotalNicUsageRxKb(List nics) {
+ return nics.stream().mapToDouble(s -> {
+ try {
+ return Double.parseDouble(new String(Files.readAllBytes(getNicRxPath(s))));
+ } catch (IOException e) {
+ LOG.error("Failed to read rx_bytes for NIC " + s, e);
+ return 0d;
+ }
+ }).sum() * 8 / 1024;
+ }
+
+ private double getTotalNicUsageTxKb(List nics) {
+ return nics.stream().mapToDouble(s -> {
+ try {
+ return Double.parseDouble(new String(Files.readAllBytes(getNicTxPath(s))));
+ } catch (IOException e) {
+ LOG.error("Failed to read tx_bytes for NIC " + s, e);
+ return 0d;
+ }
+ }).sum() * 8 / 1024;
+ }
+
+ private class CpuStat {
+ private long totalTime;
+ private long usage;
+
+ CpuStat(long totalTime, long usage) {
+ this.totalTime = totalTime;
+ this.usage = usage;
+ }
+
+ long getTotalTime() {
+ return totalTime;
+ }
+
+ long getUsage() {
+ return usage;
+ }
+ }
+}
diff --git a/pulsar-broker/src/main/java/com/yahoo/pulsar/broker/loadbalance/impl/SimpleLoadManagerImpl.java b/pulsar-broker/src/main/java/com/yahoo/pulsar/broker/loadbalance/impl/SimpleLoadManagerImpl.java
index cbb9476d50fde403bc807bb167d0bcb48db79ff7..9a2443d564129137d66ba7f0bccc064376e762c3 100644
--- a/pulsar-broker/src/main/java/com/yahoo/pulsar/broker/loadbalance/impl/SimpleLoadManagerImpl.java
+++ b/pulsar-broker/src/main/java/com/yahoo/pulsar/broker/loadbalance/impl/SimpleLoadManagerImpl.java
@@ -37,6 +37,7 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import org.apache.bookkeeper.util.ZkUtils;
+import org.apache.commons.lang3.SystemUtils;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.ZooDefs.Ids;
@@ -176,8 +177,12 @@ public class SimpleLoadManagerImpl implements LoadManager, ZooKeeperCacheListene
this.realtimeAvgResourceQuota = new ResourceQuota();
placementStrategy = new WRRPlacementStrategy();
lastLoadReport = new LoadReport(pulsar.getWebServiceAddress(), pulsar.getWebServiceAddressTls(),
- pulsar.getBrokerServiceUrl(), pulsar.getBrokerServiceUrlTls());
- brokerHostUsage = new BrokerHostUsage(pulsar);
+ pulsar.getBrokerServiceUrl(), pulsar.getBrokerServiceUrlTls());
+ if (SystemUtils.IS_OS_LINUX) {
+ brokerHostUsage = new LinuxBrokerHostUsageImpl(pulsar);
+ } else {
+ brokerHostUsage = new GenericBrokerHostUsageImpl(pulsar);
+ }
loadReportCacheZk = new ZooKeeperDataCache(pulsar.getLocalZkCache()) {
@Override
public LoadReport deserialize(String key, byte[] content) throws Exception {
@@ -200,17 +205,17 @@ public class SimpleLoadManagerImpl implements LoadManager, ZooKeeperCacheListene
public PulsarAdmin load(String key) throws Exception {
// key - broker name already is valid URL, has prefix "http://"
return new PulsarAdmin(new URL(key), pulsar.getConfiguration().getBrokerClientAuthenticationPlugin(),
- pulsar.getConfiguration().getBrokerClientAuthenticationParameters());
+ pulsar.getConfiguration().getBrokerClientAuthenticationParameters());
}
});
int entryExpiryTime = (int) pulsar.getConfiguration().getLoadBalancerSheddingGracePeriodMinutes();
unloadedHotNamespaceCache = CacheBuilder.newBuilder().expireAfterWrite(entryExpiryTime, TimeUnit.MINUTES)
- .build(new CacheLoader() {
- @Override
- public Long load(String key) throws Exception {
- return System.currentTimeMillis();
- }
- });
+ .build(new CacheLoader() {
+ @Override
+ public Long load(String key) throws Exception {
+ return System.currentTimeMillis();
+ }
+ });
availableActiveBrokers = new ZooKeeperChildrenCache(pulsar.getLocalZkCache(), LOADBALANCE_BROKERS_ROOT);
availableActiveBrokers.registerListener(new ZooKeeperCacheListener>() {
@Override
@@ -233,7 +238,7 @@ public class SimpleLoadManagerImpl implements LoadManager, ZooKeeperCacheListene
if (pulsar.getZkClient().exists(LOADBALANCE_BROKERS_ROOT, false) == null) {
try {
ZkUtils.createFullPathOptimistic(pulsar.getZkClient(), LOADBALANCE_BROKERS_ROOT, new byte[0],
- Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
+ Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
} catch (KeeperException.NodeExistsException e) {
// ignore the exception, node might be present already
}
@@ -254,7 +259,7 @@ public class SimpleLoadManagerImpl implements LoadManager, ZooKeeperCacheListene
}
try {
ZkUtils.createFullPathOptimistic(pulsar.getZkClient(), brokerZnodePath,
- loadReportJson.getBytes(Charsets.UTF_8), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
+ loadReportJson.getBytes(Charsets.UTF_8), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
} catch (Exception e) {
// Catching excption here to print the right error message
log.error("Unable to create znode - [{}] for load balance on zookeeper ", brokerZnodePath, e);
@@ -268,11 +273,11 @@ public class SimpleLoadManagerImpl implements LoadManager, ZooKeeperCacheListene
this.realtimeAvgResourceQuota = pulsar.getLocalZkCacheService().getResourceQuotaCache().getDefaultQuota();
this.lastResourceQuotaUpdateTimestamp = System.currentTimeMillis();
this.realtimeCpuLoadFactor = getDynamicConfigurationDouble(
- LOADBALANCER_DYNAMIC_SETTING_LOAD_FACTOR_CPU_ZPATH, SETTING_NAME_LOAD_FACTOR_CPU,
- this.realtimeCpuLoadFactor);
+ LOADBALANCER_DYNAMIC_SETTING_LOAD_FACTOR_CPU_ZPATH, SETTING_NAME_LOAD_FACTOR_CPU,
+ this.realtimeCpuLoadFactor);
this.realtimeMemoryLoadFactor = getDynamicConfigurationDouble(
- LOADBALANCER_DYNAMIC_SETTING_LOAD_FACTOR_MEM_ZPATH, SETTING_NAME_LOAD_FACTOR_MEM,
- this.realtimeMemoryLoadFactor);
+ LOADBALANCER_DYNAMIC_SETTING_LOAD_FACTOR_MEM_ZPATH, SETTING_NAME_LOAD_FACTOR_MEM,
+ this.realtimeMemoryLoadFactor);
} catch (Exception e) {
log.error("Unable to create znode - [{}] for load balance on zookeeper ", brokerZnodePath, e);
throw new PulsarServerException(e);
@@ -301,7 +306,7 @@ public class SimpleLoadManagerImpl implements LoadManager, ZooKeeperCacheListene
pulsar.getZkClient().setData(zkPath, settingBytes, -1);
} else {
ZkUtils.createFullPathOptimistic(pulsar.getZkClient(), zkPath, settingBytes, Ids.OPEN_ACL_UNSAFE,
- CreateMode.PERSISTENT);
+ CreateMode.PERSISTENT);
}
} catch (Exception e) {
log.warn("Got exception when writing to ZooKeeper path [{}]:", zkPath, e);
@@ -346,7 +351,7 @@ public class SimpleLoadManagerImpl implements LoadManager, ZooKeeperCacheListene
private String getLoadBalancerPlacementStrategy() {
String strategy = this.getDynamicConfigurationFromZK(LOADBALANCER_DYNAMIC_SETTING_STRATEGY_ZPATH,
- SETTING_NAME_STRATEGY, pulsar.getConfiguration().getLoadBalancerPlacementStrategy());
+ SETTING_NAME_STRATEGY, pulsar.getConfiguration().getLoadBalancerPlacementStrategy());
if (!LOADBALANCER_STRATEGY_LLS.equals(strategy) && !LOADBALANCER_STRATEGY_RAND.equals(strategy)) {
strategy = LOADBALANCER_STRATEGY_RAND;
}
@@ -355,12 +360,12 @@ public class SimpleLoadManagerImpl implements LoadManager, ZooKeeperCacheListene
private double getCpuLoadFactorFromZK(double defaultValue) {
return getDynamicConfigurationDouble(LOADBALANCER_DYNAMIC_SETTING_LOAD_FACTOR_CPU_ZPATH,
- SETTING_NAME_LOAD_FACTOR_CPU, defaultValue);
+ SETTING_NAME_LOAD_FACTOR_CPU, defaultValue);
}
private double getMemoryLoadFactorFromZK(double defaultValue) {
return getDynamicConfigurationDouble(LOADBALANCER_DYNAMIC_SETTING_LOAD_FACTOR_MEM_ZPATH,
- SETTING_NAME_LOAD_FACTOR_MEM, defaultValue);
+ SETTING_NAME_LOAD_FACTOR_MEM, defaultValue);
}
@Override
@@ -371,26 +376,26 @@ public class SimpleLoadManagerImpl implements LoadManager, ZooKeeperCacheListene
private long getLoadBalancerBrokerUnderloadedThresholdPercentage() {
return (long) this.getDynamicConfigurationDouble(LOADBALANCER_DYNAMIC_SETTING_UNDERLOAD_THRESHOLD_ZPATH,
- SETTING_NAME_UNDERLOAD_THRESHOLD,
- pulsar.getConfiguration().getLoadBalancerBrokerUnderloadedThresholdPercentage());
+ SETTING_NAME_UNDERLOAD_THRESHOLD,
+ pulsar.getConfiguration().getLoadBalancerBrokerUnderloadedThresholdPercentage());
}
private long getLoadBalancerBrokerOverloadedThresholdPercentage() {
return (long) this.getDynamicConfigurationDouble(LOADBALANCER_DYNAMIC_SETTING_OVERLOAD_THRESHOLD_ZPATH,
- SETTING_NAME_OVERLOAD_THRESHOLD,
- pulsar.getConfiguration().getLoadBalancerBrokerOverloadedThresholdPercentage());
+ SETTING_NAME_OVERLOAD_THRESHOLD,
+ pulsar.getConfiguration().getLoadBalancerBrokerOverloadedThresholdPercentage());
}
private long getLoadBalancerBrokerComfortLoadThresholdPercentage() {
return (long) this.getDynamicConfigurationDouble(LOADBALANCER_DYNAMIC_SETTING_COMFORT_LOAD_THRESHOLD_ZPATH,
- SETTING_NAME_COMFORTLOAD_THRESHOLD,
- pulsar.getConfiguration().getLoadBalancerBrokerComfortLoadLevelPercentage());
+ SETTING_NAME_COMFORTLOAD_THRESHOLD,
+ pulsar.getConfiguration().getLoadBalancerBrokerComfortLoadLevelPercentage());
}
private boolean getLoadBalancerAutoBundleSplitEnabled() {
return this.getDynamicConfigurationBoolean(LOADBALANCER_DYNAMIC_SETTING_AUTO_BUNDLE_SPLIT_ENABLED,
- SETTING_NAME_AUTO_BUNDLE_SPLIT_ENABLED,
- pulsar.getConfiguration().getLoadBalancerAutoBundleSplitEnabled());
+ SETTING_NAME_AUTO_BUNDLE_SPLIT_ENABLED,
+ pulsar.getConfiguration().getLoadBalancerAutoBundleSplitEnabled());
}
/*
@@ -451,19 +456,19 @@ public class SimpleLoadManagerImpl implements LoadManager, ZooKeeperCacheListene
}
private ResourceQuota timeSmoothQuota(ResourceQuota oldQuota, double msgRateIn, double msgRateOut,
- double bandwidthIn, double bandwidthOut, double memory, long timePast) {
+ double bandwidthIn, double bandwidthOut, double memory, long timePast) {
if (oldQuota.getDynamic()) {
ResourceQuota newQuota = new ResourceQuota();
newQuota.setMsgRateIn(timeSmoothValue(oldQuota.getMsgRateIn(), msgRateIn, RESOURCE_QUOTA_MIN_MSGRATE_IN,
- RESOURCE_QUOTA_MAX_MSGRATE_IN, timePast));
+ RESOURCE_QUOTA_MAX_MSGRATE_IN, timePast));
newQuota.setMsgRateOut(timeSmoothValue(oldQuota.getMsgRateOut(), msgRateOut, RESOURCE_QUOTA_MIN_MSGRATE_OUT,
- RESOURCE_QUOTA_MAX_MSGRATE_OUT, timePast));
+ RESOURCE_QUOTA_MAX_MSGRATE_OUT, timePast));
newQuota.setBandwidthIn(timeSmoothValue(oldQuota.getBandwidthIn(), bandwidthIn,
- RESOURCE_QUOTA_MIN_BANDWIDTH_IN, RESOURCE_QUOTA_MAX_BANDWIDTH_IN, timePast));
+ RESOURCE_QUOTA_MIN_BANDWIDTH_IN, RESOURCE_QUOTA_MAX_BANDWIDTH_IN, timePast));
newQuota.setBandwidthOut(timeSmoothValue(oldQuota.getBandwidthOut(), bandwidthOut,
- RESOURCE_QUOTA_MIN_BANDWIDTH_OUT, RESOURCE_QUOTA_MAX_BANDWIDTH_OUT, timePast));
+ RESOURCE_QUOTA_MIN_BANDWIDTH_OUT, RESOURCE_QUOTA_MAX_BANDWIDTH_OUT, timePast));
newQuota.setMemory(timeSmoothValue(oldQuota.getMemory(), memory, RESOURCE_QUOTA_MIN_MEMORY,
- RESOURCE_QUOTA_MAX_MEMORY, timePast));
+ RESOURCE_QUOTA_MAX_MEMORY, timePast));
return newQuota;
} else {
return oldQuota;
@@ -500,7 +505,7 @@ public class SimpleLoadManagerImpl implements LoadManager, ZooKeeperCacheListene
totalBundles++;
NamespaceBundleStats stats = statsEntry.getValue();
totalMemGroups += (1
- + (stats.topics + stats.producerCount + stats.consumerCount) / memObjectGroupSize);
+ + (stats.topics + stats.producerCount + stats.consumerCount) / memObjectGroupSize);
totalBandwidthIn += stats.msgThroughputIn;
totalBandwidthOut += stats.msgThroughputOut;
}
@@ -517,18 +522,18 @@ public class SimpleLoadManagerImpl implements LoadManager, ZooKeeperCacheListene
this.lastResourceQuotaUpdateTimestamp = loadReportTimestamp;
if (totalMsgRate > 1000 && totalMemGroups > 30) {
this.realtimeCpuLoadFactor = timeSmoothValue(this.realtimeCpuLoadFactor, totalCpuUsage / totalMsgRate,
- RESOURCE_QUOTA_MIN_CPU_FACTOR, RESOURCE_QUOTA_MAX_CPU_FACTOR, timePast);
+ RESOURCE_QUOTA_MIN_CPU_FACTOR, RESOURCE_QUOTA_MAX_CPU_FACTOR, timePast);
this.realtimeMemoryLoadFactor = timeSmoothValue(this.realtimeMemoryLoadFactor,
- totalMemoryUsage / totalMemGroups, RESOURCE_QUOTA_MIN_MEM_FACTOR, RESOURCE_QUOTA_MAX_MEM_FACTOR,
- timePast);
+ totalMemoryUsage / totalMemGroups, RESOURCE_QUOTA_MIN_MEM_FACTOR, RESOURCE_QUOTA_MAX_MEM_FACTOR,
+ timePast);
}
// calculate average bundle
if (totalBundles > 30 && this.realtimeAvgResourceQuota.getDynamic()) {
ResourceQuota oldQuota = this.realtimeAvgResourceQuota;
ResourceQuota newQuota = timeSmoothQuota(oldQuota, totalMsgRateIn / totalBundles,
- totalMsgRateOut / totalBundles, totalBandwidthIn / totalBundles,
- totalBandwidthOut / totalBundles, totalMemoryUsage / totalBundles, timePast);
+ totalMsgRateOut / totalBundles, totalBandwidthIn / totalBundles,
+ totalBandwidthOut / totalBundles, totalMemoryUsage / totalBundles, timePast);
this.realtimeAvgResourceQuota = newQuota;
}
@@ -546,12 +551,12 @@ public class SimpleLoadManagerImpl implements LoadManager, ZooKeeperCacheListene
String bundle = statsEntry.getKey();
NamespaceBundleStats stats = statsEntry.getValue();
long memGroupCount = (1
- + (stats.topics + stats.producerCount + stats.consumerCount) / memObjectGroupSize);
+ + (stats.topics + stats.producerCount + stats.consumerCount) / memObjectGroupSize);
double newMemoryQuota = memGroupCount * this.realtimeMemoryLoadFactor;
ResourceQuota oldQuota = getResourceQuota(bundle);
ResourceQuota newQuota = timeSmoothQuota(oldQuota, stats.msgRateIn, stats.msgRateOut,
- stats.msgThroughputIn, stats.msgThroughputOut, newMemoryQuota, timePast);
+ stats.msgThroughputIn, stats.msgThroughputOut, newMemoryQuota, timePast);
newQuotas.put(bundle, newQuota);
}
}
@@ -562,20 +567,20 @@ public class SimpleLoadManagerImpl implements LoadManager, ZooKeeperCacheListene
private void compareAndWriteQuota(String bundle, ResourceQuota oldQuota, ResourceQuota newQuota) throws Exception {
boolean needUpdate = true;
if (!oldQuota.getDynamic() || (Math
- .abs(newQuota.getMsgRateIn() - oldQuota.getMsgRateIn()) < RESOURCE_QUOTA_MIN_MSGRATE_IN
- && Math.abs(newQuota.getMsgRateOut() - oldQuota.getMsgRateOut()) < RESOURCE_QUOTA_MIN_MSGRATE_OUT
- && Math.abs(newQuota.getBandwidthIn() - oldQuota.getBandwidthOut()) < RESOURCE_QUOTA_MIN_BANDWIDTH_IN
- && Math.abs(newQuota.getBandwidthOut() - oldQuota.getBandwidthOut()) < RESOURCE_QUOTA_MIN_BANDWIDTH_OUT
- && Math.abs(newQuota.getMemory() - oldQuota.getMemory()) < RESOURCE_QUOTA_MIN_MEMORY)) {
+ .abs(newQuota.getMsgRateIn() - oldQuota.getMsgRateIn()) < RESOURCE_QUOTA_MIN_MSGRATE_IN
+ && Math.abs(newQuota.getMsgRateOut() - oldQuota.getMsgRateOut()) < RESOURCE_QUOTA_MIN_MSGRATE_OUT
+ && Math.abs(newQuota.getBandwidthIn() - oldQuota.getBandwidthOut()) < RESOURCE_QUOTA_MIN_BANDWIDTH_IN
+ && Math.abs(newQuota.getBandwidthOut() - oldQuota.getBandwidthOut()) < RESOURCE_QUOTA_MIN_BANDWIDTH_OUT
+ && Math.abs(newQuota.getMemory() - oldQuota.getMemory()) < RESOURCE_QUOTA_MIN_MEMORY)) {
needUpdate = false;
}
if (needUpdate) {
if (log.isDebugEnabled()) {
log.debug(String.format(
- "Update quota %s - msgRateIn: %.1f, msgRateOut: %.1f, bandwidthIn: %.1f, bandwidthOut: %.1f, memory: %.1f",
- (bundle == null) ? "default" : bundle, newQuota.getMsgRateIn(), newQuota.getMsgRateOut(),
- newQuota.getBandwidthIn(), newQuota.getBandwidthOut(), newQuota.getMemory()));
+ "Update quota %s - msgRateIn: %.1f, msgRateOut: %.1f, bandwidthIn: %.1f, bandwidthOut: %.1f, memory: %.1f",
+ (bundle == null) ? "default" : bundle, newQuota.getMsgRateIn(), newQuota.getMsgRateOut(),
+ newQuota.getBandwidthIn(), newQuota.getBandwidthOut(), newQuota.getMemory()));
}
if (bundle == null) {
@@ -656,13 +661,13 @@ public class SimpleLoadManagerImpl implements LoadManager, ZooKeeperCacheListene
ResourceQuota allocatedQuota = getTotalAllocatedQuota(loadedBundles);
ResourceQuota preAllocatedQuota = getTotalAllocatedQuota(preAllocatedBundles);
ResourceUnitRanking ranking = new ResourceUnitRanking(loadReport.getSystemResourceUsage(),
- loadedBundles, allocatedQuota, preAllocatedBundles, preAllocatedQuota);
+ loadedBundles, allocatedQuota, preAllocatedBundles, preAllocatedQuota);
newResourceUnitRankings.put(resourceUnit, ranking);
// generated sorted ranking
double loadPercentage = ranking.getEstimatedLoadPercentage();
long maxCapacity = ranking.estimateMaxCapacity(
- pulsar.getLocalZkCacheService().getResourceQuotaCache().getDefaultQuota());
+ pulsar.getLocalZkCacheService().getResourceQuotaCache().getDefaultQuota());
long finalRank = 0;
if (strategy.equals(LOADBALANCER_STRATEGY_LLS)) {
finalRank = (long) loadPercentage;
@@ -689,7 +694,7 @@ public class SimpleLoadManagerImpl implements LoadManager, ZooKeeperCacheListene
}
} else {
log.info("Leader broker[{}] No ResourceUnits to rank this run, Using Old Ranking",
- pulsar.getWebServiceAddress());
+ pulsar.getWebServiceAddress());
}
}
@@ -834,15 +839,15 @@ public class SimpleLoadManagerImpl implements LoadManager, ZooKeeperCacheListene
}
private Multimap getFinalCandidatesWithPolicy(NamespaceName namespace,
- Multimap primaries, Multimap shared) {
+ Multimap primaries, Multimap shared) {
Multimap finalCandidates = TreeMultimap.create();
// if not enough primary then it should be union of primaries and secondaries
finalCandidates.putAll(primaries);
if (policies.shouldFailoverToSecondaries(namespace, primaries.size())) {
log.debug(
- "Not enough of primaries [{}] available for namespace - [{}], "
- + "adding shared [{}] as possible candidate owners",
- primaries.size(), namespace.toString(), shared.size());
+ "Not enough of primaries [{}] available for namespace - [{}], "
+ + "adding shared [{}] as possible candidate owners",
+ primaries.size(), namespace.toString(), shared.size());
finalCandidates.putAll(shared);
}
return finalCandidates;
@@ -856,7 +861,7 @@ public class SimpleLoadManagerImpl implements LoadManager, ZooKeeperCacheListene
}
private Multimap getFinalCandidates(ServiceUnitId serviceUnit,
- Map> availableBrokers) {
+ Map> availableBrokers) {
// need multimap or at least set of RUs
Multimap matchedPrimaries = TreeMultimap.create();
Multimap matchedShared = TreeMultimap.create();
@@ -869,7 +874,7 @@ public class SimpleLoadManagerImpl implements LoadManager, ZooKeeperCacheListene
for (Map.Entry> entry : availableBrokers.entrySet()) {
for (ResourceUnit ru : entry.getValue()) {
log.debug("Considering Resource Unit [{}] with Rank [{}] for serviceUnit [{}]", ru.getResourceId(),
- entry.getKey(), serviceUnit);
+ entry.getKey(), serviceUnit);
URL brokerUrl = null;
try {
brokerUrl = new URL(String.format(ru.getResourceId()));
@@ -884,22 +889,22 @@ public class SimpleLoadManagerImpl implements LoadManager, ZooKeeperCacheListene
matchedPrimaries.put(entry.getKey(), ru);
if (log.isDebugEnabled()) {
log.debug(
- "Added Primary Broker - [{}] as possible Candidates for"
- + " namespace - [{}] with policies",
- brokerUrl.getHost(), namespace.toString());
+ "Added Primary Broker - [{}] as possible Candidates for"
+ + " namespace - [{}] with policies",
+ brokerUrl.getHost(), namespace.toString());
}
} else if (policies.isSharedBroker(brokerUrl.getHost())) {
matchedShared.put(entry.getKey(), ru);
if (log.isDebugEnabled()) {
log.debug(
- "Added Shared Broker - [{}] as possible "
- + "Candidates for namespace - [{}] with policies",
- brokerUrl.getHost(), namespace.toString());
+ "Added Shared Broker - [{}] as possible "
+ + "Candidates for namespace - [{}] with policies",
+ brokerUrl.getHost(), namespace.toString());
}
} else {
if (log.isDebugEnabled()) {
log.debug("Skipping Broker - [{}] not primary broker and not shared"
- + " for namespace - [{}] ", brokerUrl.getHost(), namespace.toString());
+ + " for namespace - [{}] ", brokerUrl.getHost(), namespace.toString());
}
}
@@ -907,7 +912,7 @@ public class SimpleLoadManagerImpl implements LoadManager, ZooKeeperCacheListene
if (policies.isSharedBroker(brokerUrl.getHost())) {
matchedShared.put(entry.getKey(), ru);
log.debug("Added Shared Broker - [{}] as possible Candidates for namespace - [{}]",
- brokerUrl.getHost(), namespace.toString());
+ brokerUrl.getHost(), namespace.toString());
}
}
}
@@ -916,9 +921,9 @@ public class SimpleLoadManagerImpl implements LoadManager, ZooKeeperCacheListene
return getFinalCandidatesWithPolicy(namespace, matchedPrimaries, matchedShared);
} else {
log.debug(
- "Policies not present for namespace - [{}] so only "
- + "considering shared [{}] brokers for possible owner",
- namespace.toString(), matchedShared.size());
+ "Policies not present for namespace - [{}] so only "
+ + "considering shared [{}] brokers for possible owner",
+ namespace.toString(), matchedShared.size());
return getFinalCandidatesNoPolicy(matchedShared);
}
}
@@ -946,7 +951,7 @@ public class SimpleLoadManagerImpl implements LoadManager, ZooKeeperCacheListene
availableBrokers = Maps.newTreeMap();
for (String broker : activeBrokers) {
ResourceUnit resourceUnit = new SimpleResourceUnit(String.format("http://%s", broker),
- new PulsarResourceDescription());
+ new PulsarResourceDescription());
availableBrokers.computeIfAbsent(0L, key -> Sets.newTreeSet()).add(resourceUnit);
}
log.info("Choosing at random from broker list: [{}]", availableBrokers.values());
@@ -955,7 +960,7 @@ public class SimpleLoadManagerImpl implements LoadManager, ZooKeeperCacheListene
}
private ResourceUnit getLeastLoadedBroker(ServiceUnitId serviceUnit,
- Map> availableBrokers) {
+ Map> availableBrokers) {
ResourceUnit selectedBroker = null;
Multimap finalCandidates = getFinalCandidates(serviceUnit, availableBrokers);
// Remove candidates that point to inactive brokers
@@ -982,7 +987,7 @@ public class SimpleLoadManagerImpl implements LoadManager, ZooKeeperCacheListene
selectedBroker = placementStrategy.findBrokerForPlacement(finalCandidates);
}
log.debug("Selected : [{}] for ServiceUnit : [{}]", selectedBroker.getResourceId(),
- serviceUnit.getNamespaceObject().toString());
+ serviceUnit.getNamespaceObject().toString());
return selectedBroker;
} else {
// No available broker found
@@ -1012,9 +1017,9 @@ public class SimpleLoadManagerImpl implements LoadManager, ZooKeeperCacheListene
try {
String key = String.format("%s/%s", LOADBALANCE_BROKERS_ROOT, broker);
LoadReport lr = loadReportCacheZk.get(key)
- .orElseThrow(() -> new KeeperException.NoNodeException());
+ .orElseThrow(() -> new KeeperException.NoNodeException());
ResourceUnit ru = new SimpleResourceUnit(String.format("http://%s", lr.getName()),
- fromLoadReport(lr));
+ fromLoadReport(lr));
this.currentLoadReports.put(ru, lr);
} catch (Exception e) {
log.warn("Error reading load report from Cache for broker - [{}], [{}]", broker, e);
@@ -1030,14 +1035,14 @@ public class SimpleLoadManagerImpl implements LoadManager, ZooKeeperCacheListene
public static boolean isAboveLoadLevel(SystemResourceUsage usage, float thresholdPercentage) {
return (usage.bandwidthOut.percentUsage() > thresholdPercentage
- || usage.bandwidthIn.percentUsage() > thresholdPercentage
- || usage.cpu.percentUsage() > thresholdPercentage || usage.memory.percentUsage() > thresholdPercentage);
+ || usage.bandwidthIn.percentUsage() > thresholdPercentage
+ || usage.cpu.percentUsage() > thresholdPercentage || usage.memory.percentUsage() > thresholdPercentage);
}
public static boolean isBelowLoadLevel(SystemResourceUsage usage, float thresholdPercentage) {
return (usage.bandwidthOut.percentUsage() < thresholdPercentage
- && usage.bandwidthIn.percentUsage() < thresholdPercentage
- && usage.cpu.percentUsage() < thresholdPercentage && usage.memory.percentUsage() < thresholdPercentage);
+ && usage.bandwidthIn.percentUsage() < thresholdPercentage
+ && usage.cpu.percentUsage() < thresholdPercentage && usage.memory.percentUsage() < thresholdPercentage);
}
private static long getRealtimeJvmHeapUsageMBytes() {
@@ -1060,22 +1065,18 @@ public class SimpleLoadManagerImpl implements LoadManager, ZooKeeperCacheListene
}
private SystemResourceUsage getSystemResourceUsage() throws IOException {
- SystemResourceUsage systemResourceUsage = new SystemResourceUsage();
- if (isNotEmpty(pulsar.getConfiguration().getLoadBalancerHostUsageScriptPath())) {
- systemResourceUsage = ObjectMapperFactory.getThreadLocal().readValue(brokerHostUsage.getBrokerHostUsage(),
- SystemResourceUsage.class);
-
- // Override System memory usage and limit with JVM heap usage and limit
- long maxHeapMemoryInBytes = Runtime.getRuntime().maxMemory();
- long memoryUsageInMBytes = getAverageJvmHeapUsageMBytes();
- systemResourceUsage.memory.usage = (double) memoryUsageInMBytes;
- systemResourceUsage.memory.limit = (double) (maxHeapMemoryInBytes) / MBytes;
-
- // Collect JVM direct memory
- systemResourceUsage.directMemory.usage = (double) (sun.misc.SharedSecrets.getJavaNioAccess()
- .getDirectBufferPool().getMemoryUsed() / MBytes);
- systemResourceUsage.directMemory.limit = (double) (sun.misc.VM.maxDirectMemory() / MBytes);
- }
+ SystemResourceUsage systemResourceUsage = brokerHostUsage.getBrokerHostUsage();
+
+ // Override System memory usage and limit with JVM heap usage and limit
+ long maxHeapMemoryInBytes = Runtime.getRuntime().maxMemory();
+ long memoryUsageInMBytes = getAverageJvmHeapUsageMBytes();
+ systemResourceUsage.memory.usage = (double) memoryUsageInMBytes;
+ systemResourceUsage.memory.limit = (double) (maxHeapMemoryInBytes) / MBytes;
+
+ // Collect JVM direct memory
+ systemResourceUsage.directMemory.usage = (double) (sun.misc.SharedSecrets.getJavaNioAccess()
+ .getDirectBufferPool().getMemoryUsed() / MBytes);
+ systemResourceUsage.directMemory.limit = (double) (sun.misc.VM.maxDirectMemory() / MBytes);
return systemResourceUsage;
}
@@ -1089,13 +1090,13 @@ public class SimpleLoadManagerImpl implements LoadManager, ZooKeeperCacheListene
try {
LoadReport loadReport = new LoadReport(pulsar.getWebServiceAddress(), pulsar.getWebServiceAddressTls(),
- pulsar.getBrokerServiceUrl(), pulsar.getBrokerServiceUrlTls());
+ pulsar.getBrokerServiceUrl(), pulsar.getBrokerServiceUrlTls());
loadReport.setName(String.format("%s:%s", pulsar.getAdvertisedAddress(), pulsar.getConfiguration().getWebServicePort()));
SystemResourceUsage systemResourceUsage = this.getSystemResourceUsage();
loadReport.setOverLoaded(
- isAboveLoadLevel(systemResourceUsage, this.getLoadBalancerBrokerOverloadedThresholdPercentage()));
+ isAboveLoadLevel(systemResourceUsage, this.getLoadBalancerBrokerOverloadedThresholdPercentage()));
loadReport.setUnderLoaded(
- isBelowLoadLevel(systemResourceUsage, this.getLoadBalancerBrokerUnderloadedThresholdPercentage()));
+ isBelowLoadLevel(systemResourceUsage, this.getLoadBalancerBrokerUnderloadedThresholdPercentage()));
loadReport.setSystemResourceUsage(systemResourceUsage);
loadReport.setBundleStats(pulsar.getBrokerService().getBundleStats());
@@ -1144,49 +1145,49 @@ public class SimpleLoadManagerImpl implements LoadManager, ZooKeeperCacheListene
long newBundleCount = pulsar.getBrokerService().getNumberOfNamespaceBundles();
long bundleCountChange = Math.abs(oldBundleCount - newBundleCount);
long maxCapacity = ResourceUnitRanking.calculateBrokerMaxCapacity(
- lastLoadReport.getSystemResourceUsage(),
- pulsar.getLocalZkCacheService().getResourceQuotaCache().getDefaultQuota());
+ lastLoadReport.getSystemResourceUsage(),
+ pulsar.getLocalZkCacheService().getResourceQuotaCache().getDefaultQuota());
double bundlePercentageChange = (maxCapacity > 0) ? (bundleCountChange * 100 / maxCapacity) : 0;
if (newBundleCount < oldBundleCount || bundlePercentageChange > pulsar.getConfiguration()
- .getLoadBalancerReportUpdateThresholdPercentage()) {
+ .getLoadBalancerReportUpdateThresholdPercentage()) {
needUpdate = true;
}
// check resource usage comparing with last LoadReport
if (!needUpdate && timestampNow - this.lastResourceUsageTimestamp > TimeUnit.MINUTES
- .toMillis(pulsar.getConfiguration().getLoadBalancerHostUsageCheckIntervalMinutes())) {
+ .toMillis(pulsar.getConfiguration().getLoadBalancerHostUsageCheckIntervalMinutes())) {
SystemResourceUsage oldUsage = lastLoadReport.getSystemResourceUsage();
SystemResourceUsage newUsage = this.getSystemResourceUsage();
this.lastResourceUsageTimestamp = timestampNow;
// calculate percentage of change
double cpuChange = (newUsage.cpu.limit > 0)
- ? ((newUsage.cpu.usage - oldUsage.cpu.usage) * 100 / newUsage.cpu.limit) : 0;
+ ? ((newUsage.cpu.usage - oldUsage.cpu.usage) * 100 / newUsage.cpu.limit) : 0;
double memChange = (newUsage.memory.limit > 0)
- ? ((newUsage.memory.usage - oldUsage.memory.usage) * 100 / newUsage.memory.limit) : 0;
+ ? ((newUsage.memory.usage - oldUsage.memory.usage) * 100 / newUsage.memory.limit) : 0;
double directMemChange = (newUsage.directMemory.limit > 0)
- ? ((newUsage.directMemory.usage - oldUsage.directMemory.usage) * 100
- / newUsage.directMemory.limit)
- : 0;
+ ? ((newUsage.directMemory.usage - oldUsage.directMemory.usage) * 100
+ / newUsage.directMemory.limit)
+ : 0;
double bandwidthOutChange = (newUsage.bandwidthOut.limit > 0)
- ? ((newUsage.bandwidthOut.usage - oldUsage.bandwidthOut.usage) * 100
- / newUsage.bandwidthOut.limit)
- : 0;
+ ? ((newUsage.bandwidthOut.usage - oldUsage.bandwidthOut.usage) * 100
+ / newUsage.bandwidthOut.limit)
+ : 0;
double bandwidthInChange = (newUsage.bandwidthIn.limit > 0)
- ? ((newUsage.bandwidthIn.usage - oldUsage.bandwidthIn.usage) * 100
- / newUsage.bandwidthIn.limit)
- : 0;
+ ? ((newUsage.bandwidthIn.usage - oldUsage.bandwidthIn.usage) * 100
+ / newUsage.bandwidthIn.limit)
+ : 0;
long resourceChange = (long) Math.min(100.0,
- Math.max(Math.abs(cpuChange),
- Math.max(Math.abs(directMemChange), Math.max(Math.abs(memChange),
- Math.max(Math.abs(bandwidthOutChange), Math.abs(bandwidthInChange))))));
+ Math.max(Math.abs(cpuChange),
+ Math.max(Math.abs(directMemChange), Math.max(Math.abs(memChange),
+ Math.max(Math.abs(bandwidthOutChange), Math.abs(bandwidthInChange))))));
if (resourceChange > pulsar.getConfiguration().getLoadBalancerReportUpdateThresholdPercentage()) {
needUpdate = true;
log.info("LoadReport update triggered by change on resource usage, detal ({}).",
- String.format(
- "cpu: %.1f%%, mem: %.1f%%, directMemory: %.1f%%, bandwidthIn: %.1f%%, bandwidthOut: %.1f%%)",
- cpuChange, memChange, directMemChange, bandwidthInChange, bandwidthOutChange));
+ String.format(
+ "cpu: %.1f%%, mem: %.1f%%, directMemory: %.1f%%, bandwidthIn: %.1f%%, bandwidthOut: %.1f%%)",
+ cpuChange, memChange, directMemChange, bandwidthInChange, bandwidthOutChange));
}
}
}
@@ -1195,7 +1196,7 @@ public class SimpleLoadManagerImpl implements LoadManager, ZooKeeperCacheListene
if (needUpdate) {
LoadReport lr = generateLoadReport();
pulsar.getZkClient().setData(brokerZnodePath, ObjectMapperFactory.getThreadLocal().writeValueAsBytes(lr),
- -1);
+ -1);
this.lastLoadReport = lr;
this.lastResourceUsageTimestamp = lr.getTimestamp();
}
@@ -1247,10 +1248,10 @@ public class SimpleLoadManagerImpl implements LoadManager, ZooKeeperCacheListene
boolean unloadDisabledInLoadShedding = false;
try {
unloadDisabledInLoadShedding = pulsar.getGlobalZkCache()
- .exists(AdminResource.LOAD_SHEDDING_UNLOAD_DISABLED_FLAG_PATH);
+ .exists(AdminResource.LOAD_SHEDDING_UNLOAD_DISABLED_FLAG_PATH);
} catch (Exception e) {
log.warn("Unable to fetch contents of [{}] from global zookeeper",
- AdminResource.LOAD_SHEDDING_UNLOAD_DISABLED_FLAG_PATH, e);
+ AdminResource.LOAD_SHEDDING_UNLOAD_DISABLED_FLAG_PATH, e);
}
return unloadDisabledInLoadShedding;
}
@@ -1264,17 +1265,17 @@ public class SimpleLoadManagerImpl implements LoadManager, ZooKeeperCacheListene
if (!isUnloadDisabledInLoadShedding()) {
log.info("Unloading namespace {} from overloaded broker {}", bundleName, brokerName);
adminCache.get(brokerName).namespaces().unloadNamespaceBundle(
- getNamespaceNameFromBundleName(bundleName), getBundleRangeFromBundleName(bundleName));
+ getNamespaceNameFromBundleName(bundleName), getBundleRangeFromBundleName(bundleName));
log.info("Successfully unloaded namespace {} from broker {}", bundleName, brokerName);
} else {
log.info("DRY RUN: Unload in Load Shedding is disabled. Namespace {} would have been "
- + "unloaded from overloaded broker {} otherwise.", bundleName, brokerName);
+ + "unloaded from overloaded broker {} otherwise.", bundleName, brokerName);
}
unloadedHotNamespaceCache.put(bundleName, System.currentTimeMillis());
} else {
// we can't unload this namespace so move to next one
log.info("Can't unload Namespace {} because it was unloaded last at {} and unload interval has "
- + "not exceeded.", bundleName, LocalDateTime.now());
+ + "not exceeded.", bundleName, LocalDateTime.now());
}
} catch (Exception e) {
log.warn("ERROR failed to unload the bundle {} from overloaded broker {}", bundleName, brokerName, e);
@@ -1287,7 +1288,7 @@ public class SimpleLoadManagerImpl implements LoadManager, ZooKeeperCacheListene
long overloadThreshold = this.getLoadBalancerBrokerOverloadedThresholdPercentage();
long comfortLoadLevel = this.getLoadBalancerBrokerComfortLoadThresholdPercentage();
log.info("Running load shedding task as leader broker, overload threshold {}, comfort loadlevel {}",
- overloadThreshold, comfortLoadLevel);
+ overloadThreshold, comfortLoadLevel);
// overloadedRU --> bundleName
Map namespaceBundlesToBeUnloaded = new HashMap<>();
synchronized (currentLoadReports) {
@@ -1302,9 +1303,9 @@ public class SimpleLoadManagerImpl implements LoadManager, ZooKeeperCacheListene
// can't unload one namespace, just issue a warning message
String bundleName = lr.getBundleStats().keySet().iterator().next();
log.warn(
- "HIGH USAGE WARNING : Sole namespace bundle {} is overloading broker {}. "
- + "No Load Shedding will be done on this broker",
- bundleName, overloadedRU.getResourceId());
+ "HIGH USAGE WARNING : Sole namespace bundle {} is overloading broker {}. "
+ + "No Load Shedding will be done on this broker",
+ bundleName, overloadedRU.getResourceId());
continue;
}
for (Map.Entry bundleStat : bundleStats.entrySet()) {
@@ -1313,14 +1314,14 @@ public class SimpleLoadManagerImpl implements LoadManager, ZooKeeperCacheListene
// We need at least one underloaded RU from list of candidates that can host this bundle
if (isBrokerAvailableForRebalancing(bundleStat.getKey(), comfortLoadLevel)) {
log.info(
- "Namespace bundle {} will be unloaded from overloaded broker {}, bundle stats (topics: {}, producers {}, "
- + "consumers {}, bandwidthIn {}, bandwidthOut {})",
- bundleName, overloadedRU.getResourceId(), stats.topics, stats.producerCount,
- stats.consumerCount, stats.msgThroughputIn, stats.msgThroughputOut);
+ "Namespace bundle {} will be unloaded from overloaded broker {}, bundle stats (topics: {}, producers {}, "
+ + "consumers {}, bandwidthIn {}, bandwidthOut {})",
+ bundleName, overloadedRU.getResourceId(), stats.topics, stats.producerCount,
+ stats.consumerCount, stats.msgThroughputIn, stats.msgThroughputOut);
namespaceBundlesToBeUnloaded.put(overloadedRU, bundleName);
} else {
log.info("Unable to shed load from broker {}, no brokers with enough capacity available "
- + "for re-balancing {}", overloadedRU.getResourceId(), bundleName);
+ + "for re-balancing {}", overloadedRU.getResourceId(), bundleName);
}
break;
}
@@ -1342,8 +1343,8 @@ public class SimpleLoadManagerImpl implements LoadManager, ZooKeeperCacheListene
long maxBundleBandwidth = pulsar.getConfiguration().getLoadBalancerNamespaceBundleMaxBandwidthMbytes() * MBytes;
log.info(
- "Running namespace bundle split with thresholds: topics {}, sessions {}, msgRate {}, bandwidth {}, maxBundles {}",
- maxBundleTopics, maxBundleSessions, maxBundleMsgRate, maxBundleBandwidth, maxBundleCount);
+ "Running namespace bundle split with thresholds: topics {}, sessions {}, msgRate {}, bandwidth {}, maxBundles {}",
+ maxBundleTopics, maxBundleSessions, maxBundleMsgRate, maxBundleBandwidth, maxBundleCount);
if (this.lastLoadReport == null || this.lastLoadReport.getBundleStats() == null) {
return;
}
@@ -1360,7 +1361,7 @@ public class SimpleLoadManagerImpl implements LoadManager, ZooKeeperCacheListene
boolean needSplit = false;
if (stats.topics > maxBundleTopics || totalSessions > maxBundleSessions || totalMsgRate > maxBundleMsgRate
- || totalBandwidth > maxBundleBandwidth) {
+ || totalBandwidth > maxBundleBandwidth) {
if (stats.topics <= 1) {
log.info("Unable to split hot namespace bundle {} since there is only one topic.", bundleName);
} else {
@@ -1368,7 +1369,7 @@ public class SimpleLoadManagerImpl implements LoadManager, ZooKeeperCacheListene
int numBundles = pulsar.getNamespaceService().getBundleCount(namespaceName);
if (numBundles >= maxBundleCount) {
log.info("Unable to split hot namespace bundle {} since the namespace has too many bundles.",
- bundleName);
+ bundleName);
} else {
needSplit = true;
}
@@ -1378,13 +1379,13 @@ public class SimpleLoadManagerImpl implements LoadManager, ZooKeeperCacheListene
if (needSplit) {
if (this.getLoadBalancerAutoBundleSplitEnabled()) {
log.info(
- "Will split hot namespace bundle {}, topics {}, producers+consumers {}, msgRate in+out {}, bandwidth in+out {}",
- bundleName, stats.topics, totalSessions, totalMsgRate, totalBandwidth);
+ "Will split hot namespace bundle {}, topics {}, producers+consumers {}, msgRate in+out {}, bandwidth in+out {}",
+ bundleName, stats.topics, totalSessions, totalMsgRate, totalBandwidth);
bundlesToBeSplit.add(bundleName);
} else {
log.info(
- "DRY RUN - split hot namespace bundle {}, topics {}, producers+consumers {}, msgRate in+out {}, bandwidth in+out {}",
- bundleName, stats.topics, totalSessions, totalMsgRate, totalBandwidth);
+ "DRY RUN - split hot namespace bundle {}, topics {}, producers+consumers {}, msgRate in+out {}, bandwidth in+out {}",
+ bundleName, stats.topics, totalSessions, totalMsgRate, totalBandwidth);
}
}
}
@@ -1393,7 +1394,7 @@ public class SimpleLoadManagerImpl implements LoadManager, ZooKeeperCacheListene
for (String bundleName : bundlesToBeSplit) {
try {
pulsar.getAdminClient().namespaces().splitNamespaceBundle(
- getNamespaceNameFromBundleName(bundleName), getBundleRangeFromBundleName(bundleName));
+ getNamespaceNameFromBundleName(bundleName), getBundleRangeFromBundleName(bundleName));
log.info("Successfully split namespace bundle {}", bundleName);
} catch (Exception e) {
log.error("Failed to split namespace bundle {}", bundleName, e);
diff --git a/pulsar-broker/src/test/java/com/yahoo/pulsar/PulsarBrokerStarterTest.java b/pulsar-broker/src/test/java/com/yahoo/pulsar/PulsarBrokerStarterTest.java
index 27ab45081bdc991441b62a1ab93a22faf92f6590..bae5dd4f1d95a10388e6681023dd5da26d90d360 100644
--- a/pulsar-broker/src/test/java/com/yahoo/pulsar/PulsarBrokerStarterTest.java
+++ b/pulsar-broker/src/test/java/com/yahoo/pulsar/PulsarBrokerStarterTest.java
@@ -168,7 +168,6 @@ public class PulsarBrokerStarterTest {
printWriter.println("managedLedgerDefaultAckQuorum=1");
printWriter.println("loadBalancerEnabled=false");
- printWriter.println("loadBalancerHostUsageScriptPath=/usr/bin/my_pulsar-broker-host-usage");
printWriter.println("loadBalancerHostUsageCheckIntervalMinutes=4");
printWriter.println("loadBalancerReportUpdateThresholdPercentage=15");
printWriter.println("loadBalancerReportUpdateMaxIntervalMinutes=20");
@@ -186,7 +185,6 @@ public class PulsarBrokerStarterTest {
Assert.assertTrue(ServiceConfiguration.class.isInstance(returnValue));
ServiceConfiguration serviceConfig = (ServiceConfiguration) returnValue;
Assert.assertEquals(serviceConfig.isLoadBalancerEnabled(), false);
- Assert.assertEquals(serviceConfig.getLoadBalancerHostUsageScriptPath(), "/usr/bin/my_pulsar-broker-host-usage");
Assert.assertEquals(serviceConfig.getLoadBalancerHostUsageCheckIntervalMinutes(), 4);
Assert.assertEquals(serviceConfig.getLoadBalancerReportUpdateThresholdPercentage(), 15);
Assert.assertEquals(serviceConfig.getLoadBalancerReportUpdateMaxIntervalMinutes(), 20);
diff --git a/pulsar-broker/src/test/java/com/yahoo/pulsar/broker/loadbalance/LoadBalancerTest.java b/pulsar-broker/src/test/java/com/yahoo/pulsar/broker/loadbalance/LoadBalancerTest.java
index c0e173f070ae43cb040b7f45aae63e8fd2ab7799..45f591c7e624257a435c596686ce5a43e61d8a21 100644
--- a/pulsar-broker/src/test/java/com/yahoo/pulsar/broker/loadbalance/LoadBalancerTest.java
+++ b/pulsar-broker/src/test/java/com/yahoo/pulsar/broker/loadbalance/LoadBalancerTest.java
@@ -211,9 +211,11 @@ public class LoadBalancerTest {
printSortedRanking(sortedRanking);
// all brokers have same rank to it would be 0 --> set-of-all-the-brokers
- assertEquals(sortedRanking.get().size(), 1);
- assertTrue(sortedRanking.get().get(0L) != null);
- assertEquals(sortedRanking.get().get(0L).size(), BROKER_COUNT);
+ int brokerCount = 0;
+ for (Map.Entry> entry : sortedRanking.get().entrySet()) {
+ brokerCount += entry.getValue().size();
+ }
+ assertEquals(brokerCount, BROKER_COUNT);
DestinationName fqdn = DestinationName.get("persistent://pulsar/use/primary-ns/test-topic");
ResourceUnit found = pulsarServices[i].getLoadManager()
.getLeastLoaded(pulsarServices[i].getNamespaceService().getBundle(fqdn));
diff --git a/pulsar-broker/src/test/java/com/yahoo/pulsar/broker/loadbalance/SimpleLoadManagerImplTest.java b/pulsar-broker/src/test/java/com/yahoo/pulsar/broker/loadbalance/SimpleLoadManagerImplTest.java
index 68b644cd6ab8c180c1cc11c75aa9f5dd3cd8e92e..564837b41e7a5008188b63af4f7211786af91dfa 100644
--- a/pulsar-broker/src/test/java/com/yahoo/pulsar/broker/loadbalance/SimpleLoadManagerImplTest.java
+++ b/pulsar-broker/src/test/java/com/yahoo/pulsar/broker/loadbalance/SimpleLoadManagerImplTest.java
@@ -27,6 +27,7 @@ import static org.testng.Assert.assertNull;
import static org.testng.Assert.assertTrue;
import static org.testng.Assert.fail;
+import org.apache.commons.lang3.SystemUtils;
import java.io.IOException;
import java.lang.reflect.Field;
import java.net.URL;
@@ -41,6 +42,7 @@ import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
+import com.yahoo.pulsar.broker.loadbalance.impl.*;
import org.apache.bookkeeper.test.PortManager;
import org.apache.bookkeeper.util.ZkUtils;
import org.apache.zookeeper.CreateMode;
@@ -57,12 +59,6 @@ import com.google.common.collect.Sets;
import com.yahoo.pulsar.broker.PulsarService;
import com.yahoo.pulsar.broker.ServiceConfiguration;
import com.yahoo.pulsar.broker.admin.AdminResource;
-import com.yahoo.pulsar.broker.loadbalance.impl.PulsarLoadReportImpl;
-import com.yahoo.pulsar.broker.loadbalance.impl.PulsarResourceDescription;
-import com.yahoo.pulsar.broker.loadbalance.impl.ResourceAvailabilityRanker;
-import com.yahoo.pulsar.broker.loadbalance.impl.SimpleLoadCalculatorImpl;
-import com.yahoo.pulsar.broker.loadbalance.impl.SimpleLoadManagerImpl;
-import com.yahoo.pulsar.broker.loadbalance.impl.SimpleResourceUnit;
import com.yahoo.pulsar.client.admin.BrokerStats;
import com.yahoo.pulsar.client.admin.PulsarAdmin;
import com.yahoo.pulsar.client.api.Authentication;
@@ -443,14 +439,14 @@ public class SimpleLoadManagerImplTest {
@Test
public void testBrokerHostUsage() {
- when(pulsar1.getConfiguration().getLoadBalancerHostUsageScriptPath()).thenReturn("usageScript");
- BrokerHostUsage brokerUsage = new BrokerHostUsage(pulsar1);
- try {
- brokerUsage.getBrokerHostUsage();
- fail();
- } catch (IOException e) {
- // Ok
+ BrokerHostUsage brokerUsage;
+ if (SystemUtils.IS_OS_LINUX) {
+ brokerUsage = new LinuxBrokerHostUsageImpl(pulsar1);
+ } else {
+ brokerUsage = new GenericBrokerHostUsageImpl(pulsar1);
}
+ brokerUsage.getBrokerHostUsage();
+ // Ok
}
@Test
diff --git a/pulsar-broker/src/test/resources/configurations/pulsar_broker_test.conf b/pulsar-broker/src/test/resources/configurations/pulsar_broker_test.conf
index e31a429e6a6cb3480b607836f6b0ded0bfd6d13a..413160a54235b9488b15f8d7ed3799375a2e4f32 100644
--- a/pulsar-broker/src/test/resources/configurations/pulsar_broker_test.conf
+++ b/pulsar-broker/src/test/resources/configurations/pulsar_broker_test.conf
@@ -52,7 +52,6 @@ managedLedgerCursorRolloverTimeInSeconds=14400
loadBalancerEnabled=false
loadBalancerReportUpdateThresholdPercentage=10
loadBalancerReportUpdateMaxIntervalMinutes=15
-loadBalancerHostUsageScriptPath=/usr/bin/pulsar-broker-host-usage
loadBalancerHostUsageCheckIntervalMinutes=1
loadBalancerSheddingIntervalMinutes=30
loadBalancerSheddingGracePeriodMinutes=30
diff --git a/pulsar-broker/src/test/resources/test-script-pulsar-broker-host-usage b/pulsar-broker/src/test/resources/test-script-pulsar-broker-host-usage
deleted file mode 100644
index a53466f290ffd46cdb8c1066e3e783e7ad41cfaf..0000000000000000000000000000000000000000
--- a/pulsar-broker/src/test/resources/test-script-pulsar-broker-host-usage
+++ /dev/null
@@ -1,7 +0,0 @@
-#!/bin/bash
-
-# response to test the load-report API : BrokerHostUsage execs a script that returns system resource usage;
-# Since the original script is system dependent and also requires /home/y location, this script is the dummy one
-# returning a fake/mock response, so we can test the load-report API end-to-end including the path executing the script
-
-echo "{\"bandwidthIn\":{\"usage\": 0,\"limit\": 0},\"bandwidthOut\":{\"usage\": 0,\"limit\": 0},\"memory\":{\"usage\":3240,\"limit\": 12829},\"cpu\": {\"usage\": 10.0,\"limit\": 16}, \"threads\":{\"usage\":10,\"limit\":100}}"