diff --git a/embedded-tests/src/test/java/org/apache/druid/testing/embedded/indexing/KafkaClusterMetricsTest.java b/embedded-tests/src/test/java/org/apache/druid/testing/embedded/indexing/KafkaClusterMetricsTest.java index bc7385e6bf6e..41e2ee2b73e8 100644 --- a/embedded-tests/src/test/java/org/apache/druid/testing/embedded/indexing/KafkaClusterMetricsTest.java +++ b/embedded-tests/src/test/java/org/apache/druid/testing/embedded/indexing/KafkaClusterMetricsTest.java @@ -428,6 +428,7 @@ private KafkaSupervisorSpec createKafkaSupervisor( .withConsumerProperties(kafkaServer.consumerProperties()) .withTaskCount(taskCount) ) + .withContext(Map.of("useConcurrentLocks", true)) .withId(supervisorId) .build(dataSource, TOPIC); } diff --git a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/supervisor/SupervisorManager.java b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/supervisor/SupervisorManager.java index 7c75e5a96423..32c712f31e54 100644 --- a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/supervisor/SupervisorManager.java +++ b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/supervisor/SupervisorManager.java @@ -26,6 +26,7 @@ import com.google.common.collect.ImmutableMap; import com.google.common.util.concurrent.ListenableFuture; import com.google.inject.Inject; +import org.apache.druid.common.config.Configs; import org.apache.druid.common.guava.FutureUtils; import org.apache.druid.common.utils.IdUtils; import org.apache.druid.error.DruidException; @@ -40,6 +41,9 @@ import org.apache.druid.indexing.seekablestream.supervisor.BoundedStreamConfig; import org.apache.druid.indexing.seekablestream.supervisor.SeekableStreamSupervisor; import org.apache.druid.indexing.seekablestream.supervisor.SeekableStreamSupervisorSpec; +import org.apache.druid.indexing.seekablestream.supervisor.autoscaler.CostBasedAutoScaler; +import org.apache.druid.indexing.seekablestream.supervisor.autoscaler.CostBasedAutoScalerConfig; +import org.apache.druid.indexing.seekablestream.supervisor.autoscaler.CostMetrics; import org.apache.druid.java.util.common.IAE; import org.apache.druid.java.util.common.ISE; import org.apache.druid.java.util.common.Pair; @@ -72,6 +76,9 @@ public class SupervisorManager implements SupervisorStatsProvider { private static final EmittingLogger log = new EmittingLogger(SupervisorManager.class); + // Caps the autoscaler simulation's range (we're not aware of workloads higher than 1k partitions).. + private static final int MAX_SIMULATION_TASK_COUNT = 1000; + private final MetadataSupervisorManager metadataSupervisorManager; private final ConcurrentHashMap> supervisors = new ConcurrentHashMap<>(); // SupervisorTaskAutoScaler could be null @@ -644,6 +651,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 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 + ); + InvalidInput.conditionalException( + config.getTaskCountMin() >= 1, + "Value of taskCountMin[%d] must be 1 or more", + config.getTaskCountMin() + ); + InvalidInput.conditionalException( + config.getTaskCountMax() <= MAX_SIMULATION_TASK_COUNT, + "Value of taskCountMax[%d] must be [%d] or less", + config.getTaskCountMax(), MAX_SIMULATION_TASK_COUNT + ); + + // 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 = (double) lagStepSize * i; + final CostMetrics costMetrics = new CostMetrics( + observedAggregateLag / partitionCount, + simulationTaskCount, + partitionCount, + idleRatio, + taskDurationSeconds, + avgProcessingRatePerTask, + maxProcessingRatePerTask * 1.0 + ); + final int optimalTaskCount = autoscaleSimulator.computeOptimalTaskCountInternal(costMetrics, true); + rows[i] = Map.of("lag", observedAggregateLag, "taskCount", optimalTaskCount); + } + + // Collect the results and return + return Map.of("data", rows); + } + /** * Stops a supervisor with a given id and then removes it from the list. *

diff --git a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/supervisor/SupervisorResource.java b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/supervisor/SupervisorResource.java index 10297e25a88e..60e50fc01b69 100644 --- a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/supervisor/SupervisorResource.java +++ b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/supervisor/SupervisorResource.java @@ -39,6 +39,7 @@ import org.apache.druid.indexing.overlord.DataSourceMetadata; import org.apache.druid.indexing.overlord.TaskMaster; import org.apache.druid.indexing.overlord.http.security.SupervisorResourceFilter; +import org.apache.druid.indexing.seekablestream.supervisor.autoscaler.CostBasedAutoScalerConfig; import org.apache.druid.java.util.common.StringUtils; import org.apache.druid.java.util.common.UOE; import org.apache.druid.segment.incremental.ParseExceptionReport; @@ -183,7 +184,9 @@ public Response specPost( ); } - /** Audits supervisor spec submissions that changed or restarted the supervisor. */ + /** + * Audits supervisor spec submissions that changed or restarted the supervisor. + */ private void auditSupervisorUpdate(final SupervisorSpec spec, final HttpServletRequest req) { final String auditPayload @@ -534,6 +537,33 @@ public Response terminateAll(@Context final HttpServletRequest req) ); } + @POST + @Path("/{id}/autoscaler") + @Consumes(MediaType.APPLICATION_JSON) + @Produces(MediaType.APPLICATION_JSON) + @ResourceFilters(SupervisorResourceFilter.class) + public Response simulateAutoscaling( + @PathParam("id") String supervisorId, + CostBasedAutoScalerConfig autoScalerConfig, + @QueryParam("maxProcessingRatePerTask") int maxProcessingRatePerTask, + @QueryParam("criticalLag") int criticalLag, + @QueryParam("currentTaskCount") Integer currentTaskCount, + @Context HttpServletRequest request + ) + { + return asLeaderWithSupervisorManager( + manager -> Response.ok( + manager.simulateAutoscaling( + supervisorId, + autoScalerConfig, + criticalLag, + maxProcessingRatePerTask, + currentTaskCount + ) + ).build() + ); + } + @GET @Path("/history") @Produces(MediaType.APPLICATION_JSON) @@ -562,7 +592,12 @@ public Response specGetHistory( { if (count != null && count <= 0) { return Response.status(Response.Status.BAD_REQUEST) - .entity(ImmutableMap.of("error", StringUtils.format("Count must be greater than zero if set (count was %d)", count))) + .entity(ImmutableMap.of("error", + StringUtils.format( + "Count must be greater than zero if set (count was %d)", + count + ) + )) .build(); } 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..a6b9004393f5 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 @@ -63,6 +63,7 @@ public class CostBasedAutoScaler implements SupervisorTaskAutoScaler { private static final EmittingLogger log = new EmittingLogger(CostBasedAutoScaler.class); + public static final String AUTOSCALER_TYPE_NAME = "costBased"; public static final String LAG_WEIGHT_METRIC = "task/autoScaler/costBased/lagWeight"; public static final String IDLE_WEIGHT_METRIC = "task/autoScaler/costBased/idleWeight"; public static final String CURRENT_COST_METRIC = "task/autoScaler/costBased/currentCost"; @@ -98,6 +99,7 @@ public class CostBasedAutoScaler implements SupervisorTaskAutoScaler private final String supervisorId; private final SeekableStreamSupervisor supervisor; private final ServiceEmitter emitter; + private final boolean isSimulator; private final SupervisorSpec spec; private final CostBasedAutoScalerConfig config; private final ScheduledExecutorService autoscalerExecutor; @@ -123,10 +125,34 @@ public CostBasedAutoScaler( this.costFunction = new WeightedCostFunction(); this.autoscalerExecutor = Execs.scheduledSingleThreaded("CostBasedAutoScaler-" + StringUtils.encodeForFormat(spec.getId())); + this.isSimulator = false; + } + + private CostBasedAutoScaler(CostBasedAutoScalerConfig config, String supervisorId) + { + this.config = config; + this.costFunction = new WeightedCostFunction(); + this.isSimulator = true; + this.supervisorId = "simulator__" + supervisorId; + + this.spec = null; + this.supervisor = null; + this.emitter = null; + this.processingRateSamples = null; + this.autoscalerExecutor = null; + } + + public static CostBasedAutoScaler createSimulator(CostBasedAutoScalerConfig config, String supervisorId) + { + return new CostBasedAutoScaler(config, supervisorId); } private ServiceMetricEvent.Builder getMetricBuilder() { + if (isSimulator) { + return ServiceMetricEvent.builder(); + } + return ServiceMetricEvent.builder() .setDimension(DruidMetrics.SUPERVISOR_ID, supervisorId) @@ -219,12 +245,24 @@ public CostBasedAutoScalerConfig getConfig() * metrics are unusable. Returning the current task count means the current count is already * optimal (or no better candidate could be evaluated). */ - int computeOptimalTaskCount(CostMetrics metrics) + public int computeOptimalTaskCount(CostMetrics metrics) + { + return computeOptimalTaskCountInternal(metrics, false); + } + + /** + * 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 + * optimal (or no better candidate could be evaluated). + * + * @param isSimulation enables or disables task count computation logs and metrics + */ + public int computeOptimalTaskCountInternal(CostMetrics metrics, boolean isSimulation) { final Either result = validateMetricsForScaling(metrics); if (result.isError()) { log.debug("Valid metrics are not yet available for scaling supervisor[%s]", supervisorId); - emitter.emit( + emitMetric( getMetricBuilder() .setDimension(DruidMetrics.DESCRIPTION, result.error()) .setMetric(INVALID_METRICS_COUNT, 1L) @@ -256,20 +294,22 @@ int computeOptimalTaskCount(CostMetrics metrics) CostResult optimalCost = currentCost; final double idleRatioEstimatedFromRate = metrics.estimateIdleRatioFromProcessingRate(); - log.info( - "Computing optimal taskCount for supervisor[%s] with metrics:" - + " currentTaskCount[%d], avgPartitionLag[%.1f], avgProcessingRate[%.1f], maxProcessingRate[%.1f]" - + " idleRatio[%.1f], pollIdleRatio[%.1f], lagWeight[%.2f], idleWeight[%.2f].", - supervisorId, - currentTaskCount, - metrics.getAvgPartitionLag(), - metrics.getAvgProcessingRate(), - metrics.getMaxObservedRate(), - idleRatioEstimatedFromRate, - metrics.getPollIdleRatio(), - config.getLagWeight(), - config.getIdleWeight() - ); + if (!isSimulation) { + log.info( + "Computing optimal taskCount for supervisor[%s] with metrics:" + + " currentTaskCount[%d], avgPartitionLag[%.1f], avgProcessingRate[%.1f], maxProcessingRate[%.1f]" + + " idleRatio[%.1f], pollIdleRatio[%.1f], lagWeight[%.2f], idleWeight[%.2f].", + supervisorId, + currentTaskCount, + metrics.getAvgPartitionLag(), + metrics.getAvgProcessingRate(), + metrics.getMaxObservedRate(), + idleRatioEstimatedFromRate, + metrics.getPollIdleRatio(), + config.getLagWeight(), + config.getIdleWeight() + ); + } // Find the evaluated task count with the lowest cost. final CostResult[] costResults = new CostResult[validTaskCounts.length]; @@ -304,44 +344,46 @@ int computeOptimalTaskCount(CostMetrics metrics) } } - emitter.emit(getMetricBuilder().setMetric(OPTIMAL_TASK_COUNT_METRIC, (long) optimalTaskCount)); - emitter.emit(getMetricBuilder().setMetric(LAG_WEIGHT_METRIC, config.getLagWeight())); - emitter.emit(getMetricBuilder().setMetric(IDLE_WEIGHT_METRIC, config.getIdleWeight())); - emitter.emit(getMetricBuilder().setMetric(CURRENT_LAG_COST_METRIC, currentCost.lagCost())); - emitter.emit(getMetricBuilder().setMetric(CURRENT_IDLE_COST_METRIC, currentCost.idleCost())); - emitter.emit(getMetricBuilder().setMetric(CURRENT_COST_METRIC, currentCost.totalCost())); - emitter.emit(getMetricBuilder().setMetric(OPTIMAL_COST_METRIC, optimalCost.totalCost())); - - // Emit avg rate and idle metrics only if they are available - if (metrics.getAvgProcessingRate() >= 0) { - emitter.emit(getMetricBuilder().setMetric(AVG_PROCESSING_RATE_METRIC, metrics.getAvgProcessingRate())); - } - if (metrics.getPollIdleRatio() >= 0) { - emitter.emit(getMetricBuilder().setMetric(AVG_POLL_IDLE_RATIO, metrics.getPollIdleRatio())); - } - if (idleRatioEstimatedFromRate >= 0) { - emitter.emit(getMetricBuilder().setMetric(IDLE_RATIO_ESTIMATED_FROM_RATE, idleRatioEstimatedFromRate)); - } - - if (optimalTaskCount != currentTaskCount) { - log.info( - "Optimal taskCount[%d] for supervisor[%s] has lowest cost[%.4f] out of the following candidates: %n%s", - optimalTaskCount, supervisorId, optimalCost.totalCost(), constructCostTable(validTaskCounts, costResults) - ); - emitter.emit(getMetricBuilder().setMetric(OPTIMAL_LAG_COST_METRIC, optimalCost.lagCost())); - emitter.emit(getMetricBuilder().setMetric(OPTIMAL_IDLE_COST_METRIC, optimalCost.idleCost())); + if (!isSimulation) { + emitMetric(getMetricBuilder().setMetric(OPTIMAL_TASK_COUNT_METRIC, (long) optimalTaskCount)); + emitMetric(getMetricBuilder().setMetric(LAG_WEIGHT_METRIC, config.getLagWeight())); + emitMetric(getMetricBuilder().setMetric(IDLE_WEIGHT_METRIC, config.getIdleWeight())); + emitMetric(getMetricBuilder().setMetric(CURRENT_LAG_COST_METRIC, currentCost.lagCost())); + emitMetric(getMetricBuilder().setMetric(CURRENT_IDLE_COST_METRIC, currentCost.idleCost())); + emitMetric(getMetricBuilder().setMetric(CURRENT_COST_METRIC, currentCost.totalCost())); + emitMetric(getMetricBuilder().setMetric(OPTIMAL_COST_METRIC, optimalCost.totalCost())); + + // Emit avg rate and idle metrics only if they are available + if (metrics.getAvgProcessingRate() >= 0) { + emitMetric(getMetricBuilder().setMetric(AVG_PROCESSING_RATE_METRIC, metrics.getAvgProcessingRate())); + } + if (metrics.getPollIdleRatio() >= 0) { + emitMetric(getMetricBuilder().setMetric(AVG_POLL_IDLE_RATIO, metrics.getPollIdleRatio())); + } + if (idleRatioEstimatedFromRate >= 0) { + emitMetric(getMetricBuilder().setMetric(IDLE_RATIO_ESTIMATED_FROM_RATE, idleRatioEstimatedFromRate)); + } - final double costDropPercent - = 100.0 * (currentCost.totalCost() - optimalCost.totalCost()) / currentCost.totalCost(); - if (costDropPercent < config.getMinCostDropPercentForScaling()) { + if (optimalTaskCount != currentTaskCount) { log.info( - "Skipping scaling since cost drop percent[%.2f] is less than required minCostDropPercentForScaling[%d]", - costDropPercent, config.getMinCostDropPercentForScaling() + "Optimal taskCount[%d] for supervisor[%s] has lowest cost[%.4f] out of the following candidates: %n%s", + optimalTaskCount, supervisorId, optimalCost.totalCost(), constructCostTable(validTaskCounts, costResults) ); - return currentTaskCount; + emitMetric(getMetricBuilder().setMetric(OPTIMAL_LAG_COST_METRIC, optimalCost.lagCost())); + emitMetric(getMetricBuilder().setMetric(OPTIMAL_IDLE_COST_METRIC, optimalCost.idleCost())); } } + final double costDropPercent + = 100.0 * (currentCost.totalCost() - optimalCost.totalCost()) / currentCost.totalCost(); + if (costDropPercent < config.getMinCostDropPercentForScaling()) { + log.info( + "Skipping scaling since cost drop percent[%.2f] is less than required minCostDropPercentForScaling[%d]", + costDropPercent, config.getMinCostDropPercentForScaling() + ); + return currentTaskCount; + } + // Scale-up is applied eagerly; scale-down may be deferred by computeTaskCountForScaleAction(). return optimalTaskCount; } @@ -556,4 +598,13 @@ private Either validateMetricsForScaling(CostMetrics metrics) } } + /** + * Emits metric for the given builder if this is not a simulator. + */ + private void emitMetric(ServiceMetricEvent.Builder eventBuilder) + { + if (!isSimulator) { + emitter.emit(eventBuilder); + } + } } 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..97e21d8fe31b 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 @@ -153,6 +153,31 @@ public static Builder builder() return new Builder(); } + /** + * Config used to simulate the cost function without running an actual supervisor. + */ + public static CostBasedAutoScalerConfig forSimulation( + int taskCountMin, + int taskCountMax, + double optimalTaskIdleRatio, + @Nullable Double lagWeight, + @Nullable Double idleWeight + ) + { + final Builder builder = builder() + .taskCountMin(taskCountMin) + .taskCountMax(taskCountMax) + .optimalTaskIdleRatio(optimalTaskIdleRatio) + .enableTaskAutoScaler(true); + if (lagWeight != null) { + builder.lagWeight(lagWeight); + } + if (idleWeight != null) { + builder.idleWeight(idleWeight); + } + return builder.build(); + } + @Override @JsonProperty public boolean getEnableTaskAutoScaler() diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/overlord/supervisor/SupervisorManagerTest.java b/indexing-service/src/test/java/org/apache/druid/indexing/overlord/supervisor/SupervisorManagerTest.java index 7fbb130d2b07..fc22d1c2665b 100644 --- a/indexing-service/src/test/java/org/apache/druid/indexing/overlord/supervisor/SupervisorManagerTest.java +++ b/indexing-service/src/test/java/org/apache/druid/indexing/overlord/supervisor/SupervisorManagerTest.java @@ -47,6 +47,7 @@ import org.apache.druid.indexing.seekablestream.supervisor.SeekableStreamSupervisorIngestionSpec; import org.apache.druid.indexing.seekablestream.supervisor.SeekableStreamSupervisorSpec; import org.apache.druid.indexing.seekablestream.supervisor.SupervisorIOConfigBuilder; +import org.apache.druid.indexing.seekablestream.supervisor.autoscaler.CostBasedAutoScalerConfig; import org.apache.druid.jackson.DefaultObjectMapper; import org.apache.druid.java.util.common.DateTimes; import org.apache.druid.java.util.common.Intervals; @@ -628,6 +629,72 @@ private ConcurrentHashMap> getSuperviso return (ConcurrentHashMap>) field.get(manager); } + @Test + public void testSimulateAutoscalingUsesLiveTaskCountAboveConfiguredMaximum() throws Exception + { + final String supervisorId = "supervisor"; + final SeekableStreamSupervisor supervisor = EasyMock.createMock( + SeekableStreamSupervisor.class + ); + final SeekableStreamSupervisorIOConfig ioConfig = EasyMock.createMock(SeekableStreamSupervisorIOConfig.class); + EasyMock.expect(supervisor.getIoConfig()).andReturn(ioConfig); + EasyMock.expect(ioConfig.getTaskCount()).andReturn(20); + EasyMock.replay(supervisor, ioConfig); + getSupervisorsMap().put(supervisorId, Pair.of(supervisor, new TestSupervisorSpec(supervisorId, supervisor))); + + final Map result = manager.simulateAutoscaling( + supervisorId, + CostBasedAutoScalerConfig.forSimulation(1, 10, 0.1, null, null), + 1000, + 100, + null + ); + + final Object[] data = (Object[]) result.get("data"); + Assert.assertEquals(40, data.length); + Assert.assertTrue(data[0] instanceof Map); + final Map firstDataPoint = (Map) data[0]; + Assert.assertTrue(firstDataPoint.containsKey("lag")); + Assert.assertTrue(firstDataPoint.containsKey("taskCount")); + for (Object dataPoint : data) { + Assert.assertTrue(dataPoint instanceof Map); + final Number taskCount = (Number) ((Map) dataPoint).get("taskCount"); + Assert.assertTrue(taskCount.intValue() >= 1 && taskCount.intValue() <= 10); + } + EasyMock.verify(supervisor, ioConfig); + } + + @Test + public void testSimulateAutoscalingRejectsExplicitTaskCountAboveConfiguredMaximum() throws Exception + { + final String supervisorId = "supervisor"; + final SeekableStreamSupervisor supervisor = EasyMock.createMock( + SeekableStreamSupervisor.class + ); + final SeekableStreamSupervisorIOConfig ioConfig = EasyMock.createMock(SeekableStreamSupervisorIOConfig.class); + EasyMock.expect(supervisor.getIoConfig()).andReturn(ioConfig); + EasyMock.expect(ioConfig.getTaskCount()).andReturn(20); + EasyMock.replay(supervisor, ioConfig); + getSupervisorsMap().put(supervisorId, Pair.of(supervisor, new TestSupervisorSpec(supervisorId, supervisor))); + + MatcherAssert.assertThat( + Assert.assertThrows( + DruidException.class, + () -> manager.simulateAutoscaling( + supervisorId, + CostBasedAutoScalerConfig.forSimulation(1, 10, 0.1, null, null), + 1000, + 100, + 11 + ) + ), + DruidExceptionMatcher.invalidInput().expectMessageIs( + "Value of currentTaskCount[11] must be within taskCountMin[1] and taskCountMax[10]" + ) + ); + EasyMock.verify(supervisor, ioConfig); + } + @Test public void testHandoffTaskGroupsEarly() { diff --git a/web-console/src/dialogs/supervisor-table-action-dialog/auto-scaler-panel/auto-scaler-panel.scss b/web-console/src/dialogs/supervisor-table-action-dialog/auto-scaler-panel/auto-scaler-panel.scss new file mode 100644 index 000000000000..58bf23a47faf --- /dev/null +++ b/web-console/src/dialogs/supervisor-table-action-dialog/auto-scaler-panel/auto-scaler-panel.scss @@ -0,0 +1,68 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +.auto-scaler-panel { + display: flex; + flex-direction: column; + height: 100%; + padding: 15px; + overflow: auto; + + .auto-scaler-controls { + display: flex; + flex-wrap: wrap; + gap: 40px; + margin-bottom: 15px; + + .bp4-form-group { + margin: 0; + min-width: 220px; + + .bp4-label { + white-space: nowrap; + } + + .bp4-numeric-input { + width: 100px; + } + + .bp4-slider { + width: 200px; + min-width: 200px; + margin-top: -15px; + } + } + } + + .auto-scaler-chart-area { + position: relative; + flex: 1; + min-height: 300px; + + .auto-scaler-echart { + width: 100%; + height: 100%; + min-height: 300px; + } + + .auto-scaler-error { + color: #d5100a; + padding: 10px 0; + } + } +} diff --git a/web-console/src/dialogs/supervisor-table-action-dialog/auto-scaler-panel/auto-scaler-panel.spec.tsx b/web-console/src/dialogs/supervisor-table-action-dialog/auto-scaler-panel/auto-scaler-panel.spec.tsx new file mode 100644 index 000000000000..dbae5a662b8a --- /dev/null +++ b/web-console/src/dialogs/supervisor-table-action-dialog/auto-scaler-panel/auto-scaler-panel.spec.tsx @@ -0,0 +1,41 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, + * either express or implied. See the License for the specific + * language governing permissions and limitations under the License. + */ + +import { getAutoScalerValidationError } from './auto-scaler-panel'; + +describe('getAutoScalerValidationError', () => { + const validValues = { + taskCountMin: 1, + taskCountMax: 10, + maxProcessingRatePerTask: 100, + optimalTaskIdleRatio: 0.2, + criticalLag: 1000, + currentTaskCount: undefined, + }; + + it('validates the simulator inputs before making a request', () => { + expect(getAutoScalerValidationError(validValues)).toBeUndefined(); + expect(getAutoScalerValidationError({ ...validValues, taskCountMin: 11 })).toBeDefined(); + expect( + getAutoScalerValidationError({ ...validValues, maxProcessingRatePerTask: 99 }), + ).toBeDefined(); + expect(getAutoScalerValidationError({ ...validValues, optimalTaskIdleRatio: 0 })).toBeDefined(); + expect(getAutoScalerValidationError({ ...validValues, criticalLag: 999 })).toBeDefined(); + expect(getAutoScalerValidationError({ ...validValues, currentTaskCount: 11 })).toBeDefined(); + }); +}); diff --git a/web-console/src/dialogs/supervisor-table-action-dialog/auto-scaler-panel/auto-scaler-panel.tsx b/web-console/src/dialogs/supervisor-table-action-dialog/auto-scaler-panel/auto-scaler-panel.tsx new file mode 100644 index 000000000000..9e0c3172c3ee --- /dev/null +++ b/web-console/src/dialogs/supervisor-table-action-dialog/auto-scaler-panel/auto-scaler-panel.tsx @@ -0,0 +1,303 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +import { FormGroup, NumericInput, Slider } from '@blueprintjs/core'; +import type { ECharts } from 'echarts'; +import * as echarts from 'echarts'; +import React, { useEffect, useMemo, useRef, useState } from 'react'; + +import { Loader } from '../../../components/loader/loader'; +import { useQueryManager } from '../../../hooks'; +import { Api } from '../../../singletons'; + +import './auto-scaler-panel.scss'; + +interface AutoScalerRow { + lag: number; + taskCount: number; +} + +interface AutoScalerPanelProps { + supervisorId: string; +} + +export function getAutoScalerValidationError({ + taskCountMin, + taskCountMax, + maxProcessingRatePerTask, + optimalTaskIdleRatio, + criticalLag, + currentTaskCount, +}: { + taskCountMin: number; + taskCountMax: number; + maxProcessingRatePerTask: number; + optimalTaskIdleRatio: number; + criticalLag: number; + currentTaskCount: number | undefined; +}): string | undefined { + if (taskCountMin > taskCountMax) return 'Minimum task count must not exceed maximum task count'; + if (maxProcessingRatePerTask < 100) return 'Max processing rate / task must be at least 100'; + if (optimalTaskIdleRatio <= 0 || optimalTaskIdleRatio >= 1) { + return 'Optimal task idle ratio must be greater than 0 and less than 1'; + } + if (criticalLag < 1000) return 'Critical lag must be at least 1000'; + if ( + currentTaskCount !== undefined && + (currentTaskCount < taskCountMin || currentTaskCount > taskCountMax) + ) { + return 'Current task count must be within the minimum and maximum task count'; + } + return undefined; +} + +export const AutoScalerPanel = React.memo(function AutoScalerPanel(props: AutoScalerPanelProps) { + const { supervisorId } = props; + + const [taskCountMin, setTaskCountMin] = useState(1); + const [taskCountMax, setTaskCountMax] = useState(10); + const [maxProcessingRatePerTask, setMaxProcessingRatePerTask] = useState(10000); + const [optimalTaskIdleRatio, setOptimalTaskIdleRatio] = useState(0.2); + const [lagWeight, setLagWeight] = useState(0.4); + // Idle weight is the complement of lag weight; one slider drives both. + const idleWeight = Math.round((1 - lagWeight) * 10) / 10; + const [criticalLag, setCriticalLag] = useState(100000); + // Undefined means "let the server use the supervisor's live task count". + const [currentTaskCount, setCurrentTaskCount] = useState(undefined); + + const chartContainerRef = useRef(undefined); + const chartRef = useRef(undefined); + const query = useMemo( + () => ({ + supervisorId, + taskCountMin, + taskCountMax, + maxProcessingRatePerTask, + optimalTaskIdleRatio, + lagWeight, + idleWeight, + criticalLag, + currentTaskCount, + }), + [ + supervisorId, + taskCountMin, + taskCountMax, + maxProcessingRatePerTask, + optimalTaskIdleRatio, + lagWeight, + idleWeight, + criticalLag, + currentTaskCount, + ], + ); + const validationError = getAutoScalerValidationError(query); + + const [dataState] = useQueryManager< + { + supervisorId: string; + taskCountMin: number; + taskCountMax: number; + maxProcessingRatePerTask: number; + optimalTaskIdleRatio: number; + lagWeight: number; + idleWeight: number; + criticalLag: number; + currentTaskCount: number | undefined; + }, + AutoScalerRow[] + >({ + query: validationError ? undefined : query, + debounceIdle: 300, + debounceLoading: 500, + processQuery: async (params, signal) => { + const resp = await Api.instance.post<{ data: AutoScalerRow[] }>( + `/druid/indexer/v1/supervisor/${Api.encodePath(params.supervisorId)}/autoscaler`, + { + enableTaskAutoScaler: true, + taskCountMin: params.taskCountMin, + taskCountMax: params.taskCountMax, + optimalTaskIdleRatio: params.optimalTaskIdleRatio, + lagWeight: params.lagWeight, + idleWeight: params.idleWeight, + }, + { + params: { + maxProcessingRatePerTask: params.maxProcessingRatePerTask, + criticalLag: params.criticalLag, + currentTaskCount: params.currentTaskCount, + }, + signal, + }, + ); + return resp.data.data ?? (resp.data as any); + }, + }); + + function setupChart(container: HTMLDivElement): ECharts { + const myChart = echarts.init(container, 'dark'); + myChart.setOption({ + tooltip: { + trigger: 'axis', + }, + grid: { + left: '3%', + right: '4%', + bottom: '3%', + containLabel: true, + }, + xAxis: { + type: 'value', + name: 'Lag (records)', + nameLocation: 'middle', + nameGap: 30, + }, + yAxis: { + type: 'value', + name: 'Task count', + nameLocation: 'middle', + nameGap: 40, + }, + series: [ + { + name: 'Task count', + type: 'line', + showSymbol: false, + data: [], + }, + ], + }); + return myChart; + } + + useEffect(() => { + return () => { + chartRef.current?.dispose(); + }; + }, []); + + useEffect(() => { + const myChart = chartRef.current; + const data = dataState.data; + if (!myChart || !data) return; + + myChart.setOption({ + series: [ + { + data: data.map(row => [row.lag, row.taskCount]), + }, + ], + }); + }, [dataState.data]); + + useEffect(() => { + const myChart = chartRef.current; + if (!myChart) return; + myChart.resize(); + }, []); + + const errorMessage = validationError ?? dataState.getErrorMessage(); + + return ( +

+
+ + setTaskCountMin(v)} + buttonPosition="none" + fill + /> + + + setTaskCountMax(v)} + buttonPosition="none" + fill + /> + + + setMaxProcessingRatePerTask(v)} + buttonPosition="none" + fill + /> + + + setOptimalTaskIdleRatio(v)} + fill + /> + + + setCurrentTaskCount(isNaN(v) ? undefined : v)} + buttonPosition="none" + fill + /> + + + setCriticalLag(v)} + buttonPosition="none" + fill + /> + + + setLagWeight(Math.round(v * 10) / 10)} + /> + +
+
+ {errorMessage &&
{errorMessage}
} + {dataState.loading && } +
{ + if (chartRef.current || !container) return; + chartContainerRef.current = container; + chartRef.current = setupChart(container); + }} + /> +
+
+ ); +}); diff --git a/web-console/src/dialogs/supervisor-table-action-dialog/supervisor-table-action-dialog.tsx b/web-console/src/dialogs/supervisor-table-action-dialog/supervisor-table-action-dialog.tsx index 44ef05d5d59f..57e21c0cb3b6 100644 --- a/web-console/src/dialogs/supervisor-table-action-dialog/supervisor-table-action-dialog.tsx +++ b/web-console/src/dialogs/supervisor-table-action-dialog/supervisor-table-action-dialog.tsx @@ -26,12 +26,14 @@ import type { BasicAction } from '../../utils/basic-action'; import type { SideButtonMetaData } from '../table-action-dialog/table-action-dialog'; import { TableActionDialog } from '../table-action-dialog/table-action-dialog'; +import { AutoScalerPanel } from './auto-scaler-panel/auto-scaler-panel'; import { SupervisorStatisticsTable } from './supervisor-statistics-table/supervisor-statistics-table'; -type SupervisorTableActionDialogTab = 'status' | 'stats' | 'spec' | 'history'; +type SupervisorTableActionDialogTab = 'status' | 'stats' | 'spec' | 'history' | 'auto-scaler'; interface SupervisorTableActionDialogProps { supervisorId: string; + supervisorType?: string; actions: BasicAction[]; onClose: () => void; } @@ -39,9 +41,11 @@ interface SupervisorTableActionDialogProps { export const SupervisorTableActionDialog = React.memo(function SupervisorTableActionDialog( props: SupervisorTableActionDialogProps, ) { - const { supervisorId, actions, onClose } = props; + const { supervisorId, supervisorType, actions, onClose } = props; const [activeTab, setActiveTab] = useState('status'); + const isKafka = supervisorType === 'kafka'; + const supervisorTableSideButtonMetadata: SideButtonMetaData[] = [ { icon: 'dashboard', @@ -67,6 +71,16 @@ export const SupervisorTableActionDialog = React.memo(function SupervisorTableAc active: activeTab === 'history', onClick: () => setActiveTab('history'), }, + ...(isKafka + ? [ + { + icon: 'predictive-analysis' as const, + text: 'Auto-scaler', + active: activeTab === 'auto-scaler', + onClick: () => setActiveTab('auto-scaler'), + }, + ] + : []), ]; const supervisorEndpointBase = `/druid/indexer/v1/supervisor/${Api.encodePath(supervisorId)}`; @@ -98,6 +112,7 @@ export const SupervisorTableActionDialog = React.memo(function SupervisorTableAc /> )} {activeTab === 'history' && } + {activeTab === 'auto-scaler' && isKafka && } ); }); diff --git a/web-console/src/views/supervisors-view/supervisors-view.tsx b/web-console/src/views/supervisors-view/supervisors-view.tsx index 544b1b99f183..49a7362b7502 100644 --- a/web-console/src/views/supervisors-view/supervisors-view.tsx +++ b/web-console/src/views/supervisors-view/supervisors-view.tsx @@ -212,6 +212,7 @@ export interface SupervisorsViewState { alertErrorMsg?: string; supervisorTableActionDialogId?: string; + supervisorTableActionDialogType?: string; supervisorTableActionDialogActions: BasicAction[]; visibleColumns: LocalStorageBackedVisibility; @@ -810,6 +811,7 @@ export class SupervisorsView extends React.PureComponent< private onSupervisorDetail(supervisor: SupervisorQueryResultRow) { this.setState({ supervisorTableActionDialogId: supervisor.supervisor_id, + supervisorTableActionDialogType: supervisor.type, supervisorTableActionDialogActions: this.getSupervisorActions(supervisor), }); } @@ -1326,6 +1328,7 @@ export class SupervisorsView extends React.PureComponent< supervisorSpecDialogOpen, alertErrorMsg, supervisorTableActionDialogId, + supervisorTableActionDialogType, supervisorTableActionDialogActions, visibleColumns, } = this.state; @@ -1393,8 +1396,14 @@ export class SupervisorsView extends React.PureComponent< {supervisorTableActionDialogId && ( this.setState({ supervisorTableActionDialogId: undefined })} + onClose={() => + this.setState({ + supervisorTableActionDialogId: undefined, + supervisorTableActionDialogType: undefined, + }) + } /> )}