gyfora commented on code in PR #686:
URL:
https://github.com/apache/flink-kubernetes-operator/pull/686#discussion_r1368342398
##########
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);
- var stableTime =
jobUpdateTs.plus(conf.get(AutoScalerOptions.STABILIZATION_INTERVAL));
- if (now.isBefore(stableTime)) {
- // As long as we are stabilizing, collect no metrics at all
- LOG.info("Skipping metric collection during stabilization period
until {}", stableTime);
- return new CollectedMetricHistory(topology,
Collections.emptySortedMap());
- }
+ // Trim metrics outside the metric window from metrics history
+ metricHistory.headMap(now.minus(metricWindowSize)).clear();
Review Comment:
makes sense to simplify
--
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
