Event Notification and Polling, Inbound Event Sources

operator-framework-core/pom.xml

@@ -95,7 +95,6 @@
       <artifactId>assertj-core</artifactId>
       <scope>test</scope>
     </dependency>
-
     <dependency>
       <groupId>io.fabric8</groupId>
       <artifactId>kubernetes-server-mock</artifactId>
@@ -106,5 +105,20 @@
       <artifactId>awaitility</artifactId>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>javax.cache</groupId>
+      <artifactId>cache-api</artifactId>
+      <version>${jcache.version}</version>
+    </dependency>
+    <dependency>
+      <groupId>com.github.ben-manes.caffeine</groupId>
+      <artifactId>caffeine</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>com.github.ben-manes.caffeine</groupId>
+      <artifactId>jcache</artifactId>
+      <scope>test</scope>
+    </dependency>
   </dependencies>
 </project>

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

@@ -7,8 +7,8 @@
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.fabric8.kubernetes.client.CustomResource;
 import io.javaoperatorsdk.operator.ReconcilerUtils;
-import io.javaoperatorsdk.operator.processing.event.source.ResourceEventFilter;
-import io.javaoperatorsdk.operator.processing.event.source.ResourceEventFilters;
+import io.javaoperatorsdk.operator.processing.event.source.controller.ResourceEventFilter;
+import io.javaoperatorsdk.operator.processing.event.source.controller.ResourceEventFilters;
 
 public interface ControllerConfiguration<R extends HasMetadata> {
 

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

@@ -5,7 +5,7 @@
 import java.util.Set;
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
-import io.javaoperatorsdk.operator.processing.event.source.ResourceEventFilter;
+import io.javaoperatorsdk.operator.processing.event.source.controller.ResourceEventFilter;
 
 public class ControllerConfigurationOverrider<R extends HasMetadata> {
 

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

@@ -4,7 +4,7 @@
 import java.util.Set;
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
-import io.javaoperatorsdk.operator.processing.event.source.ResourceEventFilter;
+import io.javaoperatorsdk.operator.processing.event.source.controller.ResourceEventFilter;
 
 public class DefaultControllerConfiguration<R extends HasMetadata>
     implements ControllerConfiguration<R> {

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

@@ -5,7 +5,7 @@
 import java.lang.annotation.RetentionPolicy;
 import java.lang.annotation.Target;
 
-import io.javaoperatorsdk.operator.processing.event.source.ResourceEventFilter;
+import io.javaoperatorsdk.operator.processing.event.source.controller.ResourceEventFilter;
 
 @Retention(RetentionPolicy.RUNTIME)
 @Target({ElementType.TYPE})

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

@@ -212,6 +212,10 @@ private boolean failOnMissingCurrentNS() {
     return false;
   }
 
+  public EventSourceManager<R> getEventSourceManager() {
+    return eventSourceManager;
+  }
+
   public void stop() {
     if (eventSourceManager != null) {
       eventSourceManager.stop();

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

@@ -1,5 +1,7 @@
 package io.javaoperatorsdk.operator.processing.event;
 
+import java.util.Objects;
+
 public class Event {
 
   private final ResourceID relatedCustomResource;
@@ -18,4 +20,19 @@ public String toString() {
         "relatedCustomResource=" + relatedCustomResource +
         '}';
   }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o)
+      return true;
+    if (o == null || getClass() != o.getClass())
+      return false;
+    Event event = (Event) o;
+    return Objects.equals(relatedCustomResource, event.relatedCustomResource);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(relatedCustomResource);
+  }
 }

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

@@ -20,10 +20,10 @@
 import io.javaoperatorsdk.operator.api.reconciler.RetryInfo;
 import io.javaoperatorsdk.operator.processing.LifecycleAware;
 import io.javaoperatorsdk.operator.processing.MDCUtils;
-import io.javaoperatorsdk.operator.processing.event.source.ResourceAction;
-import io.javaoperatorsdk.operator.processing.event.source.ResourceCache;
-import io.javaoperatorsdk.operator.processing.event.source.ResourceEvent;
-import io.javaoperatorsdk.operator.processing.event.source.TimerEventSource;
+import io.javaoperatorsdk.operator.processing.event.source.controller.ResourceAction;
+import io.javaoperatorsdk.operator.processing.event.source.controller.ResourceCache;
+import io.javaoperatorsdk.operator.processing.event.source.controller.ResourceEvent;
+import io.javaoperatorsdk.operator.processing.event.source.timer.TimerEventSource;
 import io.javaoperatorsdk.operator.processing.retry.GenericRetry;
 import io.javaoperatorsdk.operator.processing.retry.Retry;
 import io.javaoperatorsdk.operator.processing.retry.RetryExecution;
@@ -297,7 +297,6 @@ private RetryExecution getOrInitRetryExecution(ExecutionScope<R> executionScope)
   }
 
   private void cleanupForDeletedEvent(ResourceID customResourceUid) {
-    eventSourceManager.cleanupForCustomResource(customResourceUid);
     eventMarker.cleanup(customResourceUid);
     metrics.cleanupDoneFor(customResourceUid);
   }

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

@@ -11,10 +11,12 @@
 import io.javaoperatorsdk.operator.OperatorException;
 import io.javaoperatorsdk.operator.processing.Controller;
 import io.javaoperatorsdk.operator.processing.LifecycleAware;
-import io.javaoperatorsdk.operator.processing.event.source.ControllerResourceEventSource;
 import io.javaoperatorsdk.operator.processing.event.source.EventSource;
 import io.javaoperatorsdk.operator.processing.event.source.EventSourceRegistry;
-import io.javaoperatorsdk.operator.processing.event.source.TimerEventSource;
+import io.javaoperatorsdk.operator.processing.event.source.ResourceEventAware;
+import io.javaoperatorsdk.operator.processing.event.source.controller.ControllerResourceEventSource;
+import io.javaoperatorsdk.operator.processing.event.source.controller.ResourceAction;
+import io.javaoperatorsdk.operator.processing.event.source.timer.TimerEventSource;
 
 public class EventSourceManager<R extends HasMetadata>
     implements EventSourceRegistry<R>, LifecycleAware {
@@ -107,14 +109,22 @@ public final void registerEventSource(EventSource eventSource)
     }
   }
 
-  public void cleanupForCustomResource(ResourceID customResourceUid) {
-    lock.lock();
-    try {
-      for (EventSource eventSource : this.eventSources) {
-        eventSource.cleanupForResource(customResourceUid);
+  public void broadcastOnResourceEvent(ResourceAction action, R resource, R oldResource) {
+    for (EventSource eventSource : this.eventSources) {
+      if (eventSource instanceof ResourceEventAware) {
+        var lifecycleAwareES = ((ResourceEventAware<R>) eventSource);
+        switch (action) {
+          case ADDED:
+            lifecycleAwareES.onResourceCreated(resource);
+            break;
+          case UPDATED:
+            lifecycleAwareES.onResourceUpdated(resource, oldResource);
+            break;
+          case DELETED:
+            lifecycleAwareES.onResourceDeleted(resource);
+            break;
+        }
       }
-    } finally {
-      lock.unlock();
     }
   }
 

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

@@ -1,11 +1,12 @@
 package io.javaoperatorsdk.operator.processing.event;
 
+import java.io.Serializable;
 import java.util.Objects;
 import java.util.Optional;
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
 
-public class ResourceID {
+public class ResourceID implements Serializable {
 
   public static ResourceID fromResource(HasMetadata resource) {
     return new ResourceID(resource.getMetadata().getName(),

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

@@ -10,4 +10,5 @@ public abstract class AbstractEventSource implements EventSource {
   public void setEventHandler(EventHandler eventHandler) {
     this.eventHandler = eventHandler;
   }
+
 }

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

@@ -0,0 +1,79 @@
+package io.javaoperatorsdk.operator.processing.event.source;
+
+import java.util.Optional;
+
+import javax.cache.Cache;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import io.javaoperatorsdk.operator.OperatorException;
+import io.javaoperatorsdk.operator.processing.event.Event;
+import io.javaoperatorsdk.operator.processing.event.ResourceID;
+
+/**
+ * Base class for event sources with filtering and caching capabilities.
+ * <p>
+ * {@link #handleDelete(ResourceID)} - if the related resource is present in the cache it is
+ * removed. and event propagated. There is no event propagated if the resource is not in the cache.
+ * <p>
+ * {@link #handleEvent(Object, ResourceID)} - caches the resource if changed (or nto present
+ * before). Propagates an event if the resource is new or not equals to the one in the cache, and if
+ * accepted by the filter if one is present.
+ *
+ * @param <T> represents the resource (usually external non-kubernetes one) handled.
+ */
+public abstract class CachingFilteringEventSource<T> extends LifecycleAwareEventSource {
+
+  private static final Logger log = LoggerFactory.getLogger(CachingFilteringEventSource.class);
+
+  protected Cache<ResourceID, T> cache;
+  protected EventFilter<T> eventFilter;
+
+  public CachingFilteringEventSource(Cache<ResourceID, T> cache, EventFilter<T> eventFilter) {
+    this.cache = cache;
+    this.eventFilter = eventFilter;
+  }
+
+  protected void handleDelete(ResourceID relatedResourceID) {
+    if (!isRunning()) {
+      log.debug("Received event but event for {} source is not running", relatedResourceID);
+      return;
+    }
+    var cachedValue = cache.get(relatedResourceID);
+    cache.remove(relatedResourceID);
+    // we only propagate event if the resource was previously in cache
+    if (cachedValue != null
+        && (eventFilter == null || eventFilter.acceptDelete(cachedValue, relatedResourceID))) {
+      eventHandler.handleEvent(new Event(relatedResourceID));
+    }
+  }
+
+  protected void handleEvent(T value, ResourceID relatedResourceID) {
+    if (!isRunning()) {
+      log.debug("Received event but event for {} source is not running", relatedResourceID);
+      return;
+    }
+    var cachedValue = cache.get(relatedResourceID);
+    if (cachedValue == null || !cachedValue.equals(value)) {
+      cache.put(relatedResourceID, value);
+      if (eventFilter == null || eventFilter.accept(value, cachedValue, relatedResourceID)) {
+        eventHandler.handleEvent(new Event(relatedResourceID));
+      }
+    }
+  }
+
+  public Cache<ResourceID, T> getCache() {
+    return cache;
+  }
+
+  public Optional<T> getCachedValue(ResourceID resourceID) {
+    return Optional.ofNullable(cache.get(resourceID));
+  }
+
+  @Override
+  public void stop() throws OperatorException {
+    super.stop();
+    cache.close();
+  }
+}

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

@@ -0,0 +1,12 @@
+package io.javaoperatorsdk.operator.processing.event.source;
+
+import io.javaoperatorsdk.operator.processing.event.ResourceID;
+
+public interface EventFilter<T> {
+
+  boolean accept(T newValue, T oldValue, ResourceID relatedResourceID);
+
+  default boolean acceptDelete(T value, ResourceID relatedResourceID) {
+    return true;
+  }
+}

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

@@ -2,16 +2,9 @@
 
 import io.javaoperatorsdk.operator.processing.LifecycleAware;
 import io.javaoperatorsdk.operator.processing.event.EventHandler;
-import io.javaoperatorsdk.operator.processing.event.ResourceID;
 
 public interface EventSource extends LifecycleAware {
 
   void setEventHandler(EventHandler eventHandler);
 
-  /**
-   * Automatically called when a custom resource is deleted from the cluster.
-   *
-   * @param customResourceUid - id of custom resource
-   */
-  default void cleanupForResource(ResourceID customResourceUid) {}
 }

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

@@ -4,6 +4,7 @@
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.javaoperatorsdk.operator.OperatorException;
+import io.javaoperatorsdk.operator.processing.event.source.controller.ControllerResourceEventSource;
 
 public interface EventSourceRegistry<T extends HasMetadata> {
 

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

@@ -0,0 +1,22 @@
+package io.javaoperatorsdk.operator.processing.event.source;
+
+import io.javaoperatorsdk.operator.OperatorException;
+
+public abstract class LifecycleAwareEventSource extends AbstractEventSource {
+
+  private volatile boolean running = false;
+
+  public boolean isRunning() {
+    return running;
+  }
+
+  @Override
+  public void start() throws OperatorException {
+    running = true;
+  }
+
+  @Override
+  public void stop() throws OperatorException {
+    running = false;
+  }
+}

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

@@ -0,0 +1,13 @@
+package io.javaoperatorsdk.operator.processing.event.source;
+
+import io.fabric8.kubernetes.api.model.HasMetadata;
+
+public interface ResourceEventAware<T extends HasMetadata> {
+
+  default void onResourceCreated(T resource) {}
+
+  default void onResourceUpdated(T newResource, T oldResource) {}
+
+  default void onResourceDeleted(T resource) {}
+
+}

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/ControllerResourceCache.java → operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/controller/ControllerResourceCache.java

@@ -1,4 +1,4 @@
-package io.javaoperatorsdk.operator.processing.event.source;
+package io.javaoperatorsdk.operator.processing.event.source.controller;
 
 import java.util.Map;
 import java.util.Optional;
@@ -11,7 +11,7 @@
 import io.javaoperatorsdk.operator.api.config.Cloner;
 import io.javaoperatorsdk.operator.processing.event.ResourceID;
 
-import static io.javaoperatorsdk.operator.processing.event.source.ControllerResourceEventSource.ANY_NAMESPACE_MAP_KEY;
+import static io.javaoperatorsdk.operator.processing.event.source.controller.ControllerResourceEventSource.ANY_NAMESPACE_MAP_KEY;
 
 public class ControllerResourceCache<T extends HasMetadata> implements ResourceCache<T> {