[improve][broker] PIP-192 Added broker and top-bundles load reporters

apache · Feb 9, 2023 · cc6e9a3 · cc6e9a3
1 parent fd3ce8b
commit cc6e9a3
Show file tree

Hide file tree

Showing 15 changed files with 770 additions and 56 deletions.
diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
@@ -2514,6 +2514,17 @@ The delayed message index bucket time step(in seconds) in per bucket snapshot se
     )
     private long loadBalancerBrokerLoadDataTTLInSeconds = 1800;
 
+    @FieldContext(
+            dynamic = true,
+            category = CATEGORY_LOAD_BALANCER,
+            doc = "Percentage of bundles to compute topK bundle load data from each broker. "
+                    + "The load balancer distributes bundles across brokers, "
+                    + "based on topK bundle load data and other broker load data."
+                    + "The bigger value will increase the overhead of reporting many bundles in load data. "
+                    + "(only used in load balancer extension logics)"
+    )
+    private int loadBalancerBundleLoadReportPercentage = 10;
+
     /**** --- Replication. --- ****/
     @FieldContext(
         category = CATEGORY_REPLICATION,

diff --git a/.../main/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImpl.java b/.../main/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImpl.java
@@ -26,6 +26,7 @@
 import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
 import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.pulsar.broker.PulsarServerException;
@@ -39,6 +40,8 @@
 import org.apache.pulsar.broker.loadbalance.extensions.data.TopBundlesLoadData;
 import org.apache.pulsar.broker.loadbalance.extensions.filter.BrokerFilter;
 import org.apache.pulsar.broker.loadbalance.extensions.filter.BrokerVersionFilter;
+import org.apache.pulsar.broker.loadbalance.extensions.reporter.BrokerLoadDataReporter;
+import org.apache.pulsar.broker.loadbalance.extensions.reporter.TopBundleLoadDataReporter;
 import org.apache.pulsar.broker.loadbalance.extensions.store.LoadDataStore;
 import org.apache.pulsar.broker.loadbalance.extensions.store.LoadDataStoreException;
 import org.apache.pulsar.broker.loadbalance.extensions.store.LoadDataStoreFactory;
@@ -84,6 +87,13 @@ public class ExtensibleLoadManagerImpl implements ExtensibleLoadManager {
     @Getter
     private final List<BrokerFilter> brokerFilterPipeline;
 
+    /**
+     * The load data reporter.
+     */
+    private BrokerLoadDataReporter brokerLoadDataReporter;
+
+    private TopBundleLoadDataReporter topBundleLoadDataReporter;
+
     private boolean started = false;
 
     private final ConcurrentOpenHashMap<String, CompletableFuture<Optional<BrokerLookupData>>>
@@ -129,7 +139,23 @@ public void start() throws PulsarServerException {
                 .brokerRegistry(brokerRegistry)
                 .brokerLoadDataStore(brokerLoadDataStore)
                 .topBundleLoadDataStore(topBundlesLoadDataStore).build();
-        // TODO: Start load data reporter.
+
+
+        this.brokerLoadDataReporter =
+                new BrokerLoadDataReporter(pulsar, brokerRegistry.getBrokerId(), brokerLoadDataStore);
+
+        this.topBundleLoadDataReporter =
+                new TopBundleLoadDataReporter(pulsar, brokerRegistry.getBrokerId(), topBundlesLoadDataStore);
+
+        var interval = conf.getLoadBalancerReportUpdateMinIntervalMillis();
+        this.pulsar.getLoadManagerExecutor()
+                .scheduleAtFixedRate(() -> brokerLoadDataReporter.reportAsync(false),
+                        interval,
+                        interval, TimeUnit.MILLISECONDS);
+        this.pulsar.getLoadManagerExecutor()
+                .scheduleAtFixedRate(() -> topBundleLoadDataReporter.reportAsync(false),
+                        interval,
+                        interval, TimeUnit.MILLISECONDS);
 
         // TODO: Start unload scheduler and bundle split scheduler
         this.started = true;

diff --git a/...er/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/data/BrokerLoadData.java b/...er/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/data/BrokerLoadData.java
@@ -20,6 +20,8 @@
 
 import lombok.EqualsAndHashCode;
 import lombok.Getter;
+import lombok.Setter;
+import lombok.ToString;
 import org.apache.pulsar.broker.ServiceConfiguration;
 import org.apache.pulsar.policies.data.loadbalancer.LocalBrokerData;
 import org.apache.pulsar.policies.data.loadbalancer.ResourceUsage;
@@ -33,6 +35,7 @@
  */
 @Getter
 @EqualsAndHashCode
+@ToString
 public class BrokerLoadData {
 
     private static final double DEFAULT_RESOURCE_USAGE = 1.0d;
@@ -49,6 +52,7 @@ public class BrokerLoadData {
     private double msgThroughputOut;  // bytes/sec
     private double msgRateIn; // messages/sec
     private double msgRateOut; // messages/sec
+    private int bundleCount;
 
     // Load data features computed from the above resources.
     private double maxResourceUsage; // max of resource usages
@@ -68,6 +72,9 @@ public class BrokerLoadData {
     private double weightedMaxEMA;
     private long updatedAt;
 
+    @Setter
+    private long reportedAt;
+
     public BrokerLoadData() {
         cpu = new ResourceUsage();
         memory = new ResourceUsage();
@@ -91,6 +98,8 @@ public BrokerLoadData() {
      *            broker-level message input rate in messages/s.
      * @param msgRateOut
      *            broker-level message output rate in messages/s.
+     * @param bundleCount
+     *            broker-level bundle counts.
      * @param conf
      *            Service configuration to compute load data features.
      */
@@ -99,12 +108,14 @@ public void update(final SystemResourceUsage usage,
                        double msgThroughputOut,
                        double msgRateIn,
                        double msgRateOut,
+                       int bundleCount,
                        ServiceConfiguration conf) {
         updateSystemResourceUsage(usage.cpu, usage.memory, usage.directMemory, usage.bandwidthIn, usage.bandwidthOut);
         this.msgThroughputIn = msgThroughputIn;
         this.msgThroughputOut = msgThroughputOut;
         this.msgRateIn = msgRateIn;
         this.msgRateOut = msgRateOut;
+        this.bundleCount = bundleCount;
         updateFeatures(conf);
         updatedAt = System.currentTimeMillis();
     }
@@ -121,9 +132,11 @@ public void update(final BrokerLoadData other) {
         msgThroughputOut = other.msgThroughputOut;
         msgRateIn = other.msgRateIn;
         msgRateOut = other.msgRateOut;
+        bundleCount = other.bundleCount;
         weightedMaxEMA = other.weightedMaxEMA;
         maxResourceUsage = other.maxResourceUsage;
         updatedAt = other.updatedAt;
+        reportedAt = other.reportedAt;
     }
 
     // Update resource usage given each individual usage.
@@ -173,7 +186,9 @@ public String toString(ServiceConfiguration conf) {
                         + "cpuWeight= %f, memoryWeight= %f, directMemoryWeight= %f, "
                         + "bandwithInResourceWeight= %f, bandwithOutResourceWeight= %f, "
                         + "msgThroughputIn= %.2f, msgThroughputOut= %.2f, msgRateIn= %.2f, msgRateOut= %.2f, "
-                        + "maxResourceUsage= %.2f%%, weightedMaxEMA= %.2f%%, updatedAt= %d",
+                        + "bundleCount= %d, "
+                        + "maxResourceUsage= %.2f%%, weightedMaxEMA= %.2f%%, "
+                        + "updatedAt= %d, reportedAt= %d",
 
                 cpu.percentUsage(), memory.percentUsage(), directMemory.percentUsage(),
                 bandwidthIn.percentUsage(), bandwidthOut.percentUsage(),
@@ -183,7 +198,9 @@ public String toString(ServiceConfiguration conf) {
                 conf.getLoadBalancerBandwithInResourceWeight(),
                 conf.getLoadBalancerBandwithOutResourceWeight(),
                 msgThroughputIn, msgThroughputOut, msgRateIn, msgRateOut,
-                maxResourceUsage * 100, weightedMaxEMA * 100, updatedAt
+                bundleCount,
+                maxResourceUsage * 100, weightedMaxEMA * 100,
+                updatedAt, reportedAt
         );
     }
 

diff --git a/...rc/main/java/org/apache/pulsar/broker/loadbalance/extensions/data/TopBundlesLoadData.java b/...rc/main/java/org/apache/pulsar/broker/loadbalance/extensions/data/TopBundlesLoadData.java
@@ -18,41 +18,97 @@
  */
 package org.apache.pulsar.broker.loadbalance.extensions.data;
 
+import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
+import java.util.Map;
 import java.util.Objects;
-import java.util.stream.Collectors;
+import lombok.EqualsAndHashCode;
 import lombok.Getter;
+import lombok.ToString;
+import org.apache.pulsar.common.naming.NamespaceName;
 import org.apache.pulsar.policies.data.loadbalancer.NamespaceBundleStats;
 
 /**
  * Defines the information of top bundles load data.
  */
 @Getter
+@ToString
+@EqualsAndHashCode
 public class TopBundlesLoadData {
 
-    private final List<BundleLoadData> topBundlesLoadData;
+    private final List<BundleLoadData> topBundlesLoadData = new ArrayList<>();
+
+    private final List<Map.Entry<String, ? extends Comparable>> arr = new ArrayList<>();
 
     public record BundleLoadData(String bundleName, NamespaceBundleStats stats) {
         public BundleLoadData {
             Objects.requireNonNull(bundleName);
         }
     }
 
-    private TopBundlesLoadData(List<BundleLoadData> bundleStats, int topK) {
-        topBundlesLoadData = bundleStats
-                .stream()
-                .sorted((o1, o2) -> o2.stats().compareTo(o1.stats()))
-                .limit(topK)
-                .collect(Collectors.toList());
+    public TopBundlesLoadData() {
     }
 
     /**
-     * Give full bundle stats, and return the top K bundle stats.
-     *
-     * @param bundleStats full bundle stats.
-     * @param topK Top K bundles.
+     * Return TopBundlesLoadData with the given bundleStats.
+     * @param bundleStats bundle stats.
+     * @param topk top k bundle stats to select.
      */
-    public static TopBundlesLoadData of(List<BundleLoadData> bundleStats, int topK) {
-        return new TopBundlesLoadData(bundleStats, topK);
+    public void update(Map<String, NamespaceBundleStats> bundleStats, int topk) {
+        arr.clear();
+        for (var etr : bundleStats.entrySet()) {
+            if (etr.getKey().startsWith(NamespaceName.SYSTEM_NAMESPACE.toString())) {
+                continue;
+            }
+            arr.add(etr);
+        }
+        topBundlesLoadData.clear();
+        if (arr.isEmpty()) {
+            return;
+        }
+        topk = Math.min(topk, arr.size());
+        partitionSort(arr, topk);
+
+        for (int i = 0; i < topk; i++) {
+            var etr = arr.get(i);
+            topBundlesLoadData.add(new BundleLoadData(etr.getKey(), (NamespaceBundleStats) etr.getValue()));
+        }
+    }
+
+    static void partitionSort(List<Map.Entry<String, ? extends Comparable>> arr, int k) {
+        int start = 0;
+        int end = arr.size() - 1;
+        int target = k - 1;
+        while (start < end) {
+            int lo = start;
+            int hi = end;
+            int mid = lo;
+            var pivot = arr.get(hi).getValue();
+            while (mid <= hi) {
+                int cmp = pivot.compareTo(arr.get(mid).getValue());
+                if (cmp < 0) {
+                    var tmp = arr.get(lo);
+                    arr.set(lo++, arr.get(mid));
+                    arr.set(mid++, tmp);
+                } else if (cmp > 0) {
+                    var tmp = arr.get(mid);
+                    arr.set(mid, arr.get(hi));
+                    arr.set(hi--, tmp);
+                } else {
+                    mid++;
+                }
+            }
+            if (lo <= target && target < mid) {
+                end = lo;
+                break;
+            }
+            if (target < lo) {
+                end = lo - 1;
+            } else {
+                start = mid;
+            }
+        }
+        Collections.sort(arr.subList(0, end), (a, b) -> b.getValue().compareTo(a.getValue()));
     }
 }