You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "morhidi (via GitHub)" <gi...@apache.org> on 2025/02/18 22:25:34 UTC
[PR] [FLINK-36409] Publish some autoscaler metrics during stabilisation period [flink-kubernetes-operator]
morhidi opened a new pull request, #945:
URL: https://github.com/apache/flink-kubernetes-operator/pull/945
## What is the purpose of the change
Currently autoscaler metrics although collected are not published during stabilization period. We report metrics after the stabilization period only. In practice this could result in larger gaps in metric charts during scale operations that makes it hard for end users to interpret. The metrics could appear to be broken, especially when multiple scale operations executed in a row, for example:
<img width="450" alt="image" src="https://github.com/user-attachments/assets/e533cfc1-b712-498a-b9d9-3b3a2b4a700a" />
This change mitigates this issue by shortening the gaps in reported metrics. The collected metrics won't be withhold during stabilization period either.
## Brief change log
- Removing the logic to report no metrics during stabilization
- Adjusted the logging to better understand the stabilization/metric window periods
- Update the timestamp format to contain millis (unit tests operate with millis)
```
2025-02-17 16:56:13,948 o.a.f.a.ScalingMetricCollector [INFO ] Stabilizing... until 1969-12-31 16:00:00.100. 1 samples collected
2025-02-17 16:56:13,952 o.a.f.a.ScalingMetricCollector [INFO ] Stabilizing... until 1969-12-31 16:00:00.100. 2 samples collected
2025-02-17 16:56:13,957 o.a.f.a.ScalingMetricCollector [INFO ] Metric window is not full until 1969-12-31 16:00:00.250. 3 samples collected
2025-02-17 16:56:13,958 o.a.f.a.ScalingMetricCollector [INFO ] Metric window is not full until 1969-12-31 16:00:00.250. 4 samples collected
2025-02-17 16:56:13,960 o.a.f.a.ScalingMetricCollector [INFO ] Metric window is now full. Dropped 3 samples before 1969-12-31 16:00:00.160, keeping 2.
```
## Verifying this change
Updated existing unit tests.
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): no
- The public API, i.e., is any changes to the `CustomResourceDescriptors`: no
- Core observer or reconciler logic that is regularly executed: no
## Documentation
- Does this pull request introduce a new feature? no
- If yes, how is the feature documented? not applicable
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36409] Publish some autoscaler metrics during stabilisation period [flink-kubernetes-operator]
Posted by "morhidi (via GitHub)" <gi...@apache.org>.
morhidi merged PR #945:
URL: https://github.com/apache/flink-kubernetes-operator/pull/945
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36409] Publish some autoscaler metrics during stabilisation period [flink-kubernetes-operator]
Posted by "morhidi (via GitHub)" <gi...@apache.org>.
morhidi commented on code in PR #945:
URL: https://github.com/apache/flink-kubernetes-operator/pull/945#discussion_r1962074510
##########
flink-autoscaler/src/main/java/org/apache/flink/autoscaler/utils/DateTimeUtils.java:
##########
@@ -26,7 +26,7 @@
public class DateTimeUtils {
private static final DateTimeFormatter DEFAULT_FORMATTER =
- DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
+ DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss.SSS");
Review Comment:
Unit tests use milliseconds accuracy, it's easier to follow the logic when running individually.
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36409] Publish some autoscaler metrics during stabilisation period [flink-kubernetes-operator]
Posted by "morhidi (via GitHub)" <gi...@apache.org>.
morhidi commented on PR #945:
URL: https://github.com/apache/flink-kubernetes-operator/pull/945#issuecomment-2669277034
> +1 for @mxm 's suggestion to enable this by a config flag but it should probably be disabled by default (to keep the current behaviour as the default)
> I'm not 100% sure this change will yield the desired outcome.
All right, I can close this PR
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36409] Publish some autoscaler metrics during stabilisation period [flink-kubernetes-operator]
Posted by "morhidi (via GitHub)" <gi...@apache.org>.
morhidi commented on PR #945:
URL: https://github.com/apache/flink-kubernetes-operator/pull/945#issuecomment-2671952374
Thanks for the review folks, I've pushed the requested changes
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36409] Publish some autoscaler metrics during stabilisation period [flink-kubernetes-operator]
Posted by "gyfora (via GitHub)" <gi...@apache.org>.
gyfora commented on PR #945:
URL: https://github.com/apache/flink-kubernetes-operator/pull/945#issuecomment-2669500389
I am not sure that’s what Max meant, because you can already reduce the stabilization period to 0 if you want. So we wouldn’t need a new option .
I thought he meant a flag to determine whether the stabilization metrics should be reported or not. Let’s clarify offline
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36409] Publish some autoscaler metrics during stabilisation period [flink-kubernetes-operator]
Posted by "gyfora (via GitHub)" <gi...@apache.org>.
gyfora commented on PR #945:
URL: https://github.com/apache/flink-kubernetes-operator/pull/945#issuecomment-2669307305
Why would we close it? I think the PR makes sense , but it would be good to be able to configure the behavior as Max suggested
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36409] Publish some autoscaler metrics during stabilisation period [flink-kubernetes-operator]
Posted by "gyfora (via GitHub)" <gi...@apache.org>.
gyfora commented on PR #945:
URL: https://github.com/apache/flink-kubernetes-operator/pull/945#issuecomment-2671594528
Please rebase on main for the docker fix
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36409] Publish some autoscaler metrics during stabilisation period [flink-kubernetes-operator]
Posted by "morhidi (via GitHub)" <gi...@apache.org>.
morhidi commented on PR #945:
URL: https://github.com/apache/flink-kubernetes-operator/pull/945#issuecomment-2669434319
> Why would we close it? I think the PR makes sense , but it would be good to be able to configure the behavior as Max suggested
> Perhaps we can add an option to allow collecting metrics in the stabilization phase?
> Why would we close it? I think the PR makes sense , but it would be good to be able to configure the behavior as Max suggested
> I'm not 100% sure this change will yield the desired outcome. There are some issues with collecting metrics in the stabilization phase, which is why we explicitly chose to not collect any that phase:
>
> Metrics are not available yet which will be evident in exceptions from the Rest API
> Metrics may be incomplete
> Metrics values will be skewed in the stabilization phase.
> After this change, there is no way to externally asses the source-of-truth metrics which will be used for evaluation. This makes debugging the autoscaling algorithm harder.
>
> Perhaps we can add an option to allow collecting metrics in the stabilization phase?
I guess Max is under the assumption that the current logic does not collect metrics during the stabilization period. This PR does not change that logic, so not sure what should be controlled by a flag. The only thing the PR does is that it reports those metrics. Can you clarify?
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36409] Publish some autoscaler metrics during stabilisation period [flink-kubernetes-operator]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #945:
URL: https://github.com/apache/flink-kubernetes-operator/pull/945#discussion_r1963310240
##########
flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingMetricCollector.java:
##########
@@ -149,19 +149,31 @@ public CollectedMetricHistory updateMetrics(
// Add scaling metrics to history if they were computed successfully
metricHistory.put(now, scalingMetrics);
- if (isStabilizing) {
- LOG.info("Stabilizing until {}", readable(stableTime));
- stateStore.storeCollectedMetrics(ctx, metricHistory);
- return new CollectedMetricHistory(topology, Collections.emptySortedMap(), jobRunningTs);
- }
-
var collectedMetrics = new CollectedMetricHistory(topology, metricHistory, jobRunningTs);
if (now.isBefore(windowFullTime)) {
- LOG.info("Metric window not full until {}", readable(windowFullTime));
+ if (isStabilizing) {
+ LOG.info(
+ "Stabilizing... until {}. {} samples collected",
Review Comment:
True, we return the metrics from the stabilization period once the stabilization period is over. They are used to calculate the observed true processing rate. When the window is full, we clear the metrics from the stabilization period and mark the window as fully collected.
I still think that returning this logging message ("XX samples collected") to the user during stabilization is confusing. It suggests that these metrics will be evaluated for scaling decisions, which they are not. We are merely measuring the processing capacity of the sources.
##########
flink-autoscaler/src/main/java/org/apache/flink/autoscaler/utils/DateTimeUtils.java:
##########
@@ -26,7 +26,7 @@
public class DateTimeUtils {
private static final DateTimeFormatter DEFAULT_FORMATTER =
- DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
+ DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss.SSS");
Review Comment:
AFAIK we don't use any milliseconds in the tests, but advance only by full seconds. I think this may have unexpected consequences for the storage and logging layer. If this is a test-only change, then it should go into test.
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36409] Publish some autoscaler metrics during stabilisation period [flink-kubernetes-operator]
Posted by "morhidi (via GitHub)" <gi...@apache.org>.
morhidi commented on code in PR #945:
URL: https://github.com/apache/flink-kubernetes-operator/pull/945#discussion_r1962073100
##########
flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingMetricCollector.java:
##########
@@ -149,19 +149,31 @@ public CollectedMetricHistory updateMetrics(
// Add scaling metrics to history if they were computed successfully
metricHistory.put(now, scalingMetrics);
- if (isStabilizing) {
- LOG.info("Stabilizing until {}", readable(stableTime));
- stateStore.storeCollectedMetrics(ctx, metricHistory);
- return new CollectedMetricHistory(topology, Collections.emptySortedMap(), jobRunningTs);
- }
-
var collectedMetrics = new CollectedMetricHistory(topology, metricHistory, jobRunningTs);
if (now.isBefore(windowFullTime)) {
- LOG.info("Metric window not full until {}", readable(windowFullTime));
+ if (isStabilizing) {
+ LOG.info(
+ "Stabilizing... until {}. {} samples collected",
Review Comment:
We just didn't display it to the user.
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36409] Publish some autoscaler metrics during stabilisation period [flink-kubernetes-operator]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on PR #945:
URL: https://github.com/apache/flink-kubernetes-operator/pull/945#issuecomment-2671259434
I'm ok with not having flag / config option. I do see that adding the changes under a feature flag is of little use. Plus, we have too many config options already. You will likely still have gaps in the metric processing because of the job restarts for which metrics aren't available, but it may help users to better understand what's happening with the cluster during the stabilization period.
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36409] Publish some autoscaler metrics during stabilisation period [flink-kubernetes-operator]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on PR #945:
URL: https://github.com/apache/flink-kubernetes-operator/pull/945#issuecomment-2671149679
> I guess Max is under the assumption that the current logic does not collect metrics during the stabilization period. We do collect samples, and once the stabilization is over we even evaluate them. This PR does not change that logic, so not sure what should be controlled by a flag. The only thing the PR does is that it reports those metrics. Can you clarify? I might missing something obvious from the current logic.
You're right, we already return metrics from the stabilization phase, but only to measure the observed true processing rate. In the original model, we only returned metrics once the metric window was full. I think that was more elegant, but the source metrics proved not reliable enough that we had to manually measure the processing capacity instead of always relying on the processing rate and busyness metrics of sources.
I might be a bit pedantic here, but I want to see the actual metrics used for evaluation reported as autoscaler metrics. Reporting metrics during stabilization removes that clarity. You can only observe what the assumptions of the autoscaler were, if you observed what is actually used for evaluation. That's why I suggested to put reporting autoscaler metrics during the stabilization period behind a flag.
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36409] Publish some autoscaler metrics during stabilisation period [flink-kubernetes-operator]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #945:
URL: https://github.com/apache/flink-kubernetes-operator/pull/945#discussion_r1961334586
##########
flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingMetricCollector.java:
##########
@@ -149,19 +149,31 @@ public CollectedMetricHistory updateMetrics(
// Add scaling metrics to history if they were computed successfully
metricHistory.put(now, scalingMetrics);
- if (isStabilizing) {
- LOG.info("Stabilizing until {}", readable(stableTime));
- stateStore.storeCollectedMetrics(ctx, metricHistory);
- return new CollectedMetricHistory(topology, Collections.emptySortedMap(), jobRunningTs);
- }
-
var collectedMetrics = new CollectedMetricHistory(topology, metricHistory, jobRunningTs);
if (now.isBefore(windowFullTime)) {
- LOG.info("Metric window not full until {}", readable(windowFullTime));
+ if (isStabilizing) {
+ LOG.info(
+ "Stabilizing... until {}. {} samples collected",
Review Comment:
NIT: Can we keep the existing output?
```suggestion
"Stabilizing until {}",
```
I don't think we need the number of samples here. This is just confusing because we are not supposed to collect any metrics for evaluation.
##########
flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingMetricCollector.java:
##########
@@ -149,19 +149,31 @@ public CollectedMetricHistory updateMetrics(
// Add scaling metrics to history if they were computed successfully
metricHistory.put(now, scalingMetrics);
- if (isStabilizing) {
- LOG.info("Stabilizing until {}", readable(stableTime));
- stateStore.storeCollectedMetrics(ctx, metricHistory);
- return new CollectedMetricHistory(topology, Collections.emptySortedMap(), jobRunningTs);
- }
-
var collectedMetrics = new CollectedMetricHistory(topology, metricHistory, jobRunningTs);
if (now.isBefore(windowFullTime)) {
- LOG.info("Metric window not full until {}", readable(windowFullTime));
+ if (isStabilizing) {
+ LOG.info(
+ "Stabilizing... until {}. {} samples collected",
+ readable(stableTime),
+ metricHistory.size());
+ } else {
+ LOG.info(
+ "Metric window is not full until {}. {} samples collected",
+ readable(windowFullTime),
+ metricHistory.size());
+ }
} else {
collectedMetrics.setFullyCollected(true);
// Trim metrics outside the metric window from metrics history
- metricHistory.headMap(now.minus(metricWindowSize)).clear();
+ var trimBefore = now.minus(metricWindowSize);
+ var head = metricHistory.headMap(trimBefore);
+ var dropped = head.size();
+ head.clear();
+ LOG.debug(
+ "Metric window is now full. Dropped {} samples before {}, keeping {}.",
+ dropped,
+ readable(trimBefore),
+ metricHistory.size());
Review Comment:
This change is unrelated to the issue which we're trying to fix here. A couple of changes to avoid confusion:
NIT
```suggestion
var trimBefore = now.minus(metricWindowSize);
int numDropped = removeMetricsBefore(trimBefore);
LOG.debug(
"Metric window is now full. Dropped {} samples before {}, keeping {}.",
numDropped,
readable(trimBefore),
metricHistory.size());
```
```java
private int removeMetricsBefore(Instant cutOffTimestamp) {
var toBeDropped = metricHistory.headMap(cutOffTimestamp);
var numDropped = toBeDropped.size();
toBeDropped.clear();
return numDropped;
}
```
##########
flink-autoscaler/src/main/java/org/apache/flink/autoscaler/utils/DateTimeUtils.java:
##########
@@ -26,7 +26,7 @@
public class DateTimeUtils {
private static final DateTimeFormatter DEFAULT_FORMATTER =
- DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
+ DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss.SSS");
Review Comment:
Why this change? I think seconds accuracy is sufficient and more easily readable.
##########
flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingMetricCollector.java:
##########
@@ -149,19 +149,31 @@ public CollectedMetricHistory updateMetrics(
// Add scaling metrics to history if they were computed successfully
metricHistory.put(now, scalingMetrics);
- if (isStabilizing) {
- LOG.info("Stabilizing until {}", readable(stableTime));
- stateStore.storeCollectedMetrics(ctx, metricHistory);
- return new CollectedMetricHistory(topology, Collections.emptySortedMap(), jobRunningTs);
- }
-
var collectedMetrics = new CollectedMetricHistory(topology, metricHistory, jobRunningTs);
if (now.isBefore(windowFullTime)) {
- LOG.info("Metric window not full until {}", readable(windowFullTime));
+ if (isStabilizing) {
+ LOG.info(
+ "Stabilizing... until {}. {} samples collected",
+ readable(stableTime),
+ metricHistory.size());
+ } else {
+ LOG.info(
+ "Metric window is not full until {}. {} samples collected",
Review Comment:
```suggestion
"Metric window is not full until {}. {} samples collected so far",
```
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36409] Publish some autoscaler metrics during stabilisation period [flink-kubernetes-operator]
Posted by "morhidi (via GitHub)" <gi...@apache.org>.
morhidi commented on code in PR #945:
URL: https://github.com/apache/flink-kubernetes-operator/pull/945#discussion_r1962071572
##########
flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingMetricCollector.java:
##########
@@ -149,19 +149,31 @@ public CollectedMetricHistory updateMetrics(
// Add scaling metrics to history if they were computed successfully
metricHistory.put(now, scalingMetrics);
- if (isStabilizing) {
- LOG.info("Stabilizing until {}", readable(stableTime));
- stateStore.storeCollectedMetrics(ctx, metricHistory);
- return new CollectedMetricHistory(topology, Collections.emptySortedMap(), jobRunningTs);
- }
-
var collectedMetrics = new CollectedMetricHistory(topology, metricHistory, jobRunningTs);
if (now.isBefore(windowFullTime)) {
- LOG.info("Metric window not full until {}", readable(windowFullTime));
+ if (isStabilizing) {
+ LOG.info(
+ "Stabilizing... until {}. {} samples collected",
Review Comment:
Correct me if I'm wrong but samples are collected during stabilization period and used in evaluated metrics after exiting the stabilization period. We only drop the history when the window is full
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36409] Publish some autoscaler metrics during stabilisation period [flink-kubernetes-operator]
Posted by "gyfora (via GitHub)" <gi...@apache.org>.
gyfora commented on PR #945:
URL: https://github.com/apache/flink-kubernetes-operator/pull/945#issuecomment-2668549767
+1 for @mxm 's suggestion to enable this by a config flag but it should probably be disabled by default (to keep the current behaviour as the default)
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org