You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by mx...@apache.org on 2023/05/24 18:05:33 UTC
[flink-kubernetes-operator] branch main updated: [FLINK-31684] Report autoscaling metrics before metric window is full (#606)
This is an automated email from the ASF dual-hosted git repository.
mxm pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-kubernetes-operator.git
The following commit(s) were added to refs/heads/main by this push:
new b11d6e18 [FLINK-31684] Report autoscaling metrics before metric window is full (#606)
b11d6e18 is described below
commit b11d6e1861ae1e7c9f6f0fdb4446b4c76420612e
Author: Maximilian Michels <mx...@apache.org>
AuthorDate: Wed May 24 20:05:26 2023 +0200
[FLINK-31684] Report autoscaling metrics before metric window is full (#606)
The metrics get reported only after the metric window is full. This is not helpful for observability after rescaling. We need to make sure that metrics are reported even when the metric window is not
yet full.
---
.../operator/autoscaler/JobAutoScalerImpl.java | 11 ++++++-----
.../operator/autoscaler/ScalingMetricCollector.java | 15 +++++++--------
.../autoscaler/metrics/CollectedMetricHistory.java | 10 ++++++----
.../MetricsCollectionAndEvaluationTest.java | 21 +++++++++++++++------
4 files changed, 34 insertions(+), 23 deletions(-)
diff --git a/flink-kubernetes-operator-autoscaler/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/JobAutoScalerImpl.java b/flink-kubernetes-operator-autoscaler/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/JobAutoScalerImpl.java
index d340b249..20571643 100644
--- a/flink-kubernetes-operator-autoscaler/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/JobAutoScalerImpl.java
+++ b/flink-kubernetes-operator-autoscaler/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/JobAutoScalerImpl.java
@@ -101,17 +101,18 @@ public class JobAutoScalerImpl implements JobAutoScaler {
metricsCollector.updateMetrics(
resource, autoScalerInfo, ctx.getFlinkService(), conf);
- if (collectedMetrics.getMetricHistory().isEmpty()) {
- autoScalerInfo.replaceInKubernetes(kubernetesClient);
- return false;
- }
-
LOG.debug("Evaluating scaling metrics for {}", collectedMetrics);
var evaluatedMetrics = evaluator.evaluate(conf, collectedMetrics);
LOG.debug("Scaling metrics evaluated: {}", evaluatedMetrics);
lastEvaluatedMetrics.put(resourceId, evaluatedMetrics);
flinkMetrics.registerScalingMetrics(() -> lastEvaluatedMetrics.get(resourceId));
+ if (!collectedMetrics.isFullyCollected()) {
+ // We have done an upfront evaluation, but we are not ready for scaling.
+ autoScalerInfo.replaceInKubernetes(kubernetesClient);
+ return false;
+ }
+
var specAdjusted =
scalingExecutor.scaleResource(resource, autoScalerInfo, conf, evaluatedMetrics);
if (specAdjusted) {
diff --git a/flink-kubernetes-operator-autoscaler/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/ScalingMetricCollector.java b/flink-kubernetes-operator-autoscaler/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/ScalingMetricCollector.java
index 18478a9b..0fb65242 100644
--- a/flink-kubernetes-operator-autoscaler/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/ScalingMetricCollector.java
+++ b/flink-kubernetes-operator-autoscaler/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/ScalingMetricCollector.java
@@ -128,17 +128,16 @@ public abstract class ScalingMetricCollector {
metricHistory.put(now, scalingMetrics);
autoscalerInfo.updateMetricHistory(metricHistory);
+ var collectedMetrics = new CollectedMetricHistory(topology, metricHistory);
+
var windowFullTime = metricCollectionStartTs.plus(metricWindowSize);
- if (now.isBefore(windowFullTime)) {
- // As long as we haven't had time to collect a full window,
- // collect metrics but do not return any metrics
- LOG.info(
- "Waiting until {} so the initial metric window is full before starting scaling",
- windowFullTime);
- return new CollectedMetricHistory(topology, Collections.emptySortedMap());
+ collectedMetrics.setFullyCollected(!now.isBefore(windowFullTime));
+
+ if (!collectedMetrics.isFullyCollected()) {
+ LOG.info("Metric window not full until {}", windowFullTime);
}
- return new CollectedMetricHistory(topology, metricHistory);
+ return collectedMetrics;
}
protected Duration getMetricWindowSize(Configuration conf) {
diff --git a/flink-kubernetes-operator-autoscaler/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/metrics/CollectedMetricHistory.java b/flink-kubernetes-operator-autoscaler/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/metrics/CollectedMetricHistory.java
index 33243ec0..43bbf47b 100644
--- a/flink-kubernetes-operator-autoscaler/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/metrics/CollectedMetricHistory.java
+++ b/flink-kubernetes-operator-autoscaler/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/metrics/CollectedMetricHistory.java
@@ -19,14 +19,16 @@ package org.apache.flink.kubernetes.operator.autoscaler.metrics;
import org.apache.flink.kubernetes.operator.autoscaler.topology.JobTopology;
-import lombok.Value;
+import lombok.Data;
+import lombok.Setter;
import java.time.Instant;
import java.util.SortedMap;
/** Topology and collected metric history. */
-@Value
+@Data
public class CollectedMetricHistory {
- JobTopology jobTopology;
- SortedMap<Instant, CollectedMetrics> metricHistory;
+ final JobTopology jobTopology;
+ final SortedMap<Instant, CollectedMetrics> metricHistory;
+ @Setter private boolean fullyCollected;
}
diff --git a/flink-kubernetes-operator-autoscaler/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/MetricsCollectionAndEvaluationTest.java b/flink-kubernetes-operator-autoscaler/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/MetricsCollectionAndEvaluationTest.java
index 7bef5609..c019d7aa 100644
--- a/flink-kubernetes-operator-autoscaler/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/MetricsCollectionAndEvaluationTest.java
+++ b/flink-kubernetes-operator-autoscaler/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/MetricsCollectionAndEvaluationTest.java
@@ -146,14 +146,16 @@ public class MetricsCollectionAndEvaluationTest {
clock = Clock.offset(clock, conf.get(AutoScalerOptions.STABILIZATION_INTERVAL));
metricsCollector.setClock(clock);
collectedMetrics = metricsCollector.updateMetrics(app, scalingInfo, service, conf);
- assertTrue(collectedMetrics.getMetricHistory().isEmpty());
+ assertEquals(1, collectedMetrics.getMetricHistory().size());
+ assertFalse(collectedMetrics.isFullyCollected());
// We haven't collected a full window yet
// => no metrics should be reported but metrics should still get collected.
clock = Clock.offset(clock, Duration.ofSeconds(1));
metricsCollector.setClock(clock);
collectedMetrics = metricsCollector.updateMetrics(app, scalingInfo, service, conf);
- assertTrue(collectedMetrics.getMetricHistory().isEmpty());
+ assertEquals(2, collectedMetrics.getMetricHistory().size());
+ assertFalse(collectedMetrics.isFullyCollected());
// Advance time to stabilization period + full window => metrics should be present
clock =
@@ -165,6 +167,7 @@ public class MetricsCollectionAndEvaluationTest {
metricsCollector.setClock(clock);
collectedMetrics = metricsCollector.updateMetrics(app, scalingInfo, service, conf);
assertEquals(3, collectedMetrics.getMetricHistory().size());
+ assertTrue(collectedMetrics.isFullyCollected());
// Test resetting the collector and make sure we can deserialize the scalingInfo correctly
metricsCollector = new TestingMetricsCollector(topology);
@@ -172,6 +175,7 @@ public class MetricsCollectionAndEvaluationTest {
setDefaultMetrics(metricsCollector);
collectedMetrics = metricsCollector.updateMetrics(app, scalingInfo, service, conf);
assertEquals(3, collectedMetrics.getMetricHistory().size());
+ assertTrue(collectedMetrics.isFullyCollected());
var evaluation = evaluator.evaluate(conf, collectedMetrics);
scalingExecutor.scaleResource(app, scalingInfo, conf, evaluation);
@@ -318,17 +322,20 @@ public class MetricsCollectionAndEvaluationTest {
// This call will lead to metric collection but we haven't reached the window size yet
// which will hold back metrics
metricsHistory = metricsCollector.updateMetrics(app, scalingInfo, service, conf);
- assertEquals(0, metricsHistory.getMetricHistory().size());
+ assertEquals(1, metricsHistory.getMetricHistory().size());
+ assertFalse(metricsHistory.isFullyCollected());
// Collect more values in window
metricsCollector.setClock(Clock.offset(clock, Duration.ofSeconds(1)));
metricsHistory = metricsCollector.updateMetrics(app, scalingInfo, service, conf);
- assertEquals(0, metricsHistory.getMetricHistory().size());
+ assertEquals(2, metricsHistory.getMetricHistory().size());
+ assertFalse(metricsHistory.isFullyCollected());
// Window size reached
metricsCollector.setClock(Clock.offset(clock, conf.get(AutoScalerOptions.METRICS_WINDOW)));
metricsHistory = metricsCollector.updateMetrics(app, scalingInfo, service, conf);
assertEquals(3, metricsHistory.getMetricHistory().size());
+ assertTrue(metricsHistory.isFullyCollected());
// Window size + 1 will invalidate the first metric
metricsCollector.setClock(
@@ -447,12 +454,14 @@ public class MetricsCollectionAndEvaluationTest {
metricsCollector.setClock(clock);
var collectedMetrics = metricsCollector.updateMetrics(app, scalingInfo, service, conf);
- assertTrue(collectedMetrics.getMetricHistory().isEmpty());
+ assertEquals(1, collectedMetrics.getMetricHistory().size());
+ assertFalse(collectedMetrics.isFullyCollected());
metricsCollector.setClock(Clock.offset(clock, Duration.ofSeconds(2)));
collectedMetrics = metricsCollector.updateMetrics(app, scalingInfo, service, conf);
- assertFalse(collectedMetrics.getMetricHistory().isEmpty());
+ assertEquals(2, collectedMetrics.getMetricHistory().size());
+ assertTrue(collectedMetrics.isFullyCollected());
return collectedMetrics;
}