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/03/29 15:22:50 UTC
[flink-kubernetes-operator] branch main updated (9690d89c -> 3899050a)
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 9690d89c [FLINK-31614] Upgrade autoscaling example to 1.17.0 image
new c41771a2 [hotfix] Remove unnecessary client creation
new 3899050a [FLINK-31218] Improve health probe to only detect newly unhealthy informers
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/health/HealthProbe.java | 47 +++++--
.../operator/health/InformerHealthSummary.java | 61 +++++++++
.../operator/health/InformerIdentifier.java | 20 +--
.../operator/service/AbstractFlinkService.java | 7 +-
.../operator/health/HealthProbeTest.java | 140 +++++++++++++++++++--
5 files changed, 233 insertions(+), 42 deletions(-)
create mode 100644 flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/health/InformerHealthSummary.java
copy flink-kubernetes-operator-autoscaler/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/topology/VertexInfo.java => flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/health/InformerIdentifier.java (73%)
[flink-kubernetes-operator] 01/02: [hotfix] Remove unnecessary client creation
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 c41771a24c1db03b8d5513cc812ff3fb0ec43937
Author: Gyula Fora <g_...@apple.com>
AuthorDate: Fri Feb 10 08:56:06 2023 +0100
[hotfix] Remove unnecessary client creation
---
.../flink/kubernetes/operator/service/AbstractFlinkService.java | 7 +------
1 file changed, 1 insertion(+), 6 deletions(-)
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 97262daf..27578c04 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
@@ -640,12 +640,7 @@ public abstract class AbstractFlinkService implements FlinkService {
@Override
public PodList getJmPodList(FlinkDeployment deployment, Configuration conf) {
final String namespace = conf.getString(KubernetesConfigOptions.NAMESPACE);
- final String clusterId;
- try (ClusterClient<String> clusterClient = getClusterClient(conf)) {
- clusterId = clusterClient.getClusterId();
- } catch (Exception ex) {
- throw new RuntimeException(ex);
- }
+ final String clusterId = conf.getString(KubernetesConfigOptions.CLUSTER_ID);
return getJmPodList(namespace, clusterId);
}
[flink-kubernetes-operator] 02/02: [FLINK-31218] Improve health probe to only detect newly unhealthy informers
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 3899050ae09ee0590d9810a858703d8d6a9ca602
Author: Gyula Fora <g_...@apple.com>
AuthorDate: Thu Feb 16 12:04:27 2023 +0100
[FLINK-31218] Improve health probe to only detect newly unhealthy informers
---
.../kubernetes/operator/health/HealthProbe.java | 47 +++++--
.../operator/health/InformerHealthSummary.java | 61 +++++++++
.../operator/health/InformerIdentifier.java | 29 +++++
.../operator/health/HealthProbeTest.java | 140 +++++++++++++++++++--
4 files changed, 255 insertions(+), 22 deletions(-)
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 8ce574fa..1b5d561f 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
@@ -19,13 +19,12 @@ 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.ArrayList;
import java.util.List;
-import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.Set;
/** Flink operator health probe. */
public enum HealthProbe {
@@ -33,9 +32,9 @@ public enum HealthProbe {
private static final Logger LOG = LoggerFactory.getLogger(HealthProbe.class);
- private final AtomicBoolean isHealthy = new AtomicBoolean(true);
+ @Getter private RuntimeInfo runtimeInfo;
- @Setter @Getter private RuntimeInfo runtimeInfo;
+ private InformerHealthSummary previousInformerHealthSummary;
private final List<CanaryResourceManager<?>> canaryResourceManagers = new ArrayList<>();
@@ -43,20 +42,29 @@ public enum HealthProbe {
canaryResourceManagers.add(canaryResourceManager);
}
- public boolean isHealthy() {
- if (!isHealthy.get()) {
- return false;
- }
+ public void setRuntimeInfo(RuntimeInfo runtimeInfo) {
+ this.runtimeInfo = runtimeInfo;
+ previousInformerHealthSummary = InformerHealthSummary.fromRuntimeInfo(runtimeInfo);
+ LOG.info(
+ "Initially unhealthy informers: {}",
+ previousInformerHealthSummary.getUnhealthyInformers());
+ }
+ public boolean isHealthy() {
if (runtimeInfo != null) {
- LOG.debug("Checking operator health");
- if (!runtimeInfo.allEventSourcesAreHealthy()) {
- LOG.error("Unhealthy event sources: {}", runtimeInfo.unhealthyEventSources());
+ LOG.debug("Checking event source health");
+ var healthSummary = InformerHealthSummary.fromRuntimeInfo(runtimeInfo);
+ if (!healthSummary.isAnyHealthy()) {
+ LOG.error("All informers are unhealthy");
+ return false;
+ } else if (anyInformerBecameUnhealthy(healthSummary.getUnhealthyInformers())) {
return false;
+ } else {
+ previousInformerHealthSummary = healthSummary;
}
if (!runtimeInfo.isStarted()) {
- LOG.error("Operator not running");
+ LOG.error("Operator is not running");
return false;
}
}
@@ -67,6 +75,21 @@ public enum HealthProbe {
return false;
}
}
+
return true;
}
+
+ private boolean anyInformerBecameUnhealthy(Set<InformerIdentifier> unhealthyInformers) {
+ boolean unhealthy = false;
+ for (InformerIdentifier unhealthyInformer : unhealthyInformers) {
+ if (!previousInformerHealthSummary
+ .getUnhealthyInformers()
+ .contains(unhealthyInformer)) {
+ LOG.error("Informer became unhealthy: {}", unhealthyInformer);
+ unhealthy = true;
+ }
+ }
+
+ return unhealthy;
+ }
}
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/health/InformerHealthSummary.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/health/InformerHealthSummary.java
new file mode 100644
index 00000000..6af0925f
--- /dev/null
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/health/InformerHealthSummary.java
@@ -0,0 +1,61 @@
+/*
+ * 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 io.javaoperatorsdk.operator.RuntimeInfo;
+import io.javaoperatorsdk.operator.health.InformerHealthIndicator;
+import io.javaoperatorsdk.operator.health.Status;
+import lombok.Value;
+
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+/** Operator informer health summary. */
+@Value
+public class InformerHealthSummary {
+
+ boolean anyHealthy;
+ Set<InformerIdentifier> unhealthyInformers;
+
+ public static InformerHealthSummary fromRuntimeInfo(RuntimeInfo runtimeInfo) {
+ var newUnHealthy = new HashSet<InformerIdentifier>();
+ boolean anyHealthy = false;
+
+ for (var controllerEntry :
+ runtimeInfo.unhealthyInformerWrappingEventSourceHealthIndicator().entrySet()) {
+ for (var eventSourceEntry : controllerEntry.getValue().entrySet()) {
+ Map<String, InformerHealthIndicator> informers =
+ eventSourceEntry.getValue().informerHealthIndicators();
+ for (var informerEntry : informers.entrySet()) {
+ if (informerEntry.getValue().getStatus() == Status.HEALTHY) {
+ anyHealthy = true;
+ } else {
+ newUnHealthy.add(
+ new InformerIdentifier(
+ controllerEntry.getKey(),
+ eventSourceEntry.getKey(),
+ informerEntry.getKey()));
+ }
+ }
+ }
+ }
+
+ return new InformerHealthSummary(anyHealthy || newUnHealthy.isEmpty(), newUnHealthy);
+ }
+}
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/health/InformerIdentifier.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/health/InformerIdentifier.java
new file mode 100644
index 00000000..65166c82
--- /dev/null
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/health/InformerIdentifier.java
@@ -0,0 +1,29 @@
+/*
+ * 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 lombok.Value;
+
+/** Operator informer identifier. */
+@Value
+public class InformerIdentifier {
+
+ String controllerName;
+ String eventSourceName;
+ String informerName;
+}
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/health/HealthProbeTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/health/HealthProbeTest.java
index dfdd2dff..91385252 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/health/HealthProbeTest.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/health/HealthProbeTest.java
@@ -31,11 +31,18 @@ import io.fabric8.kubernetes.client.server.mock.EnableKubernetesMockClient;
import io.javaoperatorsdk.operator.Operator;
import io.javaoperatorsdk.operator.RuntimeInfo;
import io.javaoperatorsdk.operator.api.config.ConfigurationServiceProvider;
+import io.javaoperatorsdk.operator.api.config.ResourceConfiguration;
+import io.javaoperatorsdk.operator.health.InformerHealthIndicator;
+import io.javaoperatorsdk.operator.health.InformerWrappingEventSourceHealthIndicator;
+import io.javaoperatorsdk.operator.health.Status;
import io.javaoperatorsdk.operator.processing.event.ResourceID;
import org.junit.jupiter.api.Test;
import java.net.HttpURLConnection;
import java.net.URL;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;
import static org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus.OK;
@@ -74,11 +81,11 @@ public class HealthProbeTest {
}
@Test
- public void testHealthProbe() {
+ public void testHealthProbeInfomers() {
var isRunning = new AtomicBoolean(false);
- var isHealthy = new AtomicBoolean(false);
-
- HealthProbe.INSTANCE.setRuntimeInfo(
+ var unhealthyEventSources =
+ new HashMap<String, Map<String, InformerWrappingEventSourceHealthIndicator>>();
+ var runtimeInfo =
new RuntimeInfo(new Operator(client)) {
@Override
public boolean isStarted() {
@@ -86,19 +93,85 @@ public class HealthProbeTest {
}
@Override
- public boolean allEventSourcesAreHealthy() {
- return isHealthy.get();
+ public Map<String, Map<String, InformerWrappingEventSourceHealthIndicator>>
+ unhealthyInformerWrappingEventSourceHealthIndicator() {
+ return unhealthyEventSources;
}
- });
+ };
+
+ // Test if a new event source becomes unhealthy
+ HealthProbe.INSTANCE.setRuntimeInfo(runtimeInfo);
assertFalse(HealthProbe.INSTANCE.isHealthy());
isRunning.set(true);
+ assertTrue(HealthProbe.INSTANCE.isHealthy());
+ unhealthyEventSources.put(
+ "c1", Map.of("e1", informerHealthIndicator(Map.of("i1", Status.UNHEALTHY))));
assertFalse(HealthProbe.INSTANCE.isHealthy());
- isHealthy.set(true);
+ unhealthyEventSources.clear();
+ assertTrue(HealthProbe.INSTANCE.isHealthy());
+
+ // Test if we detect when there is an unhealthy at start
+ unhealthyEventSources.put(
+ "c1",
+ Map.of(
+ "e1",
+ informerHealthIndicator(
+ Map.of("i1", Status.UNHEALTHY, "i2", Status.HEALTHY))));
+ HealthProbe.INSTANCE.setRuntimeInfo(runtimeInfo);
assertTrue(HealthProbe.INSTANCE.isHealthy());
- isRunning.set(false);
+ unhealthyEventSources.put(
+ "c1",
+ Map.of(
+ "e1",
+ informerHealthIndicator(
+ Map.of("i1", Status.UNHEALTHY, "i2", Status.HEALTHY)),
+ "e2",
+ informerHealthIndicator(Map.of("i3", Status.UNHEALTHY))));
assertFalse(HealthProbe.INSTANCE.isHealthy());
- isRunning.set(true);
+ assertFalse(HealthProbe.INSTANCE.isHealthy());
+ unhealthyEventSources.put(
+ "c1",
+ Map.of(
+ "e1",
+ informerHealthIndicator(
+ Map.of("i1", Status.UNHEALTHY, "i2", Status.HEALTHY)),
+ "e2",
+ informerHealthIndicator(Map.of("i3", Status.HEALTHY))));
+ assertTrue(HealthProbe.INSTANCE.isHealthy());
+ unhealthyEventSources.put(
+ "c1",
+ Map.of(
+ "e1",
+ informerHealthIndicator(
+ Map.of("i1", Status.UNHEALTHY, "i2", Status.UNHEALTHY))));
+ assertFalse(HealthProbe.INSTANCE.isHealthy());
+ unhealthyEventSources.clear();
+ assertTrue(HealthProbe.INSTANCE.isHealthy());
+
+ // All informers unhealthy at start:
+ unhealthyEventSources.put(
+ "c1", Map.of("e1", informerHealthIndicator(Map.of("i1", Status.UNHEALTHY))));
+ HealthProbe.INSTANCE.setRuntimeInfo(runtimeInfo);
+ assertFalse(HealthProbe.INSTANCE.isHealthy());
+ }
+
+ @Test
+ public void testHealthProbeCanary() {
+ var runtimeInfo =
+ new RuntimeInfo(new Operator(client)) {
+ @Override
+ public boolean isStarted() {
+ return true;
+ }
+
+ @Override
+ public Map<String, Map<String, InformerWrappingEventSourceHealthIndicator>>
+ unhealthyInformerWrappingEventSourceHealthIndicator() {
+ return Collections.emptyMap();
+ }
+ };
+ HealthProbe.INSTANCE.setRuntimeInfo(runtimeInfo);
var canaryManager =
new CanaryResourceManager<FlinkDeployment>(
@@ -142,4 +215,51 @@ public class HealthProbeTest {
connection.connect();
return connection.getResponseCode() == OK.code();
}
+
+ private static InformerWrappingEventSourceHealthIndicator informerHealthIndicator(
+ Map<String, Status> informerStatuses) {
+ Map<String, InformerHealthIndicator> informers = new HashMap<>();
+ informerStatuses.forEach(
+ (n, s) ->
+ informers.put(
+ n,
+ new InformerHealthIndicator() {
+ @Override
+ public boolean hasSynced() {
+ return false;
+ }
+
+ @Override
+ public boolean isWatching() {
+ return false;
+ }
+
+ @Override
+ public boolean isRunning() {
+ return false;
+ }
+
+ @Override
+ public Status getStatus() {
+ return s;
+ }
+
+ @Override
+ public String getTargetNamespace() {
+ return null;
+ }
+ }));
+
+ return new InformerWrappingEventSourceHealthIndicator() {
+ @Override
+ public Map<String, InformerHealthIndicator> informerHealthIndicators() {
+ return informers;
+ }
+
+ @Override
+ public ResourceConfiguration getInformerConfiguration() {
+ return null;
+ }
+ };
+ }
}