Custom event filter for controllers (#457)

Custom event filter for controllers

Author: lburgazzoli

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

@@ -5,6 +5,8 @@
 import java.lang.annotation.RetentionPolicy;
 import java.lang.annotation.Target;
 
+import io.javaoperatorsdk.operator.processing.event.internal.CustomResourceEventFilter;
+
 @Retention(RetentionPolicy.RUNTIME)
 @Target({ElementType.TYPE})
 public @interface Controller {
@@ -46,7 +48,16 @@
    * upon. The label selector can be made of multiple comma separated requirements that acts as a
    * logical AND operator.
    *
-   * @return the finalizer name
+   * @return the label selector
    */
   String labelSelector() default NULL;
+
+
+  /**
+   * Optional list of classes providing custom {@link CustomResourceEventFilter}.
+   *
+   * @return the list of event filters.
+   */
+  @SuppressWarnings("rawtypes")
+  Class<CustomResourceEventFilter>[] eventFilters() default {};
 }

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

@@ -3,6 +3,7 @@
 import java.util.Set;
 
 import io.fabric8.kubernetes.client.CustomResource;
+import io.javaoperatorsdk.operator.processing.event.internal.CustomResourceEventFilter;
 
 /**
  * @deprecated use {@link DefaultControllerConfiguration} instead
@@ -25,9 +26,10 @@ public AbstractControllerConfiguration(String associatedControllerClassName, Str
       String crdName, String finalizer, boolean generationAware,
       Set<String> namespaces,
       RetryConfiguration retryConfiguration, String labelSelector,
+      CustomResourceEventFilter<R> customResourcePredicate,
       Class<R> customResourceClass,
       ConfigurationService service) {
     super(associatedControllerClassName, name, crdName, finalizer, generationAware, namespaces,
-        retryConfiguration, labelSelector, customResourceClass, service);
+        retryConfiguration, labelSelector, customResourcePredicate, customResourceClass, service);
   }
 }

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

@@ -7,6 +7,8 @@
 import io.fabric8.kubernetes.client.CustomResource;
 import io.javaoperatorsdk.operator.ControllerUtils;
 import io.javaoperatorsdk.operator.api.Controller;
+import io.javaoperatorsdk.operator.processing.event.internal.CustomResourceEventFilter;
+import io.javaoperatorsdk.operator.processing.event.internal.CustomResourceEventFilters;
 
 public interface ControllerConfiguration<R extends CustomResource> {
 
@@ -98,4 +100,17 @@ default void setConfigurationService(ConfigurationService service) {}
   default boolean useFinalizer() {
     return !Controller.NO_FINALIZER.equals(getFinalizer());
   }
+
+  /**
+   * Allow controllers to filter events before they are provided to the
+   * {@link io.javaoperatorsdk.operator.processing.event.EventHandler}.
+   * </p>
+   * Note that the provided filter is combined with {@link #isGenerationAware()} to compute the
+   * final set of fiolters that should be applied;
+   *
+   * @return
+   */
+  default CustomResourceEventFilter<R> getEventFilter() {
+    return CustomResourceEventFilters.passthrough();
+  }
 }

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

@@ -5,6 +5,7 @@
 import java.util.Set;
 
 import io.fabric8.kubernetes.client.CustomResource;
