mxm commented on code in PR #686:
URL:
https://github.com/apache/flink-kubernetes-operator/pull/686#discussion_r1365595263
##########
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:
```suggestion
metricHistory.headMap(jobUpdateTs).clear();
```
This should fix it.
##########
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:
We could also drop this line entirely and handle metric cleanup in the
cleanup block line 102.
--
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
