gyfora commented on code in PR #686:
URL:
https://github.com/apache/flink-kubernetes-operator/pull/686#discussion_r1368341787
##########
flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingMetricCollector.java:
##########
@@ -97,32 +99,30 @@ public CollectedMetricHistory updateMetrics(
}
});
- // The timestamp of the first metric observation marks the start
- // If we haven't collected any metrics, we are starting now
- var metricCollectionStartTs = metricHistory.isEmpty() ? now :
metricHistory.firstKey();
-
var jobDetailsInfo =
getJobDetailsInfo(ctx,
conf.get(AutoScalerOptions.FLINK_CLIENT_TIMEOUT));
var jobUpdateTs = getJobUpdateTs(jobDetailsInfo);
- if (jobUpdateTs.isAfter(metricCollectionStartTs)) {
+ // We detect job change compared to our collected metrics by checking
against the earliest
+ // metric timestamp
+ if (!metricHistory.isEmpty() &&
jobUpdateTs.isAfter(metricHistory.firstKey())) {
LOG.info("Job updated at {}. Clearing metrics.", jobUpdateTs);
stateStore.removeEvaluatedMetrics(ctx);
cleanup(ctx.getJobKey());
metricHistory.clear();
- metricCollectionStartTs = now;
}
var topology = getJobTopology(ctx, stateStore, jobDetailsInfo);
+ var stableTime =
jobUpdateTs.plus(conf.get(AutoScalerOptions.STABILIZATION_INTERVAL));
- // Trim metrics outside the metric window from metrics history
+ // Calculate timestamp when the metric windows is full
var metricWindowSize = getMetricWindowSize(conf);
- metricHistory.headMap(now.minus(metricWindowSize)).clear();
+ var metricsAfterStable = metricHistory.tailMap(stableTime);
+ var windowFullTime =
+ metricsAfterStable.isEmpty()
+ ? now.plus(metricWindowSize)
+ : metricsAfterStable.firstKey().plus(metricWindowSize);
Review Comment:
good point
--
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
