You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@druid.apache.org by "jtuglu-netflix (via GitHub)" <gi...@apache.org> on 2025/05/08 05:59:51 UTC
[PR] Allow task autoscaler to scale to nearest divisor of the partition count (druid)
jtuglu-netflix opened a new pull request, #17988:
URL: https://github.com/apache/druid/pull/17988
<!-- Thanks for trying to help us make Apache Druid be the best it can be! Please fill out as much of the following information as is possible (where relevant, and remove it when irrelevant) to help make the intention and scope of this PR clear in order to ease review. -->
<!-- Please read the doc for contribution (https://github.com/apache/druid/blob/master/CONTRIBUTING.md) before making this PR. Also, once you open a PR, please _avoid using force pushes and rebasing_ since these make it difficult for reviewers to see what you've changed in response to their reviews. See [the 'If your pull request shows conflicts with master' section](https://github.com/apache/druid/blob/master/CONTRIBUTING.md#if-your-pull-request-shows-conflicts-with-master) for more details. -->
Fixes #XXXX.
<!-- Replace XXXX with the id of the issue fixed in this PR. Remove this section if there is no corresponding issue. Don't reference the issue in the title of this pull-request. -->
<!-- If you are a committer, follow the PR action item checklist for committers:
https://github.com/apache/druid/blob/master/dev/committer-instructions.md#pr-and-issue-action-item-checklist-for-committers. -->
### Description
<!-- Describe the goal of this PR, what problem are you fixing. If there is a corresponding issue (referenced above), it's not necessary to repeat the description here, however, you may choose to keep one summary sentence. -->
<!-- Describe your patch: what did you change in code? How did you fix the problem? -->
<!-- If there are several relatively logically separate changes in this PR, create a mini-section for each of them. For example: -->
#### Fixed the bug ...
#### Renamed the class ...
#### Added a forbidden-apis entry ...
<!--
In each section, please describe design decisions made, including:
- Choice of algorithms
- Behavioral aspects. What configuration values are acceptable? How are corner cases and error conditions handled, such as when there are insufficient resources?
- Class organization and design (how the logic is split between classes, inheritance, composition, design patterns)
- Method organization and design (how the logic is split between methods, parameters and return types)
- Naming (class, method, API, configuration, HTTP endpoint, names of emitted metrics)
-->
<!-- It's good to describe an alternative design (or mention an alternative name) for every design (or naming) decision point and compare the alternatives with the designs that you've implemented (or the names you've chosen) to highlight the advantages of the chosen designs and names. -->
<!-- If there was a discussion of the design of the feature implemented in this PR elsewhere (e. g. a "Proposal" issue, any other issue, or a thread in the development mailing list), link to that discussion from this PR description and explain what have changed in your final design compared to your original proposal or the consensus version in the end of the discussion. If something hasn't changed since the original discussion, you can omit a detailed discussion of those aspects of the design here, perhaps apart from brief mentioning for the sake of readability of this PR description. -->
<!-- Some of the aspects mentioned above may be omitted for simple and small changes. -->
#### Release note
<!-- Give your best effort to summarize your changes in a couple of sentences aimed toward Druid users.
If your change doesn't have end user impact, you can skip this section.
For tips about how to write a good release note, see [Release notes](https://github.com/apache/druid/blob/master/CONTRIBUTING.md#release-notes).
-->
<hr>
##### Key changed/added classes in this PR
* `MyFoo`
* `OurBar`
* `TheirBaz`
<hr>
<!-- Check the items by putting "x" in the brackets for the done things. Not all of these items apply to every PR. Remove the items which are not done or not relevant to the PR. None of the items from the checklist below are strictly necessary, but it would be very helpful if you at least self-review the PR. -->
This PR has:
- [ ] been self-reviewed.
- [ ] using the [concurrency checklist](https://github.com/apache/druid/blob/master/dev/code-review/concurrency.md) (Remove this item if the PR doesn't have any relation to concurrency.)
- [ ] added documentation for new or modified features or behaviors.
- [ ] a release note entry in the PR description.
- [ ] added Javadocs for most classes and all non-trivial methods. Linked related entities via Javadoc links.
- [ ] added or updated version, license, or notice information in [licenses.yaml](https://github.com/apache/druid/blob/master/dev/license.md)
- [ ] added comments explaining the "why" and the intent of the code wherever would not be obvious for an unfamiliar reader.
- [ ] added unit tests or modified existing tests to cover new code paths, ensuring the threshold for [code coverage](https://github.com/apache/druid/blob/master/dev/code-review/code-coverage.md) is met.
- [ ] added integration tests.
- [ ] been tested in a test Druid cluster.
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Allow task autoscaler to scale to nearest divisor of the partition count (druid)
Posted by "github-advanced-security[bot] (via GitHub)" <gi...@apache.org>.
github-advanced-security[bot] commented on code in PR #17988:
URL: https://github.com/apache/druid/pull/17988#discussion_r2082636174
##########
extensions-core/kafka-indexing-service/src/test/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisorTest.java:
##########
@@ -5809,6 +5809,12 @@
this.consumerProperties = spec.getIoConfig().getConsumerProperties();
}
+ @Override
+ public int getPartitionCount()
Review Comment:
## Deprecated method or constructor invocation
Invoking [KafkaSupervisorSpec.getIoConfig](1) should be avoided because it has been deprecated.
[Show more details](https://github.com/apache/druid/security/code-scanning/9169)
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Allow task autoscaler to scale to nearest divisor of the partition count (druid)
Posted by "bsyk (via GitHub)" <gi...@apache.org>.
bsyk commented on code in PR #17988:
URL: https://github.com/apache/druid/pull/17988#discussion_r2176205020
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/LagBasedAutoScaler.java:
##########
@@ -200,84 +209,141 @@ private Runnable computeAndCollectLag()
* @param lags the lag metrics of Stream(Kafka/Kinesis)
* @return Integer. target number of tasksCount, -1 means skip scale action.
*/
- private int computeDesiredTaskCount(List<Long> lags)
+ @VisibleForTesting
+ int computeDesiredTaskCount(List<Long> lags)
{
- // if supervisor is not suspended, ensure required tasks are running
- // if suspended, ensure tasks have been requested to gracefully stop
log.debug("Computing desired task count for [%s], based on following lags : [%s]", dataSource, lags);
+ final int currentActiveTaskCount = supervisor.getActiveTaskGroupsCount();
+ final int partitionCount = supervisor.getPartitionCount();
+ if (partitionCount <= 0) {
+ log.warn("Partition number for [%s] <= 0 ? how can it be?", dataSource);
+ return -1;
+ }
+
+ // Cache the factorization in an immutable list for quick lookup later
+ // Partition counts *can* change externally without a new instance of this class being created
+ if (partitionFactors.isEmpty() || partitionCount != partitionFactors.get(partitionFactors.size() - 1)) {
+ log.debug("(Re)computing partitionCount factorization for partitionCount=[%d]", partitionCount);
+ partitionFactors = factorize(partitionCount);
+ }
+
+ Preconditions.checkState(!partitionFactors.isEmpty(), "partitionFactors should not be empty");
+
+ final int desiredActiveTaskCount = computeDesiredTaskCountHelper(lags, currentActiveTaskCount);
+ return applyMinMaxChecks(desiredActiveTaskCount, currentActiveTaskCount, partitionCount);
+ }
+
+ private int computeDesiredTaskCountHelper(final List<Long> lags, final int currentActiveTaskCount)
+ {
int beyond = 0;
int within = 0;
- int metricsCount = lags.size();
- for (Long lag : lags) {
+ final int metricsCount = lags.size();
+ for (final Long lag : lags) {
if (lag >= lagBasedAutoScalerConfig.getScaleOutThreshold()) {
beyond++;
}
if (lag <= lagBasedAutoScalerConfig.getScaleInThreshold()) {
within++;
}
}
- double beyondProportion = beyond * 1.0 / metricsCount;
- double withinProportion = within * 1.0 / metricsCount;
+ final double beyondProportion = beyond * 1.0 / metricsCount;
+ final double withinProportion = within * 1.0 / metricsCount;
- log.debug("Calculated beyondProportion is [%s] and withinProportion is [%s] for dataSource [%s].", beyondProportion,
+ log.debug(
+ "Calculated beyondProportion is [%s] and withinProportion is [%s] for dataSource [%s].", beyondProportion,
withinProportion, dataSource
);
- int currentActiveTaskCount = supervisor.getActiveTaskGroupsCount();
- int desiredActiveTaskCount;
- int partitionCount = supervisor.getPartitionCount();
- if (partitionCount <= 0) {
- log.warn("Partition number for [%s] <= 0 ? how can it be?", dataSource);
+ if (beyondProportion >= lagBasedAutoScalerConfig.getTriggerScaleOutFractionThreshold()) {
+ return currentActiveTaskCount + lagBasedAutoScalerConfig.getScaleOutStep();
+ } else if (withinProportion >= lagBasedAutoScalerConfig.getTriggerScaleInFractionThreshold()) {
+ return currentActiveTaskCount - lagBasedAutoScalerConfig.getScaleInStep();
+ }
+
+ return currentActiveTaskCount;
+ }
+
+
+ private int applyMinMaxChecks(int desiredActiveTaskCount, final int currentActiveTaskCount, final int partitionCount)
+ {
+ // for now, only attempt to scale to nearest factor for scale up
+ if (lagBasedAutoScalerConfig.getUseNearestFactorScaling() && desiredActiveTaskCount > currentActiveTaskCount) {
+ desiredActiveTaskCount = nearestPartitionCountFactor(desiredActiveTaskCount).orElse(desiredActiveTaskCount);
+ }
+
+ if (desiredActiveTaskCount == currentActiveTaskCount) {
+ log.debug("No change in task count for dataSource [%s].", dataSource);
return -1;
}
- if (beyondProportion >= lagBasedAutoScalerConfig.getTriggerScaleOutFractionThreshold()) {
- // Do Scale out
- int taskCount = currentActiveTaskCount + lagBasedAutoScalerConfig.getScaleOutStep();
-
- int actualTaskCountMax = Math.min(lagBasedAutoScalerConfig.getTaskCountMax(), partitionCount);
- if (currentActiveTaskCount == actualTaskCountMax) {
- log.debug("CurrentActiveTaskCount reached task count Max limit, skipping scale out action for dataSource [%s].",
- dataSource
- );
- emitter.emit(metricBuilder
- .setDimension(
- SeekableStreamSupervisor.AUTOSCALER_SKIP_REASON_DIMENSION,
- "Already at max task count"
- )
- .setMetric(SeekableStreamSupervisor.AUTOSCALER_REQUIRED_TASKS_METRIC, taskCount));
- return -1;
- } else {
- desiredActiveTaskCount = Math.min(taskCount, actualTaskCountMax);
- }
- return desiredActiveTaskCount;
+ final int actualTaskCountMax = Math.min(lagBasedAutoScalerConfig.getTaskCountMax(), partitionCount);
+ final int actualTaskCountMin = Math.min(lagBasedAutoScalerConfig.getTaskCountMin(), partitionCount);
+
+ if (currentActiveTaskCount == actualTaskCountMax && currentActiveTaskCount < desiredActiveTaskCount) {
+ log.debug(
+ "CurrentActiveTaskCount reached task count Max limit, skipping scale out action for dataSource[%s].",
+ dataSource
+ );
+ emitSkipMetric("Already at max task count", desiredActiveTaskCount);
+ return -1;
+ } else if (currentActiveTaskCount == actualTaskCountMin && currentActiveTaskCount > desiredActiveTaskCount) {
+ log.debug(
+ "CurrentActiveTaskCount reached task count Min limit, skipping scale in action for dataSource[%s].",
+ dataSource
+ );
+ emitSkipMetric("Already at min task count", desiredActiveTaskCount);
+ return -1;
}
- if (withinProportion >= lagBasedAutoScalerConfig.getTriggerScaleInFractionThreshold()) {
- // Do Scale in
- int taskCount = currentActiveTaskCount - lagBasedAutoScalerConfig.getScaleInStep();
- int actualTaskCountMin = Math.min(lagBasedAutoScalerConfig.getTaskCountMin(), partitionCount);
- if (currentActiveTaskCount == actualTaskCountMin) {
- log.debug("CurrentActiveTaskCount reached task count Min limit, skipping scale in action for dataSource[%s].",
- dataSource
- );
- emitter.emit(metricBuilder
- .setDimension(
- SeekableStreamSupervisor.AUTOSCALER_SKIP_REASON_DIMENSION,
- "Already at min task count"
- )
- .setMetric(SeekableStreamSupervisor.AUTOSCALER_REQUIRED_TASKS_METRIC, taskCount));
- return -1;
+ desiredActiveTaskCount = Math.min(desiredActiveTaskCount, actualTaskCountMax);
+ desiredActiveTaskCount = Math.max(desiredActiveTaskCount, actualTaskCountMin);
+ return desiredActiveTaskCount;
+ }
+
+ private void emitSkipMetric(final String reason, final int taskCount)
+ {
+ emitter.emit(metricBuilder
+ .setDimension(SeekableStreamSupervisor.AUTOSCALER_SKIP_REASON_DIMENSION, reason)
+ .setMetric(SeekableStreamSupervisor.AUTOSCALER_REQUIRED_TASKS_METRIC, taskCount));
+ }
+
+ /*
+ Finds the next largest factor of the topic's partition count that is ≥ `desiredTaskCount`.
+ If no such factor exists, return an empty optional.
+ */
+ private Optional<Integer> nearestPartitionCountFactor(final int desiredTaskCount)
+ {
+ int lo = 0;
+ int hi = partitionFactors.size();
+ while (lo < hi) {
+ final int mid = lo + (hi - lo) / 2;
+ if (partitionFactors.get(mid) >= desiredTaskCount) {
+ hi = mid;
} else {
- desiredActiveTaskCount = Math.max(taskCount, actualTaskCountMin);
+ lo = mid + 1;
}
- return desiredActiveTaskCount;
}
- return -1;
+
+ final Optional<Integer> factor = lo < partitionFactors.size()
+ ? Optional.of(partitionFactors.get(lo))
+ : Optional.empty();
+ log.debug("Given desiredTaskCount=[%d], found nearest task count=[%s]", desiredTaskCount, factor);
+ return factor;
}
public LagBasedAutoScalerConfig getAutoScalerConfig()
{
return lagBasedAutoScalerConfig;
}
+
+ private static ImmutableList<Integer> factorize(final int partitionCount)
+ {
+ final ImmutableList.Builder<Integer> factors = new ImmutableList.Builder<>();
+ for (int i = 1; i <= partitionCount; ++i) {
+ if (partitionCount % i == 0) {
Review Comment:
`0 % anything = 0`
Negative numbers leave the remainder as negative, but otherwise work the same as positive numbers.
We already have a [check](https://github.com/apache/druid/pull/17988/files#diff-f1b33808bb841d1e71e1f5ec3fbaeb3f94899066277b75e192942b66371667ceR218) for <= 0 prior to calling this method.
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Allow task autoscaler to scale to nearest divisor of the partition count (druid)
Posted by "jtuglu1 (via GitHub)" <gi...@apache.org>.
jtuglu1 closed pull request #17988: Allow task autoscaler to scale to nearest divisor of the partition count
URL: https://github.com/apache/druid/pull/17988
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Allow task autoscaler to scale to nearest divisor of the partition count (druid)
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] closed pull request #17988: Allow task autoscaler to scale to nearest divisor of the partition count
URL: https://github.com/apache/druid/pull/17988
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Allow task autoscaler to scale to nearest divisor of the partition count (druid)
Posted by "cryptoe (via GitHub)" <gi...@apache.org>.
cryptoe commented on PR #17988:
URL: https://github.com/apache/druid/pull/17988#issuecomment-3018315090
@jtuglu-netflix
We really want to get this PR in for druid 34.
The code freeze for that is July 8th. Can this make it please :)
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Allow task autoscaler to scale to nearest divisor of the partition count (druid)
Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on code in PR #17988:
URL: https://github.com/apache/druid/pull/17988#discussion_r2095875080
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/LagBasedAutoScaler.java:
##########
@@ -200,84 +209,141 @@ private Runnable computeAndCollectLag()
* @param lags the lag metrics of Stream(Kafka/Kinesis)
* @return Integer. target number of tasksCount, -1 means skip scale action.
*/
- private int computeDesiredTaskCount(List<Long> lags)
+ @VisibleForTesting
+ int computeDesiredTaskCount(List<Long> lags)
{
- // if supervisor is not suspended, ensure required tasks are running
- // if suspended, ensure tasks have been requested to gracefully stop
log.debug("Computing desired task count for [%s], based on following lags : [%s]", dataSource, lags);
+ final int currentActiveTaskCount = supervisor.getActiveTaskGroupsCount();
+ final int partitionCount = supervisor.getPartitionCount();
+ if (partitionCount <= 0) {
+ log.warn("Partition number for [%s] <= 0 ? how can it be?", dataSource);
+ return -1;
+ }
+
+ // Cache the factorization in an immutable list for quick lookup later
+ // Partition counts *can* change externally without a new instance of this class being created
+ if (partitionFactors.isEmpty() || partitionCount != partitionFactors.get(partitionFactors.size() - 1)) {
+ log.debug("(Re)computing partitionCount factorization for partitionCount=[%d]", partitionCount);
Review Comment:
Nit: this log can go into the `factorize` method itself.
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/LagBasedAutoScaler.java:
##########
@@ -200,84 +209,141 @@ private Runnable computeAndCollectLag()
* @param lags the lag metrics of Stream(Kafka/Kinesis)
* @return Integer. target number of tasksCount, -1 means skip scale action.
*/
- private int computeDesiredTaskCount(List<Long> lags)
+ @VisibleForTesting
+ int computeDesiredTaskCount(List<Long> lags)
{
- // if supervisor is not suspended, ensure required tasks are running
- // if suspended, ensure tasks have been requested to gracefully stop
log.debug("Computing desired task count for [%s], based on following lags : [%s]", dataSource, lags);
+ final int currentActiveTaskCount = supervisor.getActiveTaskGroupsCount();
+ final int partitionCount = supervisor.getPartitionCount();
+ if (partitionCount <= 0) {
+ log.warn("Partition number for [%s] <= 0 ? how can it be?", dataSource);
Review Comment:
Maybe this should be an alert rather than a warn?
```suggestion
log.warn("Supervisor[%s] has an invalid number of partitions[%d].", spec.getId(), partitionCount);
```
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/LagBasedAutoScaler.java:
##########
@@ -200,84 +209,141 @@ private Runnable computeAndCollectLag()
* @param lags the lag metrics of Stream(Kafka/Kinesis)
* @return Integer. target number of tasksCount, -1 means skip scale action.
*/
- private int computeDesiredTaskCount(List<Long> lags)
+ @VisibleForTesting
+ int computeDesiredTaskCount(List<Long> lags)
{
- // if supervisor is not suspended, ensure required tasks are running
- // if suspended, ensure tasks have been requested to gracefully stop
log.debug("Computing desired task count for [%s], based on following lags : [%s]", dataSource, lags);
+ final int currentActiveTaskCount = supervisor.getActiveTaskGroupsCount();
+ final int partitionCount = supervisor.getPartitionCount();
+ if (partitionCount <= 0) {
+ log.warn("Partition number for [%s] <= 0 ? how can it be?", dataSource);
+ return -1;
+ }
+
+ // Cache the factorization in an immutable list for quick lookup later
+ // Partition counts *can* change externally without a new instance of this class being created
+ if (partitionFactors.isEmpty() || partitionCount != partitionFactors.get(partitionFactors.size() - 1)) {
+ log.debug("(Re)computing partitionCount factorization for partitionCount=[%d]", partitionCount);
+ partitionFactors = factorize(partitionCount);
+ }
+
+ Preconditions.checkState(!partitionFactors.isEmpty(), "partitionFactors should not be empty");
+
+ final int desiredActiveTaskCount = computeDesiredTaskCountHelper(lags, currentActiveTaskCount);
+ return applyMinMaxChecks(desiredActiveTaskCount, currentActiveTaskCount, partitionCount);
+ }
+
+ private int computeDesiredTaskCountHelper(final List<Long> lags, final int currentActiveTaskCount)
+ {
int beyond = 0;
int within = 0;
- int metricsCount = lags.size();
- for (Long lag : lags) {
+ final int metricsCount = lags.size();
+ for (final Long lag : lags) {
if (lag >= lagBasedAutoScalerConfig.getScaleOutThreshold()) {
beyond++;
}
if (lag <= lagBasedAutoScalerConfig.getScaleInThreshold()) {
within++;
}
}
- double beyondProportion = beyond * 1.0 / metricsCount;
- double withinProportion = within * 1.0 / metricsCount;
+ final double beyondProportion = beyond * 1.0 / metricsCount;
+ final double withinProportion = within * 1.0 / metricsCount;
- log.debug("Calculated beyondProportion is [%s] and withinProportion is [%s] for dataSource [%s].", beyondProportion,
+ log.debug(
+ "Calculated beyondProportion is [%s] and withinProportion is [%s] for dataSource [%s].", beyondProportion,
withinProportion, dataSource
);
- int currentActiveTaskCount = supervisor.getActiveTaskGroupsCount();
- int desiredActiveTaskCount;
- int partitionCount = supervisor.getPartitionCount();
- if (partitionCount <= 0) {
- log.warn("Partition number for [%s] <= 0 ? how can it be?", dataSource);
+ if (beyondProportion >= lagBasedAutoScalerConfig.getTriggerScaleOutFractionThreshold()) {
+ return currentActiveTaskCount + lagBasedAutoScalerConfig.getScaleOutStep();
+ } else if (withinProportion >= lagBasedAutoScalerConfig.getTriggerScaleInFractionThreshold()) {
+ return currentActiveTaskCount - lagBasedAutoScalerConfig.getScaleInStep();
+ }
+
+ return currentActiveTaskCount;
+ }
+
+
+ private int applyMinMaxChecks(int desiredActiveTaskCount, final int currentActiveTaskCount, final int partitionCount)
+ {
+ // for now, only attempt to scale to nearest factor for scale up
+ if (lagBasedAutoScalerConfig.getUseNearestFactorScaling() && desiredActiveTaskCount > currentActiveTaskCount) {
+ desiredActiveTaskCount = nearestPartitionCountFactor(desiredActiveTaskCount).orElse(desiredActiveTaskCount);
+ }
+
+ if (desiredActiveTaskCount == currentActiveTaskCount) {
+ log.debug("No change in task count for dataSource [%s].", dataSource);
return -1;
}
- if (beyondProportion >= lagBasedAutoScalerConfig.getTriggerScaleOutFractionThreshold()) {
- // Do Scale out
- int taskCount = currentActiveTaskCount + lagBasedAutoScalerConfig.getScaleOutStep();
-
- int actualTaskCountMax = Math.min(lagBasedAutoScalerConfig.getTaskCountMax(), partitionCount);
- if (currentActiveTaskCount == actualTaskCountMax) {
- log.debug("CurrentActiveTaskCount reached task count Max limit, skipping scale out action for dataSource [%s].",
- dataSource
- );
- emitter.emit(metricBuilder
- .setDimension(
- SeekableStreamSupervisor.AUTOSCALER_SKIP_REASON_DIMENSION,
- "Already at max task count"
- )
- .setMetric(SeekableStreamSupervisor.AUTOSCALER_REQUIRED_TASKS_METRIC, taskCount));
- return -1;
- } else {
- desiredActiveTaskCount = Math.min(taskCount, actualTaskCountMax);
- }
- return desiredActiveTaskCount;
+ final int actualTaskCountMax = Math.min(lagBasedAutoScalerConfig.getTaskCountMax(), partitionCount);
+ final int actualTaskCountMin = Math.min(lagBasedAutoScalerConfig.getTaskCountMin(), partitionCount);
+
+ if (currentActiveTaskCount == actualTaskCountMax && currentActiveTaskCount < desiredActiveTaskCount) {
+ log.debug(
+ "CurrentActiveTaskCount reached task count Max limit, skipping scale out action for dataSource[%s].",
+ dataSource
+ );
+ emitSkipMetric("Already at max task count", desiredActiveTaskCount);
+ return -1;
+ } else if (currentActiveTaskCount == actualTaskCountMin && currentActiveTaskCount > desiredActiveTaskCount) {
+ log.debug(
+ "CurrentActiveTaskCount reached task count Min limit, skipping scale in action for dataSource[%s].",
+ dataSource
+ );
+ emitSkipMetric("Already at min task count", desiredActiveTaskCount);
+ return -1;
}
- if (withinProportion >= lagBasedAutoScalerConfig.getTriggerScaleInFractionThreshold()) {
- // Do Scale in
- int taskCount = currentActiveTaskCount - lagBasedAutoScalerConfig.getScaleInStep();
- int actualTaskCountMin = Math.min(lagBasedAutoScalerConfig.getTaskCountMin(), partitionCount);
- if (currentActiveTaskCount == actualTaskCountMin) {
- log.debug("CurrentActiveTaskCount reached task count Min limit, skipping scale in action for dataSource[%s].",
- dataSource
- );
- emitter.emit(metricBuilder
- .setDimension(
- SeekableStreamSupervisor.AUTOSCALER_SKIP_REASON_DIMENSION,
- "Already at min task count"
- )
- .setMetric(SeekableStreamSupervisor.AUTOSCALER_REQUIRED_TASKS_METRIC, taskCount));
- return -1;
+ desiredActiveTaskCount = Math.min(desiredActiveTaskCount, actualTaskCountMax);
+ desiredActiveTaskCount = Math.max(desiredActiveTaskCount, actualTaskCountMin);
+ return desiredActiveTaskCount;
+ }
+
+ private void emitSkipMetric(final String reason, final int taskCount)
+ {
+ emitter.emit(metricBuilder
+ .setDimension(SeekableStreamSupervisor.AUTOSCALER_SKIP_REASON_DIMENSION, reason)
+ .setMetric(SeekableStreamSupervisor.AUTOSCALER_REQUIRED_TASKS_METRIC, taskCount));
+ }
+
+ /*
+ Finds the next largest factor of the topic's partition count that is ≥ `desiredTaskCount`.
+ If no such factor exists, return an empty optional.
+ */
+ private Optional<Integer> nearestPartitionCountFactor(final int desiredTaskCount)
+ {
+ int lo = 0;
+ int hi = partitionFactors.size();
+ while (lo < hi) {
+ final int mid = lo + (hi - lo) / 2;
+ if (partitionFactors.get(mid) >= desiredTaskCount) {
+ hi = mid;
} else {
- desiredActiveTaskCount = Math.max(taskCount, actualTaskCountMin);
+ lo = mid + 1;
}
- return desiredActiveTaskCount;
}
- return -1;
+
+ final Optional<Integer> factor = lo < partitionFactors.size()
+ ? Optional.of(partitionFactors.get(lo))
+ : Optional.empty();
+ log.debug("Given desiredTaskCount=[%d], found nearest task count=[%s]", desiredTaskCount, factor);
+ return factor;
}
public LagBasedAutoScalerConfig getAutoScalerConfig()
{
return lagBasedAutoScalerConfig;
}
+
+ private static ImmutableList<Integer> factorize(final int partitionCount)
+ {
+ final ImmutableList.Builder<Integer> factors = new ImmutableList.Builder<>();
+ for (int i = 1; i <= partitionCount; ++i) {
Review Comment:
I feel it's cleaner have this method just take care of the factorization.
Further filtration can be done by the caller as needed.
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/LagBasedAutoScaler.java:
##########
@@ -51,6 +55,7 @@ public class LagBasedAutoScaler implements SupervisorTaskAutoScaler
private final LagBasedAutoScalerConfig lagBasedAutoScalerConfig;
private final ServiceEmitter emitter;
private final ServiceMetricEvent.Builder metricBuilder;
+ private ImmutableList<Integer> partitionFactors;
Review Comment:
As suggested in the other comment, this need not be cached and can be a local variable instead.
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/LagBasedAutoScalerConfig.java:
##########
@@ -100,6 +102,7 @@ public LagBasedAutoScalerConfig(
this.scaleOutStep = scaleOutStep != null ? scaleOutStep : 2;
this.minTriggerScaleActionFrequencyMillis = minTriggerScaleActionFrequencyMillis
!= null ? minTriggerScaleActionFrequencyMillis : 600000;
+ this.useNearestFactorScaling = useNearestFactorScaling != null ? useNearestFactorScaling : false;
Review Comment:
Nit: Probably more readable using `Configs.valueOrDefault()`
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/LagBasedAutoScaler.java:
##########
@@ -69,13 +74,15 @@ public LagBasedAutoScaler(
.getLagCollectionIntervalMillis()) + 1;
this.lagMetricsQueue = new CircularFifoQueue<>(slots);
this.allocationExec = Execs.scheduledSingleThreaded(StringUtils.encodeForFormat(supervisorId) + "-Allocation-%d");
- this.lagComputationExec = Execs.scheduledSingleThreaded(StringUtils.encodeForFormat(supervisorId) + "-Computation-%d");
+ this.lagComputationExec = Execs.scheduledSingleThreaded(StringUtils.encodeForFormat(supervisorId)
+ + "-Computation-%d");
Review Comment:
Nit: Doesn't seem necessary since the previous line is also nearly as long.
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/LagBasedAutoScaler.java:
##########
@@ -200,84 +209,141 @@ private Runnable computeAndCollectLag()
* @param lags the lag metrics of Stream(Kafka/Kinesis)
* @return Integer. target number of tasksCount, -1 means skip scale action.
*/
- private int computeDesiredTaskCount(List<Long> lags)
+ @VisibleForTesting
+ int computeDesiredTaskCount(List<Long> lags)
{
- // if supervisor is not suspended, ensure required tasks are running
- // if suspended, ensure tasks have been requested to gracefully stop
log.debug("Computing desired task count for [%s], based on following lags : [%s]", dataSource, lags);
+ final int currentActiveTaskCount = supervisor.getActiveTaskGroupsCount();
+ final int partitionCount = supervisor.getPartitionCount();
+ if (partitionCount <= 0) {
+ log.warn("Partition number for [%s] <= 0 ? how can it be?", dataSource);
+ return -1;
+ }
+
+ // Cache the factorization in an immutable list for quick lookup later
+ // Partition counts *can* change externally without a new instance of this class being created
+ if (partitionFactors.isEmpty() || partitionCount != partitionFactors.get(partitionFactors.size() - 1)) {
+ log.debug("(Re)computing partitionCount factorization for partitionCount=[%d]", partitionCount);
+ partitionFactors = factorize(partitionCount);
+ }
+
+ Preconditions.checkState(!partitionFactors.isEmpty(), "partitionFactors should not be empty");
+
+ final int desiredActiveTaskCount = computeDesiredTaskCountHelper(lags, currentActiveTaskCount);
+ return applyMinMaxChecks(desiredActiveTaskCount, currentActiveTaskCount, partitionCount);
+ }
+
+ private int computeDesiredTaskCountHelper(final List<Long> lags, final int currentActiveTaskCount)
+ {
int beyond = 0;
int within = 0;
- int metricsCount = lags.size();
- for (Long lag : lags) {
+ final int metricsCount = lags.size();
+ for (final Long lag : lags) {
if (lag >= lagBasedAutoScalerConfig.getScaleOutThreshold()) {
beyond++;
}
if (lag <= lagBasedAutoScalerConfig.getScaleInThreshold()) {
within++;
}
}
- double beyondProportion = beyond * 1.0 / metricsCount;
- double withinProportion = within * 1.0 / metricsCount;
+ final double beyondProportion = beyond * 1.0 / metricsCount;
+ final double withinProportion = within * 1.0 / metricsCount;
- log.debug("Calculated beyondProportion is [%s] and withinProportion is [%s] for dataSource [%s].", beyondProportion,
+ log.debug(
+ "Calculated beyondProportion is [%s] and withinProportion is [%s] for dataSource [%s].", beyondProportion,
withinProportion, dataSource
);
- int currentActiveTaskCount = supervisor.getActiveTaskGroupsCount();
- int desiredActiveTaskCount;
- int partitionCount = supervisor.getPartitionCount();
- if (partitionCount <= 0) {
- log.warn("Partition number for [%s] <= 0 ? how can it be?", dataSource);
+ if (beyondProportion >= lagBasedAutoScalerConfig.getTriggerScaleOutFractionThreshold()) {
+ return currentActiveTaskCount + lagBasedAutoScalerConfig.getScaleOutStep();
+ } else if (withinProportion >= lagBasedAutoScalerConfig.getTriggerScaleInFractionThreshold()) {
+ return currentActiveTaskCount - lagBasedAutoScalerConfig.getScaleInStep();
+ }
+
+ return currentActiveTaskCount;
+ }
+
+
+ private int applyMinMaxChecks(int desiredActiveTaskCount, final int currentActiveTaskCount, final int partitionCount)
+ {
+ // for now, only attempt to scale to nearest factor for scale up
+ if (lagBasedAutoScalerConfig.getUseNearestFactorScaling() && desiredActiveTaskCount > currentActiveTaskCount) {
+ desiredActiveTaskCount = nearestPartitionCountFactor(desiredActiveTaskCount).orElse(desiredActiveTaskCount);
+ }
+
+ if (desiredActiveTaskCount == currentActiveTaskCount) {
+ log.debug("No change in task count for dataSource [%s].", dataSource);
return -1;
}
- if (beyondProportion >= lagBasedAutoScalerConfig.getTriggerScaleOutFractionThreshold()) {
- // Do Scale out
- int taskCount = currentActiveTaskCount + lagBasedAutoScalerConfig.getScaleOutStep();
-
- int actualTaskCountMax = Math.min(lagBasedAutoScalerConfig.getTaskCountMax(), partitionCount);
- if (currentActiveTaskCount == actualTaskCountMax) {
- log.debug("CurrentActiveTaskCount reached task count Max limit, skipping scale out action for dataSource [%s].",
- dataSource
- );
- emitter.emit(metricBuilder
- .setDimension(
- SeekableStreamSupervisor.AUTOSCALER_SKIP_REASON_DIMENSION,
- "Already at max task count"
- )
- .setMetric(SeekableStreamSupervisor.AUTOSCALER_REQUIRED_TASKS_METRIC, taskCount));
- return -1;
- } else {
- desiredActiveTaskCount = Math.min(taskCount, actualTaskCountMax);
- }
- return desiredActiveTaskCount;
+ final int actualTaskCountMax = Math.min(lagBasedAutoScalerConfig.getTaskCountMax(), partitionCount);
+ final int actualTaskCountMin = Math.min(lagBasedAutoScalerConfig.getTaskCountMin(), partitionCount);
+
+ if (currentActiveTaskCount == actualTaskCountMax && currentActiveTaskCount < desiredActiveTaskCount) {
+ log.debug(
+ "CurrentActiveTaskCount reached task count Max limit, skipping scale out action for dataSource[%s].",
+ dataSource
+ );
+ emitSkipMetric("Already at max task count", desiredActiveTaskCount);
+ return -1;
+ } else if (currentActiveTaskCount == actualTaskCountMin && currentActiveTaskCount > desiredActiveTaskCount) {
+ log.debug(
+ "CurrentActiveTaskCount reached task count Min limit, skipping scale in action for dataSource[%s].",
+ dataSource
+ );
+ emitSkipMetric("Already at min task count", desiredActiveTaskCount);
+ return -1;
}
- if (withinProportion >= lagBasedAutoScalerConfig.getTriggerScaleInFractionThreshold()) {
- // Do Scale in
- int taskCount = currentActiveTaskCount - lagBasedAutoScalerConfig.getScaleInStep();
- int actualTaskCountMin = Math.min(lagBasedAutoScalerConfig.getTaskCountMin(), partitionCount);
- if (currentActiveTaskCount == actualTaskCountMin) {
- log.debug("CurrentActiveTaskCount reached task count Min limit, skipping scale in action for dataSource[%s].",
- dataSource
- );
- emitter.emit(metricBuilder
- .setDimension(
- SeekableStreamSupervisor.AUTOSCALER_SKIP_REASON_DIMENSION,
- "Already at min task count"
- )
- .setMetric(SeekableStreamSupervisor.AUTOSCALER_REQUIRED_TASKS_METRIC, taskCount));
- return -1;
+ desiredActiveTaskCount = Math.min(desiredActiveTaskCount, actualTaskCountMax);
+ desiredActiveTaskCount = Math.max(desiredActiveTaskCount, actualTaskCountMin);
+ return desiredActiveTaskCount;
+ }
+
+ private void emitSkipMetric(final String reason, final int taskCount)
+ {
+ emitter.emit(metricBuilder
+ .setDimension(SeekableStreamSupervisor.AUTOSCALER_SKIP_REASON_DIMENSION, reason)
+ .setMetric(SeekableStreamSupervisor.AUTOSCALER_REQUIRED_TASKS_METRIC, taskCount));
+ }
+
+ /*
+ Finds the next largest factor of the topic's partition count that is ≥ `desiredTaskCount`.
+ If no such factor exists, return an empty optional.
+ */
+ private Optional<Integer> nearestPartitionCountFactor(final int desiredTaskCount)
+ {
+ int lo = 0;
+ int hi = partitionFactors.size();
+ while (lo < hi) {
+ final int mid = lo + (hi - lo) / 2;
+ if (partitionFactors.get(mid) >= desiredTaskCount) {
+ hi = mid;
} else {
- desiredActiveTaskCount = Math.max(taskCount, actualTaskCountMin);
+ lo = mid + 1;
}
- return desiredActiveTaskCount;
}
- return -1;
+
+ final Optional<Integer> factor = lo < partitionFactors.size()
+ ? Optional.of(partitionFactors.get(lo))
+ : Optional.empty();
+ log.debug("Given desiredTaskCount=[%d], found nearest task count=[%s]", desiredTaskCount, factor);
+ return factor;
}
public LagBasedAutoScalerConfig getAutoScalerConfig()
{
return lagBasedAutoScalerConfig;
}
+
+ private static ImmutableList<Integer> factorize(final int partitionCount)
+ {
+ final ImmutableList.Builder<Integer> factors = new ImmutableList.Builder<>();
+ for (int i = 1; i <= partitionCount; ++i) {
+ if (partitionCount % i == 0) {
Review Comment:
I haven't tried it out, but how does `%` operator behave with 0 and negative numbers?
For completeness, probably best to have a condition at the top of this method to just return an empty in those cases.
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/LagBasedAutoScaler.java:
##########
@@ -200,84 +209,141 @@ private Runnable computeAndCollectLag()
* @param lags the lag metrics of Stream(Kafka/Kinesis)
* @return Integer. target number of tasksCount, -1 means skip scale action.
*/
- private int computeDesiredTaskCount(List<Long> lags)
+ @VisibleForTesting
+ int computeDesiredTaskCount(List<Long> lags)
{
- // if supervisor is not suspended, ensure required tasks are running
- // if suspended, ensure tasks have been requested to gracefully stop
log.debug("Computing desired task count for [%s], based on following lags : [%s]", dataSource, lags);
+ final int currentActiveTaskCount = supervisor.getActiveTaskGroupsCount();
+ final int partitionCount = supervisor.getPartitionCount();
+ if (partitionCount <= 0) {
+ log.warn("Partition number for [%s] <= 0 ? how can it be?", dataSource);
+ return -1;
+ }
+
+ // Cache the factorization in an immutable list for quick lookup later
+ // Partition counts *can* change externally without a new instance of this class being created
+ if (partitionFactors.isEmpty() || partitionCount != partitionFactors.get(partitionFactors.size() - 1)) {
+ log.debug("(Re)computing partitionCount factorization for partitionCount=[%d]", partitionCount);
+ partitionFactors = factorize(partitionCount);
+ }
+
+ Preconditions.checkState(!partitionFactors.isEmpty(), "partitionFactors should not be empty");
+
+ final int desiredActiveTaskCount = computeDesiredTaskCountHelper(lags, currentActiveTaskCount);
+ return applyMinMaxChecks(desiredActiveTaskCount, currentActiveTaskCount, partitionCount);
+ }
+
+ private int computeDesiredTaskCountHelper(final List<Long> lags, final int currentActiveTaskCount)
+ {
int beyond = 0;
int within = 0;
- int metricsCount = lags.size();
- for (Long lag : lags) {
+ final int metricsCount = lags.size();
+ for (final Long lag : lags) {
if (lag >= lagBasedAutoScalerConfig.getScaleOutThreshold()) {
beyond++;
}
if (lag <= lagBasedAutoScalerConfig.getScaleInThreshold()) {
within++;
}
}
- double beyondProportion = beyond * 1.0 / metricsCount;
- double withinProportion = within * 1.0 / metricsCount;
+ final double beyondProportion = beyond * 1.0 / metricsCount;
+ final double withinProportion = within * 1.0 / metricsCount;
- log.debug("Calculated beyondProportion is [%s] and withinProportion is [%s] for dataSource [%s].", beyondProportion,
+ log.debug(
+ "Calculated beyondProportion is [%s] and withinProportion is [%s] for dataSource [%s].", beyondProportion,
withinProportion, dataSource
);
- int currentActiveTaskCount = supervisor.getActiveTaskGroupsCount();
- int desiredActiveTaskCount;
- int partitionCount = supervisor.getPartitionCount();
- if (partitionCount <= 0) {
- log.warn("Partition number for [%s] <= 0 ? how can it be?", dataSource);
+ if (beyondProportion >= lagBasedAutoScalerConfig.getTriggerScaleOutFractionThreshold()) {
+ return currentActiveTaskCount + lagBasedAutoScalerConfig.getScaleOutStep();
+ } else if (withinProportion >= lagBasedAutoScalerConfig.getTriggerScaleInFractionThreshold()) {
+ return currentActiveTaskCount - lagBasedAutoScalerConfig.getScaleInStep();
+ }
+
+ return currentActiveTaskCount;
+ }
+
+
+ private int applyMinMaxChecks(int desiredActiveTaskCount, final int currentActiveTaskCount, final int partitionCount)
+ {
+ // for now, only attempt to scale to nearest factor for scale up
Review Comment:
We should do it for scale in too. It is okay since this is an experimental config anyway.
The reason we need the nearest factor scaling is to ensure that the task count is _always_ a factor of the number of partitions.
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/LagBasedAutoScaler.java:
##########
@@ -200,84 +209,141 @@ private Runnable computeAndCollectLag()
* @param lags the lag metrics of Stream(Kafka/Kinesis)
* @return Integer. target number of tasksCount, -1 means skip scale action.
*/
- private int computeDesiredTaskCount(List<Long> lags)
+ @VisibleForTesting
+ int computeDesiredTaskCount(List<Long> lags)
{
- // if supervisor is not suspended, ensure required tasks are running
- // if suspended, ensure tasks have been requested to gracefully stop
log.debug("Computing desired task count for [%s], based on following lags : [%s]", dataSource, lags);
+ final int currentActiveTaskCount = supervisor.getActiveTaskGroupsCount();
+ final int partitionCount = supervisor.getPartitionCount();
+ if (partitionCount <= 0) {
+ log.warn("Partition number for [%s] <= 0 ? how can it be?", dataSource);
+ return -1;
+ }
+
+ // Cache the factorization in an immutable list for quick lookup later
+ // Partition counts *can* change externally without a new instance of this class being created
+ if (partitionFactors.isEmpty() || partitionCount != partitionFactors.get(partitionFactors.size() - 1)) {
+ log.debug("(Re)computing partitionCount factorization for partitionCount=[%d]", partitionCount);
+ partitionFactors = factorize(partitionCount);
+ }
+
+ Preconditions.checkState(!partitionFactors.isEmpty(), "partitionFactors should not be empty");
+
+ final int desiredActiveTaskCount = computeDesiredTaskCountHelper(lags, currentActiveTaskCount);
+ return applyMinMaxChecks(desiredActiveTaskCount, currentActiveTaskCount, partitionCount);
+ }
+
+ private int computeDesiredTaskCountHelper(final List<Long> lags, final int currentActiveTaskCount)
+ {
int beyond = 0;
int within = 0;
- int metricsCount = lags.size();
- for (Long lag : lags) {
+ final int metricsCount = lags.size();
+ for (final Long lag : lags) {
if (lag >= lagBasedAutoScalerConfig.getScaleOutThreshold()) {
beyond++;
}
if (lag <= lagBasedAutoScalerConfig.getScaleInThreshold()) {
within++;
}
}
- double beyondProportion = beyond * 1.0 / metricsCount;
- double withinProportion = within * 1.0 / metricsCount;
+ final double beyondProportion = beyond * 1.0 / metricsCount;
+ final double withinProportion = within * 1.0 / metricsCount;
- log.debug("Calculated beyondProportion is [%s] and withinProportion is [%s] for dataSource [%s].", beyondProportion,
+ log.debug(
+ "Calculated beyondProportion is [%s] and withinProportion is [%s] for dataSource [%s].", beyondProportion,
Review Comment:
```suggestion
"Calculated beyondProportion[%s] and withinProportion[%s] for supervisor[%s].", beyondProportion,
```
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/LagBasedAutoScaler.java:
##########
@@ -200,84 +209,141 @@ private Runnable computeAndCollectLag()
* @param lags the lag metrics of Stream(Kafka/Kinesis)
* @return Integer. target number of tasksCount, -1 means skip scale action.
*/
- private int computeDesiredTaskCount(List<Long> lags)
+ @VisibleForTesting
+ int computeDesiredTaskCount(List<Long> lags)
{
- // if supervisor is not suspended, ensure required tasks are running
- // if suspended, ensure tasks have been requested to gracefully stop
log.debug("Computing desired task count for [%s], based on following lags : [%s]", dataSource, lags);
+ final int currentActiveTaskCount = supervisor.getActiveTaskGroupsCount();
+ final int partitionCount = supervisor.getPartitionCount();
+ if (partitionCount <= 0) {
+ log.warn("Partition number for [%s] <= 0 ? how can it be?", dataSource);
+ return -1;
+ }
+
+ // Cache the factorization in an immutable list for quick lookup later
+ // Partition counts *can* change externally without a new instance of this class being created
+ if (partitionFactors.isEmpty() || partitionCount != partitionFactors.get(partitionFactors.size() - 1)) {
+ log.debug("(Re)computing partitionCount factorization for partitionCount=[%d]", partitionCount);
+ partitionFactors = factorize(partitionCount);
+ }
+
+ Preconditions.checkState(!partitionFactors.isEmpty(), "partitionFactors should not be empty");
+
+ final int desiredActiveTaskCount = computeDesiredTaskCountHelper(lags, currentActiveTaskCount);
+ return applyMinMaxChecks(desiredActiveTaskCount, currentActiveTaskCount, partitionCount);
+ }
+
+ private int computeDesiredTaskCountHelper(final List<Long> lags, final int currentActiveTaskCount)
+ {
int beyond = 0;
int within = 0;
- int metricsCount = lags.size();
- for (Long lag : lags) {
+ final int metricsCount = lags.size();
+ for (final Long lag : lags) {
if (lag >= lagBasedAutoScalerConfig.getScaleOutThreshold()) {
beyond++;
}
if (lag <= lagBasedAutoScalerConfig.getScaleInThreshold()) {
within++;
}
}
- double beyondProportion = beyond * 1.0 / metricsCount;
- double withinProportion = within * 1.0 / metricsCount;
+ final double beyondProportion = beyond * 1.0 / metricsCount;
+ final double withinProportion = within * 1.0 / metricsCount;
- log.debug("Calculated beyondProportion is [%s] and withinProportion is [%s] for dataSource [%s].", beyondProportion,
+ log.debug(
+ "Calculated beyondProportion is [%s] and withinProportion is [%s] for dataSource [%s].", beyondProportion,
withinProportion, dataSource
Review Comment:
```suggestion
withinProportion, spec.getId()
```
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/LagBasedAutoScaler.java:
##########
@@ -200,84 +209,141 @@ private Runnable computeAndCollectLag()
* @param lags the lag metrics of Stream(Kafka/Kinesis)
* @return Integer. target number of tasksCount, -1 means skip scale action.
*/
- private int computeDesiredTaskCount(List<Long> lags)
+ @VisibleForTesting
+ int computeDesiredTaskCount(List<Long> lags)
{
- // if supervisor is not suspended, ensure required tasks are running
- // if suspended, ensure tasks have been requested to gracefully stop
log.debug("Computing desired task count for [%s], based on following lags : [%s]", dataSource, lags);
+ final int currentActiveTaskCount = supervisor.getActiveTaskGroupsCount();
+ final int partitionCount = supervisor.getPartitionCount();
+ if (partitionCount <= 0) {
+ log.warn("Partition number for [%s] <= 0 ? how can it be?", dataSource);
+ return -1;
+ }
+
+ // Cache the factorization in an immutable list for quick lookup later
+ // Partition counts *can* change externally without a new instance of this class being created
+ if (partitionFactors.isEmpty() || partitionCount != partitionFactors.get(partitionFactors.size() - 1)) {
Review Comment:
If the computation is fast enough, do we even need to cache it?
Even if the partition count was 1k, the computation would finish in some µs.
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Allow task autoscaler to scale to nearest divisor of the partition count (druid)
Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on code in PR #17988:
URL: https://github.com/apache/druid/pull/17988#discussion_r2175558914
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/LagBasedAutoScaler.java:
##########
@@ -200,84 +209,141 @@ private Runnable computeAndCollectLag()
* @param lags the lag metrics of Stream(Kafka/Kinesis)
* @return Integer. target number of tasksCount, -1 means skip scale action.
*/
- private int computeDesiredTaskCount(List<Long> lags)
+ @VisibleForTesting
+ int computeDesiredTaskCount(List<Long> lags)
{
- // if supervisor is not suspended, ensure required tasks are running
- // if suspended, ensure tasks have been requested to gracefully stop
log.debug("Computing desired task count for [%s], based on following lags : [%s]", dataSource, lags);
+ final int currentActiveTaskCount = supervisor.getActiveTaskGroupsCount();
+ final int partitionCount = supervisor.getPartitionCount();
+ if (partitionCount <= 0) {
+ log.warn("Partition number for [%s] <= 0 ? how can it be?", dataSource);
+ return -1;
+ }
+
+ // Cache the factorization in an immutable list for quick lookup later
+ // Partition counts *can* change externally without a new instance of this class being created
+ if (partitionFactors.isEmpty() || partitionCount != partitionFactors.get(partitionFactors.size() - 1)) {
+ log.debug("(Re)computing partitionCount factorization for partitionCount=[%d]", partitionCount);
+ partitionFactors = factorize(partitionCount);
+ }
+
+ Preconditions.checkState(!partitionFactors.isEmpty(), "partitionFactors should not be empty");
+
+ final int desiredActiveTaskCount = computeDesiredTaskCountHelper(lags, currentActiveTaskCount);
+ return applyMinMaxChecks(desiredActiveTaskCount, currentActiveTaskCount, partitionCount);
+ }
+
+ private int computeDesiredTaskCountHelper(final List<Long> lags, final int currentActiveTaskCount)
+ {
int beyond = 0;
int within = 0;
- int metricsCount = lags.size();
- for (Long lag : lags) {
+ final int metricsCount = lags.size();
+ for (final Long lag : lags) {
if (lag >= lagBasedAutoScalerConfig.getScaleOutThreshold()) {
beyond++;
}
if (lag <= lagBasedAutoScalerConfig.getScaleInThreshold()) {
within++;
}
}
- double beyondProportion = beyond * 1.0 / metricsCount;
- double withinProportion = within * 1.0 / metricsCount;
+ final double beyondProportion = beyond * 1.0 / metricsCount;
+ final double withinProportion = within * 1.0 / metricsCount;
- log.debug("Calculated beyondProportion is [%s] and withinProportion is [%s] for dataSource [%s].", beyondProportion,
+ log.debug(
+ "Calculated beyondProportion is [%s] and withinProportion is [%s] for dataSource [%s].", beyondProportion,
withinProportion, dataSource
);
- int currentActiveTaskCount = supervisor.getActiveTaskGroupsCount();
- int desiredActiveTaskCount;
- int partitionCount = supervisor.getPartitionCount();
- if (partitionCount <= 0) {
- log.warn("Partition number for [%s] <= 0 ? how can it be?", dataSource);
+ if (beyondProportion >= lagBasedAutoScalerConfig.getTriggerScaleOutFractionThreshold()) {
+ return currentActiveTaskCount + lagBasedAutoScalerConfig.getScaleOutStep();
+ } else if (withinProportion >= lagBasedAutoScalerConfig.getTriggerScaleInFractionThreshold()) {
+ return currentActiveTaskCount - lagBasedAutoScalerConfig.getScaleInStep();
+ }
+
+ return currentActiveTaskCount;
+ }
+
+
+ private int applyMinMaxChecks(int desiredActiveTaskCount, final int currentActiveTaskCount, final int partitionCount)
+ {
+ // for now, only attempt to scale to nearest factor for scale up
Review Comment:
Yes, I agree that sometimes the scale down (or even scale up) can be too aggressive, which is why I mentioned that at every evaluation, we would need to check if the increase or decrease in lag is enough to justify changing the task count.
The change can be especially drastic if someone decides to have a prime number of partitions! 😅
But having a mix of factors and fixed scale in/scale out step size is probably not the way to go either.
Having an unequal number of partitions assigned to tasks is known to cause issues, which is why the factor-based task count was considered in the first place.
@uds5501 is also working in this area to identify the causes of increase in lag when task count changes.
In conjunction to that, it would also be nice to have some setup where we can compare multiple strategies and determine if one is empirically better than the other. It is difficult to gauge just by the theoretical distribution of partitions if one distribution would really perform better or worse than another in practice.
Until then, I don't think we should rush the change in this patch since there are multiple factors at play here.
@cryptoe , what are your thoughts?
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Allow task autoscaler to scale to nearest divisor of the partition count (druid)
Posted by "jtuglu-netflix (via GitHub)" <gi...@apache.org>.
jtuglu-netflix commented on code in PR #17988:
URL: https://github.com/apache/druid/pull/17988#discussion_r2175466168
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/LagBasedAutoScaler.java:
##########
@@ -200,84 +209,141 @@ private Runnable computeAndCollectLag()
* @param lags the lag metrics of Stream(Kafka/Kinesis)
* @return Integer. target number of tasksCount, -1 means skip scale action.
*/
- private int computeDesiredTaskCount(List<Long> lags)
+ @VisibleForTesting
+ int computeDesiredTaskCount(List<Long> lags)
{
- // if supervisor is not suspended, ensure required tasks are running
- // if suspended, ensure tasks have been requested to gracefully stop
log.debug("Computing desired task count for [%s], based on following lags : [%s]", dataSource, lags);
+ final int currentActiveTaskCount = supervisor.getActiveTaskGroupsCount();
+ final int partitionCount = supervisor.getPartitionCount();
+ if (partitionCount <= 0) {
+ log.warn("Partition number for [%s] <= 0 ? how can it be?", dataSource);
+ return -1;
+ }
+
+ // Cache the factorization in an immutable list for quick lookup later
+ // Partition counts *can* change externally without a new instance of this class being created
+ if (partitionFactors.isEmpty() || partitionCount != partitionFactors.get(partitionFactors.size() - 1)) {
Review Comment:
I don't really like the idea of doing repeated work in a method that could be called on an arbitrary period specified by the user (we might call this 4 times a minute, etc.) and the effort to cache these things is low. I can remove though.
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Allow task autoscaler to scale to nearest divisor of the partition count (druid)
Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on PR #17988:
URL: https://github.com/apache/druid/pull/17988#issuecomment-2929051243
@jtuglu-netflix , I was a little caught with some other items.
I will finish the review of this PR this week.
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Allow task autoscaler to scale to nearest divisor of the partition count (druid)
Posted by "jtuglu-netflix (via GitHub)" <gi...@apache.org>.
jtuglu-netflix commented on code in PR #17988:
URL: https://github.com/apache/druid/pull/17988#discussion_r2175452459
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/LagBasedAutoScaler.java:
##########
@@ -200,84 +209,141 @@ private Runnable computeAndCollectLag()
* @param lags the lag metrics of Stream(Kafka/Kinesis)
* @return Integer. target number of tasksCount, -1 means skip scale action.
*/
- private int computeDesiredTaskCount(List<Long> lags)
+ @VisibleForTesting
+ int computeDesiredTaskCount(List<Long> lags)
{
- // if supervisor is not suspended, ensure required tasks are running
- // if suspended, ensure tasks have been requested to gracefully stop
log.debug("Computing desired task count for [%s], based on following lags : [%s]", dataSource, lags);
+ final int currentActiveTaskCount = supervisor.getActiveTaskGroupsCount();
+ final int partitionCount = supervisor.getPartitionCount();
+ if (partitionCount <= 0) {
+ log.warn("Partition number for [%s] <= 0 ? how can it be?", dataSource);
+ return -1;
+ }
+
+ // Cache the factorization in an immutable list for quick lookup later
+ // Partition counts *can* change externally without a new instance of this class being created
+ if (partitionFactors.isEmpty() || partitionCount != partitionFactors.get(partitionFactors.size() - 1)) {
+ log.debug("(Re)computing partitionCount factorization for partitionCount=[%d]", partitionCount);
+ partitionFactors = factorize(partitionCount);
+ }
+
+ Preconditions.checkState(!partitionFactors.isEmpty(), "partitionFactors should not be empty");
+
+ final int desiredActiveTaskCount = computeDesiredTaskCountHelper(lags, currentActiveTaskCount);
+ return applyMinMaxChecks(desiredActiveTaskCount, currentActiveTaskCount, partitionCount);
+ }
+
+ private int computeDesiredTaskCountHelper(final List<Long> lags, final int currentActiveTaskCount)
+ {
int beyond = 0;
int within = 0;
- int metricsCount = lags.size();
- for (Long lag : lags) {
+ final int metricsCount = lags.size();
+ for (final Long lag : lags) {
if (lag >= lagBasedAutoScalerConfig.getScaleOutThreshold()) {
beyond++;
}
if (lag <= lagBasedAutoScalerConfig.getScaleInThreshold()) {
within++;
}
}
- double beyondProportion = beyond * 1.0 / metricsCount;
- double withinProportion = within * 1.0 / metricsCount;
+ final double beyondProportion = beyond * 1.0 / metricsCount;
+ final double withinProportion = within * 1.0 / metricsCount;
- log.debug("Calculated beyondProportion is [%s] and withinProportion is [%s] for dataSource [%s].", beyondProportion,
+ log.debug(
+ "Calculated beyondProportion is [%s] and withinProportion is [%s] for dataSource [%s].", beyondProportion,
withinProportion, dataSource
);
- int currentActiveTaskCount = supervisor.getActiveTaskGroupsCount();
- int desiredActiveTaskCount;
- int partitionCount = supervisor.getPartitionCount();
- if (partitionCount <= 0) {
- log.warn("Partition number for [%s] <= 0 ? how can it be?", dataSource);
+ if (beyondProportion >= lagBasedAutoScalerConfig.getTriggerScaleOutFractionThreshold()) {
+ return currentActiveTaskCount + lagBasedAutoScalerConfig.getScaleOutStep();
+ } else if (withinProportion >= lagBasedAutoScalerConfig.getTriggerScaleInFractionThreshold()) {
+ return currentActiveTaskCount - lagBasedAutoScalerConfig.getScaleInStep();
+ }
+
+ return currentActiveTaskCount;
+ }
+
+
+ private int applyMinMaxChecks(int desiredActiveTaskCount, final int currentActiveTaskCount, final int partitionCount)
+ {
+ // for now, only attempt to scale to nearest factor for scale up
Review Comment:
I don't agree that it's *always* better to scale to the next lowest factor. In fact, it will cause ≥ as much lag as scaling down the normal step count. This is because you risk scaling down more than you need to which in the worst case can leave a task covering N partitions (the max) which is provably ≥ the # of partitions you would get by just scaling down a fixed step.
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Allow task autoscaler to scale to nearest divisor of the partition count (druid)
Posted by "jtuglu-netflix (via GitHub)" <gi...@apache.org>.
jtuglu-netflix commented on PR #17988:
URL: https://github.com/apache/druid/pull/17988#issuecomment-2889299305
@kfaraz did you have a chance to look at this?
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Allow task autoscaler to scale to nearest divisor of the partition count (druid)
Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on PR #17988:
URL: https://github.com/apache/druid/pull/17988#issuecomment-2890129745
@jtuglu-netflix , have been a little occupied lately, will take a look at this either later today or tomorrow.
Thanks for your patience!
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Allow task autoscaler to scale to nearest divisor of the partition count (druid)
Posted by "jtuglu-netflix (via GitHub)" <gi...@apache.org>.
jtuglu-netflix commented on PR #17988:
URL: https://github.com/apache/druid/pull/17988#issuecomment-2906161268
@kfaraz any thoughts here?
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Allow task autoscaler to scale to nearest divisor of the partition count (druid)
Posted by "jtuglu-netflix (via GitHub)" <gi...@apache.org>.
jtuglu-netflix commented on code in PR #17988:
URL: https://github.com/apache/druid/pull/17988#discussion_r2175452459
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/LagBasedAutoScaler.java:
##########
@@ -200,84 +209,141 @@ private Runnable computeAndCollectLag()
* @param lags the lag metrics of Stream(Kafka/Kinesis)
* @return Integer. target number of tasksCount, -1 means skip scale action.
*/
- private int computeDesiredTaskCount(List<Long> lags)
+ @VisibleForTesting
+ int computeDesiredTaskCount(List<Long> lags)
{
- // if supervisor is not suspended, ensure required tasks are running
- // if suspended, ensure tasks have been requested to gracefully stop
log.debug("Computing desired task count for [%s], based on following lags : [%s]", dataSource, lags);
+ final int currentActiveTaskCount = supervisor.getActiveTaskGroupsCount();
+ final int partitionCount = supervisor.getPartitionCount();
+ if (partitionCount <= 0) {
+ log.warn("Partition number for [%s] <= 0 ? how can it be?", dataSource);
+ return -1;
+ }
+
+ // Cache the factorization in an immutable list for quick lookup later
+ // Partition counts *can* change externally without a new instance of this class being created
+ if (partitionFactors.isEmpty() || partitionCount != partitionFactors.get(partitionFactors.size() - 1)) {
+ log.debug("(Re)computing partitionCount factorization for partitionCount=[%d]", partitionCount);
+ partitionFactors = factorize(partitionCount);
+ }
+
+ Preconditions.checkState(!partitionFactors.isEmpty(), "partitionFactors should not be empty");
+
+ final int desiredActiveTaskCount = computeDesiredTaskCountHelper(lags, currentActiveTaskCount);
+ return applyMinMaxChecks(desiredActiveTaskCount, currentActiveTaskCount, partitionCount);
+ }
+
+ private int computeDesiredTaskCountHelper(final List<Long> lags, final int currentActiveTaskCount)
+ {
int beyond = 0;
int within = 0;
- int metricsCount = lags.size();
- for (Long lag : lags) {
+ final int metricsCount = lags.size();
+ for (final Long lag : lags) {
if (lag >= lagBasedAutoScalerConfig.getScaleOutThreshold()) {
beyond++;
}
if (lag <= lagBasedAutoScalerConfig.getScaleInThreshold()) {
within++;
}
}
- double beyondProportion = beyond * 1.0 / metricsCount;
- double withinProportion = within * 1.0 / metricsCount;
+ final double beyondProportion = beyond * 1.0 / metricsCount;
+ final double withinProportion = within * 1.0 / metricsCount;
- log.debug("Calculated beyondProportion is [%s] and withinProportion is [%s] for dataSource [%s].", beyondProportion,
+ log.debug(
+ "Calculated beyondProportion is [%s] and withinProportion is [%s] for dataSource [%s].", beyondProportion,
withinProportion, dataSource
);
- int currentActiveTaskCount = supervisor.getActiveTaskGroupsCount();
- int desiredActiveTaskCount;
- int partitionCount = supervisor.getPartitionCount();
- if (partitionCount <= 0) {
- log.warn("Partition number for [%s] <= 0 ? how can it be?", dataSource);
+ if (beyondProportion >= lagBasedAutoScalerConfig.getTriggerScaleOutFractionThreshold()) {
+ return currentActiveTaskCount + lagBasedAutoScalerConfig.getScaleOutStep();
+ } else if (withinProportion >= lagBasedAutoScalerConfig.getTriggerScaleInFractionThreshold()) {
+ return currentActiveTaskCount - lagBasedAutoScalerConfig.getScaleInStep();
+ }
+
+ return currentActiveTaskCount;
+ }
+
+
+ private int applyMinMaxChecks(int desiredActiveTaskCount, final int currentActiveTaskCount, final int partitionCount)
+ {
+ // for now, only attempt to scale to nearest factor for scale up
Review Comment:
I don't agree that it's *always* better to scale to the next lowest factor. In fact, it will cause ≥ as much lag as scaling down the normal step count. This is because you risk scaling down more than you need to which in the worst case can leave a task covering N partitions (the max) which is provably ≥ the max # of partitions you would get by just scaling down a fixed step.
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Allow task autoscaler to scale to nearest divisor of the partition count (druid)
Posted by "bsyk (via GitHub)" <gi...@apache.org>.
bsyk commented on code in PR #17988:
URL: https://github.com/apache/druid/pull/17988#discussion_r2176201417
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/LagBasedAutoScaler.java:
##########
@@ -200,84 +209,141 @@ private Runnable computeAndCollectLag()
* @param lags the lag metrics of Stream(Kafka/Kinesis)
* @return Integer. target number of tasksCount, -1 means skip scale action.
*/
- private int computeDesiredTaskCount(List<Long> lags)
+ @VisibleForTesting
+ int computeDesiredTaskCount(List<Long> lags)
{
- // if supervisor is not suspended, ensure required tasks are running
- // if suspended, ensure tasks have been requested to gracefully stop
log.debug("Computing desired task count for [%s], based on following lags : [%s]", dataSource, lags);
+ final int currentActiveTaskCount = supervisor.getActiveTaskGroupsCount();
+ final int partitionCount = supervisor.getPartitionCount();
+ if (partitionCount <= 0) {
+ log.warn("Partition number for [%s] <= 0 ? how can it be?", dataSource);
+ return -1;
+ }
+
+ // Cache the factorization in an immutable list for quick lookup later
+ // Partition counts *can* change externally without a new instance of this class being created
+ if (partitionFactors.isEmpty() || partitionCount != partitionFactors.get(partitionFactors.size() - 1)) {
+ log.debug("(Re)computing partitionCount factorization for partitionCount=[%d]", partitionCount);
+ partitionFactors = factorize(partitionCount);
+ }
+
+ Preconditions.checkState(!partitionFactors.isEmpty(), "partitionFactors should not be empty");
+
+ final int desiredActiveTaskCount = computeDesiredTaskCountHelper(lags, currentActiveTaskCount);
+ return applyMinMaxChecks(desiredActiveTaskCount, currentActiveTaskCount, partitionCount);
+ }
+
+ private int computeDesiredTaskCountHelper(final List<Long> lags, final int currentActiveTaskCount)
+ {
int beyond = 0;
int within = 0;
- int metricsCount = lags.size();
- for (Long lag : lags) {
+ final int metricsCount = lags.size();
+ for (final Long lag : lags) {
if (lag >= lagBasedAutoScalerConfig.getScaleOutThreshold()) {
beyond++;
}
if (lag <= lagBasedAutoScalerConfig.getScaleInThreshold()) {
within++;
}
}
- double beyondProportion = beyond * 1.0 / metricsCount;
- double withinProportion = within * 1.0 / metricsCount;
+ final double beyondProportion = beyond * 1.0 / metricsCount;
+ final double withinProportion = within * 1.0 / metricsCount;
- log.debug("Calculated beyondProportion is [%s] and withinProportion is [%s] for dataSource [%s].", beyondProportion,
+ log.debug(
+ "Calculated beyondProportion is [%s] and withinProportion is [%s] for dataSource [%s].", beyondProportion,
withinProportion, dataSource
);
- int currentActiveTaskCount = supervisor.getActiveTaskGroupsCount();
- int desiredActiveTaskCount;
- int partitionCount = supervisor.getPartitionCount();
- if (partitionCount <= 0) {
- log.warn("Partition number for [%s] <= 0 ? how can it be?", dataSource);
+ if (beyondProportion >= lagBasedAutoScalerConfig.getTriggerScaleOutFractionThreshold()) {
+ return currentActiveTaskCount + lagBasedAutoScalerConfig.getScaleOutStep();
+ } else if (withinProportion >= lagBasedAutoScalerConfig.getTriggerScaleInFractionThreshold()) {
+ return currentActiveTaskCount - lagBasedAutoScalerConfig.getScaleInStep();
+ }
+
+ return currentActiveTaskCount;
+ }
+
+
+ private int applyMinMaxChecks(int desiredActiveTaskCount, final int currentActiveTaskCount, final int partitionCount)
+ {
+ // for now, only attempt to scale to nearest factor for scale up
Review Comment:
> This is because you risk scaling down more than you need to
If we scale to the next largest _above_ the desired, we won't scale down too far. And we'd protect some indexers from having more partitions assigned than others.
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Allow task autoscaler to scale to nearest divisor of the partition count (druid)
Posted by "jtuglu-netflix (via GitHub)" <gi...@apache.org>.
jtuglu-netflix commented on code in PR #17988:
URL: https://github.com/apache/druid/pull/17988#discussion_r2175466168
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/LagBasedAutoScaler.java:
##########
@@ -200,84 +209,141 @@ private Runnable computeAndCollectLag()
* @param lags the lag metrics of Stream(Kafka/Kinesis)
* @return Integer. target number of tasksCount, -1 means skip scale action.
*/
- private int computeDesiredTaskCount(List<Long> lags)
+ @VisibleForTesting
+ int computeDesiredTaskCount(List<Long> lags)
{
- // if supervisor is not suspended, ensure required tasks are running
- // if suspended, ensure tasks have been requested to gracefully stop
log.debug("Computing desired task count for [%s], based on following lags : [%s]", dataSource, lags);
+ final int currentActiveTaskCount = supervisor.getActiveTaskGroupsCount();
+ final int partitionCount = supervisor.getPartitionCount();
+ if (partitionCount <= 0) {
+ log.warn("Partition number for [%s] <= 0 ? how can it be?", dataSource);
+ return -1;
+ }
+
+ // Cache the factorization in an immutable list for quick lookup later
+ // Partition counts *can* change externally without a new instance of this class being created
+ if (partitionFactors.isEmpty() || partitionCount != partitionFactors.get(partitionFactors.size() - 1)) {
Review Comment:
I don't really like the idea of doing repeated work in a method that could be called on an arbitrary period specified by the user (we might call this 4 times a minute, etc.).
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Allow task autoscaler to scale to nearest divisor of the partition count (druid)
Posted by "github-advanced-security[bot] (via GitHub)" <gi...@apache.org>.
github-advanced-security[bot] commented on code in PR #17988:
URL: https://github.com/apache/druid/pull/17988#discussion_r2081066043
##########
indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/LagBasedAutoScalerTest.java:
##########
@@ -0,0 +1,357 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.indexing.seekablestream.supervisor.autoscaler;
+
+import com.google.common.collect.ImmutableList;
+import org.apache.druid.indexing.overlord.supervisor.SupervisorSpec;
+import org.apache.druid.indexing.overlord.supervisor.autoscaler.AggregateFunction;
+import org.apache.druid.indexing.overlord.supervisor.autoscaler.LagStats;
+import org.apache.druid.indexing.seekablestream.supervisor.SeekableStreamSupervisor;
+import org.apache.druid.indexing.seekablestream.supervisor.SeekableStreamSupervisorIOConfig;
+import org.apache.druid.java.util.emitter.service.ServiceEmitter;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.ArgumentMatchers;
+import org.mockito.Mock;
+import org.mockito.Mockito;
+import org.mockito.MockitoAnnotations;
+
+import java.util.List;
+import java.util.concurrent.Callable;
+
+public class LagBasedAutoScalerTest
+{
+ private static final String DATASOURCE = "testDataSource";
+ private static final String STREAM = "testStream";
+
+ private static final int DEFAULT_MIN_TASK_COUNT = 0;
+ private static final int DEFAULT_SCALE_IN = 1;
+ private static final int DEFAULT_SCALE_OUT = 1;
+ private static final long DEFAULT_SCALE_IN_THRESHOLD = 100L;
+ private static final long DEFAULT_SCALE_OUT_THRESHOLD = 1000L;
+ private static final double DEFAULT_SCALE_IN_THRESHOLD_PCT = 0.9;
+ private static final double DEFAULT_SCALE_OUT_THRESHOLD_PCT = 0.2;
+
+ @Mock
+ private SeekableStreamSupervisor supervisor;
+
+ @Mock
+ private SupervisorSpec spec;
+
+ @Mock
+ private ServiceEmitter emitter;
+
+ @Mock
+ private SeekableStreamSupervisorIOConfig ioConfig;
+
+ private LagBasedAutoScalerConfig defaultConfig;
+
+ @Before
+ public void setup()
+ {
+ MockitoAnnotations.initMocks(this);
Review Comment:
## Deprecated method or constructor invocation
Invoking [MockitoAnnotations.initMocks](1) should be avoided because it has been deprecated.
[Show more details](https://github.com/apache/druid/security/code-scanning/9167)
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Allow task autoscaler to scale to nearest divisor of the partition count (druid)
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #17988:
URL: https://github.com/apache/druid/pull/17988#issuecomment-3507236654
This pull request has been marked as stale due to 60 days of inactivity.
It will be closed in 4 weeks if no further activity occurs. If you think
that's incorrect or this pull request should instead be reviewed, please simply
write any comment. Even if closed, you can still revive the PR at any time or
discuss it on the dev@druid.apache.org list.
Thank you for your contributions.
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Allow task autoscaler to scale to nearest divisor of the partition count (druid)
Posted by "jtuglu-netflix (via GitHub)" <gi...@apache.org>.
jtuglu-netflix commented on PR #17988:
URL: https://github.com/apache/druid/pull/17988#issuecomment-2890139774
> @jtuglu-netflix , have been a little occupied lately, will take a look at this either later today or tomorrow. Thanks for your patience!
Sounds good, thank you!
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Allow task autoscaler to scale to nearest divisor of the partition count (druid)
Posted by "bsyk (via GitHub)" <gi...@apache.org>.
bsyk commented on code in PR #17988:
URL: https://github.com/apache/druid/pull/17988#discussion_r2176201417
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/LagBasedAutoScaler.java:
##########
@@ -200,84 +209,141 @@ private Runnable computeAndCollectLag()
* @param lags the lag metrics of Stream(Kafka/Kinesis)
* @return Integer. target number of tasksCount, -1 means skip scale action.
*/
- private int computeDesiredTaskCount(List<Long> lags)
+ @VisibleForTesting
+ int computeDesiredTaskCount(List<Long> lags)
{
- // if supervisor is not suspended, ensure required tasks are running
- // if suspended, ensure tasks have been requested to gracefully stop
log.debug("Computing desired task count for [%s], based on following lags : [%s]", dataSource, lags);
+ final int currentActiveTaskCount = supervisor.getActiveTaskGroupsCount();
+ final int partitionCount = supervisor.getPartitionCount();
+ if (partitionCount <= 0) {
+ log.warn("Partition number for [%s] <= 0 ? how can it be?", dataSource);
+ return -1;
+ }
+
+ // Cache the factorization in an immutable list for quick lookup later
+ // Partition counts *can* change externally without a new instance of this class being created
+ if (partitionFactors.isEmpty() || partitionCount != partitionFactors.get(partitionFactors.size() - 1)) {
+ log.debug("(Re)computing partitionCount factorization for partitionCount=[%d]", partitionCount);
+ partitionFactors = factorize(partitionCount);
+ }
+
+ Preconditions.checkState(!partitionFactors.isEmpty(), "partitionFactors should not be empty");
+
+ final int desiredActiveTaskCount = computeDesiredTaskCountHelper(lags, currentActiveTaskCount);
+ return applyMinMaxChecks(desiredActiveTaskCount, currentActiveTaskCount, partitionCount);
+ }
+
+ private int computeDesiredTaskCountHelper(final List<Long> lags, final int currentActiveTaskCount)
+ {
int beyond = 0;
int within = 0;
- int metricsCount = lags.size();
- for (Long lag : lags) {
+ final int metricsCount = lags.size();
+ for (final Long lag : lags) {
if (lag >= lagBasedAutoScalerConfig.getScaleOutThreshold()) {
beyond++;
}
if (lag <= lagBasedAutoScalerConfig.getScaleInThreshold()) {
within++;
}
}
- double beyondProportion = beyond * 1.0 / metricsCount;
- double withinProportion = within * 1.0 / metricsCount;
+ final double beyondProportion = beyond * 1.0 / metricsCount;
+ final double withinProportion = within * 1.0 / metricsCount;
- log.debug("Calculated beyondProportion is [%s] and withinProportion is [%s] for dataSource [%s].", beyondProportion,
+ log.debug(
+ "Calculated beyondProportion is [%s] and withinProportion is [%s] for dataSource [%s].", beyondProportion,
withinProportion, dataSource
);
- int currentActiveTaskCount = supervisor.getActiveTaskGroupsCount();
- int desiredActiveTaskCount;
- int partitionCount = supervisor.getPartitionCount();
- if (partitionCount <= 0) {
- log.warn("Partition number for [%s] <= 0 ? how can it be?", dataSource);
+ if (beyondProportion >= lagBasedAutoScalerConfig.getTriggerScaleOutFractionThreshold()) {
+ return currentActiveTaskCount + lagBasedAutoScalerConfig.getScaleOutStep();
+ } else if (withinProportion >= lagBasedAutoScalerConfig.getTriggerScaleInFractionThreshold()) {
+ return currentActiveTaskCount - lagBasedAutoScalerConfig.getScaleInStep();
+ }
+
+ return currentActiveTaskCount;
+ }
+
+
+ private int applyMinMaxChecks(int desiredActiveTaskCount, final int currentActiveTaskCount, final int partitionCount)
+ {
+ // for now, only attempt to scale to nearest factor for scale up
Review Comment:
> This is because you risk scaling down more than you need to
If we scale to the next largest _above_ the desired, we won't scale down too far. And we'd protect some indexers from having more partitions assigned than others.
Edit: I guess "desired" here is a bit simple though, since it's using a fixed step. It would need a proportion, possibly based on RPS rather than only lag. Otherwise you could get trapped at max scale if the step is smaller than the gap to the next factor down.
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Allow task autoscaler to scale to nearest divisor of the partition count (druid)
Posted by "bsyk (via GitHub)" <gi...@apache.org>.
bsyk commented on code in PR #17988:
URL: https://github.com/apache/druid/pull/17988#discussion_r2176201417
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/LagBasedAutoScaler.java:
##########
@@ -200,84 +209,141 @@ private Runnable computeAndCollectLag()
* @param lags the lag metrics of Stream(Kafka/Kinesis)
* @return Integer. target number of tasksCount, -1 means skip scale action.
*/
- private int computeDesiredTaskCount(List<Long> lags)
+ @VisibleForTesting
+ int computeDesiredTaskCount(List<Long> lags)
{
- // if supervisor is not suspended, ensure required tasks are running
- // if suspended, ensure tasks have been requested to gracefully stop
log.debug("Computing desired task count for [%s], based on following lags : [%s]", dataSource, lags);
+ final int currentActiveTaskCount = supervisor.getActiveTaskGroupsCount();
+ final int partitionCount = supervisor.getPartitionCount();
+ if (partitionCount <= 0) {
+ log.warn("Partition number for [%s] <= 0 ? how can it be?", dataSource);
+ return -1;
+ }
+
+ // Cache the factorization in an immutable list for quick lookup later
+ // Partition counts *can* change externally without a new instance of this class being created
+ if (partitionFactors.isEmpty() || partitionCount != partitionFactors.get(partitionFactors.size() - 1)) {
+ log.debug("(Re)computing partitionCount factorization for partitionCount=[%d]", partitionCount);
+ partitionFactors = factorize(partitionCount);
+ }
+
+ Preconditions.checkState(!partitionFactors.isEmpty(), "partitionFactors should not be empty");
+
+ final int desiredActiveTaskCount = computeDesiredTaskCountHelper(lags, currentActiveTaskCount);
+ return applyMinMaxChecks(desiredActiveTaskCount, currentActiveTaskCount, partitionCount);
+ }
+
+ private int computeDesiredTaskCountHelper(final List<Long> lags, final int currentActiveTaskCount)
+ {
int beyond = 0;
int within = 0;
- int metricsCount = lags.size();
- for (Long lag : lags) {
+ final int metricsCount = lags.size();
+ for (final Long lag : lags) {
if (lag >= lagBasedAutoScalerConfig.getScaleOutThreshold()) {
beyond++;
}
if (lag <= lagBasedAutoScalerConfig.getScaleInThreshold()) {
within++;
}
}
- double beyondProportion = beyond * 1.0 / metricsCount;
- double withinProportion = within * 1.0 / metricsCount;
+ final double beyondProportion = beyond * 1.0 / metricsCount;
+ final double withinProportion = within * 1.0 / metricsCount;
- log.debug("Calculated beyondProportion is [%s] and withinProportion is [%s] for dataSource [%s].", beyondProportion,
+ log.debug(
+ "Calculated beyondProportion is [%s] and withinProportion is [%s] for dataSource [%s].", beyondProportion,
withinProportion, dataSource
);
- int currentActiveTaskCount = supervisor.getActiveTaskGroupsCount();
- int desiredActiveTaskCount;
- int partitionCount = supervisor.getPartitionCount();
- if (partitionCount <= 0) {
- log.warn("Partition number for [%s] <= 0 ? how can it be?", dataSource);
+ if (beyondProportion >= lagBasedAutoScalerConfig.getTriggerScaleOutFractionThreshold()) {
+ return currentActiveTaskCount + lagBasedAutoScalerConfig.getScaleOutStep();
+ } else if (withinProportion >= lagBasedAutoScalerConfig.getTriggerScaleInFractionThreshold()) {
+ return currentActiveTaskCount - lagBasedAutoScalerConfig.getScaleInStep();
+ }
+
+ return currentActiveTaskCount;
+ }
+
+
+ private int applyMinMaxChecks(int desiredActiveTaskCount, final int currentActiveTaskCount, final int partitionCount)
+ {
+ // for now, only attempt to scale to nearest factor for scale up
Review Comment:
> This is because you risk scaling down more than you need to
If we scale to the next largest _above_ the desired, we won't scale down too far. And we'd protect some indexers from having more partitions assigned than others.
Edit: I guess "desired" here is a bit simple though, since it's using a fixed step. It would need a proportion, possibly based on RPS rather than only lag.
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Allow task autoscaler to scale to nearest divisor of the partition count (druid)
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #17988:
URL: https://github.com/apache/druid/pull/17988#issuecomment-3621412573
This pull request/issue has been closed due to lack of activity. If you think that
is incorrect, or the pull request requires review, you can revive the PR at any time.
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org