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/02/07 12:22:18 UTC
[flink-kubernetes-operator] branch main updated (f0578a57 -> 12986505)
This is an automated email from the ASF dual-hosted git repository.
gyfora pushed a change to branch main
in repository https://gitbox.apache.org/repos/asf/flink-kubernetes-operator.git
from f0578a57 [FLINK-30845] HttpArtifactFetcher: Fix filename
new ca231195 [FLINK-30769] Use Operator RuntimeInfo in health probe
new 12986505 [FLINK-30779] Bump josdk version to 4.2.6
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../kubernetes_operator_config_configuration.html | 6 ++
.../generated/system_advanced_section.html | 6 ++
.../flink/kubernetes/operator/FlinkOperator.java | 20 ++++++-
.../config/KubernetesOperatorConfigOptions.java | 10 ++++
.../operator/controller/FlinkControllerConfig.java | 46 ---------------
.../kubernetes/operator/health/HealthProbe.java | 25 ++++++++-
.../src/main/resources/META-INF/NOTICE | 4 +-
...HealthServiceTest.java => HealthProbeTest.java} | 65 ++++++++++++++++++----
pom.xml | 2 +-
9 files changed, 122 insertions(+), 62 deletions(-)
delete mode 100644 flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkControllerConfig.java
rename flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/{health/OperatorHealthServiceTest.java => HealthProbeTest.java} (50%)
[flink-kubernetes-operator] 01/02: [FLINK-30769] Use Operator RuntimeInfo in health probe
Posted by gy...@apache.org.
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 ca231195f05578947cb4e8fb591914e2dd1d4fed
Author: Gyula Fora <g_...@apple.com>
AuthorDate: Mon Jan 23 14:37:35 2023 +0100
[FLINK-30769] Use Operator RuntimeInfo in health probe
---
.../kubernetes_operator_config_configuration.html | 6 ++
.../generated/system_advanced_section.html | 6 ++
.../flink/kubernetes/operator/FlinkOperator.java | 20 ++++++-
.../config/KubernetesOperatorConfigOptions.java | 10 ++++
.../kubernetes/operator/health/HealthProbe.java | 25 ++++++++-
...HealthServiceTest.java => HealthProbeTest.java} | 65 ++++++++++++++++++----
6 files changed, 119 insertions(+), 13 deletions(-)
diff --git a/docs/layouts/shortcodes/generated/kubernetes_operator_config_configuration.html b/docs/layouts/shortcodes/generated/kubernetes_operator_config_configuration.html
index 69be9ba9..ddb0bed3 100644
--- a/docs/layouts/shortcodes/generated/kubernetes_operator_config_configuration.html
+++ b/docs/layouts/shortcodes/generated/kubernetes_operator_config_configuration.html
@@ -290,6 +290,12 @@
<td>Boolean</td>
<td>Whether informer errors should stop operator startup. If false, the startup will ignore recoverable errors, caused for example by RBAC issues and will retry periodically.</td>
</tr>
+ <tr>
+ <td><h5>kubernetes.operator.termination.timeout</h5></td>
+ <td style="word-wrap: break-word;">10 s</td>
+ <td>Duration</td>
+ <td>Operator shutdown timeout before reconciliation threads are killed.</td>
+ </tr>
<tr>
<td><h5>kubernetes.operator.user.artifacts.base.dir</h5></td>
<td style="word-wrap: break-word;">"/opt/flink/artifacts"</td>
diff --git a/docs/layouts/shortcodes/generated/system_advanced_section.html b/docs/layouts/shortcodes/generated/system_advanced_section.html
index 1a89d3d3..9a17c745 100644
--- a/docs/layouts/shortcodes/generated/system_advanced_section.html
+++ b/docs/layouts/shortcodes/generated/system_advanced_section.html
@@ -80,5 +80,11 @@
<td>Boolean</td>
<td>Whether informer errors should stop operator startup. If false, the startup will ignore recoverable errors, caused for example by RBAC issues and will retry periodically.</td>
</tr>
+ <tr>
+ <td><h5>kubernetes.operator.termination.timeout</h5></td>
+ <td style="word-wrap: break-word;">10 s</td>
+ <td>Duration</td>
+ <td>Operator shutdown timeout before reconciliation threads are killed.</td>
+ </tr>
</tbody>
</table>
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/FlinkOperator.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/FlinkOperator.java
index 25397faa..08bcb653 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/FlinkOperator.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/FlinkOperator.java
@@ -27,6 +27,7 @@ import org.apache.flink.kubernetes.operator.config.FlinkConfigManager;
import org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions;
import org.apache.flink.kubernetes.operator.controller.FlinkDeploymentController;
import org.apache.flink.kubernetes.operator.controller.FlinkSessionJobController;
+import org.apache.flink.kubernetes.operator.health.HealthProbe;
import org.apache.flink.kubernetes.operator.health.OperatorHealthService;
import org.apache.flink.kubernetes.operator.listener.ListenerUtils;
import org.apache.flink.kubernetes.operator.metrics.KubernetesOperatorMetricGroup;
@@ -87,7 +88,7 @@ public class FlinkOperator {
this.client =
KubernetesClientUtils.getKubernetesClient(
configManager.getOperatorConfiguration(), this.metricGroup);
- this.operator = new Operator(client, this::overrideOperatorConfigs);
+ this.operator = createOperator();
this.ctxFactory = new FlinkResourceContextFactory(client, configManager, metricGroup);
this.validators = ValidatorUtils.discoverValidators(configManager);
this.listeners = ListenerUtils.discoverListeners(configManager);
@@ -96,6 +97,11 @@ public class FlinkOperator {
this.operatorHealthService = OperatorHealthService.fromConfig(configManager);
}
+ @VisibleForTesting
+ Operator createOperator() {
+ return new Operator(client, this::overrideOperatorConfigs);
+ }
+
private void handleNamespaceChanges(Set<String> namespaces) {
registeredControllers.forEach(
controller -> {
@@ -121,6 +127,13 @@ public class FlinkOperator {
overrider.withMetrics(new OperatorJosdkMetrics(metricGroup, configManager));
}
+ overrider.withTerminationTimeoutSeconds(
+ (int)
+ configManager
+ .getDefaultConfig()
+ .get(KubernetesOperatorConfigOptions.OPERATOR_TERMINATION_TIMEOUT)
+ .toSeconds());
+
overrider.withStopOnInformerErrorDuringStartup(
configManager
.getDefaultConfig()
@@ -198,11 +211,16 @@ public class FlinkOperator {
operator.installShutdownHook();
operator.start();
if (operatorHealthService != null) {
+ HealthProbe.INSTANCE.setRuntimeInfo(operator.getRuntimeInfo());
Runtime.getRuntime().addShutdownHook(new Thread(operatorHealthService::stop));
operatorHealthService.start();
}
}
+ public void stop() {
+ operator.stop();
+ }
+
public static void main(String... args) {
EnvUtils.logEnvironmentInfo(LOG, "Flink Kubernetes Operator", args);
new FlinkOperator(null).run();
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 ea1c6f18..2b227c8b 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
@@ -355,6 +355,16 @@ public class KubernetesOperatorConfigOptions {
.withDescription(
"Whether informer errors should stop operator startup. If false, the startup will ignore recoverable errors, caused for example by RBAC issues and will retry periodically.");
+ @Documentation.Section(SECTION_ADVANCED)
+ public static final ConfigOption<Duration> OPERATOR_TERMINATION_TIMEOUT =
+ operatorConfig("termination.timeout")
+ .durationType()
+ .defaultValue(
+ Duration.ofSeconds(
+ ConfigurationService.DEFAULT_TERMINATION_TIMEOUT_SECONDS))
+ .withDescription(
+ "Operator shutdown timeout before reconciliation threads are killed.");
+
@Documentation.Section(SECTION_DYNAMIC)
public static final ConfigOption<Boolean> OPERATOR_CLUSTER_HEALTH_CHECK_ENABLED =
operatorConfig("cluster.health-check.enabled")
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/health/HealthProbe.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/health/HealthProbe.java
index bea1d8e3..1745cbc7 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/health/HealthProbe.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/health/HealthProbe.java
@@ -17,19 +17,42 @@
package org.apache.flink.kubernetes.operator.health;
+import io.javaoperatorsdk.operator.RuntimeInfo;
+import lombok.Getter;
+import lombok.Setter;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import java.util.concurrent.atomic.AtomicBoolean;
/** Flink operator health probe. */
public enum HealthProbe {
INSTANCE;
+ private static final Logger LOG = LoggerFactory.getLogger(HealthProbe.class);
+
private final AtomicBoolean isHealthy = new AtomicBoolean(true);
+ @Setter @Getter private RuntimeInfo runtimeInfo;
+
public void markUnhealthy() {
isHealthy.set(false);
}
public boolean isHealthy() {
- return isHealthy.get();
+ if (!isHealthy.get()) {
+ return false;
+ }
+
+ if (runtimeInfo != null) {
+ LOG.debug("Checking operator health");
+ if (runtimeInfo.allEventSourcesAreHealthy()) {
+ return runtimeInfo.isStarted();
+ } else {
+ LOG.error("Unhealthy event sources: {}", runtimeInfo.unhealthyEventSources());
+ return false;
+ }
+ }
+ return true;
}
}
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/health/OperatorHealthServiceTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/HealthProbeTest.java
similarity index 50%
rename from flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/health/OperatorHealthServiceTest.java
rename to flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/HealthProbeTest.java
index fea45c6f..f619a612 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/health/OperatorHealthServiceTest.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/HealthProbeTest.java
@@ -15,40 +15,83 @@
* limitations under the License.
*/
-package org.apache.flink.kubernetes.operator.health;
+package org.apache.flink.kubernetes.operator;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.kubernetes.operator.config.FlinkConfigManager;
import org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions;
+import org.apache.flink.kubernetes.operator.health.HealthProbe;
import org.apache.flink.util.NetUtils;
+import io.fabric8.kubernetes.client.KubernetesClient;
+import io.fabric8.kubernetes.client.server.mock.EnableKubernetesMockClient;
+import io.javaoperatorsdk.operator.Operator;
+import io.javaoperatorsdk.operator.RuntimeInfo;
import org.junit.jupiter.api.Test;
import java.net.HttpURLConnection;
import java.net.URL;
+import java.util.concurrent.atomic.AtomicBoolean;
import static org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus.OK;
import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
-/** Test for Health probe and endpoint logic. */
-public class OperatorHealthServiceTest {
+/** @link Health probe unit tests */
+@EnableKubernetesMockClient(crud = true)
+public class HealthProbeTest {
+ KubernetesClient client;
@Test
- public void testHealthService() throws Exception {
+ public void testHealthProbeEndpoint() throws Exception {
try (var port = NetUtils.getAvailablePort()) {
var conf = new Configuration();
conf.set(KubernetesOperatorConfigOptions.OPERATOR_HEALTH_PROBE_PORT, port.getPort());
- var healthService = new OperatorHealthService(new FlinkConfigManager(conf));
- healthService.start();
- assertTrue(callHealthEndpoint(conf));
- HealthProbe.INSTANCE.markUnhealthy();
- assertFalse(callHealthEndpoint(conf));
- healthService.stop();
+ FlinkOperator operator =
+ new FlinkOperator(conf) {
+ Operator createOperator() {
+ return new Operator(client);
+ }
+ };
+ try {
+ operator.run();
+ assertTrue(callHealthEndpoint(conf));
+ assertNotNull(HealthProbe.INSTANCE.getRuntimeInfo());
+ } finally {
+ operator.stop();
+ assertFalse(callHealthEndpoint(conf));
+ }
}
}
+ @Test
+ public void testHealthProbe() {
+ var isRunning = new AtomicBoolean(false);
+ var isHealthy = new AtomicBoolean(false);
+
+ HealthProbe.INSTANCE.setRuntimeInfo(
+ new RuntimeInfo(new Operator(client)) {
+ @Override
+ public boolean isStarted() {
+ return isRunning.get();
+ }
+
+ @Override
+ public boolean allEventSourcesAreHealthy() {
+ return isHealthy.get();
+ }
+ });
+
+ assertFalse(HealthProbe.INSTANCE.isHealthy());
+ isRunning.set(true);
+ assertFalse(HealthProbe.INSTANCE.isHealthy());
+ isHealthy.set(true);
+ assertTrue(HealthProbe.INSTANCE.isHealthy());
+ isRunning.set(false);
+ assertFalse(HealthProbe.INSTANCE.isHealthy());
+ }
+
private boolean callHealthEndpoint(Configuration conf) throws Exception {
URL u =
new URL(
[flink-kubernetes-operator] 02/02: [FLINK-30779] Bump josdk version to 4.2.6
Posted by gy...@apache.org.
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 12986505cd732c707187e81f01e86196028d1399
Author: Gyula Fora <g_...@apple.com>
AuthorDate: Fri Jan 20 12:53:38 2023 +0100
[FLINK-30779] Bump josdk version to 4.2.6
---
.../operator/controller/FlinkControllerConfig.java | 46 ----------------------
.../src/main/resources/META-INF/NOTICE | 4 +-
pom.xml | 2 +-
3 files changed, 3 insertions(+), 49 deletions(-)
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkControllerConfig.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkControllerConfig.java
deleted file mode 100644
index 044780b5..00000000
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkControllerConfig.java
+++ /dev/null
@@ -1,46 +0,0 @@
-/*
- * 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 io.fabric8.kubernetes.api.model.HasMetadata;
-import io.javaoperatorsdk.operator.api.reconciler.Reconciler;
-import io.javaoperatorsdk.operator.config.runtime.AnnotationControllerConfiguration;
-
-import java.util.Set;
-
-/** Custom config for {@link FlinkDeploymentController}. */
-public class FlinkControllerConfig<CR extends HasMetadata>
- extends AnnotationControllerConfiguration<CR> {
-
- private final Set<String> watchedNamespaces;
-
- public FlinkControllerConfig(Reconciler<CR> reconciler, Set<String> watchedNamespaces) {
- super(reconciler);
- this.watchedNamespaces = watchedNamespaces;
- }
-
- @Override
- public Set<String> getNamespaces() {
- return watchedNamespaces;
- }
-
- @Override
- public Set<String> getEffectiveNamespaces() {
- return watchedNamespaces;
- }
-}
diff --git a/flink-kubernetes-operator/src/main/resources/META-INF/NOTICE b/flink-kubernetes-operator/src/main/resources/META-INF/NOTICE
index 2ca555de..8e60f3bc 100644
--- a/flink-kubernetes-operator/src/main/resources/META-INF/NOTICE
+++ b/flink-kubernetes-operator/src/main/resources/META-INF/NOTICE
@@ -45,9 +45,9 @@ This project bundles the following dependencies under the Apache Software Licens
- io.fabric8:kubernetes-model-scheduling:jar:6.2.0
- io.fabric8:kubernetes-model-storageclass:jar:6.2.0
- io.fabric8:zjsonpatch:jar:0.3.0
-- io.javaoperatorsdk:operator-framework-core:jar:4.1.0
+- io.javaoperatorsdk:operator-framework-core:jar:4.2.6
- io.javaoperatorsdk:operator-framework-framework-core:jar:0.2.0
-- io.javaoperatorsdk:operator-framework:jar:4.1.0
+- io.javaoperatorsdk:operator-framework:jar:4.2.6
- org.apache.commons:commons-compress:1.21
- org.apache.commons:commons-lang3:3.12.0
- org.apache.commons:commons-math3:3.6.1
diff --git a/pom.xml b/pom.xml
index f4013896..897e3542 100644
--- a/pom.xml
+++ b/pom.xml
@@ -72,7 +72,7 @@ under the License.
<maven-javadoc-plugin.version>3.3.2</maven-javadoc-plugin.version>
<git-commit-id-maven-plugin.version>5.0.0</git-commit-id-maven-plugin.version>
- <operator.sdk.version>4.1.1</operator.sdk.version>
+ <operator.sdk.version>4.2.6</operator.sdk.version>
<operator.sdk.admission-controller.version>0.2.0</operator.sdk.admission-controller.version>
<fabric8.version>6.2.0</fabric8.version>