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
The following commit(s) were added to refs/heads/main by this push:
new 49f33337 [FLINK-29394] Add observe Flink job health
49f33337 is described below
commit 49f3333796b7c5dae2a441d453b67d7e8bcee3bf
Author: Gabor Somogyi <[email protected]>
AuthorDate: Mon Sep 26 10:27:45 2022 +0200
[FLINK-29394] Add observe Flink job health
---
.../content/docs/custom-resource/job-management.md | 9 +
docs/content/docs/custom-resource/reference.md | 2 +-
.../shortcodes/generated/dynamic_section.html | 18 ++
.../kubernetes_operator_config_configuration.html | 18 ++
.../config/KubernetesOperatorConfigOptions.java | 24 +++
.../operator/crd/status/FlinkDeploymentStatus.java | 2 +-
.../operator/health/ClusterHealthInfo.java | 76 ++++++++
.../operator/observer/ClusterHealthEvaluator.java | 146 +++++++++++++++
.../operator/observer/ClusterHealthObserver.java | 90 +++++++++
.../deployment/AbstractDeploymentObserver.java | 10 +-
.../observer/deployment/ApplicationObserver.java | 9 +
.../AbstractFlinkResourceReconciler.java | 30 +--
.../deployment/ApplicationReconciler.java | 47 ++++-
.../operator/service/AbstractFlinkService.java | 23 +++
.../kubernetes/operator/service/FlinkService.java | 4 +
.../operator/validation/DefaultValidator.java | 14 ++
.../kubernetes/operator/TestingFlinkService.java | 12 ++
.../controller/UnhealthyDeploymentRestartTest.java | 117 ++++++++++++
.../operator/health/ClusterHealthInfoTest.java | 53 ++++++
.../observer/ClusterHealthEvaluatorTest.java | 202 +++++++++++++++++++++
.../operator/validation/DefaultValidatorTest.java | 18 ++
21 files changed, 900 insertions(+), 24 deletions(-)
diff --git a/docs/content/docs/custom-resource/job-management.md
b/docs/content/docs/custom-resource/job-management.md
index b23880bc..ee00a117 100644
--- a/docs/content/docs/custom-resource/job-management.md
+++ b/docs/content/docs/custom-resource/job-management.md
@@ -232,6 +232,15 @@ In these cases no recovery will be performed to avoid
dataloss and an error will
Please check the [manual Recovery section](#manual-recovery) to understand how
to recover from these situations.
+## Restart of unhealthy job deployments
+
+When Kubernetes HA is enabled, the operator can restart the Flink cluster
deployments in cases when it was considered
+unhealthy. Unhealthy deployment restart can be turned on in the configuration
by setting `kubernetes.operator.cluster.health-check.enabled` to `true`
(default: `false`).
+In order this feature to work one must enable [recovery of missing job
deployments](#recovery-of-missing-job-deployments).
+
+At the moment deployment is considered unhealthy when Flink's restarts count
reaches `kubernetes.operator.cluster.health-check.restarts.threshold` (default:
`64`)
+within time window of
`kubernetes.operator.cluster.health-check.restarts.window` (default: 2 minutes).
+
## Application upgrade rollbacks (Experimental)
The operator supports upgrade rollbacks as an experimental feature.
diff --git a/docs/content/docs/custom-resource/reference.md
b/docs/content/docs/custom-resource/reference.md
index 924e89f5..36c11f65 100644
--- a/docs/content/docs/custom-resource/reference.md
+++ b/docs/content/docs/custom-resource/reference.md
@@ -197,7 +197,7 @@ This page serves as a full reference for FlinkDeployment
custom resource definit
| ----------| ---- | ---- |
| jobStatus | org.apache.flink.kubernetes.operator.crd.status.JobStatus | Last
observed status of the Flink job on Application/Session cluster. |
| error | java.lang.String | Error information about the
FlinkDeployment/FlinkSessionJob. |
-| clusterInfo | java.util.Map<java.lang.String,java.lang.String> | Config
information from running clusters. |
+| clusterInfo | java.util.Map<java.lang.String,java.lang.String> | Information
from running clusters. |
| jobManagerDeploymentStatus |
org.apache.flink.kubernetes.operator.crd.status.JobManagerDeploymentStatus |
Last observed status of the JobManager deployment. |
| reconciliationStatus |
org.apache.flink.kubernetes.operator.crd.status.FlinkDeploymentReconciliationStatus
| Status of the last reconcile operation. |
| taskManager |
org.apache.flink.kubernetes.operator.crd.status.TaskManagerInfo | Information
about the TaskManagers for the scale subresource. |
diff --git a/docs/layouts/shortcodes/generated/dynamic_section.html
b/docs/layouts/shortcodes/generated/dynamic_section.html
index 4cfac307..79696e38 100644
--- a/docs/layouts/shortcodes/generated/dynamic_section.html
+++ b/docs/layouts/shortcodes/generated/dynamic_section.html
@@ -8,6 +8,24 @@
</tr>
</thead>
<tbody>
+ <tr>
+ <td><h5>kubernetes.operator.cluster.health-check.enabled</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Whether to enable health check for clusters.</td>
+ </tr>
+ <tr>
+
<td><h5>kubernetes.operator.cluster.health-check.restarts.threshold</h5></td>
+ <td style="word-wrap: break-word;">64</td>
+ <td>Integer</td>
+ <td>The threshold which is checked against job restart count
within a configured window. If the restart count is reaching the threshold then
full cluster restart is initiated.</td>
+ </tr>
+ <tr>
+
<td><h5>kubernetes.operator.cluster.health-check.restarts.window</h5></td>
+ <td style="word-wrap: break-word;">2 min</td>
+ <td>Duration</td>
+ <td>The duration of the time window where job restart count
measured.</td>
+ </tr>
<tr>
<td><h5>kubernetes.operator.deployment.readiness.timeout</h5></td>
<td style="word-wrap: break-word;">1 min</td>
diff --git
a/docs/layouts/shortcodes/generated/kubernetes_operator_config_configuration.html
b/docs/layouts/shortcodes/generated/kubernetes_operator_config_configuration.html
index f00a1204..acdcaccb 100644
---
a/docs/layouts/shortcodes/generated/kubernetes_operator_config_configuration.html
+++
b/docs/layouts/shortcodes/generated/kubernetes_operator_config_configuration.html
@@ -8,6 +8,24 @@
</tr>
</thead>
<tbody>
+ <tr>
+ <td><h5>kubernetes.operator.cluster.health-check.enabled</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Whether to enable health check for clusters.</td>
+ </tr>
+ <tr>
+
<td><h5>kubernetes.operator.cluster.health-check.restarts.threshold</h5></td>
+ <td style="word-wrap: break-word;">64</td>
+ <td>Integer</td>
+ <td>The threshold which is checked against job restart count
within a configured window. If the restart count is reaching the threshold then
full cluster restart is initiated.</td>
+ </tr>
+ <tr>
+
<td><h5>kubernetes.operator.cluster.health-check.restarts.window</h5></td>
+ <td style="word-wrap: break-word;">2 min</td>
+ <td>Duration</td>
+ <td>The duration of the time window where job restart count
measured.</td>
+ </tr>
<tr>
<td><h5>kubernetes.operator.config.cache.size</h5></td>
<td style="word-wrap: break-word;">1000</td>
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/KubernetesOperatorConfigOptions.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/KubernetesOperatorConfigOptions.java
index 18664a1f..8c18f249 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/KubernetesOperatorConfigOptions.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/KubernetesOperatorConfigOptions.java
@@ -315,4 +315,28 @@ public class KubernetesOperatorConfigOptions {
.intType()
.defaultValue(8085)
.withDescription("The port the health probe will use to
expose the status.");
+
+ @Documentation.Section(SECTION_DYNAMIC)
+ public static final ConfigOption<Boolean>
OPERATOR_CLUSTER_HEALTH_CHECK_ENABLED =
+ operatorConfig("cluster.health-check.enabled")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription("Whether to enable health check for
clusters.");
+
+ @Documentation.Section(SECTION_DYNAMIC)
+ public static final ConfigOption<Duration>
OPERATOR_CLUSTER_HEALTH_CHECK_RESTARTS_WINDOW =
+ operatorConfig("cluster.health-check.restarts.window")
+ .durationType()
+ .defaultValue(Duration.ofMinutes(2))
+ .withDescription(
+ "The duration of the time window where job restart
count measured.");
+
+ @Documentation.Section(SECTION_DYNAMIC)
+ public static final ConfigOption<Integer>
OPERATOR_CLUSTER_HEALTH_CHECK_RESTARTS_THRESHOLD =
+ operatorConfig("cluster.health-check.restarts.threshold")
+ .intType()
+ .defaultValue(64)
+ .withDescription(
+ "The threshold which is checked against job
restart count within a configured window. "
+ + "If the restart count is reaching the
threshold then full cluster restart is initiated.");
}
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/status/FlinkDeploymentStatus.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/status/FlinkDeploymentStatus.java
index c2f6d0d2..c9367c70 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/status/FlinkDeploymentStatus.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/status/FlinkDeploymentStatus.java
@@ -40,7 +40,7 @@ import java.util.Map;
@SuperBuilder
public class FlinkDeploymentStatus extends CommonStatus<FlinkDeploymentSpec> {
- /** Config information from running clusters. */
+ /** Information from running clusters. */
private Map<String, String> clusterInfo = new HashMap<>();
/** Last observed status of the JobManager deployment. */
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/health/ClusterHealthInfo.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/health/ClusterHealthInfo.java
new file mode 100644
index 00000000..bb76f862
--- /dev/null
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/health/ClusterHealthInfo.java
@@ -0,0 +1,76 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.kubernetes.operator.health;
+
+import org.apache.flink.annotation.Experimental;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import java.time.Clock;
+
+/** Represents information about job health. */
+@Experimental
+@Data
+@AllArgsConstructor
+@NoArgsConstructor
+public class ClusterHealthInfo {
+ /** Millisecond timestamp of the last observed health information. */
+ private long timeStamp;
+
+ /** Number of restarts. */
+ private int numRestarts;
+
+ /** Calculated field whether the cluster is healthy or not. */
+ private boolean healthy;
+
+ public static ClusterHealthInfo of(int numRestarts) {
+ return of(Clock.systemDefaultZone(), numRestarts);
+ }
+
+ public static ClusterHealthInfo of(Clock clock, int numRestarts) {
+ return new ClusterHealthInfo(clock.millis(), numRestarts, true);
+ }
+
+ public static boolean isValid(ClusterHealthInfo clusterHealthInfo) {
+ return clusterHealthInfo.timeStamp != 0;
+ }
+
+ public static String serialize(ClusterHealthInfo clusterHealthInfo) {
+ ObjectMapper objectMapper = new ObjectMapper();
+ try {
+ return objectMapper.writeValueAsString(clusterHealthInfo);
+ } catch (JsonProcessingException e) {
+ throw new RuntimeException(
+ "Could not serialize ClusterHealthInfo, this indicates a
bug...", e);
+ }
+ }
+
+ public static ClusterHealthInfo deserialize(String data) {
+ ObjectMapper objectMapper = new ObjectMapper();
+ try {
+ return objectMapper.readValue(data, ClusterHealthInfo.class);
+ } catch (JsonProcessingException e) {
+ throw new RuntimeException(
+ "Could not deserialize ClusterHealthInfo, this indicates a
bug...", e);
+ }
+ }
+}
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/ClusterHealthEvaluator.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/ClusterHealthEvaluator.java
new file mode 100644
index 00000000..4f5f3dc6
--- /dev/null
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/ClusterHealthEvaluator.java
@@ -0,0 +1,146 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.kubernetes.operator.observer;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.kubernetes.operator.health.ClusterHealthInfo;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.time.Clock;
+import java.time.Duration;
+import java.util.Map;
+
+import static
org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions.OPERATOR_CLUSTER_HEALTH_CHECK_RESTARTS_THRESHOLD;
+import static
org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions.OPERATOR_CLUSTER_HEALTH_CHECK_RESTARTS_WINDOW;
+
+/** Evaluates whether the cluster is healthy. */
+public class ClusterHealthEvaluator {
+
+ private static final String CLUSTER_INFO_KEY =
ClusterHealthInfo.class.getSimpleName();
+
+ private static final Logger LOG =
LoggerFactory.getLogger(ClusterHealthEvaluator.class);
+
+ private final Clock clock;
+
+ public ClusterHealthEvaluator(Clock clock) {
+ this.clock = clock;
+ }
+
+ public static ClusterHealthInfo getLastValidClusterHealthInfo(Map<String,
String> clusterInfo) {
+ LOG.debug("Getting last valid health check info");
+ if (clusterInfo.containsKey(CLUSTER_INFO_KEY)) {
+ return
ClusterHealthInfo.deserialize(clusterInfo.get(CLUSTER_INFO_KEY));
+ } else {
+ LOG.debug("No last valid health check info");
+ return null;
+ }
+ }
+
+ public static void setLastValidClusterHealthInfo(
+ Map<String, String> clusterInfo, ClusterHealthInfo
clusterHealthInfo) {
+ LOG.debug("Setting last valid health check info");
+ clusterInfo.put(CLUSTER_INFO_KEY,
ClusterHealthInfo.serialize(clusterHealthInfo));
+ }
+
+ public static void removeLastValidClusterHealthInfo(Map<String, String>
clusterInfo) {
+ LOG.debug("Removing last valid health check info");
+ clusterInfo.remove(CLUSTER_INFO_KEY);
+ }
+
+ public void evaluate(
+ Configuration configuration,
+ Map<String, String> clusterInfo,
+ ClusterHealthInfo observedClusterHealthInfo) {
+
+ if (ClusterHealthInfo.isValid(observedClusterHealthInfo)) {
+ LOG.debug("Observed health info is valid");
+
+ var lastValidClusterHealthInfo =
getLastValidClusterHealthInfo(clusterInfo);
+ if (lastValidClusterHealthInfo == null) {
+ LOG.debug("No last valid health info, skipping health check");
+ setLastValidClusterHealthInfo(clusterInfo,
observedClusterHealthInfo);
+ } else if (observedClusterHealthInfo.getTimeStamp()
+ < lastValidClusterHealthInfo.getTimeStamp()) {
+ String msg =
+ "Observed health info timestamp is less than the last
valid health info timestamp, this indicates a bug...";
+ LOG.error(msg);
+ throw new IllegalStateException(msg);
+ } else if (observedClusterHealthInfo.getNumRestarts()
+ < lastValidClusterHealthInfo.getNumRestarts()) {
+ LOG.debug(
+ "Observed health info number of restarts is less than
the last valid health info number of restarts, skipping health check");
+ setLastValidClusterHealthInfo(clusterInfo,
observedClusterHealthInfo);
+ } else {
+ boolean isHealthy = true;
+
+ LOG.debug("Valid health info exist, checking cluster health");
+ LOG.debug("Last valid health info: {}",
lastValidClusterHealthInfo);
+ LOG.debug("Observed health info: {}",
observedClusterHealthInfo);
+
+ var timestampDiffMs =
+ observedClusterHealthInfo.getTimeStamp()
+ - lastValidClusterHealthInfo.getTimeStamp();
+ LOG.debug(
+ "Time difference between health infos: {}",
+ Duration.ofMillis(timestampDiffMs));
+
+ var restartCheckWindow =
+
configuration.get(OPERATOR_CLUSTER_HEALTH_CHECK_RESTARTS_WINDOW);
+ var restartCheckWindowMs = restartCheckWindow.toMillis();
+ double countMultiplier = (double) restartCheckWindowMs /
(double) timestampDiffMs;
+ // If the 2 health info timestamp difference is within the
window then no
+ // scaling needed
+ if (countMultiplier > 1) {
+ countMultiplier = 1;
+ }
+ long numRestarts =
+ (long)
+ ((double)
+
(observedClusterHealthInfo.getNumRestarts()
+ -
lastValidClusterHealthInfo
+
.getNumRestarts())
+ * countMultiplier);
+ LOG.debug(
+ "Calculated restart count for {} window: {}",
+ restartCheckWindow,
+ numRestarts);
+
+ if (lastValidClusterHealthInfo.getTimeStamp()
+ < clock.millis() - restartCheckWindowMs) {
+ LOG.debug("Last valid health info timestamp is outside of
the window");
+ setLastValidClusterHealthInfo(clusterInfo,
observedClusterHealthInfo);
+ }
+
+ var restartThreshold =
+
configuration.get(OPERATOR_CLUSTER_HEALTH_CHECK_RESTARTS_THRESHOLD);
+ if (numRestarts > restartThreshold) {
+ LOG.info("Restart count hit threshold: {}",
restartThreshold);
+ setLastValidClusterHealthInfo(clusterInfo,
observedClusterHealthInfo);
+ isHealthy = false;
+ }
+
+ // Update the health flag
+ lastValidClusterHealthInfo =
getLastValidClusterHealthInfo(clusterInfo);
+ lastValidClusterHealthInfo.setHealthy(isHealthy);
+ setLastValidClusterHealthInfo(clusterInfo,
lastValidClusterHealthInfo);
+ }
+ }
+ }
+}
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/ClusterHealthObserver.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/ClusterHealthObserver.java
new file mode 100644
index 00000000..8bb042f9
--- /dev/null
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/ClusterHealthObserver.java
@@ -0,0 +1,90 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.kubernetes.operator.observer;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.kubernetes.operator.crd.FlinkDeployment;
+import org.apache.flink.kubernetes.operator.health.ClusterHealthInfo;
+import org.apache.flink.kubernetes.operator.service.FlinkService;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.time.Clock;
+import java.util.List;
+
+/** An observer to observe the cluster health. */
+public class ClusterHealthObserver {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(ClusterHealthObserver.class);
+ private static final String FULL_RESTARTS_METRIC_NAME = "fullRestarts";
+ private static final String NUM_RESTARTS_METRIC_NAME = "numRestarts";
+ private final FlinkService flinkService;
+ private final ClusterHealthEvaluator clusterHealthEvaluator;
+
+ public ClusterHealthObserver(FlinkService flinkService) {
+ this.flinkService = flinkService;
+ this.clusterHealthEvaluator = new
ClusterHealthEvaluator(Clock.systemDefaultZone());
+ }
+
+ /** Observe the health of the flink cluster. */
+ public void observe(FlinkDeployment flinkApp, Configuration
deployedConfig) {
+ try {
+ LOG.info("Observing cluster health");
+ var deploymentStatus = flinkApp.getStatus();
+ var jobStatus = deploymentStatus.getJobStatus();
+ var jobId = jobStatus.getJobId();
+ var metrics =
+ flinkService.getMetrics(
+ deployedConfig,
+ jobId,
+ List.of(FULL_RESTARTS_METRIC_NAME,
NUM_RESTARTS_METRIC_NAME));
+ ClusterHealthInfo observedClusterHealthInfo;
+ if (metrics.containsKey(NUM_RESTARTS_METRIC_NAME)) {
+ LOG.debug(NUM_RESTARTS_METRIC_NAME + " metric is used");
+ observedClusterHealthInfo =
+ ClusterHealthInfo.of(
+
Integer.parseInt(metrics.get(NUM_RESTARTS_METRIC_NAME)));
+ } else if (metrics.containsKey(FULL_RESTARTS_METRIC_NAME)) {
+ LOG.debug(
+ FULL_RESTARTS_METRIC_NAME
+ + " metric is used because "
+ + NUM_RESTARTS_METRIC_NAME
+ + " is missing");
+ observedClusterHealthInfo =
+ ClusterHealthInfo.of(
+
Integer.parseInt(metrics.get(FULL_RESTARTS_METRIC_NAME)));
+ } else {
+ throw new IllegalStateException(
+ "No job restart metric found. Either "
+ + FULL_RESTARTS_METRIC_NAME
+ + "(old and deprecated in never Flink
versions) or "
+ + NUM_RESTARTS_METRIC_NAME
+ + "(new) must exist.");
+ }
+ LOG.debug("Observed cluster health: {}",
observedClusterHealthInfo);
+
+ clusterHealthEvaluator.evaluate(
+ deployedConfig, deploymentStatus.getClusterInfo(),
observedClusterHealthInfo);
+ } catch (Exception e) {
+ LOG.warn("Exception while observing cluster health: {}",
e.getMessage());
+ // Intentionally not throwing exception since we handle fetch
metrics failure as
+ // temporary issue
+ }
+ }
+}
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/deployment/AbstractDeploymentObserver.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/deployment/AbstractDeploymentObserver.java
index 744ecd64..7319e923 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/deployment/AbstractDeploymentObserver.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/deployment/AbstractDeploymentObserver.java
@@ -47,7 +47,6 @@ import io.javaoperatorsdk.operator.api.reconciler.Context;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
@@ -116,13 +115,10 @@ public abstract class AbstractDeploymentObserver
implements Observer<FlinkDeploy
}
private void observeClusterInfo(FlinkDeployment flinkApp, Configuration
configuration) {
- if (!flinkApp.getStatus().getClusterInfo().isEmpty()) {
- return;
- }
try {
Map<String, String> clusterInfo =
flinkService.getClusterInfo(configuration);
- flinkApp.getStatus().setClusterInfo(clusterInfo);
- logger.debug("ClusterInfo: {}", clusterInfo);
+ flinkApp.getStatus().getClusterInfo().putAll(clusterInfo);
+ logger.debug("ClusterInfo: {}",
flinkApp.getStatus().getClusterInfo());
} catch (Exception e) {
logger.error("Exception while fetching cluster info", e);
}
@@ -139,8 +135,6 @@ public abstract class AbstractDeploymentObserver implements
Observer<FlinkDeploy
return;
}
- flinkApp.getStatus().setClusterInfo(new HashMap<>());
-
logger.info(
"Observing JobManager deployment. Previous status: {}",
previousJmStatus.name());
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/deployment/ApplicationObserver.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/deployment/ApplicationObserver.java
index a2eac7e3..c41ee389 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/deployment/ApplicationObserver.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/deployment/ApplicationObserver.java
@@ -22,6 +22,7 @@ import
org.apache.flink.kubernetes.operator.config.FlinkConfigManager;
import org.apache.flink.kubernetes.operator.crd.FlinkDeployment;
import org.apache.flink.kubernetes.operator.crd.status.FlinkDeploymentStatus;
import org.apache.flink.kubernetes.operator.crd.status.JobStatus;
+import org.apache.flink.kubernetes.operator.observer.ClusterHealthObserver;
import org.apache.flink.kubernetes.operator.observer.JobStatusObserver;
import org.apache.flink.kubernetes.operator.observer.SavepointObserver;
import
org.apache.flink.kubernetes.operator.observer.context.ApplicationObserverContext;
@@ -35,12 +36,16 @@ import io.javaoperatorsdk.operator.api.reconciler.Context;
import java.util.List;
import java.util.Optional;
+import static
org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions.OPERATOR_CLUSTER_HEALTH_CHECK_ENABLED;
+
/** The observer of {@link
org.apache.flink.kubernetes.operator.config.Mode#APPLICATION} cluster. */
public class ApplicationObserver extends AbstractDeploymentObserver {
private final SavepointObserver<FlinkDeployment, FlinkDeploymentStatus>
savepointObserver;
private final JobStatusObserver<FlinkDeployment,
ApplicationObserverContext> jobStatusObserver;
+ private final ClusterHealthObserver clusterHealthObserver;
+
public ApplicationObserver(
FlinkService flinkService,
FlinkConfigManager configManager,
@@ -94,6 +99,7 @@ public class ApplicationObserver extends
AbstractDeploymentObserver {
err);
}
};
+ this.clusterHealthObserver = new ClusterHealthObserver(flinkService);
}
@Override
@@ -108,6 +114,9 @@ public class ApplicationObserver extends
AbstractDeploymentObserver {
new ApplicationObserverContext(flinkApp, context,
deployedConfig));
if (jobFound) {
savepointObserver.observeSavepointStatus(flinkApp, deployedConfig);
+ if
(deployedConfig.getBoolean(OPERATOR_CLUSTER_HEALTH_CHECK_ENABLED)) {
+ clusterHealthObserver.observe(flinkApp, deployedConfig);
+ }
}
}
}
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractFlinkResourceReconciler.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractFlinkResourceReconciler.java
index c915d5fd..11bb82fd 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractFlinkResourceReconciler.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractFlinkResourceReconciler.java
@@ -399,23 +399,27 @@ public abstract class AbstractFlinkResourceReconciler<
* @param deployment FlinkDeployment object.
* @return True if recovery should be executed.
*/
- protected static boolean shouldRecoverDeployment(
- Configuration conf, FlinkDeployment deployment) {
-
- if (!jmMissingForRunningDeployment(deployment)
- || !conf.get(
-
KubernetesOperatorConfigOptions.OPERATOR_JM_DEPLOYMENT_RECOVERY_ENABLED)) {
- return false;
+ protected boolean shouldRecoverDeployment(Configuration conf,
FlinkDeployment deployment) {
+ boolean result = false;
+
+ if
(conf.get(KubernetesOperatorConfigOptions.OPERATOR_JM_DEPLOYMENT_RECOVERY_ENABLED))
{
+ LOG.debug("Checking whether jobmanager deployment needs recovery");
+
+ if (jmMissingForRunningDeployment(deployment)) {
+ LOG.debug("Jobmanager deployment is missing, trying to
recover");
+ if (FlinkUtils.isKubernetesHAActivated(conf)) {
+ LOG.debug("HA is enabled, recovering lost jobmanager
deployment");
+ result = true;
+ } else {
+ LOG.warn("Could not recover lost jobmanager deployment
without HA enabled");
+ }
+ }
}
- if (!FlinkUtils.isKubernetesHAActivated(conf)) {
- LOG.warn("Could not recover lost deployment without HA enabled");
- return false;
- }
- return true;
+ return result;
}
- private static boolean jmMissingForRunningDeployment(FlinkDeployment
deployment) {
+ private boolean jmMissingForRunningDeployment(FlinkDeployment deployment) {
var deployedJob =
ReconciliationUtils.getDeployedSpec(deployment).getJob();
return (deployedJob == null || deployedJob.getState() ==
JobState.RUNNING)
&& (deployment.getStatus().getJobManagerDeploymentStatus()
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconciler.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconciler.java
index c3798786..1c7c84d8 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconciler.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconciler.java
@@ -31,6 +31,8 @@ import
org.apache.flink.kubernetes.operator.crd.spec.UpgradeMode;
import org.apache.flink.kubernetes.operator.crd.status.FlinkDeploymentStatus;
import
org.apache.flink.kubernetes.operator.crd.status.JobManagerDeploymentStatus;
import
org.apache.flink.kubernetes.operator.exception.DeploymentFailedException;
+import org.apache.flink.kubernetes.operator.health.ClusterHealthInfo;
+import org.apache.flink.kubernetes.operator.observer.ClusterHealthEvaluator;
import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
import org.apache.flink.kubernetes.operator.service.FlinkService;
import org.apache.flink.kubernetes.operator.utils.EventRecorder;
@@ -51,6 +53,8 @@ import org.slf4j.LoggerFactory;
import java.util.Optional;
import java.util.UUID;
+import static
org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions.OPERATOR_CLUSTER_HEALTH_CHECK_ENABLED;
+
/** Reconciler Flink Application deployments. */
public class ApplicationReconciler
extends AbstractJobReconciler<FlinkDeployment, FlinkDeploymentSpec,
FlinkDeploymentStatus> {
@@ -258,10 +262,17 @@ public class ApplicationReconciler
return true;
}
- if (shouldRecoverDeployment(observeConfig, deployment)) {
+ boolean shouldRestartJobBecauseUnhealthy =
+ shouldRestartJobBecauseUnhealthy(deployment, observeConfig);
+ boolean shouldRecoverDeployment =
shouldRecoverDeployment(observeConfig, deployment);
+ if (shouldRestartJobBecauseUnhealthy || shouldRecoverDeployment) {
+ if (shouldRestartJobBecauseUnhealthy) {
+ cancelJob(deployment, ctx, UpgradeMode.LAST_STATE,
observeConfig);
+ }
recoverJmDeployment(deployment, ctx, observeConfig);
return true;
}
+
return false;
}
@@ -273,6 +284,40 @@ public class ApplicationReconciler
restoreJob(deployment, specToRecover, deployment.getStatus(), ctx,
observeConfig, true);
}
+ private boolean shouldRestartJobBecauseUnhealthy(
+ FlinkDeployment deployment, Configuration observeConfig) {
+ boolean restartNeeded = false;
+
+ if (observeConfig.getBoolean(OPERATOR_CLUSTER_HEALTH_CHECK_ENABLED)) {
+ var clusterInfo = deployment.getStatus().getClusterInfo();
+ ClusterHealthInfo clusterHealthInfo =
+
ClusterHealthEvaluator.getLastValidClusterHealthInfo(clusterInfo);
+ if (clusterHealthInfo != null) {
+ LOG.debug("Cluster info contains job health info");
+ if (!clusterHealthInfo.isHealthy()) {
+ if (deployment.getSpec().getJob().getUpgradeMode() ==
UpgradeMode.STATELESS) {
+ LOG.debug("Stateless job, recovering unhealthy
jobmanager deployment");
+ restartNeeded = true;
+ } else if
(FlinkUtils.isKubernetesHAActivated(observeConfig)) {
+ LOG.debug("HA is enabled, recovering unhealthy
jobmanager deployment");
+ restartNeeded = true;
+ } else {
+ LOG.warn(
+ "Could not recover unhealthy jobmanager
deployment without HA enabled");
+ }
+
+ if (restartNeeded) {
+
ClusterHealthEvaluator.removeLastValidClusterHealthInfo(clusterInfo);
+ }
+ }
+ } else {
+ LOG.debug("Cluster info not contains job health info, skipping
health check");
+ }
+ }
+
+ return restartNeeded;
+ }
+
@Override
@SneakyThrows
protected DeleteControl cleanupInternal(FlinkDeployment deployment,
Context<?> context) {
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 6ae12f80..832013ee 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
@@ -20,6 +20,7 @@ package org.apache.flink.kubernetes.operator.service;
import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.JobStatus;
+import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.client.program.ClusterClient;
import org.apache.flink.client.program.rest.RestClusterClient;
import org.apache.flink.configuration.CheckpointingOptions;
@@ -60,6 +61,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.metrics.JobMetricsHeaders;
import
org.apache.flink.runtime.rest.messages.job.savepoints.SavepointDisposalRequest;
import
org.apache.flink.runtime.rest.messages.job.savepoints.SavepointDisposalTriggerHeaders;
import org.apache.flink.runtime.rest.messages.job.savepoints.SavepointInfo;
@@ -860,4 +862,25 @@ public abstract class AbstractFlinkService implements
FlinkService {
throw new RuntimeException("Failed to create empty jar", e);
}
}
+
+ public Map<String, String> getMetrics(
+ Configuration conf, String jobId, List<String> metricNames) throws
Exception {
+ try (RestClusterClient<String> clusterClient =
+ (RestClusterClient<String>) getClusterClient(conf)) {
+ var jobMetricsMessageParameters =
+
JobMetricsHeaders.getInstance().getUnresolvedMessageParameters();
+
jobMetricsMessageParameters.jobPathParameter.resolve(JobID.fromHexString(jobId));
+
jobMetricsMessageParameters.metricsFilterParameter.resolve(metricNames);
+ var responseBody =
+ clusterClient
+ .sendRequest(
+ JobMetricsHeaders.getInstance(),
+ jobMetricsMessageParameters,
+ EmptyRequestBody.getInstance())
+ .get();
+ return responseBody.getMetrics().stream()
+ .map(metric -> Tuple2.of(metric.getId(),
metric.getValue()))
+ .collect(Collectors.toMap((t) -> t.f0, (t) -> t.f1));
+ }
+ }
}
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 560c5cc5..3e46351d 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
@@ -38,6 +38,7 @@ import io.fabric8.kubernetes.client.KubernetesClient;
import javax.annotation.Nullable;
import java.util.Collection;
+import java.util.List;
import java.util.Map;
import java.util.Optional;
@@ -97,4 +98,7 @@ public interface FlinkService {
default boolean scale(ObjectMeta meta, JobSpec jobSpec, Configuration
conf) {
return false;
}
+
+ Map<String, String> getMetrics(Configuration conf, String jobId,
List<String> metricNames)
+ throws Exception;
}
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/validation/DefaultValidator.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/validation/DefaultValidator.java
index 03604b59..2d51888d 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/validation/DefaultValidator.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/validation/DefaultValidator.java
@@ -156,6 +156,20 @@ public class DefaultValidator implements
FlinkResourceValidator {
return Optional.of("Kubernetes HA must be enabled for rollback
support.");
}
+ if
(conf.get(KubernetesOperatorConfigOptions.OPERATOR_CLUSTER_HEALTH_CHECK_ENABLED)
+ && !conf.get(
+
KubernetesOperatorConfigOptions.OPERATOR_JM_DEPLOYMENT_RECOVERY_ENABLED)) {
+ return Optional.of(
+ "Deployment recovery ("
+ + KubernetesOperatorConfigOptions
+ .OPERATOR_JM_DEPLOYMENT_RECOVERY_ENABLED
+ .key()
+ + ") must be enabled for job health check ("
+ +
KubernetesOperatorConfigOptions.OPERATOR_CLUSTER_HEALTH_CHECK_ENABLED
+ .key()
+ + ") support.");
+ }
+
return Optional.empty();
}
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 961830c6..a0ee0dd6 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
@@ -105,6 +105,8 @@ public class TestingFlinkService extends
AbstractFlinkService {
private final Map<String, Boolean> savepointTriggers = new HashMap<>();
private int desiredReplicas = 0;
+ private Map<String, String> metricsValues = new HashMap<>();
+
public TestingFlinkService() {
super(null, new FlinkConfigManager(new Configuration()));
}
@@ -503,4 +505,14 @@ public class TestingFlinkService extends
AbstractFlinkService {
public int getDesiredReplicas() {
return desiredReplicas;
}
+
+ public void setMetricValue(String name, String value) {
+ metricsValues.put(name, value);
+ }
+
+ @Override
+ public Map<String, String> getMetrics(
+ Configuration conf, String jobId, List<String> metricNames) {
+ return metricsValues;
+ }
}
diff --git
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/UnhealthyDeploymentRestartTest.java
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/UnhealthyDeploymentRestartTest.java
new file mode 100644
index 00000000..e7a35dbc
--- /dev/null
+++
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/UnhealthyDeploymentRestartTest.java
@@ -0,0 +1,117 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.kubernetes.operator.controller;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.kubernetes.operator.TestUtils;
+import org.apache.flink.kubernetes.operator.TestingFlinkService;
+import org.apache.flink.kubernetes.operator.config.FlinkConfigManager;
+import org.apache.flink.kubernetes.operator.crd.FlinkDeployment;
+import org.apache.flink.kubernetes.operator.crd.spec.FlinkVersion;
+import org.apache.flink.kubernetes.operator.crd.spec.UpgradeMode;
+import
org.apache.flink.kubernetes.operator.crd.status.JobManagerDeploymentStatus;
+
+import io.fabric8.kubernetes.client.KubernetesClient;
+import io.fabric8.kubernetes.client.server.mock.EnableKubernetesMockClient;
+import io.javaoperatorsdk.operator.api.reconciler.Context;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.stream.Stream;
+
+import static
org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions.OPERATOR_CLUSTER_HEALTH_CHECK_ENABLED;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.params.provider.Arguments.arguments;
+
+/** @link Unhealthy deployment restart tests */
+@EnableKubernetesMockClient(crud = true)
+public class UnhealthyDeploymentRestartTest {
+
+ private static final String NUM_RESTARTS_METRIC_NAME = "numRestarts";
+
+ private FlinkConfigManager configManager;
+
+ private TestingFlinkService flinkService;
+ private Context<FlinkDeployment> context;
+ private TestingFlinkDeploymentController testController;
+
+ private KubernetesClient kubernetesClient;
+
+ @BeforeEach
+ public void setup() {
+ var configuration = new Configuration();
+ configuration.set(OPERATOR_CLUSTER_HEALTH_CHECK_ENABLED, true);
+ configManager = new FlinkConfigManager(configuration);
+ flinkService = new TestingFlinkService(kubernetesClient);
+ context = flinkService.getContext();
+ testController =
+ new TestingFlinkDeploymentController(configManager,
kubernetesClient, flinkService);
+
kubernetesClient.resource(TestUtils.buildApplicationCluster()).createOrReplace();
+ }
+
+ @ParameterizedTest
+ @MethodSource("applicationTestParams")
+ public void verifyApplicationUnhealthyJmRecovery(
+ FlinkVersion flinkVersion, UpgradeMode upgradeMode) throws
Exception {
+ FlinkDeployment appCluster =
TestUtils.buildApplicationCluster(flinkVersion);
+ appCluster.getSpec().getJob().setUpgradeMode(upgradeMode);
+
+ // Start a healthy deployment
+ flinkService.setMetricValue(NUM_RESTARTS_METRIC_NAME, "0");
+ testController.reconcile(appCluster, context);
+ testController.reconcile(appCluster, context);
+ testController.reconcile(appCluster, context);
+ assertEquals(
+ JobManagerDeploymentStatus.READY,
+ appCluster.getStatus().getJobManagerDeploymentStatus());
+ assertEquals("RUNNING",
appCluster.getStatus().getJobStatus().getState());
+
+ // Make deployment unhealthy
+ flinkService.setMetricValue(NUM_RESTARTS_METRIC_NAME, "100");
+ testController.reconcile(appCluster, context);
+ assertEquals(
+ JobManagerDeploymentStatus.DEPLOYING,
+ appCluster.getStatus().getJobManagerDeploymentStatus());
+
+ // After restart the deployment is healthy again
+ flinkService.setMetricValue(NUM_RESTARTS_METRIC_NAME, "0");
+ testController.reconcile(appCluster, context);
+ testController.reconcile(appCluster, context);
+ assertEquals(
+ JobManagerDeploymentStatus.READY,
+ appCluster.getStatus().getJobManagerDeploymentStatus());
+ assertEquals("RUNNING",
appCluster.getStatus().getJobStatus().getState());
+ assertEquals(
+ appCluster.getSpec(),
+
appCluster.getStatus().getReconciliationStatus().deserializeLastReconciledSpec());
+ }
+
+ private static Stream<Arguments> applicationTestParams() {
+ List<Arguments> args = new ArrayList<>();
+ for (FlinkVersion version : FlinkVersion.values()) {
+ for (UpgradeMode upgradeMode : UpgradeMode.values()) {
+ args.add(arguments(version, upgradeMode));
+ }
+ }
+ return args.stream();
+ }
+}
diff --git
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/health/ClusterHealthInfoTest.java
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/health/ClusterHealthInfoTest.java
new file mode 100644
index 00000000..2b6707c0
--- /dev/null
+++
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/health/ClusterHealthInfoTest.java
@@ -0,0 +1,53 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.kubernetes.operator.health;
+
+import org.junit.jupiter.api.Test;
+
+import java.time.Clock;
+import java.time.ZoneId;
+
+import static java.time.Instant.ofEpochSecond;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class ClusterHealthInfoTest {
+
+ @Test
+ public void isValidShouldReturnFalseWhenTimestampIsZero() {
+ var clock = Clock.fixed(ofEpochSecond(0), ZoneId.systemDefault());
+ assertFalse(ClusterHealthInfo.isValid(ClusterHealthInfo.of(clock, 0)));
+ }
+
+ @Test
+ public void isValidShouldReturnTrueWhenTimestampIsNonzero() {
+ var clock = Clock.fixed(ofEpochSecond(1), ZoneId.systemDefault());
+ assertTrue(ClusterHealthInfo.isValid(ClusterHealthInfo.of(clock, 0)));
+ }
+
+ @Test
+ public void serializationRoundTrip() {
+ var clock = Clock.fixed(ofEpochSecond(123), ZoneId.systemDefault());
+ var clusterHealthInfo = ClusterHealthInfo.of(clock, 456);
+ var clusterHealthInfoJson =
ClusterHealthInfo.serialize(clusterHealthInfo);
+
+ var clusterHealthInfoFromJson =
ClusterHealthInfo.deserialize(clusterHealthInfoJson);
+ assertEquals(clusterHealthInfo, clusterHealthInfoFromJson);
+ }
+}
diff --git
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/ClusterHealthEvaluatorTest.java
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/ClusterHealthEvaluatorTest.java
new file mode 100644
index 00000000..e6cf9a63
--- /dev/null
+++
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/ClusterHealthEvaluatorTest.java
@@ -0,0 +1,202 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.kubernetes.operator.observer;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.kubernetes.operator.health.ClusterHealthInfo;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.time.Clock;
+import java.time.Duration;
+import java.time.Instant;
+import java.time.ZoneId;
+import java.time.temporal.ChronoUnit;
+import java.util.HashMap;
+import java.util.Map;
+
+import static java.time.Instant.ofEpochMilli;
+import static java.time.Instant.ofEpochSecond;
+import static
org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions.OPERATOR_CLUSTER_HEALTH_CHECK_RESTARTS_THRESHOLD;
+import static
org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions.OPERATOR_CLUSTER_HEALTH_CHECK_RESTARTS_WINDOW;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+/** Tests for {@link ClusterHealthEvaluator}. */
+class ClusterHealthEvaluatorTest {
+
+ private Configuration configuration;
+ private Map<String, String> clusterInfo;
+ private ClusterHealthEvaluator clusterHealthEvaluator;
+ private final Instant invalidInstant = ofEpochMilli(0);
+ private final Instant validInstant = ofEpochMilli(1);
+ private ClusterHealthInfo invalidClusterHealthInfo;
+
+ @BeforeEach
+ public void beforeEach() {
+ configuration = new Configuration();
+
+ clusterInfo = new HashMap<>();
+
+ var now = Clock.fixed(ofEpochSecond(120), ZoneId.systemDefault());
+ clusterHealthEvaluator = new ClusterHealthEvaluator(now);
+
+ var clock = Clock.fixed(invalidInstant, ZoneId.systemDefault());
+ invalidClusterHealthInfo = ClusterHealthInfo.of(clock, 0);
+ }
+
+ @Test
+ public void evaluateShouldNotSetLastStateWhenInvalidObserved() {
+ clusterHealthEvaluator.evaluate(configuration, clusterInfo,
invalidClusterHealthInfo);
+
assertNull(ClusterHealthEvaluator.getLastValidClusterHealthInfo(clusterInfo));
+ }
+
+ @Test
+ public void evaluateShouldSetLastStateWhenValidObserved() {
+ var observedClusterHealthInfo = createClusterHealthInfo(validInstant,
0);
+
+ clusterHealthEvaluator.evaluate(configuration, clusterInfo,
observedClusterHealthInfo);
+ assertEquals(
+ observedClusterHealthInfo,
+
ClusterHealthEvaluator.getLastValidClusterHealthInfo(clusterInfo));
+ }
+
+ @Test
+ public void evaluateShouldThrowExceptionWhenObservedTimestampIsOld() {
+ var observedClusterHealthInfo1 = createClusterHealthInfo(validInstant,
0);
+ var observedClusterHealthInfo2 =
createClusterHealthInfo(validInstant.plusMillis(100), 0);
+
+ ClusterHealthEvaluator.setLastValidClusterHealthInfo(
+ clusterInfo, observedClusterHealthInfo2);
+ assertThrows(
+ IllegalStateException.class,
+ () ->
+ clusterHealthEvaluator.evaluate(
+ configuration, clusterInfo,
observedClusterHealthInfo1));
+ }
+
+ @Test
+ public void evaluateShouldOverwriteLastStateWhenRestartCountIsLess() {
+ var observedClusterHealthInfo1 = createClusterHealthInfo(validInstant,
1);
+ var observedClusterHealthInfo2 = createClusterHealthInfo(validInstant,
0);
+
+ clusterHealthEvaluator.evaluate(configuration, clusterInfo,
observedClusterHealthInfo1);
+ assertEquals(
+ observedClusterHealthInfo1,
+
ClusterHealthEvaluator.getLastValidClusterHealthInfo(clusterInfo));
+ clusterHealthEvaluator.evaluate(configuration, clusterInfo,
observedClusterHealthInfo2);
+ assertEquals(
+ observedClusterHealthInfo2,
+
ClusterHealthEvaluator.getLastValidClusterHealthInfo(clusterInfo));
+ }
+
+ @Test
+ public void evaluateShouldNotOverwriteLastStateWhenTimestampIsInWindow() {
+ configuration.set(OPERATOR_CLUSTER_HEALTH_CHECK_RESTARTS_WINDOW,
Duration.ofMinutes(2));
+ var observedClusterHealthInfo1 = createClusterHealthInfo(validInstant,
0);
+ var observedClusterHealthInfo2 =
+ createClusterHealthInfo(validInstant.plus(1,
ChronoUnit.MINUTES), 0);
+
+ ClusterHealthEvaluator.setLastValidClusterHealthInfo(
+ clusterInfo, observedClusterHealthInfo1);
+ clusterHealthEvaluator.evaluate(configuration, clusterInfo,
observedClusterHealthInfo2);
+ assertEquals(
+ observedClusterHealthInfo1,
+
ClusterHealthEvaluator.getLastValidClusterHealthInfo(clusterInfo));
+ }
+
+ @Test
+ public void evaluateShouldOverwriteLastStateWhenTimestampIsOutOfWindow() {
+ configuration.set(OPERATOR_CLUSTER_HEALTH_CHECK_RESTARTS_WINDOW,
Duration.ofMinutes(1));
+ var observedClusterHealthInfo1 = createClusterHealthInfo(validInstant,
0);
+ var observedClusterHealthInfo2 =
+ createClusterHealthInfo(validInstant.plus(1,
ChronoUnit.MINUTES), 0);
+
+ ClusterHealthEvaluator.setLastValidClusterHealthInfo(
+ clusterInfo, observedClusterHealthInfo1);
+ clusterHealthEvaluator.evaluate(configuration, clusterInfo,
observedClusterHealthInfo2);
+ assertEquals(
+ observedClusterHealthInfo2,
+
ClusterHealthEvaluator.getLastValidClusterHealthInfo(clusterInfo));
+ }
+
+ @Test
+ public void evaluateShouldMarkClusterHealthyWhenNoPreviousState() {
+ var observedClusterHealthInfo = createClusterHealthInfo(validInstant,
1);
+
+ clusterHealthEvaluator.evaluate(configuration, clusterInfo,
observedClusterHealthInfo);
+ assertClusterHealthIs(true);
+ }
+
+ @Test
+ public void evaluateShouldMarkClusterHealthyWhenThresholdNotHit() {
+ configuration.set(OPERATOR_CLUSTER_HEALTH_CHECK_RESTARTS_WINDOW,
Duration.ofMinutes(5));
+ configuration.set(OPERATOR_CLUSTER_HEALTH_CHECK_RESTARTS_THRESHOLD,
100);
+ ClusterHealthInfo observedClusterHealthInfo1 =
createClusterHealthInfo(validInstant, 0);
+ var observedClusterHealthInfo2 =
+ createClusterHealthInfo(validInstant.plus(1,
ChronoUnit.MINUTES), 100);
+
+ ClusterHealthEvaluator.setLastValidClusterHealthInfo(
+ clusterInfo, observedClusterHealthInfo1);
+ clusterHealthEvaluator.evaluate(configuration, clusterInfo,
observedClusterHealthInfo2);
+ assertClusterHealthIs(true);
+ }
+
+ @Test
+ public void
evaluateShouldMarkClusterUnhealthyWhenThresholdHitImmediately() {
+ configuration.set(OPERATOR_CLUSTER_HEALTH_CHECK_RESTARTS_WINDOW,
Duration.ofMinutes(5));
+ configuration.set(OPERATOR_CLUSTER_HEALTH_CHECK_RESTARTS_THRESHOLD,
100);
+ ClusterHealthInfo observedClusterHealthInfo1 =
createClusterHealthInfo(validInstant, 0);
+ var observedClusterHealthInfo2 =
+ createClusterHealthInfo(validInstant.plus(1,
ChronoUnit.MINUTES), 101);
+
+ ClusterHealthEvaluator.setLastValidClusterHealthInfo(
+ clusterInfo, observedClusterHealthInfo1);
+ clusterHealthEvaluator.evaluate(configuration, clusterInfo,
observedClusterHealthInfo2);
+ assertClusterHealthIs(false);
+ }
+
+ @Test
+ public void evaluateShouldMarkClusterUnhealthyWhenThresholdHitInAverage() {
+ configuration.set(OPERATOR_CLUSTER_HEALTH_CHECK_RESTARTS_WINDOW,
Duration.ofMinutes(5));
+ configuration.set(OPERATOR_CLUSTER_HEALTH_CHECK_RESTARTS_THRESHOLD,
100);
+ var observedClusterHealthInfo1 = createClusterHealthInfo(validInstant,
0);
+ var observedClusterHealthInfo2 =
+ createClusterHealthInfo(validInstant.plus(6,
ChronoUnit.MINUTES), 122);
+
+ ClusterHealthEvaluator.setLastValidClusterHealthInfo(
+ clusterInfo, observedClusterHealthInfo1);
+ clusterHealthEvaluator.evaluate(configuration, clusterInfo,
observedClusterHealthInfo2);
+ assertClusterHealthIs(false);
+ }
+
+ private ClusterHealthInfo createClusterHealthInfo(Instant instant, int
numRestarts) {
+ var clock = Clock.fixed(instant, ZoneId.systemDefault());
+ return ClusterHealthInfo.of(clock, numRestarts);
+ }
+
+ private void assertClusterHealthIs(boolean healthy) {
+ var lastValidClusterHealthInfo =
+
ClusterHealthEvaluator.getLastValidClusterHealthInfo(clusterInfo);
+ assertNotNull(lastValidClusterHealthInfo);
+ assertEquals(healthy, lastValidClusterHealthInfo.isHealthy());
+ }
+}
diff --git
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/validation/DefaultValidatorTest.java
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/validation/DefaultValidatorTest.java
index 0b2c3361..0a03da37 100644
---
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/validation/DefaultValidatorTest.java
+++
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/validation/DefaultValidatorTest.java
@@ -179,6 +179,24 @@ public class DefaultValidatorTest {
KubernetesConfigOptions.NAMESPACE.key(), "myns")),
"Forbidden Flink config key");
+ testError(
+ dep ->
+ dep.getSpec()
+ .setFlinkConfiguration(
+ Map.of(
+ KubernetesOperatorConfigOptions
+
.OPERATOR_CLUSTER_HEALTH_CHECK_ENABLED
+ .key(),
+ "true",
+ KubernetesOperatorConfigOptions
+
.OPERATOR_JM_DEPLOYMENT_RECOVERY_ENABLED
+ .key(),
+ "false")),
+ "Deployment recovery ("
+ +
KubernetesOperatorConfigOptions.OPERATOR_JM_DEPLOYMENT_RECOVERY_ENABLED
+ .key()
+ + ") must be enabled");
+
// Test log config validation
testSuccess(
dep ->
