kfaraz commented on code in PR #19622:
URL: https://github.com/apache/druid/pull/19622#discussion_r3472792643
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/CostBasedAutoScaler.java:
##########
@@ -103,6 +115,11 @@ public CostBasedAutoScaler(
this.costFunction = new WeightedCostFunction();
this.autoscalerExecutor =
Execs.scheduledSingleThreaded("CostBasedAutoScaler-"
+
StringUtils.encodeForFormat(spec.getId()));
+
+ // Bound the rate watermark to roughly one task's lifetime (in
collectMetrics() ticks).
+ final long taskDurationMillis =
supervisor.getIoConfig().getTaskDuration().getStandardSeconds() * 1000L;
+ this.lagGatedRateWindowSize = Math.max(1, (int) (taskDurationMillis /
config.getScaleActionPeriodMillis()));
Review Comment:
So, if task duration is 1 hour and scaling period is 15 mins, the window
size would be only 4?
I feel like we should be looking at a much larger window if we are trying to
determine the max, maybe even 100?
It probably doesn't have to be a function of the task duration, just gating
the sample collection on the lag condition should be enough.
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/CostBasedAutoScalerConfig.java:
##########
@@ -91,7 +92,8 @@ public CostBasedAutoScalerConfig(
@Nullable @JsonProperty("useTaskCountBoundariesOnScaleDown") Boolean
useTaskCountBoundariesOnScaleDown,
@Nullable @JsonProperty("minScaleUpDelay") Duration minScaleUpDelay,
@Nullable @JsonProperty("minScaleDownDelay") Duration minScaleDownDelay,
- @Nullable @JsonProperty("scaleDownDuringTaskRolloverOnly") Boolean
scaleDownDuringTaskRolloverOnly
+ @Nullable @JsonProperty("scaleDownDuringTaskRolloverOnly") Boolean
scaleDownDuringTaskRolloverOnly,
+ @Nullable @JsonProperty("useUtilizationRatio") Boolean
useUtilizationRatio
Review Comment:
We should invert the meaning of this flag and name it `usePollIdleRatio`
(true by default) since that is a concept already being used in this
auto-scaler. Adding another new terminology (utilization ratio) makes things
more confusing.
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/CostBasedAutoScaler.java:
##########
@@ -85,6 +87,16 @@ public class CostBasedAutoScaler implements
SupervisorTaskAutoScaler
private final CostBasedAutoScalerConfig config;
private final ScheduledExecutorService autoscalerExecutor;
private final WeightedCostFunction costFunction;
+ /**
+ * Recent {@code avgProcessingRate} readings, but only from ticks where lag
was present.
+ * "Lag-gated" because lag is the condition that decides whether a reading
counts as evidence
+ * of capacity at all: with no lag the task is only processing as fast as
data arrives (an
+ * upstream supply number, not a capacity number); with lag present, demand
isn't the
+ * bottleneck, so the rate observed is the task's own ceiling. The max of
these samples
+ * becomes {@link CostMetrics#getMaxObservedRate()}.
+ */
+ private final EvictingQueue<Double> lagGatedRateSamples;
Review Comment:
```suggestion
private final EvictingQueue<Double> lagGatedProcessingRateSamples;
```
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/CostBasedAutoScaler.java:
##########
@@ -103,6 +115,11 @@ public CostBasedAutoScaler(
this.costFunction = new WeightedCostFunction();
this.autoscalerExecutor =
Execs.scheduledSingleThreaded("CostBasedAutoScaler-"
+
StringUtils.encodeForFormat(spec.getId()));
+
+ // Bound the rate watermark to roughly one task's lifetime (in
collectMetrics() ticks).
+ final long taskDurationMillis =
supervisor.getIoConfig().getTaskDuration().getStandardSeconds() * 1000L;
Review Comment:
If task duration is specified as less than 1s (e.g. in embedded tests), will
this become 0? I think we should use `Duration.toMillis()` to be on the
safe-side. Please double check this either way.
##########
embedded-tests/src/test/java/org/apache/druid/testing/embedded/indexing/autoscaler/CostBasedAutoScalerIntegrationTest.java:
##########
@@ -337,7 +398,9 @@ private int getCurrentTaskCount(String supervisorId)
final String getSupervisorPath =
StringUtils.format("/druid/indexer/v1/supervisor/%s", supervisorId);
final KafkaSupervisorSpec supervisorSpec =
cluster.callApi().serviceClient().onLeaderOverlord(
mapper -> new RequestBuilder(HttpMethod.GET, getSupervisorPath),
- new TypeReference<>(){}
+ new TypeReference<>()
+ {
+ }
Review Comment:
Nit: extra change
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/CostBasedAutoScaler.java:
##########
@@ -85,6 +87,16 @@ public class CostBasedAutoScaler implements
SupervisorTaskAutoScaler
private final CostBasedAutoScalerConfig config;
private final ScheduledExecutorService autoscalerExecutor;
private final WeightedCostFunction costFunction;
+ /**
+ * Recent {@code avgProcessingRate} readings, but only from ticks where lag
was present.
+ * "Lag-gated" because lag is the condition that decides whether a reading
counts as evidence
+ * of capacity at all: with no lag the task is only processing as fast as
data arrives (an
+ * upstream supply number, not a capacity number); with lag present, demand
isn't the
+ * bottleneck, so the rate observed is the task's own ceiling. The max of
these samples
+ * becomes {@link CostMetrics#getMaxObservedRate()}.
Review Comment:
This can be simplified:
```suggestion
* When there is no lag i.e. tasks are able to keep up with incoming data,
the processing rate
* is dictated by the rate of incoming data alone and does not denote the
actual processing
* capacity of the task.
```
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/CostBasedAutoScalerConfig.java:
##########
@@ -282,6 +285,15 @@ public boolean isScaleDownOnTaskRolloverOnly()
return scaleDownDuringTaskRolloverOnly;
}
+ /**
+ * Opt-in flag for replacing default idle-ratio with the rate-based
utilization idle ratio.
+ */
+ @JsonProperty("useUtilizationRatio")
+ public boolean shouldUseUtilizationRatio()
Review Comment:
Use the prefix `is` for boolean configs to align with the rest of the druid
code. You can also skip the String literal inside the `@JsonProperty`
annotation.
```suggestion
public boolean isUseUtilizationRatio()
```
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/WeightedCostFunction.java:
##########
@@ -156,6 +157,19 @@ public CostResult computeCost(
return new CostResult(cost, lagCost, weightedIdleCost);
}
+ /**
+ * Derives the current idle ratio from measured utilization ({@code
avgProcessingRate / maxObservedRate}).
+ */
+ static double estimateIdleFromUtilization(CostMetrics metrics)
Review Comment:
Maybe this method should exist in `CostMetrics` itself?
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/WeightedCostFunction.java:
##########
@@ -156,6 +157,19 @@ public CostResult computeCost(
return new CostResult(cost, lagCost, weightedIdleCost);
}
+ /**
+ * Derives the current idle ratio from measured utilization ({@code
avgProcessingRate / maxObservedRate}).
+ */
+ static double estimateIdleFromUtilization(CostMetrics metrics)
+ {
+ final Double maxObservedRate = metrics.getMaxObservedRate();
+ if (maxObservedRate == null || maxObservedRate <= 0) {
+ return IDEAL_IDLE_RATIO;
Review Comment:
So, if there has been no lag for a while, we will assume that we are already
in the optimal usage band?
This would mean that scale down would not happen when, ideally, it should.
I wonder if it would be better to collect samples of `avgProcessingRate`
even when there is no lag, and simply use the max to determine the max observed
rate.
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/CostBasedAutoScaler.java:
##########
@@ -85,6 +87,16 @@ public class CostBasedAutoScaler implements
SupervisorTaskAutoScaler
private final CostBasedAutoScalerConfig config;
private final ScheduledExecutorService autoscalerExecutor;
private final WeightedCostFunction costFunction;
+ /**
+ * Recent {@code avgProcessingRate} readings, but only from ticks where lag
was present.
+ * "Lag-gated" because lag is the condition that decides whether a reading
counts as evidence
+ * of capacity at all: with no lag the task is only processing as fast as
data arrives (an
+ * upstream supply number, not a capacity number); with lag present, demand
isn't the
+ * bottleneck, so the rate observed is the task's own ceiling. The max of
these samples
+ * becomes {@link CostMetrics#getMaxObservedRate()}.
+ */
+ private final EvictingQueue<Double> lagGatedRateSamples;
+ private final int lagGatedRateWindowSize;
Review Comment:
This probably doesn't need to be a member field of this class.
--
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]
