FrankChen021 commented on code in PR #20708:
URL: https://github.com/apache/druid/pull/20708#discussion_r4377189690
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/CostBasedAutoScaler.java:
##########
@@ -476,14 +493,29 @@ static double extractPollIdleRatio(Map<String,
Map<String, Object>> taskStats)
return -1.;
}
+ // Use running average metric to smoothen out spikes
+ return extractAutoscalerMetric(
+ taskStats,
+ SeekableStreamIndexTaskRunner.POLL_IDLE_RATIO_RUNNING_AVG
+ );
+ }
+
+ /**
+ * Gets the value of the given metric averaged across all tasks that have
reported the metric.
+ */
+ private static double extractAutoscalerMetric(
+ Map<String, Map<String, Object>> taskStats,
+ String metricName
+ )
+ {
double sum = 0;
int count = 0;
for (Map<String, Object> groupMetrics : taskStats.values()) {
for (Object taskMetric : groupMetrics.values()) {
if (taskMetric instanceof Map) {
Object autoScalerMetricsMap = ((Map<?, ?>)
taskMetric).get(SeekableStreamIndexTaskRunner.AUTOSCALER_METRICS_KEY);
if (autoScalerMetricsMap instanceof Map) {
- Object pollIdleRatioAvg = ((Map<?, ?>)
autoScalerMetricsMap).get(SeekableStreamIndexTaskRunner.POLL_IDLE_RATIO_KEY);
+ Object pollIdleRatioAvg = ((Map<?, ?>)
autoScalerMetricsMap).get(metricName);
Review Comment:
The all-old fallback is now covered, but the mixed-version case remains
incomplete: if any new task has a valid running-average value, the helper
averages only new-key tasks and drops old tasks; poll-idle ratio has the same
whole-map fallback. Please select the new or legacy value per task and add
mixed-version coverage.
<!-- mergelens:review -->
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/CostBasedAutoScaler.java:
##########
@@ -463,6 +463,29 @@ static int[] computeValidTaskCounts(
return result.toIntArray();
}
+ /**
+ * Extracts the average processing rate (records per second) from the task
stats.
+ *
+ * @param taskStats the stats map from supervisor.getStats()
+ */
+ static double extractAvgProcessingRate(Map<String, Map<String, Object>>
taskStats)
+ {
+ if (taskStats == null || taskStats.isEmpty()) {
+ return -1.;
+ }
+
+ // First try using the running average if available
+ final double runningAverage = extractAutoscalerMetric(
+ taskStats,
+ SeekableStreamIndexTaskRunner.PROCESSING_RATE_RUNNING_AVG
+ );
+ if (runningAverage >= 0) {
Review Comment:
[P2] Merge legacy metrics per task during rolling upgrades
**Finding:** The fallback is selected only after aggregating the entire
taskStats map. If mixed-version tasks include one valid running-average metric
and one older task exposing only movingAverages or pollIdleRatio, the older
task is discarded instead of contributing its legacy value. This distorts
autoscaling metrics during rolling upgrades; the same issue exists for
poll-idle ratio.
**Suggestion:** Choose the running-average key or its legacy counterpart per
task before averaging, and add mixed-version tests for both metrics.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
