Export Adaptive Routing Stats as broker metrics

[timothy-e]

When multistage adaptive server selector stats collection is enabled (pinot.broker.adaptive.server.selector.enable.stats.collection=true), the broker tracks per-server routing stats (in-flight request count, latency EMA, hybrid score) in memory but has no way to observe them externally without hitting the broker's debug API. This makes it difficult to monitor the health and behavior of adaptive routing in production.

We add new metrics for the stats:

• adaptiveServerNumInFlightRequests
• adaptiveServerLatencyEma
• adaptiveServerHybridScore

This introduces 3 new metrics for each broker x server combination on a tenant. Not all users / usecases will want or warrant this, so we disable it by default.

The resulting Pinot metric names are of the form: pinot.broker.adaptiveServerLatencyEma.server.Server_pinotdb1_8098

Hybrid Score Range

ServerRoutingStatsEntry.java::computeHybridScore defines a hybrid score as:

  public double computeHybridScore() {
    double estimatedQSize = _numInFlightRequests + _inFlighRequestsEMA.getAverage();
    return Math.pow(estimatedQSize, _hybridScoreExponent) * _latencyMsEMA.getAverage();
  }

_hybridScoreExponent defaults to 3.

The hybrid score could be as low as 0 if there are no inflight requests, but as soon as there is an inflight request, we jump to at least _latencyMsEMA.getAverage(). An unhealthy server would be much higher because it would have an increased number of inflight requests (~ (inflight_reqs * infight_reqs_ema) ^ 3 => a server with 5 inflight requests would have it's latency_ema multiplied by ~ (5 + 5) ^ 3 = 1000.

Using a decimal data type would only help differentiate scores among servers with very low inflight requests, which isn't too helpful.

Testing

Unit tests verify that we:

• don't export when adaptive routing stats are disabled
• don't export when adaptive routing stats metric export is disabled

We deployed this to an internal stripe cluster with adaptive routing stats enabled, and saw metrics appearing in Grafana.

Setting the value

Currently this requires restarting the brokers. #18135 will allow these values to be toggled without a restart.

[timothy-e]

@Jackie-Jiang @vvivekiyer can you please take a look at this when you have a chance?

[codecov-commenter]

Codecov Report

❌ Patch coverage is 84.61538% with 4 lines in your changes missing coverage. Please review.
✅ Project coverage is 63.99%. Comparing base (739043c) to head (dd68062).
⚠️ Report is 306 commits behind head on master.

Files with missing lines	Patch %	Lines
...erver/routing/stats/ServerRoutingStatsManager.java	90.47%	2 Missing ⚠️
...va/org/apache/pinot/spi/utils/CommonConstants.java	0.00%	2 Missing ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #18134      +/-   ##
============================================
+ Coverage     63.22%   63.99%   +0.77%     
- Complexity     1454     1617     +163     
============================================
  Files          3176     3192      +16     
  Lines        191002   193960    +2958     
  Branches      29204    29938     +734     
============================================
+ Hits         120763   124131    +3368     
+ Misses        60818    60026     -792     
- Partials       9421     9803     +382     

Flag	Coverage Δ
custom-integration1	100.00% <ø> (ø)
integration	100.00% <ø> (ø)
integration1	100.00% <ø> (ø)
integration2	0.00% <ø> (ø)
java-11	63.97% <84.61%> (+0.79%)	⬆️
java-21	63.96% <84.61%> (+0.77%)	⬆️
temurin	63.99% <84.61%> (+0.77%)	⬆️
unittests	63.99% <84.61%> (+0.77%)	⬆️
unittests1	55.90% <84.61%> (+0.35%)	⬆️
unittests2	34.35% <15.38%> (+0.25%)	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[xiangfu0]

I found a potential thread-safety issue in the exportStatsAsMetrics() method:

The iteration over _serverQueryStatsMap.entrySet() at line 408 is not protected. While ConcurrentHashMap is used, iterating over entrySet() while the map is being concurrently modified (via recordStatsForQuerySubmission()) can result in:

1. ConcurrentModificationException
2. Missed entries in the metrics export
3. Inconsistent snapshots across iterations

Since this runs on a periodic scheduled task that could coincide with ongoing request processing, this race condition is likely to manifest under load.

Consider wrapping the iteration with explicit synchronization or taking a snapshot of the entries:

private void exportStatsAsMetrics() {
  try {
    // Option 1: Snapshot the entries
    List<Map.Entry<String, ServerRoutingStatsEntry>> entries = 
      new ArrayList<>(_serverQueryStatsMap.entrySet());
    for (Map.Entry<String, ServerRoutingStatsEntry> entry : entries) {

Or use an iterator with exception handling:

Iterator<Map.Entry<String, ServerRoutingStatsEntry>> it = 
  _serverQueryStatsMap.entrySet().iterator();
while (it.hasNext()) {
  try {
    Map.Entry<String, ServerRoutingStatsEntry> entry = it.next();

[timothy-e]

I found a potential thread-safety issue in the exportStatsAsMetrics() method:

The iteration over _serverQueryStatsMap.entrySet() at line 408 is not protected. While ConcurrentHashMap is used, iterating over entrySet() while the map is being concurrently modified (via recordStatsForQuerySubmission()) can result in:

1. ConcurrentModificationException
2. Missed entries in the metrics export
3. Inconsistent snapshots across iterations

I don't think this is actually a problem. The concurrent hash map java doc says:

Similarly, Iterators, Spliterators and Enumerations return elements reflecting the state of the hash table at some point at or since the creation of the iterator/enumeration. They do not throw ConcurrentModificationException.

There are already 4 patterns like this in the file: getServerRoutingStats, fetchNumInFlightRequestsForAllServers. fetchEMALatencyForAllServers, fetchHybridScoreForAllServers.