kfaraz commented on code in PR #19622:
URL: https://github.com/apache/druid/pull/19622#discussion_r3479297438
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/CostMetrics.java:
##########
@@ -136,6 +115,16 @@ public Double getMaxObservedRate()
return maxObservedRate;
}
+
+ /**
+ * Derives the current idle ratio from measured utilization ({@code
avgProcessingRate / maxObservedRate}).
+ */
+ double estimateIdleRatioFromProcessingRate(CostMetrics metrics)
Review Comment:
We don't need to pass the `CostMetrics` here.
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/CostBasedAutoScaler.java:
##########
@@ -78,13 +80,21 @@ public class CostBasedAutoScaler implements
SupervisorTaskAutoScaler
*/
static final int MAX_IDLENESS_PARTITION_LAG = 10_000;
+ /**
+ * The number of most recent processing rate samples to maintain for
auto-scaling decisions
+ * in case when {@link CostBasedAutoScalerConfig#usePollIdleRatio} is
disabled.
+ */
+ static final int RATE_WINDOW_SIZE = 50;
Review Comment:
```suggestion
static final int PROCESSING_RATE_WINDOW_SIZE = 50;
```
##########
indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/CostBasedAutoScalerTest.java:
##########
@@ -563,6 +571,78 @@ public void
testScalingActionSkippedWhenMovingAverageRateUnavailable()
);
}
+ @Test
+ public void
testCollectMetricsTracksRateWatermarkOnlyWhenPollIdleRatioDisabled()
+ {
+ SupervisorSpec spec = Mockito.mock(SupervisorSpec.class);
+ SeekableStreamSupervisor supervisor =
Mockito.mock(SeekableStreamSupervisor.class);
+ ServiceEmitter emitter = Mockito.mock(ServiceEmitter.class);
+ SeekableStreamSupervisorIOConfig ioConfig =
Mockito.mock(SeekableStreamSupervisorIOConfig.class);
+
+ when(spec.getId()).thenReturn("test-supervisor");
+ when(spec.getDataSources()).thenReturn(List.of("test-datasource"));
+ when(spec.isSuspended()).thenReturn(false);
+ when(supervisor.getIoConfig()).thenReturn(ioConfig);
+ when(ioConfig.getStream()).thenReturn("test-stream");
+ when(ioConfig.getTaskDuration()).thenReturn(Duration.standardHours(1));
+ when(supervisor.getPartitionCount()).thenReturn(1);
+ when(supervisor.computeLagStats()).thenReturn(new LagStats(0, 0, 0));
+
+ // usePollIdleRatio defaults to true, which disables rate-watermark
tracking entirely.
+ CostBasedAutoScalerConfig defaultConfig =
CostBasedAutoScalerConfig.builder()
+
.taskCountMax(10)
+
.taskCountMin(1)
+
.enableTaskAutoScaler(true)
+
.build();
+ when(supervisor.getStats()).thenReturn(buildTaskStatsForRate(500.0));
+ CostBasedAutoScaler defaultScaler = new CostBasedAutoScaler(supervisor,
defaultConfig, spec, emitter);
+ Assert.assertNull(
+ "With usePollIdleRatio=true (the default), samples are never
collected",
+ defaultScaler.collectMetrics().getMaxObservedRate()
+ );
+
+ // Disabling usePollIdleRatio switches the idle cost to processing-rate
utilization,
+ // which requires tracking a watermark of observed rates.
+ CostBasedAutoScalerConfig utilizationConfig =
CostBasedAutoScalerConfig.builder()
Review Comment:
Please remove usages of the word utilization as it is ambiguous.
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/CostMetrics.java:
##########
@@ -105,6 +108,26 @@ public double getAvgProcessingRate()
return avgProcessingRate;
}
+ public Double getMaxObservedRate()
Review Comment:
```suggestion
public double getMaxObservedRate()
```
##########
indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/CostBasedAutoScalerTest.java:
##########
@@ -563,6 +571,78 @@ public void
testScalingActionSkippedWhenMovingAverageRateUnavailable()
);
}
+ @Test
+ public void
testCollectMetricsTracksRateWatermarkOnlyWhenPollIdleRatioDisabled()
+ {
+ SupervisorSpec spec = Mockito.mock(SupervisorSpec.class);
+ SeekableStreamSupervisor supervisor =
Mockito.mock(SeekableStreamSupervisor.class);
+ ServiceEmitter emitter = Mockito.mock(ServiceEmitter.class);
+ SeekableStreamSupervisorIOConfig ioConfig =
Mockito.mock(SeekableStreamSupervisorIOConfig.class);
+
+ when(spec.getId()).thenReturn("test-supervisor");
+ when(spec.getDataSources()).thenReturn(List.of("test-datasource"));
+ when(spec.isSuspended()).thenReturn(false);
+ when(supervisor.getIoConfig()).thenReturn(ioConfig);
+ when(ioConfig.getStream()).thenReturn("test-stream");
+ when(ioConfig.getTaskDuration()).thenReturn(Duration.standardHours(1));
+ when(supervisor.getPartitionCount()).thenReturn(1);
+ when(supervisor.computeLagStats()).thenReturn(new LagStats(0, 0, 0));
+
+ // usePollIdleRatio defaults to true, which disables rate-watermark
tracking entirely.
+ CostBasedAutoScalerConfig defaultConfig =
CostBasedAutoScalerConfig.builder()
+
.taskCountMax(10)
+
.taskCountMin(1)
+
.enableTaskAutoScaler(true)
+
.build();
+ when(supervisor.getStats()).thenReturn(buildTaskStatsForRate(500.0));
+ CostBasedAutoScaler defaultScaler = new CostBasedAutoScaler(supervisor,
defaultConfig, spec, emitter);
+ Assert.assertNull(
+ "With usePollIdleRatio=true (the default), samples are never
collected",
+ defaultScaler.collectMetrics().getMaxObservedRate()
+ );
+
+ // Disabling usePollIdleRatio switches the idle cost to processing-rate
utilization,
+ // which requires tracking a watermark of observed rates.
+ CostBasedAutoScalerConfig utilizationConfig =
CostBasedAutoScalerConfig.builder()
+
.taskCountMax(10)
+
.taskCountMin(1)
+
.enableTaskAutoScaler(true)
+
.usePollIdleRatio(false)
+
.build();
+ when(supervisor.getStats()).thenReturn(
+ buildTaskStatsForRate(500.0),
+ buildTaskStatsForRate(9000.0),
+ buildTaskStatsForRate(300.0)
+ );
+ CostBasedAutoScaler utilizationScaler = new
CostBasedAutoScaler(supervisor, utilizationConfig, spec, emitter);
Review Comment:
Maybe rename this to `autoScalerWithoutPollIdleRatio` or similar.
##########
indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/CostBasedAutoScalerTest.java:
##########
@@ -563,6 +571,78 @@ public void
testScalingActionSkippedWhenMovingAverageRateUnavailable()
);
}
+ @Test
+ public void
testCollectMetricsTracksRateWatermarkOnlyWhenPollIdleRatioDisabled()
Review Comment:
```suggestion
public void
testCollectMetricsTracksMaxProcessingRateOnlyWhenPollIdleRatioDisabled()
```
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/CostMetrics.java:
##########
@@ -105,6 +108,26 @@ public double getAvgProcessingRate()
return avgProcessingRate;
}
+ public Double getMaxObservedRate()
+ {
+ return maxObservedRate;
+ }
+
+
+ /**
+ * Derives the current idle ratio from measured utilization ({@code
avgProcessingRate / maxObservedRate}).
+ */
+ double estimateIdleRatioFromProcessingRate(CostMetrics metrics)
+ {
+ if (maxObservedRate <= 0) {
+ // No throughput baseline yet (e.g. only zero-rate samples at startup).
Returning a negative
+ // value routes computeCost to its neutral idle branch instead of
producing NaN from 0/0.
Review Comment:
How the caller (i.e. computeCost) deals with a negative value should not be
a concern of this class. We can remove that from the comment here.
The javadoc of this method should just call out the case when a negative
value is returned.
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/CostMetrics.java:
##########
@@ -120,7 +143,8 @@ public boolean equals(Object o)
&& partitionCount == that.partitionCount
&& Double.compare(that.pollIdleRatio, pollIdleRatio) == 0
&& taskDurationSeconds == that.taskDurationSeconds
- && Double.compare(that.avgProcessingRate, avgProcessingRate) == 0;
+ && Double.compare(that.avgProcessingRate, avgProcessingRate) == 0
+ && Objects.equals(that.maxObservedRate, maxObservedRate);
Review Comment:
```suggestion
&& Double.compare(that.maxObservedRate, maxObservedRate) == 0;
```
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/CostBasedAutoScalerConfig.java:
##########
@@ -282,6 +285,16 @@ public boolean isScaleDownOnTaskRolloverOnly()
return scaleDownDuringTaskRolloverOnly;
}
+ /**
+ * If true (the default), idle cost is derived from the poll idle ratio
reported by tasks.
+ * If false, idle cost is derived from processing-rate utilization instead.
+ */
+ @JsonProperty("usePollIdleRatio")
Review Comment:
Not really needed
```suggestion
@JsonProperty
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
