Idle cost uses a U-shaped penalty with minimum at {@link #IDEAL_IDLE_RATIO}. + *
Idle cost uses a U-shaped penalty with minimum at {@link #OPTIMAL_TASK_IDLE_RATIO}. * This penalizes both under-provisioning (low idle, no safety margin, lag risk) and * over-provisioning (high idle, wasted capacity), with asymmetric severity controlled by * {@link #UNDER_PROVISIONING_PENALTY} and {@link #OVER_PROVISIONING_PENALTY}. @@ -40,7 +40,7 @@ public class WeightedCostFunction * Multiplier for a lag amplification factor; it was carefully chosen * during extensive testing as the most balanced multiplier for high-lag recovery. */ - static final double LAG_AMPLIFICATION_MULTIPLIER = 0.4; + static final double LAG_AMPLIFICATION_MULTIPLIER = 0.3; /** * Exponent (< 1) for sublinear busy redistribution in the idle projection: @@ -53,27 +53,28 @@ public class WeightedCostFunction /** * Minimum rate of processing for any task in records per second. This is used * as a placeholder if avg rate is not available to ensure that cost computations - * do not return infinitely large lag recovery times. + * do not return infinitely large lag recovery times, at the expense of underestimating the lag cost. */ static final double MIN_PROCESSING_RATE = 1_000; /** - * Target idle ratio representing the optimal operating point for the U-shaped idle cost. + * Default target idle ratio representing the optimal operating point for the U-shaped idle cost. * At this ratio the idle cost is at its minimum; both lower (risk) and higher (waste) are penalized. + * Configurable via {@link CostBasedAutoScalerConfig#getOptimalTaskIdleRatio()}. */ - static final double IDEAL_IDLE_RATIO = 0.25; + static final double OPTIMAL_TASK_IDLE_RATIO = 0.25; /** * Penalty magnitude applied when idle ratio is 0 (no safety margin). * Controls the steepness of the U-shape on the under-provisioning side. */ - static final double UNDER_PROVISIONING_PENALTY = 2.0; + static final double UNDER_PROVISIONING_PENALTY = 1.0; /** * Penalty magnitude applied when idle ratio is 1 (fully wasted capacity). * Controls the steepness of the U-shape on the over-provisioning side. */ - static final double OVER_PROVISIONING_PENALTY = 1.0; + static final double OVER_PROVISIONING_PENALTY = 2.0; /** * Computes cost for a given task count using compute time metrics. @@ -148,7 +149,7 @@ public CostResult computeCost( } final double virtualLagRecoveryTime = overrun * metrics.getTaskDurationSeconds(); - final double idleCost = uShapedIdleCost(predictedIdleRatio, proposedTaskCount); + final double idleCost = uShapedIdleCost(predictedIdleRatio, proposedTaskCount, config.getOptimalTaskIdleRatio()); final double lagCost = config.getLagWeight() * (lagRecoveryTime + virtualLagRecoveryTime); final double weightedIdleCost = config.getIdleWeight() * idleCost; final double cost = lagCost + weightedIdleCost; @@ -168,28 +169,28 @@ public CostResult computeCost( } /** - * U-shaped idle cost with minimum at {@link #IDEAL_IDLE_RATIO}. + * U-shaped idle cost with minimum at {@link #OPTIMAL_TASK_IDLE_RATIO}. * *
- * The ideal-idle baseline keeps cost non-zero at the optimum so the optimizer + * The optimal-task-idle baseline keeps cost non-zero at the optimum so the optimizer * always has a finite trade-off against lag cost. */ - double uShapedIdleCost(double predictedIdleRatio, int taskCount) + double uShapedIdleCost(double predictedIdleRatio, int taskCount, double optimalTaskIdleRatio) { final double penalty; - if (predictedIdleRatio < IDEAL_IDLE_RATIO) { - final double norm = (IDEAL_IDLE_RATIO - predictedIdleRatio) / IDEAL_IDLE_RATIO; + if (predictedIdleRatio < optimalTaskIdleRatio) { + final double norm = (optimalTaskIdleRatio - predictedIdleRatio) / optimalTaskIdleRatio; penalty = UNDER_PROVISIONING_PENALTY * norm * norm; } else { - final double norm = (predictedIdleRatio - IDEAL_IDLE_RATIO) / (1.0 - IDEAL_IDLE_RATIO); + final double norm = (predictedIdleRatio - optimalTaskIdleRatio) / (1.0 - optimalTaskIdleRatio); penalty = OVER_PROVISIONING_PENALTY * norm * norm; } - return taskCount * (IDEAL_IDLE_RATIO + penalty); + return taskCount * (optimalTaskIdleRatio + penalty); } } 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 6a8eec3e3d90..6d09221b410a 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 @@ -29,6 +29,7 @@ import static org.apache.druid.indexing.seekablestream.supervisor.autoscaler.CostBasedAutoScalerConfig.DEFAULT_LAG_WEIGHT; import static org.apache.druid.indexing.seekablestream.supervisor.autoscaler.CostBasedAutoScalerConfig.DEFAULT_MIN_SCALE_DELAY; import static org.apache.druid.indexing.seekablestream.supervisor.autoscaler.CostBasedAutoScalerConfig.DEFAULT_SCALE_ACTION_PERIOD_MILLIS; +import static org.apache.druid.indexing.seekablestream.supervisor.autoscaler.WeightedCostFunction.OPTIMAL_TASK_IDLE_RATIO; @SuppressWarnings("TextBlockMigration") public class CostBasedAutoScalerConfigTest @@ -49,13 +50,14 @@ public void testSerdeWithAllProperties() throws Exception + " \"scaleActionPeriodMillis\": 60000,\n" + " \"lagWeight\": 0.6,\n" + " \"idleWeight\": 0.4,\n" + + " \"optimalTaskIdleRatio\": 0.3,\n" + " \"minScaleUpDelay\": \"PT5M\",\n" + " \"minScaleDownDelay\": \"PT10M\",\n" + " \"scaleDownDuringTaskRolloverOnly\": true,\n" + " \"usePollIdleRatio\": false\n" + "}"; - CostBasedAutoScalerConfig config = mapper.readValue(json, CostBasedAutoScalerConfig.class); + final CostBasedAutoScalerConfig config = mapper.readValue(json, CostBasedAutoScalerConfig.class); Assert.assertTrue(config.getEnableTaskAutoScaler()); Assert.assertEquals(100, config.getTaskCountMax()); @@ -65,6 +67,7 @@ public void testSerdeWithAllProperties() throws Exception Assert.assertEquals(60000L, config.getScaleActionPeriodMillis()); Assert.assertEquals(0.6, config.getLagWeight(), 0.001); Assert.assertEquals(0.4, config.getIdleWeight(), 0.001); + Assert.assertEquals(0.3, config.getOptimalTaskIdleRatio(), 0.001); Assert.assertEquals(Duration.standardMinutes(5), config.getMinScaleUpDelay()); Assert.assertEquals(Duration.standardMinutes(10), config.getMinScaleDownDelay()); Assert.assertTrue(config.isScaleDownOnTaskRolloverOnly()); @@ -73,8 +76,8 @@ public void testSerdeWithAllProperties() throws Exception Assert.assertTrue(config.isUseTaskCountBoundariesOnScaleDown()); // Test serialization back to JSON - String serialized = mapper.writeValueAsString(config); - CostBasedAutoScalerConfig deserialized = mapper.readValue(serialized, CostBasedAutoScalerConfig.class); + final String serialized = mapper.writeValueAsString(config); + final CostBasedAutoScalerConfig deserialized = mapper.readValue(serialized, CostBasedAutoScalerConfig.class); Assert.assertEquals(config, deserialized); } @@ -89,7 +92,7 @@ public void testSerdeWithDefaults() throws Exception + " \"taskCountMin\": 2\n" + "}"; - CostBasedAutoScalerConfig config = mapper.readValue(json, CostBasedAutoScalerConfig.class); + final CostBasedAutoScalerConfig config = mapper.readValue(json, CostBasedAutoScalerConfig.class); Assert.assertTrue(config.getEnableTaskAutoScaler()); Assert.assertEquals(50, config.getTaskCountMax()); @@ -99,6 +102,7 @@ public void testSerdeWithDefaults() throws Exception Assert.assertEquals(DEFAULT_SCALE_ACTION_PERIOD_MILLIS, config.getScaleActionPeriodMillis()); Assert.assertEquals(DEFAULT_LAG_WEIGHT, config.getLagWeight(), 0.001); Assert.assertEquals(DEFAULT_IDLE_WEIGHT, config.getIdleWeight(), 0.001); + Assert.assertEquals(OPTIMAL_TASK_IDLE_RATIO, config.getOptimalTaskIdleRatio(), 0.001); // minScaleUpDelay and minScaleDownDelay each have their own independent default Assert.assertEquals(Duration.millis(DEFAULT_SCALE_ACTION_PERIOD_MILLIS), config.getMinScaleUpDelay()); Assert.assertEquals(DEFAULT_MIN_SCALE_DELAY, config.getMinScaleDownDelay()); @@ -176,25 +180,48 @@ public void testValidation_InvalidStopTaskCountRatio() .build(); } + @Test(expected = IllegalArgumentException.class) + public void testValidationZeroOptimalTaskIdleRatio() + { + CostBasedAutoScalerConfig.builder() + .taskCountMax(100) + .taskCountMin(5) + .optimalTaskIdleRatio(0.0) + .enableTaskAutoScaler(true) + .build(); + } + + @Test(expected = IllegalArgumentException.class) + public void testValidationOneOptimalTaskIdleRatio() + { + CostBasedAutoScalerConfig.builder() + .taskCountMax(100) + .taskCountMin(5) + .optimalTaskIdleRatio(1.0) + .enableTaskAutoScaler(true) + .build(); + } + @Test public void testBuilder() { - CostBasedAutoScalerConfig config = CostBasedAutoScalerConfig.builder() - .taskCountMax(100) - .taskCountMin(5) - .taskCountStart(10) - .enableTaskAutoScaler(true) - .stopTaskCountRatio(0.8) - .scaleActionPeriodMillis(60000L) - .lagWeight(0.6) - .idleWeight(0.4) - .useTaskCountBoundariesOnScaleUp(true) - .useTaskCountBoundariesOnScaleDown(true) - .minScaleUpDelay(Duration.standardMinutes(5)) - .minScaleDownDelay(Duration.standardMinutes(10)) - .scaleDownDuringTaskRolloverOnly(true) - .usePollIdleRatio(false) - .build(); + final CostBasedAutoScalerConfig config = CostBasedAutoScalerConfig.builder() + .taskCountMax(100) + .taskCountMin(5) + .taskCountStart(10) + .enableTaskAutoScaler(true) + .stopTaskCountRatio(0.8) + .scaleActionPeriodMillis(60000L) + .lagWeight(0.6) + .idleWeight(0.4) + .optimalTaskIdleRatio(0.3) + .useTaskCountBoundariesOnScaleUp(true) + .useTaskCountBoundariesOnScaleDown(true) + .minScaleUpDelay(Duration.standardMinutes(5)) + .minScaleDownDelay(Duration.standardMinutes(10)) + .scaleDownDuringTaskRolloverOnly(true) + .usePollIdleRatio(false) + .build(); Assert.assertTrue(config.getEnableTaskAutoScaler()); Assert.assertEquals(100, config.getTaskCountMax()); @@ -204,6 +231,7 @@ public void testBuilder() Assert.assertEquals(60000L, config.getScaleActionPeriodMillis()); Assert.assertEquals(0.6, config.getLagWeight(), 0.001); Assert.assertEquals(0.4, config.getIdleWeight(), 0.001); + Assert.assertEquals(0.3, config.getOptimalTaskIdleRatio(), 0.001); Assert.assertTrue(config.isUseTaskCountBoundariesOnScaleUp()); Assert.assertTrue(config.isUseTaskCountBoundariesOnScaleDown()); Assert.assertEquals(Duration.standardMinutes(5), config.getMinScaleUpDelay()); diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/WeightedCostFunctionTest.java b/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/WeightedCostFunctionTest.java index 2c9d462b6f24..6802692ade49 100644 --- a/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/WeightedCostFunctionTest.java +++ b/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/WeightedCostFunctionTest.java @@ -101,13 +101,28 @@ public void testLagCostWithAbsoluteModel() double amplification = 1.0 + WeightedCostFunction.LAG_AMPLIFICATION_MULTIPLIER * Math.log(aggregateLag / 100); double costCurrent = costFunction.computeCost(metrics, 10, lagOnlyConfig).totalCost(); - Assert.assertEquals("Cost of current tasks", aggregateLag * amplification / (10 * 1000.0), costCurrent, 0.1); + Assert.assertEquals( + "Cost of current tasks", + aggregateLag * amplification / (10 * WeightedCostFunction.MIN_PROCESSING_RATE), + costCurrent, + 0.1 + ); double costUp5 = costFunction.computeCost(metrics, 15, lagOnlyConfig).totalCost(); - Assert.assertEquals("Cost when scaling up by 5", aggregateLag * amplification / (15 * 1000.0), costUp5, 0.1); + Assert.assertEquals( + "Cost when scaling up by 5", + aggregateLag * amplification / (15 * WeightedCostFunction.MIN_PROCESSING_RATE), + costUp5, + 0.1 + ); double costUp10 = costFunction.computeCost(metrics, 20, lagOnlyConfig).totalCost(); - Assert.assertEquals("Cost when scaling up by 10", aggregateLag * amplification / (20 * 1000.0), costUp10, 0.1); + Assert.assertEquals( + "Cost when scaling up by 10", + aggregateLag * amplification / (20 * WeightedCostFunction.MIN_PROCESSING_RATE), + costUp10, + 0.1 + ); // Adding more tasks reduces lag recovery time Assert.assertTrue("Adding more tasks reduces lag cost", costUp10 < costUp5); @@ -195,10 +210,10 @@ public void testNegativeProcessingRate_throwsDefensiveException() } @Test - public void testIdleCostIsUShapedAroundIdealRatio() + public void testIdleCostIsUShapedAroundOptimalTaskIdleRatio() { - // U-shaped cost: minimum near IDEAL_IDLE_RATIO=0.25, higher on both sides. - // Current: 10 tasks with 25% idle (already at ideal). + // U-shaped cost: minimum near OPTIMAL_TASK_IDLE_RATIO=0.25, higher on both sides. + // Current: 10 tasks with 25% idle (already at optimum). CostBasedAutoScalerConfig idleOnlyConfig = CostBasedAutoScalerConfig.builder() .taskCountMax(100) .taskCountMin(1) @@ -240,7 +255,10 @@ public void testIdleRatioClampingAtBoundaries() double costAt2 = costFunction.computeCost(metrics, 2, idleOnlyConfig).totalCost(); // idle = 0: max under-provisioning penalty; cost = taskCount * (IDEAL + UNDER_PENALTY) - double expectedAt2 = 2 * (WeightedCostFunction.IDEAL_IDLE_RATIO + WeightedCostFunction.UNDER_PROVISIONING_PENALTY); + double expectedAt2 = 2 * ( + WeightedCostFunction.OPTIMAL_TASK_IDLE_RATIO + + WeightedCostFunction.UNDER_PROVISIONING_PENALTY + ); Assert.assertEquals("Idle cost at clamped-to-zero idle ratio should reflect full under-provisioning penalty", expectedAt2, costAt2, 0.0001); @@ -277,7 +295,7 @@ public void testModerateConsolidationProjectsHealthyIdle() // Far below the clamped-to-zero cost the linear model produced: idle is healthy, not clamped. double clampedToZeroCost = - 5 * (WeightedCostFunction.IDEAL_IDLE_RATIO + WeightedCostFunction.UNDER_PROVISIONING_PENALTY); + 5 * (WeightedCostFunction.OPTIMAL_TASK_IDLE_RATIO + WeightedCostFunction.UNDER_PROVISIONING_PENALTY); Assert.assertTrue( "Predicted idle should be healthy, not clamped to zero", costScaleDown < clampedToZeroCost @@ -303,7 +321,7 @@ public void testIdleRatioWithMissingData() // With missing data, predicted idle = 0.5 for all task counts regardless of proposed count. // U-shaped cost at idle=0.5: idle > IDEAL(0.25), norm=(0.5-0.25)/0.75=1/3, penalty=1*(1/3)^2=1/9 - double expectedCostPerTask = WeightedCostFunction.IDEAL_IDLE_RATIO + double expectedCostPerTask = WeightedCostFunction.OPTIMAL_TASK_IDLE_RATIO + WeightedCostFunction.OVER_PROVISIONING_PENALTY * (1.0 / 3.0) * (1.0 / 3.0); Assert.assertEquals("Cost at 10 tasks with missing idle data", 10 * expectedCostPerTask, cost10, 0.0001); Assert.assertEquals("Cost at 20 tasks with missing idle data", 20 * expectedCostPerTask, cost20, 0.0001); @@ -334,7 +352,7 @@ public void testLagAmplificationAppliedUnconditionally() double aggregateLag = 150.0 * partitionCount; double lagPerPartition = aggregateLag / partitionCount; double amplification = 1.0 + WeightedCostFunction.LAG_AMPLIFICATION_MULTIPLIER * Math.log(lagPerPartition); - double expected = aggregateLag * amplification / (proposedTaskCount * 1000.0); + double expected = aggregateLag * amplification / (proposedTaskCount * WeightedCostFunction.MIN_PROCESSING_RATE); Assert.assertEquals("Lag amplification should increase lag recovery time", expected, costWithAmp, 0.0001); } @@ -379,42 +397,57 @@ public void testUShapedIdleCostFormula() { int n = 10; - // At ideal ratio: penalty = 0, cost = n * IDEAL_IDLE_RATIO + // At optimal ratio: penalty = 0, cost = n * OPTIMAL_TASK_IDLE_RATIO Assert.assertEquals( - n * WeightedCostFunction.IDEAL_IDLE_RATIO, - costFunction.uShapedIdleCost(WeightedCostFunction.IDEAL_IDLE_RATIO, n), + n * WeightedCostFunction.OPTIMAL_TASK_IDLE_RATIO, + costFunction.uShapedIdleCost( + WeightedCostFunction.OPTIMAL_TASK_IDLE_RATIO, + n, + WeightedCostFunction.OPTIMAL_TASK_IDLE_RATIO + ), 1e-9 ); // At idle = 0 (fully under-provisioned): norm = 1, penalty = UNDER_PROVISIONING_PENALTY Assert.assertEquals( - n * (WeightedCostFunction.IDEAL_IDLE_RATIO + WeightedCostFunction.UNDER_PROVISIONING_PENALTY), - costFunction.uShapedIdleCost(0.0, n), + n * (WeightedCostFunction.OPTIMAL_TASK_IDLE_RATIO + WeightedCostFunction.UNDER_PROVISIONING_PENALTY), + costFunction.uShapedIdleCost(0.0, n, WeightedCostFunction.OPTIMAL_TASK_IDLE_RATIO), 1e-9 ); // At idle = 1 (fully over-provisioned): norm = 1, penalty = OVER_PROVISIONING_PENALTY Assert.assertEquals( - n * (WeightedCostFunction.IDEAL_IDLE_RATIO + WeightedCostFunction.OVER_PROVISIONING_PENALTY), - costFunction.uShapedIdleCost(1.0, n), + n * (WeightedCostFunction.OPTIMAL_TASK_IDLE_RATIO + WeightedCostFunction.OVER_PROVISIONING_PENALTY), + costFunction.uShapedIdleCost(1.0, n, WeightedCostFunction.OPTIMAL_TASK_IDLE_RATIO), 1e-9 ); - // Both extremes exceed the ideal cost - double idealCost = costFunction.uShapedIdleCost(WeightedCostFunction.IDEAL_IDLE_RATIO, n); - Assert.assertTrue("idle=0 costs more than ideal", costFunction.uShapedIdleCost(0.0, n) > idealCost); - Assert.assertTrue("idle=1 costs more than ideal", costFunction.uShapedIdleCost(1.0, n) > idealCost); + // Both extremes exceed the optimal cost + final double optimalCost = costFunction.uShapedIdleCost( + WeightedCostFunction.OPTIMAL_TASK_IDLE_RATIO, + n, + WeightedCostFunction.OPTIMAL_TASK_IDLE_RATIO + ); + Assert.assertTrue( + "idle=0 costs more than optimal", + costFunction.uShapedIdleCost(0.0, n, WeightedCostFunction.OPTIMAL_TASK_IDLE_RATIO) > optimalCost + ); + Assert.assertTrue( + "idle=1 costs more than optimal", + costFunction.uShapedIdleCost(1.0, n, WeightedCostFunction.OPTIMAL_TASK_IDLE_RATIO) > optimalCost + ); - // Under-provisioning is penalized more than over-provisioning (UNDER > OVER) + // Over-provisioning is penalized more than under-provisioning (OVER > UNDER) Assert.assertTrue( - "under-provisioning penalty exceeds over-provisioning penalty", - costFunction.uShapedIdleCost(0.0, n) > costFunction.uShapedIdleCost(1.0, n) + "over-provisioning penalty exceeds under-provisioning penalty", + costFunction.uShapedIdleCost(1.0, n, WeightedCostFunction.OPTIMAL_TASK_IDLE_RATIO) + > costFunction.uShapedIdleCost(0.0, n, WeightedCostFunction.OPTIMAL_TASK_IDLE_RATIO) ); // Cost scales linearly with task count at any fixed idle ratio Assert.assertEquals( - 2 * costFunction.uShapedIdleCost(0.5, n), - costFunction.uShapedIdleCost(0.5, 2 * n), + 2 * costFunction.uShapedIdleCost(0.5, n, WeightedCostFunction.OPTIMAL_TASK_IDLE_RATIO), + costFunction.uShapedIdleCost(0.5, 2 * n, WeightedCostFunction.OPTIMAL_TASK_IDLE_RATIO), 1e-9 ); } @@ -469,7 +502,7 @@ public void testComputeCostSwitchesBetweenPollIdleRatioAndUtilizationRatio() double costWithPollIdleRatio = costFunction.computeCost(metrics, 10, idleOnlyConfig).totalCost(); Assert.assertEquals( "Default config should cost using the raw pollIdleRatio", - costFunction.uShapedIdleCost(0.9, 10), + costFunction.uShapedIdleCost(0.9, 10, WeightedCostFunction.OPTIMAL_TASK_IDLE_RATIO), costWithPollIdleRatio, 0.0001 ); @@ -477,7 +510,7 @@ public void testComputeCostSwitchesBetweenPollIdleRatioAndUtilizationRatio() double costWithUtilizationRatio = costFunction.computeCost(metrics, 10, utilizationConfig).totalCost(); Assert.assertEquals( "usePollIdleRatio=false should cost using the rate-derived idle ratio instead of pollIdleRatio", - costFunction.uShapedIdleCost(0.9, 10), + costFunction.uShapedIdleCost(0.9, 10, WeightedCostFunction.OPTIMAL_TASK_IDLE_RATIO), costWithUtilizationRatio, 0.0001 ); @@ -487,13 +520,13 @@ public void testComputeCostSwitchesBetweenPollIdleRatioAndUtilizationRatio() double costStillPollIdle = costFunction.computeCost(divergingMetrics, 10, idleOnlyConfig).totalCost(); double costStillUtilization = costFunction.computeCost(divergingMetrics, 10, utilizationConfig).totalCost(); Assert.assertEquals( - costFunction.uShapedIdleCost(0.1, 10), + costFunction.uShapedIdleCost(0.1, 10, WeightedCostFunction.OPTIMAL_TASK_IDLE_RATIO), costStillPollIdle, 0.0001 ); Assert.assertEquals( "Utilization-derived idle ratio (0.9) should be used instead of the diverging pollIdleRatio (0.1)", - costFunction.uShapedIdleCost(0.9, 10), + costFunction.uShapedIdleCost(0.9, 10, WeightedCostFunction.OPTIMAL_TASK_IDLE_RATIO), costStillUtilization, 0.0001 );