feat: Introduce lag emergency option for cost-based autoscaler

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:

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,

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()}:
+   * <ul>
+   *   <li>At 75% of this value, the lag-amplification multiplier maxes out at 6.0 (instead of
+   *   unamplified normal recovery), and the scale-up candidate search bypasses
+   *   {@link #isUseTaskCountBoundariesOnScaleUp()}.</li>
+   *   <li>At 95% of this value, cost minimization is skipped entirely and the task count jumps
+   *   straight to the maximum.</li>
+   * </ul>
+   * {@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
       );
     }

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 +

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);
     }