sharath1709 commented on code in PR #686:
URL:
https://github.com/apache/flink-kubernetes-operator/pull/686#discussion_r2289296770
##########
flink-autoscaler/src/main/java/org/apache/flink/autoscaler/metrics/ScalingMetrics.java:
##########
@@ -78,6 +88,45 @@ public static void computeDataRateMetrics(
}
}
+ private static Optional<Double> getObservedTpr(
+ Map<FlinkMetric, AggregatedMetric> flinkMetrics,
+ Map<ScalingMetric, Double> scalingMetrics,
+ double numRecordsInPerSecond,
+ Configuration conf) {
+
+ // If there are no incoming records we return infinity to allow scale
down
+ if (numRecordsInPerSecond == 0) {
+ return Optional.of(Double.POSITIVE_INFINITY);
+ }
+
+ // We only measure observed tpr when we are catching up, that is when
the lag is beyond the
+ // configured observe threshold
+ boolean catchingUp =
+ scalingMetrics.getOrDefault(ScalingMetric.LAG, 0.)
+ >=
conf.get(AutoScalerOptions.OBSERVE_TRUE_PROCESSING_RATE_LAG_THRESHOLD)
+ .toSeconds()
+ * numRecordsInPerSecond;
+ if (!catchingUp) {
+ return Optional.empty();
+ }
+
+ double observedTpr =
+ computeObservedTprWithBackpressure(
+ numRecordsInPerSecond,
+
flinkMetrics.get(FlinkMetric.BACKPRESSURE_TIME_PER_SEC).getAvg());
+
+ return Double.isNaN(observedTpr) ? Optional.empty() :
Optional.of(observedTpr);
+ }
+
+ public static double computeObservedTprWithBackpressure(
Review Comment:
Let me clarify my statement. Catch-up Time is used to
[calculate](https://github.com/apache/flink-kubernetes-operator/blob/b72ad7f23658c29939682cffff77bf737fe7a1ba/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/utils/AutoScalerUtils.java#L45)
the target processing capacity. The true processing capacity is then
[calculated](https://github.com/apache/flink-kubernetes-operator/blob/b72ad7f23658c29939682cffff77bf737fe7a1ba/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingMetricEvaluator.java#L223)
based on either busy time or backpressure, whichever results in a lower value.
Next, the scale factor is
[computed](https://github.com/apache/flink-kubernetes-operator/blame/b72ad7f23658c29939682cffff77bf737fe7a1ba/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/JobVertexScaler.java#L186)
as:
double scaleFactor = targetCapacity / averageTrueProcessingRate;
If these values match, the parallelism for the vertex remains unchanged. In
effect, both the autoscaler and the user assume that the backlog will be
cleared within job.autoscaler.catch-up.duration (per the description: “The
target duration for fully processing any backlog after a scaling operation. Set
to 0 to disable backlog-based scaling.”).
However, in practice, the catch-up time almost always exceeds
job.autoscaler.catch-up.duration, which makes it difficult to reason about the
guarantees provided by the autoscaler.
--
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]
