diff --git a/operator-framework-core/pom.xml b/operator-framework-core/pom.xml index af0737bf29..eafb9ec1b1 100644 --- a/operator-framework-core/pom.xml +++ b/operator-framework-core/pom.xml @@ -95,7 +95,6 @@ assertj-core test - io.fabric8 kubernetes-server-mock @@ -106,5 +105,20 @@ awaitility test + + javax.cache + cache-api + ${jcache.version} + + + com.github.ben-manes.caffeine + caffeine + test + + + com.github.ben-manes.caffeine + jcache + test + diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ControllerConfiguration.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ControllerConfiguration.java index adc1256b79..31d701c03d 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ControllerConfiguration.java +++ b/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 { diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ControllerConfigurationOverrider.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ControllerConfigurationOverrider.java index 515addcad0..e8e2ef1162 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ControllerConfigurationOverrider.java +++ b/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 { diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/DefaultControllerConfiguration.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/DefaultControllerConfiguration.java index 965ac32f76..860152745b 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/DefaultControllerConfiguration.java +++ b/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 implements ControllerConfiguration { diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/ControllerConfiguration.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/ControllerConfiguration.java index 7abd5f7b43..b5c6265ae1 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/ControllerConfiguration.java +++ b/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}) diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/Controller.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/Controller.java index 1bbb48873c..c8c96cfd6a 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/Controller.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/Controller.java @@ -212,6 +212,10 @@ private boolean failOnMissingCurrentNS() { return false; } + public EventSourceManager getEventSourceManager() { + return eventSourceManager; + } + public void stop() { if (eventSourceManager != null) { eventSourceManager.stop(); diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/Event.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/Event.java index deacb85956..0f13b71e11 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/Event.java +++ b/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); + } } diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/EventProcessor.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/EventProcessor.java index b3a2fd2896..1f0161aeef 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/EventProcessor.java +++ b/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 executionScope) } private void cleanupForDeletedEvent(ResourceID customResourceUid) { - eventSourceManager.cleanupForCustomResource(customResourceUid); eventMarker.cleanup(customResourceUid); metrics.cleanupDoneFor(customResourceUid); } diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/EventSourceManager.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/EventSourceManager.java index 50f2af8641..6848fcbb37 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/EventSourceManager.java +++ b/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 implements EventSourceRegistry, 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) 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(); } } diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/ResourceID.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/ResourceID.java index 515be245b8..38c9055ff1 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/ResourceID.java +++ b/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(), diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/AbstractEventSource.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/AbstractEventSource.java index 555da0d1b1..ddc787ad2d 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/AbstractEventSource.java +++ b/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; } + } diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/CachingEventSource.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/CachingEventSource.java new file mode 100644 index 0000000000..9a2be41a70 --- /dev/null +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/CachingEventSource.java @@ -0,0 +1,72 @@ +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 caching capabilities. + *

