This is an automated email from the ASF dual-hosted git repository.
kfaraz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/druid.git
The following commit(s) were added to refs/heads/master by this push:
new 9c26bb75430 minor: Add metrics to costBased auto-scaler (#19631)
9c26bb75430 is described below
commit 9c26bb75430dd2e853d666ba67707647e61f392e
Author: Kashif Faraz <[email protected]>
AuthorDate: Fri Jun 26 15:03:14 2026 +0530
minor: Add metrics to costBased auto-scaler (#19631)
Changes
---------
- Add metric `task/autoScaler/costBased/avgProcessingRate`
- Add metric `task/autoScaler/costBased/avgPollIdleRatio`
- Add metric `task/autoScaler/costBased/lagWeight`
- Add metric `task/autoScaler/costBased/costWeight`
- Remove metric `task/autoScaler/costBased/lagCost`
- Remove metric `task/autoScaler/costBased/idleCost`
- Minor cleanup
---
docs/ingestion/supervisor.md | 2 --
docs/operations/metrics.md | 6 ++--
.../supervisor/autoscaler/CostBasedAutoScaler.java | 16 +++++++----
.../autoscaler/CostBasedAutoScalerConfig.java | 32 +++++++---------------
.../autoscaler/CostBasedAutoScalerConfigTest.java | 12 ++++----
5 files changed, 30 insertions(+), 38 deletions(-)
diff --git a/docs/ingestion/supervisor.md b/docs/ingestion/supervisor.md
index 4579532e09d..b642aefec45 100644
--- a/docs/ingestion/supervisor.md
+++ b/docs/ingestion/supervisor.md
@@ -204,8 +204,6 @@ The sweet spot is roughly 25% idle, giving headroom to
absorb traffic spikes wit
At every evaluation interval, Druid computes the score for each candidate task
count and picks the one with the lowest total cost.
-Note: Kinesis is not supported yet, support is in progress.
-
The following table outlines the configuration properties related to the
`costBased` autoscaler strategy:
| Property | Description | Required | Default |
diff --git a/docs/operations/metrics.md b/docs/operations/metrics.md
index 31c6813be6c..4ce8e85061f 100644
--- a/docs/operations/metrics.md
+++ b/docs/operations/metrics.md
@@ -318,8 +318,10 @@ batch ingestion emit the following metrics. These metrics
are deltas for each em
|`task/autoScaler/requiredCount`|Count of required tasks based on the
calculations of the auto scaler.|`supervisorId`, `dataSource`, `stream`,
`scalingSkipReason`|Depends on auto scaler config.|
|`task/autoScaler/scaleActionTime`|Time taken in milliseconds to complete the
scale action.|`supervisorId`, `dataSource`, `stream`, `tags`|Depends on auto
scaler config.|
|`task/autoScaler/costBased/optimalTaskCount`|Optimal task count computed by
the cost-based auto scaler.|`supervisorId`, `dataSource`, `stream`|Depends on
auto scaler config.|
-|`task/autoScaler/costBased/lagCost`|Lag cost component of the cost-based auto
scaler's cost function.|`supervisorId`, `dataSource`, `stream`|Depends on auto
scaler config.|
-|`task/autoScaler/costBased/idleCost`|Idle cost component of the cost-based
auto scaler's cost function.|`supervisorId`, `dataSource`, `stream`|Depends on
auto scaler config.|
+|`task/autoScaler/costBased/lagWeight`|Lag weight used in the cost function
for the auto scaler.|`supervisorId`, `dataSource`, `stream`|Depends on auto
scaler config.|
+|`task/autoScaler/costBased/idleWeight`|Idle weight used in the cost function
for the auto scaler.|`supervisorId`, `dataSource`, `stream`|Depends on auto
scaler config.|
+|`task/autoScaler/costBased/avgProcessingRate`|Windowed average rate of
processing records across all tasks in the supervisor.|`supervisorId`,
`dataSource`, `stream`|Depends on rate of incoming data and processing capacity
of the tasks.|
+|`task/autoScaler/costBased/avgPollIdleRatio`|Poll-to-idle-ratio as reported
by the streaming consumer averaged across all tasks in the supervisor.
Currently supported only with Kafka supervisors.|`supervisorId`, `dataSource`,
`stream`|0 if the consumer is never idle, 1 if the consumer is always idle, -1
if ratio is not available.|
If the JVM does not support CPU time measurement for the current thread,
`ingest/merge/cpu` and `ingest/persists/cpu` will be 0.
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/CostBasedAutoScaler.java
b/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/CostBasedAutoScaler.java
index 7dc9dd0c2d6..32efeadd4c7 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/CostBasedAutoScaler.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/CostBasedAutoScaler.java
@@ -61,10 +61,12 @@ public class CostBasedAutoScaler implements
SupervisorTaskAutoScaler
{
private static final EmittingLogger log = new
EmittingLogger(CostBasedAutoScaler.class);
- public static final String LAG_COST_METRIC =
"task/autoScaler/costBased/lagCost";
- public static final String IDLE_COST_METRIC =
"task/autoScaler/costBased/idleCost";
+ public static final String LAG_WEIGHT_METRIC =
"task/autoScaler/costBased/lagWeight";
+ public static final String IDLE_WEIGHT_METRIC =
"task/autoScaler/costBased/idleWeight";
public static final String OPTIMAL_TASK_COUNT_METRIC =
"task/autoScaler/costBased/optimalTaskCount";
public static final String INVALID_METRICS_COUNT =
"task/autoScaler/costBased/invalidMetrics";
+ public static final String AVG_PROCESSING_RATE_METRIC =
"task/autoScaler/costBased/avgProcessingRate";
+ public static final String AVG_POLL_IDLE_RATIO =
"task/autoScaler/costBased/avgPollIdleRatio";
/**
* Maximum number of candidate task counts to evaluate above or below the
current task count
@@ -255,14 +257,14 @@ public class CostBasedAutoScaler implements
SupervisorTaskAutoScaler
int startIndex = 0;
int endIndex = validTaskCounts.length - 1;
- if (config.shouldUseTaskCountBoundariesOnScaleUp()) {
+ if (config.isUseTaskCountBoundariesOnScaleUp()) {
int currentTaskCountIndex = Arrays.binarySearch(validTaskCounts,
currentTaskCount);
endIndex = currentTaskCountIndex >= 0
? Math.min(currentTaskCountIndex +
BOUNDARY_LIMIT_IN_PARTITIONS_PER_TASK, endIndex)
: endIndex;
}
- if (config.shouldUseTaskCountBoundariesOnScaleDown()) {
+ if (config.isUseTaskCountBoundariesOnScaleDown()) {
int currentTaskCountIndex = Arrays.binarySearch(validTaskCounts,
currentTaskCount);
startIndex = currentTaskCountIndex >= 0
? Math.max(currentTaskCountIndex -
BOUNDARY_LIMIT_IN_PARTITIONS_PER_TASK, startIndex)
@@ -282,8 +284,10 @@ public class CostBasedAutoScaler implements
SupervisorTaskAutoScaler
}
emitter.emit(getMetricBuilder().setMetric(OPTIMAL_TASK_COUNT_METRIC,
(long) optimalTaskCount));
- emitter.emit(getMetricBuilder().setMetric(LAG_COST_METRIC,
optimalCost.lagCost()));
- emitter.emit(getMetricBuilder().setMetric(IDLE_COST_METRIC,
optimalCost.idleCost()));
+ emitter.emit(getMetricBuilder().setMetric(LAG_WEIGHT_METRIC,
optimalCost.lagCost()));
+ emitter.emit(getMetricBuilder().setMetric(IDLE_WEIGHT_METRIC,
optimalCost.idleCost()));
+ emitter.emit(getMetricBuilder().setMetric(AVG_PROCESSING_RATE_METRIC,
metrics.getAvgProcessingRate()));
+ emitter.emit(getMetricBuilder().setMetric(AVG_POLL_IDLE_RATIO,
metrics.getPollIdleRatio()));
if (optimalTaskCount != currentTaskCount) {
log.info(
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/CostBasedAutoScalerConfig.java
b/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/CostBasedAutoScalerConfig.java
index 68f9f5a5ad7..7987c935b8e 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/CostBasedAutoScalerConfig.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/CostBasedAutoScalerConfig.java
@@ -85,8 +85,6 @@ public class CostBasedAutoScalerConfig implements
AutoScalerConfig
@Nullable @JsonProperty("scaleActionPeriodMillis") Long
scaleActionPeriodMillis,
@Nullable @JsonProperty("lagWeight") Double lagWeight,
@Nullable @JsonProperty("idleWeight") Double idleWeight,
- @Nullable @JsonProperty("useTaskCountBoundaries") Boolean
useTaskCountBoundaries,
- @Nullable @JsonProperty("highLagThreshold") Integer highLagThreshold,
@Nullable @JsonProperty("useTaskCountBoundariesOnScaleUp") Boolean
useTaskCountBoundariesOnScaleUp,
@Nullable @JsonProperty("useTaskCountBoundariesOnScaleDown") Boolean
useTaskCountBoundariesOnScaleDown,
@Nullable @JsonProperty("minScaleUpDelay") Duration minScaleUpDelay,
@@ -118,14 +116,6 @@ public class CostBasedAutoScalerConfig implements
AutoScalerConfig
this.scaleDownDuringTaskRolloverOnly =
Configs.valueOrDefault(scaleDownDuringTaskRolloverOnly, false);
if (this.enableTaskAutoScaler) {
- if (useTaskCountBoundaries != null) {
- LOG.warn("useTaskCountBoundaries is removed, "
- + "use useTaskCountBoundariesOnScaleUp and
useTaskCountBoundariesOnScaleDown instead");
- }
- if (highLagThreshold != null) {
- LOG.warn("highLagThreshold is removed, the autoscaler behavior is good
enough just with cost function");
- }
-
Preconditions.checkNotNull(taskCountMax, "taskCountMax is required when
enableTaskAutoScaler is true");
Preconditions.checkNotNull(taskCountMin, "taskCountMin is required when
enableTaskAutoScaler is true");
Preconditions.checkArgument(taskCountMax >= taskCountMin, "taskCountMax
must be >= taskCountMin");
@@ -231,23 +221,23 @@ public class CostBasedAutoScalerConfig implements
AutoScalerConfig
}
/**
- * Enables or disables the task-count evaluation window when considering
scale-up candidates.
- * If enabled, the optimizer only evaluates a small number of candidate task
counts above the current count,
- * which prevents large scale-up jumps.
+ * If true, the auto-scaler evaluates a small number of candidate task counts
+ * (dictated by {@link
CostBasedAutoScaler#BOUNDARY_LIMIT_IN_PARTITIONS_PER_TASK})
+ * above the current count, which prevents large scale-up jumps.
*/
- @JsonProperty("useTaskCountBoundariesOnScaleUp")
- public boolean shouldUseTaskCountBoundariesOnScaleUp()
+ @JsonProperty
+ public boolean isUseTaskCountBoundariesOnScaleUp()
{
return useTaskCountBoundariesOnScaleUp;
}
/**
- * Enables or disables the task-count evaluation window when considering
scale-down candidates.
- * If enabled, the optimizer only evaluates a small number of candidate task
counts below the current count,
- * which prevents large scale-down drops.
+ * If true, the auto-scaler evaluates a small number of candidate task counts
+ * (dictated by {@link
CostBasedAutoScaler#BOUNDARY_LIMIT_IN_PARTITIONS_PER_TASK})
+ * below the current count, which prevents large scale-down drops.
*/
- @JsonProperty("useTaskCountBoundariesOnScaleDown")
- public boolean shouldUseTaskCountBoundariesOnScaleDown()
+ @JsonProperty
+ public boolean isUseTaskCountBoundariesOnScaleDown()
{
return useTaskCountBoundariesOnScaleDown;
}
@@ -479,8 +469,6 @@ public class CostBasedAutoScalerConfig implements
AutoScalerConfig
scaleActionPeriodMillis,
lagWeight,
idleWeight,
- null,
- null,
useTaskCountBoundariesOnScaleUp,
useTaskCountBoundariesOnScaleDown,
minScaleUpDelay,
diff --git
a/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/CostBasedAutoScalerConfigTest.java
b/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/CostBasedAutoScalerConfigTest.java
index 4755b3cf496..5a1c6ec98fc 100644
---
a/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/CostBasedAutoScalerConfigTest.java
+++
b/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/CostBasedAutoScalerConfigTest.java
@@ -67,8 +67,8 @@ public class CostBasedAutoScalerConfigTest
Assert.assertEquals(Duration.standardMinutes(5),
config.getMinScaleUpDelay());
Assert.assertEquals(Duration.standardMinutes(10),
config.getMinScaleDownDelay());
Assert.assertTrue(config.isScaleDownOnTaskRolloverOnly());
- Assert.assertFalse(config.shouldUseTaskCountBoundariesOnScaleUp());
- Assert.assertTrue(config.shouldUseTaskCountBoundariesOnScaleDown());
+ Assert.assertFalse(config.isUseTaskCountBoundariesOnScaleUp());
+ Assert.assertTrue(config.isUseTaskCountBoundariesOnScaleDown());
// Test serialization back to JSON
String serialized = mapper.writeValueAsString(config);
@@ -101,8 +101,8 @@ public class CostBasedAutoScalerConfigTest
Assert.assertEquals(Duration.millis(DEFAULT_SCALE_ACTION_PERIOD_MILLIS),
config.getMinScaleUpDelay());
Assert.assertEquals(DEFAULT_MIN_SCALE_DELAY,
config.getMinScaleDownDelay());
Assert.assertFalse(config.isScaleDownOnTaskRolloverOnly());
- Assert.assertFalse(config.shouldUseTaskCountBoundariesOnScaleUp());
- Assert.assertTrue(config.shouldUseTaskCountBoundariesOnScaleDown());
+ Assert.assertFalse(config.isUseTaskCountBoundariesOnScaleUp());
+ Assert.assertTrue(config.isUseTaskCountBoundariesOnScaleDown());
Assert.assertNull(config.getTaskCountStart());
Assert.assertNull(config.getStopTaskCountRatio());
}
@@ -200,8 +200,8 @@ public class CostBasedAutoScalerConfigTest
Assert.assertEquals(60000L, config.getScaleActionPeriodMillis());
Assert.assertEquals(0.6, config.getLagWeight(), 0.001);
Assert.assertEquals(0.4, config.getIdleWeight(), 0.001);
- Assert.assertTrue(config.shouldUseTaskCountBoundariesOnScaleUp());
- Assert.assertTrue(config.shouldUseTaskCountBoundariesOnScaleDown());
+ Assert.assertTrue(config.isUseTaskCountBoundariesOnScaleUp());
+ Assert.assertTrue(config.isUseTaskCountBoundariesOnScaleDown());
Assert.assertEquals(Duration.standardMinutes(5),
config.getMinScaleUpDelay());
Assert.assertEquals(Duration.standardMinutes(10),
config.getMinScaleDownDelay());
Assert.assertTrue(config.isScaleDownOnTaskRolloverOnly());
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
