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 2025/08/15 12:34:43 UTC
(flink-kubernetes-operator) branch main updated: [FLINK-37193] Ingress recreated after perform a change to CR FlinkDeployment
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 bb58a0a0 [FLINK-37193] Ingress recreated after perform a change to CR FlinkDeployment
bb58a0a0 is described below
commit bb58a0a0388fbb080e25cd91ba5b061c5d00966e
Author: Attila Mészáros <a_...@apple.com>
AuthorDate: Fri Aug 15 14:34:37 2025 +0200
[FLINK-37193] Ingress recreated after perform a change to CR FlinkDeployment
---
.../flink/kubernetes/operator/FlinkOperator.java | 15 ++-
.../controller/FlinkDeploymentController.java | 3 +-
.../deployment/ApplicationReconciler.java | 3 +-
.../reconciler/deployment/SessionReconciler.java | 3 +-
.../operator/utils/EventSourceUtils.java | 28 ++++-
.../kubernetes/operator/utils/IngressUtils.java | 51 +++++---
.../kubernetes/operator/FlinkOperatorTest.java | 18 +--
.../controller/FlinkDeploymentControllerTest.java | 4 +
.../deployment/SessionReconcilerTest.java | 2 +-
.../operator/utils/IngressUtilsTest.java | 85 ++++++++++----
.../operator/utils/TestingJosdkContext.java | 130 +++++++++++++++++++++
11 files changed, 282 insertions(+), 60 deletions(-)
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/FlinkOperator.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/FlinkOperator.java
index 06c361ae..e2cca039 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/FlinkOperator.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/FlinkOperator.java
@@ -91,6 +91,11 @@ public class FlinkOperator {
private final Configuration baseConfig;
public FlinkOperator(@Nullable Configuration conf) {
+ this(conf, null);
+ }
+
+ @VisibleForTesting
+ FlinkOperator(@Nullable Configuration conf, KubernetesClient client) {
this.configManager =
conf != null
? new FlinkConfigManager(conf) // For testing only
@@ -100,9 +105,13 @@ public class FlinkOperator {
baseConfig = configManager.getDefaultConfig();
this.metricGroup = OperatorMetricUtils.initOperatorMetrics(baseConfig);
- this.client =
- KubernetesClientUtils.getKubernetesClient(
- configManager.getOperatorConfiguration(), this.metricGroup);
+ if (client == null) {
+ this.client =
+ KubernetesClientUtils.getKubernetesClient(
+ configManager.getOperatorConfiguration(), this.metricGroup);
+ } else {
+ this.client = client;
+ }
this.operator = createOperator();
this.validators = ValidatorUtils.discoverValidators(configManager);
this.listeners = ListenerUtils.discoverListeners(configManager);
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkDeploymentController.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkDeploymentController.java
index e5d418b8..165c313d 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkDeploymentController.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkDeploymentController.java
@@ -184,7 +184,7 @@ public class FlinkDeploymentController
List<EventSource<?, FlinkDeployment>> eventSources = new ArrayList<>();
eventSources.add(EventSourceUtils.getSessionJobInformerEventSource(context));
eventSources.add(EventSourceUtils.getDeploymentInformerEventSource(context));
-
+ eventSources.add(EventSourceUtils.getIngressInformerEventSource(context));
if (KubernetesClientUtils.isCrdInstalled(FlinkStateSnapshot.class)) {
eventSources.add(
EventSourceUtils.getStateSnapshotForFlinkResourceInformerEventSource(context));
@@ -192,7 +192,6 @@ public class FlinkDeploymentController
LOG.warn(
"Could not initialize informer for snapshots as the CRD has not been installed!");
}
-
return eventSources;
}
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconciler.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconciler.java
index 199a5030..f2a1b71e 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconciler.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconciler.java
@@ -190,8 +190,7 @@ public class ApplicationReconciler
status.getJobStatus().setState(org.apache.flink.api.common.JobStatus.RECONCILING);
status.setJobManagerDeploymentStatus(JobManagerDeploymentStatus.DEPLOYING);
- IngressUtils.updateIngressRules(
- relatedResource.getMetadata(), spec, deployConfig, ctx.getKubernetesClient());
+ IngressUtils.reconcileIngress(ctx, spec, deployConfig, ctx.getKubernetesClient());
}
private void setJobIdIfNecessary(
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/SessionReconciler.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/SessionReconciler.java
index 809fbfbd..139a9993 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/SessionReconciler.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/SessionReconciler.java
@@ -106,8 +106,7 @@ public class SessionReconciler
setOwnerReference(cr, deployConfig);
ctx.getFlinkService().submitSessionCluster(deployConfig);
cr.getStatus().setJobManagerDeploymentStatus(JobManagerDeploymentStatus.DEPLOYING);
- IngressUtils.updateIngressRules(
- cr.getMetadata(), spec, deployConfig, ctx.getKubernetesClient());
+ IngressUtils.reconcileIngress(ctx, spec, deployConfig, ctx.getKubernetesClient());
}
@Override
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/EventSourceUtils.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/EventSourceUtils.java
index f53096a0..efcf2ab6 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/EventSourceUtils.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/EventSourceUtils.java
@@ -30,6 +30,7 @@ import org.apache.flink.kubernetes.utils.Constants;
import io.fabric8.kubernetes.api.model.HasMetadata;
import io.fabric8.kubernetes.api.model.apps.Deployment;
+import io.fabric8.kubernetes.api.model.networking.v1.Ingress;
import io.javaoperatorsdk.operator.api.config.informer.InformerEventSourceConfiguration;
import io.javaoperatorsdk.operator.api.reconciler.EventSourceContext;
import io.javaoperatorsdk.operator.processing.event.ResourceID;
@@ -45,9 +46,13 @@ import java.util.Set;
import java.util.stream.Collectors;
import java.util.stream.Stream;
+import static org.apache.flink.kubernetes.operator.utils.IngressUtils.ingressInNetworkingV1;
+
/** Utility class to locate secondary resources. */
public class EventSourceUtils {
+ public static final String LABEL_COMPONENT_INGRESS = "ingress";
+
private static final String FLINK_DEPLOYMENT_IDX = FlinkDeploymentController.class.getName();
private static final String FLINK_SESSIONJOB_IDX = FlinkSessionJobController.class.getName();
private static final String FLINK_STATE_SNAPSHOT_IDX = FlinkStateSnapshot.class.getName();
@@ -95,7 +100,28 @@ public class EventSourceUtils {
var configuration =
InformerEventSourceConfiguration.from(Deployment.class, FlinkDeployment.class)
.withLabelSelector(labelSelector)
- .withSecondaryToPrimaryMapper(fromLabel(Constants.LABEL_APP_KEY))
+ .withNamespacesInheritedFromController()
+ .withFollowControllerNamespacesChanges(true)
+ .build();
+
+ return new InformerEventSource<>(configuration, context);
+ }
+
+ public static InformerEventSource<?, FlinkDeployment> getIngressInformerEventSource(
+ EventSourceContext<FlinkDeployment> context) {
+ final String labelSelector =
+ Map.of(Constants.LABEL_COMPONENT_KEY, LABEL_COMPONENT_INGRESS).entrySet().stream()
+ .map(Object::toString)
+ .collect(Collectors.joining(","));
+
+ var ingressClass =
+ ingressInNetworkingV1(context.getClient())
+ ? Ingress.class
+ : io.fabric8.kubernetes.api.model.networking.v1beta1.Ingress.class;
+
+ var configuration =
+ InformerEventSourceConfiguration.from(ingressClass, FlinkDeployment.class)
+ .withLabelSelector(labelSelector)
.withNamespacesInheritedFromController()
.withFollowControllerNamespacesChanges(true)
.build();
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 9d3314d7..89036d05 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
@@ -20,7 +20,9 @@ package org.apache.flink.kubernetes.operator.utils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.kubernetes.operator.api.spec.FlinkDeploymentSpec;
+import org.apache.flink.kubernetes.operator.controller.FlinkResourceContext;
import org.apache.flink.kubernetes.operator.exception.ReconciliationException;
+import org.apache.flink.kubernetes.utils.Constants;
import org.apache.flink.util.Preconditions;
import io.fabric8.kubernetes.api.model.HasMetadata;
@@ -28,13 +30,14 @@ import io.fabric8.kubernetes.api.model.IntOrString;
import io.fabric8.kubernetes.api.model.ObjectMeta;
import io.fabric8.kubernetes.api.model.OwnerReference;
import io.fabric8.kubernetes.api.model.OwnerReferenceBuilder;
-import io.fabric8.kubernetes.api.model.apps.Deployment;
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.Ingress;
import io.fabric8.kubernetes.api.model.networking.v1beta1.IngressTLS;
import io.fabric8.kubernetes.client.KubernetesClient;
+import io.fabric8.kubernetes.client.dsl.NonDeletingOperation;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -43,11 +46,15 @@ import java.lang.module.ModuleDescriptor;
import java.net.MalformedURLException;
import java.net.URL;
import java.util.Collections;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
import java.util.Optional;
import java.util.regex.Pattern;
import java.util.stream.Collectors;
+import static org.apache.flink.kubernetes.operator.utils.EventSourceUtils.LABEL_COMPONENT_INGRESS;
+
/** Ingress utilities. */
public class IngressUtils {
@@ -62,29 +69,32 @@ public class IngressUtils {
private static final Logger LOG = LoggerFactory.getLogger(IngressUtils.class);
- public static void updateIngressRules(
- ObjectMeta objectMeta,
+ public static void reconcileIngress(
+ FlinkResourceContext<?> ctx,
FlinkDeploymentSpec spec,
Configuration effectiveConfig,
KubernetesClient client) {
+ var objectMeta = ctx.getResource().getMetadata();
if (spec.getIngress() != null) {
HasMetadata ingress = getIngress(objectMeta, spec, effectiveConfig, client);
-
- Deployment deployment =
- client.apps()
- .deployments()
- .inNamespace(objectMeta.getNamespace())
- .withName(objectMeta.getName())
- .get();
- if (deployment == null) {
- LOG.error("Could not find deployment {}", objectMeta.getName());
+ setOwnerReference(ctx.getResource(), Collections.singletonList(ingress));
+ LOG.info("Updating ingress rules {}", ingress);
+ client.resource(ingress)
+ .inNamespace(objectMeta.getNamespace())
+ .createOr(NonDeletingOperation::update);
+ } else {
+ Optional<? extends HasMetadata> ingress;
+ if (ingressInNetworkingV1(client)) {
+ ingress =
+ ctx.getJosdkContext()
+ .getSecondaryResource(
+ io.fabric8.kubernetes.api.model.networking.v1.Ingress
+ .class);
} else {
- setOwnerReference(deployment, Collections.singletonList(ingress));
+ ingress = ctx.getJosdkContext().getSecondaryResource(Ingress.class);
}
-
- LOG.info("Updating ingress rules {}", ingress);
- client.resourceList(ingress).inNamespace(objectMeta.getNamespace()).createOrReplace();
+ ingress.ifPresent(i -> ctx.getKubernetesClient().resource(i).delete());
}
}
@@ -93,10 +103,15 @@ public class IngressUtils {
FlinkDeploymentSpec spec,
Configuration effectiveConfig,
KubernetesClient client) {
+ Map<String, String> labels =
+ spec.getIngress().getLabels() == null
+ ? new HashMap<>()
+ : new HashMap<>(spec.getIngress().getLabels());
+ labels.put(Constants.LABEL_COMPONENT_KEY, LABEL_COMPONENT_INGRESS);
if (ingressInNetworkingV1(client)) {
return new IngressBuilder()
.withNewMetadata()
- .withLabels(spec.getIngress().getLabels())
+ .withLabels(labels)
.withAnnotations(spec.getIngress().getAnnotations())
.withName(objectMeta.getName())
.withNamespace(objectMeta.getNamespace())
@@ -128,7 +143,7 @@ public class IngressUtils {
return new io.fabric8.kubernetes.api.model.networking.v1beta1.IngressBuilder()
.withNewMetadata()
.withAnnotations(spec.getIngress().getAnnotations())
- .withLabels(spec.getIngress().getLabels())
+ .withLabels(labels)
.withName(objectMeta.getName())
.withNamespace(objectMeta.getNamespace())
.endMetadata()
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/FlinkOperatorTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/FlinkOperatorTest.java
index 423b2af3..5d56518b 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/FlinkOperatorTest.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/FlinkOperatorTest.java
@@ -21,12 +21,13 @@ import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.IllegalConfigurationException;
import org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions;
+import io.fabric8.kubeapitest.junit.EnableKubeAPIServer;
import io.fabric8.kubernetes.client.Config;
+import io.fabric8.kubernetes.client.KubernetesClient;
import io.javaoperatorsdk.operator.RegisteredController;
import io.javaoperatorsdk.operator.api.config.ControllerConfiguration;
import io.javaoperatorsdk.operator.processing.event.rate.LinearRateLimiter;
import org.junit.jupiter.api.Assertions;
-import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
import java.util.concurrent.ThreadPoolExecutor;
@@ -39,14 +40,13 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
* ConfigurationServiceProvider) we write multiple tests as a single function, please provide
* ample comments.
*/
-public class FlinkOperatorTest {
- @BeforeAll
- public static void setAutoTryKubeConfig() {
- System.setProperty(Config.KUBERNETES_AUTH_TRYKUBECONFIG_SYSTEM_PROPERTY, "false");
- }
+@EnableKubeAPIServer
+class FlinkOperatorTest {
+
+ static KubernetesClient kubernetesClient;
@Test
- public void testConfigurationPassedToJOSDK() {
+ void testConfigurationPassedToJOSDK() {
var testParallelism = 42;
var testSelector = "flink=enabled";
var testLeaseName = "test-lease";
@@ -64,7 +64,7 @@ public class FlinkOperatorTest {
operatorConfig.set(
KubernetesOperatorConfigOptions.OPERATOR_LEADER_ELECTION_LEASE_NAME, testLeaseName);
- var testOperator = new FlinkOperator(operatorConfig);
+ var testOperator = new FlinkOperator(operatorConfig, kubernetesClient);
testOperator.registerDeploymentController();
testOperator.registerSessionJobController();
@@ -99,7 +99,7 @@ public class FlinkOperatorTest {
}
@Test
- public void testLeaderElectionConfig() {
+ void testLeaderElectionConfig() {
var operatorConfig = new Configuration();
operatorConfig.set(KubernetesOperatorConfigOptions.OPERATOR_LEADER_ELECTION_ENABLED, true);
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/FlinkDeploymentControllerTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/FlinkDeploymentControllerTest.java
index d0544b10..4d379139 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/FlinkDeploymentControllerTest.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/FlinkDeploymentControllerTest.java
@@ -68,6 +68,7 @@ import static org.apache.flink.kubernetes.operator.TestUtils.MAX_RECONCILE_TIMES
import static org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions.OPERATOR_JOB_UPGRADE_LAST_STATE_FALLBACK_ENABLED;
import static org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions.SNAPSHOT_RESOURCE_ENABLED;
import static org.apache.flink.kubernetes.operator.utils.EventRecorder.Reason.ValidationError;
+import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotEquals;
@@ -983,6 +984,9 @@ public class FlinkDeploymentControllerTest {
appWithIngress.getMetadata().getName(),
appWithIngress.getMetadata().getNamespace())
.getHost());
+ assertThat(ingress.getMetadata().getOwnerReferences()).hasSize(1);
+ assertThat(ingress.getMetadata().getOwnerReferences().get(0).getKind())
+ .isEqualTo(FlinkDeployment.class.getSimpleName());
}
@Test
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/deployment/SessionReconcilerTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/deployment/SessionReconcilerTest.java
index 9f84f2c1..284af221 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/deployment/SessionReconcilerTest.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/deployment/SessionReconcilerTest.java
@@ -75,7 +75,7 @@ public class SessionReconcilerTest extends OperatorTestBase {
public void testStartSession() throws Exception {
var count = new AtomicInteger(0);
flinkService =
- new TestingFlinkService() {
+ new TestingFlinkService(kubernetesClient) {
@Override
public void submitSessionCluster(Configuration conf) throws Exception {
super.submitSessionCluster(conf);
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 2257ee21..78b6c3a1 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
@@ -23,11 +23,14 @@ import org.apache.flink.kubernetes.operator.TestUtils;
import org.apache.flink.kubernetes.operator.api.FlinkDeployment;
import org.apache.flink.kubernetes.operator.api.spec.IngressSpec;
import org.apache.flink.kubernetes.operator.config.FlinkConfigManager;
+import org.apache.flink.kubernetes.operator.controller.FlinkDeploymentContext;
+import org.apache.flink.kubernetes.operator.controller.FlinkResourceContext;
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.api.model.networking.v1beta1.IngressBuilder;
import io.fabric8.kubernetes.client.KubernetesClient;
import io.fabric8.kubernetes.client.server.mock.EnableKubernetesMockClient;
import org.apache.commons.lang3.StringUtils;
@@ -38,6 +41,7 @@ import java.util.ArrayList;
import java.util.List;
import java.util.Map;
+import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
@@ -45,20 +49,27 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
/** Test class for {@link IngressUtils}. */
@EnableKubernetesMockClient(crud = true)
-public class IngressUtilsTest {
+class IngressUtilsTest {
KubernetesClient client;
+ TestingJosdkContext testingJosdkContext;
+
+ private FlinkResourceContext<?> createResourceContext(FlinkDeployment appCluster) {
+ testingJosdkContext = new TestingJosdkContext<>(client);
+ return new FlinkDeploymentContext(appCluster, testingJosdkContext, null, null, null, null);
+ }
+
@Test
- public void testIngress() {
+ void testIngress() {
FlinkDeployment appCluster = TestUtils.buildApplicationCluster();
Configuration config =
new FlinkConfigManager(new Configuration())
.getDeployConfig(appCluster.getMetadata(), appCluster.getSpec());
// no ingress when ingressDomain is empty
- IngressUtils.updateIngressRules(
- appCluster.getMetadata(), appCluster.getSpec(), config, client);
+ IngressUtils.reconcileIngress(
+ createResourceContext(appCluster), appCluster.getSpec(), config, client);
if (IngressUtils.ingressInNetworkingV1(client)) {
assertNull(
client.network()
@@ -81,8 +92,8 @@ public class IngressUtilsTest {
IngressSpec.IngressSpecBuilder builder = IngressSpec.builder();
builder.template("{{name}}.{{namespace}}.example.com");
appCluster.getSpec().setIngress(builder.build());
- IngressUtils.updateIngressRules(
- appCluster.getMetadata(), appCluster.getSpec(), config, client);
+ IngressUtils.reconcileIngress(
+ createResourceContext(appCluster), appCluster.getSpec(), config, client);
Ingress ingress = null;
io.fabric8.kubernetes.api.model.networking.v1beta1.Ingress ingressV1beta1 = null;
if (IngressUtils.ingressInNetworkingV1(client)) {
@@ -130,8 +141,8 @@ public class IngressUtilsTest {
builder.className("nginx");
builder.annotations(Map.of("nginx.ingress.kubernetes.io/rewrite-target", "/$2"));
appCluster.getSpec().setIngress(builder.build());
- IngressUtils.updateIngressRules(
- appCluster.getMetadata(), appCluster.getSpec(), config, client);
+ IngressUtils.reconcileIngress(
+ createResourceContext(appCluster), appCluster.getSpec(), config, client);
if (IngressUtils.ingressInNetworkingV1(client)) {
ingress =
client.network()
@@ -187,8 +198,8 @@ public class IngressUtilsTest {
builder.template("example.com/{{namespace}}/{{name}}(/|$)(.*)");
builder.className("nginx");
appCluster.getSpec().setIngress(builder.build());
- IngressUtils.updateIngressRules(
- appCluster.getMetadata(), appCluster.getSpec(), config, client);
+ IngressUtils.reconcileIngress(
+ createResourceContext(appCluster), appCluster.getSpec(), config, client);
if (IngressUtils.ingressInNetworkingV1(client)) {
ingress =
client.network()
@@ -271,8 +282,8 @@ public class IngressUtilsTest {
builder.template("{{name}}.{{namespace}}.example.com");
builder.tls(new ArrayList<>());
appCluster.getSpec().setIngress(builder.build());
- IngressUtils.updateIngressRules(
- appCluster.getMetadata(), appCluster.getSpec(), config, client);
+ IngressUtils.reconcileIngress(
+ createResourceContext(appCluster), appCluster.getSpec(), config, client);
Ingress ingress = null;
io.fabric8.kubernetes.api.model.networking.v1beta1.Ingress ingressV1beta1 = null;
if (IngressUtils.ingressInNetworkingV1(client)) {
@@ -308,8 +319,8 @@ public class IngressUtilsTest {
ingressTlsSpecSecretOnly.setSecretName("secret");
builder.tls(List.of(ingressTlsSpecSecretOnly));
appCluster.getSpec().setIngress(builder.build());
- IngressUtils.updateIngressRules(
- appCluster.getMetadata(), appCluster.getSpec(), config, client);
+ IngressUtils.reconcileIngress(
+ createResourceContext(appCluster), appCluster.getSpec(), config, client);
if (IngressUtils.ingressInNetworkingV1(client)) {
ingress =
client.network()
@@ -346,8 +357,8 @@ public class IngressUtilsTest {
ingressTlsSpecHostsOnly.setHosts(List.of("example.com"));
builder.tls(List.of(ingressTlsSpecHostsOnly));
appCluster.getSpec().setIngress(builder.build());
- IngressUtils.updateIngressRules(
- appCluster.getMetadata(), appCluster.getSpec(), config, client);
+ IngressUtils.reconcileIngress(
+ createResourceContext(appCluster), appCluster.getSpec(), config, client);
if (IngressUtils.ingressInNetworkingV1(client)) {
ingress =
client.network()
@@ -384,8 +395,8 @@ public class IngressUtilsTest {
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);
+ IngressUtils.reconcileIngress(
+ createResourceContext(appCluster), appCluster.getSpec(), config, client);
if (IngressUtils.ingressInNetworkingV1(client)) {
ingress =
client.network()
@@ -426,8 +437,8 @@ public class IngressUtilsTest {
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);
+ IngressUtils.reconcileIngress(
+ createResourceContext(appCluster), appCluster.getSpec(), config, client);
if (IngressUtils.ingressInNetworkingV1(client)) {
ingress =
client.network()
@@ -473,8 +484,8 @@ public class IngressUtilsTest {
builder.tls(
List.of(ingressTlsSpecMultipleTLSWithHosts1, ingressTlsSpecMultipleTLSWithHosts2));
appCluster.getSpec().setIngress(builder.build());
- IngressUtils.updateIngressRules(
- appCluster.getMetadata(), appCluster.getSpec(), config, client);
+ IngressUtils.reconcileIngress(
+ createResourceContext(appCluster), appCluster.getSpec(), config, client);
if (IngressUtils.ingressInNetworkingV1(client)) {
ingress =
client.network()
@@ -519,4 +530,34 @@ public class IngressUtilsTest {
assertEquals("example4.com", tlsV1beta1.get(1).getHosts().get(1));
}
}
+
+ @Test
+ void testDeletesIngress() {
+ FlinkDeployment appCluster = TestUtils.buildApplicationCluster();
+ appCluster.getSpec().setIngress(null);
+ io.fabric8.kubernetes.api.model.networking.v1beta1.Ingress ingress =
+ new IngressBuilder()
+ .withNewMetadata()
+ .withName(appCluster.getMetadata().getName())
+ .withNamespace(appCluster.getMetadata().getNamespace())
+ .endMetadata()
+ .build();
+ client.network().v1beta1().ingresses().resource(ingress).create();
+ var context = createResourceContext(appCluster);
+ testingJosdkContext.setSecondaryResources(
+ Map.of(
+ io.fabric8.kubernetes.api.model.networking.v1beta1.Ingress.class,
+ List.of(ingress)));
+
+ IngressUtils.reconcileIngress(context, appCluster.getSpec(), null, client);
+
+ var ingressV1beta1 =
+ client.network()
+ .v1beta1()
+ .ingresses()
+ .inNamespace(appCluster.getMetadata().getNamespace())
+ .withName(appCluster.getMetadata().getName())
+ .get();
+ assertThat(ingressV1beta1).isNull();
+ }
}
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/utils/TestingJosdkContext.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/utils/TestingJosdkContext.java
new file mode 100644
index 00000000..be87f3b5
--- /dev/null
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/utils/TestingJosdkContext.java
@@ -0,0 +1,130 @@
+/*
+ * 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.utils;
+
+import io.fabric8.kubernetes.api.model.HasMetadata;
+import io.fabric8.kubernetes.client.KubernetesClient;
+import io.javaoperatorsdk.operator.api.config.ControllerConfiguration;
+import io.javaoperatorsdk.operator.api.reconciler.Context;
+import io.javaoperatorsdk.operator.api.reconciler.IndexedResourceCache;
+import io.javaoperatorsdk.operator.api.reconciler.RetryInfo;
+import io.javaoperatorsdk.operator.api.reconciler.dependent.managed.ManagedWorkflowAndDependentResourceContext;
+import io.javaoperatorsdk.operator.processing.event.EventSourceRetriever;
+import lombok.Setter;
+
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+
+/**
+ * JOSDK Context for unit testing.
+ *
+ * @param <P>
+ */
+public class TestingJosdkContext<P extends HasMetadata> implements Context<P> {
+
+ @Setter private KubernetesClient kubernetesClient;
+
+ @Setter
+ private Map<Class<? extends HasMetadata>, List<? extends HasMetadata>> secondaryResources =
+ new ConcurrentHashMap<>();
+
+ public TestingJosdkContext() {}
+
+ public TestingJosdkContext(KubernetesClient kubernetesClient) {
+ this.kubernetesClient = kubernetesClient;
+ }
+
+ public TestingJosdkContext(
+ KubernetesClient kubernetesClient,
+ Map<Class<? extends HasMetadata>, List<? extends HasMetadata>> secondaryResources) {
+ this.kubernetesClient = kubernetesClient;
+ this.secondaryResources = secondaryResources;
+ }
+
+ @Override
+ public Optional<RetryInfo> getRetryInfo() {
+ return Optional.empty();
+ }
+
+ @Override
+ public <R> Set<R> getSecondaryResources(Class<R> expectedType) {
+ var res = secondaryResources.get(expectedType);
+ if (res == null) {
+ return Set.of();
+ }
+ return (Set<R>) new HashSet<>(res);
+ }
+
+ @Override
+ public <R> Optional<R> getSecondaryResource(Class<R> expectedType, String eventSourceName) {
+ var resources = getSecondaryResources(expectedType);
+ if (resources.isEmpty()) {
+ return Optional.empty();
+ } else if (resources.size() == 1) {
+ return Optional.of(resources.iterator().next());
+ } else {
+ throw new IllegalStateException("Multiple secondary resources found: " + resources);
+ }
+ }
+
+ @Override
+ public ControllerConfiguration<P> getControllerConfiguration() {
+ return null;
+ }
+
+ @Override
+ public ManagedWorkflowAndDependentResourceContext managedWorkflowAndDependentResourceContext() {
+ return null;
+ }
+
+ @Override
+ public EventSourceRetriever<P> eventSourceRetriever() {
+ return null;
+ }
+
+ @Override
+ public KubernetesClient getClient() {
+ return kubernetesClient;
+ }
+
+ @Override
+ public ExecutorService getWorkflowExecutorService() {
+ return null;
+ }
+
+ @Override
+ public P getPrimaryResource() {
+ return null;
+ }
+
+ @Override
+ public IndexedResourceCache<P> getPrimaryCache() {
+ return null;
+ }
+
+ @Override
+ public boolean isNextReconciliationImminent() {
+ return false;
+ }
+}