diff --git a/docs/ingestion/supervisor.md b/docs/ingestion/supervisor.md index 2a4f2a260061..2e8352d2142a 100644 --- a/docs/ingestion/supervisor.md +++ b/docs/ingestion/supervisor.md @@ -208,13 +208,13 @@ The following table outlines the configuration properties related to the `costBa | Property | Description | Required | Default | |----------|-------------|----------|---------------------------| -|`scaleActionPeriodMillis`|How often, in milliseconds, Druid evaluates whether to scale.|No| `600000` (10 min) | +|`scaleActionPeriodMillis`|How often, in milliseconds, Druid evaluates whether to scale.|No| `120000` (2 min) | |`lagWeight`|How much weight to give the lag cost relative to the idle cost. Higher values make the autoscaler more aggressive about adding tasks to drain backlog.|No| `0.4` | |`idleWeight`|How much weight to give the idle cost relative to the lag cost. Higher values make the autoscaler more aggressive about removing over-provisioned tasks.|No| `0.6` | |`useTaskCountBoundariesOnScaleUp`|Limits scale-up to a small step relative to the current task count, preventing large jumps. Disable to allow the autoscaler to jump directly to any task count.|No| `false` | |`useTaskCountBoundariesOnScaleDown`|Limits scale-down to a small step relative to the current task count, preventing large drops. Disable to allow the autoscaler to drop directly to any task count.|No| `true` | -|`minScaleUpDelay`|Minimum cooldown after a scale-up before the next scale-up is allowed. Specified as an ISO-8601 duration.|No| `scaleActionPeriodMillis` | -|`minScaleDownDelay`|Minimum cooldown after a scale-down before the next scale-down is allowed. Specified as an ISO-8601 duration.|No| `PT30M` | +|`minScaleUpDelay`|Minimum cooldown after a scale-up before the next scale-up is allowed. Specified as an ISO-8601 duration.|No| `PT15M` | +|`minScaleDownDelay`|Minimum cooldown after a scale-down before the next scale-down is allowed. Specified as an ISO-8601 duration.|No| `PT20M` | |`scaleDownDuringTaskRolloverOnly`|If `true`, scale-down actions are deferred until the next task rollover. This avoids disrupting in-progress ingestion.|No| `false` | The following example shows a supervisor spec with `costBased` autoscaler: 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 abada2f9b746..d16d425d833e 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 @@ -214,6 +214,25 @@ public CostBasedAutoScalerConfig getConfig() return config; } + private boolean isCriticalLag(CostMetrics metrics) + { + final Long criticalLagThreshold = config.getCriticalLagThreshold(); + return metrics != null && criticalLagThreshold != null + && metrics.getAggregateLag() >= criticalLagThreshold * WeightedCostFunction.CRITICAL_LAG_TIER1_FRACTION; + } + + /** + * Whether the last collected metrics crossed {@link WeightedCostFunction#CRITICAL_LAG_TIER2_FRACTION} of + * {@link CostBasedAutoScalerConfig#getCriticalLagThreshold()}, meaning the argmin search should be + * skipped entirely in favor of jumping straight to the maximum task count. + */ + private boolean isEmergencyLag(CostMetrics metrics) + { + final Long criticalLagThreshold = config.getCriticalLagThreshold(); + return metrics != null && criticalLagThreshold != null + && metrics.getAggregateLag() >= criticalLagThreshold * WeightedCostFunction.CRITICAL_LAG_TIER2_FRACTION; + } + /** * Returns the lowest-cost task count given {@code metrics}, or {@link #CANNOT_COMPUTE} when * metrics are unusable. Returning the current task count means the current count is already @@ -250,6 +269,24 @@ int computeOptimalTaskCount(CostMetrics metrics) return currentTaskCount; } + final boolean criticalLag = isCriticalLag(metrics); + final boolean emergencyLag = isEmergencyLag(metrics); + if (emergencyLag) { + log.info( + "Supervisor[%s] aggregateLag[%.0f] crossed [%.0f%%] of criticalLagThreshold[%d]: skipping the argmin" + + " search and jumping straight to the maximum task count.", + supervisorId, metrics.getAggregateLag(), WeightedCostFunction.CRITICAL_LAG_TIER2_FRACTION * 100, + config.getCriticalLagThreshold() + ); + } else if (criticalLag) { + log.info( + "Supervisor[%s] aggregateLag[%.0f] crossed [%.0f%%] of criticalLagThreshold[%d]: widening scale-up" + + " candidates and maxing out the lag-amplification multiplier.", + supervisorId, metrics.getAggregateLag(), WeightedCostFunction.CRITICAL_LAG_TIER1_FRACTION * 100, + config.getCriticalLagThreshold() + ); + } + // Start with the current task count as optimal final CostResult currentCost = costFunction.computeCost(metrics, currentTaskCount, config); int optimalTaskCount = currentTaskCount; @@ -278,7 +315,7 @@ int computeOptimalTaskCount(CostMetrics metrics) int startIndex = 0; int endIndex = validTaskCounts.length - 1; - if (config.isUseTaskCountBoundariesOnScaleUp()) { + if (config.isUseTaskCountBoundariesOnScaleUp() && !criticalLag) { int currentTaskCountIndex = Arrays.binarySearch(validTaskCounts, currentTaskCount); endIndex = currentTaskCountIndex >= 0 ? Math.min(currentTaskCountIndex + BOUNDARY_LIMIT_IN_PARTITIONS_PER_TASK, endIndex) @@ -292,6 +329,12 @@ int computeOptimalTaskCount(CostMetrics metrics) : startIndex; } + // Emergency (tier 2) lag skips the argmin search entirely: evaluate only the maximum valid task count. + if (emergencyLag) { + startIndex = validTaskCounts.length - 1; + endIndex = validTaskCounts.length - 1; + } + for (int i = startIndex; i <= endIndex; ++i) { final int taskCount = validTaskCounts[i]; CostResult costResult = costFunction.computeCost(metrics, taskCount, config); @@ -477,11 +520,14 @@ CostMetrics collectMetrics() final LagStats lagStats = supervisor.computeLagStats(); final double avgPartitionLag; + final double aggregateLag; if (lagStats == null) { log.debug("Lag stats unavailable for supervisorId [%s], skipping collection", supervisorId); avgPartitionLag = -1; + aggregateLag = -1; } else { avgPartitionLag = lagStats.getAvgLag(); + aggregateLag = lagStats.getTotalLag(); } final int currentTaskCount = supervisor.getIoConfig().getTaskCount(); @@ -497,6 +543,7 @@ CostMetrics collectMetrics() return new CostMetrics( avgPartitionLag, + aggregateLag, currentTaskCount, partitionCount, pollIdleRatio, 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 b93c354b6a5c..47aadf644328 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 @@ -46,8 +46,9 @@ public class CostBasedAutoScalerConfig implements AutoScalerConfig { static final double DEFAULT_LAG_WEIGHT = 0.4; static final double DEFAULT_IDLE_WEIGHT = 0.6; - static final Duration DEFAULT_MIN_SCALE_UP_DELAY = Duration.standardMinutes(10); - static final Duration DEFAULT_MIN_SCALE_DOWN_DELAY = Duration.standardMinutes(30); + static final Duration DEFAULT_MIN_SCALE_UP_DELAY = Duration.standardMinutes(15); + static final Duration DEFAULT_MIN_SCALE_DOWN_DELAY = Duration.standardMinutes(20); + static final Duration DEFAULT_SCALE_ACTION_PERIOD = Duration.standardMinutes(2); private final boolean enableTaskAutoScaler; private final int taskCountMax; @@ -65,6 +66,7 @@ public class CostBasedAutoScalerConfig implements AutoScalerConfig private final Duration minScaleDownDelay; private final boolean scaleDownDuringTaskRolloverOnly; private final boolean usePollIdleRatio; + private final Long criticalLagThreshold; private final int minCostDropPercentForScaling; /** @@ -87,11 +89,12 @@ public CostBasedAutoScalerConfig( @Nullable @JsonProperty("minScaleDownDelay") Duration minScaleDownDelay, @Nullable @JsonProperty("scaleDownDuringTaskRolloverOnly") Boolean scaleDownDuringTaskRolloverOnly, @Nullable @JsonProperty("usePollIdleRatio") Boolean usePollIdleRatio, + @Nullable @JsonProperty("criticalLagThreshold") Long criticalLagThreshold, @Nullable @JsonProperty("minCostDropPercentForScaling") Integer minCostDropPercentForScaling ) { this.enableTaskAutoScaler = Configs.valueOrDefault(enableTaskAutoScaler, false); - this.scaleActionPeriodMillis = Configs.valueOrDefault(scaleActionPeriodMillis, DEFAULT_MIN_SCALE_UP_DELAY.getMillis()); + this.scaleActionPeriodMillis = Configs.valueOrDefault(scaleActionPeriodMillis, DEFAULT_SCALE_ACTION_PERIOD.getMillis()); this.lagWeight = Configs.valueOrDefault(lagWeight, DEFAULT_LAG_WEIGHT); this.idleWeight = Configs.valueOrDefault(idleWeight, DEFAULT_IDLE_WEIGHT); @@ -105,6 +108,12 @@ public CostBasedAutoScalerConfig( this.minScaleDownDelay = Configs.valueOrDefault(minScaleDownDelay, DEFAULT_MIN_SCALE_DOWN_DELAY); this.scaleDownDuringTaskRolloverOnly = Configs.valueOrDefault(scaleDownDuringTaskRolloverOnly, false); this.usePollIdleRatio = Configs.valueOrDefault(usePollIdleRatio, true); + this.criticalLagThreshold = criticalLagThreshold; + + Preconditions.checkArgument( + criticalLagThreshold == null || criticalLagThreshold > 0, + "criticalLagThreshold must be > 0" + ); this.minCostDropPercentForScaling = Configs.valueOrDefault(minCostDropPercentForScaling, 0); if (this.enableTaskAutoScaler) { @@ -289,7 +298,25 @@ public boolean isUsePollIdleRatio() } /** - * Minimum percentage drop from current cost that is required by the auto-scaler + * Aggregate (sum-across-partitions) lag threshold driving a two-tier SLA-critical fast path, + * relative to {@link CostMetrics#getAggregateLag()}: + * + * {@code null} disables the feature. + */ + @JsonProperty + @Nullable + public Long getCriticalLagThreshold() + { + return criticalLagThreshold; + } + + /* Minimum percentage drop from current cost that is required by the auto-scaler * to choose a new task count. */ @JsonProperty @@ -331,7 +358,8 @@ public boolean equals(Object o) && usePollIdleRatio == that.usePollIdleRatio && minCostDropPercentForScaling == that.minCostDropPercentForScaling && Objects.equals(taskCountStart, that.taskCountStart) - && Objects.equals(stopTaskCountRatio, that.stopTaskCountRatio); + && Objects.equals(stopTaskCountRatio, that.stopTaskCountRatio) + && Objects.equals(criticalLagThreshold, that.criticalLagThreshold); } @Override @@ -353,6 +381,7 @@ public int hashCode() minScaleDownDelay, scaleDownDuringTaskRolloverOnly, usePollIdleRatio, + criticalLagThreshold, minCostDropPercentForScaling ); } @@ -376,6 +405,7 @@ public String toString() ", minScaleDownDelay=" + minScaleDownDelay + ", scaleDownDuringTaskRolloverOnly=" + scaleDownDuringTaskRolloverOnly + ", usePollIdleRatio=" + usePollIdleRatio + + ", criticalLagThreshold=" + criticalLagThreshold + ", minCostDropPercentForScaling=" + minCostDropPercentForScaling + '}'; } @@ -401,6 +431,7 @@ public static class Builder private Duration minScaleDownDelay; private Boolean scaleDownDuringTaskRolloverOnly; private Boolean usePollIdleRatio; + private Long criticalLagThreshold; private Integer minCostDropPercentForScaling; private Builder() @@ -485,6 +516,12 @@ public Builder usePollIdleRatio(boolean usePollIdleRatio) return this; } + public Builder criticalLagThreshold(Long criticalLagThreshold) + { + this.criticalLagThreshold = criticalLagThreshold; + return this; + } + public Builder useTaskCountBoundariesOnScaleUp(boolean useTaskCountBoundariesOnScaleUp) { this.useTaskCountBoundariesOnScaleUp = useTaskCountBoundariesOnScaleUp; @@ -521,6 +558,7 @@ public CostBasedAutoScalerConfig build() minScaleDownDelay, scaleDownDuringTaskRolloverOnly, usePollIdleRatio, + criticalLagThreshold, minCostDropPercentForScaling ); } diff --git a/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/CostMetrics.java b/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/CostMetrics.java index 33bf7eff33d5..f3e898497664 100644 --- a/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/CostMetrics.java +++ b/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/CostMetrics.java @@ -41,6 +41,7 @@ public class CostMetrics public CostMetrics( double avgPartitionLag, + double aggregateLag, int currentTaskCount, int partitionCount, double pollIdleRatio, @@ -55,7 +56,7 @@ public CostMetrics( this.pollIdleRatio = pollIdleRatio; this.taskDurationSeconds = taskDurationSeconds; this.avgProcessingRate = avgProcessingRate; - this.aggregateLag = avgPartitionLag * partitionCount; + this.aggregateLag = aggregateLag; this.maxObservedRate = maxObservedRate; } @@ -89,7 +90,6 @@ public double getPollIdleRatio() /** * Returns the aggregated lag across all partitions. - * Pre-computed as avgPartitionLag * partitionCount. */ public double getAggregateLag() { @@ -142,6 +142,7 @@ public boolean equals(Object o) } CostMetrics that = (CostMetrics) o; return Double.compare(that.avgPartitionLag, avgPartitionLag) == 0 + && Double.compare(that.aggregateLag, aggregateLag) == 0 && currentTaskCount == that.currentTaskCount && partitionCount == that.partitionCount && Double.compare(that.pollIdleRatio, pollIdleRatio) == 0 @@ -155,6 +156,7 @@ public int hashCode() { return Objects.hash( avgPartitionLag, + aggregateLag, currentTaskCount, partitionCount, pollIdleRatio, @@ -169,6 +171,7 @@ public String toString() { return "CostMetrics{" + "avgPartitionLag=" + avgPartitionLag + + ", aggregateLag=" + aggregateLag + ", currentTaskCount=" + currentTaskCount + ", partitionCount=" + partitionCount + ", pollIdleRatio=" + pollIdleRatio + diff --git a/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/WeightedCostFunction.java b/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/WeightedCostFunction.java index beaf0a5b9d55..99484d971581 100644 --- a/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/WeightedCostFunction.java +++ b/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/WeightedCostFunction.java @@ -37,10 +37,30 @@ public class WeightedCostFunction private static final Logger log = new Logger(WeightedCostFunction.class); /** - * Multiplier for a lag amplification factor; it was carefully chosen - * during extensive testing as the most balanced multiplier for high-lag recovery. + * Normal-path lag amplification multiplier. Critical-lag tiers provide the + * urgency amplification, so normal lag uses unamplified recovery time. */ - static final double LAG_AMPLIFICATION_MULTIPLIER = 0.3; + static final double LAG_AMPLIFICATION_MULTIPLIER = 0.0; + + /** + * Amplification multiplier used once aggregate lag crosses {@link #CRITICAL_LAG_TIER1_FRACTION} of + * {@link CostBasedAutoScalerConfig#getCriticalLagThreshold()} (tier 1 of the critical-lag fast path). + */ + static final double CRITICAL_LAG_AMPLIFICATION_MULTIPLIER = 6.0; + + /** + * Fraction of {@link CostBasedAutoScalerConfig#getCriticalLagThreshold()} at which tier 1 of the + * critical-lag fast path engages: amplification maxes out at {@link #CRITICAL_LAG_AMPLIFICATION_MULTIPLIER} + * and the scale-up candidate boundary is bypassed. + */ + static final double CRITICAL_LAG_TIER1_FRACTION = 0.75; + + /** + * Fraction of {@link CostBasedAutoScalerConfig#getCriticalLagThreshold()} at which tier 2 of the + * critical-lag fast path engages: the cost-minimization search is skipped entirely and the task + * count jumps straight to the maximum. + */ + static final double CRITICAL_LAG_TIER2_FRACTION = 0.95; /** * Exponent (< 1) for sublinear busy redistribution in the idle projection: @@ -106,14 +126,20 @@ public CostResult computeCost( } // Lag recovery time is decreasing by adding tasks and increasing by ejecting tasks. - // In case of increasing lag, we apply an amplification factor to reflect the urgency of addressing lag. - // Caution: we rely only on the metrics, the real issues may be absolutely different, up to hardware failure. + // Critical lag uses extra amplification; normal lag uses raw recovery time so capacity cost remains meaningful. + // Once aggregate lag crosses CRITICAL_LAG_TIER1_FRACTION of criticalLagThreshold, the multiplier is + // maxed out at CRITICAL_LAG_AMPLIFICATION_MULTIPLIER (vs unamplified normal recovery). final double lagRecoveryTime; if (metrics.getAggregateLag() <= 0) { lagRecoveryTime = 0; } else { final double lagPerPartition = metrics.getAggregateLag() / metrics.getPartitionCount(); - final double amplification = Math.max(1.0, 1.0 + LAG_AMPLIFICATION_MULTIPLIER * Math.log(lagPerPartition)); + final Long criticalLagThreshold = config.getCriticalLagThreshold(); + final boolean criticalLag = criticalLagThreshold != null + && metrics.getAggregateLag() >= criticalLagThreshold * CRITICAL_LAG_TIER1_FRACTION; + + final double amplificationMultiplier = criticalLag ? CRITICAL_LAG_AMPLIFICATION_MULTIPLIER : LAG_AMPLIFICATION_MULTIPLIER; + final double amplification = Math.max(1.0, 1.0 + amplificationMultiplier * Math.log(lagPerPartition)); final double adjustedProcessingRate = Math.max(avgProcessingRate, MIN_PROCESSING_RATE); lagRecoveryTime = metrics.getAggregateLag() * amplification / (proposedTaskCount * adjustedProcessingRate); } 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 295dd1efe882..472a55320920 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_DOWN_DELAY; import static org.apache.druid.indexing.seekablestream.supervisor.autoscaler.CostBasedAutoScalerConfig.DEFAULT_MIN_SCALE_UP_DELAY; +import static org.apache.druid.indexing.seekablestream.supervisor.autoscaler.CostBasedAutoScalerConfig.DEFAULT_SCALE_ACTION_PERIOD; import static org.apache.druid.indexing.seekablestream.supervisor.autoscaler.WeightedCostFunction.OPTIMAL_TASK_IDLE_RATIO; @SuppressWarnings("TextBlockMigration") @@ -55,6 +56,7 @@ public void testSerdeWithAllProperties() throws Exception + " \"minScaleDownDelay\": \"PT10M\",\n" + " \"scaleDownDuringTaskRolloverOnly\": true,\n" + " \"usePollIdleRatio\": false,\n" + + " \"criticalLagThreshold\": 500000,\n" + " \"minCostDropPercentForScaling\": 10\n" + "}"; @@ -75,6 +77,7 @@ public void testSerdeWithAllProperties() throws Exception Assert.assertFalse(config.isUsePollIdleRatio()); Assert.assertFalse(config.isUseTaskCountBoundariesOnScaleUp()); Assert.assertTrue(config.isUseTaskCountBoundariesOnScaleDown()); + Assert.assertEquals(Long.valueOf(500000), config.getCriticalLagThreshold()); Assert.assertEquals(10, config.getMinCostDropPercentForScaling()); // Test serialization back to JSON @@ -101,7 +104,7 @@ public void testSerdeWithDefaults() throws Exception Assert.assertEquals(2, config.getTaskCountMin()); // Check defaults - Assert.assertEquals(DEFAULT_MIN_SCALE_UP_DELAY.getMillis(), config.getScaleActionPeriodMillis()); + Assert.assertEquals(DEFAULT_SCALE_ACTION_PERIOD.getMillis(), 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); @@ -114,6 +117,7 @@ public void testSerdeWithDefaults() throws Exception Assert.assertTrue(config.isUseTaskCountBoundariesOnScaleDown()); Assert.assertNull(config.getTaskCountStart()); Assert.assertNull(config.getStopTaskCountRatio()); + Assert.assertNull(config.getCriticalLagThreshold()); Assert.assertEquals(0, config.getMinCostDropPercentForScaling()); } @@ -224,6 +228,7 @@ public void testBuilder() .minScaleDownDelay(Duration.standardMinutes(10)) .scaleDownDuringTaskRolloverOnly(true) .usePollIdleRatio(false) + .criticalLagThreshold(500000L) .build(); Assert.assertTrue(config.getEnableTaskAutoScaler()); @@ -241,6 +246,18 @@ public void testBuilder() Assert.assertEquals(Duration.standardMinutes(10), config.getMinScaleDownDelay()); Assert.assertTrue(config.isScaleDownOnTaskRolloverOnly()); Assert.assertFalse(config.isUsePollIdleRatio()); + Assert.assertEquals(Long.valueOf(500000), config.getCriticalLagThreshold()); + } + + @Test(expected = IllegalArgumentException.class) + public void testValidation_ZeroCriticalLagThreshold() + { + CostBasedAutoScalerConfig.builder() + .taskCountMax(100) + .taskCountMin(5) + .criticalLagThreshold(0L) + .enableTaskAutoScaler(true) + .build(); } @Test diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/CostBasedAutoScalerMockTest.java b/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/CostBasedAutoScalerMockTest.java index fa86fbf1d3cb..3430842517dd 100644 --- a/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/CostBasedAutoScalerMockTest.java +++ b/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/CostBasedAutoScalerMockTest.java @@ -367,6 +367,7 @@ private void setupMocksForMetricsCollection( { CostMetrics metrics = new CostMetrics( avgLag, + avgLag * PARTITION_COUNT, taskCount, PARTITION_COUNT, pollIdleRatio, diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/CostBasedAutoScalerTest.java b/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/CostBasedAutoScalerTest.java index 435427fa37b2..993789dbe94d 100644 --- a/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/CostBasedAutoScalerTest.java +++ b/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/CostBasedAutoScalerTest.java @@ -240,6 +240,84 @@ public void testComputeOptimalTaskCountLimitsTaskCountJumps() ); } + @Test + public void testCriticalLagThresholdBypassesScaleUpBoundary() + { + // aggregateLag = 100_000 * 100 = 10,000,000. With threshold=12,000,000: tier1=9,000,000 (crossed), + // tier2=11,400,000 (not crossed), so this exercises tier1 (boundary bypass) without triggering + // tier2's emergency jump-to-max. + final CostBasedAutoScalerConfig boundedScaleUpConfig = CostBasedAutoScalerConfig + .builder() + .taskCountMax(100) + .taskCountMin(1) + .enableTaskAutoScaler(true) + .lagWeight(1.0) + .idleWeight(0.0) + .useTaskCountBoundariesOnScaleUp(true) + .criticalLagThreshold(12_000_000L) + .build(); + final CostBasedAutoScaler scaler = createAutoScaler(boundedScaleUpConfig); + + Assert.assertEquals( + "Critical lag should bypass the scale-up boundary and jump straight to the argmin", + 100, + scaler.computeOptimalTaskCount(createMetrics(100_000.0, 10, 100, 0.25)) + ); + + // Below the threshold, the boundary still applies as usual. + Assert.assertEquals( + "Below criticalLagThreshold, the scale-up boundary still limits candidates", + 13, + scaler.computeOptimalTaskCount(createMetrics(10.0, 10, 100, 0.25)) + ); + } + + @Test + public void testCriticalLagThresholdUsesExactAggregateLag() + { + final CostBasedAutoScalerConfig boundedScaleUpConfig = CostBasedAutoScalerConfig + .builder() + .taskCountMax(1_000) + .taskCountMin(1) + .enableTaskAutoScaler(true) + .lagWeight(1.0) + .idleWeight(0.0) + .useTaskCountBoundariesOnScaleUp(true) + .criticalLagThreshold(1_000L) + .build(); + final CostBasedAutoScaler scaler = createAutoScaler(boundedScaleUpConfig); + + Assert.assertEquals( + "Exact aggregate lag should engage tier 1 even when integer average lag is zero", + 1_000, + scaler.computeOptimalTaskCount(createMetrics(0.0, 999.0, 10, 1_000, 0.25)) + ); + } + + @Test + public void testEmergencyLagJumpsStraightToMaxTaskCount() + { + // aggregateLag = 100_000 * 500 = 50,000,000. With threshold=10,000,000: tier2=9,500,000 is + // comfortably crossed, so the argmin search is skipped entirely in favor of the maximum task count. + final CostBasedAutoScalerConfig config = CostBasedAutoScalerConfig + .builder() + .taskCountMax(500) + .taskCountMin(1) + .enableTaskAutoScaler(true) + .lagWeight(0.1) + .idleWeight(0.9) + .criticalLagThreshold(10_000_000L) + .build(); + final CostBasedAutoScaler scaler = createAutoScaler(config); + + // Idle-heavy weights would normally argue for scaling down, but emergency lag overrides that entirely. + Assert.assertEquals( + "Emergency lag should jump straight to the maximum task count regardless of idle-favoring weights", + 500, + scaler.computeOptimalTaskCount(createMetrics(100_000.0, 10, 500, 0.9)) + ); + } + @Test public void testExtractPollIdleRatio() { @@ -553,6 +631,35 @@ public void testScalingActionSkippedWhenMovingAverageRateUnavailable() ); } + @Test + public void testCollectMetricsPreservesExactAggregateLag() + { + final SupervisorSpec spec = Mockito.mock(SupervisorSpec.class); + final SeekableStreamSupervisor supervisor = Mockito.mock(SeekableStreamSupervisor.class); + final ServiceEmitter emitter = Mockito.mock(ServiceEmitter.class); + final SeekableStreamSupervisorIOConfig ioConfig = Mockito.mock(SeekableStreamSupervisorIOConfig.class); + + when(spec.getId()).thenReturn("test-supervisor"); + when(spec.getDataSources()).thenReturn(List.of("test-datasource")); + when(spec.isSuspended()).thenReturn(false); + when(supervisor.getIoConfig()).thenReturn(ioConfig); + when(ioConfig.getStream()).thenReturn("test-stream"); + when(ioConfig.getTaskDuration()).thenReturn(Duration.standardHours(1)); + when(ioConfig.getTaskCount()).thenReturn(10); + when(supervisor.getPartitionCount()).thenReturn(1_000); + when(supervisor.computeLagStats()).thenReturn(new LagStats(999, 999, 0)); + when(supervisor.getStats()).thenReturn(Collections.emptyMap()); + + final CostBasedAutoScalerConfig config = CostBasedAutoScalerConfig.builder() + .taskCountMax(1_000) + .taskCountMin(1) + .enableTaskAutoScaler(true) + .build(); + final CostBasedAutoScaler scaler = new CostBasedAutoScaler(supervisor, config, spec, emitter); + + Assert.assertEquals(999.0, scaler.collectMetrics().getAggregateLag(), 0.0); + } + @Test public void testCollectMetricsTracksMaxProcessingRateOnlyWhenPollIdleRatioDisabled() { @@ -635,6 +742,27 @@ private CostMetrics createMetrics( { return new CostMetrics( avgPartitionLag, + avgPartitionLag * partitionCount, + currentTaskCount, + partitionCount, + pollIdleRatio, + 3600, + 1000.0, + 0. + ); + } + + private CostMetrics createMetrics( + double avgPartitionLag, + double aggregateLag, + int currentTaskCount, + int partitionCount, + double pollIdleRatio + ) + { + return new CostMetrics( + avgPartitionLag, + aggregateLag, currentTaskCount, partitionCount, pollIdleRatio, 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 6802692ade49..c3aaf8e01ae1 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 @@ -329,7 +329,7 @@ public void testIdleRatioWithMissingData() } @Test - public void testLagAmplificationAppliedUnconditionally() + public void testNormalLagCostUsesUnamplifiedRecoveryTime() { CostBasedAutoScalerConfig lagOnly = CostBasedAutoScalerConfig.builder() .taskCountMax(100) @@ -344,23 +344,86 @@ public void testLagAmplificationAppliedUnconditionally() int partitionCount = 10; double pollIdleRatio = 0.1; - // lagPerPartition = 150 * 10 / 10 = 150, amplification = 1 + 0.2 * ln(150) + // Normal lag uses raw recovery time; critical lag is tested separately below. CostMetrics metrics = createMetrics(150.0, currentTaskCount, partitionCount, pollIdleRatio); double costWithAmp = costFunction.computeCost(metrics, proposedTaskCount, lagOnly).totalCost(); double aggregateLag = 150.0 * partitionCount; + double expected = aggregateLag / (proposedTaskCount * WeightedCostFunction.MIN_PROCESSING_RATE); + + Assert.assertEquals("Normal lag cost should use raw recovery time", expected, costWithAmp, 0.0001); + } + + @Test + public void testCriticalLagThresholdMaxesOutAmplificationMultiplier() + { + int currentTaskCount = 10; + int proposedTaskCount = 10; + int partitionCount = 10; + double avgPartitionLag = 150.0; + double aggregateLag = avgPartitionLag * partitionCount; + + CostMetrics metrics = createMetrics(avgPartitionLag, currentTaskCount, partitionCount, 0.1); + + // aggregateLag sits at exactly tier1Fraction (75%) of this threshold. + long tier1Threshold = (long) (aggregateLag / WeightedCostFunction.CRITICAL_LAG_TIER1_FRACTION); + + CostBasedAutoScalerConfig noThreshold = CostBasedAutoScalerConfig.builder() + .taskCountMax(100) + .taskCountMin(1) + .enableTaskAutoScaler(true) + .lagWeight(1.0) + .idleWeight(0.0) + .build(); + CostBasedAutoScalerConfig belowTier1 = CostBasedAutoScalerConfig.builder() + .taskCountMax(100) + .taskCountMin(1) + .enableTaskAutoScaler(true) + .lagWeight(1.0) + .idleWeight(0.0) + .criticalLagThreshold(tier1Threshold + 100) + .build(); + CostBasedAutoScalerConfig atTier1 = CostBasedAutoScalerConfig.builder() + .taskCountMax(100) + .taskCountMin(1) + .enableTaskAutoScaler(true) + .lagWeight(1.0) + .idleWeight(0.0) + .criticalLagThreshold(tier1Threshold) + .build(); + + double costBelowTier1 = costFunction.computeCost(metrics, proposedTaskCount, belowTier1).totalCost(); + Assert.assertEquals( + "Below tier1, amplification uses the default multiplier", + costFunction.computeCost(metrics, proposedTaskCount, noThreshold).totalCost(), + costBelowTier1, + 0.0001 + ); + double lagPerPartition = aggregateLag / partitionCount; - double amplification = 1.0 + WeightedCostFunction.LAG_AMPLIFICATION_MULTIPLIER * Math.log(lagPerPartition); - double expected = aggregateLag * amplification / (proposedTaskCount * WeightedCostFunction.MIN_PROCESSING_RATE); + double criticalAmplification = + 1.0 + WeightedCostFunction.CRITICAL_LAG_AMPLIFICATION_MULTIPLIER * Math.log(lagPerPartition); + double expectedCriticalCost = + aggregateLag * criticalAmplification / (proposedTaskCount * WeightedCostFunction.MIN_PROCESSING_RATE); - Assert.assertEquals("Lag amplification should increase lag recovery time", expected, costWithAmp, 0.0001); + double costAtTier1 = costFunction.computeCost(metrics, proposedTaskCount, atTier1).totalCost(); + Assert.assertEquals( + "At/above tier1, the amplification multiplier maxes out at CRITICAL_LAG_AMPLIFICATION_MULTIPLIER", + expectedCriticalCost, + costAtTier1, + 0.0001 + ); + Assert.assertTrue( + "Critical-lag cost should exceed the default-multiplier cost for the same lag", + costAtTier1 > costBelowTier1 + ); } @Test - public void testAmplificationGrowsWithLag() + public void testNormalLagCostScalesLinearlyWithLag() { - // Verify that higher lag produces proportionally higher cost due to log amplification + // Without normal-path amplification, cost grows linearly with lag. CostBasedAutoScalerConfig lagOnly = CostBasedAutoScalerConfig.builder() .taskCountMax(100) .taskCountMin(1) @@ -382,12 +445,14 @@ public void testAmplificationGrowsWithLag() Assert.assertTrue("Higher lag should produce higher cost", highCost > lowCost); - // The ratio of costs should be more than the ratio of raw lags (due to amplification) + // The ratio of costs matches the ratio of raw lags. double lagRatio = 10_000.0 / 100.0; double costRatio = highCost / lowCost; - Assert.assertTrue( - "Amplification should make cost grow faster than linear with lag", - costRatio > lagRatio + Assert.assertEquals( + "Normal lag cost should grow linearly with lag", + lagRatio, + costRatio, + 0.0001 ); } @@ -541,6 +606,7 @@ private CostMetrics createMetrics( { return new CostMetrics( avgPartitionLag, + avgPartitionLag * partitionCount, currentTaskCount, partitionCount, pollIdleRatio, @@ -556,7 +622,7 @@ private CostMetrics createMetricsWithMaxObservedRate( double pollIdleRatio ) { - return new CostMetrics(0.0, 10, 100, pollIdleRatio, 3600, avgProcessingRate, maxObservedRate); + return new CostMetrics(0.0, 0.0, 10, 100, pollIdleRatio, 3600, avgProcessingRate, maxObservedRate); } private CostMetrics createMetricsWithRate( @@ -569,6 +635,7 @@ private CostMetrics createMetricsWithRate( { return new CostMetrics( avgPartitionLag, + avgPartitionLag * partitionCount, currentTaskCount, partitionCount, pollIdleRatio,