Feature flag for managing ingress

csviri · csviri · commit e2e5e045f2c4 · 2025-08-22T12:34:55.000+02:00
Signed-off-by: Attila Mészáros &lt;a_meszaros@apple.com&gt;
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
@@ -199,7 +199,8 @@ void registerDeploymentController() {
                         observerFactory,
                         statusRecorder,
                         eventRecorder,
-                        canaryResourceManager);
+                        canaryResourceManager,
+                        configManager);
         registeredControllers.add(operator.register(controller, this::overrideControllerConfigs));
     }
 
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkOperatorConfiguration.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkOperatorConfiguration.java
@@ -79,6 +79,7 @@ public class FlinkOperatorConfiguration {
     Duration slowRequestThreshold;
     int reportedExceptionEventsMaxCount;
     int reportedExceptionEventsMaxStackTraceLength;
+    boolean manageIngress;
 
     public static FlinkOperatorConfiguration fromConfiguration(Configuration operatorConfig) {
         Duration reconcileInterval =
@@ -203,6 +204,9 @@ public static FlinkOperatorConfiguration fromConfiguration(Configuration operato
                 operatorConfig.get(
                         KubernetesOperatorConfigOptions.OPERATOR_EVENT_EXCEPTION_STACKTRACE_LINES);
 
+        boolean manageIngress =
+                operatorConfig.get(KubernetesOperatorConfigOptions.OPERATOR_MANAGE_INGRESS);
+
         return new FlinkOperatorConfiguration(
                 reconcileInterval,
                 reconcilerMaxParallelism,
@@ -234,7 +238,8 @@ public static FlinkOperatorConfiguration fromConfiguration(Configuration operato
                 snapshotResourcesEnabled,
                 slowRequestThreshold,
                 reportedExceptionEventsMaxCount,
-                reportedExceptionEventsMaxStackTraceLength);
+                reportedExceptionEventsMaxStackTraceLength,
+                manageIngress);
     }
 
     private static GenericRetry getRetryConfig(Configuration conf) {
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/KubernetesOperatorConfigOptions.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/KubernetesOperatorConfigOptions.java
@@ -679,4 +679,12 @@ public static String operatorConfigKey(String key) {
                     .defaultValue(10)
                     .withDescription(
                             "Maximum number of exception-related Kubernetes events emitted per reconciliation cycle.");
+
+    @Documentation.Section(SECTION_ADVANCED)
+    public static final ConfigOption<Boolean> OPERATOR_MANAGE_INGRESS =
+            operatorConfig("ingress.manage")
+                    .booleanType()
+                    .defaultValue(true)
+                    .withDescription(
+                            "Feature flag if operator will manage the Ingress resource. If false, no InformerEventSource will be registered for Ingress, and Ingress won't be created.");
 }
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
@@ -23,6 +23,7 @@
 import org.apache.flink.kubernetes.operator.api.lifecycle.ResourceLifecycleState;
 import org.apache.flink.kubernetes.operator.api.status.FlinkDeploymentStatus;
 import org.apache.flink.kubernetes.operator.api.status.JobManagerDeploymentStatus;
+import org.apache.flink.kubernetes.operator.config.FlinkConfigManager;
 import org.apache.flink.kubernetes.operator.exception.DeploymentFailedException;
 import org.apache.flink.kubernetes.operator.exception.ReconciliationException;
 import org.apache.flink.kubernetes.operator.exception.UpgradeFailureException;
@@ -69,6 +70,7 @@ public class FlinkDeploymentController
     private final StatusRecorder<FlinkDeployment, FlinkDeploymentStatus> statusRecorder;
     private final EventRecorder eventRecorder;
     private final CanaryResourceManager<FlinkDeployment> canaryResourceManager;
+    private final FlinkConfigManager flinkConfigManager;
 
     public FlinkDeploymentController(
             Set<FlinkResourceValidator> validators,
@@ -77,14 +79,16 @@ public FlinkDeploymentController(
             FlinkDeploymentObserverFactory observerFactory,
             StatusRecorder<FlinkDeployment, FlinkDeploymentStatus> statusRecorder,
             EventRecorder eventRecorder,
-            CanaryResourceManager<FlinkDeployment> canaryResourceManager) {
+            CanaryResourceManager<FlinkDeployment> canaryResourceManager,
+            FlinkConfigManager flinkConfigManager) {
         this.validators = validators;
         this.ctxFactory = ctxFactory;
         this.reconcilerFactory = reconcilerFactory;
         this.observerFactory = observerFactory;
         this.statusRecorder = statusRecorder;
         this.eventRecorder = eventRecorder;
         this.canaryResourceManager = canaryResourceManager;
+        this.flinkConfigManager = flinkConfigManager;
     }
 
     @Override
@@ -184,7 +188,9 @@ public List<EventSource<?, FlinkDeployment>> prepareEventSources(
         List<EventSource<?, FlinkDeployment>> eventSources = new ArrayList<>();
         eventSources.add(EventSourceUtils.getSessionJobInformerEventSource(context));
         eventSources.add(EventSourceUtils.getDeploymentInformerEventSource(context));
-        eventSources.add(EventSourceUtils.getIngressInformerEventSource(context));
+        if (flinkConfigManager.getOperatorConfiguration().isManageIngress()) {
+            eventSources.add(EventSourceUtils.getIngressInformerEventSource(context));
+        }
         if (KubernetesClientUtils.isCrdInstalled(FlinkStateSnapshot.class)) {
             eventSources.add(
                     EventSourceUtils.getStateSnapshotForFlinkResourceInformerEventSource(context));
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
@@ -74,7 +74,9 @@ public static void reconcileIngress(
             FlinkDeploymentSpec spec,
             Configuration effectiveConfig,
             KubernetesClient client) {
-
+        if (!ctx.getOperatorConfig().isManageIngress()) {
+            return;
+        }
         var objectMeta = ctx.getResource().getMetadata();
         if (spec.getIngress() != null) {
             HasMetadata ingress = getIngress(objectMeta, spec, effectiveConfig, client);
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/TestingFlinkDeploymentController.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/TestingFlinkDeploymentController.java
@@ -110,7 +110,8 @@ public TestingFlinkDeploymentController(
                         new FlinkDeploymentObserverFactory(eventRecorder),
                         statusRecorder,
                         eventRecorder,
-                        canaryResourceManager);
+                        canaryResourceManager,
+                        new FlinkConfigManager(Configuration.fromMap(Map.of())));
     }
 
     @Override
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
@@ -38,9 +38,11 @@
 
 import java.net.URL;
 import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
+import static org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions.OPERATOR_MANAGE_INGRESS;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNull;
@@ -57,7 +59,20 @@ class IngressUtilsTest {
 
     private FlinkResourceContext<?> createResourceContext(FlinkDeployment appCluster) {
         testingJosdkContext = new TestingJosdkContext<>(client);
-        return new FlinkDeploymentContext(appCluster, testingJosdkContext, null, null, null, null);
+        return new FlinkDeploymentContext(
+                appCluster,
+                testingJosdkContext,
+                null,
+                new FlinkConfigManager(Configuration.fromMap(new HashMap<>())),
+                null,
+                null);
+    }
+
+    private FlinkResourceContext<?> createResourceContext(
+            FlinkDeployment appCluster, FlinkConfigManager configManager) {
+        testingJosdkContext = new TestingJosdkContext<>(client);
+        return new FlinkDeploymentContext(
+                appCluster, testingJosdkContext, null, configManager, null, null);
     }
 
     @Test
@@ -560,4 +575,33 @@ void testDeletesIngress() {
                         .get();
         assertThat(ingressV1beta1).isNull();
     }
+
+    @Test
+    void skipIngressReconciliationIfFeatureFlagOff() {
+        FlinkDeployment appCluster = TestUtils.buildApplicationCluster();
+        FlinkConfigManager manager =
+                new FlinkConfigManager(
+                        Configuration.fromMap(Map.of(OPERATOR_MANAGE_INGRESS.key(), "false")));
+        var context = createResourceContext(appCluster, manager);
+        context.getOperatorConfig();
+        Configuration config =
+                new FlinkConfigManager(new Configuration())
+                        .getDeployConfig(appCluster.getMetadata(), appCluster.getSpec());
+
+        IngressSpec.IngressSpecBuilder builder = IngressSpec.builder();
+        builder.template("{{name}}.{{namespace}}.example.com");
+        builder.tls(new ArrayList<>());
+        appCluster.getSpec().setIngress(builder.build());
+
+        IngressUtils.reconcileIngress(context, appCluster.getSpec(), config, client);
+
+        var ingressV1beta1 =
+                client.network()
+                        .v1beta1()
+                        .ingresses()
+                        .inNamespace(appCluster.getMetadata().getNamespace())
+                        .withName(appCluster.getMetadata().getName())
+                        .get();
+        assertThat(ingressV1beta1).isNull();
+    }
 }

Original file line number	Diff line number	Diff line change
`@@ -199,7 +199,8 @@ void registerDeploymentController() {`
`199`	`199`	`observerFactory,`
`200`	`200`	`statusRecorder,`
`201`	`201`	`eventRecorder,`
`202`		`- canaryResourceManager);`
	`202`	`+ canaryResourceManager,`
	`203`	`+ configManager);`
`203`	`204`	`registeredControllers.add(operator.register(controller, this::overrideControllerConfigs));`
`204`	`205`	`}`
`205`	`206`
Original file line number	Diff line number	Diff line change
`@@ -110,7 +110,8 @@ public TestingFlinkDeploymentController(`
`110`	`110`	`new FlinkDeploymentObserverFactory(eventRecorder),`
`111`	`111`	`statusRecorder,`
`112`	`112`	`eventRecorder,`
`113`		`- canaryResourceManager);`
	`113`	`+ canaryResourceManager,`
	`114`	`+ new FlinkConfigManager(Configuration.fromMap(Map.of())));`
`114`	`115`	`}`
`115`	`116`
`116`	`117`	`@Override`