+ * {@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. + *

+ * {@link #handleEvent(Object, ResourceID)} - caches the resource if changed or missing. 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 represents the type of resources (usually external non-kubernetes ones) being handled. + */ +public abstract class CachingEventSource extends LifecycleAwareEventSource { + + private static final Logger log = LoggerFactory.getLogger(CachingEventSource.class); + + protected Cache cache; + + public CachingEventSource(Cache cache) { + this.cache = cache; + } + + protected void handleDelete(ResourceID relatedResourceID) { + if (!isRunning()) { + return; + } + var cachedValue = cache.get(relatedResourceID); + cache.remove(relatedResourceID); + // we only propagate event if the resource was previously in cache + if (cachedValue != null) { + eventHandler.handleEvent(new Event(relatedResourceID)); + } + } + + protected void handleEvent(T value, ResourceID relatedResourceID) { + if (!isRunning()) { + return; + } + var cachedValue = cache.get(relatedResourceID); + if (cachedValue == null || !cachedValue.equals(value)) { + cache.put(relatedResourceID, value); + eventHandler.handleEvent(new Event(relatedResourceID)); + } + } + + public Cache getCache() { + return cache; + } + + public Optional getCachedValue(ResourceID resourceID) { + return Optional.ofNullable(cache.get(resourceID)); + } + + @Override + public void stop() throws OperatorException { + super.stop(); + cache.close(); + } +} diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/EventSource.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/EventSource.java index e0d0bc45e1..18e47d03db 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/EventSource.java +++ b/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) {} } diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/EventSourceRegistry.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/EventSourceRegistry.java index a2d38690cc..dca5436427 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/EventSourceRegistry.java +++ b/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 { diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/LifecycleAwareEventSource.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/LifecycleAwareEventSource.java new file mode 100644 index 0000000000..6b2d79fd1a --- /dev/null +++ b/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; + } +} diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/ResourceAction.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/ResourceAction.java deleted file mode 100644 index ff0b4673be..0000000000 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/ResourceAction.java +++ /dev/null @@ -1,5 +0,0 @@ -package io.javaoperatorsdk.operator.processing.event.source; - -public enum ResourceAction { - ADDED, UPDATED, DELETED -} diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/ResourceEventAware.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/ResourceEventAware.java new file mode 100644 index 0000000000..dcb15a4229 --- /dev/null +++ b/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 { + + default void onResourceCreated(T resource) {} + + default void onResourceUpdated(T newResource, T oldResource) {} + + default void onResourceDeleted(T resource) {} + +} diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/ControllerResourceCache.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/controller/ControllerResourceCache.java similarity index 94% rename from operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/ControllerResourceCache.java rename to operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/controller/ControllerResourceCache.java index 2186277e32..2397af9573 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/ControllerResourceCache.java +++ b/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 implements ResourceCache { diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/ControllerResourceEventSource.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/controller/ControllerResourceEventSource.java similarity index 83% rename from operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/ControllerResourceEventSource.java rename to operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/controller/ControllerResourceEventSource.java index faf0d8f33b..2386cd4dbc 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/ControllerResourceEventSource.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/controller/ControllerResourceEventSource.java @@ -1,4 +1,4 @@ -package io.javaoperatorsdk.operator.processing.event.source; +package io.javaoperatorsdk.operator.processing.event.source.controller; import java.util.Collections; import java.util.Map; @@ -20,6 +20,7 @@ import io.javaoperatorsdk.operator.processing.Controller; import io.javaoperatorsdk.operator.processing.MDCUtils; import io.javaoperatorsdk.operator.processing.event.ResourceID; +import io.javaoperatorsdk.operator.processing.event.source.AbstractEventSource; import static io.javaoperatorsdk.operator.processing.KubernetesResourceUtils.getName; import static io.javaoperatorsdk.operator.processing.KubernetesResourceUtils.getUID; @@ -75,30 +76,20 @@ public void start() { try { if (ControllerConfiguration.allNamespacesWatched(targetNamespaces)) { - final var filteredBySelectorClient = client.inAnyNamespace() - .withLabelSelector(labelSelector); final var informer = - createAndRunInformerFor(filteredBySelectorClient, ANY_NAMESPACE_MAP_KEY); + createAndRunInformerFor(client.inAnyNamespace() + .withLabelSelector(labelSelector), ANY_NAMESPACE_MAP_KEY); log.debug("Registered {} -> {} for any namespace", controller, informer); } else { - targetNamespaces.forEach( - ns -> { - final var informer = createAndRunInformerFor( - client.inNamespace(ns).withLabelSelector(labelSelector), ns); - log.debug("Registered {} -> {} for namespace: {}", controller, informer, - ns); - }); + targetNamespaces.forEach(ns -> { + final var informer = createAndRunInformerFor( + client.inNamespace(ns).withLabelSelector(labelSelector), ns); + log.debug("Registered {} -> {} for namespace: {}", controller, informer, ns); + }); } } catch (Exception e) { if (e instanceof KubernetesClientException) { - KubernetesClientException ke = (KubernetesClientException) e; - if (404 == ke.getCode()) { - // only throw MissingCRDException if the 404 error occurs on the target CRD - final var targetCRDName = controller.getConfiguration().getResourceTypeName(); - if (targetCRDName.equals(ke.getFullResourceName())) { - throw new MissingCRDException(targetCRDName, null, e.getMessage(), e); - } - } + handleKubernetesClientException(e); } throw e; } @@ -130,6 +121,8 @@ public void eventReceived(ResourceAction action, T customResource, T oldResource log.debug( "Event received for resource: {}", getName(customResource)); MDCUtils.addResourceInfo(customResource); + controller.getEventSourceManager().broadcastOnResourceEvent(action, customResource, + oldResource); if (filter.acceptChange(controller.getConfiguration(), oldResource, customResource)) { eventHandler.handleEvent( new ResourceEvent(action, ResourceID.fromResource(customResource))); @@ -191,4 +184,16 @@ public void whitelistNextEvent(ResourceID resourceID) { } } + + private void handleKubernetesClientException(Exception e) { + KubernetesClientException ke = (KubernetesClientException) e; + if (404 == ke.getCode()) { + // only throw MissingCRDException if the 404 error occurs on the target CRD + final var targetCRDName = controller.getConfiguration().getResourceTypeName(); + if (targetCRDName.equals(ke.getFullResourceName())) { + throw new MissingCRDException(targetCRDName, null, e.getMessage(), e); + } + } + } + } diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/OnceWhitelistEventFilterEventFilter.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/controller/OnceWhitelistEventFilterEventFilter.java similarity index 94% rename from operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/OnceWhitelistEventFilterEventFilter.java rename to operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/controller/OnceWhitelistEventFilterEventFilter.java index aba822dcf6..8262ff1c21 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/OnceWhitelistEventFilterEventFilter.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/controller/OnceWhitelistEventFilterEventFilter.java @@ -1,4 +1,4 @@ -package io.javaoperatorsdk.operator.processing.event.source; +package io.javaoperatorsdk.operator.processing.event.source.controller; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/controller/ResourceAction.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/controller/ResourceAction.java new file mode 100644 index 0000000000..7a04dc9164 --- /dev/null +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/controller/ResourceAction.java @@ -0,0 +1,5 @@ +package io.javaoperatorsdk.operator.processing.event.source.controller; + +public enum ResourceAction { + ADDED, UPDATED, DELETED +} diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/ResourceCache.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/controller/ResourceCache.java similarity index 89% rename from operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/ResourceCache.java rename to operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/controller/ResourceCache.java index af156e24ec..5508dd43eb 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/ResourceCache.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/controller/ResourceCache.java @@ -1,4 +1,4 @@ -package io.javaoperatorsdk.operator.processing.event.source; +package io.javaoperatorsdk.operator.processing.event.source.controller; import java.util.Optional; import java.util.function.Predicate; diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/ResourceEvent.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/controller/ResourceEvent.java similarity index 88% rename from operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/ResourceEvent.java rename to operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/controller/ResourceEvent.java index 7610a880b7..ad1d85330c 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/ResourceEvent.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/controller/ResourceEvent.java @@ -1,4 +1,4 @@ -package io.javaoperatorsdk.operator.processing.event.source; +package io.javaoperatorsdk.operator.processing.event.source.controller; import io.javaoperatorsdk.operator.processing.event.Event; import io.javaoperatorsdk.operator.processing.event.ResourceID; diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/ResourceEventFilter.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/controller/ResourceEventFilter.java similarity index 97% rename from operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/ResourceEventFilter.java rename to operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/controller/ResourceEventFilter.java index 01f7e1aec0..497c9016b7 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/ResourceEventFilter.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/controller/ResourceEventFilter.java @@ -1,4 +1,4 @@ -package io.javaoperatorsdk.operator.processing.event.source; +package io.javaoperatorsdk.operator.processing.event.source.controller; import io.fabric8.kubernetes.api.model.HasMetadata; import io.javaoperatorsdk.operator.api.config.ControllerConfiguration; diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/ResourceEventFilters.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/controller/ResourceEventFilters.java similarity index 98% rename from operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/ResourceEventFilters.java rename to operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/controller/ResourceEventFilters.java index 9d90ebd963..43fe410fbc 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/ResourceEventFilters.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/controller/ResourceEventFilters.java @@ -1,4 +1,4 @@ -package io.javaoperatorsdk.operator.processing.event.source; +package io.javaoperatorsdk.operator.processing.event.source.controller; import io.fabric8.kubernetes.api.model.HasMetadata; import io.fabric8.kubernetes.client.CustomResource; diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/inbound/CachingInboundEventSource.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/inbound/CachingInboundEventSource.java new file mode 100644 index 0000000000..51d1ed287d --- /dev/null +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/inbound/CachingInboundEventSource.java @@ -0,0 +1,21 @@ +package io.javaoperatorsdk.operator.processing.event.source.inbound; + +import javax.cache.Cache; + +import io.javaoperatorsdk.operator.processing.event.ResourceID; +import io.javaoperatorsdk.operator.processing.event.source.CachingEventSource; + +public class CachingInboundEventSource extends CachingEventSource { + + public CachingInboundEventSource(Cache cache) { + super(cache); + } + + public void handleResourceEvent(T resource, ResourceID relatedResourceID) { + super.handleEvent(resource, relatedResourceID); + } + + public void handleResourceDeleteEvent(ResourceID resourceID) { + super.handleDelete(resourceID); + } +} diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/inbound/SimpleInboundEventSource.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/inbound/SimpleInboundEventSource.java new file mode 100644 index 0000000000..475cfee916 --- /dev/null +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/inbound/SimpleInboundEventSource.java @@ -0,0 +1,22 @@ +package io.javaoperatorsdk.operator.processing.event.source.inbound; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import io.javaoperatorsdk.operator.processing.event.Event; +import io.javaoperatorsdk.operator.processing.event.ResourceID; +import io.javaoperatorsdk.operator.processing.event.source.LifecycleAwareEventSource; + +public class SimpleInboundEventSource extends LifecycleAwareEventSource { + + private static final Logger log = LoggerFactory.getLogger(SimpleInboundEventSource.class); + + public void propagateEvent(ResourceID resourceID) { + if (isRunning()) { + eventHandler.handleEvent(new Event(resourceID)); + } else { + log.debug("Event source not started yet, not propagating event for: {}", resourceID); + } + } + +} diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/InformerEventSource.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerEventSource.java similarity index 96% rename from operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/InformerEventSource.java rename to operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerEventSource.java index 3bdc8fb764..8095289221 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/InformerEventSource.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerEventSource.java @@ -1,4 +1,4 @@ -package io.javaoperatorsdk.operator.processing.event.source; +package io.javaoperatorsdk.operator.processing.event.source.informer; import java.util.Objects; import java.util.Set; @@ -15,6 +15,7 @@ import io.fabric8.kubernetes.client.informers.cache.Store; import io.javaoperatorsdk.operator.processing.event.Event; import io.javaoperatorsdk.operator.processing.event.ResourceID; +import io.javaoperatorsdk.operator.processing.event.source.AbstractEventSource; public class InformerEventSource extends AbstractEventSource { diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/Mappers.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/Mappers.java similarity index 95% rename from operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/Mappers.java rename to operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/Mappers.java index d428bb26cc..c578490147 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/Mappers.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/Mappers.java @@ -1,4 +1,4 @@ -package io.javaoperatorsdk.operator.processing.event.source; +package io.javaoperatorsdk.operator.processing.event.source.informer; import java.util.Collections; import java.util.Set; diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/polling/PerResourcePollingEventSource.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/polling/PerResourcePollingEventSource.java new file mode 100644 index 0000000000..bf9b41cf0e --- /dev/null +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/polling/PerResourcePollingEventSource.java @@ -0,0 +1,146 @@ +package io.javaoperatorsdk.operator.processing.event.source.polling; + +import java.util.Map; +import java.util.Optional; +import java.util.Timer; +import java.util.TimerTask; +import java.util.concurrent.ConcurrentHashMap; +import java.util.function.Predicate; + +import javax.cache.Cache; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import io.fabric8.kubernetes.api.model.HasMetadata; +import io.javaoperatorsdk.operator.OperatorException; +import io.javaoperatorsdk.operator.processing.event.ResourceID; +import io.javaoperatorsdk.operator.processing.event.source.CachingEventSource; +import io.javaoperatorsdk.operator.processing.event.source.ResourceEventAware; +import io.javaoperatorsdk.operator.processing.event.source.controller.ResourceCache; + +/** + * + * Polls the supplier for each controlled resource registered. Resource is registered when created + * if there is no registerPredicate provided. If register predicate provided it is evaluated on + * resource create and/or update to register polling for the event source. + *

+ * For other behavior see {@link CachingEventSource} + * + * @param the resource polled by the event source + * @param related custom resource + */ +public class PerResourcePollingEventSource + extends CachingEventSource + implements ResourceEventAware { + + private static final Logger log = LoggerFactory.getLogger(PerResourcePollingEventSource.class); + + private final Timer timer = new Timer(); + private final Map timerTasks = new ConcurrentHashMap<>(); + private final ResourceSupplier resourceSupplier; + private final ResourceCache resourceCache; + private final Predicate registerPredicate; + private final long period; + + public PerResourcePollingEventSource(ResourceSupplier resourceSupplier, + ResourceCache resourceCache, long period, Cache cache) { + this(resourceSupplier, resourceCache, period, cache, null); + } + + public PerResourcePollingEventSource(ResourceSupplier resourceSupplier, + ResourceCache resourceCache, long period, Cache cache, + Predicate registerPredicate) { + super(cache); + this.resourceSupplier = resourceSupplier; + this.resourceCache = resourceCache; + this.period = period; + this.registerPredicate = registerPredicate; + } + + private void pollForResource(R resource) { + var value = resourceSupplier.getResources(resource); + var resourceID = ResourceID.fromResource(resource); + if (value.isEmpty()) { + super.handleDelete(resourceID); + } else { + super.handleEvent(value.get(), resourceID); + } + } + + private Optional getAndCacheResource(ResourceID resourceID) { + var resource = resourceCache.get(resourceID); + if (resource.isPresent()) { + var value = resourceSupplier.getResources(resource.get()); + value.ifPresent(v -> cache.put(resourceID, v)); + return value; + } + return Optional.empty(); + } + + @Override + public void onResourceCreated(R resource) { + checkAndRegisterTask(resource); + } + + @Override + public void onResourceUpdated(R newResource, R oldResource) { + checkAndRegisterTask(newResource); + } + + @Override + public void onResourceDeleted(R resource) { + var resourceID = ResourceID.fromResource(resource); + TimerTask task = timerTasks.remove(resourceID); + if (task != null) { + task.cancel(); + } + cache.remove(resourceID); + } + + private void checkAndRegisterTask(R resource) { + var resourceID = ResourceID.fromResource(resource); + if (timerTasks.get(resourceID) == null && (registerPredicate == null + || registerPredicate.test(resource))) { + timer.schedule(new TimerTask() { + @Override + public void run() { + if (!isRunning()) { + log.debug("Event source not yet started. Will not run for: {}", resourceID); + return; + } + // always use up-to-date resource from cache + var res = resourceCache.get(resourceID); + res.ifPresentOrElse(r -> pollForResource(r), + () -> log.warn("No resource in cache for resource ID: {}", resourceID)); + } + }, 0, period); + } + } + + /** + * + * @param resourceID of the target related resource + * @return the cached value of the resource, if not present it gets the resource from the + * supplier. The value provided from the supplier is cached, but no new event is + * propagated. + */ + public Optional getValueFromCacheOrSupplier(ResourceID resourceID) { + var cachedValue = getCachedValue(resourceID); + if (cachedValue.isPresent()) { + return cachedValue; + } else { + return getAndCacheResource(resourceID); + } + } + + public interface ResourceSupplier { + Optional getResources(R resource); + } + + @Override + public void stop() throws OperatorException { + super.stop(); + timer.cancel(); + } +} diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/polling/PollingEventSource.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/polling/PollingEventSource.java new file mode 100644 index 0000000000..b2c3fdff78 --- /dev/null +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/polling/PollingEventSource.java @@ -0,0 +1,68 @@ +package io.javaoperatorsdk.operator.processing.event.source.polling; + +import java.util.*; +import java.util.function.Supplier; +import java.util.stream.StreamSupport; + +import javax.cache.Cache; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import io.javaoperatorsdk.operator.OperatorException; +import io.javaoperatorsdk.operator.processing.event.ResourceID; +import io.javaoperatorsdk.operator.processing.event.source.CachingEventSource; + +public class PollingEventSource extends CachingEventSource { + + private static final Logger log = LoggerFactory.getLogger(PollingEventSource.class); + + private final Timer timer = new Timer(); + private final Supplier> supplierToPoll; + private final long period; + + public PollingEventSource(Supplier> supplier, + long period, Cache cache) { + super(cache); + this.supplierToPoll = supplier; + this.period = period; + } + + @Override + public void start() throws OperatorException { + super.start(); + timer.schedule(new TimerTask() { + @Override + public void run() { + if (!isRunning()) { + log.debug("Event source not yet started. Will not run."); + return; + } + getStateAndFillCache(); + } + }, period, period); + } + + protected void getStateAndFillCache() { + var values = supplierToPoll.get(); + values.forEach((k, v) -> super.handleEvent(v, k)); + StreamSupport.stream(cache.spliterator(), false) + .filter(e -> !values.containsKey(e.getKey())).map(Cache.Entry::getKey) + .forEach(super::handleDelete); + } + + @Override + public void stop() throws OperatorException { + super.stop(); + timer.cancel(); + } + + public Optional getValueFromCacheOrSupplier(ResourceID resourceID) { + var resource = getCachedValue(resourceID); + if (resource.isPresent()) { + return resource; + } + getStateAndFillCache(); + return getCachedValue(resourceID); + } +} diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/TimerEventSource.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/timer/TimerEventSource.java similarity index 84% rename from operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/TimerEventSource.java rename to operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/timer/TimerEventSource.java index ebac87cdf2..2ab8b2f128 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/TimerEventSource.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/timer/TimerEventSource.java @@ -1,4 +1,4 @@ -package io.javaoperatorsdk.operator.processing.event.source; +package io.javaoperatorsdk.operator.processing.event.source.timer; import java.util.Map; import java.util.Timer; @@ -12,8 +12,11 @@ import io.fabric8.kubernetes.api.model.HasMetadata; import io.javaoperatorsdk.operator.processing.event.Event; import io.javaoperatorsdk.operator.processing.event.ResourceID; +import io.javaoperatorsdk.operator.processing.event.source.AbstractEventSource; +import io.javaoperatorsdk.operator.processing.event.source.ResourceEventAware; -public class TimerEventSource extends AbstractEventSource { +public class TimerEventSource extends AbstractEventSource + implements ResourceEventAware { private static final Logger log = LoggerFactory.getLogger(TimerEventSource.class); private final Timer timer = new Timer(); @@ -35,8 +38,8 @@ public void scheduleOnce(R resource, long delay) { } @Override - public void cleanupForResource(ResourceID resourceUid) { - cancelOnceSchedule(resourceUid); + public void onResourceDeleted(R resource) { + cancelOnceSchedule(ResourceID.fromResource(resource)); } public void cancelOnceSchedule(ResourceID customResourceUid) { diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/EventProcessorTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/EventProcessorTest.java index 9f99b1bdd0..f837f5eea2 100644 --- a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/EventProcessorTest.java +++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/EventProcessorTest.java @@ -12,12 +12,15 @@ import org.slf4j.LoggerFactory; import io.fabric8.kubernetes.api.model.HasMetadata; -import io.javaoperatorsdk.operator.processing.event.source.*; +import io.javaoperatorsdk.operator.processing.event.source.controller.ControllerResourceCache; +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.controller.ResourceEvent; +import io.javaoperatorsdk.operator.processing.event.source.timer.TimerEventSource; import io.javaoperatorsdk.operator.processing.retry.GenericRetry; import io.javaoperatorsdk.operator.sample.simple.TestCustomResource; import static io.javaoperatorsdk.operator.TestUtils.testCustomResource; -import static io.javaoperatorsdk.operator.processing.event.source.ResourceAction.DELETED; import static org.assertj.core.api.Assertions.assertThat; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.any; @@ -205,31 +208,6 @@ public void doNotFireEventsIfClosing() { verify(reconciliationDispatcherMock, timeout(50).times(0)).handleExecution(any()); } - @Test - public void cleansUpWhenDeleteEventReceivedAndNoEventPresent() { - Event deleteEvent = - new ResourceEvent(DELETED, prepareCREvent().getRelatedCustomResourceID()); - - eventProcessor.handleEvent(deleteEvent); - - verify(eventSourceManagerMock, times(1)) - .cleanupForCustomResource(eq(deleteEvent.getRelatedCustomResourceID())); - } - - @Test - public void cleansUpAfterExecutionIfOnlyDeleteEventMarkLeft() { - var cr = testCustomResource(); - var crEvent = prepareCREvent(ResourceID.fromResource(cr)); - eventProcessor.getEventMarker().markDeleteEventReceived(crEvent.getRelatedCustomResourceID()); - var executionScope = new ExecutionScope(cr, null); - - eventProcessor.eventProcessingFinished(executionScope, - PostExecutionControl.defaultDispatch()); - - verify(eventSourceManagerMock, times(1)) - .cleanupForCustomResource(eq(crEvent.getRelatedCustomResourceID())); - } - @Test public void whitelistNextEventIfTheCacheIsNotPropagatedAfterAnUpdate() { var crID = new ResourceID("test-cr", TEST_NAMESPACE); diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/EventSourceManagerTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/EventSourceManagerTest.java index 164a66b970..79311a7057 100644 --- a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/EventSourceManagerTest.java +++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/EventSourceManagerTest.java @@ -5,8 +5,6 @@ import org.junit.jupiter.api.Test; -import io.fabric8.kubernetes.client.CustomResource; -import io.javaoperatorsdk.operator.TestUtils; import io.javaoperatorsdk.operator.processing.event.source.EventSource; import static org.assertj.core.api.Assertions.assertThat; @@ -57,17 +55,4 @@ public void startCascadesToEventSources() { verify(eventSource, times(1)).start(); verify(eventSource2, times(1)).start(); } - - @Test - public void deRegistersEventSources() { - CustomResource customResource = TestUtils.testCustomResource(); - EventSource eventSource = mock(EventSource.class); - eventSourceManager.registerEventSource(eventSource); - - eventSourceManager - .cleanupForCustomResource(ResourceID.fromResource(customResource)); - - verify(eventSource, times(1)) - .cleanupForResource(eq(ResourceID.fromResource(customResource))); - } } diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/CachingEventSourceTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/CachingEventSourceTest.java new file mode 100644 index 0000000000..15fcca7253 --- /dev/null +++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/CachingEventSourceTest.java @@ -0,0 +1,97 @@ +package io.javaoperatorsdk.operator.processing.event.source; + +import javax.cache.Cache; +import javax.cache.CacheManager; +import javax.cache.configuration.MutableConfiguration; +import javax.cache.spi.CachingProvider; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import io.javaoperatorsdk.operator.processing.event.Event; +import io.javaoperatorsdk.operator.processing.event.EventHandler; +import io.javaoperatorsdk.operator.processing.event.ResourceID; + +import com.github.benmanes.caffeine.jcache.spi.CaffeineCachingProvider; + +import static io.javaoperatorsdk.operator.processing.event.source.SampleExternalResource.*; +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.Mockito.*; + +class CachingEventSourceTest { + + private CachingEventSource cachingEventSource; + private Cache cache; + private EventHandler eventHandlerMock = mock(EventHandler.class); + + @BeforeEach + public void setup() { + CachingProvider cachingProvider = new CaffeineCachingProvider(); + CacheManager cacheManager = cachingProvider.getCacheManager(); + cache = cacheManager.createCache("test-caching", new MutableConfiguration<>()); + + cachingEventSource = new SimpleCachingEventSource(cache); + cachingEventSource.setEventHandler(eventHandlerMock); + cachingEventSource.start(); + } + + @AfterEach + public void tearDown() { + cachingEventSource.stop(); + } + + @Test + public void putsNewResourceIntoCacheAndProducesEvent() { + cachingEventSource.handleEvent(testResource1(), testResource1ID()); + + verify(eventHandlerMock, times(1)).handleEvent(eq(new Event(testResource1ID()))); + assertThat(cachingEventSource.getCachedValue(testResource1ID())).isPresent(); + } + + @Test + public void propagatesEventIfResourceChanged() { + var res2 = testResource1(); + res2.setValue("changedValue"); + cachingEventSource.handleEvent(testResource1(), testResource1ID()); + cachingEventSource.handleEvent(res2, testResource1ID()); + + + verify(eventHandlerMock, times(2)).handleEvent(eq(new Event(testResource1ID()))); + assertThat(cachingEventSource.getCachedValue(testResource1ID()).get()).isEqualTo(res2); + } + + @Test + public void noEventPropagatedIfTheResourceIsNotChanged() { + cachingEventSource.handleEvent(testResource1(), testResource1ID()); + cachingEventSource.handleEvent(testResource1(), testResource1ID()); + + verify(eventHandlerMock, times(1)).handleEvent(eq(new Event(testResource1ID()))); + assertThat(cachingEventSource.getCachedValue(testResource1ID())).isPresent(); + } + + @Test + public void propagatesEventOnDeleteIfThereIsPrevResourceInCache() { + cachingEventSource.handleEvent(testResource1(), testResource1ID()); + cachingEventSource.handleDelete(testResource1ID()); + + verify(eventHandlerMock, times(2)).handleEvent(eq(new Event(testResource1ID()))); + assertThat(cachingEventSource.getCachedValue(testResource1ID())).isNotPresent(); + } + + @Test + public void noEventOnDeleteIfResourceWasNotInCacheBefore() { + cachingEventSource.handleDelete(testResource1ID()); + + verify(eventHandlerMock, times(0)).handleEvent(eq(new Event(testResource1ID()))); + } + + + public static class SimpleCachingEventSource + extends CachingEventSource { + public SimpleCachingEventSource(Cache cache) { + super(cache); + } + } + +} diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/OnceWhitelistEventFilterEventFilterTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/OnceWhitelistEventFilterEventFilterTest.java index 4595f1b649..3d2582fd3e 100644 --- a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/OnceWhitelistEventFilterEventFilterTest.java +++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/OnceWhitelistEventFilterEventFilterTest.java @@ -4,6 +4,7 @@ import io.javaoperatorsdk.operator.TestUtils; import io.javaoperatorsdk.operator.processing.event.ResourceID; +import io.javaoperatorsdk.operator.processing.event.source.controller.OnceWhitelistEventFilterEventFilter; import static io.javaoperatorsdk.operator.TestUtils.testCustomResource; import static org.assertj.core.api.Assertions.assertThat; diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/ResourceEventFilterTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/ResourceEventFilterTest.java index 571d9433c6..be6d9f803b 100644 --- a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/ResourceEventFilterTest.java +++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/ResourceEventFilterTest.java @@ -17,6 +17,10 @@ import io.javaoperatorsdk.operator.api.config.DefaultControllerConfiguration; import io.javaoperatorsdk.operator.processing.Controller; import io.javaoperatorsdk.operator.processing.event.EventHandler; +import io.javaoperatorsdk.operator.processing.event.EventSourceManager; +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.controller.ResourceEventFilter; import io.javaoperatorsdk.operator.sample.observedgeneration.ObservedGenCustomResource; import io.javaoperatorsdk.operator.sample.simple.TestCustomResource; @@ -195,6 +199,11 @@ public TestController(ControllerConfiguration configuration) super(null, configuration, null); } + @Override + public EventSourceManager getEventSourceManager() { + return mock(EventSourceManager.class); + } + @Override public MixedOperation, Resource> getCRClient() { return mock(MixedOperation.class); @@ -209,6 +218,11 @@ public ObservedGenController( super(null, configuration, null); } + @Override + public EventSourceManager getEventSourceManager() { + return mock(EventSourceManager.class); + } + @Override public MixedOperation, Resource> getCRClient() { return mock(MixedOperation.class); diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/SampleExternalResource.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/SampleExternalResource.java new file mode 100644 index 0000000000..bf2ff42c96 --- /dev/null +++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/SampleExternalResource.java @@ -0,0 +1,71 @@ +package io.javaoperatorsdk.operator.processing.event.source; + +import java.io.Serializable; +import java.util.Objects; + +import io.javaoperatorsdk.operator.processing.event.ResourceID; + +public class SampleExternalResource implements Serializable { + + public static final String DEFAULT_VALUE_1 = "value1"; + public static final String DEFAULT_VALUE_2 = "value2"; + public static final String NAME_1 = "name1"; + public static final String NAME_2 = "name2"; + + public static SampleExternalResource testResource1() { + return new SampleExternalResource(NAME_1, DEFAULT_VALUE_1); + } + + public static SampleExternalResource testResource2() { + return new SampleExternalResource(NAME_2, DEFAULT_VALUE_2); + } + + public static ResourceID testResource1ID() { + return new ResourceID(NAME_1, "testns"); + } + + public static ResourceID testResource2ID() { + return new ResourceID(NAME_2, "testns"); + } + + private String name; + private String value; + + public SampleExternalResource(String name, String value) { + this.name = name; + this.value = value; + } + + public String getName() { + return name; + } + + public SampleExternalResource setName(String name) { + this.name = name; + return this; + } + + public String getValue() { + return value; + } + + public SampleExternalResource setValue(String value) { + this.value = value; + return this; + } + + @Override + public boolean equals(Object o) { + if (this == o) + return true; + if (o == null || getClass() != o.getClass()) + return false; + SampleExternalResource that = (SampleExternalResource) o; + return Objects.equals(name, that.name) && Objects.equals(value, that.value); + } + + @Override + public int hashCode() { + return Objects.hash(name, value); + } +} diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/ControllerResourceEventSourceTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/controller/ControllerResourceEventSourceTest.java similarity index 86% rename from operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/ControllerResourceEventSourceTest.java rename to operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/controller/ControllerResourceEventSourceTest.java index e63c415e20..5eb26db524 100644 --- a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/ControllerResourceEventSourceTest.java +++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/controller/ControllerResourceEventSourceTest.java @@ -1,4 +1,4 @@ -package io.javaoperatorsdk.operator.processing.event.source; +package io.javaoperatorsdk.operator.processing.event.source.controller; import java.time.LocalDateTime; import java.util.List; @@ -15,9 +15,11 @@ import io.javaoperatorsdk.operator.api.monitoring.Metrics; import io.javaoperatorsdk.operator.processing.Controller; import io.javaoperatorsdk.operator.processing.event.EventHandler; +import io.javaoperatorsdk.operator.processing.event.EventSourceManager; import io.javaoperatorsdk.operator.processing.event.ResourceID; import io.javaoperatorsdk.operator.sample.simple.TestCustomResource; +import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.any; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.times; @@ -31,8 +33,9 @@ class ControllerResourceEventSourceTest { mock(MixedOperation.class); EventHandler eventHandler = mock(EventHandler.class); + private TestController testController = new TestController(true); private ControllerResourceEventSource controllerResourceEventSource = - new ControllerResourceEventSource<>(new TestController(true)); + new ControllerResourceEventSource<>(testController); @BeforeEach public void setup() { @@ -136,12 +139,32 @@ public void notHandlesNextEventIfNotWhitelisted() { verify(eventHandler, times(0)).handleEvent(any()); } + @Test + public void callsBroadcastsOnResourceEvents() { + TestCustomResource customResource1 = TestUtils.testCustomResource(); + + controllerResourceEventSource.eventReceived(ResourceAction.UPDATED, customResource1, + customResource1); + + verify(testController.getEventSourceManager(), times(1)) + .broadcastOnResourceEvent(eq(ResourceAction.UPDATED), eq(customResource1), + eq(customResource1)); + } + private static class TestController extends Controller { + private EventSourceManager eventSourceManager = + mock(EventSourceManager.class); + public TestController(boolean generationAware) { super(null, new TestConfiguration(generationAware), null); } + @Override + public EventSourceManager getEventSourceManager() { + return eventSourceManager; + } + @Override public MixedOperation, Resource> getCRClient() { return client; diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/polling/PerResourcePollingEventSourceTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/polling/PerResourcePollingEventSourceTest.java new file mode 100644 index 0000000000..6eb4b98e3b --- /dev/null +++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/polling/PerResourcePollingEventSourceTest.java @@ -0,0 +1,114 @@ +package io.javaoperatorsdk.operator.processing.event.source.polling; + +import java.util.Optional; + +import javax.cache.Cache; +import javax.cache.CacheManager; +import javax.cache.configuration.MutableConfiguration; +import javax.cache.spi.CachingProvider; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import io.javaoperatorsdk.operator.TestUtils; +import io.javaoperatorsdk.operator.processing.event.EventHandler; +import io.javaoperatorsdk.operator.processing.event.ResourceID; +import io.javaoperatorsdk.operator.processing.event.source.SampleExternalResource; +import io.javaoperatorsdk.operator.processing.event.source.controller.ResourceCache; +import io.javaoperatorsdk.operator.sample.simple.TestCustomResource; + +import com.github.benmanes.caffeine.jcache.spi.CaffeineCachingProvider; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.junit.jupiter.api.Assertions.*; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.*; + +class PerResourcePollingEventSourceTest { + + public static final int PERIOD = 80; + private PerResourcePollingEventSource pollingEventSource; + private PerResourcePollingEventSource.ResourceSupplier supplier = + mock(PerResourcePollingEventSource.ResourceSupplier.class); + private ResourceCache resourceCache = mock(ResourceCache.class); + private Cache cache; + private EventHandler eventHandler = mock(EventHandler.class); + private TestCustomResource testCustomResource = TestUtils.testCustomResource(); + + @BeforeEach + public void setup() { + CachingProvider cachingProvider = new CaffeineCachingProvider(); + CacheManager cacheManager = cachingProvider.getCacheManager(); + cache = cacheManager.createCache("test-caching", new MutableConfiguration<>()); + + when(resourceCache.get(any())).thenReturn(Optional.of(testCustomResource)); + when(supplier.getResources(any())) + .thenReturn(Optional.of(SampleExternalResource.testResource1())); + + pollingEventSource = + new PerResourcePollingEventSource<>(supplier, resourceCache, PERIOD, cache); + pollingEventSource.setEventHandler(eventHandler); + } + + @Test + public void pollsTheResourceAfterAwareOfIt() throws InterruptedException { + pollingEventSource.start(); + pollingEventSource.onResourceCreated(testCustomResource); + + Thread.sleep(3 * PERIOD); + verify(supplier, atLeast(2)).getResources(eq(testCustomResource)); + verify(eventHandler, times(1)).handleEvent(any()); + } + + @Test + public void registeringTaskOnAPredicate() throws InterruptedException { + pollingEventSource = new PerResourcePollingEventSource<>(supplier, resourceCache, PERIOD, cache, + testCustomResource -> testCustomResource.getMetadata().getGeneration() > 1); + pollingEventSource.setEventHandler(eventHandler); + pollingEventSource.start(); + pollingEventSource.onResourceCreated(testCustomResource); + Thread.sleep(2 * PERIOD); + + verify(supplier, times(0)).getResources(eq(testCustomResource)); + testCustomResource.getMetadata().setGeneration(2L); + pollingEventSource.onResourceUpdated(testCustomResource, testCustomResource); + + Thread.sleep(2 * PERIOD); + + verify(supplier, atLeast(1)).getResources(eq(testCustomResource)); + } + + @Test + public void propagateEventOnDeletedResource() throws InterruptedException { + pollingEventSource.start(); + pollingEventSource.onResourceCreated(testCustomResource); + when(supplier.getResources(any())) + .thenReturn(Optional.of(SampleExternalResource.testResource1())) + .thenReturn(Optional.empty()); + + Thread.sleep(3 * PERIOD); + verify(supplier, atLeast(2)).getResources(eq(testCustomResource)); + verify(eventHandler, times(2)).handleEvent(any()); + } + + @Test + public void getsValueFromCacheOrSupplier() throws InterruptedException { + pollingEventSource.start(); + pollingEventSource.onResourceCreated(testCustomResource); + when(supplier.getResources(any())) + .thenReturn(Optional.empty()) + .thenReturn(Optional.of(SampleExternalResource.testResource1())); + + Thread.sleep(PERIOD / 2); + + var value = + pollingEventSource.getValueFromCacheOrSupplier(ResourceID.fromResource(testCustomResource)); + + Thread.sleep(PERIOD * 2); + + assertThat(value).isPresent(); + verify(eventHandler, never()).handleEvent(any()); + } + +} diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/polling/PollingEventSourceTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/polling/PollingEventSourceTest.java new file mode 100644 index 0000000000..177760f94e --- /dev/null +++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/polling/PollingEventSourceTest.java @@ -0,0 +1,91 @@ +package io.javaoperatorsdk.operator.processing.event.source.polling; + +import java.util.HashMap; +import java.util.Map; +import java.util.function.Supplier; + +import javax.cache.Cache; +import javax.cache.CacheManager; +import javax.cache.configuration.MutableConfiguration; +import javax.cache.spi.CachingProvider; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import io.javaoperatorsdk.operator.processing.event.EventHandler; +import io.javaoperatorsdk.operator.processing.event.ResourceID; +import io.javaoperatorsdk.operator.processing.event.source.SampleExternalResource; + +import com.github.benmanes.caffeine.jcache.spi.CaffeineCachingProvider; + +import static io.javaoperatorsdk.operator.processing.event.source.SampleExternalResource.*; +import static org.mockito.Mockito.*; + +class PollingEventSourceTest { + + private PollingEventSource pollingEventSource; + private Supplier> supplier = mock(Supplier.class); + private Cache cache; + private EventHandler eventHandler = mock(EventHandler.class); + + @BeforeEach + public void setup() { + CachingProvider cachingProvider = new CaffeineCachingProvider(); + CacheManager cacheManager = cachingProvider.getCacheManager(); + cache = cacheManager.createCache("test-caching", new MutableConfiguration<>()); + + pollingEventSource = new PollingEventSource<>(supplier, 50, cache); + pollingEventSource.setEventHandler(eventHandler); + } + + @AfterEach + public void teardown() { + pollingEventSource.stop(); + } + + @Test + public void pollsAndProcessesEvents() throws InterruptedException { + when(supplier.get()).thenReturn(testResponseWithTwoValues()); + pollingEventSource.start(); + + Thread.sleep(100); + + verify(eventHandler, times(2)).handleEvent(any()); + } + + @Test + public void propagatesEventForRemovedResources() throws InterruptedException { + when(supplier.get()).thenReturn(testResponseWithTwoValues()) + .thenReturn(testResponseWithOneValue()); + pollingEventSource.start(); + + Thread.sleep(150); + + verify(eventHandler, times(3)).handleEvent(any()); + } + + @Test + public void doesNotPropagateEventIfResourceNotChanged() throws InterruptedException { + when(supplier.get()).thenReturn(testResponseWithTwoValues()); + pollingEventSource.start(); + + Thread.sleep(250); + + verify(eventHandler, times(2)).handleEvent(any()); + } + + private Map testResponseWithOneValue() { + Map res = new HashMap<>(); + res.put(testResource1ID(), testResource1()); + return res; + } + + private Map testResponseWithTwoValues() { + Map res = new HashMap<>(); + res.put(testResource1ID(), testResource1()); + res.put(testResource2ID(), testResource2()); + return res; + } + +} diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/TimerEventSourceTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/timer/TimerEventSourceTest.java similarity index 96% rename from operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/TimerEventSourceTest.java rename to operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/timer/TimerEventSourceTest.java index d1768a8547..0f259f77cb 100644 --- a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/TimerEventSourceTest.java +++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/timer/TimerEventSourceTest.java @@ -1,4 +1,4 @@ -package io.javaoperatorsdk.operator.processing.event.source; +package io.javaoperatorsdk.operator.processing.event.source.timer; import java.io.IOException; import java.util.List; @@ -73,7 +73,7 @@ public void deRegistersOnceEventSources() { timerEventSource.scheduleOnce(customResource, PERIOD); timerEventSource - .cleanupForResource(ResourceID.fromResource(customResource)); + .onResourceDeleted(customResource); untilAsserted(() -> assertThat(eventHandlerMock.events).isEmpty()); } diff --git a/operator-framework/src/main/java/io/javaoperatorsdk/operator/config/runtime/AnnotationConfiguration.java b/operator-framework/src/main/java/io/javaoperatorsdk/operator/config/runtime/AnnotationConfiguration.java index 22d864158a..9a5577a90d 100644 --- a/operator-framework/src/main/java/io/javaoperatorsdk/operator/config/runtime/AnnotationConfiguration.java +++ b/operator-framework/src/main/java/io/javaoperatorsdk/operator/config/runtime/AnnotationConfiguration.java @@ -8,8 +8,8 @@ import io.javaoperatorsdk.operator.api.config.ConfigurationService; import io.javaoperatorsdk.operator.api.reconciler.ControllerConfiguration; import io.javaoperatorsdk.operator.api.reconciler.Reconciler; -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 class AnnotationConfiguration implements io.javaoperatorsdk.operator.api.config.ControllerConfiguration { diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/informereventsource/InformerEventSourceTestCustomReconciler.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/informereventsource/InformerEventSourceTestCustomReconciler.java index 71463e13d0..b6807358f1 100644 --- a/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/informereventsource/InformerEventSourceTestCustomReconciler.java +++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/informereventsource/InformerEventSourceTestCustomReconciler.java @@ -12,8 +12,8 @@ import io.javaoperatorsdk.operator.api.reconciler.UpdateControl; import io.javaoperatorsdk.operator.junit.KubernetesClientAware; import io.javaoperatorsdk.operator.processing.event.source.EventSourceRegistry; -import io.javaoperatorsdk.operator.processing.event.source.InformerEventSource; -import io.javaoperatorsdk.operator.processing.event.source.Mappers; +import io.javaoperatorsdk.operator.processing.event.source.informer.InformerEventSource; +import io.javaoperatorsdk.operator.processing.event.source.informer.Mappers; import static io.javaoperatorsdk.operator.api.reconciler.ControllerConfiguration.NO_FINALIZER; diff --git a/pom.xml b/pom.xml index fe9a567707..fe5a48e293 100644 --- a/pom.xml +++ b/pom.xml @@ -70,6 +70,8 @@ 2.17.1 1.0 1.6.2 + 1.1.1 + 3.0.4 @@ -168,6 +170,21 @@ operator-framework ${project.version} + + javax.cache + cache-api + ${jcache.version} + + + com.github.ben-manes.caffeine + caffeine + ${caffein.version} + + + com.github.ben-manes.caffeine + jcache + ${caffein.version} + diff --git a/sample-operators/mysql-schema/pom.xml b/sample-operators/mysql-schema/pom.xml index 8693cf6154..f5df325685 100644 --- a/sample-operators/mysql-schema/pom.xml +++ b/sample-operators/mysql-schema/pom.xml @@ -67,6 +67,19 @@ jackson-dataformat-yaml 2.13.0 + + javax.cache + cache-api + ${jcache.version} + + + com.github.ben-manes.caffeine + caffeine + + + com.github.ben-manes.caffeine + jcache + diff --git a/sample-operators/mysql-schema/src/main/java/io/javaoperatorsdk/operator/sample/MySQLSchemaReconciler.java b/sample-operators/mysql-schema/src/main/java/io/javaoperatorsdk/operator/sample/MySQLSchemaReconciler.java index 47d44246e3..33b8df52df 100644 --- a/sample-operators/mysql-schema/src/main/java/io/javaoperatorsdk/operator/sample/MySQLSchemaReconciler.java +++ b/sample-operators/mysql-schema/src/main/java/io/javaoperatorsdk/operator/sample/MySQLSchemaReconciler.java @@ -2,128 +2,102 @@ import java.sql.Connection; import java.sql.DriverManager; -import java.sql.PreparedStatement; -import java.sql.ResultSet; import java.sql.SQLException; -import java.sql.Statement; import java.util.Base64; +import java.util.Optional; + +import javax.cache.Cache; +import javax.cache.CacheManager; +import javax.cache.configuration.MutableConfiguration; +import javax.cache.spi.CachingProvider; import org.apache.commons.lang3.RandomStringUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import io.fabric8.kubernetes.api.model.OwnerReference; import io.fabric8.kubernetes.api.model.Secret; import io.fabric8.kubernetes.api.model.SecretBuilder; import io.fabric8.kubernetes.client.KubernetesClient; import io.javaoperatorsdk.operator.api.reconciler.*; +import io.javaoperatorsdk.operator.processing.event.ResourceID; +import io.javaoperatorsdk.operator.processing.event.source.EventSourceRegistry; +import io.javaoperatorsdk.operator.processing.event.source.polling.PerResourcePollingEventSource; +import io.javaoperatorsdk.operator.sample.schema.Schema; +import io.javaoperatorsdk.operator.sample.schema.SchemaService; + +import com.github.benmanes.caffeine.jcache.spi.CaffeineCachingProvider; import static java.lang.String.format; @ControllerConfiguration -public class MySQLSchemaReconciler implements Reconciler { - static final String USERNAME_FORMAT = "%s-user"; - static final String SECRET_FORMAT = "%s-secret"; - +public class MySQLSchemaReconciler + implements Reconciler, ErrorStatusHandler, + EventSourceInitializer { + public static final String SECRET_FORMAT = "%s-secret"; + public static final String USERNAME_FORMAT = "%s-user"; + public static final int POLL_PERIOD = 500; private final Logger log = LoggerFactory.getLogger(getClass()); private final KubernetesClient kubernetesClient; private final MySQLDbConfig mysqlDbConfig; + PerResourcePollingEventSource perResourcePollingEventSource; public MySQLSchemaReconciler(KubernetesClient kubernetesClient, MySQLDbConfig mysqlDbConfig) { this.kubernetesClient = kubernetesClient; this.mysqlDbConfig = mysqlDbConfig; } + @Override + public void prepareEventSources(EventSourceRegistry eventSourceRegistry) { + CachingProvider cachingProvider = new CaffeineCachingProvider(); + CacheManager cacheManager = cachingProvider.getCacheManager(); + Cache schemaCache = + cacheManager.createCache("schema-cache", new MutableConfiguration<>()); + + perResourcePollingEventSource = + new PerResourcePollingEventSource<>(new SchemaPollingResourceSupplier(mysqlDbConfig), + eventSourceRegistry.getControllerResourceEventSource().getResourceCache(), POLL_PERIOD, + schemaCache); + + eventSourceRegistry.registerEventSource(perResourcePollingEventSource); + } + @Override public UpdateControl reconcile(MySQLSchema schema, Context context) { + var dbSchema = perResourcePollingEventSource + .getValueFromCacheOrSupplier(ResourceID.fromResource(schema)); try (Connection connection = getConnection()) { - if (!schemaExists(connection, schema.getMetadata().getName())) { - try (Statement statement = connection.createStatement()) { - statement.execute( - format( - "CREATE SCHEMA `%1$s` DEFAULT CHARACTER SET %2$s", - schema.getMetadata().getName(), schema.getSpec().getEncoding())); - } - + if (!dbSchema.isPresent()) { + var schemaName = schema.getMetadata().getName(); String password = RandomStringUtils.randomAlphanumeric(16); - String userName = String.format(USERNAME_FORMAT, schema.getMetadata().getName()); - String secretName = String.format(SECRET_FORMAT, schema.getMetadata().getName()); - try (Statement statement = connection.createStatement()) { - statement.execute(format("CREATE USER '%1$s' IDENTIFIED BY '%2$s'", userName, password)); - } - try (Statement statement = connection.createStatement()) { - statement.execute( - format("GRANT ALL ON `%1$s`.* TO '%2$s'", schema.getMetadata().getName(), userName)); - } - Secret credentialsSecret = - new SecretBuilder() - .withNewMetadata() - .withName(secretName) - .endMetadata() - .addToData( - "MYSQL_USERNAME", Base64.getEncoder().encodeToString(userName.getBytes())) - .addToData( - "MYSQL_PASSWORD", Base64.getEncoder().encodeToString(password.getBytes())) - .build(); - this.kubernetesClient - .secrets() - .inNamespace(schema.getMetadata().getNamespace()) - .create(credentialsSecret); - - SchemaStatus status = new SchemaStatus(); - status.setUrl( - format( - "jdbc:mysql://%1$s/%2$s", - System.getenv("MYSQL_HOST"), schema.getMetadata().getName())); - status.setUserName(userName); - status.setSecretName(secretName); - status.setStatus("CREATED"); - schema.setStatus(status); - log.info("Schema {} created - updating CR status", schema.getMetadata().getName()); + String secretName = String.format(SECRET_FORMAT, schemaName); + String userName = String.format(USERNAME_FORMAT, schemaName); + SchemaService.createSchemaAndRelatedUser(connection, schemaName, + schema.getSpec().getEncoding(), userName, password); + createSecret(schema, password, secretName, userName); + updateStatusPojo(schema, secretName, userName); + log.info("Schema {} created - updating CR status", schema.getMetadata().getName()); return UpdateControl.updateStatus(schema); } return UpdateControl.noUpdate(); } catch (SQLException e) { log.error("Error while creating Schema", e); - - SchemaStatus status = new SchemaStatus(); - status.setUrl(null); - status.setUserName(null); - status.setSecretName(null); - status.setStatus("ERROR: " + e.getMessage()); - schema.setStatus(status); - - return UpdateControl.updateStatus(schema); + throw new IllegalStateException(e); } } @Override public DeleteControl cleanup(MySQLSchema schema, Context context) { log.info("Execution deleteResource for: {}", schema.getMetadata().getName()); - try (Connection connection = getConnection()) { - if (schemaExists(connection, schema.getMetadata().getName())) { - try (Statement statement = connection.createStatement()) { - statement.execute(format("DROP DATABASE `%1$s`", schema.getMetadata().getName())); - } - log.info("Deleted Schema '{}'", schema.getMetadata().getName()); - - if (schema.getStatus() != null) { - if (userExists(connection, schema.getStatus().getUserName())) { - try (Statement statement = connection.createStatement()) { - statement.execute(format("DROP USER '%1$s'", schema.getStatus().getUserName())); - } - log.info("Deleted User '{}'", schema.getStatus().getUserName()); - } - } - - this.kubernetesClient - .secrets() - .inNamespace(schema.getMetadata().getNamespace()) - .withName(schema.getStatus().getSecretName()) - .delete(); + var dbSchema = SchemaService.getSchema(connection, schema.getMetadata().getName()); + if (dbSchema.isPresent()) { + var userName = schema.getStatus() != null ? schema.getStatus().getUserName() : null; + SchemaService.deleteSchemaAndRelatedUser(connection, schema.getMetadata().getName(), + userName); } else { log.info( "Delete event ignored for schema '{}', real schema doesn't exist", @@ -136,33 +110,65 @@ public DeleteControl cleanup(MySQLSchema schema, Context context) { } } + @Override + public Optional updateErrorStatus(MySQLSchema schema, RetryInfo retryInfo, + RuntimeException e) { + SchemaStatus status = new SchemaStatus(); + status.setUrl(null); + status.setUserName(null); + status.setSecretName(null); + status.setStatus("ERROR: " + e.getMessage()); + schema.setStatus(status); + return Optional.empty(); + } + private Connection getConnection() throws SQLException { String connectionString = format("jdbc:mysql://%1$s:%2$s", mysqlDbConfig.getHost(), mysqlDbConfig.getPort()); - log.info("Connecting to '{}' with user '{}'", connectionString, mysqlDbConfig.getUser()); + log.debug("Connecting to '{}' with user '{}'", connectionString, mysqlDbConfig.getUser()); return DriverManager.getConnection(connectionString, mysqlDbConfig.getUser(), mysqlDbConfig.getPassword()); } - private boolean schemaExists(Connection connection, String schemaName) throws SQLException { - try (PreparedStatement ps = - connection.prepareStatement( - "SELECT schema_name FROM information_schema.schemata WHERE schema_name = ?")) { - ps.setString(1, schemaName); - try (ResultSet resultSet = ps.executeQuery()) { - return resultSet.next(); - } - } + private void updateStatusPojo(MySQLSchema schema, String secretName, String userName) { + SchemaStatus status = new SchemaStatus(); + status.setUrl( + format( + "jdbc:mysql://%1$s/%2$s", + System.getenv("MYSQL_HOST"), schema.getMetadata().getName())); + status.setUserName(userName); + status.setSecretName(secretName); + status.setStatus("CREATED"); + schema.setStatus(status); } - private boolean userExists(Connection connection, String userName) throws SQLException { - try (PreparedStatement ps = - connection.prepareStatement("SELECT User FROM mysql.user WHERE User = ?")) { - ps.setString(1, userName); - try (ResultSet resultSet = ps.executeQuery()) { - return resultSet.first(); - } + private void createSecret(MySQLSchema schema, String password, String secretName, + String userName) { + + var currentSecret = kubernetesClient.secrets().inNamespace(schema.getMetadata().getNamespace()) + .withName(secretName).get(); + if (currentSecret != null) { + return; } + Secret credentialsSecret = + new SecretBuilder() + .withNewMetadata() + .withName(secretName) + .withOwnerReferences(new OwnerReference("mysql.sample.javaoperatorsdk/v1", + false, false, "MySQLSchema", + schema.getMetadata().getName(), schema.getMetadata().getUid())) + .endMetadata() + .addToData( + "MYSQL_USERNAME", Base64.getEncoder().encodeToString(userName.getBytes())) + .addToData( + "MYSQL_PASSWORD", Base64.getEncoder().encodeToString(password.getBytes())) + .build(); + this.kubernetesClient + .secrets() + .inNamespace(schema.getMetadata().getNamespace()) + .create(credentialsSecret); } + + } diff --git a/sample-operators/mysql-schema/src/main/java/io/javaoperatorsdk/operator/sample/SchemaPollingResourceSupplier.java b/sample-operators/mysql-schema/src/main/java/io/javaoperatorsdk/operator/sample/SchemaPollingResourceSupplier.java new file mode 100644 index 0000000000..52f3f301c8 --- /dev/null +++ b/sample-operators/mysql-schema/src/main/java/io/javaoperatorsdk/operator/sample/SchemaPollingResourceSupplier.java @@ -0,0 +1,22 @@ +package io.javaoperatorsdk.operator.sample; + +import java.util.Optional; + +import io.javaoperatorsdk.operator.processing.event.source.polling.PerResourcePollingEventSource; +import io.javaoperatorsdk.operator.sample.schema.Schema; +import io.javaoperatorsdk.operator.sample.schema.SchemaService; + +public class SchemaPollingResourceSupplier + implements PerResourcePollingEventSource.ResourceSupplier { + + private final SchemaService schemaService; + + public SchemaPollingResourceSupplier(MySQLDbConfig mySQLDbConfig) { + this.schemaService = new SchemaService(mySQLDbConfig); + } + + @Override + public Optional getResources(MySQLSchema resource) { + return schemaService.getSchema(resource.getMetadata().getName()); + } +} diff --git a/sample-operators/mysql-schema/src/main/java/io/javaoperatorsdk/operator/sample/SchemaStatus.java b/sample-operators/mysql-schema/src/main/java/io/javaoperatorsdk/operator/sample/SchemaStatus.java index 168cd8db15..92ddb67a63 100644 --- a/sample-operators/mysql-schema/src/main/java/io/javaoperatorsdk/operator/sample/SchemaStatus.java +++ b/sample-operators/mysql-schema/src/main/java/io/javaoperatorsdk/operator/sample/SchemaStatus.java @@ -1,6 +1,8 @@ package io.javaoperatorsdk.operator.sample; -public class SchemaStatus { +import io.javaoperatorsdk.operator.api.ObservedGenerationAwareStatus; + +public class SchemaStatus extends ObservedGenerationAwareStatus { private String url; diff --git a/sample-operators/mysql-schema/src/main/java/io/javaoperatorsdk/operator/sample/schema/Schema.java b/sample-operators/mysql-schema/src/main/java/io/javaoperatorsdk/operator/sample/schema/Schema.java new file mode 100644 index 0000000000..836951a004 --- /dev/null +++ b/sample-operators/mysql-schema/src/main/java/io/javaoperatorsdk/operator/sample/schema/Schema.java @@ -0,0 +1,38 @@ +package io.javaoperatorsdk.operator.sample.schema; + +import java.io.Serializable; +import java.util.Objects; + +public class Schema implements Serializable { + + private String name; + private String characterSet; + + public Schema(String name, String characterSet) { + this.name = name; + this.characterSet = characterSet; + } + + public String getName() { + return name; + } + + public String getCharacterSet() { + return characterSet; + } + + @Override + public boolean equals(Object o) { + if (this == o) + return true; + if (o == null || getClass() != o.getClass()) + return false; + Schema schema = (Schema) o; + return Objects.equals(name, schema.name) && Objects.equals(characterSet, schema.characterSet); + } + + @Override + public int hashCode() { + return Objects.hash(name, characterSet); + } +} diff --git a/sample-operators/mysql-schema/src/main/java/io/javaoperatorsdk/operator/sample/schema/SchemaService.java b/sample-operators/mysql-schema/src/main/java/io/javaoperatorsdk/operator/sample/schema/SchemaService.java new file mode 100644 index 0000000000..8c6cd31b70 --- /dev/null +++ b/sample-operators/mysql-schema/src/main/java/io/javaoperatorsdk/operator/sample/schema/SchemaService.java @@ -0,0 +1,122 @@ +package io.javaoperatorsdk.operator.sample.schema; + +import java.sql.*; +import java.util.Optional; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import io.javaoperatorsdk.operator.sample.MySQLDbConfig; + +import static java.lang.String.format; + +public class SchemaService { + + + private static final Logger log = LoggerFactory.getLogger(SchemaService.class); + + private final MySQLDbConfig mySQLDbConfig; + + public SchemaService(MySQLDbConfig mySQLDbConfig) { + this.mySQLDbConfig = mySQLDbConfig; + } + + public Optional getSchema(String name) { + try (Connection connection = getConnection()) { + return getSchema(connection, name); + } catch (SQLException e) { + throw new IllegalStateException(e); + } + } + + public static void createSchemaAndRelatedUser(Connection connection, String schemaName, + String encoding, + String userName, + String password) { + try { + try (Statement statement = connection.createStatement()) { + statement.execute( + format( + "CREATE SCHEMA `%1$s` DEFAULT CHARACTER SET %2$s", + schemaName, encoding)); + } + if (!userExists(connection, userName)) { + try (Statement statement = connection.createStatement()) { + statement.execute(format("CREATE USER '%1$s' IDENTIFIED BY '%2$s'", userName, password)); + } + } + try (Statement statement = connection.createStatement()) { + statement.execute( + format("GRANT ALL ON `%1$s`.* TO '%2$s'", schemaName, userName)); + } + } catch (SQLException e) { + throw new IllegalStateException(e); + } + } + + public static void deleteSchemaAndRelatedUser(Connection connection, String schemaName, + String userName) { + try { + try (Statement statement = connection.createStatement()) { + statement.execute(format("DROP DATABASE `%1$s`", schemaName)); + } + log.info("Deleted Schema '{}'", schemaName); + if (userName != null) { + try (Statement statement = connection.createStatement()) { + statement.execute(format("DROP USER '%1$s'", userName)); + } + log.info("Deleted User '{}'", userName); + } + + } catch (SQLException e) { + throw new IllegalStateException(e); + } + } + + private static boolean userExists(Connection connection, String username) { + try (PreparedStatement ps = + connection.prepareStatement( + "SELECT EXISTS(SELECT 1 FROM mysql.user WHERE user = ?)")) { + ps.setString(1, username); + try (ResultSet resultSet = ps.executeQuery()) { + return resultSet.next(); + } + } catch (SQLException e) { + throw new IllegalStateException(e); + } + } + + public static Optional getSchema(Connection connection, String schemaName) { + try (PreparedStatement ps = + connection.prepareStatement( + "SELECT * FROM information_schema.schemata WHERE schema_name = ?")) { + ps.setString(1, schemaName); + try (ResultSet resultSet = ps.executeQuery()) { + // CATALOG_NAME, SCHEMA_NAME, DEFAULT_CHARACTER_SET_NAME, DEFAULT_COLLATION_NAME, SQL_PATH + var exists = resultSet.next(); + if (!exists) { + return Optional.empty(); + } else { + return Optional.of(new Schema(resultSet.getString("SCHEMA_NAME"), + resultSet.getString("DEFAULT_CHARACTER_SET_NAME"))); + } + } + } catch (SQLException e) { + throw new IllegalStateException(e); + } + } + + private Connection getConnection() { + try { + String connectionString = + format("jdbc:mysql://%1$s:%2$s", mySQLDbConfig.getHost(), mySQLDbConfig.getPort()); + + log.debug("Connecting to '{}' with user '{}'", connectionString, mySQLDbConfig.getUser()); + return DriverManager.getConnection(connectionString, mySQLDbConfig.getUser(), + mySQLDbConfig.getPassword()); + } catch (SQLException e) { + throw new IllegalStateException(e); + } + } + +} diff --git a/sample-operators/tomcat-operator/src/main/java/io/javaoperatorsdk/operator/sample/TomcatReconciler.java b/sample-operators/tomcat-operator/src/main/java/io/javaoperatorsdk/operator/sample/TomcatReconciler.java index ce3803534a..1d2c8d187b 100644 --- a/sample-operators/tomcat-operator/src/main/java/io/javaoperatorsdk/operator/sample/TomcatReconciler.java +++ b/sample-operators/tomcat-operator/src/main/java/io/javaoperatorsdk/operator/sample/TomcatReconciler.java @@ -18,7 +18,7 @@ import io.javaoperatorsdk.operator.api.reconciler.*; import io.javaoperatorsdk.operator.processing.event.ResourceID; import io.javaoperatorsdk.operator.processing.event.source.EventSourceRegistry; -import io.javaoperatorsdk.operator.processing.event.source.InformerEventSource; +import io.javaoperatorsdk.operator.processing.event.source.informer.InformerEventSource; import static io.javaoperatorsdk.operator.api.reconciler.ControllerConfiguration.NO_FINALIZER; import static java.util.Collections.EMPTY_SET; diff --git a/sample-operators/tomcat-operator/src/main/java/io/javaoperatorsdk/operator/sample/WebappReconciler.java b/sample-operators/tomcat-operator/src/main/java/io/javaoperatorsdk/operator/sample/WebappReconciler.java index 115ba9f648..2c71dd9815 100644 --- a/sample-operators/tomcat-operator/src/main/java/io/javaoperatorsdk/operator/sample/WebappReconciler.java +++ b/sample-operators/tomcat-operator/src/main/java/io/javaoperatorsdk/operator/sample/WebappReconciler.java @@ -26,7 +26,7 @@ import io.javaoperatorsdk.operator.api.reconciler.UpdateControl; import io.javaoperatorsdk.operator.processing.event.ResourceID; import io.javaoperatorsdk.operator.processing.event.source.EventSourceRegistry; -import io.javaoperatorsdk.operator.processing.event.source.InformerEventSource; +import io.javaoperatorsdk.operator.processing.event.source.informer.InformerEventSource; import okhttp3.Response;