event source: use a predicate to select the custom resource for which a reconcile should be triggered operator-framework#430

Author: lburgazzoli

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

@@ -6,12 +6,15 @@
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import io.fabric8.kubernetes.client.CustomResource;
+import java.util.List;
 import java.util.Optional;
+import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
 import java.util.function.Predicate;
+import java.util.stream.Collectors;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -64,6 +67,30 @@ public Optional<CustomResource> getLatestResource(String uuid) {
     return Optional.ofNullable(resources.get(uuid)).map(this::clone);
   }
 
+  public List<CustomResource> getLatestResources(Predicate<CustomResource> selector) {
+    try {
+      lock.lock();
+      return resources.values().stream()
+          .filter(selector)
+          .map(this::clone)
+          .collect(Collectors.toList());
+    } finally {
+      lock.unlock();
+    }
+  }
+
+  public Set<String> getLatestResourcesUids(Predicate<CustomResource> selector) {
+    try {
+      lock.lock();
+      return resources.values().stream()
+          .filter(selector)
+          .map(r -> r.getMetadata().getUid())
+          .collect(Collectors.toSet());
+    } finally {
+      lock.unlock();
+    }
+  }
+
   private CustomResource clone(CustomResource customResource) {
     try {
       return objectMapper.readValue(

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

@@ -18,6 +18,7 @@
 import io.javaoperatorsdk.operator.processing.retry.RetryExecution;
 import java.util.HashMap;
 import java.util.HashSet;
+import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
@@ -109,12 +110,26 @@ public void handleEvent(Event event) {
       lock.lock();
       log.debug("Received event: {}", event);
       eventBuffer.addEvent(event);
-      executeBufferedEvents(event.getRelatedCustomResourceUid());
+
+      if (event.getRelatedCustomResourceUid() != null) {
+        executeBufferedEvents(event.getRelatedCustomResourceUid());
+      } else if (event.getCustomResourcesSelector() != null) {
+        for (String uid :
+            eventSourceManager.getLatestResourceUids(event.getCustomResourcesSelector())) {
+          executeBufferedEvents(uid);
+        }
+      }
     } finally {
       lock.unlock();
     }
   }
 
+  private void executeBufferedEvents(List<String> customResourceUids) {
+    for (String customResourceUid : customResourceUids) {
+      executeBufferedEvents(customResourceUid);
+    }
+  }
+
   private void executeBufferedEvents(String customResourceUid) {
     boolean newEventForResourceId = eventBuffer.containsEvents(customResourceUid);
     boolean controllerUnderExecution = isControllerUnderExecution(customResourceUid);

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

@@ -1,13 +1,24 @@
 package io.javaoperatorsdk.operator.processing.event;
 
+import io.fabric8.kubernetes.client.CustomResource;
+import java.util.function.Predicate;
+
+@SuppressWarnings("rawtypes")
 public abstract class AbstractEvent implements Event {
 
   private final String relatedCustomResourceUid;
-
+  private final Predicate<CustomResource> customResourcesSelector;
   private final EventSource eventSource;
 
   public AbstractEvent(String relatedCustomResourceUid, EventSource eventSource) {
     this.relatedCustomResourceUid = relatedCustomResourceUid;
+    this.customResourcesSelector = null;
+    this.eventSource = eventSource;
+  }
+
+  public AbstractEvent(Predicate<CustomResource> customResourcesSelector, EventSource eventSource) {
+    this.relatedCustomResourceUid = null;
+    this.customResourcesSelector = customResourcesSelector;
     this.eventSource = eventSource;
   }
 
@@ -16,6 +27,10 @@ public String getRelatedCustomResourceUid() {
     return relatedCustomResourceUid;
   }
 
+  public Predicate<CustomResource> getCustomResourcesSelector() {
+    return customResourcesSelector;
+  }
+
   @Override
   public EventSource getEventSource() {
     return eventSource;
@@ -27,6 +42,8 @@ public String toString() {
         + this.getClass().getName()
         + ", relatedCustomResourceUid="
         + relatedCustomResourceUid
+        + ", customResourcesSelector="
+        + customResourcesSelector
         + ", eventSource="
         + eventSource
         + " }";

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

@@ -12,9 +12,11 @@
 import io.javaoperatorsdk.operator.processing.event.internal.CustomResourceEventSource;
 import io.javaoperatorsdk.operator.processing.event.internal.TimerEventSource;
 import java.util.Collections;
+import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
+import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.locks.ReentrantLock;
 import java.util.function.Predicate;
@@ -163,6 +165,16 @@ public Optional<CustomResource> getLatestResource(String customResourceUid) {
     return getCache().getLatestResource(customResourceUid);
   }
 
+  // todo: remove
+  public List<CustomResource> getLatestResources(Predicate<CustomResource> selector) {
+    return getCache().getLatestResources(selector);
+  }
+
+  // todo: remove
+  public Set<String> getLatestResourceUids(Predicate<CustomResource> selector) {
+    return getCache().getLatestResourcesUids(selector);
+  }
+
   // todo: remove
   public void cacheResource(CustomResource resource, Predicate<CustomResource> predicate) {
     getCache().cacheResource(resource, predicate);

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

@@ -1,8 +1,24 @@
 package io.javaoperatorsdk.operator.processing.event;
 
+import io.fabric8.kubernetes.client.CustomResource;
+import java.util.function.Predicate;
+
 public interface Event {
 
+  /**
+   * @return the UID of the the {@link CustomResource} for which a reconcile loop should be
+   *     triggered.
+   * @deprecated use {@link #getCustomResourcesSelector()}
+   */
+  @Deprecated
   String getRelatedCustomResourceUid();
 
+  /**
+   * The selector used to determine the {@link CustomResource} for which a reconcile loop should be
+   * triggered.
+   */
+  Predicate<CustomResource> getCustomResourcesSelector();
+
+  /** @return the {@link EventSource} that has generated the event. */
   EventSource getEventSource();
 }