Introduce new load manager API

bin/pulsar-perf

@@ -137,6 +137,14 @@ if [ "$COMMAND" == "produce" ]; then
     exec $JAVA $OPTS com.yahoo.pulsar.testclient.PerformanceProducer --conf-file $PULSAR_PERFTEST_CONF "$@"
 elif [ "$COMMAND" == "consume" ]; then
     exec $JAVA $OPTS com.yahoo.pulsar.testclient.PerformanceConsumer --conf-file $PULSAR_PERFTEST_CONF "$@"
+elif [ "$COMMAND" == "monitor" ]; then
+    exec $JAVA $OPTS com.yahoo.pulsar.testclient.BrokerMonitor "$@"
+elif [ "$COMMAND" == "new-monitor" ]; then
+    exec $JAVA $OPTS com.yahoo.pulsar.testclient.NewBrokerMonitor "$@"
+elif [ "$COMMAND" == "simulation-server" ]; then
+    exec $JAVA $OPTS com.yahoo.pulsar.testclient.LoadSimulationServer "$@"
+elif [ "$COMMAND" == "simulation-controller" ]; then
+    exec $JAVA $OPTS com.yahoo.pulsar.testclient.LoadSimulationController "$@"
 elif [ "$COMMAND" == "help" ]; then
     pulsar_help;
 else

conf/broker.conf

@@ -293,3 +293,30 @@ keepAliveIntervalSeconds=30
 
 # How often broker checks for inactive topics to be deleted (topics with no subscriptions and no one connected)
 brokerServicePurgeInactiveFrequencyInSeconds=60
+
+# Number of samples to use for short term time window
+numShortSamples=50
+
+# Number of samples to use for long term time window
+numLongSamples=1000
+
+# How often in seconds to update the broker data
+brokerDataUpdateIntervalSeconds=60
+
+# How often in seconds to update the bundle data
+bundleDataUpdateIntervalSeconds=60
+
+# Default throughput in to assume for new bundles
+defaultMsgThroughputIn=50000
+
+# Default throughput out to assume for new bundles
+defaultMsgThroughputOut=50000
+
+# Default message rate in to assume for new bundles
+defaultMsgRateIn=50
+
+# Default message rate out to assume for new bundles
+defaultMsgRateOut=50
+
+# Name of load manager to use
+loadManagerName=SimpleLoadManager

pulsar-broker-common/src/main/java/com/yahoo/pulsar/broker/ServiceConfiguration.java

@@ -30,7 +30,7 @@
  * Pulsar service configuration object.
  *
  */
