ingress labels and default label selector

Signed-off-by: Attila Mészáros <[email protected]>

Author: csviri

flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/EventSourceUtils.java

@@ -51,6 +51,8 @@
 /** 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();
@@ -107,13 +109,10 @@ public static InformerEventSource<Deployment, FlinkDeployment> getDeploymentInfo
 
     public static InformerEventSource<?, FlinkDeployment> getIngressInformerEventSource(
             EventSourceContext<FlinkDeployment> context) {
-        //        final String labelSelector =
-        //                Map.of(Constants.LABEL_COMPONENT_KEY,
-        // Constants.LABEL_COMPONENT_JOB_MANAGER)
-        //                        .entrySet()
-        //                        .stream()
-        //                        .map(Object::toString)
-        //                        .collect(Collectors.joining(","));
+        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())
@@ -122,7 +121,7 @@ public static InformerEventSource<?, FlinkDeployment> getIngressInformerEventSou
 
         var configuration =
                 InformerEventSourceConfiguration.from(ingressClass, FlinkDeployment.class)
-                        //                        .withLabelSelector(labelSelector)
+                        .withLabelSelector(labelSelector)
                         .withNamespacesInheritedFromController()
                         .withFollowControllerNamespacesChanges(true)
                         .build();

flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/IngressUtils.java

@@ -22,6 +22,7 @@
 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;
@@ -45,11 +46,14 @@
 import java.net.MalformedURLException;
 import java.net.URL;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.List;
 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 {
 
@@ -98,10 +102,12 @@ private static HasMetadata getIngress(
             FlinkDeploymentSpec spec,
             Configuration effectiveConfig,
             KubernetesClient client) {
+        var labels = 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())
@@ -133,7 +139,7 @@ private static HasMetadata getIngress(
             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()