FrankChen021 commented on code in PR #19687:
URL: https://github.com/apache/druid/pull/19687#discussion_r3728799562
##########
indexing-service/src/main/java/org/apache/druid/indexing/overlord/supervisor/SupervisorResource.java:
##########
@@ -534,6 +537,31 @@ public Response terminateAll(@Context final
HttpServletRequest req)
);
}
+ @POST
+ @Path("/{id}/autoscaler/simulate")
+ @Consumes(MediaType.APPLICATION_JSON)
+ @Produces(MediaType.APPLICATION_JSON)
+ @ResourceFilters(SupervisorResourceFilter.class)
Review Comment:
[P2] Authorize simulation as read-only
`SupervisorResourceFilter` maps every POST to `Action.WRITE`, so read-only
users who can inspect this supervisor receive 403 for a computation that
changes no state. The prior authorization finding remains unresolved; use an
explicit datasource `READ` authorization path for this endpoint.
##########
indexing-service/src/main/java/org/apache/druid/indexing/overlord/supervisor/SupervisorManager.java:
##########
@@ -644,6 +648,90 @@ 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 maxProcessingRatePerTask,
+ @Nullable Integer requestedTaskCount
+ )
+ {
+ // Validate that this is a SeekableStreamSupervisor
+ final Pair<Supervisor, SupervisorSpec> supervisorPair =
supervisors.get(supervisorId);
+ if (supervisorPair == null || supervisorPair.rhs == null ||
supervisorPair.lhs == null) {
+ throw NotFound.exception("Invalid supervisor[%s]", supervisorId);
+ } else if (!(supervisorPair.rhs instanceof SeekableStreamSupervisorSpec)) {
+ throw InvalidInput.exception(
+ "Cannot simulate autoscaling for supervisor[%s] of type[%s]",
+ supervisorId, supervisorPair.rhs.getType()
+ );
+ }
+
+ final SeekableStreamSupervisorSpec supervisorSpec =
(SeekableStreamSupervisorSpec) supervisorPair.rhs;
+
+ // Validate the inputs
+ final long criticalLag =
Configs.valueOrDefault(config.getCriticalLagThreshold(), 1_000_000);
+ 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
+ );
+ InvalidInput.conditionalException(
+ requestedTaskCount == null
+ || (requestedTaskCount >= config.getTaskCountMin() &&
requestedTaskCount <= config.getTaskCountMax()),
+ "Value of currentTaskCount[%d] must be within taskCountMin[%d] and
taskCountMax[%d]",
+ requestedTaskCount, config.getTaskCountMin(), config.getTaskCountMax()
+ );
+
+ // Simulate from the supervisor's live task count unless the caller pins
one.
+ final int currentTaskCount = supervisorSpec.getIoConfig().getTaskCount();
+ final int simulationTaskCount = Math.max(
+ config.getTaskCountMin(),
+ Math.min(
+ Configs.valueOrDefault(requestedTaskCount, currentTaskCount),
+ config.getTaskCountMax()
+ )
+ );
+
+ // Use the partition count and task duration from the supervisor spec
+ final int partitionCount = ((SeekableStreamSupervisor<?, ?, ?>)
supervisorPair.lhs).getKnownPartitionCount();
+ final long taskDurationSeconds =
supervisorSpec.getIoConfig().getTaskDuration().getStandardSeconds();
+
+ // Assume that the tasks are fully used since there is some lag
+ final double idleRatio = config.getOptimalTaskIdleRatio();
+
+ // Invoke the cost function for lag in the range [0, 2 *
criticalLagThreshold)
+ final Object[] rows = new Object[200];
+ final int lagStepSize = (int) (criticalLag / 100);
Review Comment:
[P3] Preserve the long critical-lag range
`criticalLagThreshold` is a `long`, but its step is narrowed to `int`.
Thresholds above 214,748,364,700 therefore produce wrapped or truncated lag
points instead of the promised 0-to-2x range. Keep `lagStepSize` as `long` or
`double`.
##########
indexing-service/src/main/java/org/apache/druid/indexing/overlord/supervisor/SupervisorManager.java:
##########
@@ -644,6 +648,90 @@ 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 maxProcessingRatePerTask,
+ @Nullable Integer requestedTaskCount
+ )
+ {
+ // Validate that this is a SeekableStreamSupervisor
+ final Pair<Supervisor, SupervisorSpec> supervisorPair =
supervisors.get(supervisorId);
+ if (supervisorPair == null || supervisorPair.rhs == null ||
supervisorPair.lhs == null) {
+ throw NotFound.exception("Invalid supervisor[%s]", supervisorId);
+ } else if (!(supervisorPair.rhs instanceof SeekableStreamSupervisorSpec)) {
+ throw InvalidInput.exception(
+ "Cannot simulate autoscaling for supervisor[%s] of type[%s]",
+ supervisorId, supervisorPair.rhs.getType()
+ );
+ }
+
+ final SeekableStreamSupervisorSpec supervisorSpec =
(SeekableStreamSupervisorSpec) supervisorPair.rhs;
+
+ // Validate the inputs
+ final long criticalLag =
Configs.valueOrDefault(config.getCriticalLagThreshold(), 1_000_000);
+ 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
+ );
+ InvalidInput.conditionalException(
Review Comment:
[P2] Reject non-positive task-count bounds
This validation checks only the optional current count. The config still
accepts `taskCountMin=0`; with a positive live task count,
`computeValidTaskCounts` divides by that minimum and the request fails with
HTTP 500. Require both task-count bounds to be at least one before simulation.
##########
indexing-service/src/main/java/org/apache/druid/indexing/overlord/supervisor/SupervisorManager.java:
##########
@@ -644,6 +648,90 @@ 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 maxProcessingRatePerTask,
+ @Nullable Integer requestedTaskCount
+ )
+ {
+ // Validate that this is a SeekableStreamSupervisor
+ final Pair<Supervisor, SupervisorSpec> supervisorPair =
supervisors.get(supervisorId);
+ if (supervisorPair == null || supervisorPair.rhs == null ||
supervisorPair.lhs == null) {
+ throw NotFound.exception("Invalid supervisor[%s]", supervisorId);
+ } else if (!(supervisorPair.rhs instanceof SeekableStreamSupervisorSpec)) {
+ throw InvalidInput.exception(
+ "Cannot simulate autoscaling for supervisor[%s] of type[%s]",
+ supervisorId, supervisorPair.rhs.getType()
+ );
+ }
+
+ final SeekableStreamSupervisorSpec supervisorSpec =
(SeekableStreamSupervisorSpec) supervisorPair.rhs;
+
+ // Validate the inputs
+ final long criticalLag =
Configs.valueOrDefault(config.getCriticalLagThreshold(), 1_000_000);
+ 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
+ );
+ InvalidInput.conditionalException(
+ requestedTaskCount == null
+ || (requestedTaskCount >= config.getTaskCountMin() &&
requestedTaskCount <= config.getTaskCountMax()),
+ "Value of currentTaskCount[%d] must be within taskCountMin[%d] and
taskCountMax[%d]",
+ requestedTaskCount, config.getTaskCountMin(), config.getTaskCountMax()
+ );
+
+ // Simulate from the supervisor's live task count unless the caller pins
one.
+ final int currentTaskCount = supervisorSpec.getIoConfig().getTaskCount();
+ final int simulationTaskCount = Math.max(
+ config.getTaskCountMin(),
+ Math.min(
+ Configs.valueOrDefault(requestedTaskCount, currentTaskCount),
+ config.getTaskCountMax()
+ )
+ );
+
+ // Use the partition count and task duration from the supervisor spec
+ final int partitionCount = ((SeekableStreamSupervisor<?, ?, ?>)
supervisorPair.lhs).getKnownPartitionCount();
+ final long taskDurationSeconds =
supervisorSpec.getIoConfig().getTaskDuration().getStandardSeconds();
+
+ // Assume that the tasks are fully used since there is some lag
+ final double idleRatio = config.getOptimalTaskIdleRatio();
+
+ // Invoke the cost function for lag in the range [0, 2 *
criticalLagThreshold)
+ final Object[] rows = new Object[200];
+ final int lagStepSize = (int) (criticalLag / 100);
+ final CostBasedAutoScaler autoscaleSimulator =
CostBasedAutoScaler.createSimulator(config, supervisorId);
+ for (int i = 0; i < 200; ++i) {
+ final double observedAggregateLag = (double) lagStepSize * i;
+ final CostMetrics costMetrics = new CostMetrics(
+ observedAggregateLag / partitionCount,
+ observedAggregateLag,
+ simulationTaskCount,
+ partitionCount,
+ idleRatio,
+ taskDurationSeconds,
+ maxProcessingRatePerTask,
+ maxProcessingRatePerTask * 1.0
+ );
+ final int optimalTaskCount =
autoscaleSimulator.computeOptimalTaskCountInternal(costMetrics, true);
Review Comment:
[P2] Suppress every simulator log path
Passing `true` suppresses the normal computation logs, but the high- and
critical-lag branches in `computeOptimalTaskCountInternal` still log
unconditionally. Each request evaluates 200 points; with the standard 0-to-2x
range, 125 cross the 0.75x threshold and emit INFO records. Guard those
branches, and the minimum-cost-drop log path, during simulation.
##########
indexing-service/src/main/java/org/apache/druid/indexing/overlord/supervisor/SupervisorManager.java:
##########
@@ -644,6 +648,90 @@ 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 maxProcessingRatePerTask,
+ @Nullable Integer requestedTaskCount
+ )
+ {
+ // Validate that this is a SeekableStreamSupervisor
+ final Pair<Supervisor, SupervisorSpec> supervisorPair =
supervisors.get(supervisorId);
+ if (supervisorPair == null || supervisorPair.rhs == null ||
supervisorPair.lhs == null) {
+ throw NotFound.exception("Invalid supervisor[%s]", supervisorId);
+ } else if (!(supervisorPair.rhs instanceof SeekableStreamSupervisorSpec)) {
+ throw InvalidInput.exception(
+ "Cannot simulate autoscaling for supervisor[%s] of type[%s]",
+ supervisorId, supervisorPair.rhs.getType()
+ );
+ }
+
+ final SeekableStreamSupervisorSpec supervisorSpec =
(SeekableStreamSupervisorSpec) supervisorPair.rhs;
+
+ // Validate the inputs
+ final long criticalLag =
Configs.valueOrDefault(config.getCriticalLagThreshold(), 1_000_000);
+ 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
+ );
+ InvalidInput.conditionalException(
+ requestedTaskCount == null
+ || (requestedTaskCount >= config.getTaskCountMin() &&
requestedTaskCount <= config.getTaskCountMax()),
+ "Value of currentTaskCount[%d] must be within taskCountMin[%d] and
taskCountMax[%d]",
+ requestedTaskCount, config.getTaskCountMin(), config.getTaskCountMax()
+ );
+
+ // Simulate from the supervisor's live task count unless the caller pins
one.
+ final int currentTaskCount = supervisorSpec.getIoConfig().getTaskCount();
+ final int simulationTaskCount = Math.max(
+ config.getTaskCountMin(),
+ Math.min(
+ Configs.valueOrDefault(requestedTaskCount, currentTaskCount),
+ config.getTaskCountMax()
+ )
+ );
+
+ // Use the partition count and task duration from the supervisor spec
+ final int partitionCount = ((SeekableStreamSupervisor<?, ?, ?>)
supervisorPair.lhs).getKnownPartitionCount();
Review Comment:
[P2] Reject unavailable live partition topology
`getKnownPartitionCount()` is zero during startup and temporarily while
assignments are cleared. In that state all 200 samples return `taskCount=-1` in
a successful response, producing a bogus chart. Return a retryable or
invalid-input error until a positive live partition count is available.
--
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]
