FrankChen021 commented on code in PR #19687:
URL: https://github.com/apache/druid/pull/19687#discussion_r3645259935
##########
indexing-service/src/main/java/org/apache/druid/indexing/overlord/supervisor/SupervisorManager.java:
##########
@@ -644,6 +648,80 @@ public boolean isAnotherTaskGroupPublishingToPartitions(
}
}
+ /**
+ * Simulates the effects of the {@code costBased} auto-scaler by computing
the optimal
+ * task count under various values of aggregate lag.
+ */
+ public Map<String, Object> simulateAutoscaling(
+ String supervisorId,
+ CostBasedAutoScalerConfig config,
+ int criticalLag,
+ int maxProcessingRatePerTask,
+ @Nullable Integer requestedTaskCount
+ )
+ {
+ // Validate that this is a streaming supervisor
+ final StreamSupervisor supervisor = requireStreamSupervisor(supervisorId,
"simulateAutoscaling");
+
+ // Validate the inputs
+ InvalidInput.conditionalException(
+ criticalLag >= 1000,
+ "Value of critical lag[%d] must be 1000 or more",
+ criticalLag
+ );
+ InvalidInput.conditionalException(
+ maxProcessingRatePerTask >= 100,
+ "Value of maxProcessingRatePerTask[%d] must be 100 events per second
or more",
+ maxProcessingRatePerTask
+ );
+
+ // Simulate from the supervisor's live task count unless the caller pins
one.
+ final int currentTaskCount = Configs.valueOrDefault(
+ requestedTaskCount,
+ ((SeekableStreamSupervisor<?, ?, ?>)
supervisor).getIoConfig().getTaskCount()
+ );
+ InvalidInput.conditionalException(
Review Comment:
[P1] Bound the simulation task counts at the API boundary
The request body can set `taskCountMin` to zero, which the config
constructor accepts; with a positive live task count this reaches
`computeValidTaskCounts` and divides by zero. It can also set an arbitrarily
large `taskCountMax`, which is reused as `partitionCount` and makes each of the
40 samples scan linearly through that range, allowing an authorized request to
monopolize the Overlord CPU. Require a positive minimum and impose a practical
maximum before running the simulation.
##########
indexing-service/src/main/java/org/apache/druid/indexing/overlord/supervisor/SupervisorManager.java:
##########
@@ -644,6 +648,80 @@ public boolean isAnotherTaskGroupPublishingToPartitions(
}
}
+ /**
+ * Simulates the effects of the {@code costBased} auto-scaler by computing
the optimal
+ * task count under various values of aggregate lag.
+ */
+ public Map<String, Object> simulateAutoscaling(
+ String supervisorId,
+ CostBasedAutoScalerConfig config,
+ int criticalLag,
+ int maxProcessingRatePerTask,
+ @Nullable Integer requestedTaskCount
+ )
+ {
+ // Validate that this is a streaming supervisor
+ final StreamSupervisor supervisor = requireStreamSupervisor(supervisorId,
"simulateAutoscaling");
+
+ // Validate the inputs
+ InvalidInput.conditionalException(
+ criticalLag >= 1000,
+ "Value of critical lag[%d] must be 1000 or more",
+ criticalLag
+ );
+ InvalidInput.conditionalException(
+ maxProcessingRatePerTask >= 100,
+ "Value of maxProcessingRatePerTask[%d] must be 100 events per second
or more",
+ maxProcessingRatePerTask
+ );
+
+ // Simulate from the supervisor's live task count unless the caller pins
one.
+ final int currentTaskCount = Configs.valueOrDefault(
+ requestedTaskCount,
+ ((SeekableStreamSupervisor<?, ?, ?>)
supervisor).getIoConfig().getTaskCount()
+ );
+ InvalidInput.conditionalException(
+ requestedTaskCount == null
+ || (currentTaskCount >= config.getTaskCountMin() && currentTaskCount
<= config.getTaskCountMax()),
+ "Value of currentTaskCount[%d] must be within taskCountMin[%d] and
taskCountMax[%d]",
+ currentTaskCount, config.getTaskCountMin(), config.getTaskCountMax()
+ );
+ final int simulationTaskCount = Math.max(
+ config.getTaskCountMin(),
+ Math.min(currentTaskCount, config.getTaskCountMax())
+ );
+
+ // Assumption: enough partitions to reach taskCountMax.
+ final int partitionCount = config.getTaskCountMax();
Review Comment:
[P2] Simulate the supervisor's actual topology
The live cost scaler uses `supervisor.getPartitionCount()` and the
supervisor IO config's task duration, but this endpoint substitutes
`taskCountMax` and a hard-coded hour. For example, a two-partition supervisor
configured with a maximum of ten is shown recommendations that cannot run, and
a custom task duration changes the lag-recovery cost curve. Read both values
from the selected supervisor so the chart predicts what its scaler would
actually choose.
##########
indexing-service/src/main/java/org/apache/druid/indexing/overlord/supervisor/SupervisorManager.java:
##########
@@ -644,6 +648,80 @@ public boolean isAnotherTaskGroupPublishingToPartitions(
}
}
+ /**
+ * Simulates the effects of the {@code costBased} auto-scaler by computing
the optimal
+ * task count under various values of aggregate lag.
+ */
+ public Map<String, Object> simulateAutoscaling(
+ String supervisorId,
+ CostBasedAutoScalerConfig config,
+ int criticalLag,
+ int maxProcessingRatePerTask,
+ @Nullable Integer requestedTaskCount
+ )
+ {
+ // Validate that this is a streaming supervisor
+ final StreamSupervisor supervisor = requireStreamSupervisor(supervisorId,
"simulateAutoscaling");
+
+ // Validate the inputs
+ InvalidInput.conditionalException(
+ criticalLag >= 1000,
+ "Value of critical lag[%d] must be 1000 or more",
+ criticalLag
+ );
+ InvalidInput.conditionalException(
+ maxProcessingRatePerTask >= 100,
+ "Value of maxProcessingRatePerTask[%d] must be 100 events per second
or more",
+ maxProcessingRatePerTask
+ );
+
+ // Simulate from the supervisor's live task count unless the caller pins
one.
+ final int currentTaskCount = Configs.valueOrDefault(
+ requestedTaskCount,
+ ((SeekableStreamSupervisor<?, ?, ?>)
supervisor).getIoConfig().getTaskCount()
+ );
+ InvalidInput.conditionalException(
+ requestedTaskCount == null
+ || (currentTaskCount >= config.getTaskCountMin() && currentTaskCount
<= config.getTaskCountMax()),
+ "Value of currentTaskCount[%d] must be within taskCountMin[%d] and
taskCountMax[%d]",
+ currentTaskCount, config.getTaskCountMin(), config.getTaskCountMax()
+ );
+ final int simulationTaskCount = Math.max(
+ config.getTaskCountMin(),
+ Math.min(currentTaskCount, config.getTaskCountMax())
+ );
+
+ // Assumption: enough partitions to reach taskCountMax.
+ final int partitionCount = config.getTaskCountMax();
+ final int taskDurationSeconds = 3600;
+
+ // Assume that the tasks are fully used since there is some lag
+ final double avgProcessingRatePerTask = maxProcessingRatePerTask;
+ final double idleRatio = config.getOptimalTaskIdleRatio();
+
+ // Invoke the cost function for a variety of input values of lag
+ final Object[] rows = new Object[40];
+ final int lagStepSize = criticalLag / 20;
+ final CostBasedAutoScaler autoscaleSimulator =
CostBasedAutoScaler.createSimulator(config, supervisorId);
+ for (int i = 0; i < 40; ++i) {
+ final double observedAggregateLag = lagStepSize * i * 1.0;
Review Comment:
[P2] Avoid overflowing the generated lag values
`lagStepSize * i` is evaluated as `int` before conversion to `double`. Large
but valid `criticalLag` query values therefore wrap to negative lag in later
samples, producing nonsensical task-count recommendations. Promote either
operand before multiplying, or keep the calculation in `long`/`double`
throughout.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
