This is an automated email from the ASF dual-hosted git repository.
gortiz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new db47c062b04 Emit per-table broker metrics for routing rebuilds on
assignment change (#19784)
db47c062b04 is described below
commit db47c062b04158cf599a44d597c8dc4aba294cb8
Author: Gonzalo Ortiz Jaureguizar <[email protected]>
AuthorDate: Tue Oct 13 12:23:32 2026 +0200
Emit per-table broker metrics for routing rebuilds on assignment change
(#19784)
* Emit per-table metrics for broker routing rebuilds on assignment change
The broker rebuilds the routing of a table on every ideal state or external
view change, but it reported nothing about how often that happens or how
long it takes. BrokerTimer.ROUTING_TABLE_UPDATE_TIME was defined (and has a
JMX exporter rule) but was never emitted.
BaseBrokerRoutingManager.processAssignmentChangeForTable now:
- records ROUTING_TABLE_UPDATE_TIME per table and globally. The time covers
reading the ideal state and external view and
RoutingEntry.onAssignmentChange.
- increments the new per-table meter BrokerMeter.ROUTING_TABLE_UPDATES.
- sets the new per-table gauge BrokerGauge.ROUTING_TABLE_SEGMENTS to the
number
of pre-selected online segments the routing entry tracks.
RoutingEntry keeps that segment count, set at build time and on each
assignment change. buildRoutingInternal reports the gauge after it stores
the
entry, and removeRoutingInternal removes it with the replica health gauges.
* Count failed routing updates separately from successful ones
BaseBrokerRoutingManager.processAssignmentChangeForTable recorded
ROUTING_TABLE_UPDATE_TIME and ROUTING_TABLE_UPDATES after the try/catch, so
an
update that threw in RoutingEntry.onAssignmentChange was timed and counted
as a
normal update. The routing entry keeps its old ideal state and external view
versions on failure, so the broker retries on every later change, and a
table
whose update keeps failing looked like a table with many assignment changes.
The timer and ROUTING_TABLE_UPDATES are now recorded only after a successful
onAssignmentChange. The catch increments the new per-table meter
BrokerMeter.ROUTING_TABLE_UPDATE_FAILURES. ROUTING_TABLE_SEGMENTS and the
replica health gauges are still reported in both cases.
BrokerRoutingManagerTest.testFailedAssignmentChangeReportsFailureMetric
covers
the failure path.
* Rename routing update metrics and routing entry segment count
Address review naming suggestions:
- BrokerMeter ROUTING_TABLE_UPDATES -> ROUTING_TABLE_UPDATE_SUCCESS and
ROUTING_TABLE_UPDATE_FAILURES -> ROUTING_TABLE_UPDATE_FAILURE, matching
existing SUCCESS/FAILURE pairs such as IDEAL_STATE_UPDATE_SUCCESS.
- RoutingEntry._numSegments -> _numPreSelectedOnlineSegments (and getter),
naming the exact population the ROUTING_TABLE_SEGMENTS gauge reports.
- Rename the test to testAssignmentChangeReportsRoutingUpdateMetrics, since
full routing builds do not emit these metrics.
---------
Co-authored-by: Gonzalo Ortiz <[email protected]>
---
.../routing/manager/BaseBrokerRoutingManager.java | 32 ++++++++++++-
.../routing/manager/BrokerRoutingManagerTest.java | 56 ++++++++++++++++++++--
.../apache/pinot/common/metrics/BrokerGauge.java | 6 ++-
.../apache/pinot/common/metrics/BrokerMeter.java | 15 ++++++
.../apache/pinot/common/metrics/BrokerTimer.java | 3 ++
5 files changed, 106 insertions(+), 6 deletions(-)
diff --git
a/pinot-broker/src/main/java/org/apache/pinot/broker/routing/manager/BaseBrokerRoutingManager.java
b/pinot-broker/src/main/java/org/apache/pinot/broker/routing/manager/BaseBrokerRoutingManager.java
index 829e20c8e82..5aabc29e179 100644
---
a/pinot-broker/src/main/java/org/apache/pinot/broker/routing/manager/BaseBrokerRoutingManager.java
+++
b/pinot-broker/src/main/java/org/apache/pinot/broker/routing/manager/BaseBrokerRoutingManager.java
@@ -77,6 +77,7 @@ import org.apache.pinot.common.metadata.ZKMetadataProvider;
import org.apache.pinot.common.metrics.BrokerGauge;
import org.apache.pinot.common.metrics.BrokerMeter;
import org.apache.pinot.common.metrics.BrokerMetrics;
+import org.apache.pinot.common.metrics.BrokerTimer;
import org.apache.pinot.common.request.BrokerRequest;
import org.apache.pinot.common.utils.HashUtil;
import org.apache.pinot.common.utils.config.QueryOptionsUtils;
@@ -391,6 +392,7 @@ public abstract class BaseBrokerRoutingManager
if (idealStateVersion != routingEntry.getLastUpdateIdealStateVersion()
|| externalViewVersion !=
routingEntry.getLastUpdateExternalViewVersion()) {
String tableNameWithType = routingEntry.getTableNameWithType();
+ long startTimeNs = System.nanoTime();
try {
IdealState idealState = getIdealState(routingEntry._idealStatePath);
if (idealState == null) {
@@ -403,10 +405,19 @@ public abstract class BaseBrokerRoutingManager
return true;
}
routingEntry.onAssignmentChange(idealState, externalView);
+ // Only successful updates are timed and counted, so that a table
whose update keeps failing does not look
+ // like a table with many assignment changes
+ long updateTimeNs = System.nanoTime() - startTimeNs;
+ _brokerMetrics.addTimedTableValue(tableNameWithType,
BrokerTimer.ROUTING_TABLE_UPDATE_TIME, updateTimeNs,
+ TimeUnit.NANOSECONDS);
+ _brokerMetrics.addTimedValue(BrokerTimer.ROUTING_TABLE_UPDATE_TIME,
updateTimeNs, TimeUnit.NANOSECONDS);
+ _brokerMetrics.addMeteredTableValue(tableNameWithType,
BrokerMeter.ROUTING_TABLE_UPDATE_SUCCESS, 1L);
} catch (Exception e) {
LOGGER.error("Caught unexpected exception while updating routing entry
on segment assignment change for "
+ "table: {}", tableNameWithType, e);
+ _brokerMetrics.addMeteredTableValue(tableNameWithType,
BrokerMeter.ROUTING_TABLE_UPDATE_FAILURE, 1L);
}
+ updateRoutingSegmentsMetric(routingEntry);
updateReplicaHealthMetrics(routingEntry);
return true;
}
@@ -781,6 +792,12 @@ public abstract class BaseBrokerRoutingManager
replicaHealth.getNumUnavailableSegments());
}
+ /// Reports the number of segments the routing entry tracks. Callers must
hold the table's routing build lock.
+ private void updateRoutingSegmentsMetric(RoutingEntry routingEntry) {
+ _brokerMetrics.setValueOfTableGauge(routingEntry.getTableNameWithType(),
BrokerGauge.ROUTING_TABLE_SEGMENTS,
+ routingEntry.getNumPreSelectedOnlineSegments());
+ }
+
/// Stops reporting the table's replica health gauges, so that they do not
keep being exported frozen at a
/// value that no longer describes the table.
private void removeReplicaHealthMetrics(String tableNameWithType) {
@@ -980,7 +997,7 @@ public abstract class BaseBrokerRoutingManager
new RoutingEntry(tableNameWithType, idealStatePath,
externalViewPath, segmentPreSelector, segmentSelector,
segmentPruners, instanceSelector, idealStateVersion,
externalViewVersion, segmentZkMetadataFetcher,
timeBoundaryManager, partitionMetadataManager, queryTimeoutMs,
samplerInfos,
- !idealState.isEnabled());
+ !idealState.isEnabled(), preSelectedOnlineSegments.size());
if (_routingEntryMap.put(tableNameWithType, routingEntry) == null) {
LOGGER.info("Built routing for table: {}", tableNameWithType);
} else {
@@ -989,6 +1006,7 @@ public abstract class BaseBrokerRoutingManager
// Reported only once the entry is stored, so that a build that failed
earlier cannot leave gauges
// behind with no routing entry to ever clean them up. The IS / EV
re-check below reports again if it
// ends up updating the entry.
+ updateRoutingSegmentsMetric(routingEntry);
updateReplicaHealthMetrics(routingEntry);
// Check for updates to the IS / EV after adding the routing entry, as
it is possible that the
@@ -1074,6 +1092,7 @@ public abstract class BaseBrokerRoutingManager
// Stop reporting the table level gauges owned by the routing,
otherwise they keep being exported
// for a table this broker no longer serves
removeReplicaHealthMetrics(tableNameWithType);
+ _brokerMetrics.removeTableGauge(tableNameWithType,
BrokerGauge.ROUTING_TABLE_SEGMENTS);
// Remove time boundary manager for the offline part routing if the
removed routing is the real-time part of a
// hybrid table
@@ -1427,12 +1446,15 @@ public abstract class BaseBrokerRoutingManager
transient boolean _disabled;
+ // Number of pre-selected online segments at the last build or assignment
change
+ transient int _numPreSelectedOnlineSegments;
+
RoutingEntry(String tableNameWithType, String idealStatePath, String
externalViewPath,
SegmentPreSelector segmentPreSelector, SegmentSelector
segmentSelector, List<SegmentPruner> segmentPruners,
InstanceSelector instanceSelector, int lastUpdateIdealStateVersion,
int lastUpdateExternalViewVersion,
SegmentZkMetadataFetcher segmentZkMetadataFetcher, @Nullable
TimeBoundaryManager timeBoundaryManager,
@Nullable SegmentPartitionMetadataManager partitionMetadataManager,
@Nullable Long queryTimeoutMs,
- Map<String, SamplerInfo> samplerInfos, boolean disabled) {
+ Map<String, SamplerInfo> samplerInfos, boolean disabled, int
numPreSelectedOnlineSegments) {
_tableNameWithType = tableNameWithType;
_idealStatePath = idealStatePath;
_externalViewPath = externalViewPath;
@@ -1448,6 +1470,7 @@ public abstract class BaseBrokerRoutingManager
_samplerInfos = samplerInfos;
_segmentZkMetadataFetcher = segmentZkMetadataFetcher;
_disabled = disabled;
+ _numPreSelectedOnlineSegments = numPreSelectedOnlineSegments;
}
String getTableNameWithType() {
@@ -1484,6 +1507,10 @@ public abstract class BaseBrokerRoutingManager
return _disabled;
}
+ int getNumPreSelectedOnlineSegments() {
+ return _numPreSelectedOnlineSegments;
+ }
+
private void updateSamplerInfos(IdealState idealState, ExternalView
externalView,
Set<String> preSelectedOnlineSegments) {
if (_samplerInfos.isEmpty()) {
@@ -1528,6 +1555,7 @@ public abstract class BaseBrokerRoutingManager
_disabled = !idealState.isEnabled();
Set<String> onlineSegments = getOnlineSegments(idealState);
Set<String> preSelectedOnlineSegments =
_segmentPreSelector.preSelect(onlineSegments);
+ _numPreSelectedOnlineSegments = preSelectedOnlineSegments.size();
_segmentZkMetadataFetcher.onAssignmentChange(idealState, externalView,
preSelectedOnlineSegments);
_segmentSelector.onAssignmentChange(idealState, externalView,
preSelectedOnlineSegments);
_instanceSelector.onAssignmentChange(idealState, externalView,
preSelectedOnlineSegments);
diff --git
a/pinot-broker/src/test/java/org/apache/pinot/broker/routing/manager/BrokerRoutingManagerTest.java
b/pinot-broker/src/test/java/org/apache/pinot/broker/routing/manager/BrokerRoutingManagerTest.java
index 57592fd309b..e82fac469fc 100644
---
a/pinot-broker/src/test/java/org/apache/pinot/broker/routing/manager/BrokerRoutingManagerTest.java
+++
b/pinot-broker/src/test/java/org/apache/pinot/broker/routing/manager/BrokerRoutingManagerTest.java
@@ -24,6 +24,7 @@ import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import org.apache.helix.AccessOption;
import org.apache.helix.BaseDataAccessor;
@@ -51,6 +52,7 @@ import
org.apache.pinot.common.config.DefaultClusterConfigChangeHandler;
import org.apache.pinot.common.metrics.BrokerGauge;
import org.apache.pinot.common.metrics.BrokerMeter;
import org.apache.pinot.common.metrics.BrokerMetrics;
+import org.apache.pinot.common.metrics.BrokerTimer;
import org.apache.pinot.common.request.BrokerRequest;
import org.apache.pinot.common.request.QuerySource;
import org.apache.pinot.common.utils.config.TableConfigSerDeUtils;
@@ -525,13 +527,14 @@ public class BrokerRoutingManagerTest {
Constructor<?> constructor =
routingEntryClass.getDeclaredConstructor(String.class, String.class,
String.class,
SegmentPreSelector.class, SegmentSelector.class, List.class,
InstanceSelector.class, int.class, int.class,
SegmentZkMetadataFetcher.class, TimeBoundaryManager.class,
SegmentPartitionMetadataManager.class, Long.class,
- Map.class, boolean.class);
+ Map.class, boolean.class, int.class);
constructor.setAccessible(true);
+ SegmentPreSelector segmentPreSelector = onlineSegments -> onlineSegments;
return constructor.newInstance(tableNameWithType, "/IDEALSTATES/" +
tableNameWithType,
- "/EXTERNALVIEW/" + tableNameWithType, mock(SegmentPreSelector.class),
segmentSelector, segmentPruners,
+ "/EXTERNALVIEW/" + tableNameWithType, segmentPreSelector,
segmentSelector, segmentPruners,
instanceSelector, 1, 1,
mock(SegmentZkMetadataFetcher.class), timeBoundaryManager,
partitionMetadataManager, null, samplerInfos,
- disabled);
+ disabled, 0);
}
private static Object createSamplerInfo(InstanceSelector instanceSelector)
@@ -573,6 +576,52 @@ public class BrokerRoutingManagerTest {
_routingManager.removeRouting(TEST_TABLE);
verifyReplicaHealthGaugesRemoved();
+ verify(_brokerMetrics).removeTableGauge(TEST_TABLE,
BrokerGauge.ROUTING_TABLE_SEGMENTS);
+ }
+
+ @Test
+ public void testAssignmentChangeReportsRoutingUpdateMetrics()
+ throws Exception {
+ // Every routing update triggered by an ideal state or external view
change is timed, counted and reports the
+ // number of segments the routing tracks
+ InstanceSelector instanceSelector = mock(InstanceSelector.class);
+ putRoutingEntry(TEST_TABLE, createRoutingEntry(TEST_TABLE, null, null,
Map.of(), instanceSelector, false));
+ IdealState idealState = createIdealState(true);
+ idealState.getRecord().setMapField("seg0", Map.of(SERVER_INSTANCE_ID,
"ONLINE"));
+ idealState.getRecord().setMapField("seg1", Map.of(SERVER_INSTANCE_ID,
"ONLINE"));
+ idealState.getRecord().setMapField("offlineSeg",
Map.of(SERVER_INSTANCE_ID, "OFFLINE"));
+ stubSegmentAssignmentChange(idealState);
+
+ _routingManager.processSegmentAssignmentChangeInternal();
+
+ verify(_brokerMetrics).addTimedTableValue(eq(TEST_TABLE),
eq(BrokerTimer.ROUTING_TABLE_UPDATE_TIME), anyLong(),
+ eq(TimeUnit.NANOSECONDS));
+
verify(_brokerMetrics).addTimedValue(eq(BrokerTimer.ROUTING_TABLE_UPDATE_TIME),
anyLong(),
+ eq(TimeUnit.NANOSECONDS));
+ verify(_brokerMetrics).addMeteredTableValue(TEST_TABLE,
BrokerMeter.ROUTING_TABLE_UPDATE_SUCCESS, 1L);
+ verify(_brokerMetrics).setValueOfTableGauge(TEST_TABLE,
BrokerGauge.ROUTING_TABLE_SEGMENTS, 2L);
+ verify(_brokerMetrics, never()).addMeteredTableValue(TEST_TABLE,
BrokerMeter.ROUTING_TABLE_UPDATE_FAILURE, 1L);
+ }
+
+ @Test
+ public void testFailedAssignmentChangeReportsFailureMetric()
+ throws Exception {
+ // A failed update is counted as a failure, not as an update, and is not
timed, so that a table whose update
+ // keeps failing does not look like a table with many assignment changes
+ InstanceSelector instanceSelector = mock(InstanceSelector.class);
+ doThrow(new
RuntimeException("test")).when(instanceSelector).onAssignmentChange(any(),
any(), any());
+ putRoutingEntry(TEST_TABLE, createRoutingEntry(TEST_TABLE, null, null,
Map.of(), instanceSelector, false));
+ IdealState idealState = createIdealState(true);
+ idealState.getRecord().setMapField("seg0", Map.of(SERVER_INSTANCE_ID,
"ONLINE"));
+ stubSegmentAssignmentChange(idealState);
+
+ _routingManager.processSegmentAssignmentChangeInternal();
+
+ verify(_brokerMetrics).addMeteredTableValue(TEST_TABLE,
BrokerMeter.ROUTING_TABLE_UPDATE_FAILURE, 1L);
+ verify(_brokerMetrics, never()).addMeteredTableValue(TEST_TABLE,
BrokerMeter.ROUTING_TABLE_UPDATE_SUCCESS, 1L);
+ verify(_brokerMetrics, never()).addTimedTableValue(eq(TEST_TABLE),
eq(BrokerTimer.ROUTING_TABLE_UPDATE_TIME),
+ anyLong(), any());
+ verify(_brokerMetrics,
never()).addTimedValue(eq(BrokerTimer.ROUTING_TABLE_UPDATE_TIME), anyLong(),
any());
}
@Test
@@ -657,6 +706,7 @@ public class BrokerRoutingManagerTest {
verify(_brokerMetrics).setValueOfTableGauge(TEST_TABLE,
BrokerGauge.PERCENT_OF_REPLICAS, 100);
verify(_brokerMetrics).setValueOfTableGauge(TEST_TABLE,
BrokerGauge.SEGMENTS_AT_MIN_PERCENT_OF_REPLICAS, 0);
verify(_brokerMetrics).setValueOfTableGauge(TEST_TABLE,
BrokerGauge.UNAVAILABLE_SEGMENTS, 0);
+ verify(_brokerMetrics).setValueOfTableGauge(TEST_TABLE,
BrokerGauge.ROUTING_TABLE_SEGMENTS, 0);
}
@Test
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/metrics/BrokerGauge.java
b/pinot-common/src/main/java/org/apache/pinot/common/metrics/BrokerGauge.java
index 83289ca40c8..b7924004cf1 100644
---
a/pinot-common/src/main/java/org/apache/pinot/common/metrics/BrokerGauge.java
+++
b/pinot-common/src/main/java/org/apache/pinot/common/metrics/BrokerGauge.java
@@ -141,7 +141,11 @@ public enum BrokerGauge implements AbstractMetrics.Gauge {
/// Number of the table's measured segments that are replicated as poorly as
[#PERCENT_OF_REPLICAS] reports,
/// i.e. how many segments that percentage speaks for. The same populations
are excluded, so this reads `0`
/// exactly when the table has nothing to measure.
- SEGMENTS_AT_MIN_PERCENT_OF_REPLICAS("segments", false);
+ SEGMENTS_AT_MIN_PERCENT_OF_REPLICAS("segments", false),
+
+ /// Number of the table's segments that the routing of this broker tracks:
the online segments of the ideal state
+ /// that the segment pre-selector keeps. Updated on every routing build and
every ideal state or external view change.
+ ROUTING_TABLE_SEGMENTS("segments", false);
private final String _brokerGaugeName;
private final String _unit;
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/metrics/BrokerMeter.java
b/pinot-common/src/main/java/org/apache/pinot/common/metrics/BrokerMeter.java
index bc2a8b354e9..5fa2c6cda29 100644
---
a/pinot-common/src/main/java/org/apache/pinot/common/metrics/BrokerMeter.java
+++
b/pinot-common/src/main/java/org/apache/pinot/common/metrics/BrokerMeter.java
@@ -201,6 +201,21 @@ public class BrokerMeter implements AbstractMetrics.Meter {
public static final BrokerMeter SERVER_MISSING_FOR_ROUTING = create(
"SERVER_MISSING_FOR_ROUTING", "badResponses", false);
+ /// Number of times the broker updated the routing of a table after a change
to its ideal state or external view.
+ /// Failed updates are counted in [#ROUTING_TABLE_UPDATE_FAILURE] instead.
+ ///
+ /// This metric is not global and is attached to a particular table.
+ public static final BrokerMeter ROUTING_TABLE_UPDATE_SUCCESS =
+ create("ROUTING_TABLE_UPDATE_SUCCESS", "updates", false);
+
+ /// Number of times the broker failed to update the routing of a table after
a change to its ideal state or external
+ /// view. The broker retries the update on the next change. Failed updates
are not counted in
+ /// [#ROUTING_TABLE_UPDATE_SUCCESS].
+ ///
+ /// This metric is not global and is attached to a particular table.
+ public static final BrokerMeter ROUTING_TABLE_UPDATE_FAILURE =
+ create("ROUTING_TABLE_UPDATE_FAILURE", "failures", false);
+
// Netty connection metrics
public static final BrokerMeter NETTY_CONNECTION_REQUESTS_SENT = create(
"NETTY_CONNECTION_REQUESTS_SENT", "nettyConnection", true);
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/metrics/BrokerTimer.java
b/pinot-common/src/main/java/org/apache/pinot/common/metrics/BrokerTimer.java
index 7b64714c452..297249a853e 100644
---
a/pinot-common/src/main/java/org/apache/pinot/common/metrics/BrokerTimer.java
+++
b/pinot-common/src/main/java/org/apache/pinot/common/metrics/BrokerTimer.java
@@ -23,6 +23,9 @@ import org.apache.pinot.common.Utils;
/// Enumeration containing all the timers exposed by the Pinot broker.
public enum BrokerTimer implements AbstractMetrics.Timer {
+ /// Time the broker spends to update the routing of a table after a change
to its ideal state or external view.
+ /// Includes reading both znodes and rebuilding the routing components.
Emitted per table and globally, only for
+ /// successful updates.
ROUTING_TABLE_UPDATE_TIME(true),
CLUSTER_CHANGE_QUEUE_TIME(true), // metric tracking the freshness lag for
consuming segments
FRESHNESS_LAG_MS(false),
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
