From f16ee188b1870d220b3545de4c233a65192062ae Mon Sep 17 00:00:00 2001 From: Kashif Faraz Date: Fri, 10 Jul 2026 10:24:52 +0530 Subject: [PATCH 1/6] wip: temp changes --- .../indexing/KafkaClusterMetricsTest.java | 30 +-- .../SegmentTransactionalReplaceAction.java | 4 + .../supervisor/SupervisorManager.java | 60 +++++ .../supervisor/SupervisorResource.java | 45 ++++ .../autoscaler/CostBasedAutoScaler.java | 48 +++- .../auto-scaler-panel/auto-scaler-panel.scss | 62 +++++ .../auto-scaler-panel/auto-scaler-panel.tsx | 215 ++++++++++++++++++ .../supervisor-table-action-dialog.tsx | 21 +- .../supervisors-view/supervisors-view.tsx | 11 +- 9 files changed, 472 insertions(+), 24 deletions(-) create mode 100644 web-console/src/dialogs/supervisor-table-action-dialog/auto-scaler-panel/auto-scaler-panel.scss create mode 100644 web-console/src/dialogs/supervisor-table-action-dialog/auto-scaler-panel/auto-scaler-panel.tsx 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..4be5adb3270f 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 @@ -141,10 +141,10 @@ public void stop() public static Stream getCompactionSupervisorTestParams() { return Stream.of( - Arguments.of( - CompactionEngine.NATIVE, - new NewestSegmentFirstPolicy(null) - ), +// Arguments.of( +// CompactionEngine.NATIVE, +// new NewestSegmentFirstPolicy(null) +// ), Arguments.of( CompactionEngine.MSQ, new MostFragmentedIntervalFirstPolicy(1, HumanReadableBytes.valueOf(1), null, null, 80, null) @@ -208,8 +208,8 @@ public void test_ingestClusterMetrics_withConcurrentCompactionSupervisor_andSkip CompactionCandidateSearchPolicy policy ) { - final int maxRowsPerSegment = 500; - final int compactedMaxRowsPerSegment = 5000; + final int maxRowsPerSegment = 10_000; + final int compactedMaxRowsPerSegment = 5_000; final int taskCount = 2; @@ -223,15 +223,15 @@ public void test_ingestClusterMetrics_withConcurrentCompactionSupervisor_andSkip cluster.callApi().postSupervisor(kafkaSupervisorSpec); // Wait for a task to succeed - overlord.latchableEmitter().waitForEventAggregate( - event -> event.hasMetricName("task/success/count") - .hasDimension(DruidMetrics.DATASOURCE, dataSource) - .hasDimension(DruidMetrics.SUPERVISOR_ID, supervisorId), - agg -> agg.hasSumAtLeast(1) - ); +// overlord.latchableEmitter().waitForEventAggregate( +// event -> event.hasMetricName("task/success/count") +// .hasDimension(DruidMetrics.DATASOURCE, dataSource) +// .hasDimension(DruidMetrics.SUPERVISOR_ID, supervisorId), +// agg -> agg.hasSumAtLeast(1) +// ); // Wait for some segments to be published - overlord.latchableEmitter().waitForEvent( - event -> event.hasMetricName("segment/txn/success") + indexer.latchableEmitter().waitForEvent( + event -> event.hasMetricName("ingest/events/processed") .hasDimension(DruidMetrics.DATASOURCE, dataSource) ); @@ -427,7 +427,9 @@ private KafkaSupervisorSpec createKafkaSupervisor( ioConfig -> ioConfig .withConsumerProperties(kafkaServer.consumerProperties()) .withTaskCount(taskCount) + .withTaskDuration(Period.minutes(10)) ) + .withContext(Map.of("useConcurrentLocks", true)) .withId(supervisorId) .build(dataSource, TOPIC); } diff --git a/indexing-service/src/main/java/org/apache/druid/indexing/common/actions/SegmentTransactionalReplaceAction.java b/indexing-service/src/main/java/org/apache/druid/indexing/common/actions/SegmentTransactionalReplaceAction.java index 6546ed80d9d9..022721719b47 100644 --- a/indexing-service/src/main/java/org/apache/druid/indexing/common/actions/SegmentTransactionalReplaceAction.java +++ b/indexing-service/src/main/java/org/apache/druid/indexing/common/actions/SegmentTransactionalReplaceAction.java @@ -169,6 +169,10 @@ private void registerUpgradedPendingSegmentsOnSupervisor( List upgradedPendingSegments ) { + if (upgradedPendingSegments.isEmpty()) { + return; + } + final SupervisorManager supervisorManager = toolbox.getSupervisorManager(); final Optional activeSupervisorIdWithAppendLock = supervisorManager.getActiveSupervisorIdForDatasourceWithAppendLock(task.getDataSource()); 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 fa7d96634ae6..87d383a347a0 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 @@ -40,6 +40,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; @@ -598,6 +601,63 @@ public boolean isAnotherTaskGroupPublishingToPartitions( } } + /** + * Simulates the effects of the auto-scaler by computing the optimal task count + * under various values of aggregate lag. + */ + public Map simulateAutoscalingWithLag( + String supervisorId, + CostBasedAutoScalerConfig config, + int criticalLag, + int maxProcessingRatePerTask + ) + { + // Validate that this is a streaming supervisor + requireStreamSupervisor(supervisorId, "simulateAutoscalingWithLag"); + + // 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 + ); + + // Assumptions + 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; + CostMetrics costMetrics = new CostMetrics( + observedAggregateLag / partitionCount, + 1, + partitionCount, + idleRatio, + taskDurationSeconds, + avgProcessingRatePerTask, + maxProcessingRatePerTask * 1.0 + ); + int optimalTaskCount = autoscaleSimulator.computeOptimalTaskCount(costMetrics); + rows[i] = Map.of("lag", observedAggregateLag, "task_count", optimalTaskCount, "usage_percent", 100.0); + } + + // 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 8d0e04eb7988..6e7da2a900d6 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; @@ -523,6 +524,50 @@ public Response terminateAll(@Context final HttpServletRequest req) ); } + @GET + @Path("/{id}/autoscaler") + @Produces(MediaType.APPLICATION_JSON) + @ResourceFilters(SupervisorResourceFilter.class) + public Response simulateAutoscalingWithLag( + @PathParam("id") String supervisorId, + @QueryParam("taskCountMin") int taskCountMin, + @QueryParam("taskCountMax") int taskCountMax, + @QueryParam("maxProcessingRatePerTask") int maxProcessingRatePerTask, + @QueryParam("optimalTaskIdleRatio") double optimalTaskIdleRatio, + @QueryParam("criticalLag") int criticalLag, + @Context HttpServletRequest request + ) + { + final CostBasedAutoScalerConfig autoScalerConfig = new CostBasedAutoScalerConfig( + taskCountMax, + taskCountMin, + true, + null, + null, + null, + null, + null, + null, + optimalTaskIdleRatio, + null, + null, + null, + null, + null, + false + ); + return asLeaderWithSupervisorManager( + manager -> Response.ok( + manager.simulateAutoscalingWithLag( + supervisorId, + autoScalerConfig, + criticalLag, + maxProcessingRatePerTask + ) + ).build() + ); + } + @GET @Path("/history") @Produces(MediaType.APPLICATION_JSON) 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 bf07545aaa53..ab12e9efa9bd 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 @@ -91,6 +91,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; @@ -116,10 +117,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) @@ -212,12 +237,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) @@ -296,11 +321,11 @@ int computeOptimalTaskCount(CostMetrics metrics) } } - emitter.emit(getMetricBuilder().setMetric(OPTIMAL_TASK_COUNT_METRIC, (long) optimalTaskCount)); - emitter.emit(getMetricBuilder().setMetric(LAG_WEIGHT_METRIC, optimalCost.lagCost())); - emitter.emit(getMetricBuilder().setMetric(IDLE_WEIGHT_METRIC, optimalCost.idleCost())); - emitter.emit(getMetricBuilder().setMetric(AVG_PROCESSING_RATE_METRIC, metrics.getAvgProcessingRate())); - emitter.emit(getMetricBuilder().setMetric(AVG_POLL_IDLE_RATIO, metrics.getPollIdleRatio())); + emitMetric(getMetricBuilder().setMetric(OPTIMAL_TASK_COUNT_METRIC, (long) optimalTaskCount)); + emitMetric(getMetricBuilder().setMetric(LAG_WEIGHT_METRIC, optimalCost.lagCost())); + emitMetric(getMetricBuilder().setMetric(IDLE_WEIGHT_METRIC, optimalCost.idleCost())); + emitMetric(getMetricBuilder().setMetric(AVG_PROCESSING_RATE_METRIC, metrics.getAvgProcessingRate())); + emitMetric(getMetricBuilder().setMetric(AVG_POLL_IDLE_RATIO, metrics.getPollIdleRatio())); if (optimalTaskCount != currentTaskCount) { log.info( @@ -516,4 +541,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/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..04ac93bf2933 --- /dev/null +++ b/web-console/src/dialogs/supervisor-table-action-dialog/auto-scaler-panel/auto-scaler-panel.scss @@ -0,0 +1,62 @@ +/* + * 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: 10px; + margin-bottom: 15px; + + .bp4-form-group { + margin: 0; + min-width: 220px; + + .bp4-label { + white-space: nowrap; + } + + .bp4-numeric-input { + width: 100px; + } + } + } + + .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..0791e2f748c6 --- /dev/null +++ b/web-console/src/dialogs/supervisor-table-action-dialog/auto-scaler-panel/auto-scaler-panel.tsx @@ -0,0 +1,215 @@ +/* + * 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 } 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; + task_count: number; + usage_percent: 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 [criticalLag, setCriticalLag] = useState(100000); + + const chartContainerRef = useRef(undefined); + const chartRef = useRef(undefined); + const query = useMemo( + () => ({supervisorId, taskCountMin, taskCountMax, maxProcessingRatePerTask, optimalTaskIdleRatio, criticalLag}), + [supervisorId, taskCountMin, taskCountMax, maxProcessingRatePerTask, optimalTaskIdleRatio, criticalLag] + ) + + const [dataState] = useQueryManager< + { + supervisorId: string; + taskCountMin: number; + taskCountMax: number; + maxProcessingRatePerTask: number; + optimalTaskIdleRatio: number; + criticalLag: number; + }, + AutoScalerRow[] + >({ + query, + 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, + criticalLag: params.criticalLag, + }, + 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.task_count]), + }, + ], + }); + }, [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)} + buttonPosition="none" + fill + /> + + + setCriticalLag(v)} + buttonPosition="none" + fill + /> + +
+
+ {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..4f5ec3253c4c 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 any, + text: 'Auto-scaler', + active: activeTab === 'auto-scaler', + onClick: () => setActiveTab('auto-scaler'), + }, + ] + : []), ]; const supervisorEndpointBase = `/druid/indexer/v1/supervisor/${Api.encodePath(supervisorId)}`; @@ -98,6 +112,9 @@ 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..cd41308693ee 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 + }) + } /> )}
From febfc600a7125fdf28eb1bb3530a48292f317f60 Mon Sep 17 00:00:00 2001 From: Kashif Faraz Date: Wed, 15 Jul 2026 14:07:51 +0530 Subject: [PATCH 2/6] Revert extra changes --- .../indexing/KafkaClusterMetricsTest.java | 20 +++++++++---------- 1 file changed, 10 insertions(+), 10 deletions(-) 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 4be5adb3270f..6ccff17c1239 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 @@ -141,10 +141,10 @@ public void stop() public static Stream getCompactionSupervisorTestParams() { return Stream.of( -// Arguments.of( -// CompactionEngine.NATIVE, -// new NewestSegmentFirstPolicy(null) -// ), + Arguments.of( + CompactionEngine.NATIVE, + new NewestSegmentFirstPolicy(null) + ), Arguments.of( CompactionEngine.MSQ, new MostFragmentedIntervalFirstPolicy(1, HumanReadableBytes.valueOf(1), null, null, 80, null) @@ -223,12 +223,12 @@ public void test_ingestClusterMetrics_withConcurrentCompactionSupervisor_andSkip cluster.callApi().postSupervisor(kafkaSupervisorSpec); // Wait for a task to succeed -// overlord.latchableEmitter().waitForEventAggregate( -// event -> event.hasMetricName("task/success/count") -// .hasDimension(DruidMetrics.DATASOURCE, dataSource) -// .hasDimension(DruidMetrics.SUPERVISOR_ID, supervisorId), -// agg -> agg.hasSumAtLeast(1) -// ); + overlord.latchableEmitter().waitForEventAggregate( + event -> event.hasMetricName("task/success/count") + .hasDimension(DruidMetrics.DATASOURCE, dataSource) + .hasDimension(DruidMetrics.SUPERVISOR_ID, supervisorId), + agg -> agg.hasSumAtLeast(1) + ); // Wait for some segments to be published indexer.latchableEmitter().waitForEvent( event -> event.hasMetricName("ingest/events/processed") From c70f681d9c82d896a5a440003d5583423444962d Mon Sep 17 00:00:00 2001 From: Kashif Faraz Date: Wed, 15 Jul 2026 14:25:04 +0530 Subject: [PATCH 3/6] Revert extra changes --- .../embedded/indexing/KafkaClusterMetricsTest.java | 9 ++++----- .../actions/SegmentTransactionalReplaceAction.java | 14 +++++++++++++- 2 files changed, 17 insertions(+), 6 deletions(-) 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 6ccff17c1239..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 @@ -208,8 +208,8 @@ public void test_ingestClusterMetrics_withConcurrentCompactionSupervisor_andSkip CompactionCandidateSearchPolicy policy ) { - final int maxRowsPerSegment = 10_000; - final int compactedMaxRowsPerSegment = 5_000; + final int maxRowsPerSegment = 500; + final int compactedMaxRowsPerSegment = 5000; final int taskCount = 2; @@ -230,8 +230,8 @@ public void test_ingestClusterMetrics_withConcurrentCompactionSupervisor_andSkip agg -> agg.hasSumAtLeast(1) ); // Wait for some segments to be published - indexer.latchableEmitter().waitForEvent( - event -> event.hasMetricName("ingest/events/processed") + overlord.latchableEmitter().waitForEvent( + event -> event.hasMetricName("segment/txn/success") .hasDimension(DruidMetrics.DATASOURCE, dataSource) ); @@ -427,7 +427,6 @@ private KafkaSupervisorSpec createKafkaSupervisor( ioConfig -> ioConfig .withConsumerProperties(kafkaServer.consumerProperties()) .withTaskCount(taskCount) - .withTaskDuration(Period.minutes(10)) ) .withContext(Map.of("useConcurrentLocks", true)) .withId(supervisorId) diff --git a/indexing-service/src/main/java/org/apache/druid/indexing/common/actions/SegmentTransactionalReplaceAction.java b/indexing-service/src/main/java/org/apache/druid/indexing/common/actions/SegmentTransactionalReplaceAction.java index cc7eefeba250..b4ad8accac55 100644 --- a/indexing-service/src/main/java/org/apache/druid/indexing/common/actions/SegmentTransactionalReplaceAction.java +++ b/indexing-service/src/main/java/org/apache/druid/indexing/common/actions/SegmentTransactionalReplaceAction.java @@ -24,12 +24,14 @@ import com.fasterxml.jackson.core.type.TypeReference; import com.google.common.annotations.VisibleForTesting; import com.google.common.collect.ImmutableSet; +import org.apache.druid.indexing.common.SegmentUpgradeMetrics; import org.apache.druid.indexing.common.task.IndexTaskUtils; import org.apache.druid.indexing.common.task.Task; import org.apache.druid.indexing.overlord.CriticalAction; import org.apache.druid.indexing.overlord.SegmentPublishResult; import org.apache.druid.indexing.overlord.supervisor.SupervisorManager; import org.apache.druid.java.util.common.logger.Logger; +import org.apache.druid.java.util.emitter.service.ServiceMetricEvent; import org.apache.druid.metadata.PendingSegmentRecord; import org.apache.druid.metadata.ReplaceTaskLock; import org.apache.druid.segment.SegmentSchemaMapping; @@ -180,10 +182,20 @@ void registerUpgradedPendingSegmentsOnSupervisor( List upgradedPendingSegments ) { - if (upgradedPendingSegments.isEmpty()) { + if (upgradedPendingSegments == null || upgradedPendingSegments.isEmpty()) { return; } + // Emit one persisted event per upgraded segment (rather than a single aggregate) regardless of whether a supervisor + // exists to receive them, so the total can be compared against the count actually announced by tasks and so each + // event carries the segment's interval and version. + for (PendingSegmentRecord upgradedPendingSegment : upgradedPendingSegments) { + final ServiceMetricEvent.Builder metricBuilder = new ServiceMetricEvent.Builder(); + IndexTaskUtils.setTaskDimensions(metricBuilder, task); + IndexTaskUtils.setPendingSegmentDimensions(metricBuilder, upgradedPendingSegment); + toolbox.getEmitter().emit(metricBuilder.setMetric(SegmentUpgradeMetrics.PERSISTED, 1)); + } + final SupervisorManager supervisorManager = toolbox.getSupervisorManager(); final List activeSupervisorIdWithAppendLock = supervisorManager.getActiveSupervisorIdsForDatasourceWithAppendLock(task.getDataSource()); From d763878aa9a701acadf2fc3cf14f7280f68eac84 Mon Sep 17 00:00:00 2001 From: Sasha Syrotenko Date: Thu, 16 Jul 2026 12:27:26 +0300 Subject: [PATCH 4/6] Enhance UI and backend codebase --- distribution/docker/docker-kafka-compose.yml | 224 ++++++++++++++++++ .../supervisor/SupervisorManager.java | 33 ++- .../supervisor/SupervisorResource.java | 59 +++-- .../autoscaler/CostBasedAutoScaler.java | 5 +- .../autoscaler/CostBasedAutoScalerConfig.java | 25 ++ .../auto-scaler-panel/auto-scaler-panel.tsx | 75 +++++- .../supervisor-table-action-dialog.tsx | 6 +- .../supervisors-view/supervisors-view.tsx | 4 +- 8 files changed, 383 insertions(+), 48 deletions(-) create mode 100644 distribution/docker/docker-kafka-compose.yml 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/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 8492547fc014..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; @@ -645,18 +646,19 @@ public boolean isAnotherTaskGroupPublishingToPartitions( } /** - * Simulates the effects of the auto-scaler by computing the optimal task count - * under various values of aggregate lag. + * Simulates the effects of the {@code costBased} auto-scaler by computing the optimal + * task count under various values of aggregate lag. */ - public Map simulateAutoscalingWithLag( + public Map simulateAutoscaling( String supervisorId, CostBasedAutoScalerConfig config, int criticalLag, - int maxProcessingRatePerTask + int maxProcessingRatePerTask, + @Nullable Integer requestedTaskCount ) { // Validate that this is a streaming supervisor - requireStreamSupervisor(supervisorId, "simulateAutoscalingWithLag"); + final StreamSupervisor supervisor = requireStreamSupervisor(supervisorId, "simulateAutoscaling"); // Validate the inputs InvalidInput.conditionalException( @@ -670,7 +672,18 @@ public Map simulateAutoscalingWithLag( maxProcessingRatePerTask ); - // Assumptions + // 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; @@ -684,17 +697,17 @@ public Map simulateAutoscalingWithLag( final CostBasedAutoScaler autoscaleSimulator = CostBasedAutoScaler.createSimulator(config, supervisorId); for (int i = 0; i < 40; ++i) { final double observedAggregateLag = lagStepSize * i * 1.0; - CostMetrics costMetrics = new CostMetrics( + final CostMetrics costMetrics = new CostMetrics( observedAggregateLag / partitionCount, - 1, + currentTaskCount, partitionCount, idleRatio, taskDurationSeconds, avgProcessingRatePerTask, maxProcessingRatePerTask * 1.0 ); - int optimalTaskCount = autoscaleSimulator.computeOptimalTaskCount(costMetrics); - rows[i] = Map.of("lag", observedAggregateLag, "task_count", optimalTaskCount, "usage_percent", 100.0); + final int optimalTaskCount = autoscaleSimulator.computeOptimalTaskCount(costMetrics); + rows[i] = Map.of("lag", observedAggregateLag, "taskCount", optimalTaskCount); } // Collect the results and return 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 7bd48282e29c..1bdf4f775222 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,6 +36,7 @@ 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; @@ -58,6 +59,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; @@ -75,6 +77,8 @@ import java.util.Set; import java.util.stream.Collectors; +import static org.apache.druid.indexing.seekablestream.supervisor.autoscaler.CostBasedAutoScaler.AUTOSCALER_TYPE_NAME; + /** * Endpoints for submitting and starting a {@link SupervisorSpec}, getting running supervisors, stopping supervisors, * and getting supervisor history. @@ -184,7 +188,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 @@ -539,41 +545,43 @@ public Response terminateAll(@Context final HttpServletRequest req) @Path("/{id}/autoscaler") @Produces(MediaType.APPLICATION_JSON) @ResourceFilters(SupervisorResourceFilter.class) - public Response simulateAutoscalingWithLag( + public Response simulateAutoscaling( @PathParam("id") String supervisorId, + @QueryParam("autoScalerStrategy") @DefaultValue(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 ) { - final CostBasedAutoScalerConfig autoScalerConfig = new CostBasedAutoScalerConfig( - taskCountMax, - taskCountMin, - true, - null, - null, - null, - null, - null, - null, - optimalTaskIdleRatio, - null, - null, - null, - null, - null, - false - ); + if (!AUTOSCALER_TYPE_NAME.equals(autoScalerStrategy)) { + throw InvalidInput.exception( + "Cannot simulate autoScalerStrategy[%s]. Only [%s] is supported.", + autoScalerStrategy, + AUTOSCALER_TYPE_NAME + ); + } + final CostBasedAutoScalerConfig autoScalerConfig = + CostBasedAutoScalerConfig.forSimulation( + taskCountMin, + taskCountMax, + optimalTaskIdleRatio, + lagWeight, + idleWeight + ); return asLeaderWithSupervisorManager( manager -> Response.ok( - manager.simulateAutoscalingWithLag( + manager.simulateAutoscaling( supervisorId, autoScalerConfig, criticalLag, - maxProcessingRatePerTask + maxProcessingRatePerTask, + currentTaskCount ) ).build() ); @@ -607,7 +615,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 2ba65254aa41..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"; @@ -124,7 +125,7 @@ public CostBasedAutoScaler( + StringUtils.encodeForFormat(spec.getId())); this.isSimulator = false; } - + private CostBasedAutoScaler(CostBasedAutoScalerConfig config, String supervisorId) { this.config = config; @@ -138,7 +139,7 @@ private CostBasedAutoScaler(CostBasedAutoScalerConfig config, String supervisorI this.processingRateSamples = null; this.autoscalerExecutor = null; } - + public static CostBasedAutoScaler createSimulator(CostBasedAutoScalerConfig config, String supervisorId) { return new CostBasedAutoScaler(config, supervisorId); 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.tsx b/web-console/src/dialogs/supervisor-table-action-dialog/auto-scaler-panel/auto-scaler-panel.tsx index 0791e2f748c6..65b6e3a7803a 100644 --- 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 @@ -29,8 +29,7 @@ import './auto-scaler-panel.scss'; interface AutoScalerRow { lag: number; - task_count: number; - usage_percent: number; + taskCount: number; } interface AutoScalerPanelProps { @@ -44,14 +43,38 @@ export const AutoScalerPanel = React.memo(function AutoScalerPanel(props: AutoSc const [taskCountMax, setTaskCountMax] = useState(10); const [maxProcessingRatePerTask, setMaxProcessingRatePerTask] = useState(10000); const [optimalTaskIdleRatio, setOptimalTaskIdleRatio] = useState(0.2); + const [lagWeight, setLagWeight] = useState(0.4); + const [idleWeight, setIdleWeight] = useState(0.6); 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, criticalLag}), - [supervisorId, taskCountMin, taskCountMax, maxProcessingRatePerTask, optimalTaskIdleRatio, criticalLag] - ) + () => ({ + supervisorId, + taskCountMin, + taskCountMax, + maxProcessingRatePerTask, + optimalTaskIdleRatio, + lagWeight, + idleWeight, + criticalLag, + currentTaskCount, + }), + [ + supervisorId, + taskCountMin, + taskCountMax, + maxProcessingRatePerTask, + optimalTaskIdleRatio, + lagWeight, + idleWeight, + criticalLag, + currentTaskCount, + ], + ); const [dataState] = useQueryManager< { @@ -60,11 +83,16 @@ export const AutoScalerPanel = React.memo(function AutoScalerPanel(props: AutoSc 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`, @@ -74,7 +102,10 @@ export const AutoScalerPanel = React.memo(function AutoScalerPanel(props: AutoSc taskCountMax: params.taskCountMax, maxProcessingRatePerTask: params.maxProcessingRatePerTask, optimalTaskIdleRatio: params.optimalTaskIdleRatio, + lagWeight: params.lagWeight, + idleWeight: params.idleWeight, criticalLag: params.criticalLag, + currentTaskCount: params.currentTaskCount, }, signal, }, @@ -133,7 +164,7 @@ export const AutoScalerPanel = React.memo(function AutoScalerPanel(props: AutoSc myChart.setOption({ series: [ { - data: data.map(row => [row.lag, row.task_count]), + data: data.map(row => [row.lag, row.taskCount]), }, ], }); @@ -182,12 +213,42 @@ export const AutoScalerPanel = React.memo(function AutoScalerPanel(props: AutoSc value={optimalTaskIdleRatio} min={0} max={1} - stepSize={0.05} + stepSize={0.1} + minorStepSize={0.01} onValueChange={v => setOptimalTaskIdleRatio(v)} + fill + /> + + + setCurrentTaskCount(isNaN(v) ? undefined : v)} buttonPosition="none" fill /> + + setLagWeight(v)} + fill + /> + + + setIdleWeight(v)} + fill + /> + setActiveTab('auto-scaler'), @@ -112,9 +112,7 @@ export const SupervisorTableActionDialog = React.memo(function SupervisorTableAc /> )} {activeTab === 'history' && } - {activeTab === 'auto-scaler' && isKafka && ( - - )} + {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 cd41308693ee..3860fe8feb78 100644 --- a/web-console/src/views/supervisors-view/supervisors-view.tsx +++ b/web-console/src/views/supervisors-view/supervisors-view.tsx @@ -207,7 +207,7 @@ export interface SupervisorsViewState { alertErrorMsg?: string; supervisorTableActionDialogId?: string; - supervisorTableActionDialogType?: string, + supervisorTableActionDialogType?: string; supervisorTableActionDialogActions: BasicAction[]; visibleColumns: LocalStorageBackedVisibility; @@ -1386,7 +1386,7 @@ export class SupervisorsView extends React.PureComponent< onClose={() => this.setState({ supervisorTableActionDialogId: undefined, - supervisorTableActionDialogType: undefined + supervisorTableActionDialogType: undefined, }) } /> From 1ab2cd358f3a9b61826cfaee3bc8e8cc1cbf6cab Mon Sep 17 00:00:00 2001 From: Sasha Syrotenko Date: Thu, 16 Jul 2026 16:09:57 +0300 Subject: [PATCH 5/6] Use slider instead --- .../auto-scaler-panel/auto-scaler-panel.scss | 8 ++++- .../auto-scaler-panel/auto-scaler-panel.tsx | 35 +++++++------------ 2 files changed, 20 insertions(+), 23 deletions(-) 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 index 04ac93bf2933..58bf23a47faf 100644 --- 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 @@ -26,7 +26,7 @@ .auto-scaler-controls { display: flex; flex-wrap: wrap; - gap: 10px; + gap: 40px; margin-bottom: 15px; .bp4-form-group { @@ -40,6 +40,12 @@ .bp4-numeric-input { width: 100px; } + + .bp4-slider { + width: 200px; + min-width: 200px; + margin-top: -15px; + } } } 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 index 65b6e3a7803a..9bd76cd23cc0 100644 --- 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 @@ -16,7 +16,7 @@ * limitations under the License. */ -import { FormGroup, NumericInput } from '@blueprintjs/core'; +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'; @@ -44,7 +44,8 @@ export const AutoScalerPanel = React.memo(function AutoScalerPanel(props: AutoSc const [maxProcessingRatePerTask, setMaxProcessingRatePerTask] = useState(10000); const [optimalTaskIdleRatio, setOptimalTaskIdleRatio] = useState(0.2); const [lagWeight, setLagWeight] = useState(0.4); - const [idleWeight, setIdleWeight] = useState(0.6); + // 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); @@ -229,26 +230,6 @@ export const AutoScalerPanel = React.memo(function AutoScalerPanel(props: AutoSc fill /> - - setLagWeight(v)} - fill - /> - - - setIdleWeight(v)} - fill - /> - + + setLagWeight(Math.round(v * 10) / 10)} + /> +
{errorMessage &&
{errorMessage}
} From 2134fceae5d0732bff9c566553780a880505273c Mon Sep 17 00:00:00 2001 From: Sasha Syrotenko Date: Mon, 20 Jul 2026 10:44:36 +0300 Subject: [PATCH 6/6] Checkstyle --- .../indexing/overlord/supervisor/SupervisorResource.java | 9 ++++----- 1 file changed, 4 insertions(+), 5 deletions(-) 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 1bdf4f775222..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 @@ -40,6 +40,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.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; @@ -77,8 +78,6 @@ import java.util.Set; import java.util.stream.Collectors; -import static org.apache.druid.indexing.seekablestream.supervisor.autoscaler.CostBasedAutoScaler.AUTOSCALER_TYPE_NAME; - /** * Endpoints for submitting and starting a {@link SupervisorSpec}, getting running supervisors, stopping supervisors, * and getting supervisor history. @@ -547,7 +546,7 @@ public Response terminateAll(@Context final HttpServletRequest req) @ResourceFilters(SupervisorResourceFilter.class) public Response simulateAutoscaling( @PathParam("id") String supervisorId, - @QueryParam("autoScalerStrategy") @DefaultValue(AUTOSCALER_TYPE_NAME) String autoScalerStrategy, + @QueryParam("autoScalerStrategy") @DefaultValue(CostBasedAutoScaler.AUTOSCALER_TYPE_NAME) String autoScalerStrategy, @QueryParam("taskCountMin") int taskCountMin, @QueryParam("taskCountMax") int taskCountMax, @QueryParam("maxProcessingRatePerTask") int maxProcessingRatePerTask, @@ -559,11 +558,11 @@ public Response simulateAutoscaling( @Context HttpServletRequest request ) { - if (!AUTOSCALER_TYPE_NAME.equals(autoScalerStrategy)) { + if (!CostBasedAutoScaler.AUTOSCALER_TYPE_NAME.equals(autoScalerStrategy)) { throw InvalidInput.exception( "Cannot simulate autoScalerStrategy[%s]. Only [%s] is supported.", autoScalerStrategy, - AUTOSCALER_TYPE_NAME + CostBasedAutoScaler.AUTOSCALER_TYPE_NAME ); } final CostBasedAutoScalerConfig autoScalerConfig =