gyfora commented on code in PR #774:
URL:
https://github.com/apache/flink-kubernetes-operator/pull/774#discussion_r1489408092
##########
flink-autoscaler/src/main/java/org/apache/flink/autoscaler/metrics/ScalingMetrics.java:
##########
@@ -43,49 +44,61 @@ public static void computeLoadMetrics(
JobVertexID jobVertexID,
Map<FlinkMetric, AggregatedMetric> flinkMetrics,
Map<ScalingMetric, Double> scalingMetrics,
+ IOMetrics ioMetrics,
Configuration conf) {
- double busyTimeMsPerSecond = getBusyTimeMsPerSecond(flinkMetrics,
conf, jobVertexID);
- scalingMetrics.put(ScalingMetric.LOAD, busyTimeMsPerSecond / 1000);
+ scalingMetrics.put(
+ ScalingMetric.LOAD,
+ getBusyTimeMsPerSecond(flinkMetrics, conf, jobVertexID) /
1000.);
+ scalingMetrics.put(ScalingMetric.ACCUMULATED_BUSY_TIME,
ioMetrics.getAccumulatedBusyTime());
+ }
+
+ private static double getBusyTimeMsPerSecond(
+ Map<FlinkMetric, AggregatedMetric> flinkMetrics,
+ Configuration conf,
+ JobVertexID jobVertexId) {
+ var busyTimeAggregator =
conf.get(AutoScalerOptions.BUSY_TIME_AGGREGATOR);
+ var busyTimeMsPerSecond =
+
busyTimeAggregator.get(flinkMetrics.get(FlinkMetric.BUSY_TIME_PER_SEC));
+ if (!Double.isFinite(busyTimeMsPerSecond)) {
+ if (AutoScalerUtils.excludeVertexFromScaling(conf, jobVertexId)) {
+ // We only want to log this once
+ LOG.warn(
+ "No busyTimeMsPerSecond metric available for {}. No
scaling will be performed for this vertex.",
+ jobVertexId);
+ }
+ return Double.NaN;
+ }
+ return Math.max(0, busyTimeMsPerSecond);
}
public static void computeDataRateMetrics(
JobVertexID jobVertexID,
Map<FlinkMetric, AggregatedMetric> flinkMetrics,
Map<ScalingMetric, Double> scalingMetrics,
JobTopology topology,
- double lagGrowthRate,
Configuration conf,
Supplier<Double> observedTprAvg) {
var isSource = topology.isSource(jobVertexID);
+ var ioMetrics = topology.get(jobVertexID).getIoMetrics();
double numRecordsInPerSecond =
- getNumRecordsInPerSecond(flinkMetrics, jobVertexID, isSource);
+ getNumRecordsIn(flinkMetrics, ioMetrics, jobVertexID,
isSource, true);
Review Comment:
will do
--
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: issues-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