-public class ServiceConfiguration implements PulsarConfiguration{
+public class ServiceConfiguration implements PulsarConfiguration {
 
     /***** --- pulsar configuration --- ****/
     // Zookeeper quorum connection string
@@ -196,10 +196,14 @@ public class ServiceConfiguration implements PulsarConfiguration{
     private boolean loadBalancerEnabled = false;
     // load placement strategy
     private String loadBalancerPlacementStrategy = "weightedRandomSelection"; // weighted random selection
+    // load placement secondary strategy (used to silently test an alternate strategy)
+    private String loadBalancerSecondaryStrategy = null;
+    // are all bundle placement operations forwarded to a lead broker
+    private boolean loadBalancerIsCentralized = false;
     // Percentage of change to trigger load report update
     private int loadBalancerReportUpdateThresholdPercentage = 10;
     // maximum interval to update load report
-    private int loadBalancerReportUpdateMaxIntervalMinutes = 15;
+    private int loadBalancerReportUpdateMaxIntervalMinutes = 1;
     // Frequency of report to collect
     private int loadBalancerHostUsageCheckIntervalMinutes = 1;
     // Load shedding interval. Broker periodically checks whether some traffic
@@ -724,6 +728,22 @@ public String getLoadBalancerPlacementStrategy() {
         return this.loadBalancerPlacementStrategy;
     }
 
+    public void setLoadBalancerSecondaryStrategy(String secondaryStrategy) {
+        this.loadBalancerSecondaryStrategy = secondaryStrategy;
+    }
+
+    public String getLoadBalancerSecondaryStrategy() {
+        return this.loadBalancerSecondaryStrategy;
+    }
+
+    public void setLoadBalancerIsCentralized(boolean isCentralized) {
+        this.loadBalancerIsCentralized = isCentralized;
+    }
+
+    public boolean getLoadBalancerIsCentralized() {
+        return this.loadBalancerIsCentralized;
+    }
+
     public int getLoadBalancerReportUpdateThresholdPercentage() {
         return loadBalancerReportUpdateThresholdPercentage;
     }
@@ -941,4 +961,117 @@ public String getReplicatorPrefix() {
     public void setReplicatorPrefix(String replicatorPrefix) {
         this.replicatorPrefix = replicatorPrefix;
     }
+
+
+    // Configurations for new load manager API
+
+    // Number of samples to use for short term time window
+    private int numShortSamples = 50;
+
+    // Number of samples to use for long term time window
+    private int numLongSamples = 1000;
+
+    // How often in seconds to update the broker data
+    private long brokerDataUpdateIntervalSeconds = 60;
+
+    // How often in seconds to update the bundle data
+    private long bundleDataUpdateIntervalSeconds = 60;
+
+    // Default throughput in to assume for new bundles
+    private double defaultMsgThroughputIn = 50000;
+
+    // Default throughput out to assume for new bundles
+    private double defaultMsgThroughputOut = 50000;
+
+    // Default message rate in to assume for new bundles
+    private double defaultMsgRateIn = 50;
+
+    // Default message rate out to assume for new bundles
+    private double defaultMsgRateOut = 50;
+
+    // Name of load manager to use
+    private String loadManagerName = "SimpleLoadManager";
+
+    // Name of placement strategy to use for new loadbalancer API.
+    private String newPlacementStrategyName = "LeastLongTermMessageRate";
+
+    public int getNumShortSamples() {
+        return numShortSamples;
+    }
+
+    public void setNumShortSamples(int numShortSamples) {
+        this.numShortSamples = numShortSamples;
+    }
+
+    public int getNumLongSamples() {
+        return numLongSamples;
+    }
+
+    public void setNumLongSamples(int numLongSamples) {
+        this.numLongSamples = numLongSamples;
+    }
+
+    public long getBrokerDataUpdateIntervalSeconds() {
+        return brokerDataUpdateIntervalSeconds;
+    }
+
+    public void setBrokerDataUpdateIntervalSeconds(long brokerDataUpdateIntervalSeconds) {
+        this.brokerDataUpdateIntervalSeconds = brokerDataUpdateIntervalSeconds;
+    }
+
+    public long getBundleDataUpdateIntervalSeconds() {
+        return bundleDataUpdateIntervalSeconds;
+    }
+
+    public void setBundleDataUpdateIntervalSeconds(long bundleDataUpdateIntervalSeconds) {
+        this.bundleDataUpdateIntervalSeconds = bundleDataUpdateIntervalSeconds;
+    }
+
+    public double getDefaultMsgThroughputIn() {
+        return defaultMsgThroughputIn;
+    }
+
+    public void setDefaultMsgThroughputIn(double defaultMsgThroughputIn) {
+        this.defaultMsgThroughputIn = defaultMsgThroughputIn;
+    }
+
+    public double getDefaultMsgThroughputOut() {
+        return defaultMsgThroughputOut;
+    }
+
+    public void setDefaultMsgThroughputOut(double defaultMsgThroughputOut) {
+        this.defaultMsgThroughputOut = defaultMsgThroughputOut;
+    }
+
+    public double getDefaultMsgRateIn() {
+        return defaultMsgRateIn;
+    }
+
+    public void setDefaultMsgRateIn(double defaultMsgRateIn) {
+        this.defaultMsgRateIn = defaultMsgRateIn;
+    }
+
+    public double getDefaultMsgRateOut() {
+        return defaultMsgRateOut;
+    }
+
+    public void setDefaultMsgRateOut(double defaultMsgRateOut) {
+        this.defaultMsgRateOut = defaultMsgRateOut;
+    }
+
+    public String getLoadManagerName() {
+        return loadManagerName;
+    }
+
+    public void setLoadManagerName(String loadManagerName) {
+        this.loadManagerName = loadManagerName;
+    }
+
+    public String getNewPlacementStrategyName() {
+        return newPlacementStrategyName;
+    }
+
+    public void setNewPlacementStrategyName(String newPlacementStrategyName) {
+        this.newPlacementStrategyName = newPlacementStrategyName;
+    }
 }

pulsar-broker/src/main/java/com/yahoo/pulsar/broker/BrokerData.java

@@ -0,0 +1,51 @@
+package com.yahoo.pulsar.broker;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * Data class containing three components comprising all the data available for the leader broker about other brokers:
+ * - The local broker data which is written to ZooKeeper by each individual broker (LocalBrokerData).
+ * - The time average bundle data which is written to ZooKeeper by the leader broker (TimeAverageBrokerData).
+ * - The preallocated bundles which are not written to ZooKeeper but are maintained by the leader broker
+ *   (Map<String, BundleData>).
+ */
+public class BrokerData {
+    private LocalBrokerData localData;
+    private TimeAverageBrokerData timeAverageData;
+    private Map<String, BundleData> preallocatedBundleData;
+
+    /**
+     * Initialize this BrokerData using the most recent local data.
+     * @param localData The data local for the broker.
+     */
+    public BrokerData(final LocalBrokerData localData) {
+        this.localData = localData;
+        timeAverageData = new TimeAverageBrokerData();
+        preallocatedBundleData = new ConcurrentHashMap<>();
+    }
+
+    public LocalBrokerData getLocalData() {
+        return localData;
+    }
+
+    public void setLocalData(LocalBrokerData localData) {
+        this.localData = localData;
+    }
+
+    public TimeAverageBrokerData getTimeAverageData() {
+        return timeAverageData;
+    }
+
+    public void setTimeAverageData(TimeAverageBrokerData timeAverageData) {
+        this.timeAverageData = timeAverageData;
+    }
+
+    public Map<String, BundleData> getPreallocatedBundleData() {
+        return preallocatedBundleData;
+    }
+
+    public void setPreallocatedBundleData(Map<String, BundleData> preallocatedBundleData) {
+        this.preallocatedBundleData = preallocatedBundleData;
+    }
+}

pulsar-broker/src/main/java/com/yahoo/pulsar/broker/BundleData.java

@@ -0,0 +1,66 @@
+package com.yahoo.pulsar.broker;
+
+import com.yahoo.pulsar.common.policies.data.loadbalancer.NamespaceBundleStats;
+
+/**
+ * Data class comprising the short term and long term historical data for this bundle.
+ */
+public class BundleData extends JSONWritable {
+    // Short term data for this bundle. The time frame of this data is determined by the number of short term samples
+    // and the bundle update period.
+    private TimeAverageMessageData shortTermData;
+
+    // Long term data for this bundle. The time frame of this data is determined by the number of long term samples
+    // and the bundle update period.
+    private TimeAverageMessageData longTermData;
+
+    // For JSON only.
+    public BundleData(){}
+
+    /**
+     * Initialize the bundle data.
+     * @param numShortSamples Number of short term samples to use.
+     * @param numLongSamples Number of long term samples to use.
+     */
+    public BundleData(final int numShortSamples, final int numLongSamples) {
+        shortTermData = new TimeAverageMessageData(numShortSamples);
+        longTermData = new TimeAverageMessageData(numLongSamples);
+    }
+
+    /**
+     * Initialize this bundle data and have its histories default to the given stats before the first sample is
+     * received.
+     * @param numShortSamples Number of short term samples to use.
+     * @param numLongSamples Number of long term samples to use.
+     * @param defaultStats The stats to default to before the first sample is received.
+     */
+    public BundleData(final int numShortSamples, final int numLongSamples, final NamespaceBundleStats defaultStats) {
+        shortTermData = new TimeAverageMessageData(numShortSamples, defaultStats);
+        longTermData = new TimeAverageMessageData(numLongSamples, defaultStats);
+    }
+
+    /**
+     * Update the historical data for this bundle.
+     * @param newSample The bundle stats to update this data with.
+     */
+    public void update(final NamespaceBundleStats newSample) {
+        shortTermData.update(newSample);
+        longTermData.update(newSample);
+    }
+
+    public TimeAverageMessageData getShortTermData() {
+        return shortTermData;
+    }
+
+    public void setShortTermData(TimeAverageMessageData shortTermData) {
+        this.shortTermData = shortTermData;
+    }
+
+    public TimeAverageMessageData getLongTermData() {
+        return longTermData;
+    }
+
+    public void setLongTermData(TimeAverageMessageData longTermData) {
+        this.longTermData = longTermData;
+    }
+}

pulsar-broker/src/main/java/com/yahoo/pulsar/broker/JSONWritable.java

@@ -0,0 +1,31 @@
+package com.yahoo.pulsar.broker;
+
+import com.fasterxml.jackson.annotation.JsonIgnore;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.yahoo.pulsar.common.util.ObjectMapperFactory;
+
+/**
+ * Helper class used to conveniently convert a data class to a JSON.
+ */
+public class JSONWritable {
+
+    /**
+     * Get the JSON of this object as a byte[].
+     * @return A byte[] of this object's JSON.
+     * @throws JsonProcessingException
+     */
+    @JsonIgnore
+    public byte[] getJsonBytes() throws JsonProcessingException {
+        return ObjectMapperFactory.getThreadLocal().writeValueAsBytes(this);
+    }
+
+    /**
+     * Get the JSON of this object as a String.
+     * @return A String of this object's JSON.
+     * @throws JsonProcessingException
+     */
+    @JsonIgnore
+    public String getJsonString() throws JsonProcessingException {
+        return ObjectMapperFactory.getThreadLocal().writeValueAsString(this);
+    }
+}