Visitar URL original
[improve][broker] PIP-192 Added --extensions option in BrokerMonitor … · apache/pulsar@579f22c · GitHub
Skip to content

Commit 579f22c

Browse files
authored
[improve][broker] PIP-192 Added --extensions option in BrokerMonitor (#19654)
1 parent a12f554 commit 579f22c

1 file changed

Lines changed: 88 additions & 6 deletions

File tree

‎pulsar-testclient/src/main/java/org/apache/pulsar/testclient/BrokerMonitor.java‎

Lines changed: 88 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@
1818
*/
1919
package org.apache.pulsar.testclient;
2020

21+
import static org.apache.pulsar.broker.loadbalance.extensions.ExtensibleLoadManagerImpl.BROKER_LOAD_DATA_STORE_TOPIC;
2122
import com.beust.jcommander.JCommander;
2223
import com.beust.jcommander.Parameter;
2324
import com.beust.jcommander.ParameterException;
@@ -31,7 +32,13 @@
3132
import java.util.Map;
3233
import java.util.Set;
3334
import java.util.concurrent.ConcurrentHashMap;
35+
import java.util.concurrent.TimeUnit;
36+
import org.apache.pulsar.broker.loadbalance.extensions.data.BrokerLoadData;
3437
import org.apache.pulsar.broker.loadbalance.impl.ModularLoadManagerImpl;
38+
import org.apache.pulsar.client.api.PulsarClient;
39+
import org.apache.pulsar.client.api.Schema;
40+
import org.apache.pulsar.client.api.SizeUnit;
41+
import org.apache.pulsar.client.api.TableView;
3542
import org.apache.pulsar.policies.data.loadbalancer.LoadReport;
3643
import org.apache.pulsar.policies.data.loadbalancer.LocalBrokerData;
3744
import org.apache.pulsar.policies.data.loadbalancer.ResourceUsage;
@@ -54,7 +61,7 @@ public class BrokerMonitor {
5461
private static final String BROKER_ROOT = "/loadbalance/brokers";
5562
private static final int ZOOKEEPER_TIMEOUT_MILLIS = 30000;
5663
private static final int GLOBAL_STATS_PRINT_PERIOD_MILLIS = 60000;
57-
private final ZooKeeper zkClient;
64+
private ZooKeeper zkClient;
5865
private static final Gson gson = new Gson();
5966

6067
// Fields common for message rows.
@@ -77,7 +84,7 @@ public class BrokerMonitor {
7784
private static final Object[] ALLOC_MESSAGE_ROW = makeMessageRow("ALLOC MSG");
7885
private static final Object[] GLOBAL_HEADER = { "BROKER", "BUNDLE", "MSG/S", "LONG/S", "KB/S", "MAX %" };
7986

80-
private final Map<String, Object> loadData;
87+
private Map<String, Object> loadData;
8188

8289
private static final FixedColumnLengthTableMaker localTableMaker = new FixedColumnLengthTableMaker();
8390
static {
@@ -434,8 +441,11 @@ private static class Arguments {
434441
@Parameter(names = { "-h", "--help" }, description = "Help message", help = true)
435442
boolean help;
436443

437-
@Parameter(names = { "--connect-string" }, description = "Zookeeper connect string", required = true)
444+
@Parameter(names = { "--connect-string" }, description = "Zookeeper or broker connect string", required = true)
438445
public String connectString = null;
446+
447+
@Parameter(names = { "--extensions" }, description = "true to monitor Load Balance Extensions.")
448+
boolean extensions = false;
439449
}
440450

441451
/**
@@ -464,6 +474,71 @@ public void start() {
464474
}
465475
}
466476

477+
private TableView<BrokerLoadData> brokerLoadDataTableView;
478+
479+
private BrokerMonitor(String brokerServiceUrl) {
480+
try {
481+
PulsarClient client = PulsarClient.builder()
482+
.memoryLimit(0, SizeUnit.BYTES)
483+
.serviceUrl(brokerServiceUrl)
484+
.connectionsPerBroker(4)
485+
.ioThreads(Runtime.getRuntime().availableProcessors())
486+
.statsInterval(0, TimeUnit.SECONDS)
487+
.build();
488+
this.brokerLoadDataTableView = client
489+
.newTableView(Schema.JSON(BrokerLoadData.class))
490+
.topic(BROKER_LOAD_DATA_STORE_TOPIC).create();
491+
} catch (Throwable e) {
492+
log.info("Failed to start BrokerMonitor", e);
493+
throw new RuntimeException(e);
494+
}
495+
}
496+
497+
private synchronized void printBrokerLoadData(final String broker, final BrokerLoadData brokerLoadData) {
498+
499+
// Initialize the constant rows.
500+
final Object[][] rows = new Object[6][];
501+
rows[0] = SYSTEM_ROW;
502+
rows[2] = COUNT_ROW;
503+
rows[4] = LATEST_ROW;
504+
505+
// First column is a label, so start at the second column at index 1.
506+
// System row.
507+
rows[1] = new Object[SYSTEM_ROW.length];
508+
initRow(rows[1], brokerLoadData.getCpu().percentUsage(), brokerLoadData.getMemory().percentUsage(),
509+
brokerLoadData.getDirectMemory().percentUsage(), brokerLoadData.getBandwidthIn().percentUsage(),
510+
brokerLoadData.getBandwidthOut().percentUsage(), brokerLoadData.getMaxResourceUsage() * 100);
511+
512+
// Count row.
513+
rows[3] = new Object[COUNT_ROW.length];
514+
initRow(rows[3], null, brokerLoadData.getBundleCount(),
515+
null, null,
516+
null, null);
517+
518+
// Latest message data row.
519+
rows[5] = new Object[LATEST_ROW.length];
520+
initMessageRow(rows[5], brokerLoadData.getMsgRateIn(), brokerLoadData.getMsgRateOut(),
521+
brokerLoadData.getMsgThroughputIn(), brokerLoadData.getMsgThroughputOut());
522+
523+
final String table = localTableMaker.make(rows);
524+
log.info("\nBroker Data for {}:\n{}\n", broker, table);
525+
}
526+
527+
private synchronized void printBrokerLoadDataStore() {
528+
brokerLoadDataTableView.forEach(this::printBrokerLoadData);
529+
}
530+
531+
private void startBrokerLoadDataStoreMonitor() {
532+
try {
533+
while (true) {
534+
Thread.sleep(GLOBAL_STATS_PRINT_PERIOD_MILLIS);
535+
printBrokerLoadDataStore();
536+
}
537+
} catch (Exception ex) {
538+
throw new RuntimeException(ex);
539+
}
540+
}
541+
467542
/**
468543
* Run a monitor from command line arguments.
469544
*
@@ -481,8 +556,15 @@ public static void main(String[] args) throws Exception {
481556
jc.usage();
482557
PerfClientUtils.exit(1);
483558
}
484-
final ZooKeeper zkClient = new ZooKeeper(arguments.connectString, ZOOKEEPER_TIMEOUT_MILLIS, null);
485-
final BrokerMonitor monitor = new BrokerMonitor(zkClient);
486-
monitor.start();
559+
560+
561+
if (arguments.extensions) {
562+
final BrokerMonitor monitor = new BrokerMonitor(arguments.connectString);
563+
monitor.startBrokerLoadDataStoreMonitor();
564+
} else {
565+
final ZooKeeper zkClient = new ZooKeeper(arguments.connectString, ZOOKEEPER_TIMEOUT_MILLIS, null);
566+
final BrokerMonitor monitor = new BrokerMonitor(zkClient);
567+
monitor.start();
568+
}
487569
}
488570
}

0 commit comments

Comments
 (0)