diff --git a/distribution/docker/docker-kafka-compose.yml b/distribution/docker/docker-kafka-compose.yml new file mode 100644 index 000000000000..277a2c908c3f --- /dev/null +++ b/distribution/docker/docker-kafka-compose.yml @@ -0,0 +1,224 @@ +# +# 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. +# + +# All Druid services share one locally-built image. Build context is the repo +# root because distribution/docker/Dockerfile does `COPY . /src`. +# +# On Apple Silicon the builder stage is pinned to linux/amd64 and emulating it +# is painfully slow. Prefer building the distribution natively first: +# mvn clean package -DskipTests -Pdist +# DRUID_BUILD_FROM_SOURCE=false docker compose -f distribution/docker/docker-kafka-compose.yml build +# Env shared by every Druid service. Kept in its own anchor because YAML merge +# keys do not deep-merge: any service declaring `environment:` replaces the +# anchor's wholesale, so it must re-merge this explicitly. +x-druid-env: &druid-env + # Overrides the loadList in `environment`, which is shared with docker-compose.yml. + # Same list plus druid-kafka-indexing-service, needed for Kafka supervisors. + druid_extensions_loadList: '["druid-histogram", "druid-datasketches", "druid-lookups-cached-global", "postgresql-metadata-storage", "druid-kafka-indexing-service"]' + +x-druid: &druid + image: apache/druid:local + build: + context: ../.. + dockerfile: distribution/docker/Dockerfile + args: + BUILD_FROM_SOURCE: "${DRUID_BUILD_FROM_SOURCE:-true}" + env_file: + - environment + environment: + <<: *druid-env + +volumes: + metadata_data: {} + middle_var: {} + historical_var: {} + broker_var: {} + coordinator_var: {} + overlord_var: {} + router_var: {} + druid_shared: {} + + +services: + postgres: + container_name: postgres + image: postgres:17.6 + ports: + - "5432:5432" + volumes: + - metadata_data:/var/lib/postgresql/data + environment: + - POSTGRES_PASSWORD=FoolishPassword + - POSTGRES_USER=druid + - POSTGRES_DB=druid + + # Needed by Druid only. Kafka 4.x is KRaft-only and does not use ZooKeeper. + zookeeper: + container_name: zookeeper + image: zookeeper:3.5.10 + ports: + - "2181:2181" + environment: + - ZOO_MY_ID=1 + + # Single-node Kafka in KRaft combined mode (broker + controller in one process). + # Two listeners: PLAINTEXT for in-network clients (Druid), EXTERNAL for the host. + kafka: + container_name: kafka + image: apache/kafka:4.3.0 + ports: + - "29092:29092" + environment: + KAFKA_NODE_ID: 1 + KAFKA_PROCESS_ROLES: broker,controller + KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093,EXTERNAL://0.0.0.0:29092 + KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092,EXTERNAL://localhost:29092 + KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,EXTERNAL:PLAINTEXT + KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:9093 + KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER + KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT + KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1 + KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1 + KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1 + KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0 + + # Creates the topic, then emits exactly one JSON event per second, forever. + kafka-producer: + container_name: kafka-producer + image: apache/kafka:4.3.0 + depends_on: + - kafka + entrypoint: ["/bin/sh", "-c"] + # $$ escapes compose interpolation so the shell sees a single $. + command: + - | + until /opt/kafka/bin/kafka-topics.sh --bootstrap-server kafka:9092 --list >/dev/null 2>&1; do + echo "waiting for kafka..." + sleep 1 + done + /opt/kafka/bin/kafka-topics.sh --bootstrap-server kafka:9092 \ + --create --if-not-exists --topic druid-events \ + --partitions 4 --replication-factor 1 + i=0 + while true; do + i=$$((i + 1)) + echo "{\"timestamp\":\"$$(date -u +%Y-%m-%dT%H:%M:%SZ)\",\"page\":\"page-$$((i % 10))\",\"user\":\"user-$$((i % 100))\",\"latency\":$$((i % 500)),\"count\":1}" + sleep 1 + done | /opt/kafka/bin/kafka-console-producer.sh \ + --bootstrap-server kafka:9092 --topic druid-events + + coordinator: + <<: *druid + container_name: coordinator + volumes: + - druid_shared:/opt/shared + - coordinator_var:/opt/druid/var + depends_on: + - zookeeper + - postgres + ports: + - "8081:8081" + command: + - coordinator + environment: + <<: *druid-env + # The shared coordinator-overlord config enables this; turn it off so the + # standalone overlord below is the only one. + druid_coordinator_asOverlord_enabled: "false" + + # druid.sh maps both `coordinator` and `overlord` to the coordinator-overlord + # config dir, which hardcodes druid.service=druid/coordinator on port 8081. + # Override both so this announces itself as a real, separate overlord. + overlord: + <<: *druid + container_name: overlord + volumes: + - druid_shared:/opt/shared + - overlord_var:/opt/druid/var + depends_on: + - zookeeper + - postgres + ports: + - "8090:8090" + command: + - overlord + environment: + <<: *druid-env + druid_service: druid/overlord + druid_plaintextPort: "8090" + + broker: + <<: *druid + container_name: broker + volumes: + - broker_var:/opt/druid/var + depends_on: + - zookeeper + - postgres + - coordinator + ports: + - "8082:8082" + command: + - broker + + historical: + <<: *druid + container_name: historical + volumes: + - druid_shared:/opt/shared + - historical_var:/opt/druid/var + depends_on: + - zookeeper + - postgres + - coordinator + ports: + - "8083:8083" + command: + - historical + + middlemanager: + <<: *druid + container_name: middlemanager + volumes: + - druid_shared:/opt/shared + - middle_var:/opt/druid/var + depends_on: + - zookeeper + - postgres + - coordinator + - overlord + ports: + - "8091:8091" + - "8100-8105:8100-8105" + command: + - middleManager + + router: + <<: *druid + container_name: router + volumes: + - router_var:/opt/druid/var + depends_on: + - zookeeper + - postgres + - coordinator + ports: + - "8888:8888" + command: + - router 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 9f599f3aee8f..942c0f38ae66 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; @@ -641,6 +645,75 @@ 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 + ); + + // 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( + currentTaskCount >= config.getTaskCountMin() && currentTaskCount <= config.getTaskCountMax(), + "Value of currentTaskCount[%d] must be within taskCountMin[%d] and taskCountMax[%d]", + currentTaskCount, config.getTaskCountMin(), 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; + final CostMetrics costMetrics = new CostMetrics( + observedAggregateLag / partitionCount, + currentTaskCount, + partitionCount, + idleRatio, + taskDurationSeconds, + avgProcessingRatePerTask, + maxProcessingRatePerTask * 1.0 + ); + final int optimalTaskCount = autoscaleSimulator.computeOptimalTaskCount(costMetrics); + 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..abf62c87f922 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 @@ -36,9 +36,12 @@ import org.apache.druid.audit.AuditEntry; import org.apache.druid.audit.AuditManager; import org.apache.druid.error.DruidException; +import org.apache.druid.error.InvalidInput; 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.CostBasedAutoScaler; +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; @@ -57,6 +60,7 @@ import javax.annotation.Nullable; import javax.servlet.http.HttpServletRequest; import javax.ws.rs.Consumes; +import javax.ws.rs.DefaultValue; import javax.ws.rs.GET; import javax.ws.rs.POST; import javax.ws.rs.Path; @@ -183,7 +187,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 +540,52 @@ public Response terminateAll(@Context final HttpServletRequest req) ); } + @GET + @Path("/{id}/autoscaler") + @Produces(MediaType.APPLICATION_JSON) + @ResourceFilters(SupervisorResourceFilter.class) + public Response simulateAutoscaling( + @PathParam("id") String supervisorId, + @QueryParam("autoScalerStrategy") @DefaultValue(CostBasedAutoScaler.AUTOSCALER_TYPE_NAME) String autoScalerStrategy, + @QueryParam("taskCountMin") int taskCountMin, + @QueryParam("taskCountMax") int taskCountMax, + @QueryParam("maxProcessingRatePerTask") int maxProcessingRatePerTask, + @QueryParam("optimalTaskIdleRatio") double optimalTaskIdleRatio, + @QueryParam("lagWeight") Double lagWeight, + @QueryParam("idleWeight") Double idleWeight, + @QueryParam("criticalLag") int criticalLag, + @QueryParam("currentTaskCount") Integer currentTaskCount, + @Context HttpServletRequest request + ) + { + if (!CostBasedAutoScaler.AUTOSCALER_TYPE_NAME.equals(autoScalerStrategy)) { + throw InvalidInput.exception( + "Cannot simulate autoScalerStrategy[%s]. Only [%s] is supported.", + autoScalerStrategy, + CostBasedAutoScaler.AUTOSCALER_TYPE_NAME + ); + } + final CostBasedAutoScalerConfig autoScalerConfig = + CostBasedAutoScalerConfig.forSimulation( + taskCountMin, + taskCountMax, + optimalTaskIdleRatio, + lagWeight, + idleWeight + ); + return asLeaderWithSupervisorManager( + manager -> Response.ok( + manager.simulateAutoscaling( + supervisorId, + autoScalerConfig, + criticalLag, + maxProcessingRatePerTask, + currentTaskCount + ) + ).build() + ); + } + @GET @Path("/history") @Produces(MediaType.APPLICATION_JSON) @@ -562,7 +614,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 b089df3bcd1a..f35f76005d89 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_LAG_COST_METRIC = "task/autoScaler/costBased/currentLagCost"; @@ -96,6 +97,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; @@ -121,10 +123,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) @@ -217,12 +243,12 @@ 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) { 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) @@ -302,21 +328,21 @@ 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())); + 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())); // 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())); + emitMetric(getMetricBuilder().setMetric(AVG_PROCESSING_RATE_METRIC, metrics.getAvgProcessingRate())); } if (metrics.getPollIdleRatio() >= 0) { - emitter.emit(getMetricBuilder().setMetric(AVG_POLL_IDLE_RATIO, metrics.getPollIdleRatio())); + emitMetric(getMetricBuilder().setMetric(AVG_POLL_IDLE_RATIO, metrics.getPollIdleRatio())); } if (idleRatioEstimatedFromRate >= 0) { - emitter.emit(getMetricBuilder().setMetric(IDLE_RATIO_ESTIMATED_FROM_RATE, idleRatioEstimatedFromRate)); + emitMetric(getMetricBuilder().setMetric(IDLE_RATIO_ESTIMATED_FROM_RATE, idleRatioEstimatedFromRate)); } if (optimalTaskCount != currentTaskCount) { @@ -324,8 +350,8 @@ int computeOptimalTaskCount(CostMetrics metrics) "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())); + emitMetric(getMetricBuilder().setMetric(OPTIMAL_LAG_COST_METRIC, optimalCost.lagCost())); + emitMetric(getMetricBuilder().setMetric(OPTIMAL_IDLE_COST_METRIC, optimalCost.idleCost())); } // Scale-up is applied eagerly; scale-down may be deferred by computeTaskCountForScaleAction(). @@ -542,4 +568,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 785fa3257b5c..6c64d804b968 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 @@ -150,6 +150,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/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.tsx b/web-console/src/dialogs/supervisor-table-action-dialog/auto-scaler-panel/auto-scaler-panel.tsx new file mode 100644 index 000000000000..9bd76cd23cc0 --- /dev/null +++ b/web-console/src/dialogs/supervisor-table-action-dialog/auto-scaler-panel/auto-scaler-panel.tsx @@ -0,0 +1,267 @@ +/* + * 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 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 [dataState] = useQueryManager< + { + supervisorId: string; + taskCountMin: number; + taskCountMax: number; + maxProcessingRatePerTask: number; + optimalTaskIdleRatio: number; + lagWeight: number; + idleWeight: number; + criticalLag: number; + currentTaskCount: number | undefined; + }, + AutoScalerRow[] + >({ + query, + debounceIdle: 300, + debounceLoading: 500, + processQuery: async (params, signal) => { + const resp = await Api.instance.get<{ data: AutoScalerRow[] }>( + `/druid/indexer/v1/supervisor/${Api.encodePath(params.supervisorId)}/autoscaler`, + { + params: { + taskCountMin: params.taskCountMin, + taskCountMax: params.taskCountMax, + maxProcessingRatePerTask: params.maxProcessingRatePerTask, + optimalTaskIdleRatio: params.optimalTaskIdleRatio, + lagWeight: params.lagWeight, + idleWeight: params.idleWeight, + 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 = 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 31a64f1bf0c2..3860fe8feb78 100644 --- a/web-console/src/views/supervisors-view/supervisors-view.tsx +++ b/web-console/src/views/supervisors-view/supervisors-view.tsx @@ -207,6 +207,7 @@ export interface SupervisorsViewState { alertErrorMsg?: string; supervisorTableActionDialogId?: string; + supervisorTableActionDialogType?: string; supervisorTableActionDialogActions: BasicAction[]; visibleColumns: LocalStorageBackedVisibility; @@ -802,6 +803,7 @@ export class SupervisorsView extends React.PureComponent< private onSupervisorDetail(supervisor: SupervisorQueryResultRow) { this.setState({ supervisorTableActionDialogId: supervisor.supervisor_id, + supervisorTableActionDialogType: supervisor.type, supervisorTableActionDialogActions: this.getSupervisorActions(supervisor), }); } @@ -1311,6 +1313,7 @@ export class SupervisorsView extends React.PureComponent< supervisorSpecDialogOpen, alertErrorMsg, supervisorTableActionDialogId, + supervisorTableActionDialogType, supervisorTableActionDialogActions, visibleColumns, } = this.state; @@ -1378,8 +1381,14 @@ export class SupervisorsView extends React.PureComponent< {supervisorTableActionDialogId && ( this.setState({ supervisorTableActionDialogId: undefined })} + onClose={() => + this.setState({ + supervisorTableActionDialogId: undefined, + supervisorTableActionDialogType: undefined, + }) + } /> )}