feat: field selectors support for InformerEventSource (#2835)

Signed-off-by: Attila Mészáros <a_meszaros@apple.com>

Author: csviri

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/informer/Field.java

@@ -0,0 +1,10 @@
+package io.javaoperatorsdk.operator.api.config.informer;
+
+public @interface Field {
+
+  String path();
+
+  String value();
+
+  boolean negated() default false;
+}

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/informer/FieldSelector.java

@@ -0,0 +1,26 @@
+package io.javaoperatorsdk.operator.api.config.informer;
+
+import java.util.Arrays;
+import java.util.List;
+
+public class FieldSelector {
+  private final List<Field> fields;
+
+  public FieldSelector(List<Field> fields) {
+    this.fields = fields;
+  }
+
+  public FieldSelector(Field... fields) {
+    this.fields = Arrays.asList(fields);
+  }
+
+  public List<Field> getFields() {
+    return fields;
+  }
+
+  public record Field(String path, String value, boolean negated) {
+    public Field(String path, String value) {
+      this(path, value, false);
+    }
+  }
+}

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/informer/FieldSelectorBuilder.java

@@ -0,0 +1,23 @@
+package io.javaoperatorsdk.operator.api.config.informer;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class FieldSelectorBuilder {
+
+  private final List<FieldSelector.Field> fields = new ArrayList<>();
+
+  public FieldSelectorBuilder withField(String path, String value) {
+    fields.add(new FieldSelector.Field(path, value));
+    return this;
+  }
+
+  public FieldSelectorBuilder withoutField(String path, String value) {
+    fields.add(new FieldSelector.Field(path, value, true));
+    return this;
+  }
+
+  public FieldSelector build() {
+    return new FieldSelector(fields);
+  }
+}

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/informer/Informer.java

@@ -113,4 +113,7 @@
    * the informer cache.
    */
   long informerListLimit() default NO_LONG_VALUE_SET;
+
+  /** Kubernetes field selector for additional resource filtering */
+  Field[] fieldSelector() default {};
 }

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/informer/InformerConfiguration.java

@@ -1,5 +1,6 @@
 package io.javaoperatorsdk.operator.api.config.informer;
 
+import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.Set;
@@ -36,6 +37,7 @@ public class InformerConfiguration<R extends HasMetadata> {
   private GenericFilter<? super R> genericFilter;
   private ItemStore<R> itemStore;
   private Long informerListLimit;
+  private FieldSelector fieldSelector;
 
   protected InformerConfiguration(
       Class<R> resourceClass,
@@ -48,7 +50,8 @@ protected InformerConfiguration(
       OnDeleteFilter<? super R> onDeleteFilter,
       GenericFilter<? super R> genericFilter,
       ItemStore<R> itemStore,
-      Long informerListLimit) {
+      Long informerListLimit,
+      FieldSelector fieldSelector) {
     this(resourceClass);
     this.name = name;
     this.namespaces = namespaces;
@@ -60,6 +63,7 @@ protected InformerConfiguration(
     this.genericFilter = genericFilter;
     this.itemStore = itemStore;
     this.informerListLimit = informerListLimit;
+    this.fieldSelector = fieldSelector;
   }
 
   private InformerConfiguration(Class<R> resourceClass) {
@@ -93,7 +97,8 @@ public static <R extends HasMetadata> InformerConfiguration<R>.Builder builder(
             original.onDeleteFilter,
             original.genericFilter,
             original.itemStore,
-            original.informerListLimit)
+            original.informerListLimit,
+            original.fieldSelector)
         .builder;
   }
 
@@ -264,6 +269,10 @@ public Long getInformerListLimit() {
     return informerListLimit;
   }
 
+  public FieldSelector getFieldSelector() {
+    return fieldSelector;
+  }
+
   @SuppressWarnings("UnusedReturnValue")
   public class Builder {
 
@@ -329,6 +338,12 @@ public InformerConfiguration<R>.Builder initFromAnnotation(
         final var informerListLimit =
             informerListLimitValue == Constants.NO_LONG_VALUE_SET ? null : informerListLimitValue;
         withInformerListLimit(informerListLimit);
+
+        withFieldSelector(
+            new FieldSelector(
+                Arrays.stream(informerConfig.fieldSelector())
+                    .map(f -> new FieldSelector.Field(f.path(), f.value(), f.negated()))
+                    .toList()));
       }
       return this;
     }
@@ -424,5 +439,10 @@ public Builder withInformerListLimit(Long informerListLimit) {
       InformerConfiguration.this.informerListLimit = informerListLimit;
       return this;
     }
+
+    public Builder withFieldSelector(FieldSelector fieldSelector) {
+      InformerConfiguration.this.fieldSelector = fieldSelector;
+      return this;
+    }
   }
 }

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/informer/InformerEventSourceConfiguration.java

