You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tr...@apache.org on 2020/08/25 08:05:43 UTC
[flink] branch master updated: [FLINK-16699][k8s] Support accessing
secured services via K8s secrets
This is an automated email from the ASF dual-hosted git repository.
trohrmann pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 8d2c969 [FLINK-16699][k8s] Support accessing secured services via K8s secrets
8d2c969 is described below
commit 8d2c969a83a6830f52b230a00424ddd4f4795a3d
Author: totorooo <77...@qq.com>
AuthorDate: Mon Aug 24 16:10:36 2020 +0800
[FLINK-16699][k8s] Support accessing secured services via K8s secrets
This closes #12899.
---
.../generated/kubernetes_config_configuration.html | 12 +++
docs/ops/deployment/native_kubernetes.md | 75 ++++++++++++++++++
docs/ops/deployment/native_kubernetes.zh.md | 75 ++++++++++++++++++
.../configuration/KubernetesConfigOptions.java | 23 ++++++
.../kubeclient/decorators/EnvSecretsDecorator.java | 61 ++++++++++++++
.../decorators/MountSecretsDecorator.java | 92 ++++++++++++++++++++++
.../factory/KubernetesJobManagerFactory.java | 4 +
.../factory/KubernetesTaskManagerFactory.java | 4 +
.../parameters/AbstractKubernetesParameters.java | 9 +++
.../parameters/KubernetesParameters.java | 10 +++
.../resources/KubernetesSecretEnvVar.java | 50 ++++++++++++
.../apache/flink/kubernetes/VolumeTestUtils.java | 46 +++++++++++
.../decorators/EnvSecretsDecoratorTest.java | 64 +++++++++++++++
.../decorators/MountSecretsDecoratorTest.java | 67 ++++++++++++++++
14 files changed, 592 insertions(+)
diff --git a/docs/_includes/generated/kubernetes_config_configuration.html b/docs/_includes/generated/kubernetes_config_configuration.html
index c95ef2f..338ad6f 100644
--- a/docs/_includes/generated/kubernetes_config_configuration.html
+++ b/docs/_includes/generated/kubernetes_config_configuration.html
@@ -164,5 +164,17 @@
<td>List<Map></td>
<td>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</td>
</tr>
+ <tr>
+ <td><h5>kubernetes.secrets</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>Map</td>
+ <td>The user-specified secrets that will be mounted into Flink container. The value should be in the form of <span markdown="span">`foo:/opt/secrets-foo,bar:/opt/secrets-bar`</span>.</td>
+ </tr>
+ <tr>
+ <td><h5>kubernetes.env.secretKeyRef</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>List<Map></td>
+ <td>The user-specified secrets to set env variables in Flink container. The value should be in the form of <span markdown="span">`env:FOO_ENV,secret:foo_secret,key:foo_key;env:BAR_ENV,secret:bar_secret,key:bar_key`</span>.</td>
+ </tr>
</tbody>
</table>
diff --git a/docs/ops/deployment/native_kubernetes.md b/docs/ops/deployment/native_kubernetes.md
index 1b7654d..775f31f 100644
--- a/docs/ops/deployment/native_kubernetes.md
+++ b/docs/ops/deployment/native_kubernetes.md
@@ -211,6 +211,81 @@ $ ./bin/flink run-application -p 8 -t kubernetes-application \
local:///opt/flink/usrlib/my-flink-job.jar
{% endhighlight %}
+## Using Secrets
+
+[Kubernetes Secrets](https://kubernetes.io/docs/concepts/configuration/secret/) is an object that contains a small amount of sensitive data such as a password, a token, or a key.
+Such information might otherwise be put in a Pod specification or in an image. Flink on Kubernetes can use Secrets in two ways:
+
+- Using Secrets as files from a pod;
+
+- Using Secrets as environment variables;
+
+### Using Secrets as files from a pod
+
+Here is an example of a Pod that mounts a Secret in a volume:
+
+{% highlight yaml %}
+apiVersion: v1
+kind: Pod
+metadata:
+ name: foo
+spec:
+ containers:
+ - name: foo
+ image: foo
+ volumeMounts:
+ - name: foo
+ mountPath: "/opt/foo"
+ volumes:
+ - name: foo
+ secret:
+ secretName: foo
+{% endhighlight %}
+
+By applying this yaml, each key in foo Secrets becomes the filename under `/opt/foo` path. Flink on Kubernetes can enable this feature by the following command:
+
+{% highlight bash %}
+$ ./bin/kubernetes-session.sh \
+ -Dkubernetes.cluster-id=<ClusterId> \
+ -Dkubernetes.container.image=<CustomImageName> \
+ -Dkubernetes.secrets=foo:/opt/foo
+{% endhighlight %}
+
+For more details see the [official Kubernetes documentation](https://kubernetes.io/docs/concepts/configuration/secret/#using-secrets-as-files-from-a-pod).
+
+### Using Secrets as environment variables
+
+Here is an example of a Pod that uses secrets from environment variables:
+
+{% highlight yaml %}
+apiVersion: v1
+kind: Pod
+metadata:
+ name: foo
+spec:
+ containers:
+ - name: foo
+ image: foo
+ env:
+ - name: FOO_ENV
+ valueFrom:
+ secretKeyRef:
+ name: foo_secret
+ key: foo_key
+{% endhighlight %}
+
+By applying this yaml, an environment variable named `FOO_ENV` is added into `foo` container, and `FOO_ENV` consumes the value of `foo_key` which is defined in Secrets `foo_secret`.
+Flink on Kubernetes can enable this feature by the following command:
+
+{% highlight bash %}
+$ ./bin/kubernetes-session.sh \
+ -Dkubernetes.cluster-id=<ClusterId> \
+ -Dkubernetes.container.image=<CustomImageName> \
+ -Dkubernetes.env.secretKeyRef=env:FOO_ENV,secret:foo_secret,key:foo_key
+{% endhighlight %}
+
+For more details see the [official Kubernetes documentation](https://kubernetes.io/docs/concepts/configuration/secret/#using-secrets-as-environment-variables).
+
## Kubernetes concepts
### Namespaces
diff --git a/docs/ops/deployment/native_kubernetes.zh.md b/docs/ops/deployment/native_kubernetes.zh.md
index 14748a1..5461ec0 100644
--- a/docs/ops/deployment/native_kubernetes.zh.md
+++ b/docs/ops/deployment/native_kubernetes.zh.md
@@ -208,6 +208,81 @@ $ ./bin/flink run-application -p 8 -t kubernetes-application \
local:///opt/flink/usrlib/my-flink-job.jar
{% endhighlight %}
+## Using Secrets
+
+[Kubernetes Secrets](https://kubernetes.io/docs/concepts/configuration/secret/) is an object that contains a small amount of sensitive data such as a password, a token, or a key.
+Such information might otherwise be put in a Pod specification or in an image. Flink on Kubernetes can use Secrets in two ways:
+
+- Using Secrets as files from a pod;
+
+- Using Secrets as environment variables;
+
+### Using Secrets as files from a pod
+
+Here is an example of a Pod that mounts a Secret in a volume:
+
+{% highlight yaml %}
+apiVersion: v1
+kind: Pod
+metadata:
+ name: foo
+spec:
+ containers:
+ - name: foo
+ image: foo
+ volumeMounts:
+ - name: foo
+ mountPath: "/opt/foo"
+ volumes:
+ - name: foo
+ secret:
+ secretName: foo
+{% endhighlight %}
+
+By applying this yaml, each key in foo Secrets becomes the filename under `/opt/foo` path. Flink on Kubernetes can enable this feature by the following command:
+
+{% highlight bash %}
+$ ./bin/kubernetes-session.sh \
+ -Dkubernetes.cluster-id=<ClusterId> \
+ -Dkubernetes.container.image=<CustomImageName> \
+ -Dkubernetes.secrets=foo:/opt/foo
+{% endhighlight %}
+
+For more details see the [official Kubernetes documentation](https://kubernetes.io/docs/concepts/configuration/secret/#using-secrets-as-files-from-a-pod).
+
+### Using Secrets as environment variables
+
+Here is an example of a Pod that uses secrets from environment variables:
+
+{% highlight yaml %}
+apiVersion: v1
+kind: Pod
+metadata:
+ name: foo
+spec:
+ containers:
+ - name: foo
+ image: foo
+ env:
+ - name: FOO_ENV
+ valueFrom:
+ secretKeyRef:
+ name: foo_secret
+ key: foo_key
+{% endhighlight %}
+
+By applying this yaml, an environment variable named `FOO_ENV` is added into `foo` container, and `FOO_ENV` consumes the value of `foo_key` which is defined in Secrets `foo_secret`.
+Flink on Kubernetes can enable this feature by the following command:
+
+{% highlight bash %}
+$ ./bin/kubernetes-session.sh \
+ -Dkubernetes.cluster-id=<ClusterId> \
+ -Dkubernetes.container.image=<CustomImageName> \
+ -Dkubernetes.env.secretKeyRef=env:FOO_ENV,secret:foo_secret,key:foo_key
+{% endhighlight %}
+
+For more details see the [official Kubernetes documentation](https://kubernetes.io/docs/concepts/configuration/secret/#using-secrets-as-environment-variables).
+
## Kubernetes 概念
### 命名空间
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 d4aa93a..5bc7cc6 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
@@ -22,6 +22,8 @@ import org.apache.flink.annotation.PublicEvolving;
import org.apache.flink.annotation.docs.Documentation;
import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.ExternalResourceOptions;
+import org.apache.flink.configuration.description.Description;
+import org.apache.flink.configuration.description.TextElement;
import org.apache.flink.runtime.util.EnvironmentInformation;
import java.util.List;
@@ -221,6 +223,27 @@ public class KubernetesConfigOptions {
/** Defines the configuration key of that external resource in Kubernetes. This is used as a suffix in an actual config. */
public static final String EXTERNAL_RESOURCE_KUBERNETES_CONFIG_KEY_SUFFIX = "kubernetes.config-key";
+ public static final ConfigOption<Map<String, String>> KUBERNETES_SECRETS =
+ key("kubernetes.secrets")
+ .mapType()
+ .noDefaultValue()
+ .withDescription(
+ Description.builder()
+ .text("The user-specified secrets that will be mounted into Flink container. The value should be in " +
+ "the form of %s.", TextElement.code("foo:/opt/secrets-foo,bar:/opt/secrets-bar"))
+ .build());
+
+ public static final ConfigOption<List<Map<String, String>>> KUBERNETES_ENV_SECRET_KEY_REF =
+ key("kubernetes.env.secretKeyRef")
+ .mapType()
+ .asList()
+ .noDefaultValue()
+ .withDescription(
+ Description.builder()
+ .text("The user-specified secrets to set env variables in Flink container. The value should be in " +
+ "the form of %s.", TextElement.code("env:FOO_ENV,secret:foo_secret,key:foo_key;env:BAR_ENV,secret:bar_secret,key:bar_key"))
+ .build());
+
/**
* If configured, Flink will add "resources.limits.>config-key<" and "resources.requests.>config-key<" to the main
* container of TaskExecutor and set the value to {@link ExternalResourceOptions#EXTERNAL_RESOURCE_AMOUNT}.
diff --git a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/EnvSecretsDecorator.java b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/EnvSecretsDecorator.java
new file mode 100644
index 0000000..216e549
--- /dev/null
+++ b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/EnvSecretsDecorator.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.kubeclient.decorators;
+
+import org.apache.flink.kubernetes.kubeclient.FlinkPod;
+import org.apache.flink.kubernetes.kubeclient.parameters.AbstractKubernetesParameters;
+import org.apache.flink.kubernetes.kubeclient.resources.KubernetesSecretEnvVar;
+
+import io.fabric8.kubernetes.api.model.Container;
+import io.fabric8.kubernetes.api.model.ContainerBuilder;
+import io.fabric8.kubernetes.api.model.EnvVar;
+
+import java.util.List;
+import java.util.stream.Collectors;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * Support setting environment variables via Secrets.
+ */
+public class EnvSecretsDecorator extends AbstractKubernetesStepDecorator {
+
+ private final AbstractKubernetesParameters kubernetesComponentConf;
+
+ public EnvSecretsDecorator(AbstractKubernetesParameters kubernetesComponentConf) {
+ this.kubernetesComponentConf = checkNotNull(kubernetesComponentConf);
+ }
+
+ @Override
+ public FlinkPod decorateFlinkPod(FlinkPod flinkPod) {
+ final Container basicMainContainer = new ContainerBuilder(flinkPod.getMainContainer())
+ .addAllToEnv(getSecretEnvs())
+ .build();
+
+ return new FlinkPod.Builder(flinkPod)
+ .withMainContainer(basicMainContainer)
+ .build();
+ }
+
+ private List<EnvVar> getSecretEnvs() {
+ return kubernetesComponentConf.getEnvironmentsFromSecrets().stream()
+ .map(e -> KubernetesSecretEnvVar.fromMap(e).getInternalResource())
+ .collect(Collectors.toList());
+ }
+}
diff --git a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/MountSecretsDecorator.java b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/MountSecretsDecorator.java
new file mode 100644
index 0000000..e0a78ee0
--- /dev/null
+++ b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/MountSecretsDecorator.java
@@ -0,0 +1,92 @@
+/*
+ * 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.decorators;
+
+import org.apache.flink.kubernetes.kubeclient.FlinkPod;
+import org.apache.flink.kubernetes.kubeclient.parameters.AbstractKubernetesParameters;
+
+import io.fabric8.kubernetes.api.model.Container;
+import io.fabric8.kubernetes.api.model.ContainerBuilder;
+import io.fabric8.kubernetes.api.model.Pod;
+import io.fabric8.kubernetes.api.model.PodBuilder;
+import io.fabric8.kubernetes.api.model.Volume;
+import io.fabric8.kubernetes.api.model.VolumeBuilder;
+import io.fabric8.kubernetes.api.model.VolumeMount;
+import io.fabric8.kubernetes.api.model.VolumeMountBuilder;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * Support mounting Secrets on the JobManager or TaskManager pod..
+ */
+public class MountSecretsDecorator extends AbstractKubernetesStepDecorator {
+
+ private final AbstractKubernetesParameters kubernetesComponentConf;
+
+ public MountSecretsDecorator(AbstractKubernetesParameters kubernetesComponentConf) {
+ this.kubernetesComponentConf = checkNotNull(kubernetesComponentConf);
+ }
+
+ @Override
+ public FlinkPod decorateFlinkPod(FlinkPod flinkPod) {
+ final Pod podWithMount = decoratePod(flinkPod.getPod());
+ final Container containerWithMount = decorateMainContainer(flinkPod.getMainContainer());
+
+ return new FlinkPod.Builder(flinkPod)
+ .withPod(podWithMount)
+ .withMainContainer(containerWithMount)
+ .build();
+ }
+
+ private Container decorateMainContainer(Container container) {
+ final VolumeMount[] volumeMounts = kubernetesComponentConf.getSecretNamesToMountPaths().entrySet().stream()
+ .map(secretNameToPath -> new VolumeMountBuilder()
+ .withName(secretVolumeName(secretNameToPath.getKey()))
+ .withMountPath(secretNameToPath.getValue())
+ .build()
+ )
+ .toArray(VolumeMount[]::new);
+
+ return new ContainerBuilder(container)
+ .addToVolumeMounts(volumeMounts)
+ .build();
+ }
+
+ private Pod decoratePod(Pod pod) {
+ final Volume[] volumes = kubernetesComponentConf.getSecretNamesToMountPaths().keySet().stream()
+ .map(secretName -> new VolumeBuilder()
+ .withName(secretVolumeName(secretName))
+ .withNewSecret()
+ .withSecretName(secretName)
+ .endSecret()
+ .build()
+ )
+ .toArray(Volume[]::new);
+
+ return new PodBuilder(pod)
+ .editOrNewSpec()
+ .addToVolumes(volumes)
+ .endSpec()
+ .build();
+ }
+
+ private String secretVolumeName(String secretName) {
+ return secretName + "-volume";
+ }
+}
diff --git a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/factory/KubernetesJobManagerFactory.java b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/factory/KubernetesJobManagerFactory.java
index 66d55a3..e3d9c9b 100644
--- a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/factory/KubernetesJobManagerFactory.java
+++ b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/factory/KubernetesJobManagerFactory.java
@@ -20,6 +20,7 @@ package org.apache.flink.kubernetes.kubeclient.factory;
import org.apache.flink.kubernetes.kubeclient.FlinkPod;
import org.apache.flink.kubernetes.kubeclient.KubernetesJobManagerSpecification;
+import org.apache.flink.kubernetes.kubeclient.decorators.EnvSecretsDecorator;
import org.apache.flink.kubernetes.kubeclient.decorators.ExternalServiceDecorator;
import org.apache.flink.kubernetes.kubeclient.decorators.FlinkConfMountDecorator;
import org.apache.flink.kubernetes.kubeclient.decorators.HadoopConfMountDecorator;
@@ -27,6 +28,7 @@ import org.apache.flink.kubernetes.kubeclient.decorators.InitJobManagerDecorator
import org.apache.flink.kubernetes.kubeclient.decorators.InternalServiceDecorator;
import org.apache.flink.kubernetes.kubeclient.decorators.JavaCmdJobManagerDecorator;
import org.apache.flink.kubernetes.kubeclient.decorators.KubernetesStepDecorator;
+import org.apache.flink.kubernetes.kubeclient.decorators.MountSecretsDecorator;
import org.apache.flink.kubernetes.kubeclient.parameters.KubernetesJobManagerParameters;
import org.apache.flink.kubernetes.utils.Constants;
import org.apache.flink.kubernetes.utils.KubernetesUtils;
@@ -56,6 +58,8 @@ public class KubernetesJobManagerFactory {
final KubernetesStepDecorator[] stepDecorators = new KubernetesStepDecorator[] {
new InitJobManagerDecorator(kubernetesJobManagerParameters),
+ new EnvSecretsDecorator(kubernetesJobManagerParameters),
+ new MountSecretsDecorator(kubernetesJobManagerParameters),
new JavaCmdJobManagerDecorator(kubernetesJobManagerParameters),
new InternalServiceDecorator(kubernetesJobManagerParameters),
new ExternalServiceDecorator(kubernetesJobManagerParameters),
diff --git a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/factory/KubernetesTaskManagerFactory.java b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/factory/KubernetesTaskManagerFactory.java
index a7b8617..3f9debb 100644
--- a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/factory/KubernetesTaskManagerFactory.java
+++ b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/factory/KubernetesTaskManagerFactory.java
@@ -19,11 +19,13 @@
package org.apache.flink.kubernetes.kubeclient.factory;
import org.apache.flink.kubernetes.kubeclient.FlinkPod;
+import org.apache.flink.kubernetes.kubeclient.decorators.EnvSecretsDecorator;
import org.apache.flink.kubernetes.kubeclient.decorators.FlinkConfMountDecorator;
import org.apache.flink.kubernetes.kubeclient.decorators.HadoopConfMountDecorator;
import org.apache.flink.kubernetes.kubeclient.decorators.InitTaskManagerDecorator;
import org.apache.flink.kubernetes.kubeclient.decorators.JavaCmdTaskManagerDecorator;
import org.apache.flink.kubernetes.kubeclient.decorators.KubernetesStepDecorator;
+import org.apache.flink.kubernetes.kubeclient.decorators.MountSecretsDecorator;
import org.apache.flink.kubernetes.kubeclient.parameters.KubernetesTaskManagerParameters;
import org.apache.flink.kubernetes.kubeclient.resources.KubernetesPod;
@@ -40,6 +42,8 @@ public class KubernetesTaskManagerFactory {
final KubernetesStepDecorator[] stepDecorators = new KubernetesStepDecorator[] {
new InitTaskManagerDecorator(kubernetesTaskManagerParameters),
+ new EnvSecretsDecorator(kubernetesTaskManagerParameters),
+ new MountSecretsDecorator(kubernetesTaskManagerParameters),
new JavaCmdTaskManagerDecorator(kubernetesTaskManagerParameters),
new HadoopConfMountDecorator(kubernetesTaskManagerParameters),
new FlinkConfMountDecorator(kubernetesTaskManagerParameters)};
diff --git a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/parameters/AbstractKubernetesParameters.java b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/parameters/AbstractKubernetesParameters.java
index 85a33d6..6f539f9 100644
--- a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/parameters/AbstractKubernetesParameters.java
+++ b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/parameters/AbstractKubernetesParameters.java
@@ -177,4 +177,13 @@ public abstract class AbstractKubernetesParameters implements KubernetesParamete
return Optional.empty();
}
+
+ public Map<String, String> getSecretNamesToMountPaths() {
+ return flinkConfig.getOptional(KubernetesConfigOptions.KUBERNETES_SECRETS).orElse(Collections.emptyMap());
+ }
+
+ @Override
+ public List<Map<String, String>> getEnvironmentsFromSecrets() {
+ return flinkConfig.getOptional(KubernetesConfigOptions.KUBERNETES_ENV_SECRET_KEY_REF).orElse(Collections.emptyList());
+ }
}
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 2e5ee0a..f67c330 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
@@ -110,4 +110,14 @@ public interface KubernetesParameters {
* The local directory to locate the custom Hadoop configuration.
*/
Optional<String> getLocalHadoopConfigurationDirectory();
+
+ /**
+ * A collection of secret and path pairs that are mounted to the JobManager and TaskManager container(s).
+ */
+ Map<String, String> getSecretNamesToMountPaths();
+
+ /**
+ * A collection of customized environments that are attached to the JobManager and TaskManager container(s).
+ */
+ List<Map<String, String>> getEnvironmentsFromSecrets();
}
diff --git a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/resources/KubernetesSecretEnvVar.java b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/resources/KubernetesSecretEnvVar.java
new file mode 100644
index 0000000..5f5ab9e
--- /dev/null
+++ b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/resources/KubernetesSecretEnvVar.java
@@ -0,0 +1,50 @@
+/*
+ * 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.EnvVar;
+import io.fabric8.kubernetes.api.model.EnvVarBuilder;
+
+import java.util.Map;
+
+/**
+ * Represents EnvVar resource in Kubernetes.
+ */
+public class KubernetesSecretEnvVar extends KubernetesResource<EnvVar> {
+
+ private static final String ENV = "env";
+ private static final String SECRET = "secret";
+ private static final String KEY = "key";
+
+ private KubernetesSecretEnvVar(EnvVar envVar) {
+ super(envVar);
+ }
+
+ public static KubernetesSecretEnvVar fromMap(Map<String, String> stringMap) {
+ final EnvVarBuilder envVarBuilder = new EnvVarBuilder()
+ .withName(stringMap.get(ENV))
+ .withNewValueFrom()
+ .withNewSecretKeyRef()
+ .withName(stringMap.get(SECRET))
+ .withKey(stringMap.get(KEY))
+ .endSecretKeyRef()
+ .endValueFrom();
+ return new KubernetesSecretEnvVar(envVarBuilder.build());
+ }
+}
diff --git a/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/VolumeTestUtils.java b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/VolumeTestUtils.java
new file mode 100644
index 0000000..d5c1f72
--- /dev/null
+++ b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/VolumeTestUtils.java
@@ -0,0 +1,46 @@
+/*
+ * 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;
+
+import io.fabric8.kubernetes.api.model.Container;
+import io.fabric8.kubernetes.api.model.Pod;
+
+/**
+ * Utilities for the Kubernetes tests.
+ */
+public class VolumeTestUtils {
+
+ public static boolean podHasVolume(Pod pod, String volumeName){
+ return pod.getSpec().getVolumes().stream()
+ .filter(volume -> volume
+ .getName()
+ .equals(volumeName)
+ )
+ .count() == 1L;
+ }
+
+ public static boolean containerHasVolume(Container container, String volumeName, String mountPath){
+ return container.getVolumeMounts().stream()
+ .filter(volumeMount -> volumeMount.getName().equals(volumeName)
+ && volumeMount.getMountPath().equals(mountPath)
+ )
+ .count() == 1L;
+ }
+
+}
diff --git a/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/decorators/EnvSecretsDecoratorTest.java b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/decorators/EnvSecretsDecoratorTest.java
new file mode 100644
index 0000000..a77e1a3f
--- /dev/null
+++ b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/decorators/EnvSecretsDecoratorTest.java
@@ -0,0 +1,64 @@
+/*
+ * 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.decorators;
+
+import org.apache.flink.kubernetes.configuration.KubernetesConfigOptions;
+import org.apache.flink.kubernetes.kubeclient.FlinkPod;
+import org.apache.flink.kubernetes.kubeclient.KubernetesJobManagerTestBase;
+
+import io.fabric8.kubernetes.api.model.EnvVar;
+import org.junit.Test;
+
+import java.util.List;
+
+import static org.junit.Assert.assertEquals;
+
+/**
+ * General tests for the {@link EnvSecretsDecorator}.
+ */
+public class EnvSecretsDecoratorTest extends KubernetesJobManagerTestBase {
+
+ private static final String ENV_NAME = "MY_FOO";
+ private static final String ENV_SERCET_KEY = "env:MY_FOO,secret:foo,key:key_foo";
+
+ private EnvSecretsDecorator envSecretsDecorator;
+
+ @Override
+ protected void setupFlinkConfig() {
+ super.setupFlinkConfig();
+
+ this.flinkConfig.setString(KubernetesConfigOptions.KUBERNETES_ENV_SECRET_KEY_REF.key(), ENV_SERCET_KEY);
+ }
+
+ @Override
+ protected void onSetup() throws Exception {
+ super.onSetup();
+ this.envSecretsDecorator = new EnvSecretsDecorator(kubernetesJobManagerParameters);
+ }
+
+ @Test
+ public void testWhetherPodOrContainerIsDecorated() {
+ final FlinkPod resultFlinkPod = envSecretsDecorator.decorateFlinkPod(baseFlinkPod);
+ List<EnvVar> envVarList = resultFlinkPod.getMainContainer().getEnv();
+
+ assertEquals(1, envVarList.size());
+ assertEquals(ENV_NAME, envVarList.get(0).getName());
+ }
+
+}
diff --git a/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/decorators/MountSecretsDecoratorTest.java b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/decorators/MountSecretsDecoratorTest.java
new file mode 100644
index 0000000..f11d4a4
--- /dev/null
+++ b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/decorators/MountSecretsDecoratorTest.java
@@ -0,0 +1,67 @@
+/*
+ * 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.decorators;
+
+import org.apache.flink.kubernetes.VolumeTestUtils;
+import org.apache.flink.kubernetes.configuration.KubernetesConfigOptions;
+import org.apache.flink.kubernetes.kubeclient.FlinkPod;
+import org.apache.flink.kubernetes.kubeclient.KubernetesJobManagerTestBase;
+
+import org.junit.Test;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * General tests for the {@link MountSecretsDecorator}.
+ */
+public class MountSecretsDecoratorTest extends KubernetesJobManagerTestBase {
+
+ private static final String SECRET_NAME = "test";
+ private static final String SECRET_MOUNT_PATH = "/opt/flink/secret";
+
+ private MountSecretsDecorator mountSecretsDecorator;
+
+ @Override
+ protected void setupFlinkConfig() {
+ super.setupFlinkConfig();
+
+ this.flinkConfig.setString(KubernetesConfigOptions.KUBERNETES_SECRETS.key(), SECRET_NAME + ":" + SECRET_MOUNT_PATH);
+ }
+
+ @Override
+ protected void onSetup() throws Exception {
+ super.onSetup();
+
+ this.mountSecretsDecorator = new MountSecretsDecorator(kubernetesJobManagerParameters);
+ }
+
+ @Test
+ public void testWhetherPodOrContainerIsDecorated() {
+ final FlinkPod resultFlinkPod = mountSecretsDecorator.decorateFlinkPod(baseFlinkPod);
+
+ assertFalse(VolumeTestUtils.podHasVolume(baseFlinkPod.getPod(), SECRET_NAME + "-volume"));
+ assertTrue(VolumeTestUtils.podHasVolume(resultFlinkPod.getPod(), SECRET_NAME + "-volume"));
+
+ assertFalse(VolumeTestUtils.containerHasVolume(baseFlinkPod.getMainContainer(),
+ SECRET_NAME + "-volume", SECRET_MOUNT_PATH));
+ assertTrue(VolumeTestUtils.containerHasVolume(resultFlinkPod.getMainContainer(),
+ SECRET_NAME + "-volume", SECRET_MOUNT_PATH));
+ }
+}