Realtime segment size threshold metrics (apache#14485)

* Realtime segment size threshold metrics * Addessed PR comment
y-scope · Nov 19, 2024 · b9965c8 · b9965c8
1 parent 8300838
commit b9965c8
Show file tree

Hide file tree

Showing 4 changed files with 33 additions and 7 deletions.
diff --git a/pinot-common/src/main/java/org/apache/pinot/common/metrics/ControllerGauge.java b/pinot-common/src/main/java/org/apache/pinot/common/metrics/ControllerGauge.java
@@ -165,6 +165,16 @@ public enum ControllerGauge implements AbstractMetrics.Gauge {
 
   TABLE_DISABLED("tableDisabled", false),
 
+  // A per-table metric that shows the number of rows we expect to consume for the next segment of
+  // any partition in the realtime table. This metric is emitted from the segment size based threshold
+  // computer.
+  NUM_ROWS_THRESHOLD("numRowsThreshold", false),
+
+  // The actual segment size for committing segments. These may be shorter than expected when the administrator
+  // issues a force-commit, or zero when new partitions are detected in the stream (since there is no completing
+  // segment when the partition is first detected).
+  COMMITTING_SEGMENT_SIZE("committingSegmentSize", false),
+
   TABLE_REBALANCE_IN_PROGRESS("tableRebalanceInProgress", false);
 
   private final String _gaugeName;

diff --git a/.../org/apache/pinot/controller/helix/core/realtime/segment/FlushThresholdUpdateManager.java b/.../org/apache/pinot/controller/helix/core/realtime/segment/FlushThresholdUpdateManager.java
@@ -61,7 +61,7 @@ public FlushThresholdUpdater getFlushThresholdUpdater(StreamConfig streamConfig)
     long flushThresholdSegmentSizeBytes = streamConfig.getFlushThresholdSegmentSizeBytes();
     if (flushThresholdRows == 0 || flushThresholdSegmentSizeBytes > 0) {
       return _flushThresholdUpdaterMap.computeIfAbsent(realtimeTableName,
-          k -> new SegmentSizeBasedFlushThresholdUpdater());
+          k -> new SegmentSizeBasedFlushThresholdUpdater(realtimeTableName));
     } else {
       _flushThresholdUpdaterMap.remove(realtimeTableName);
       return new DefaultFlushThresholdUpdater(StreamConfig.DEFAULT_FLUSH_THRESHOLD_ROWS);

diff --git a/...e/pinot/controller/helix/core/realtime/segment/SegmentSizeBasedFlushThresholdUpdater.java b/...e/pinot/controller/helix/core/realtime/segment/SegmentSizeBasedFlushThresholdUpdater.java
@@ -20,7 +20,10 @@
 
 import javax.annotation.Nullable;
 import org.apache.pinot.common.metadata.segment.SegmentZKMetadata;
+import org.apache.pinot.common.metrics.ControllerGauge;
+import org.apache.pinot.common.metrics.ControllerMetrics;
 import org.apache.pinot.spi.stream.StreamConfig;
+import org.apache.pinot.spi.utils.builder.TableNameBuilder;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -36,9 +39,13 @@
 public class SegmentSizeBasedFlushThresholdUpdater implements FlushThresholdUpdater {
   public static final Logger LOGGER = LoggerFactory.getLogger(SegmentSizeBasedFlushThresholdUpdater.class);
   private final SegmentFlushThresholdComputer _flushThresholdComputer;
+  private final String _rawTableName;
 
-  public SegmentSizeBasedFlushThresholdUpdater() {
+  private final ControllerMetrics _controllerMetrics = ControllerMetrics.get();
+
+  public SegmentSizeBasedFlushThresholdUpdater(String realtimeTableName) {
     _flushThresholdComputer = new SegmentFlushThresholdComputer();
+    _rawTableName = TableNameBuilder.extractRawTableName(realtimeTableName);
   }
 
   // synchronized since this method could be called for multiple partitions of the same table in different threads
@@ -50,5 +57,9 @@ public synchronized void updateFlushThreshold(StreamConfig streamConfig, Segment
         _flushThresholdComputer.computeThreshold(streamConfig, committingSegmentDescriptor, committingSegmentZKMetadata,
             newSegmentZKMetadata.getSegmentName());
     newSegmentZKMetadata.setSizeThresholdToFlushSegment(threshold);
+
+    _controllerMetrics.setOrUpdateTableGauge(_rawTableName, ControllerGauge.NUM_ROWS_THRESHOLD, threshold);
+    _controllerMetrics.setOrUpdateTableGauge(_rawTableName, ControllerGauge.COMMITTING_SEGMENT_SIZE,
+        committingSegmentDescriptor.getSegmentSizeBytes());
   }
 }
diff --git a/...va/org/apache/pinot/controller/helix/core/realtime/segment/FlushThresholdUpdaterTest.java b/...va/org/apache/pinot/controller/helix/core/realtime/segment/FlushThresholdUpdaterTest.java
@@ -139,7 +139,8 @@ public void testSegmentSizeBasedFlushThreshold() {
 
     for (long[] segmentSizesMB : Arrays.asList(EXPONENTIAL_GROWTH_SEGMENT_SIZES_MB, LOGARITHMIC_GROWTH_SEGMENT_SIZES_MB,
         STEPS_SEGMENT_SIZES_MB)) {
-      SegmentSizeBasedFlushThresholdUpdater flushThresholdUpdater = new SegmentSizeBasedFlushThresholdUpdater();
+      SegmentSizeBasedFlushThresholdUpdater flushThresholdUpdater =
+          new SegmentSizeBasedFlushThresholdUpdater(REALTIME_TABLE_NAME);
 
       // Start consumption
       SegmentZKMetadata newSegmentZKMetadata = getNewSegmentZKMetadata(0);
@@ -176,7 +177,8 @@ public void testSegmentSizeBasedFlushThresholdMinPartition() {
 
     for (long[] segmentSizesMB : Arrays.asList(EXPONENTIAL_GROWTH_SEGMENT_SIZES_MB, LOGARITHMIC_GROWTH_SEGMENT_SIZES_MB,
         STEPS_SEGMENT_SIZES_MB)) {
-      SegmentSizeBasedFlushThresholdUpdater flushThresholdUpdater = new SegmentSizeBasedFlushThresholdUpdater();
+      SegmentSizeBasedFlushThresholdUpdater flushThresholdUpdater =
+          new SegmentSizeBasedFlushThresholdUpdater(REALTIME_TABLE_NAME);
 
       // Start consumption
       SegmentZKMetadata newSegmentZKMetadata = getNewSegmentZKMetadata(1);
@@ -236,7 +238,8 @@ private long getSegmentSizeBytes(int numRowsConsumed, long[] segmentSizesMB) {
 
   @Test
   public void testTimeThreshold() {
-    SegmentSizeBasedFlushThresholdUpdater flushThresholdUpdater = new SegmentSizeBasedFlushThresholdUpdater();
+    SegmentSizeBasedFlushThresholdUpdater flushThresholdUpdater =
+        new SegmentSizeBasedFlushThresholdUpdater(REALTIME_TABLE_NAME);
     StreamConfig streamConfig = mockDefaultAutotuneStreamConfig();
 
     // Start consumption
@@ -269,7 +272,8 @@ public void testTimeThreshold() {
 
   @Test
   public void testMinThreshold() {
-    SegmentSizeBasedFlushThresholdUpdater flushThresholdUpdater = new SegmentSizeBasedFlushThresholdUpdater();
+    SegmentSizeBasedFlushThresholdUpdater flushThresholdUpdater =
+        new SegmentSizeBasedFlushThresholdUpdater(REALTIME_TABLE_NAME);
     StreamConfig streamConfig = mockDefaultAutotuneStreamConfig();
 
     // Start consumption
@@ -301,7 +305,8 @@ public void testMinThreshold() {
 
   @Test
   public void testSegmentSizeBasedUpdaterWithModifications() {
-    SegmentSizeBasedFlushThresholdUpdater flushThresholdUpdater = new SegmentSizeBasedFlushThresholdUpdater();
+    SegmentSizeBasedFlushThresholdUpdater flushThresholdUpdater
+        = new SegmentSizeBasedFlushThresholdUpdater(REALTIME_TABLE_NAME);
 
     // Use customized stream config
     long flushSegmentDesiredSizeBytes = StreamConfig.DEFAULT_FLUSH_THRESHOLD_SEGMENT_SIZE_BYTES / 2;