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/03/24 14:22:35 UTC
[flink-kubernetes-operator] branch main updated: [FLINK-30575] Limit output ratio in case of near zero values (#554)
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 10faab31 [FLINK-30575] Limit output ratio in case of near zero values (#554)
10faab31 is described below
commit 10faab31766b0f7ee3dc558a226503cd3baf27da
Author: Maximilian Michels <mx...@apache.org>
AuthorDate: Fri Mar 24 15:22:19 2023 +0100
[FLINK-30575] Limit output ratio in case of near zero values (#554)
When the rates drop to zero, we set near zero values to continue to use the same
scaling logic. There is a gap in this logic when only the input rate drops to
zero but the output rate is still measured to be greater than zero. This is
because we do not pass on the output rates directly but the output / input ratio
which becomes very large in this special scenario.
In this change we cap the output rate in case we use near zero input rates. This
has been tested to effectively prevent scale up spikes in pipelines which only
occasionally receive data.
---
.../autoscaler/metrics/ScalingMetrics.java | 28 +++++++++++------
.../autoscaler/metrics/ScalingMetricsTest.java | 35 +++++++++++++++++-----
2 files changed, 47 insertions(+), 16 deletions(-)
diff --git a/flink-kubernetes-operator-autoscaler/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/metrics/ScalingMetrics.java b/flink-kubernetes-operator-autoscaler/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/metrics/ScalingMetrics.java
index bf508c71..2329a50d 100644
--- a/flink-kubernetes-operator-autoscaler/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/metrics/ScalingMetrics.java
+++ b/flink-kubernetes-operator-autoscaler/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/metrics/ScalingMetrics.java
@@ -77,8 +77,6 @@ public class ScalingMetrics {
double numRecordsInPerSecond =
getNumRecordsInPerSecond(flinkMetrics, jobVertexID, isSource);
- double outputPerSecond =
- getNumRecordsOutPerSecond(flinkMetrics, jobVertexID, isSource, isSink);
if (isSource) {
double sourceDataRate = Math.max(0, numRecordsInPerSecond + lagGrowthRate);
@@ -99,11 +97,15 @@ public class ScalingMetrics {
}
if (!isSink) {
- if (!Double.isNaN(outputPerSecond)) {
+ double numRecordsOutPerSecond =
+ getNumRecordsOutPerSecond(
+ flinkMetrics, jobVertexID, isSource, numRecordsInPerSecond);
+ if (!Double.isNaN(numRecordsOutPerSecond)) {
scalingMetrics.put(
- ScalingMetric.OUTPUT_RATIO, outputPerSecond / numRecordsInPerSecond);
+ ScalingMetric.OUTPUT_RATIO, numRecordsOutPerSecond / numRecordsInPerSecond);
scalingMetrics.put(
- ScalingMetric.TRUE_OUTPUT_RATE, busyTimeMultiplier * outputPerSecond);
+ ScalingMetric.TRUE_OUTPUT_RATE,
+ busyTimeMultiplier * numRecordsOutPerSecond);
} else {
LOG.error(
"Cannot compute processing and input rate without numRecordsOutPerSecond");
@@ -163,7 +165,13 @@ public class ScalingMetrics {
Map<FlinkMetric, AggregatedMetric> flinkMetrics,
JobVertexID jobVertexID,
boolean isSource,
- boolean isSink) {
+ double numRecordsInPerSecond) {
+ if (isEffectivelyZero(numRecordsInPerSecond)) {
+ // If the input rate is zero, we also need to flatten the output rate.
+ // Otherwise, the OUTPUT_RATIO would be outrageously large, leading to
+ // a rapid scale up.
+ return EFFECTIVELY_ZERO;
+ }
AggregatedMetric numRecordsOutPerSecond =
flinkMetrics.get(FlinkMetric.NUM_RECORDS_OUT_PER_SEC);
if (numRecordsOutPerSecond == null) {
@@ -172,9 +180,7 @@ public class ScalingMetrics {
flinkMetrics.get(FlinkMetric.SOURCE_TASK_NUM_RECORDS_OUT_PER_SEC);
}
if (numRecordsOutPerSecond == null) {
- if (!isSink) {
- LOG.warn("Received null output rate for {}. Returning NaN.", jobVertexID);
- }
+ LOG.warn("Received null output rate for {}. Returning NaN.", jobVertexID);
return Double.NaN;
}
}
@@ -189,6 +195,10 @@ public class ScalingMetrics {
return value;
}
+ private static boolean isEffectivelyZero(double value) {
+ return value <= EFFECTIVELY_ZERO;
+ }
+
private static void excludeVertexFromScaling(Configuration conf, JobVertexID jobVertexId) {
Set<String> excludedIds = new HashSet<>(conf.get(AutoScalerOptions.VERTEX_EXCLUDE_IDS));
excludedIds.add(jobVertexId.toHexString());
diff --git a/flink-kubernetes-operator-autoscaler/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/metrics/ScalingMetricsTest.java b/flink-kubernetes-operator-autoscaler/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/metrics/ScalingMetricsTest.java
index 62bc5a51..4574bac0 100644
--- a/flink-kubernetes-operator-autoscaler/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/metrics/ScalingMetricsTest.java
+++ b/flink-kubernetes-operator-autoscaler/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/metrics/ScalingMetricsTest.java
@@ -223,7 +223,8 @@ public class ScalingMetricsTest {
@Test
public void testZeroValuesForBusyness() {
double dataRate = 10;
- Map<ScalingMetric, Double> scalingMetrics = testZeroValuesForRatesOrBusyness(dataRate, 0);
+ Map<ScalingMetric, Double> scalingMetrics =
+ testZeroValuesForRatesOrBusyness(dataRate, dataRate, 0);
assertEquals(
Map.of(
ScalingMetric.TRUE_PROCESSING_RATE,
@@ -243,7 +244,7 @@ public class ScalingMetricsTest {
public void testZeroValuesForRates() {
double busyMillisecondPerSec = 100;
Map<ScalingMetric, Double> scalingMetrics =
- testZeroValuesForRatesOrBusyness(0, busyMillisecondPerSec);
+ testZeroValuesForRatesOrBusyness(0, 0, busyMillisecondPerSec);
assertEquals(
Map.of(
ScalingMetric.TRUE_PROCESSING_RATE,
@@ -260,8 +261,27 @@ public class ScalingMetricsTest {
}
@Test
- public void testZeroValuesForRateAndBusyness() {
- Map<ScalingMetric, Double> scalingMetrics = testZeroValuesForRatesOrBusyness(0, 0);
+ public void testZeroProcessingRateOnly() {
+ Map<ScalingMetric, Double> scalingMetrics = testZeroValuesForRatesOrBusyness(0, 1, 100);
+ assertEquals(
+ Map.of(
+ // Output ratio should be kept to 1
+ ScalingMetric.OUTPUT_RATIO,
+ 1.,
+ ScalingMetric.TRUE_PROCESSING_RATE,
+ ScalingMetrics.EFFECTIVELY_ZERO * 10,
+ ScalingMetric.TRUE_OUTPUT_RATE,
+ ScalingMetrics.EFFECTIVELY_ZERO * 10,
+ ScalingMetric.SOURCE_DATA_RATE,
+ ScalingMetrics.EFFECTIVELY_ZERO,
+ ScalingMetric.CURRENT_PROCESSING_RATE,
+ ScalingMetrics.EFFECTIVELY_ZERO),
+ scalingMetrics);
+ }
+
+ @Test
+ public void testZeroValuesForRatesAndBusyness() {
+ Map<ScalingMetric, Double> scalingMetrics = testZeroValuesForRatesOrBusyness(0, 0, 0);
assertEquals(
Map.of(
ScalingMetric.TRUE_PROCESSING_RATE,
@@ -278,7 +298,7 @@ public class ScalingMetricsTest {
}
private static Map<ScalingMetric, Double> testZeroValuesForRatesOrBusyness(
- double rate, double busyness) {
+ double processingRate, double outputRate, double busyness) {
var source = new JobVertexID();
var op = new JobVertexID();
var sink = new JobVertexID();
@@ -296,9 +316,10 @@ public class ScalingMetricsTest {
FlinkMetric.BUSY_TIME_PER_SEC,
new AggregatedMetric("", Double.NaN, busyness, Double.NaN, Double.NaN),
FlinkMetric.SOURCE_TASK_NUM_RECORDS_OUT_PER_SEC,
- new AggregatedMetric("", Double.NaN, Double.NaN, Double.NaN, rate),
+ new AggregatedMetric(
+ "", Double.NaN, Double.NaN, Double.NaN, processingRate),
FlinkMetric.NUM_RECORDS_OUT_PER_SEC,
- new AggregatedMetric("", Double.NaN, Double.NaN, Double.NaN, rate)),
+ new AggregatedMetric("", Double.NaN, Double.NaN, Double.NaN, outputRate)),
scalingMetrics,
topology,
0.,