Introduce new load manager API

bin/pulsar-perf

@@ -73,9 +73,14 @@ pulsar_help() {
     cat <<EOF
 Usage: pulsar <command>
 where command is one of:
-    produce             Run a producer
-    consume             Run a consumer
-    help                This help message
+    produce                        Run a producer
+    consume                        Run a consumer
+    monitor-simple-load-manager    Continuously receive broker data when using SimpleLoadManagerImpl
+    monitor-modular-load-manager   Continuously receive broker data when using ModularLoadManagerImpl
+    simulation-client              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" == "monitor-simple-load-manager" ]; then
+    exec $JAVA $OPTS com.yahoo.pulsar.testclient.SimpleLoadManagerBrokerMonitor "$@"
+elif [ "$COMMAND" == "monitor-modular-load-manager" ]; then
+    exec $JAVA $OPTS com.yahoo.pulsar.testclient.ModularLoadManagerBrokerMonitor "$@"
+elif [ "$COMMAND" == "simulation-client" ]; then
+    exec $JAVA $OPTS com.yahoo.pulsar.testclient.LoadSimulationClient "$@"
+elif [ "$COMMAND" == "simulation-controller" ]; then
+    exec $JAVA $OPTS com.yahoo.pulsar.testclient.LoadSimulationController "$@"
 elif [ "$COMMAND" == "help" ]; then
     pulsar_help;
 else

conf/broker.conf

@@ -296,3 +296,6 @@ keepAliveIntervalSeconds=30
 
 # How often broker checks for inactive topics to be deleted (topics with no subscriptions and no one connected)
 brokerServicePurgeInactiveFrequencyInSeconds=60
+
+# Name of load manager to use
+loadManagerClassName=com.yahoo.pulsar.broker.loadbalance.impl.SimpleLoadManagerImpl

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
@@ -255,6 +255,9 @@ public class ServiceConfiguration implements PulsarConfiguration{
     private int brokerServicePurgeInactiveFrequencyInSeconds = 60;
     private List<String> bootstrapNamespaces = new ArrayList<String>();
     private Properties properties = new Properties();
+    // Name of load manager to use
+    @FieldContext(dynamic = true)
+    private String loadManagerClassName = "com.yahoo.pulsar.broker.loadbalance.impl.SimpleLoadManagerImpl";
 
     public String getZookeeperServers() {
         return zookeeperServers;
@@ -952,4 +955,12 @@ public String getReplicatorPrefix() {
     public void setReplicatorPrefix(String replicatorPrefix) {
         this.replicatorPrefix = replicatorPrefix;
     }
+
+    public String getLoadManagerClassName() {
+        return loadManagerClassName;
+    }
+
+    public void setLoadManagerClassName(String loadManagerClassName) {
+        this.loadManagerClassName = loadManagerClassName;
+    }
 }

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

@@ -0,0 +1,67 @@
+/**
+ * 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;
+
+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,93 @@
+/**
+ * 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;
+
+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,48 @@
+/**
+ * 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;
+
+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);
+    }
+}