+import io.javaoperatorsdk.operator.processing.event.internal.CustomResourceEventFilter;
 
 public class ControllerConfigurationOverrider<R extends CustomResource<?, ?>> {
 
@@ -13,6 +14,7 @@ public class ControllerConfigurationOverrider<R extends CustomResource<?, ?>> {
   private final Set<String> namespaces;
   private RetryConfiguration retry;
   private String labelSelector;
+  private CustomResourceEventFilter<R> customResourcePredicate;
   private final ControllerConfiguration<R> original;
 
   private ControllerConfigurationOverrider(ControllerConfiguration<R> original) {
@@ -21,6 +23,7 @@ private ControllerConfigurationOverrider(ControllerConfiguration<R> original) {
     namespaces = new HashSet<>(original.getNamespaces());
     retry = original.getRetryConfiguration();
     labelSelector = original.getLabelSelector();
+    customResourcePredicate = original.getEventFilter();
     this.original = original;
   }
 
@@ -65,6 +68,12 @@ public ControllerConfigurationOverrider<R> withLabelSelector(String labelSelecto
     return this;
   }
 
+  public ControllerConfigurationOverrider<R> withCustomResourcePredicate(
+      CustomResourceEventFilter<R> customResourcePredicate) {
+    this.customResourcePredicate = customResourcePredicate;
+    return this;
+  }
+
   public ControllerConfiguration<R> build() {
     return new DefaultControllerConfiguration<>(
         original.getAssociatedControllerClassName(),
@@ -75,6 +84,7 @@ public ControllerConfiguration<R> build() {
         namespaces,
         retry,
         labelSelector,
+        customResourcePredicate,
         original.getCustomResourceClass(),
         original.getConfigurationService());
   }

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

@@ -4,6 +4,7 @@
 import java.util.Set;
 
 import io.fabric8.kubernetes.client.CustomResource;
+import io.javaoperatorsdk.operator.processing.event.internal.CustomResourceEventFilter;
 
 public class DefaultControllerConfiguration<R extends CustomResource<?, ?>>
     implements ControllerConfiguration<R> {
@@ -17,6 +18,7 @@ public class DefaultControllerConfiguration<R extends CustomResource<?, ?>>
   private final boolean watchAllNamespaces;
   private final RetryConfiguration retryConfiguration;
   private final String labelSelector;
+  private final CustomResourceEventFilter<R> customResourceEventFilter;
   private Class<R> customResourceClass;
   private ConfigurationService service;
 
@@ -29,6 +31,7 @@ public DefaultControllerConfiguration(
       Set<String> namespaces,
       RetryConfiguration retryConfiguration,
       String labelSelector,
+      CustomResourceEventFilter<R> customResourceEventFilter,
       Class<R> customResourceClass,
       ConfigurationService service) {
     this.associatedControllerClassName = associatedControllerClassName;
@@ -44,6 +47,7 @@ public DefaultControllerConfiguration(
             ? ControllerConfiguration.super.getRetryConfiguration()
             : retryConfiguration;
     this.labelSelector = labelSelector;
+    this.customResourceEventFilter = customResourceEventFilter;
     this.customResourceClass =
         customResourceClass == null ? ControllerConfiguration.super.getCustomResourceClass()
             : customResourceClass;
@@ -52,7 +56,7 @@ public DefaultControllerConfiguration(
 
   /**
    * @deprecated use
-   *             {@link #DefaultControllerConfiguration(String, String, String, String, boolean, Set, RetryConfiguration, String, Class, ConfigurationService)}
+   *             {@link #DefaultControllerConfiguration(String, String, String, String, boolean, Set, RetryConfiguration)}
    *             instead
    */
   @Deprecated
@@ -64,8 +68,18 @@ public DefaultControllerConfiguration(
       boolean generationAware,
       Set<String> namespaces,
       RetryConfiguration retryConfiguration) {
-    this(associatedControllerClassName, name, crdName, finalizer, generationAware, namespaces,
-        retryConfiguration, null, null, null);
+    this(
+        associatedControllerClassName,
+        name,
+        crdName,
+        finalizer,
+        generationAware,
+        namespaces,
+        retryConfiguration,
+        null,
+        null,
+        null,
+        null);
   }
 
   @Override
@@ -131,4 +145,9 @@ public String getLabelSelector() {
   public Class<R> getCustomResourceClass() {
     return customResourceClass;
   }
+
+  @Override
+  public CustomResourceEventFilter<R> getEventFilter() {
+    return customResourceEventFilter;
+  }
 }

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/CustomResourceCache.java

@@ -23,13 +23,13 @@
 import static io.javaoperatorsdk.operator.processing.KubernetesResourceUtils.getUID;
 
 @SuppressWarnings("rawtypes")
-public class CustomResourceCache {
+public class CustomResourceCache<T extends CustomResource<?, ?>> {
 
   private static final Logger log = LoggerFactory.getLogger(CustomResourceCache.class);
   private static final Predicate passthrough = o -> true;
 
   private final ObjectMapper objectMapper;
-  private final ConcurrentMap<String, CustomResource> resources;
+  private final ConcurrentMap<String, T> resources;
   private final Lock lock = new ReentrantLock();
 
   public CustomResourceCache() {
@@ -44,19 +44,21 @@ public CustomResourceCache(ObjectMapper objectMapper, Metrics metrics) {
     resources = metrics.monitorSizeOf(new ConcurrentHashMap<>(), "cache");
   }
 
-  public void cacheResource(CustomResource resource) {
+  @SuppressWarnings("unchecked")
+  public void cacheResource(T resource) {
     cacheResource(resource, passthrough);
   }
 
-  public void cacheResource(CustomResource resource, Predicate<CustomResource> predicate) {
+  public void cacheResource(T resource, Predicate<T> predicate) {
     try {
       lock.lock();
       final var uid = getUID(resource);
       if (predicate.test(resources.get(uid))) {
         if (passthrough != predicate) {
           log.trace("Update cache after condition is true: {}", getName(resource));
         }
-        resources.put(uid, resource);
+        // defensive copy
+        resources.put(getUID(resource), clone(resource));
       }
     } finally {
       lock.unlock();
@@ -70,11 +72,11 @@ public void cacheResource(CustomResource resource, Predicate<CustomResource> pre
    * @param uuid
    * @return
    */
-  public Optional<CustomResource> getLatestResource(String uuid) {
+  public Optional<T> getLatestResource(String uuid) {
     return Optional.ofNullable(resources.get(uuid)).map(this::clone);
   }
 
-  public List<CustomResource> getLatestResources(Predicate<CustomResource> selector) {
+  public List<T> getLatestResources(Predicate<CustomResource> selector) {
     try {
       lock.lock();
       return resources.values().stream()
@@ -98,16 +100,17 @@ public Set<String> getLatestResourcesUids(Predicate<CustomResource> selector) {
     }
   }
 
-  private CustomResource clone(CustomResource customResource) {
+  @SuppressWarnings("unchecked")
+  private T clone(CustomResource customResource) {
     try {
-      return objectMapper.readValue(
+      return (T) objectMapper.readValue(
           objectMapper.writeValueAsString(customResource), customResource.getClass());
     } catch (JsonProcessingException e) {
       throw new IllegalStateException(e);
     }
   }
 
-  public CustomResource cleanup(String customResourceUid) {
+  public T cleanup(String customResourceUid) {
     return resources.remove(customResourceUid);
   }
 }

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/internal/CustomResourceEventFilter.java

@@ -0,0 +1,58 @@
+package io.javaoperatorsdk.operator.processing.event.internal;
+
+import io.fabric8.kubernetes.client.CustomResource;
+import io.javaoperatorsdk.operator.api.config.ControllerConfiguration;
+
+/**
+ * A functional interface to determine whether resource events should be processed by the SDK. This
+ * allows users to more finely tuned which events trigger a reconciliation than was previously
+ * possible (where the logic was limited to generation-based checking).
+ *
+ * @param <T> the type of custom resources handled by this filter
+ */
+@FunctionalInterface
+public interface CustomResourceEventFilter<T extends CustomResource> {
+
+  /**
+   * Determines whether the change between the old version of the resource and the new one needs to
+   * be propagated to the controller or not.
+   *
+   * @param configuration the target controller's configuration
+   * @param oldResource the old version of the resource, null if no old resource available
+   * @param newResource the new version of the resource
+   * @return {@code true} if the change needs to be propagated to the controller, {@code false}
+   *         otherwise
+   */
+  boolean acceptChange(ControllerConfiguration<T> configuration, T oldResource, T newResource);
+
+  /**
+   * Combines this filter with the provided one with an AND logic, i.e. the resulting filter will
+   * only accept the change if both this and the other filter accept it, reject it otherwise.
+   *
+   * @param other the possibly {@code null} other filter to combine this one with
+   * @return a composite filter implementing the AND logic between this and the provided filter
+   */
+  default CustomResourceEventFilter<T> and(CustomResourceEventFilter<T> other) {
+    return other == null ? this
+        : (ControllerConfiguration<T> configuration, T oldResource, T newResource) -> {
+          boolean result = acceptChange(configuration, oldResource, newResource);
+          return result && other.acceptChange(configuration, oldResource, newResource);
+        };
+  }
+
+  /**
+   * Combines this filter with the provided one with an OR logic, i.e. the resulting filter will
+   * accept the change if any of this or the other filter accept it, rejecting it only if both
+   * reject it.
+   *
+   * @param other the possibly {@code null} other filter to combine this one with
+   * @return a composite filter implementing the OR logic between this and the provided filter
+   */
+  default CustomResourceEventFilter<T> or(CustomResourceEventFilter<T> other) {
+    return other == null ? this
+        : (ControllerConfiguration<T> configuration, T oldResource, T newResource) -> {
+          boolean result = acceptChange(configuration, oldResource, newResource);
+          return result || other.acceptChange(configuration, oldResource, newResource);
+        };
+  }
+}