gyfora commented on code in PR #686:
URL:
https://github.com/apache/flink-kubernetes-operator/pull/686#discussion_r2289950705
##########
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:
I agree that the name catch up duration is a bit incorrect, we can fix the
name or tweak the behaviour but I would like to continue this discussion on a
mailing list / JIRA instead. These old PRs are not the place for it :)
--
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]
