feat: explicit event filters for all event sources

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

@@ -8,7 +8,9 @@
 import java.util.List;
 import java.util.Optional;
 import java.util.Set;
+import java.util.function.BiPredicate;
 import java.util.function.Function;
+import java.util.function.Predicate;
 import java.util.stream.Collectors;
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
@@ -27,19 +29,23 @@
 import io.javaoperatorsdk.operator.processing.dependent.workflow.Condition;
 import io.javaoperatorsdk.operator.processing.event.source.controller.ResourceEventFilter;
 import io.javaoperatorsdk.operator.processing.event.source.controller.ResourceEventFilters;
+import io.javaoperatorsdk.operator.processing.event.source.filter.VoidGenericFilter;
+import io.javaoperatorsdk.operator.processing.event.source.filter.VoidOnAddFilter;
+import io.javaoperatorsdk.operator.processing.event.source.filter.VoidOnDeleteFilter;
+import io.javaoperatorsdk.operator.processing.event.source.filter.VoidOnUpdateFilter;
 
 import static io.javaoperatorsdk.operator.api.reconciler.Constants.DEFAULT_NAMESPACES_SET;
 
 @SuppressWarnings("rawtypes")
-public class AnnotationControllerConfiguration<R extends HasMetadata>
-    implements io.javaoperatorsdk.operator.api.config.ControllerConfiguration<R> {
+public class AnnotationControllerConfiguration<P extends HasMetadata>
+    implements io.javaoperatorsdk.operator.api.config.ControllerConfiguration<P> {
 
-  protected final Reconciler<R> reconciler;
+  protected final Reconciler<P> reconciler;
   private final ControllerConfiguration annotation;
   private List<DependentResourceSpec> specs;
-  private Class<R> resourceClass;
+  private Class<P> resourceClass;
 
-  public AnnotationControllerConfiguration(Reconciler<R> reconciler) {
+  public AnnotationControllerConfiguration(Reconciler<P> reconciler) {
     this.reconciler = reconciler;
     this.annotation = reconciler.getClass().getAnnotation(ControllerConfiguration.class);
     if (annotation == null) {
@@ -84,10 +90,10 @@ public Set<String> getNamespaces() {
 
   @Override
   @SuppressWarnings("unchecked")
-  public Class<R> getResourceClass() {
+  public Class<P> getResourceClass() {
     if (resourceClass == null) {
       resourceClass =
-          (Class<R>) Utils.getFirstTypeArgumentFromSuperClassOrInterface(reconciler.getClass(),
+          (Class<P>) Utils.getFirstTypeArgumentFromSuperClassOrInterface(reconciler.getClass(),
               Reconciler.class);
     }
     return resourceClass;
@@ -105,16 +111,16 @@ public String getAssociatedReconcilerClassName() {
 
   @SuppressWarnings("unchecked")
   @Override
-  public ResourceEventFilter<R> getEventFilter() {
-    ResourceEventFilter<R> answer = null;
+  public ResourceEventFilter<P> getEventFilter() {
+    ResourceEventFilter<P> answer = null;
 
-    Class<ResourceEventFilter<R>>[] filterTypes =
-        (Class<ResourceEventFilter<R>>[]) valueOrDefault(annotation,
+    Class<ResourceEventFilter<P>>[] filterTypes =
+        (Class<ResourceEventFilter<P>>[]) valueOrDefault(annotation,
             ControllerConfiguration::eventFilters, new Object[] {});
     if (filterTypes.length > 0) {
       for (var filterType : filterTypes) {
         try {
-          ResourceEventFilter<R> filter = filterType.getConstructor().newInstance();
+          ResourceEventFilter<P> filter = filterType.getConstructor().newInstance();
 
           if (answer == null) {
             answer = filter;
@@ -144,17 +150,55 @@ public Optional<Duration> reconciliationMaxInterval() {
     }
   }
 
-  public static <T> T valueOrDefault(
-      ControllerConfiguration controllerConfiguration,
-      Function<ControllerConfiguration, T> mapper,
-      T defaultValue) {
-    if (controllerConfiguration == null) {
-      return defaultValue;
+  @Override
+  @SuppressWarnings("unchecked")
+  public Optional<Predicate<P>> onAddFilter() {
+    return (Optional<Predicate<P>>) createFilter(annotation.onAddFilter(), FilterType.onAdd,
+        annotation.getClass().getSimpleName());
+  }
+
+  private enum FilterType {
+    onAdd(VoidOnAddFilter.class), onUpdate(VoidOnUpdateFilter.class), onDelete(
+        VoidOnDeleteFilter.class), generic(VoidGenericFilter.class);
+
+    final Class<?> defaultValue;
+
+    FilterType(Class<?> defaultValue) {
+      this.defaultValue = defaultValue;
+    }
+  }
+
+  private <T> Optional<T> createFilter(Class<T> filter, FilterType filterType, String origin) {
+    if (filterType.defaultValue.equals(filter)) {
+      return Optional.empty();
     } else {
-      return mapper.apply(controllerConfiguration);
+      try {
+        var instance = (T) filter.getDeclaredConstructor().newInstance();
+        return Optional.of(instance);
+      } catch (InstantiationException | IllegalAccessException | InvocationTargetException
+          | NoSuchMethodException e) {
+        throw new OperatorException(
+            "Couldn't create " + filterType + " filter from " + filter.getName() + " class in "
+                + origin + " for reconciler " + getName(),
+            e);
+      }
     }
   }
 
+  @SuppressWarnings("unchecked")
+  @Override
+  public Optional<BiPredicate<P, P>> onUpdateFilter() {
+    return (Optional<BiPredicate<P, P>>) createFilter(annotation.onUpdateFilter(),
+        FilterType.onUpdate, annotation.getClass().getSimpleName());
+  }
+
+  @SuppressWarnings("unchecked")
+  @Override
+  public Optional<Predicate<P>> genericFilter() {
+    return (Optional<Predicate<P>>) createFilter(annotation.genericFilter(),
+        FilterType.generic, annotation.getClass().getSimpleName());
+  }
+
   @SuppressWarnings({"rawtypes", "unchecked"})
   @Override
   public List<DependentResourceSpec> getDependentResources() {
@@ -223,26 +267,59 @@ private String getName(Dependent dependent, Class<? extends DependentResource> d
     return name;
   }
 
+  @SuppressWarnings("rawtypes")
   private Object createKubernetesResourceConfig(Class<? extends DependentResource> dependentType) {
+
     Object config;
     final var kubeDependent = dependentType.getAnnotation(KubernetesDependent.class);
 
     var namespaces = getNamespaces();
     var configuredNS = false;
-    if (kubeDependent != null && !Arrays.equals(KubernetesDependent.DEFAULT_NAMESPACES,
-        kubeDependent.namespaces())) {
-      namespaces = Set.of(kubeDependent.namespaces());
-      configuredNS = true;
-    }
-
     String labelSelector = null;
+    Predicate<? extends HasMetadata> onAddFilter = null;
+    BiPredicate<? extends HasMetadata, ? extends HasMetadata> onUpdateFilter = null;
+    BiPredicate<? extends HasMetadata, Boolean> onDeleteFilter = null;
+    Predicate<? extends HasMetadata> genericFilter = null;
     if (kubeDependent != null) {
+      if (!Arrays.equals(KubernetesDependent.DEFAULT_NAMESPACES,
+          kubeDependent.namespaces())) {
+        namespaces = Set.of(kubeDependent.namespaces());
+        configuredNS = true;
+      }
+
       final var fromAnnotation = kubeDependent.labelSelector();
       labelSelector = Constants.NO_VALUE_SET.equals(fromAnnotation) ? null : fromAnnotation;
+
+      final var kubeDependentName = KubernetesDependent.class.getSimpleName();
+      onAddFilter = createFilter(kubeDependent.onAddFilter(), FilterType.onAdd, kubeDependentName)
+          .orElse(null);
+      onUpdateFilter =
+          createFilter(kubeDependent.onUpdateFilter(), FilterType.onUpdate, kubeDependentName)
+              .orElse(null);
+      onDeleteFilter =
+          createFilter(kubeDependent.onDeleteFilter(), FilterType.onDelete, kubeDependentName)
+              .orElse(null);
+      genericFilter =
+          createFilter(kubeDependent.genericFilter(), FilterType.generic, kubeDependentName)
+              .orElse(null);
     }
 
     config =
-        new KubernetesDependentResourceConfig(namespaces, labelSelector, configuredNS);
+        new KubernetesDependentResourceConfig(namespaces, labelSelector, configuredNS, onAddFilter,
+            onUpdateFilter, onDeleteFilter, genericFilter);
+
     return config;
   }
+
+  public static <T> T valueOrDefault(
+      ControllerConfiguration controllerConfiguration,
+      Function<ControllerConfiguration, T> mapper,
+      T defaultValue) {
+    if (controllerConfiguration == null) {
+      return defaultValue;
+    } else {
+      return mapper.apply(controllerConfiguration);
+    }
+  }
+
 }

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

@@ -1,7 +1,12 @@
 package io.javaoperatorsdk.operator.api.config;
 
 import java.time.Duration;
-import java.util.*;
+import java.util.HashSet;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Optional;
+import java.util.Set;
+import java.util.function.BiPredicate;
 import java.util.function.Predicate;
 import java.util.stream.Collectors;
 
@@ -27,6 +32,9 @@ public class ControllerConfigurationOverrider<R extends HasMetadata> {
   private final ControllerConfiguration<R> original;
   private Duration reconciliationMaxInterval;
   private final LinkedHashMap<String, DependentResourceSpec> namedDependentResourceSpecs;
+  private Predicate<R> onAddFilter;
+  private BiPredicate<R, R> onUpdateFilter;
+  private Predicate<R> genericFilter;
 
   private ControllerConfigurationOverrider(ControllerConfiguration<R> original) {
     finalizer = original.getFinalizerName();
@@ -39,6 +47,9 @@ private ControllerConfigurationOverrider(ControllerConfiguration<R> original) {
     // make the original specs modifiable
     final var dependentResources = original.getDependentResources();
     namedDependentResourceSpecs = new LinkedHashMap<>(dependentResources.size());
+    this.onAddFilter = original.onAddFilter().orElse(null);
+    this.onUpdateFilter = original.onUpdateFilter().orElse(null);
+    this.genericFilter = original.genericFilter().orElse(null);
     dependentResources.forEach(drs -> namedDependentResourceSpecs.put(drs.getName(), drs));
     this.original = original;
   }
@@ -120,6 +131,21 @@ public ControllerConfigurationOverrider<R> withReconciliationMaxInterval(
     return this;
   }
 
+  public ControllerConfigurationOverrider<R> withOnAddFilter(Predicate<R> onAddFilter) {
+    this.onAddFilter = onAddFilter;
+    return this;
+  }
+
+  public ControllerConfigurationOverrider<R> withOnUpdateFilter(BiPredicate<R, R> onUpdateFilter) {
+    this.onUpdateFilter = onUpdateFilter;
+    return this;
+  }
+
+  public ControllerConfigurationOverrider<R> withGenericFilter(Predicate<R> genericFilter) {
+    this.genericFilter = genericFilter;
+    return this;
+  }
+
   public ControllerConfigurationOverrider<R> replacingNamedDependentResourceConfig(String name,
       Object dependentResourceConfig) {
 
@@ -167,6 +193,9 @@ public ControllerConfiguration<R> build() {
         customResourcePredicate,
         original.getResourceClass(),
         reconciliationMaxInterval,
+        onAddFilter,
+        onUpdateFilter,
+        genericFilter,
         newDependentSpecs);
   }
 

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

@@ -5,6 +5,8 @@
 import java.util.List;
 import java.util.Optional;
 import java.util.Set;
+import java.util.function.BiPredicate;
+import java.util.function.Predicate;
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.javaoperatorsdk.operator.api.config.dependent.DependentResourceSpec;
@@ -39,8 +41,11 @@ public DefaultControllerConfiguration(
       ResourceEventFilter<R> resourceEventFilter,
       Class<R> resourceClass,
       Duration reconciliationMaxInterval,
+      Predicate<R> onAddFilter,
+      BiPredicate<R, R> onUpdateFilter,
+      Predicate<R> genericFilter,
       List<DependentResourceSpec> dependents) {
-    super(labelSelector, resourceClass, namespaces);
+    super(labelSelector, resourceClass, onAddFilter, onUpdateFilter, genericFilter, namespaces);
     this.associatedControllerClassName = associatedControllerClassName;
     this.name = name;
     this.crdName = crdName;

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

@@ -1,6 +1,9 @@
 package io.javaoperatorsdk.operator.api.config;
 
+import java.util.Optional;
 import java.util.Set;
+import java.util.function.BiPredicate;
+import java.util.function.Predicate;
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
 
@@ -12,18 +15,26 @@ public class DefaultResourceConfiguration<R extends HasMetadata>
   private final String labelSelector;
   private final Set<String> namespaces;
   private final Class<R> resourceClass;
+  private final Predicate<R> onAddFilter;
+  private final BiPredicate<R, R> onUpdateFilter;
+  private final Predicate<R> genericFilter;
 
   public DefaultResourceConfiguration(String labelSelector, Class<R> resourceClass,
-      String... namespaces) {
-    this(labelSelector, resourceClass,
+      Predicate<R> onAddFilter,
+      BiPredicate<R, R> onUpdateFilter, Predicate<R> genericFilter, String... namespaces) {
+    this(labelSelector, resourceClass, onAddFilter, onUpdateFilter, genericFilter,
         namespaces == null || namespaces.length == 0 ? DEFAULT_NAMESPACES_SET
             : Set.of(namespaces));
   }
 
   public DefaultResourceConfiguration(String labelSelector, Class<R> resourceClass,
-      Set<String> namespaces) {
+      Predicate<R> onAddFilter,
+      BiPredicate<R, R> onUpdateFilter, Predicate<R> genericFilter, Set<String> namespaces) {
     this.labelSelector = labelSelector;
     this.resourceClass = resourceClass;
+    this.onAddFilter = onAddFilter;
+    this.onUpdateFilter = onUpdateFilter;
+    this.genericFilter = genericFilter;
     this.namespaces =
         namespaces == null || namespaces.isEmpty() ? DEFAULT_NAMESPACES_SET
             : namespaces;
@@ -48,4 +59,18 @@ public Set<String> getNamespaces() {
   public Class<R> getResourceClass() {
     return resourceClass;
   }
+
+  @Override
+  public Optional<Predicate<R>> onAddFilter() {
+    return Optional.ofNullable(onAddFilter);
+  }
+
+  @Override
+  public Optional<BiPredicate<R, R>> onUpdateFilter() {
+    return Optional.ofNullable(onUpdateFilter);
+  }
+
+  public Optional<Predicate<R>> genericFilter() {
+    return Optional.ofNullable(genericFilter);
+  }
 }

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

@@ -1,7 +1,10 @@
 package io.javaoperatorsdk.operator.api.config;
 
 import java.util.Collections;
+import java.util.Optional;
 import java.util.Set;
+import java.util.function.BiPredicate;
+import java.util.function.Predicate;
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.javaoperatorsdk.operator.OperatorException;
@@ -17,6 +20,18 @@ default String getResourceTypeName() {
     return ReconcilerUtils.getResourceTypeName(getResourceClass());
   }
 
+  default Optional<Predicate<R>> onAddFilter() {
+    return Optional.empty();
+  }
+
+  default Optional<BiPredicate<R, R>> onUpdateFilter() {
+    return Optional.empty();
+  }
+
+  default Optional<Predicate<R>> genericFilter() {
+    return Optional.empty();
+  }
+
   /**
    * Retrieves the label selector that is used to filter which resources are actually watched by the
    * associated event source. See the official documentation on the