1818 */
1919package org .apache .pulsar .testclient ;
2020
21+ import static org .apache .pulsar .broker .loadbalance .extensions .ExtensibleLoadManagerImpl .BROKER_LOAD_DATA_STORE_TOPIC ;
2122import com .beust .jcommander .JCommander ;
2223import com .beust .jcommander .Parameter ;
2324import com .beust .jcommander .ParameterException ;
3132import java .util .Map ;
3233import java .util .Set ;
3334import java .util .concurrent .ConcurrentHashMap ;
35+ import java .util .concurrent .TimeUnit ;
36+ import org .apache .pulsar .broker .loadbalance .extensions .data .BrokerLoadData ;
3437import 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 ;
3542import org .apache .pulsar .policies .data .loadbalancer .LoadReport ;
3643import org .apache .pulsar .policies .data .loadbalancer .LocalBrokerData ;
3744import 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 ("\n Broker 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