You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by gy...@apache.org on 2023/06/25 14:15:07 UTC
[flink-kubernetes-operator] 04/07: [FLINK-32057] Improve job update / restart tracking in autoscaler
This is an automated email from the ASF dual-hosted git repository.
gyfora pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-kubernetes-operator.git
commit 0e78af24a2563ac0f7165f729b650d52f022335b
Author: Gyula Fora <g_...@apple.com>
AuthorDate: Wed Jun 14 22:15:06 2023 +0200
[FLINK-32057] Improve job update / restart tracking in autoscaler
---
.../operator/autoscaler/JobAutoScalerImpl.java | 12 ++-
.../autoscaler/RestApiMetricsCollector.java | 3 +-
.../operator/autoscaler/ScalingExecutor.java | 30 ------
.../autoscaler/ScalingMetricCollector.java | 118 ++++++++++-----------
.../autoscaler/BacklogBasedScalingTest.java | 21 ++--
.../operator/autoscaler/JobAutoScalerImplTest.java | 1 +
.../MetricsCollectionAndEvaluationTest.java | 11 +-
.../autoscaler/RecommendedParallelismTest.java | 5 +-
.../operator/autoscaler/ScalingExecutorTest.java | 47 --------
.../autoscaler/ScalingMetricCollectorTest.java | 35 ++++--
.../autoscaler/TestingMetricsCollector.java | 11 +-
.../operator/service/AbstractFlinkService.java | 16 +++
.../kubernetes/operator/service/FlinkService.java | 3 +
.../kubernetes/operator/TestingFlinkService.java | 33 ++----
14 files changed, 155 insertions(+), 191 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 44c361d7..a1879e9e 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
@@ -20,6 +20,7 @@ package org.apache.flink.kubernetes.operator.autoscaler;
import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.api.common.JobStatus;
import org.apache.flink.kubernetes.operator.api.AbstractFlinkResource;
+import org.apache.flink.kubernetes.operator.api.lifecycle.ResourceLifecycleState;
import org.apache.flink.kubernetes.operator.autoscaler.metrics.EvaluatedScalingMetric;
import org.apache.flink.kubernetes.operator.autoscaler.metrics.ScalingMetric;
import org.apache.flink.kubernetes.operator.controller.FlinkResourceContext;
@@ -97,7 +98,9 @@ public class JobAutoScalerImpl implements JobAutoScaler {
// Initialize metrics only if autoscaler is enabled
- if (!resource.getStatus().getJobStatus().getState().equals(JobStatus.RUNNING.name())) {
+ var status = resource.getStatus();
+ if (status.getLifecycleState() != ResourceLifecycleState.STABLE
+ || !status.getJobStatus().getState().equals(JobStatus.RUNNING.name())) {
LOG.info("Job autoscaler is waiting for RUNNING job state");
lastEvaluatedMetrics.remove(resourceId);
return false;
@@ -114,10 +117,10 @@ public class JobAutoScalerImpl implements JobAutoScaler {
return false;
}
- LOG.debug("Evaluating scaling metrics for {}", collectedMetrics);
+ LOG.debug("Collected metrics: {}", collectedMetrics);
evaluatedMetrics = evaluator.evaluate(conf, collectedMetrics);
+ LOG.debug("Evaluated metrics: {}", evaluatedMetrics);
initRecommendedParallelism(evaluatedMetrics);
- LOG.debug("Scaling metrics evaluated: {}", evaluatedMetrics);
if (!collectedMetrics.isFullyCollected()) {
// We have done an upfront evaluation, but we are not ready for scaling.
@@ -128,11 +131,13 @@ public class JobAutoScalerImpl implements JobAutoScaler {
var specAdjusted =
scalingExecutor.scaleResource(resource, autoScalerInfo, conf, evaluatedMetrics);
+
if (specAdjusted) {
flinkMetrics.numScalings.inc();
} else {
flinkMetrics.numBalanced.inc();
}
+
autoScalerInfo.replaceInKubernetes(kubernetesClient);
return specAdjusted;
} catch (Throwable e) {
@@ -147,7 +152,6 @@ public class JobAutoScalerImpl implements JobAutoScaler {
return false;
} finally {
if (evaluatedMetrics != null) {
- LOG.debug("Storing evaluated metrics {}", evaluatedMetrics);
lastEvaluatedMetrics.put(resourceId, evaluatedMetrics);
flinkMetrics.registerScalingMetrics(() -> lastEvaluatedMetrics.get(resourceId));
}
diff --git a/flink-kubernetes-operator-autoscaler/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/RestApiMetricsCollector.java b/flink-kubernetes-operator-autoscaler/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/RestApiMetricsCollector.java
index 68924e94..158bb5a2 100644
--- a/flink-kubernetes-operator-autoscaler/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/RestApiMetricsCollector.java
+++ b/flink-kubernetes-operator-autoscaler/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/RestApiMetricsCollector.java
@@ -18,7 +18,6 @@
package org.apache.flink.kubernetes.operator.autoscaler;
import org.apache.flink.api.common.JobID;
-import org.apache.flink.client.program.rest.RestClusterClient;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.kubernetes.operator.api.AbstractFlinkResource;
import org.apache.flink.kubernetes.operator.autoscaler.metrics.FlinkMetric;
@@ -83,7 +82,7 @@ public class RestApiMetricsCollector extends ScalingMetricCollector {
.next()
.resolveFromString(StringUtils.join(metrics.keySet(), ","));
- try (var restClient = (RestClusterClient<String>) flinkService.getClusterClient(conf)) {
+ try (var restClient = flinkService.getClusterClient(conf)) {
var responseBody =
restClient
diff --git a/flink-kubernetes-operator-autoscaler/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/ScalingExecutor.java b/flink-kubernetes-operator-autoscaler/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/ScalingExecutor.java
index 03267797..b917a821 100644
--- a/flink-kubernetes-operator-autoscaler/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/ScalingExecutor.java
+++ b/flink-kubernetes-operator-autoscaler/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/ScalingExecutor.java
@@ -18,7 +18,6 @@
package org.apache.flink.kubernetes.operator.autoscaler;
import org.apache.flink.annotation.VisibleForTesting;
-import org.apache.flink.api.common.JobStatus;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.kubernetes.operator.api.AbstractFlinkResource;
import org.apache.flink.kubernetes.operator.autoscaler.config.AutoScalerOptions;
@@ -43,7 +42,6 @@ import java.util.SortedMap;
import static org.apache.flink.configuration.PipelineOptions.PARALLELISM_OVERRIDES;
import static org.apache.flink.kubernetes.operator.autoscaler.config.AutoScalerOptions.SCALING_ENABLED;
-import static org.apache.flink.kubernetes.operator.autoscaler.config.AutoScalerOptions.STABILIZATION_INTERVAL;
import static org.apache.flink.kubernetes.operator.autoscaler.metrics.ScalingMetric.EXPECTED_PROCESSING_RATE;
import static org.apache.flink.kubernetes.operator.autoscaler.metrics.ScalingMetric.SCALE_DOWN_RATE_THRESHOLD;
import static org.apache.flink.kubernetes.operator.autoscaler.metrics.ScalingMetric.SCALE_UP_RATE_THRESHOLD;
@@ -83,10 +81,6 @@ public class ScalingExecutor {
Configuration conf,
Map<JobVertexID, Map<ScalingMetric, EvaluatedScalingMetric>> evaluatedMetrics) {
- if (!stabilizationPeriodPassed(resource, conf)) {
- return false;
- }
-
var now = Instant.now();
var scalingHistory = scalingInformation.getScalingHistory(now, conf);
var scalingSummaries =
@@ -165,30 +159,6 @@ public class ScalingExecutor {
return sb.toString();
}
- private boolean stabilizationPeriodPassed(
- AbstractFlinkResource<?, ?> resource, Configuration conf) {
- var jobStatus = resource.getStatus().getJobStatus();
-
- if (!JobStatus.RUNNING.name().equals(jobStatus.getState())) {
- // Never consider a non-running job stable
- return false;
- }
-
- var startTs =
- Instant.ofEpochMilli(
- // Use the update time which will reflect the latest job state update
- // Do not use the start time because it doesn't tell when the job went to
- // RUNNING
- Long.parseLong(jobStatus.getUpdateTime()));
- var stableTime = startTs.plus(conf.get(STABILIZATION_INTERVAL));
-
- if (stableTime.isAfter(clock.instant())) {
- LOG.info("Waiting until {} to stabilize before new scale operation.", stableTime);
- return false;
- }
- return true;
- }
-
protected static boolean allVerticesWithinUtilizationTarget(
Map<JobVertexID, Map<ScalingMetric, EvaluatedScalingMetric>> evaluatedMetrics,
Map<JobVertexID, ScalingSummary> scalingSummaries) {
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 99c0c0da..4d75b528 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
@@ -84,6 +84,8 @@ public abstract class ScalingMetricCollector {
throws Exception {
var resourceID = ResourceID.fromResource(cr);
+ var jobId = JobID.fromHexString(cr.getStatus().getJobStatus().getJobId());
+
var now = clock.instant();
var metricHistory =
@@ -92,15 +94,18 @@ public abstract class ScalingMetricCollector {
// 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 jobUpdateTs = getJobUpdateTs(cr);
+
+ var jobDetailsInfo = flinkService.getJobDetailsInfo(jobId, conf);
+ var jobUpdateTs = getJobUpdateTs(jobDetailsInfo);
if (jobUpdateTs.isAfter(metricCollectionStartTs)) {
- LOG.info("Job updated. Clearing metrics.");
+ LOG.info("Job updated at {}. Clearing metrics.", jobUpdateTs);
autoscalerInfo.clearMetricHistory();
cleanup(cr);
metricHistory.clear();
metricCollectionStartTs = now;
}
- var topology = getJobTopology(flinkService, cr, conf, autoscalerInfo);
+ var topology =
+ getJobTopology(flinkService, resourceID, conf, autoscalerInfo, jobDetailsInfo);
// Trim metrics outside the metric window from metrics history
var metricWindowSize = getMetricWindowSize(conf);
@@ -144,76 +149,71 @@ public abstract class ScalingMetricCollector {
return conf.get(AutoScalerOptions.METRICS_WINDOW);
}
- private static Instant getJobUpdateTs(AbstractFlinkResource<?, ?> cr) {
- return Instant.ofEpochMilli(Long.parseLong(cr.getStatus().getJobStatus().getUpdateTime()));
+ @VisibleForTesting
+ protected Instant getJobUpdateTs(JobDetailsInfo jobDetailsInfo) {
+ return Instant.ofEpochMilli(
+ jobDetailsInfo.getTimestamps().values().stream().max(Long::compare).get());
}
protected JobTopology getJobTopology(
FlinkService flinkService,
- AbstractFlinkResource<?, ?> cr,
+ ResourceID resourceID,
Configuration conf,
- AutoScalerInfo scalerInfo)
+ AutoScalerInfo scalerInfo,
+ JobDetailsInfo jobDetailsInfo)
throws Exception {
- try (var restClient = (RestClusterClient<String>) flinkService.getClusterClient(conf)) {
- var jobId = JobID.fromHexString(cr.getStatus().getJobStatus().getJobId());
- var topology =
- topologies.computeIfAbsent(
- ResourceID.fromResource(cr),
- r -> {
- var t = queryJobTopology(restClient, jobId);
- scalerInfo.updateVertexList(
- t.getVerticesInTopologicalOrder(), clock.instant(), conf);
- return t;
- });
- updateKafkaSourceMaxParallelisms(restClient, jobId, topology);
- return topology;
- }
+ var topology =
+ topologies.computeIfAbsent(
+ resourceID,
+ r -> {
+ var t = getJobTopology(jobDetailsInfo);
+ scalerInfo.updateVertexList(
+ t.getVerticesInTopologicalOrder(), clock.instant(), conf);
+ return t;
+ });
+ updateKafkaSourceMaxParallelisms(flinkService, conf, jobDetailsInfo.getJobId(), topology);
+ return topology;
}
@VisibleForTesting
- protected JobTopology queryJobTopology(RestClusterClient<String> restClient, JobID jobId) {
- try {
- var jobDetailsInfo = restClient.getJobDetails(jobId).get();
-
- Map<JobVertexID, Integer> maxParallelismMap =
- jobDetailsInfo.getJobVertexInfos().stream()
- .collect(
- Collectors.toMap(
- JobDetailsInfo.JobVertexDetailsInfo::getJobVertexID,
- JobDetailsInfo.JobVertexDetailsInfo
- ::getMaxParallelism));
-
- // Flink 1.17 introduced a strange behaviour where the JobDetailsInfo#getJsonPlan()
- // doesn't return the jsonPlan correctly but it wraps with RawJson{json='plan'}
- var rawPlan = jobDetailsInfo.getJsonPlan();
- var json =
- rawPlan.substring("RawJson{json='".length(), rawPlan.length() - "'}".length());
-
- return JobTopology.fromJsonPlan(json, maxParallelismMap);
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
+ @SneakyThrows
+ protected JobTopology getJobTopology(JobDetailsInfo jobDetailsInfo) {
+ Map<JobVertexID, Integer> maxParallelismMap =
+ jobDetailsInfo.getJobVertexInfos().stream()
+ .collect(
+ Collectors.toMap(
+ JobDetailsInfo.JobVertexDetailsInfo::getJobVertexID,
+ JobDetailsInfo.JobVertexDetailsInfo::getMaxParallelism));
+
+ // Flink 1.17 introduced a strange behaviour where the JobDetailsInfo#getJsonPlan()
+ // doesn't return the jsonPlan correctly but it wraps with RawJson{json='plan'}
+ var rawPlan = jobDetailsInfo.getJsonPlan();
+ var json = rawPlan.substring("RawJson{json='".length(), rawPlan.length() - "'}".length());
+
+ return JobTopology.fromJsonPlan(json, maxParallelismMap);
}
private void updateKafkaSourceMaxParallelisms(
- RestClusterClient<String> restClient, JobID jobId, JobTopology topology)
+ FlinkService flinkService, Configuration conf, JobID jobId, JobTopology topology)
throws Exception {
- var partitionRegex = Pattern.compile("^.*\\.partition\\.\\d+\\.currentOffset$");
- for (Map.Entry<JobVertexID, Set<JobVertexID>> entry : topology.getInputs().entrySet()) {
- if (entry.getValue().isEmpty()) {
- var sourceVertex = entry.getKey();
- var numPartitions =
- queryAggregatedMetricNames(restClient, jobId, sourceVertex).stream()
- .map(AggregatedMetric::getId)
- .filter(partitionRegex.asMatchPredicate())
- .count();
- if (numPartitions > 0) {
- LOG.debug(
- "Updating source {} max parallelism based on available partitions to {}",
- sourceVertex,
- numPartitions);
- topology.updateMaxParallelism(sourceVertex, (int) numPartitions);
+ try (var restClient = (RestClusterClient<String>) flinkService.getClusterClient(conf)) {
+ var partitionRegex = Pattern.compile("^.*\\.partition\\.\\d+\\.currentOffset$");
+ for (Map.Entry<JobVertexID, Set<JobVertexID>> entry : topology.getInputs().entrySet()) {
+ if (entry.getValue().isEmpty()) {
+ var sourceVertex = entry.getKey();
+ var numPartitions =
+ queryAggregatedMetricNames(restClient, jobId, sourceVertex).stream()
+ .map(AggregatedMetric::getId)
+ .filter(partitionRegex.asMatchPredicate())
+ .count();
+ if (numPartitions > 0) {
+ LOG.debug(
+ "Updating source {} max parallelism based on available partitions to {}",
+ sourceVertex,
+ numPartitions);
+ topology.updateMaxParallelism(sourceVertex, (int) numPartitions);
+ }
}
}
}
diff --git a/flink-kubernetes-operator-autoscaler/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/BacklogBasedScalingTest.java b/flink-kubernetes-operator-autoscaler/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/BacklogBasedScalingTest.java
index a4506d7b..bd9385ee 100644
--- a/flink-kubernetes-operator-autoscaler/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/BacklogBasedScalingTest.java
+++ b/flink-kubernetes-operator-autoscaler/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/BacklogBasedScalingTest.java
@@ -110,6 +110,7 @@ public class BacklogBasedScalingTest extends OperatorTestBase {
ReconciliationUtils.updateStatusForDeployedSpec(
app, configManager.getDeployConfig(app.getMetadata(), app.getSpec()));
app.getStatus().getJobStatus().setState(JobStatus.RUNNING.name());
+ app.getStatus().getReconciliationStatus().markReconciledSpecAsStable();
autoscaler =
new JobAutoScalerImpl(
@@ -130,7 +131,7 @@ public class BacklogBasedScalingTest extends OperatorTestBase {
/* Test scaling up. */
var now = Instant.ofEpochMilli(0);
setClocksTo(now);
- redeployJob(now);
+ metricsCollector.setJobUpdateTs(now);
// Adjust metric window size, so we can fill the metric window with two metrics
metricsCollector.setTestMetricWindowSize(Duration.ofSeconds(1));
metricsCollector.setCurrentMetrics(
@@ -200,7 +201,7 @@ public class BacklogBasedScalingTest extends OperatorTestBase {
now = now.plusSeconds(1);
setClocksTo(now);
// Redeploying which erases metric history
- redeployJob(now);
+ metricsCollector.setJobUpdateTs(now);
// Adjust metric window size, so we can fill the metric window with three metrics
metricsCollector.setTestMetricWindowSize(Duration.ofSeconds(2));
@@ -276,6 +277,7 @@ public class BacklogBasedScalingTest extends OperatorTestBase {
assertEquals(2, scaledParallelism.get(source1));
assertEquals(2, scaledParallelism.get(sink));
+ metricsCollector.setJobUpdateTs(now);
metricsCollector.setJobTopology(
new JobTopology(
new VertexInfo(source1, Set.of(), 2, 24),
@@ -304,9 +306,8 @@ public class BacklogBasedScalingTest extends OperatorTestBase {
"", Double.NaN, Double.NaN, Double.NaN, 900.))));
now = now.plus(Duration.ofSeconds(1));
setClocksTo(now);
- app.getStatus().getJobStatus().setStartTime(String.valueOf(now.toEpochMilli()));
autoscaler.scale(getResourceContext(app, ctx));
- assertFlinkMetricsCount(2, 1, ctx);
+ assertFlinkMetricsCount(2, 0, ctx);
scaledParallelism = ScalingExecutorTest.getScaledParallelism(app);
assertEquals(2, scaledParallelism.get(source1));
assertEquals(2, scaledParallelism.get(sink));
@@ -333,7 +334,7 @@ public class BacklogBasedScalingTest extends OperatorTestBase {
now = now.plus(Duration.ofSeconds(1));
setClocksTo(now);
autoscaler.scale(getResourceContext(app, ctx));
- assertFlinkMetricsCount(2, 2, ctx);
+ assertFlinkMetricsCount(2, 0, ctx);
scaledParallelism = ScalingExecutorTest.getScaledParallelism(app);
assertEquals(2, scaledParallelism.get(source1));
@@ -361,7 +362,7 @@ public class BacklogBasedScalingTest extends OperatorTestBase {
now = now.plus(Duration.ofSeconds(1));
setClocksTo(now);
autoscaler.scale(getResourceContext(app, ctx));
- assertFlinkMetricsCount(2, 3, ctx);
+ assertFlinkMetricsCount(2, 1, ctx);
scaledParallelism = ScalingExecutorTest.getScaledParallelism(app);
assertEquals(2, scaledParallelism.get(source1));
assertEquals(2, scaledParallelism.get(sink));
@@ -372,7 +373,7 @@ public class BacklogBasedScalingTest extends OperatorTestBase {
var ctx = createAutoscalerTestContext();
var now = Instant.ofEpochMilli(0);
setClocksTo(now);
- app.getStatus().getJobStatus().setUpdateTime(String.valueOf(now.toEpochMilli()));
+ metricsCollector.setJobUpdateTs(now);
metricsCollector.setCurrentMetrics(
Map.of(
source1,
@@ -418,16 +419,12 @@ public class BacklogBasedScalingTest extends OperatorTestBase {
var ctx = createAutoscalerTestContext();
var now = Instant.ofEpochMilli(0);
setClocksTo(now);
- app.getStatus().getJobStatus().setUpdateTime(String.valueOf(now.toEpochMilli()));
+ metricsCollector.setJobUpdateTs(now);
assertFalse(autoscaler.scale(getResourceContext(app, ctx)));
assertTrue(AutoScalerInfo.forResource(app, kubernetesClient).getMetricHistory().isEmpty());
assertTrue(eventCollector.events.isEmpty());
}
- private void redeployJob(Instant now) {
- app.getStatus().getJobStatus().setUpdateTime(String.valueOf(now.toEpochMilli()));
- }
-
private void setClocksTo(Instant time) {
var clock = Clock.fixed(time, ZoneId.systemDefault());
metricsCollector.setClock(clock);
diff --git a/flink-kubernetes-operator-autoscaler/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/JobAutoScalerImplTest.java b/flink-kubernetes-operator-autoscaler/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/JobAutoScalerImplTest.java
index 8f478bf0..d60ebcd8 100644
--- a/flink-kubernetes-operator-autoscaler/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/JobAutoScalerImplTest.java
+++ b/flink-kubernetes-operator-autoscaler/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/JobAutoScalerImplTest.java
@@ -57,6 +57,7 @@ public class JobAutoScalerImplTest extends OperatorTestBase {
configManager = new FlinkConfigManager(defaultConf);
ReconciliationUtils.updateStatusForDeployedSpec(
app, configManager.getDeployConfig(app.getMetadata(), app.getSpec()));
+ app.getStatus().getReconciliationStatus().markReconciledSpecAsStable();
}
@Test
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 c019d7aa..f0f3f318 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
@@ -124,8 +124,7 @@ public class MetricsCollectionAndEvaluationTest {
clock = Clock.fixed(Instant.ofEpochSecond(0), ZoneId.systemDefault());
metricsCollector.setClock(clock);
startTime = clock.instant();
- app.getStatus().getJobStatus().setStartTime(String.valueOf(startTime.toEpochMilli()));
- app.getStatus().getJobStatus().setUpdateTime(String.valueOf(startTime.toEpochMilli()));
+ metricsCollector.setJobUpdateTs(startTime);
app.getStatus().getJobStatus().setState(JobStatus.RUNNING.name());
}
@@ -171,6 +170,7 @@ public class MetricsCollectionAndEvaluationTest {
// Test resetting the collector and make sure we can deserialize the scalingInfo correctly
metricsCollector = new TestingMetricsCollector(topology);
+ metricsCollector.setJobUpdateTs(startTime);
metricsCollector.setClock(clock);
setDefaultMetrics(metricsCollector);
collectedMetrics = metricsCollector.updateMetrics(app, scalingInfo, service, conf);
@@ -350,10 +350,7 @@ public class MetricsCollectionAndEvaluationTest {
assertEquals(1, metricsHistory.getMetricHistory().size());
// Existing metrics should be cleared on job updates
- app.getStatus()
- .getJobStatus()
- .setUpdateTime(
- String.valueOf(clock.instant().plus(Duration.ofDays(10)).toEpochMilli()));
+ metricsCollector.setJobUpdateTs(clock.instant().plus(Duration.ofDays(10)));
metricsHistory = metricsCollector.updateMetrics(app, scalingInfo, service, conf);
assertEquals(0, metricsHistory.getMetricHistory().size());
}
@@ -376,6 +373,7 @@ public class MetricsCollectionAndEvaluationTest {
var topology = new JobTopology(new VertexInfo(source1, Set.of(), 5, 720));
metricsCollector = new TestingMetricsCollector(topology);
+ metricsCollector.setJobUpdateTs(startTime);
metricsCollector.setCurrentMetrics(
Map.of(
// Set source1 metrics without the PENDING_RECORDS metric
@@ -415,6 +413,7 @@ public class MetricsCollectionAndEvaluationTest {
var topology = new JobTopology(new VertexInfo(source1, Set.of(), 10, 720));
metricsCollector = new TestingMetricsCollector(topology);
+ metricsCollector.setJobUpdateTs(startTime);
metricsCollector.setCurrentMetrics(
Map.of(
// Set source1 metrics without the PENDING_RECORDS metric
diff --git a/flink-kubernetes-operator-autoscaler/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/RecommendedParallelismTest.java b/flink-kubernetes-operator-autoscaler/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/RecommendedParallelismTest.java
index c0dd8fa9..44210a55 100644
--- a/flink-kubernetes-operator-autoscaler/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/RecommendedParallelismTest.java
+++ b/flink-kubernetes-operator-autoscaler/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/RecommendedParallelismTest.java
@@ -112,6 +112,7 @@ public class RecommendedParallelismTest extends OperatorTestBase {
ReconciliationUtils.updateStatusForDeployedSpec(
app, configManager.getDeployConfig(app.getMetadata(), app.getSpec()));
app.getStatus().getJobStatus().setState(JobStatus.RUNNING.name());
+ app.getStatus().getReconciliationStatus().markReconciledSpecAsStable();
autoscaler =
new JobAutoScalerImpl(
@@ -262,12 +263,12 @@ public class RecommendedParallelismTest extends OperatorTestBase {
}
private void restart(Instant now) {
- app.getStatus().getJobStatus().setUpdateTime(String.valueOf(now.toEpochMilli()));
+ metricsCollector.setJobUpdateTs(now);
app.getStatus().getJobStatus().setState(JobStatus.CREATED.name());
}
private void running(Instant now) {
- app.getStatus().getJobStatus().setUpdateTime(String.valueOf(now.toEpochMilli()));
+ metricsCollector.setJobUpdateTs(now);
app.getStatus().getJobStatus().setState(JobStatus.RUNNING.name());
}
diff --git a/flink-kubernetes-operator-autoscaler/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/ScalingExecutorTest.java b/flink-kubernetes-operator-autoscaler/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/ScalingExecutorTest.java
index dbdf1451..fb9ec4e1 100644
--- a/flink-kubernetes-operator-autoscaler/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/ScalingExecutorTest.java
+++ b/flink-kubernetes-operator-autoscaler/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/ScalingExecutorTest.java
@@ -36,10 +36,7 @@ import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;
-import java.time.Clock;
import java.time.Duration;
-import java.time.Instant;
-import java.time.ZoneId;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -85,50 +82,6 @@ public class ScalingExecutorTest {
jobStatus.setState(JobStatus.RUNNING.name());
}
- @Test
- public void testStabilizationPeriod() throws Exception {
- conf.set(AutoScalerOptions.STABILIZATION_INTERVAL, Duration.ofMinutes(1));
- conf.set(AutoScalerOptions.SCALING_EFFECTIVENESS_DETECTION_ENABLED, false);
-
- var metrics = Map.of(new JobVertexID(), evaluated(1, 110, 100));
-
- var scalingInfo = new AutoScalerInfo(new HashMap<>());
- var clock = Clock.fixed(Instant.now(), ZoneId.systemDefault());
- var jobStatus = flinkDep.getStatus().getJobStatus();
- jobStatus.setUpdateTime(String.valueOf(clock.instant().toEpochMilli()));
-
- scalingDecisionExecutor.setClock(clock);
- assertFalse(scalingDecisionExecutor.scaleResource(flinkDep, scalingInfo, conf, metrics));
-
- clock = Clock.offset(clock, Duration.ofSeconds(30));
- scalingDecisionExecutor.setClock(clock);
- assertFalse(scalingDecisionExecutor.scaleResource(flinkDep, scalingInfo, conf, metrics));
-
- clock = Clock.offset(clock, Duration.ofSeconds(20));
- scalingDecisionExecutor.setClock(clock);
- assertFalse(scalingDecisionExecutor.scaleResource(flinkDep, scalingInfo, conf, metrics));
-
- clock = Clock.offset(clock, Duration.ofSeconds(20));
- scalingDecisionExecutor.setClock(clock);
- assertTrue(scalingDecisionExecutor.scaleResource(flinkDep, scalingInfo, conf, metrics));
-
- // A job should not be considered stable in a non-RUNNING state
- jobStatus.setState(JobStatus.FAILING.name());
- assertFalse(scalingDecisionExecutor.scaleResource(flinkDep, scalingInfo, conf, metrics));
-
- jobStatus.setState(JobStatus.RUNNING.name());
- jobStatus.setUpdateTime(String.valueOf(clock.instant().toEpochMilli()));
- assertFalse(scalingDecisionExecutor.scaleResource(flinkDep, scalingInfo, conf, metrics));
-
- clock = Clock.offset(clock, Duration.ofSeconds(59));
- scalingDecisionExecutor.setClock(clock);
- assertFalse(scalingDecisionExecutor.scaleResource(flinkDep, scalingInfo, conf, metrics));
-
- clock = Clock.offset(clock, Duration.ofSeconds(2));
- scalingDecisionExecutor.setClock(clock);
- assertTrue(scalingDecisionExecutor.scaleResource(flinkDep, scalingInfo, conf, metrics));
- }
-
@Test
public void testUtilizationBoundaries() {
// Restart time should not affect utilization boundary
diff --git a/flink-kubernetes-operator-autoscaler/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/ScalingMetricCollectorTest.java b/flink-kubernetes-operator-autoscaler/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/ScalingMetricCollectorTest.java
index a6e5665c..f6da361c 100644
--- a/flink-kubernetes-operator-autoscaler/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/ScalingMetricCollectorTest.java
+++ b/flink-kubernetes-operator-autoscaler/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/ScalingMetricCollectorTest.java
@@ -18,15 +18,19 @@
package org.apache.flink.kubernetes.operator.autoscaler;
import org.apache.flink.api.common.JobID;
-import org.apache.flink.configuration.Configuration;
-import org.apache.flink.kubernetes.operator.TestingClusterClient;
+import org.apache.flink.api.common.JobStatus;
+import org.apache.flink.runtime.rest.messages.JobPlanInfo;
import org.apache.flink.runtime.rest.messages.job.JobDetailsInfo;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
import org.junit.jupiter.api.Test;
-import java.util.concurrent.CompletableFuture;
+import java.time.Instant;
+import java.util.List;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
/** Tests for {@link ScalingMetricCollector}. */
public class ScalingMetricCollectorTest {
@@ -38,8 +42,27 @@ public class ScalingMetricCollectorTest {
JobDetailsInfo jobDetailsInfo = new ObjectMapper().readValue(s, JobDetailsInfo.class);
var metricsCollector = new RestApiMetricsCollector();
- var client = new TestingClusterClient<>(new Configuration(), "test");
- client.setRequestProcessor((h, p, b) -> CompletableFuture.completedFuture(jobDetailsInfo));
- metricsCollector.queryJobTopology(client, new JobID());
+ metricsCollector.getJobTopology(jobDetailsInfo);
+ }
+
+ @Test
+ public void testJobUpdateTsLogic() {
+ var details =
+ new JobDetailsInfo(
+ new JobID(),
+ "",
+ false,
+ org.apache.flink.api.common.JobStatus.RUNNING,
+ 0,
+ 0,
+ 0,
+ 0,
+ 0,
+ Map.of(JobStatus.RUNNING, 3L, JobStatus.CREATED, 2L),
+ List.of(),
+ Map.of(),
+ new JobPlanInfo.RawJson(""));
+ var metricsCollector = new RestApiMetricsCollector();
+ assertEquals(Instant.ofEpochMilli(3L), metricsCollector.getJobUpdateTs(details));
}
}
diff --git a/flink-kubernetes-operator-autoscaler/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/TestingMetricsCollector.java b/flink-kubernetes-operator-autoscaler/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/TestingMetricsCollector.java
index 2d9eb534..9a3c1d13 100644
--- a/flink-kubernetes-operator-autoscaler/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/TestingMetricsCollector.java
+++ b/flink-kubernetes-operator-autoscaler/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/TestingMetricsCollector.java
@@ -25,11 +25,13 @@ import org.apache.flink.kubernetes.operator.autoscaler.metrics.FlinkMetric;
import org.apache.flink.kubernetes.operator.autoscaler.topology.JobTopology;
import org.apache.flink.kubernetes.operator.service.FlinkService;
import org.apache.flink.runtime.jobgraph.JobVertexID;
+import org.apache.flink.runtime.rest.messages.job.JobDetailsInfo;
import org.apache.flink.runtime.rest.messages.job.metrics.AggregatedMetric;
import lombok.Setter;
import java.time.Duration;
+import java.time.Instant;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
@@ -42,6 +44,8 @@ public class TestingMetricsCollector extends ScalingMetricCollector {
@Setter private Duration testMetricWindowSize;
+ @Setter private Instant jobUpdateTs;
+
@Setter
private Map<JobVertexID, Map<FlinkMetric, AggregatedMetric>> currentMetrics = new HashMap<>();
@@ -52,7 +56,7 @@ public class TestingMetricsCollector extends ScalingMetricCollector {
}
@Override
- protected JobTopology queryJobTopology(RestClusterClient<String> restClient, JobID jobId) {
+ protected JobTopology getJobTopology(JobDetailsInfo jobDetailsInfo) {
return jobTopology;
}
@@ -87,4 +91,9 @@ public class TestingMetricsCollector extends ScalingMetricCollector {
}
return super.getMetricWindowSize(conf);
}
+
+ @Override
+ protected Instant getJobUpdateTs(JobDetailsInfo jobDetailsInfo) {
+ return jobUpdateTs;
+ }
}
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkService.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkService.java
index b48f6bb1..bc1e4da4 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkService.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkService.java
@@ -62,6 +62,7 @@ import org.apache.flink.runtime.rest.messages.EmptyMessageParameters;
import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
import org.apache.flink.runtime.rest.messages.JobsOverviewHeaders;
import org.apache.flink.runtime.rest.messages.TriggerId;
+import org.apache.flink.runtime.rest.messages.job.JobDetailsInfo;
import org.apache.flink.runtime.rest.messages.job.metrics.JobMetricsHeaders;
import org.apache.flink.runtime.rest.messages.job.savepoints.SavepointDisposalRequest;
import org.apache.flink.runtime.rest.messages.job.savepoints.SavepointDisposalTriggerHeaders;
@@ -809,6 +810,21 @@ public abstract class AbstractFlinkService implements FlinkService {
}
}
+ @Override
+ public JobDetailsInfo getJobDetailsInfo(JobID jobID, Configuration conf) throws Exception {
+
+ try (var restClient = getClusterClient(conf)) {
+ return restClient
+ .getJobDetails(jobID)
+ .get(
+ configManager
+ .getOperatorConfiguration()
+ .getFlinkClientTimeout()
+ .toSeconds(),
+ TimeUnit.SECONDS);
+ }
+ }
+
/** Wait until the FLink cluster has completely shut down. */
@VisibleForTesting
void waitForClusterShutdown(String namespace, String clusterId, long shutdownTimeout) {
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/FlinkService.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/FlinkService.java
index 01271496..edfb322f 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/FlinkService.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/FlinkService.java
@@ -34,6 +34,7 @@ import org.apache.flink.kubernetes.operator.controller.FlinkResourceContext;
import org.apache.flink.kubernetes.operator.observer.SavepointFetchResult;
import org.apache.flink.runtime.client.JobStatusMessage;
import org.apache.flink.runtime.jobmaster.JobResult;
+import org.apache.flink.runtime.rest.messages.job.JobDetailsInfo;
import io.fabric8.kubernetes.api.model.ObjectMeta;
import io.fabric8.kubernetes.api.model.PodList;
@@ -115,4 +116,6 @@ public interface FlinkService {
throws Exception;
RestClusterClient<String> getClusterClient(Configuration conf) throws Exception;
+
+ JobDetailsInfo getJobDetailsInfo(JobID jobID, Configuration conf) throws Exception;
}
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestingFlinkService.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestingFlinkService.java
index f3471496..22d6be41 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestingFlinkService.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestingFlinkService.java
@@ -46,6 +46,7 @@ import org.apache.flink.kubernetes.operator.exception.RecoveryFailureException;
import org.apache.flink.kubernetes.operator.observer.SavepointFetchResult;
import org.apache.flink.kubernetes.operator.service.AbstractFlinkService;
import org.apache.flink.kubernetes.operator.service.CheckpointHistoryWrapper;
+import org.apache.flink.kubernetes.operator.service.NativeFlinkServiceTest;
import org.apache.flink.kubernetes.operator.standalone.StandaloneKubernetesConfigOptionsInternal;
import org.apache.flink.kubernetes.operator.utils.FlinkUtils;
import org.apache.flink.kubernetes.operator.utils.SavepointUtils;
@@ -62,6 +63,7 @@ import org.apache.flink.runtime.messages.webmonitor.MultipleJobsDetails;
import org.apache.flink.runtime.rest.messages.DashboardConfiguration;
import org.apache.flink.runtime.rest.messages.EmptyResponseBody;
import org.apache.flink.runtime.rest.messages.JobsOverviewHeaders;
+import org.apache.flink.runtime.rest.messages.job.JobDetailsInfo;
import org.apache.flink.runtime.rest.messages.job.metrics.AggregatedMetric;
import org.apache.flink.runtime.rest.messages.job.metrics.AggregatedMetricsResponseBody;
import org.apache.flink.runtime.rest.messages.job.metrics.AggregatedSubtaskMetricsHeaders;
@@ -534,22 +536,6 @@ public class TestingFlinkService extends AbstractFlinkService {
jobErrors.put(jobID, error);
}
- /** The information collector of a submitted job. */
- public static class SubmittedJobInfo {
- public final String savepointPath;
- public final JobStatusMessage jobStatusMessage;
- public final Configuration effectiveConfig;
-
- public SubmittedJobInfo(
- String savepointPath,
- JobStatusMessage jobStatusMessage,
- Configuration effectiveConfig) {
- this.savepointPath = savepointPath;
- this.jobStatusMessage = jobStatusMessage;
- this.effectiveConfig = effectiveConfig;
- }
- }
-
@Override
public Map<String, String> getClusterInfo(Configuration conf) {
return CLUSTER_INFO;
@@ -559,13 +545,11 @@ public class TestingFlinkService extends AbstractFlinkService {
public boolean scale(FlinkResourceContext<?> ctx) {
boolean standalone = ctx.getDeploymentMode() == KubernetesDeploymentMode.STANDALONE;
boolean session = ctx.getResource().getSpec().getJob() == null;
- if (!standalone) {
- return false;
- }
+ boolean reactive =
+ ctx.getObserveConfig().get(JobManagerOptions.SCHEDULER_MODE)
+ == SchedulerExecutionMode.REACTIVE;
- if (session
- || ctx.getObserveConfig().get(JobManagerOptions.SCHEDULER_MODE)
- == SchedulerExecutionMode.REACTIVE) {
+ if (standalone && (session || reactive)) {
desiredReplicas =
ctx.getDeployConfig(ctx.getResource().getSpec())
.get(
@@ -591,4 +575,9 @@ public class TestingFlinkService extends AbstractFlinkService {
Configuration conf, String jobId, List<String> metricNames) {
return metricsValues;
}
+
+ @Override
+ public JobDetailsInfo getJobDetailsInfo(JobID jobID, Configuration conf) {
+ return NativeFlinkServiceTest.createJobDetailsFor(List.of());
+ }
}