You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by wa...@apache.org on 2022/05/16 03:57:44 UTC
[flink-kubernetes-operator] branch main updated: [FLINK-27614] Use informer in webhook to avoid query apiserver at each update
This is an automated email from the ASF dual-hosted git repository.
wangyang0918 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 6c9c452 [FLINK-27614] Use informer in webhook to avoid query apiserver at each update
6c9c452 is described below
commit 6c9c45250a49b7ebf5484aae820822e8c5e73fbc
Author: Aitozi <yu...@alibaba-inc.com>
AuthorDate: Sat May 14 16:35:40 2022 +0800
[FLINK-27614] Use informer in webhook to avoid query apiserver at each update
---
.github/workflows/ci.yml | 4 +-
.../operator/informer/InformerManager.java | 65 ++++++++++++++++++++--
.../kubernetes/operator/utils/OperatorUtils.java | 34 +++--------
.../operator/informer/InformerManagerTest.java | 51 +++++++++++++++++
.../operator/admission/FlinkOperatorWebhook.java | 3 +-
.../operator/admission/FlinkValidator.java | 24 +++++---
.../operator/admission/AdmissionHandlerTest.java | 3 +-
.../templates/flink-operator.yaml | 2 +
8 files changed, 142 insertions(+), 44 deletions(-)
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 7b69f82..c6e6f6c 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -67,7 +67,7 @@ jobs:
docker images
- name: Start the operator
run: |
- helm install flink-kubernetes-operator helm/flink-kubernetes-operator --set image.repository=flink-kubernetes-operator --set image.tag=ci-latest
+ helm --debug install flink-kubernetes-operator helm/flink-kubernetes-operator --set image.repository=flink-kubernetes-operator --set image.tag=ci-latest
kubectl wait --for=condition=Available --timeout=120s deploy/flink-kubernetes-operator
kubectl get pods
- name: Tests in flink-kubernetes-operator
@@ -145,7 +145,7 @@ jobs:
docker images
- name: Start the operator
run: |
- helm install flink-kubernetes-operator -n ${{ matrix.config.namespace }} helm/flink-kubernetes-operator --set image.repository=flink-kubernetes-operator --set image.tag=ci-latest ${{ matrix.config.extraArgs }}
+ helm --debug install flink-kubernetes-operator -n ${{ matrix.config.namespace }} helm/flink-kubernetes-operator --set image.repository=flink-kubernetes-operator --set image.tag=ci-latest ${{ matrix.config.extraArgs }}
kubectl wait --for=condition=Available --timeout=120s -n ${{ matrix.config.namespace }} deploy/flink-kubernetes-operator
kubectl get pods
- name: Run Flink e2e tests
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/informer/InformerManager.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/informer/InformerManager.java
index 0f8d86d..a7d0aa3 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/informer/InformerManager.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/informer/InformerManager.java
@@ -17,6 +17,7 @@
package org.apache.flink.kubernetes.operator.informer;
+import org.apache.flink.kubernetes.operator.crd.FlinkDeployment;
import org.apache.flink.kubernetes.operator.crd.FlinkSessionJob;
import org.apache.flink.kubernetes.operator.utils.OperatorUtils;
import org.apache.flink.util.Preconditions;
@@ -26,6 +27,7 @@ import io.fabric8.kubernetes.client.informers.SharedIndexInformer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -36,14 +38,20 @@ public class InformerManager {
private final Set<String> watchedNamespaces;
private final KubernetesClient kubernetesClient;
private volatile Map<String, SharedIndexInformer<FlinkSessionJob>> sessionJobInformers;
+ private volatile Map<String, SharedIndexInformer<FlinkDeployment>> flinkDepInformers;
public InformerManager(Set<String> watchedNamespaces, KubernetesClient kubernetesClient) {
this.watchedNamespaces = watchedNamespaces;
this.kubernetesClient = kubernetesClient;
+ LOG.info(
+ "Created informer manager with watchedNamespaces: {}",
+ watchedNamespaces.isEmpty()
+ ? "[" + OperatorUtils.ALL_NAMESPACE + "]"
+ : watchedNamespaces);
}
public SharedIndexInformer<FlinkSessionJob> getSessionJobInformer(String namespace) {
- initSessionJobInformer();
+ initSessionJobInformersWithIndexer();
var effectiveNamespace =
watchedNamespaces.isEmpty() ? OperatorUtils.ALL_NAMESPACE : namespace;
var informer = sessionJobInformers.get(effectiveNamespace);
@@ -52,14 +60,59 @@ public class InformerManager {
return informer;
}
- private void initSessionJobInformer() {
+ public SharedIndexInformer<FlinkDeployment> getFlinkDepInformer(String namespace) {
+ initFlinkDepInformers();
+ var effectiveNamespace =
+ watchedNamespaces.isEmpty() ? OperatorUtils.ALL_NAMESPACE : namespace;
+ var informer = flinkDepInformers.get(effectiveNamespace);
+ Preconditions.checkNotNull(
+ informer, String.format("The informer for %s should not be null", namespace));
+ return informer;
+ }
+
+ private void initSessionJobInformersWithIndexer() {
if (sessionJobInformers == null) {
synchronized (this) {
if (sessionJobInformers == null) {
- this.sessionJobInformers =
- OperatorUtils.createSessionJobInformersWithIndexer(
- watchedNamespaces, kubernetesClient);
- LOG.info("Created informer for {}", sessionJobInformers.keySet());
+ var runnableInformers =
+ OperatorUtils.createRunnableInformer(
+ FlinkSessionJob.class, watchedNamespaces, kubernetesClient);
+ for (Map.Entry<String, SharedIndexInformer<FlinkSessionJob>> runnableInformer :
+ runnableInformers.entrySet()) {
+ runnableInformer
+ .getValue()
+ .addIndexers(
+ Map.of(
+ OperatorUtils.CLUSTER_ID_INDEX,
+ sessionJob ->
+ List.of(
+ sessionJob
+ .getSpec()
+ .getDeploymentName())));
+ runnableInformer.getValue().run();
+ }
+ this.sessionJobInformers = runnableInformers;
+ LOG.info("Created session job informers for {}", sessionJobInformers.keySet());
+ }
+ }
+ }
+ }
+
+ private void initFlinkDepInformers() {
+ if (flinkDepInformers == null) {
+ synchronized (this) {
+ if (flinkDepInformers == null) {
+ var runnableInformers =
+ OperatorUtils.createRunnableInformer(
+ FlinkDeployment.class, watchedNamespaces, kubernetesClient);
+ for (Map.Entry<String, SharedIndexInformer<FlinkDeployment>> runnableInformer :
+ runnableInformers.entrySet()) {
+ runnableInformer.getValue().run();
+ }
+ this.flinkDepInformers = runnableInformers;
+ LOG.info(
+ "Created flink deployment informers for {}",
+ flinkDepInformers.keySet());
}
}
}
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/OperatorUtils.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/OperatorUtils.java
index 5844522..457fa98 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/OperatorUtils.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/OperatorUtils.java
@@ -40,11 +40,9 @@ import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
-import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
-import java.util.function.Function;
/** Operator SDK related utility functions. */
public class OperatorUtils {
@@ -106,40 +104,26 @@ public class OperatorUtils {
return context.getSecondaryResource(FlinkDeployment.class, identifier);
}
- /**
- * Create informers for session job to build indexer for cluster to session job relations.
- *
- * @return The different namespace's index informer.
- */
- public static Map<String, SharedIndexInformer<FlinkSessionJob>>
- createSessionJobInformersWithIndexer(
- Set<String> effectiveNamespaces, KubernetesClient kubernetesClient) {
+ public static <CR extends HasMetadata>
+ Map<String, SharedIndexInformer<CR>> createRunnableInformer(
+ Class<CR> resourceClass,
+ Set<String> effectiveNamespaces,
+ KubernetesClient kubernetesClient) {
if (effectiveNamespaces.isEmpty()) {
return Map.of(
ALL_NAMESPACE,
- kubernetesClient
- .resources(FlinkSessionJob.class)
- .inAnyNamespace()
- .withIndexers(clusterToSessionJobIndexer())
- .inform());
+ kubernetesClient.resources(resourceClass).inAnyNamespace().runnableInformer(0));
} else {
- var informers = new HashMap<String, SharedIndexInformer<FlinkSessionJob>>();
+ var informers = new HashMap<String, SharedIndexInformer<CR>>();
for (String effectiveNamespace : effectiveNamespaces) {
informers.put(
effectiveNamespace,
kubernetesClient
- .resources(FlinkSessionJob.class)
+ .resources(resourceClass)
.inNamespace(effectiveNamespace)
- .withIndexers(clusterToSessionJobIndexer())
- .inform());
+ .runnableInformer(0));
}
return informers;
}
}
-
- private static Map<String, Function<FlinkSessionJob, List<String>>>
- clusterToSessionJobIndexer() {
- return Map.of(
- CLUSTER_ID_INDEX, sessionJob -> List.of(sessionJob.getSpec().getDeploymentName()));
- }
}
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/informer/InformerManagerTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/informer/InformerManagerTest.java
new file mode 100644
index 0000000..09715f1
--- /dev/null
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/informer/InformerManagerTest.java
@@ -0,0 +1,51 @@
+/*
+ * 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.informer;
+
+import org.apache.flink.kubernetes.operator.utils.OperatorUtils;
+
+import io.fabric8.kubernetes.client.KubernetesClient;
+import io.fabric8.kubernetes.client.server.mock.EnableKubernetesMockClient;
+import io.fabric8.kubernetes.client.server.mock.KubernetesMockServer;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.HashSet;
+import java.util.Set;
+
+/** Test for {@link InformerManager}. */
+@EnableKubernetesMockClient(crud = true)
+public class InformerManagerTest {
+
+ private KubernetesMockServer mockServer;
+ private KubernetesClient kubernetesClient;
+
+ @Test
+ public void testNamespacedInformerCreated() {
+ var informerManager = new InformerManager(new HashSet<>(), kubernetesClient);
+ Assertions.assertNotNull(
+ informerManager.getSessionJobInformer(OperatorUtils.ALL_NAMESPACE));
+ Assertions.assertNotNull(informerManager.getFlinkDepInformer(OperatorUtils.ALL_NAMESPACE));
+
+ informerManager = new InformerManager(Set.of("ns1", "ns2"), kubernetesClient);
+ Assertions.assertNotNull(informerManager.getSessionJobInformer("ns1"));
+ Assertions.assertNotNull(informerManager.getSessionJobInformer("ns2"));
+ Assertions.assertNotNull(informerManager.getFlinkDepInformer("ns1"));
+ Assertions.assertNotNull(informerManager.getFlinkDepInformer("ns2"));
+ }
+}
diff --git a/flink-kubernetes-webhook/src/main/java/org/apache/flink/kubernetes/operator/admission/FlinkOperatorWebhook.java b/flink-kubernetes-webhook/src/main/java/org/apache/flink/kubernetes/operator/admission/FlinkOperatorWebhook.java
index cb018fe..21a8a21 100644
--- a/flink-kubernetes-webhook/src/main/java/org/apache/flink/kubernetes/operator/admission/FlinkOperatorWebhook.java
+++ b/flink-kubernetes-webhook/src/main/java/org/apache/flink/kubernetes/operator/admission/FlinkOperatorWebhook.java
@@ -60,7 +60,8 @@ public class FlinkOperatorWebhook {
EnvUtils.logEnvironmentInfo(LOG, "Flink Kubernetes Webhook", args);
FlinkConfigManager configManager = new FlinkConfigManager();
Set<FlinkResourceValidator> validators = ValidatorUtils.discoverValidators(configManager);
- AdmissionHandler endpoint = new AdmissionHandler(new FlinkValidator(validators));
+ AdmissionHandler endpoint =
+ new AdmissionHandler(new FlinkValidator(validators, configManager));
ChannelInitializer<SocketChannel> initializer = createChannelInitializer(endpoint);
NioEventLoopGroup bossGroup = new NioEventLoopGroup(1);
NioEventLoopGroup workerGroup = new NioEventLoopGroup();
diff --git a/flink-kubernetes-webhook/src/main/java/org/apache/flink/kubernetes/operator/admission/FlinkValidator.java b/flink-kubernetes-webhook/src/main/java/org/apache/flink/kubernetes/operator/admission/FlinkValidator.java
index ded2659..5882a54 100644
--- a/flink-kubernetes-webhook/src/main/java/org/apache/flink/kubernetes/operator/admission/FlinkValidator.java
+++ b/flink-kubernetes-webhook/src/main/java/org/apache/flink/kubernetes/operator/admission/FlinkValidator.java
@@ -20,16 +20,18 @@ package org.apache.flink.kubernetes.operator.admission;
import org.apache.flink.kubernetes.operator.admission.admissioncontroller.NotAllowedException;
import org.apache.flink.kubernetes.operator.admission.admissioncontroller.Operation;
import org.apache.flink.kubernetes.operator.admission.admissioncontroller.validation.Validator;
+import org.apache.flink.kubernetes.operator.config.FlinkConfigManager;
import org.apache.flink.kubernetes.operator.crd.CrdConstants;
import org.apache.flink.kubernetes.operator.crd.FlinkDeployment;
import org.apache.flink.kubernetes.operator.crd.FlinkSessionJob;
+import org.apache.flink.kubernetes.operator.informer.InformerManager;
import org.apache.flink.kubernetes.operator.validation.FlinkResourceValidator;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.fabric8.kubernetes.api.model.GroupVersionKind;
import io.fabric8.kubernetes.api.model.KubernetesResource;
import io.fabric8.kubernetes.client.DefaultKubernetesClient;
-import io.fabric8.kubernetes.client.KubernetesClient;
+import io.fabric8.kubernetes.client.informers.cache.Cache;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -42,11 +44,15 @@ public class FlinkValidator implements Validator<KubernetesResource> {
private static final ObjectMapper objectMapper = new ObjectMapper();
private final Set<FlinkResourceValidator> validators;
- private final KubernetesClient client;
+ private final InformerManager informerManager;
- public FlinkValidator(Set<FlinkResourceValidator> validators) {
+ public FlinkValidator(
+ Set<FlinkResourceValidator> validators, FlinkConfigManager configManager) {
this.validators = validators;
- this.client = new DefaultKubernetesClient();
+ this.informerManager =
+ new InformerManager(
+ configManager.getOperatorConfiguration().getWatchedNamespaces(),
+ new DefaultKubernetesClient());
}
@Override
@@ -76,11 +82,11 @@ public class FlinkValidator implements Validator<KubernetesResource> {
private void validateSessionJob(KubernetesResource resource) {
FlinkSessionJob sessionJob = objectMapper.convertValue(resource, FlinkSessionJob.class);
- var deployment =
- client.resources(FlinkDeployment.class)
- .inNamespace(sessionJob.getMetadata().getNamespace())
- .withName(sessionJob.getSpec().getDeploymentName())
- .get();
+ var namespace = sessionJob.getMetadata().getNamespace();
+ var deploymentName = sessionJob.getSpec().getDeploymentName();
+
+ var key = Cache.namespaceKeyFunc(namespace, deploymentName);
+ var deployment = informerManager.getFlinkDepInformer(namespace).getStore().getByKey(key);
for (FlinkResourceValidator validator : validators) {
Optional<String> validationError =
diff --git a/flink-kubernetes-webhook/src/test/java/org/apache/flink/kubernetes/operator/admission/AdmissionHandlerTest.java b/flink-kubernetes-webhook/src/test/java/org/apache/flink/kubernetes/operator/admission/AdmissionHandlerTest.java
index 09a8c46..2e260be 100644
--- a/flink-kubernetes-webhook/src/test/java/org/apache/flink/kubernetes/operator/admission/AdmissionHandlerTest.java
+++ b/flink-kubernetes-webhook/src/test/java/org/apache/flink/kubernetes/operator/admission/AdmissionHandlerTest.java
@@ -52,7 +52,8 @@ public class AdmissionHandlerTest {
private final AdmissionHandler admissionHandler =
new AdmissionHandler(
new FlinkValidator(
- ValidatorUtils.discoverValidators(new FlinkConfigManager())));
+ ValidatorUtils.discoverValidators(new FlinkConfigManager()),
+ new FlinkConfigManager()));
@Test
public void testHandleIllegalRequest() {
diff --git a/helm/flink-kubernetes-operator/templates/flink-operator.yaml b/helm/flink-kubernetes-operator/templates/flink-operator.yaml
index 66f4731..3a962c0 100644
--- a/helm/flink-kubernetes-operator/templates/flink-operator.yaml
+++ b/helm/flink-kubernetes-operator/templates/flink-operator.yaml
@@ -109,6 +109,8 @@ spec:
value: /opt/flink/plugins
- name: OPERATOR_NAMESPACE
value: {{ .Release.Namespace }}
+ - name: FLINK_OPERATOR_WATCH_NAMESPACES
+ value: {{ join "," .Values.watchNamespaces }}
securityContext:
{{- toYaml .Values.webhookSecurityContext | nindent 12 }}
volumeMounts: