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/12/12 15:56:16 UTC
(flink-kubernetes-operator) branch main updated: [FLINK-33454] Add IngressTls to support TLS within the managed Ingress, also add Label Passthrough
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
The following commit(s) were added to refs/heads/main by this push:
new 30097254 [FLINK-33454] Add IngressTls to support TLS within the managed Ingress, also add Label Passthrough
30097254 is described below
commit 300972547cc2d4f78434a8729f85866e64505b34
Author: Ryan van Huuksloot <ry...@gmail.com>
AuthorDate: Tue Dec 12 10:56:10 2023 -0500
[FLINK-33454] Add IngressTls to support TLS within the managed Ingress, also add Label Passthrough
---
docs/content/docs/custom-resource/reference.md | 2 +
.../kubernetes/operator/api/spec/IngressSpec.java | 8 +
.../kubernetes/operator/utils/IngressUtils.java | 24 ++
.../operator/utils/IngressUtilsTest.java | 263 +++++++++++++++++++++
.../crds/flinkdeployments.flink.apache.org-v1.yml | 15 ++
5 files changed, 312 insertions(+)
diff --git a/docs/content/docs/custom-resource/reference.md b/docs/content/docs/custom-resource/reference.md
index dfe6e0d7..649e5fbc 100644
--- a/docs/content/docs/custom-resource/reference.md
+++ b/docs/content/docs/custom-resource/reference.md
@@ -96,6 +96,8 @@ This page serves as a full reference for FlinkDeployment custom resource definit
| template | java.lang.String | Ingress template for the JobManager service. |
| className | java.lang.String | Ingress className for the Flink deployment. |
| annotations | java.util.Map<java.lang.String,java.lang.String> | Ingress annotations. |
+| labels | java.util.Map<java.lang.String,java.lang.String> | Ingress labels. |
+| tls | java.util.List<io.fabric8.kubernetes.api.model.networking.v1.IngressTLS> | Ingress tls. |
### JobManagerSpec
**Class**: org.apache.flink.kubernetes.operator.api.spec.JobManagerSpec
diff --git a/flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/spec/IngressSpec.java b/flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/spec/IngressSpec.java
index d0111e92..d5dce55d 100644
--- a/flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/spec/IngressSpec.java
+++ b/flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/spec/IngressSpec.java
@@ -20,11 +20,13 @@ package org.apache.flink.kubernetes.operator.api.spec;
import org.apache.flink.annotation.Experimental;
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import io.fabric8.kubernetes.api.model.networking.v1.IngressTLS;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
+import java.util.List;
import java.util.Map;
/** Ingress spec. */
@@ -44,4 +46,10 @@ public class IngressSpec {
/** Ingress annotations. */
private Map<String, String> annotations;
+
+ /** Ingress labels. */
+ private Map<String, String> labels;
+
+ /** Ingress tls. */
+ private List<IngressTLS> tls;
}
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/IngressUtils.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/IngressUtils.java
index 076a8480..9d3314d7 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/IngressUtils.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/IngressUtils.java
@@ -33,6 +33,7 @@ import io.fabric8.kubernetes.api.model.networking.v1.HTTPIngressRuleValueBuilder
import io.fabric8.kubernetes.api.model.networking.v1.IngressBuilder;
import io.fabric8.kubernetes.api.model.networking.v1.IngressRule;
import io.fabric8.kubernetes.api.model.networking.v1.IngressRuleBuilder;
+import io.fabric8.kubernetes.api.model.networking.v1beta1.IngressTLS;
import io.fabric8.kubernetes.client.KubernetesClient;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
@@ -43,7 +44,9 @@ import java.net.MalformedURLException;
import java.net.URL;
import java.util.Collections;
import java.util.List;
+import java.util.Optional;
import java.util.regex.Pattern;
+import java.util.stream.Collectors;
/** Ingress utilities. */
public class IngressUtils {
@@ -93,24 +96,45 @@ public class IngressUtils {
if (ingressInNetworkingV1(client)) {
return new IngressBuilder()
.withNewMetadata()
+ .withLabels(spec.getIngress().getLabels())
.withAnnotations(spec.getIngress().getAnnotations())
.withName(objectMeta.getName())
.withNamespace(objectMeta.getNamespace())
.endMetadata()
.withNewSpec()
.withIngressClassName(spec.getIngress().getClassName())
+ .withTls(spec.getIngress().getTls())
.withRules(getIngressRule(objectMeta, spec, effectiveConfig))
.endSpec()
.build();
} else {
+ List<IngressTLS> ingressTLS =
+ Optional.ofNullable(spec.getIngress().getTls())
+ .map(
+ list ->
+ list.stream()
+ .map(
+ v1Tls -> {
+ IngressTLS v1beta1Tls =
+ new IngressTLS();
+ v1beta1Tls.setHosts(
+ v1Tls.getHosts());
+ v1beta1Tls.setSecretName(
+ v1Tls.getSecretName());
+ return v1beta1Tls;
+ })
+ .collect(Collectors.toList()))
+ .orElse(Collections.emptyList());
return new io.fabric8.kubernetes.api.model.networking.v1beta1.IngressBuilder()
.withNewMetadata()
.withAnnotations(spec.getIngress().getAnnotations())
+ .withLabels(spec.getIngress().getLabels())
.withName(objectMeta.getName())
.withNamespace(objectMeta.getNamespace())
.endMetadata()
.withNewSpec()
.withIngressClassName(spec.getIngress().getClassName())
+ .withTls(ingressTLS)
.withRules(getIngressRuleForV1beta1(objectMeta, spec, effectiveConfig))
.endSpec()
.build();
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/utils/IngressUtilsTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/utils/IngressUtilsTest.java
index fd35661f..2257ee21 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/utils/IngressUtilsTest.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/utils/IngressUtilsTest.java
@@ -27,12 +27,14 @@ import org.apache.flink.kubernetes.operator.exception.ReconciliationException;
import io.fabric8.kubernetes.api.model.networking.v1.Ingress;
import io.fabric8.kubernetes.api.model.networking.v1.IngressRule;
+import io.fabric8.kubernetes.api.model.networking.v1.IngressTLS;
import io.fabric8.kubernetes.client.KubernetesClient;
import io.fabric8.kubernetes.client.server.mock.EnableKubernetesMockClient;
import org.apache.commons.lang3.StringUtils;
import org.junit.jupiter.api.Test;
import java.net.URL;
+import java.util.ArrayList;
import java.util.List;
import java.util.Map;
@@ -256,4 +258,265 @@ public class IngressUtilsTest {
ReconciliationException.class,
() -> IngressUtils.getIngressUrl("example.com:port", "basic-ingress", "default"));
}
+
+ @Test
+ public void testIngressTls() {
+ FlinkDeployment appCluster = TestUtils.buildApplicationCluster();
+ Configuration config =
+ new FlinkConfigManager(new Configuration())
+ .getDeployConfig(appCluster.getMetadata(), appCluster.getSpec());
+
+ // no tls when tls spec is empty
+ IngressSpec.IngressSpecBuilder builder = IngressSpec.builder();
+ builder.template("{{name}}.{{namespace}}.example.com");
+ builder.tls(new ArrayList<>());
+ appCluster.getSpec().setIngress(builder.build());
+ IngressUtils.updateIngressRules(
+ appCluster.getMetadata(), appCluster.getSpec(), config, client);
+ Ingress ingress = null;
+ io.fabric8.kubernetes.api.model.networking.v1beta1.Ingress ingressV1beta1 = null;
+ if (IngressUtils.ingressInNetworkingV1(client)) {
+ ingress =
+ client.network()
+ .v1()
+ .ingresses()
+ .inNamespace(appCluster.getMetadata().getNamespace())
+ .withName(appCluster.getMetadata().getName())
+ .get();
+ } else {
+ ingressV1beta1 =
+ client.network()
+ .v1beta1()
+ .ingresses()
+ .inNamespace(appCluster.getMetadata().getNamespace())
+ .withName(appCluster.getMetadata().getName())
+ .get();
+ }
+ List<IngressTLS> tls = null;
+ List<io.fabric8.kubernetes.api.model.networking.v1beta1.IngressTLS> tlsV1beta1 = null;
+ if (IngressUtils.ingressInNetworkingV1(client)) {
+ tls = ingress.getSpec().getTls();
+ } else {
+ tlsV1beta1 = ingressV1beta1.getSpec().getTls();
+ }
+ assertEquals(
+ 0, IngressUtils.ingressInNetworkingV1(client) ? tls.size() : tlsV1beta1.size());
+
+ // no tls when hosts spec is empty
+ builder.template("{{name}}.{{namespace}}.example.com");
+ IngressTLS ingressTlsSpecSecretOnly = new IngressTLS();
+ ingressTlsSpecSecretOnly.setSecretName("secret");
+ builder.tls(List.of(ingressTlsSpecSecretOnly));
+ appCluster.getSpec().setIngress(builder.build());
+ IngressUtils.updateIngressRules(
+ appCluster.getMetadata(), appCluster.getSpec(), config, client);
+ if (IngressUtils.ingressInNetworkingV1(client)) {
+ ingress =
+ client.network()
+ .v1()
+ .ingresses()
+ .inNamespace(appCluster.getMetadata().getNamespace())
+ .withName(appCluster.getMetadata().getName())
+ .get();
+ } else {
+ ingressV1beta1 =
+ client.network()
+ .v1beta1()
+ .ingresses()
+ .inNamespace(appCluster.getMetadata().getNamespace())
+ .withName(appCluster.getMetadata().getName())
+ .get();
+ }
+ if (IngressUtils.ingressInNetworkingV1(client)) {
+ tls = ingress.getSpec().getTls();
+ } else {
+ tlsV1beta1 = ingressV1beta1.getSpec().getTls();
+ }
+ assertEquals(
+ 1, IngressUtils.ingressInNetworkingV1(client) ? tls.size() : tlsV1beta1.size());
+ assertEquals(
+ "secret",
+ IngressUtils.ingressInNetworkingV1(client)
+ ? tls.get(0).getSecretName()
+ : tlsV1beta1.get(0).getSecretName());
+
+ // tls with no secretName
+ builder.template("{{name}}.{{namespace}}.example.com");
+ IngressTLS ingressTlsSpecHostsOnly = new IngressTLS();
+ ingressTlsSpecHostsOnly.setHosts(List.of("example.com"));
+ builder.tls(List.of(ingressTlsSpecHostsOnly));
+ appCluster.getSpec().setIngress(builder.build());
+ IngressUtils.updateIngressRules(
+ appCluster.getMetadata(), appCluster.getSpec(), config, client);
+ if (IngressUtils.ingressInNetworkingV1(client)) {
+ ingress =
+ client.network()
+ .v1()
+ .ingresses()
+ .inNamespace(appCluster.getMetadata().getNamespace())
+ .withName(appCluster.getMetadata().getName())
+ .get();
+ } else {
+ ingressV1beta1 =
+ client.network()
+ .v1beta1()
+ .ingresses()
+ .inNamespace(appCluster.getMetadata().getNamespace())
+ .withName(appCluster.getMetadata().getName())
+ .get();
+ }
+ if (IngressUtils.ingressInNetworkingV1(client)) {
+ tls = ingress.getSpec().getTls();
+ } else {
+ tlsV1beta1 = ingressV1beta1.getSpec().getTls();
+ }
+ assertEquals(
+ 1, IngressUtils.ingressInNetworkingV1(client) ? tls.size() : tlsV1beta1.size());
+ assertEquals(
+ "example.com",
+ IngressUtils.ingressInNetworkingV1(client)
+ ? tls.get(0).getHosts().get(0)
+ : tlsV1beta1.get(0).getHosts().get(0));
+
+ // tls with secretName and hosts
+ builder.template("{{name}}.{{namespace}}.example.com");
+ IngressTLS ingressTlsSpecSingleTLSWithHost =
+ new IngressTLS(List.of("example.com"), "secret");
+ builder.tls(List.of(ingressTlsSpecSingleTLSWithHost));
+ appCluster.getSpec().setIngress(builder.build());
+ IngressUtils.updateIngressRules(
+ appCluster.getMetadata(), appCluster.getSpec(), config, client);
+ if (IngressUtils.ingressInNetworkingV1(client)) {
+ ingress =
+ client.network()
+ .v1()
+ .ingresses()
+ .inNamespace(appCluster.getMetadata().getNamespace())
+ .withName(appCluster.getMetadata().getName())
+ .get();
+ } else {
+ ingressV1beta1 =
+ client.network()
+ .v1beta1()
+ .ingresses()
+ .inNamespace(appCluster.getMetadata().getNamespace())
+ .withName(appCluster.getMetadata().getName())
+ .get();
+ }
+ if (IngressUtils.ingressInNetworkingV1(client)) {
+ tls = ingress.getSpec().getTls();
+ } else {
+ tlsV1beta1 = ingressV1beta1.getSpec().getTls();
+ }
+ assertEquals(
+ 1, IngressUtils.ingressInNetworkingV1(client) ? tls.size() : tlsV1beta1.size());
+ if (IngressUtils.ingressInNetworkingV1(client)) {
+ assertEquals("secret", tls.get(0).getSecretName());
+ assertEquals(1, tls.get(0).getHosts().size());
+ assertEquals("example.com", tls.get(0).getHosts().get(0));
+ } else {
+ assertEquals("secret", tlsV1beta1.get(0).getSecretName());
+ assertEquals(1, tlsV1beta1.get(0).getHosts().size());
+ assertEquals("example.com", tlsV1beta1.get(0).getHosts().get(0));
+ }
+
+ // tls with secretName and multiple hosts
+ builder.template("{{name}}.{{namespace}}.example.com");
+ IngressTLS ingressTlsSpecSingleTLSWithHosts =
+ new IngressTLS(List.of("example.com", "example2.com"), "secret");
+ builder.tls(List.of(ingressTlsSpecSingleTLSWithHosts));
+ appCluster.getSpec().setIngress(builder.build());
+ IngressUtils.updateIngressRules(
+ appCluster.getMetadata(), appCluster.getSpec(), config, client);
+ if (IngressUtils.ingressInNetworkingV1(client)) {
+ ingress =
+ client.network()
+ .v1()
+ .ingresses()
+ .inNamespace(appCluster.getMetadata().getNamespace())
+ .withName(appCluster.getMetadata().getName())
+ .get();
+ } else {
+ ingressV1beta1 =
+ client.network()
+ .v1beta1()
+ .ingresses()
+ .inNamespace(appCluster.getMetadata().getNamespace())
+ .withName(appCluster.getMetadata().getName())
+ .get();
+ }
+ if (IngressUtils.ingressInNetworkingV1(client)) {
+ tls = ingress.getSpec().getTls();
+ } else {
+ tlsV1beta1 = ingressV1beta1.getSpec().getTls();
+ }
+ assertEquals(
+ 1, IngressUtils.ingressInNetworkingV1(client) ? tls.size() : tlsV1beta1.size());
+ if (IngressUtils.ingressInNetworkingV1(client)) {
+ assertEquals("secret", tls.get(0).getSecretName());
+ assertEquals(2, tls.get(0).getHosts().size());
+ assertEquals("example.com", tls.get(0).getHosts().get(0));
+ assertEquals("example2.com", tls.get(0).getHosts().get(1));
+ } else {
+ assertEquals("secret", tlsV1beta1.get(0).getSecretName());
+ assertEquals(2, tlsV1beta1.get(0).getHosts().size());
+ assertEquals("example.com", tlsV1beta1.get(0).getHosts().get(0));
+ assertEquals("example2.com", tlsV1beta1.get(0).getHosts().get(1));
+ }
+
+ // tls with secretName and multiple hosts and multiple tls
+ builder.template("{{name}}.{{namespace}}.example.com");
+ IngressTLS ingressTlsSpecMultipleTLSWithHosts1 =
+ new IngressTLS(List.of("example.com", "example2.com"), "secret");
+ IngressTLS ingressTlsSpecMultipleTLSWithHosts2 =
+ new IngressTLS(List.of("example3.com", "example4.com"), "secret2");
+ builder.tls(
+ List.of(ingressTlsSpecMultipleTLSWithHosts1, ingressTlsSpecMultipleTLSWithHosts2));
+ appCluster.getSpec().setIngress(builder.build());
+ IngressUtils.updateIngressRules(
+ appCluster.getMetadata(), appCluster.getSpec(), config, client);
+ if (IngressUtils.ingressInNetworkingV1(client)) {
+ ingress =
+ client.network()
+ .v1()
+ .ingresses()
+ .inNamespace(appCluster.getMetadata().getNamespace())
+ .withName(appCluster.getMetadata().getName())
+ .get();
+ } else {
+ ingressV1beta1 =
+ client.network()
+ .v1beta1()
+ .ingresses()
+ .inNamespace(appCluster.getMetadata().getNamespace())
+ .withName(appCluster.getMetadata().getName())
+ .get();
+ }
+ if (IngressUtils.ingressInNetworkingV1(client)) {
+ tls = ingress.getSpec().getTls();
+ } else {
+ tlsV1beta1 = ingressV1beta1.getSpec().getTls();
+ }
+ assertEquals(
+ 2, IngressUtils.ingressInNetworkingV1(client) ? tls.size() : tlsV1beta1.size());
+ if (IngressUtils.ingressInNetworkingV1(client)) {
+ assertEquals("secret", tls.get(0).getSecretName());
+ assertEquals(2, tls.get(0).getHosts().size());
+ assertEquals("example.com", tls.get(0).getHosts().get(0));
+ assertEquals("example2.com", tls.get(0).getHosts().get(1));
+ assertEquals("secret2", tls.get(1).getSecretName());
+ assertEquals(2, tls.get(1).getHosts().size());
+ assertEquals("example3.com", tls.get(1).getHosts().get(0));
+ assertEquals("example4.com", tls.get(1).getHosts().get(1));
+ } else {
+ assertEquals("secret", tlsV1beta1.get(0).getSecretName());
+ assertEquals(2, tlsV1beta1.get(0).getHosts().size());
+ assertEquals("example.com", tlsV1beta1.get(0).getHosts().get(0));
+ assertEquals("example2.com", tlsV1beta1.get(0).getHosts().get(1));
+ assertEquals("secret2", tlsV1beta1.get(1).getSecretName());
+ assertEquals(2, tlsV1beta1.get(1).getHosts().size());
+ assertEquals("example3.com", tlsV1beta1.get(1).getHosts().get(0));
+ assertEquals("example4.com", tlsV1beta1.get(1).getHosts().get(1));
+ }
+ }
}
diff --git a/helm/flink-kubernetes-operator/crds/flinkdeployments.flink.apache.org-v1.yml b/helm/flink-kubernetes-operator/crds/flinkdeployments.flink.apache.org-v1.yml
index 43af6e61..14de115c 100644
--- a/helm/flink-kubernetes-operator/crds/flinkdeployments.flink.apache.org-v1.yml
+++ b/helm/flink-kubernetes-operator/crds/flinkdeployments.flink.apache.org-v1.yml
@@ -57,6 +57,21 @@ spec:
additionalProperties:
type: string
type: object
+ labels:
+ additionalProperties:
+ type: string
+ type: object
+ tls:
+ items:
+ properties:
+ hosts:
+ items:
+ type: string
+ type: array
+ secretName:
+ type: string
+ type: object
+ type: array
type: object
podTemplate:
properties: