Introduce new load manager API

bin/pulsar-perf

@@ -73,8 +73,13 @@ pulsar_help() {
     cat <<EOF
 Usage: pulsar <command>
 where command is one of:
-    produce             Run a producer
-    consume             Run a consumer
+    produce                 Run a producer
+    consume                 Run a consumer
+    simple-monitor          Continuously receive broker data when using SimpleLoadManagerImpl
+    modular-monitor         Continuously receive broker data when using ModularLoadManagerImpl
+    simulation-server       Run a simulation server acting as a Pulsar client
+    simulation-controller   Run a simulation controller to give commands to servers
+
     help                This help message
 
 or command is the full name of a class with a defined main() method.
@@ -137,6 +142,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" == "simple-monitor" ]; then
+    exec $JAVA $OPTS com.yahoo.pulsar.testclient.SimpleLoadManagerBrokerMonitor "$@"
+elif [ "$COMMAND" == "modular-monitor" ]; then
+    exec $JAVA $OPTS com.yahoo.pulsar.testclient.ModularLoadManagerBrokerMonitor "$@"
+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,33 @@ 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
+loadManagerNumberOfSamplesShortTermWindow=50
+
+# Number of samples to use for long term time window
+loadManagerNumberOfSamplesLongTermWindow=1000
+
+# How often in seconds to update the broker data
+loadManagerBrokerDataUpdateIntervalInSeconds=60
+
+# How often in seconds to update the bundle data
+loadManagerBundleDataUpdateIntervalInSeconds=60
+
+# Default throughput in to assume for new bundles
+loadManagerDefaultMessageThroughputIn=50000
+
+# Default throughput out to assume for new bundles
+loadManagerDefaultMessageThroughputOut=50000
+
+# Default message rate in to assume for new bundles
+loadManagerDefaultMessageRateIn=50
+
+# Default message rate out to assume for new bundles
+loadManagerDefaultMessageRateOut=50
+
+# Name of load manager to use
+loadManagerClassName=com.yahoo.pulsar.broker.loadbalance.impl.SimpleLoadManagerImpl
+
+# Name of placement strategy to use for new load manager API
+loadManagerPlacementStrategyClassName=com.yahoo.pulsar.broker.loadbalance.impl.LeastLongTermMessageRate

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,6 +196,10 @@ 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
@@ -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,118 @@ 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 loadManagerNumberOfSamplesShortTermWindow = 50;
+
+    // Number of samples to use for long term time window
+    private int loadManagerNumberOfSamplesLongTermWindow = 1000;
+
+    // How often in seconds to update the broker data
+    private long loadManagerBrokerDataUpdateIntervalInSeconds = 60;
+
+    // How often in seconds to update the bundle data
+    private long loadManagerBundleDataUpdateIntervalInSeconds = 60;
+
+    // Default throughput in to assume for new bundles
+    private double loadManagerDefaultMessageThroughputIn = 50000;
+
+    // Default throughput out to assume for new bundles
+    private double loadManagerDefaultMessageThroughputOut = 50000;
+
+    // Default message rate in to assume for new bundles
+    private double loadManagerDefaultMessageRateIn = 50;
+
+    // Default message rate out to assume for new bundles
+    private double loadManagerDefaultMessageRateOut = 50;
+
+    // Name of load manager to use
+    private String loadManagerClassName = "com.yahoo.pulsar.broker.loadbalance.impl.SimpleLoadManagerImpl";
+
+    // Name of placement strategy to use for new loadbalancer API.
+    private String modularPlacementStrategyClassName =
+            "com.yahoo.pulsar.broker.loadbalance.impl.LeastLongTermMessageRate";
+
+    public int getLoadManagerNumberOfSamplesShortTermWindow() {
+        return loadManagerNumberOfSamplesShortTermWindow;
+    }
+
+    public void setLoadManagerNumberOfSamplesShortTermWindow(int loadManagerNumberOfSamplesShortTermWindow) {
+        this.loadManagerNumberOfSamplesShortTermWindow = loadManagerNumberOfSamplesShortTermWindow;
+    }
+
+    public int getLoadManagerNumberOfSamplesLongTermWindow() {
+        return loadManagerNumberOfSamplesLongTermWindow;
+    }
+
+    public void setLoadManagerNumberOfSamplesLongTermWindow(int loadManagerNumberOfSamplesLongTermWindow) {
+        this.loadManagerNumberOfSamplesLongTermWindow = loadManagerNumberOfSamplesLongTermWindow;
+    }
+
+    public long getLoadManagerBrokerDataUpdateIntervalInSeconds() {
+        return loadManagerBrokerDataUpdateIntervalInSeconds;
+    }
+
+    public void setLoadManagerBrokerDataUpdateIntervalInSeconds(long loadManagerBrokerDataUpdateIntervalInSeconds) {
+        this.loadManagerBrokerDataUpdateIntervalInSeconds = loadManagerBrokerDataUpdateIntervalInSeconds;
+    }
+
+    public long getLoadManagerBundleDataUpdateIntervalInSeconds() {
+        return loadManagerBundleDataUpdateIntervalInSeconds;
+    }
+
+    public void setLoadManagerBundleDataUpdateIntervalInSeconds(long loadManagerBundleDataUpdateIntervalInSeconds) {
+        this.loadManagerBundleDataUpdateIntervalInSeconds = loadManagerBundleDataUpdateIntervalInSeconds;
+    }
+
+    public double getLoadManagerDefaultMessageThroughputIn() {
+        return loadManagerDefaultMessageThroughputIn;
+    }
+
+    public void setLoadManagerDefaultMessageThroughputIn(double loadManagerDefaultMessageThroughputIn) {
+        this.loadManagerDefaultMessageThroughputIn = loadManagerDefaultMessageThroughputIn;
+    }
+
+    public double getLoadManagerDefaultMessageThroughputOut() {
+        return loadManagerDefaultMessageThroughputOut;
+    }
+
+    public void setLoadManagerDefaultMessageThroughputOut(double loadManagerDefaultMessageThroughputOut) {
+        this.loadManagerDefaultMessageThroughputOut = loadManagerDefaultMessageThroughputOut;
+    }
+
+    public double getLoadManagerDefaultMessageRateIn() {
+        return loadManagerDefaultMessageRateIn;
+    }
+
+    public void setLoadManagerDefaultMessageRateIn(double loadManagerDefaultMessageRateIn) {
+        this.loadManagerDefaultMessageRateIn = loadManagerDefaultMessageRateIn;
+    }
+
+    public double getLoadManagerDefaultMessageRateOut() {
+        return loadManagerDefaultMessageRateOut;
+    }
+
+    public void setLoadManagerDefaultMessageRateOut(double loadManagerDefaultMessageRateOut) {
+        this.loadManagerDefaultMessageRateOut = loadManagerDefaultMessageRateOut;
+    }
+
+    public String getLoadManagerClassName() {
+        return loadManagerClassName;
+    }
+
+    public void setLoadManagerClassName(String loadManagerClassName) {
+        this.loadManagerClassName = loadManagerClassName;
+    }
+
+    public String getModularPlacementStrategyClassName() {
+        return modularPlacementStrategyClassName;
+    }
+
+    public void setModularPlacementStrategyClassName(String modularPlacementStrategyClassName) {
+        this.modularPlacementStrategyClassName = modularPlacementStrategyClassName;
+    }
 }

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

@@ -0,0 +1,52 @@
+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,78 @@
+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,33 @@
+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);
+    }
+}