From 30311ecdf470db4bb6d23c7883aed6aad901a898 Mon Sep 17 00:00:00 2001 From: wangyang0918 Date: Tue, 7 Apr 2020 20:10:40 +0800 Subject: [PATCH] [FLINK-15639][k8s] Support to set tolerations for jobmanager and taskmanger pod Taints and tolerations work together to ensure that pods are not scheduled onto inappropriate nodes. One or more taints are applied to a node; this marks that the node should not accept any pods that do not tolerate the taints. Tolerations are applied to pods, and allow (but do not require) the pods to schedule onto nodes with matching taints. This closes #11606 . --- .../kubernetes_config_configuration.html | 12 ++++ .../KubernetesConfigOptions.java | 18 +++++ .../decorators/InitJobManagerDecorator.java | 4 ++ .../decorators/InitTaskManagerDecorator.java | 4 ++ .../KubernetesJobManagerParameters.java | 6 ++ .../parameters/KubernetesParameters.java | 7 ++ .../KubernetesTaskManagerParameters.java | 6 ++ .../resources/KubernetesToleration.java | 65 +++++++++++++++++++ .../InitJobManagerDecoratorTest.java | 13 ++++ .../InitTaskManagerDecoratorTest.java | 13 ++++ 10 files changed, 148 insertions(+) create mode 100644 flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/resources/KubernetesToleration.java diff --git a/docs/_includes/generated/kubernetes_config_configuration.html b/docs/_includes/generated/kubernetes_config_configuration.html index 6ef559f6dc1d4..114a05b9022bb 100644 --- a/docs/_includes/generated/kubernetes_config_configuration.html +++ b/docs/_includes/generated/kubernetes_config_configuration.html @@ -104,6 +104,12 @@ String Service account that is used by jobmanager within kubernetes cluster. The job manager uses this service account when requesting taskmanager pods from the API server. + +
kubernetes.jobmanager.tolerations
+ (none) + List<Map> + The user-specified tolerations to be set to the JobManager pod. The value should be in the form of key:key1,operator:Equal,value:value1,effect:NoSchedule;key:key2,operator:Exists,effect:NoExecute,tolerationSeconds:6000 +
kubernetes.namespace
"default" @@ -140,5 +146,11 @@ Map The node selector to be set for TaskManager pods. Specified as key:value pairs separated by commas. For example, environment:production,disk:ssd. + +
kubernetes.taskmanager.tolerations
+ (none) + List<Map> + The user-specified tolerations to be set to the TaskManager pod. The value should be in the form of key:key1,operator:Equal,value:value1,effect:NoSchedule;key:key2,operator:Exists,effect:NoExecute,tolerationSeconds:6000 + diff --git a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/configuration/KubernetesConfigOptions.java b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/configuration/KubernetesConfigOptions.java index 39b2cad4e3338..70969b7d911d9 100644 --- a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/configuration/KubernetesConfigOptions.java +++ b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/configuration/KubernetesConfigOptions.java @@ -186,6 +186,24 @@ public class KubernetesConfigOptions { .withDescription("The user-specified annotations that are set to the TaskManager pod. The value could be " + "in the form of a1:v1,a2:v2"); + public static final ConfigOption>> JOB_MANAGER_TOLERATIONS = + key("kubernetes.jobmanager.tolerations") + .mapType() + .asList() + .noDefaultValue() + .withDescription("The user-specified tolerations to be set to the JobManager pod. The value should be " + + "in the form of key:key1,operator:Equal,value:value1,effect:NoSchedule;" + + "key:key2,operator:Exists,effect:NoExecute,tolerationSeconds:6000"); + + public static final ConfigOption>> TASK_MANAGER_TOLERATIONS = + key("kubernetes.taskmanager.tolerations") + .mapType() + .asList() + .noDefaultValue() + .withDescription("The user-specified tolerations to be set to the TaskManager pod. The value should be " + + "in the form of key:key1,operator:Equal,value:value1,effect:NoSchedule;" + + "key:key2,operator:Exists,effect:NoExecute,tolerationSeconds:6000"); + /** * The flink rest service exposed type. */ diff --git a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/InitJobManagerDecorator.java b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/InitJobManagerDecorator.java index 45fa10d2d7e53..ad54d71dd53a2 100644 --- a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/InitJobManagerDecorator.java +++ b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/InitJobManagerDecorator.java @@ -20,6 +20,7 @@ import org.apache.flink.kubernetes.kubeclient.FlinkPod; import org.apache.flink.kubernetes.kubeclient.parameters.KubernetesJobManagerParameters; +import org.apache.flink.kubernetes.kubeclient.resources.KubernetesToleration; import org.apache.flink.kubernetes.utils.Constants; import org.apache.flink.kubernetes.utils.KubernetesUtils; @@ -66,6 +67,9 @@ public FlinkPod decorateFlinkPod(FlinkPod flinkPod) { .withServiceAccountName(kubernetesJobManagerParameters.getServiceAccount()) .withImagePullSecrets(kubernetesJobManagerParameters.getImagePullSecrets()) .withNodeSelector(kubernetesJobManagerParameters.getNodeSelector()) + .withTolerations(kubernetesJobManagerParameters.getTolerations().stream() + .map(e -> KubernetesToleration.fromMap(e).getInternalResource()) + .collect(Collectors.toList())) .endSpec() .build(); diff --git a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/InitTaskManagerDecorator.java b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/InitTaskManagerDecorator.java index e789b60dcf2b8..f968b750ad1fa 100644 --- a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/InitTaskManagerDecorator.java +++ b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/InitTaskManagerDecorator.java @@ -20,6 +20,7 @@ import org.apache.flink.kubernetes.kubeclient.FlinkPod; import org.apache.flink.kubernetes.kubeclient.parameters.KubernetesTaskManagerParameters; +import org.apache.flink.kubernetes.kubeclient.resources.KubernetesToleration; import org.apache.flink.kubernetes.utils.Constants; import org.apache.flink.kubernetes.utils.KubernetesUtils; @@ -60,6 +61,9 @@ public FlinkPod decorateFlinkPod(FlinkPod flinkPod) { .editOrNewSpec() .withImagePullSecrets(kubernetesTaskManagerParameters.getImagePullSecrets()) .withNodeSelector(kubernetesTaskManagerParameters.getNodeSelector()) + .withTolerations(kubernetesTaskManagerParameters.getTolerations().stream() + .map(e -> KubernetesToleration.fromMap(e).getInternalResource()) + .collect(Collectors.toList())) .endSpec() .build(); diff --git a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/parameters/KubernetesJobManagerParameters.java b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/parameters/KubernetesJobManagerParameters.java index 89f15f38247bb..6540cdb1aca16 100644 --- a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/parameters/KubernetesJobManagerParameters.java +++ b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/parameters/KubernetesJobManagerParameters.java @@ -33,6 +33,7 @@ import java.util.Collections; import java.util.HashMap; +import java.util.List; import java.util.Map; import static org.apache.flink.util.Preconditions.checkArgument; @@ -79,6 +80,11 @@ public Map getAnnotations() { return flinkConfig.getOptional(KubernetesConfigOptions.JOB_MANAGER_ANNOTATIONS).orElse(Collections.emptyMap()); } + @Override + public List> getTolerations() { + return flinkConfig.getOptional(KubernetesConfigOptions.JOB_MANAGER_TOLERATIONS).orElse(Collections.emptyList()); + } + public String getJobManagerMainContainerName() { return JOB_MANAGER_MAIN_CONTAINER_NAME; } diff --git a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/parameters/KubernetesParameters.java b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/parameters/KubernetesParameters.java index 2f66101090f38..44443c421f868 100644 --- a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/parameters/KubernetesParameters.java +++ b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/parameters/KubernetesParameters.java @@ -22,6 +22,7 @@ import io.fabric8.kubernetes.api.model.LocalObjectReference; +import java.util.List; import java.util.Map; import java.util.Optional; @@ -67,6 +68,12 @@ public interface KubernetesParameters { */ Map getAnnotations(); + /** + * A collection of tolerations that are set to the JobManager and TaskManager Pod(s). Kubernetes taints and + * tolerations work together to ensure that pods are not scheduled onto inappropriate nodes. + */ + List> getTolerations(); + /** * Directory in Pod that stores the flink-conf.yaml, log4j.properties, and the logback.xml. */ diff --git a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/parameters/KubernetesTaskManagerParameters.java b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/parameters/KubernetesTaskManagerParameters.java index 6929d3a809357..663db57a63b78 100644 --- a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/parameters/KubernetesTaskManagerParameters.java +++ b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/parameters/KubernetesTaskManagerParameters.java @@ -26,6 +26,7 @@ import java.util.Collections; import java.util.HashMap; +import java.util.List; import java.util.Map; import static org.apache.flink.util.Preconditions.checkArgument; @@ -84,6 +85,11 @@ public Map getAnnotations() { return flinkConfig.getOptional(KubernetesConfigOptions.TASK_MANAGER_ANNOTATIONS).orElse(Collections.emptyMap()); } + @Override + public List> getTolerations() { + return flinkConfig.getOptional(KubernetesConfigOptions.TASK_MANAGER_TOLERATIONS).orElse(Collections.emptyList()); + } + public String getTaskManagerMainContainerName() { return TASK_MANAGER_MAIN_CONTAINER_NAME; } diff --git a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/resources/KubernetesToleration.java b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/resources/KubernetesToleration.java new file mode 100644 index 0000000000000..2d43208d3fd13 --- /dev/null +++ b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/resources/KubernetesToleration.java @@ -0,0 +1,65 @@ +/* + * 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.kubeclient.resources; + +import io.fabric8.kubernetes.api.model.Toleration; +import io.fabric8.kubernetes.api.model.TolerationBuilder; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.Map; + +/** + * Represent Toleration resource in kubernetes. + */ +public class KubernetesToleration extends KubernetesResource { + + private static final Logger LOG = LoggerFactory.getLogger(KubernetesToleration.class); + + private KubernetesToleration(Toleration toleration) { + super(toleration); + } + + public static KubernetesToleration fromMap(Map stringMap) { + final TolerationBuilder tolerationBuilder = new TolerationBuilder(); + stringMap.forEach((k, v) -> { + switch (k.toLowerCase()) { + case "effect": + tolerationBuilder.withEffect(v); + break; + case "key": + tolerationBuilder.withKey(v); + break; + case "operator": + tolerationBuilder.withOperator(v); + break; + case "tolerationseconds": + tolerationBuilder.withTolerationSeconds(Long.valueOf(v)); + break; + case "value": + tolerationBuilder.withValue(v); + break; + default: + LOG.warn("Unrecognized key({}) of toleration, will ignore.", k); + break; + } + }); + return new KubernetesToleration(tolerationBuilder.build()); + } +} diff --git a/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/decorators/InitJobManagerDecoratorTest.java b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/decorators/InitJobManagerDecoratorTest.java index 78d1976f2d609..02981298837e5 100644 --- a/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/decorators/InitJobManagerDecoratorTest.java +++ b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/decorators/InitJobManagerDecoratorTest.java @@ -32,6 +32,8 @@ import io.fabric8.kubernetes.api.model.Pod; import io.fabric8.kubernetes.api.model.Quantity; import io.fabric8.kubernetes.api.model.ResourceRequirements; +import io.fabric8.kubernetes.api.model.Toleration; +import org.hamcrest.Matchers; import org.junit.Before; import org.junit.Test; @@ -60,6 +62,11 @@ public class InitJobManagerDecoratorTest extends KubernetesJobManagerTestBase { put("a2", "v2"); } }; + private static final String TOLERATION_STRING = "key:key1,operator:Equal,value:value1,effect:NoSchedule;" + + "KEY:key2,operator:Exists,Effect:NoExecute,tolerationSeconds:6000"; + private static final List TOLERATION = Arrays.asList( + new Toleration("NoSchedule", "key1", "Equal", null, "value1"), + new Toleration("NoExecute", "key2", "Exists", 6000L, null)); private Pod resultPod; private Container resultMainContainer; @@ -70,6 +77,7 @@ public void setup() throws Exception { this.flinkConfig.set(KubernetesConfigOptions.JOB_MANAGER_SERVICE_ACCOUNT, SERVICE_ACCOUNT_NAME); this.flinkConfig.set(KubernetesConfigOptions.CONTAINER_IMAGE_PULL_SECRETS, IMAGE_PULL_SECRETS); this.flinkConfig.set(KubernetesConfigOptions.JOB_MANAGER_ANNOTATIONS, ANNOTATIONS); + this.flinkConfig.setString(KubernetesConfigOptions.JOB_MANAGER_TOLERATIONS.key(), TOLERATION_STRING); final InitJobManagerDecorator initJobManagerDecorator = new InitJobManagerDecorator(this.kubernetesJobManagerParameters); @@ -174,4 +182,9 @@ public void testImagePullSecrets() { public void testNodeSelector() { assertThat(this.resultPod.getSpec().getNodeSelector(), is(equalTo(nodeSelector))); } + + @Test + public void testPodTolerations() { + assertThat(this.resultPod.getSpec().getTolerations(), Matchers.containsInAnyOrder(TOLERATION.toArray())); + } } diff --git a/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/decorators/InitTaskManagerDecoratorTest.java b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/decorators/InitTaskManagerDecoratorTest.java index 287548c2f3f70..1b91bc206cc8e 100644 --- a/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/decorators/InitTaskManagerDecoratorTest.java +++ b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/decorators/InitTaskManagerDecoratorTest.java @@ -31,6 +31,8 @@ import io.fabric8.kubernetes.api.model.Pod; import io.fabric8.kubernetes.api.model.Quantity; import io.fabric8.kubernetes.api.model.ResourceRequirements; +import io.fabric8.kubernetes.api.model.Toleration; +import org.hamcrest.Matchers; import org.junit.Before; import org.junit.Test; @@ -58,6 +60,11 @@ public class InitTaskManagerDecoratorTest extends KubernetesTaskManagerTestBase put("a2", "v2"); } }; + private static final String TOLERATION_STRING = "key:key1,operator:Equal,value:value1,effect:NoSchedule;" + + "KEY:key2,operator:Exists,Effect:NoExecute,tolerationSeconds:6000"; + private static final List TOLERATION = Arrays.asList( + new Toleration("NoSchedule", "key1", "Equal", null, "value1"), + new Toleration("NoExecute", "key2", "Exists", 6000L, null)); private Pod resultPod; private Container resultMainContainer; @@ -67,6 +74,7 @@ public void setup() throws Exception { super.setup(); this.flinkConfig.set(KubernetesConfigOptions.CONTAINER_IMAGE_PULL_SECRETS, IMAGE_PULL_SECRETS); this.flinkConfig.set(KubernetesConfigOptions.TASK_MANAGER_ANNOTATIONS, ANNOTATIONS); + this.flinkConfig.setString(KubernetesConfigOptions.TASK_MANAGER_TOLERATIONS.key(), TOLERATION_STRING); final InitTaskManagerDecorator initTaskManagerDecorator = new InitTaskManagerDecorator(kubernetesTaskManagerParameters); @@ -168,4 +176,9 @@ public void testImagePullSecrets() { public void testNodeSelector() { assertThat(this.resultPod.getSpec().getNodeSelector(), is(equalTo(nodeSelector))); } + + @Test + public void testPodTolerations() { + assertThat(this.resultPod.getSpec().getTolerations(), Matchers.containsInAnyOrder(TOLERATION.toArray())); + } }