@@ -265,6 +265,11 @@ public Builder<R> withInformerListLimit(Long informerListLimit) {
       return this;
     }
 
+    public Builder<R> withFieldSelector(FieldSelector fieldSelector) {
+      config.withFieldSelector(fieldSelector);
+      return this;
+    }
+
     public void updateFrom(InformerConfiguration<R> informerConfig) {
       if (informerConfig != null) {
         final var informerConfigName = informerConfig.getName();
@@ -281,7 +286,8 @@ public void updateFrom(InformerConfiguration<R> informerConfig) {
             .withOnUpdateFilter(informerConfig.getOnUpdateFilter())
             .withOnDeleteFilter(informerConfig.getOnDeleteFilter())
             .withGenericFilter(informerConfig.getGenericFilter())
-            .withInformerListLimit(informerConfig.getInformerListLimit());
+            .withInformerListLimit(informerConfig.getInformerListLimit())
+            .withFieldSelector(informerConfig.getFieldSelector());
       }
     }
 

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerManager.java

@@ -134,6 +134,18 @@ private InformerWrapper<R> createEventSource(
       ResourceEventHandler<R> eventHandler,
       String namespaceIdentifier) {
     final var informerConfig = configuration.getInformerConfig();
+
+    if (informerConfig.getFieldSelector() != null
+        && !informerConfig.getFieldSelector().getFields().isEmpty()) {
+      for (var f : informerConfig.getFieldSelector().getFields()) {
+        if (f.negated()) {
+          filteredBySelectorClient = filteredBySelectorClient.withoutField(f.path(), f.value());
+        } else {
+          filteredBySelectorClient = filteredBySelectorClient.withField(f.path(), f.value());
+        }
+      }
+    }
+
     var informer =
         Optional.ofNullable(informerConfig.getInformerListLimit())
             .map(filteredBySelectorClient::withLimit)

operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/fieldselector/FieldSelectorIT.java

@@ -0,0 +1,73 @@
+package io.javaoperatorsdk.operator.baseapi.fieldselector;
+
+import java.time.Duration;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import io.fabric8.kubernetes.api.model.ObjectMetaBuilder;
+import io.fabric8.kubernetes.api.model.SecretBuilder;
+import io.javaoperatorsdk.operator.junit.LocallyRunOperatorExtension;
+import io.javaoperatorsdk.operator.processing.event.ResourceID;
+
+import static io.javaoperatorsdk.operator.baseapi.fieldselector.FieldSelectorTestReconciler.MY_SECRET_TYPE;
+import static io.javaoperatorsdk.operator.baseapi.fieldselector.FieldSelectorTestReconciler.OTHER_SECRET_TYPE;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
+
+class FieldSelectorIT {
+
+  public static final String TEST_1 = "test1";
+  public static final String TEST_2 = "test2";
+  public static final String TEST_3 = "test3";
+
+  @RegisterExtension
+  LocallyRunOperatorExtension extension =
+      LocallyRunOperatorExtension.builder()
+          .withReconciler(new FieldSelectorTestReconciler())
+          .build();
+
+  @Test
+  void filtersCustomResourceByLabel() {
+
+    var customPrimarySecret =
+        extension.create(
+            new SecretBuilder()
+                .withMetadata(new ObjectMetaBuilder().withName(TEST_1).build())
+                .withType(MY_SECRET_TYPE)
+                .build());
+
+    var otherSecret =
+        extension.create(
+            new SecretBuilder()
+                .withMetadata(new ObjectMetaBuilder().withName(TEST_2).build())
+                .build());
+
+    var dependentSecret =
+        extension.create(
+            new SecretBuilder()
+                .withMetadata(new ObjectMetaBuilder().withName(TEST_3).build())
+                .withType(OTHER_SECRET_TYPE)
+                .build());
+
+    await()
+        .pollDelay(Duration.ofMillis(150))
+        .untilAsserted(
+            () -> {
+              var r = extension.getReconcilerOfType(FieldSelectorTestReconciler.class);
+              assertThat(r.getReconciledSecrets()).containsExactly(TEST_1);
+
+              assertThat(
+                      r.getDependentSecretEventSource()
+                          .get(ResourceID.fromResource(dependentSecret)))
+                  .isPresent();
+              assertThat(
+                      r.getDependentSecretEventSource()
+                          .get(ResourceID.fromResource(customPrimarySecret)))
+                  .isNotPresent();
+              assertThat(
+                      r.getDependentSecretEventSource().get(ResourceID.fromResource(otherSecret)))
+                  .isNotPresent();
+            });
+  }
+}

operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/fieldselector/FieldSelectorTestReconciler.java

@@ -0,0 +1,69 @@
+package io.javaoperatorsdk.operator.baseapi.fieldselector;
+
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import io.fabric8.kubernetes.api.model.Secret;
+import io.javaoperatorsdk.operator.api.config.informer.Field;
+import io.javaoperatorsdk.operator.api.config.informer.FieldSelectorBuilder;
+import io.javaoperatorsdk.operator.api.config.informer.Informer;
+import io.javaoperatorsdk.operator.api.config.informer.InformerEventSourceConfiguration;
+import io.javaoperatorsdk.operator.api.reconciler.Context;
+import io.javaoperatorsdk.operator.api.reconciler.ControllerConfiguration;
+import io.javaoperatorsdk.operator.api.reconciler.EventSourceContext;
+import io.javaoperatorsdk.operator.api.reconciler.Reconciler;
+import io.javaoperatorsdk.operator.api.reconciler.UpdateControl;
+import io.javaoperatorsdk.operator.processing.event.source.EventSource;
+import io.javaoperatorsdk.operator.processing.event.source.informer.InformerEventSource;
+import io.javaoperatorsdk.operator.support.TestExecutionInfoProvider;
+
+@ControllerConfiguration(
+    informer =
+        @Informer(
+            fieldSelector =
+                @Field(path = "type", value = FieldSelectorTestReconciler.MY_SECRET_TYPE)))
+public class FieldSelectorTestReconciler implements Reconciler<Secret>, TestExecutionInfoProvider {
+
+  public static final String MY_SECRET_TYPE = "my-secret-type";
+  public static final String OTHER_SECRET_TYPE = "my-dependent-secret-type";
+  private final AtomicInteger numberOfExecutions = new AtomicInteger(0);
+
+  private Set<String> reconciledSecrets = Collections.synchronizedSet(new HashSet<>());
+  private InformerEventSource<Secret, Secret> dependentSecretEventSource;
+
+  @Override
+  public UpdateControl<Secret> reconcile(Secret resource, Context<Secret> context) {
+    reconciledSecrets.add(resource.getMetadata().getName());
+    numberOfExecutions.addAndGet(1);
+    return UpdateControl.noUpdate();
+  }
+
+  public int getNumberOfExecutions() {
+    return numberOfExecutions.get();
+  }
+
+  public Set<String> getReconciledSecrets() {
+    return reconciledSecrets;
+  }
+
+  @Override
+  public List<EventSource<?, Secret>> prepareEventSources(EventSourceContext<Secret> context) {
+    dependentSecretEventSource =
+        new InformerEventSource<>(
+            InformerEventSourceConfiguration.from(Secret.class, Secret.class)
+                .withNamespacesInheritedFromController()
+                .withFieldSelector(
+                    new FieldSelectorBuilder().withField("type", OTHER_SECRET_TYPE).build())
+                .build(),
+            context);
+
+    return List.of(dependentSecretEventSource);
+  }
+
+  public InformerEventSource<Secret, Secret> getDependentSecretEventSource() {
+    return dependentSecretEventSource;
+  }
+}