diff --git a/docs/content/en/docs/documentation/eventing.md b/docs/content/en/docs/documentation/eventing.md index d2a104737b..e2c57d6217 100644 --- a/docs/content/en/docs/documentation/eventing.md +++ b/docs/content/en/docs/documentation/eventing.md @@ -346,4 +346,57 @@ for [primary resources](https://github.com/operator-framework/java-operator-sdk/ See also [CaffeineBoundedItemStores](https://github.com/operator-framework/java-operator-sdk/blob/main/caffeine-bounded-cache-support/src/main/java/io/javaoperatorsdk/operator/processing/event/source/cache/CaffeineBoundedItemStores.java) -for more details. \ No newline at end of file +for more details. + +### Sharing Informers Between Controllers (Informer Pool) + +{{% alert title="Experimental" color="warning" %}} +Informer pooling is marked `@Experimental`: the feature itself is production ready, but its +configuration API may still change in a non-backwards-compatible way. +{{% /alert %}} + +By default JOSDK maintains an *informer pool* so that informers are **shared** across controllers +and event sources. When several `InformerEventSource`s (whether belonging to different controllers, +or dynamically registered at runtime) watch the same resource type with an equivalent configuration, +they are all backed by a single underlying `SharedIndexInformer` instead of one informer each. This +reduces memory usage and the number of watch connections opened against the API server — which +matters in operators where many controllers watch the same secondary resource type (for example +`ConfigMap` or `Secret`). + +Two event sources share an informer when their effective informer configuration matches on all of: + +- the target cluster (API server URL, including the remote client used for + [multi-cluster](#informereventsource-multi-cluster-support) event sources), +- the resource type (or the group/version/kind for generic resources), +- the watched namespace, +- the label, field and shard selectors, +- the configured [item store](#bounded-caches-for-informers). + +The `informerListLimit` is intentionally *not* part of this identity: if two otherwise-equivalent +event sources request a different list limit, the existing informer is reused (a warning is logged +and the first-configured limit is kept). Indexers are also not part of the identity, since they can +be added to a shared informer independently; just make sure indexer names don't collide. + +The pool is reference-counted: the shared informer is created on first use and only stopped once the +last event source using it is de-registered (or its controller stops). Dynamically registering an +event source for a resource that is already backed by a running informer reuses that informer, and +the initial state already in its cache is replayed to the newly added handler. + +#### Selecting the pooling strategy + +The strategy is provided by the +[`InformerPool`](https://github.com/operator-framework/java-operator-sdk/blob/main/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/pool/InformerPool.java) +configured on the `ConfigurationService`. Two implementations are available: + +- [`DefaultInformerPool`](https://github.com/operator-framework/java-operator-sdk/blob/main/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/pool/DefaultInformerPool.java) + (the default): shares informers as described above. +- [`AlwaysNewInformerPool`](https://github.com/operator-framework/java-operator-sdk/blob/main/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/pool/AlwaysNewInformerPool.java): + never shares informers, creating a dedicated informer for every event source. Use this to opt out + of pooling and restore the pre-pooling behavior. + +You can override the strategy through the `ConfigurationService`: + +```java +Operator operator = new Operator(overrider -> + overrider.withInformerPool(new AlwaysNewInformerPool())); +``` \ No newline at end of file diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/AbstractConfigurationService.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/AbstractConfigurationService.java index a1b37d6fe9..9c8c52a4bc 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/AbstractConfigurationService.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/AbstractConfigurationService.java @@ -24,6 +24,8 @@ import io.fabric8.kubernetes.client.KubernetesClient; import io.javaoperatorsdk.operator.ReconcilerUtilsInternal; import io.javaoperatorsdk.operator.api.reconciler.Reconciler; +import io.javaoperatorsdk.operator.processing.event.source.informer.pool.DefaultInformerPool; +import io.javaoperatorsdk.operator.processing.event.source.informer.pool.InformerPool; /** * An abstract implementation of {@link ConfigurationService} meant to ease custom implementations @@ -35,6 +37,7 @@ public class AbstractConfigurationService implements ConfigurationService { private KubernetesClient client; private Cloner cloner; private ExecutorServiceManager executorServiceManager; + private InformerPool informerPool; protected AbstractConfigurationService(Version version) { this(version, null); @@ -190,4 +193,16 @@ public ExecutorServiceManager getExecutorServiceManager() { } return executorServiceManager; } + + @Override + public synchronized InformerPool informerPool() { + // cached so that all controllers backed by this ConfigurationService share the same pool and + // can therefore share the underlying informers; synchronized so concurrent first-access from + // multiple controllers cannot create (and share out) more than one pool instance + if (informerPool == null) { + informerPool = new DefaultInformerPool(); + informerPool.setConfigurationService(this); + } + return informerPool; + } } diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ConfigurationService.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ConfigurationService.java index 6ed9b7ff64..28a15ec3d6 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ConfigurationService.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ConfigurationService.java @@ -43,6 +43,7 @@ import io.javaoperatorsdk.operator.processing.dependent.kubernetes.KubernetesDependentResourceConfig; import io.javaoperatorsdk.operator.processing.dependent.workflow.ManagedWorkflowFactory; import io.javaoperatorsdk.operator.processing.event.source.controller.ControllerEventSource; +import io.javaoperatorsdk.operator.processing.event.source.informer.pool.InformerPool; /** An interface from which to retrieve configuration information. */ public interface ConfigurationService { @@ -476,4 +477,21 @@ default boolean useSSAToPatchPrimaryResource() { default boolean cloneSecondaryResourcesWhenGettingFromCache() { return false; } + + /** + * The {@link InformerPool} used to create and (when using the default, sharing pool) share the + * informers backing the event sources of all controllers managed by this {@code + * ConfigurationService}. + * + *

Implementations must return the same instance on every call. The pool is + * effectively a per-{@code ConfigurationService} singleton: controllers share informers only if + * they resolve the same pool, and reference counting / informer shutdown are only correct if + * {@code getInformer} and {@code releaseInformer} operate on that same instance. This is + * intentionally not a {@code default} method, since a {@code default} could not cache the result + * and would hand out a fresh (unshared) pool on each call; {@link AbstractConfigurationService} + * provides a cached implementation backed by the default sharing pool. + * + * @return the informer pool for this configuration service + */ + InformerPool informerPool(); } diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ConfigurationServiceOverrider.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ConfigurationServiceOverrider.java index cd9cdafb39..8d0835cd5e 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ConfigurationServiceOverrider.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ConfigurationServiceOverrider.java @@ -29,6 +29,7 @@ import io.javaoperatorsdk.operator.Operator; import io.javaoperatorsdk.operator.api.monitoring.Metrics; import io.javaoperatorsdk.operator.api.reconciler.dependent.DependentResourceFactory; +import io.javaoperatorsdk.operator.processing.event.source.informer.pool.InformerPool; @SuppressWarnings({"unused", "UnusedReturnValue"}) public class ConfigurationServiceOverrider { @@ -53,6 +54,7 @@ public class ConfigurationServiceOverrider { private Set> defaultNonSSAResource; private Boolean useSSAToPatchPrimaryResource; private Boolean cloneSecondaryResourcesWhenGettingFromCache; + private InformerPool informerPool; @SuppressWarnings("rawtypes") private DependentResourceFactory dependentResourceFactory; @@ -176,6 +178,15 @@ public ConfigurationServiceOverrider withCloneSecondaryResourcesWhenGettingFromC return this; } + /** + * Overrides the {@link InformerPool} strategy used to create/share the informers backing the + * event sources. When not set, the default (informer-sharing) pool is used. + */ + public ConfigurationServiceOverrider withInformerPool(InformerPool informerPool) { + this.informerPool = informerPool; + return this; + } + public ConfigurationService build() { return new BaseConfigurationService(original.getVersion(), cloner, client) { @Override @@ -309,6 +320,15 @@ public boolean cloneSecondaryResourcesWhenGettingFromCache() { cloneSecondaryResourcesWhenGettingFromCache, ConfigurationService::cloneSecondaryResourcesWhenGettingFromCache); } + + @Override + public InformerPool informerPool() { + if (informerPool == null) { + return super.informerPool(); + } + informerPool.setConfigurationService(this); + return informerPool; + } }; } } diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/informer/FieldSelector.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/informer/FieldSelector.java index 022bb59ef0..26102dc314 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/informer/FieldSelector.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/informer/FieldSelector.java @@ -17,6 +17,7 @@ import java.util.Arrays; import java.util.List; +import java.util.Objects; public class FieldSelector { private final List fields; @@ -38,4 +39,16 @@ public Field(String path, String value) { this(path, value, false); } } + + @Override + public boolean equals(Object o) { + if (o == null || getClass() != o.getClass()) return false; + FieldSelector that = (FieldSelector) o; + return Objects.equals(fields, that.fields); + } + + @Override + public int hashCode() { + return Objects.hashCode(fields); + } } diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/informer/InformerConfiguration.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/informer/InformerConfiguration.java index 6c92dcdcc1..a1d9f0a746 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/informer/InformerConfiguration.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/informer/InformerConfiguration.java @@ -30,6 +30,7 @@ import io.javaoperatorsdk.operator.api.config.ControllerConfiguration; import io.javaoperatorsdk.operator.api.config.Utils; import io.javaoperatorsdk.operator.api.reconciler.Constants; +import io.javaoperatorsdk.operator.processing.GroupVersionKind; import io.javaoperatorsdk.operator.processing.event.source.cache.BoundedItemStore; import io.javaoperatorsdk.operator.processing.event.source.filter.GenericFilter; import io.javaoperatorsdk.operator.processing.event.source.filter.OnAddFilter; @@ -42,6 +43,7 @@ public class InformerConfiguration { private final Builder builder = new Builder(); private final Class resourceClass; + private final GroupVersionKind resourceGroupVersionKind; private final String resourceTypeName; private String name; private Set namespaces; @@ -59,6 +61,7 @@ public class InformerConfiguration { protected InformerConfiguration( Class resourceClass, + GroupVersionKind resourceGroupVersionKind, String name, Set namespaces, boolean followControllerNamespaceChanges, @@ -74,7 +77,7 @@ protected InformerConfiguration( Boolean comparableResourceVersions, // TODO for removal in major release Duration ghostResourceCacheCheckInterval) { - this(resourceClass); + this(resourceClass, resourceGroupVersionKind); this.name = name; this.namespaces = namespaces; this.followControllerNamespaceChanges = followControllerNamespaceChanges; @@ -90,8 +93,9 @@ protected InformerConfiguration( this.comparableResourceVersions = comparableResourceVersions; } - private InformerConfiguration(Class resourceClass) { + private InformerConfiguration(Class resourceClass, GroupVersionKind resourceGroupVersionKind) { this.resourceClass = resourceClass; + this.resourceGroupVersionKind = resourceGroupVersionKind; this.resourceTypeName = resourceClass.isAssignableFrom(GenericKubernetesResource.class) // in general this is irrelevant now for secondary resources it is used just by @@ -101,10 +105,16 @@ private InformerConfiguration(Class resourceClass) { : ReconcilerUtilsInternal.getResourceTypeName(resourceClass); } + @SuppressWarnings({"rawtypes", "unchecked"}) + public static InformerConfiguration.Builder builder( + Class resourceClass, GroupVersionKind groupVersionKind) { + return new InformerConfiguration(resourceClass, groupVersionKind).builder; + } + @SuppressWarnings({"rawtypes", "unchecked"}) public static InformerConfiguration.Builder builder( Class resourceClass) { - return new InformerConfiguration(resourceClass).builder; + return new InformerConfiguration(resourceClass, null).builder; } @SuppressWarnings({"rawtypes", "unchecked"}) @@ -112,6 +122,7 @@ public static InformerConfiguration.Builder builder( InformerConfiguration original) { return new InformerConfiguration( original.resourceClass, + original.resourceGroupVersionKind, original.name, original.namespaces, original.followControllerNamespaceChanges, @@ -305,6 +316,10 @@ public Long getInformerListLimit() { return informerListLimit; } + public GroupVersionKind getResourceGroupVersionKind() { + return resourceGroupVersionKind; + } + public FieldSelector getFieldSelector() { return fieldSelector; } diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/informer/InformerEventSourceConfiguration.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/informer/InformerEventSourceConfiguration.java index ab1ad2b8eb..b6f7939728 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/informer/InformerEventSourceConfiguration.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/informer/InformerEventSourceConfiguration.java @@ -76,6 +76,7 @@ default boolean followControllerNamespaceChanges() {

PrimaryToSecondaryMapper

getPrimaryToSecondaryMapper(); + // todo deprecate Optional getGroupVersionKind(); default String name() { @@ -167,7 +168,7 @@ private Builder( this.resourceClass = resourceClass; this.groupVersionKind = groupVersionKind; this.primaryResourceClass = primaryResourceClass; - this.config = InformerConfiguration.builder(resourceClass); + this.config = InformerConfiguration.builder(resourceClass, groupVersionKind); } public Builder withName(String name) { diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/GroupVersionKind.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/GroupVersionKind.java index be3869a64f..7d182cf1e6 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/GroupVersionKind.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/GroupVersionKind.java @@ -136,4 +136,8 @@ public int hashCode() { public String toString() { return toGVKString(); } + + public String getApiVersion() { + return apiVersion; + } } 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 441d3cf178..b03d69369e 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 @@ -148,7 +148,7 @@ private Void stopEventSource(EventSource eventSource) { return null; } - @SuppressWarnings("rawtypes") + @SuppressWarnings({"rawtypes", "unchecked"}) public final synchronized void registerEventSource(EventSource eventSource) throws OperatorException { Objects.requireNonNull(eventSource, "EventSource must not be null"); @@ -249,8 +249,6 @@ public EventSource dynamicallyRegisterEventSource(EventSource ev registerEventSource(eventSource); } } - // The start itself is blocking thus blocking only the threads which are attempt to start the - // actual event source. Think of this as a form of lock striping. eventSource.start(); return eventSource; } diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/controller/ControllerEventSource.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/controller/ControllerEventSource.java index 2f624d1150..13d199bb59 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/controller/ControllerEventSource.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/controller/ControllerEventSource.java @@ -48,7 +48,7 @@ public class ControllerEventSource @SuppressWarnings({"unchecked", "rawtypes"}) public ControllerEventSource(Controller controller) { - super(NAME, controller.getCRClient(), controller.getConfiguration()); + super(NAME, controller.getConfiguration()); this.controller = controller; final var config = controller.getConfiguration(); diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerEventSource.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerEventSource.java index b03a22e894..cb0fdaa8dd 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerEventSource.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerEventSource.java @@ -24,8 +24,6 @@ import org.slf4j.LoggerFactory; import io.fabric8.kubernetes.api.model.HasMetadata; -import io.fabric8.kubernetes.client.KubernetesClient; -import io.fabric8.kubernetes.client.dsl.MixedOperation; import io.fabric8.kubernetes.client.informers.ResourceEventHandler; import io.javaoperatorsdk.operator.api.config.informer.InformerEventSourceConfiguration; import io.javaoperatorsdk.operator.api.reconciler.EventSourceContext; @@ -54,20 +52,18 @@ public class InformerEventSource private final PrimaryToSecondaryIndex primaryToSecondaryIndex; private final PrimaryToSecondaryMapper

primaryToSecondaryMapper; + /** + * @deprecated use {@link InformerEventSource(InformerEventSourceConfiguration)} + */ + // todo migrate sample, separate PR? + @Deprecated(forRemoval = true) public InformerEventSource( InformerEventSourceConfiguration configuration, EventSourceContext

context) { - this(configuration, configuration.getKubernetesClient().orElse(context.getClient())); + this(configuration); } - @SuppressWarnings({"unchecked", "rawtypes"}) - InformerEventSource(InformerEventSourceConfiguration configuration, KubernetesClient client) { - super( - configuration.name(), - configuration - .getGroupVersionKind() - .map(gvk -> client.genericKubernetesResources(gvk.apiVersion(), gvk.getKind())) - .orElseGet(() -> (MixedOperation) client.resources(configuration.getResourceClass())), - configuration); + public InformerEventSource(InformerEventSourceConfiguration configuration) { + super(configuration.name(), configuration); // If there is a primary to secondary mapper there is no need for primary to secondary index. primaryToSecondaryMapper = configuration.getPrimaryToSecondaryMapper(); if (useSecondaryToPrimaryIndex()) { diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerManager.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerManager.java index 8e7054b231..a18af8eca9 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerManager.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerManager.java @@ -26,50 +26,45 @@ import org.slf4j.LoggerFactory; import io.fabric8.kubernetes.api.model.HasMetadata; -import io.fabric8.kubernetes.api.model.KubernetesResourceList; -import io.fabric8.kubernetes.client.dsl.FilterWatchListDeletable; -import io.fabric8.kubernetes.client.dsl.MixedOperation; -import io.fabric8.kubernetes.client.dsl.Resource; +import io.fabric8.kubernetes.client.KubernetesClient; import io.fabric8.kubernetes.client.informers.ResourceEventHandler; import io.javaoperatorsdk.operator.OperatorException; import io.javaoperatorsdk.operator.ReconcilerUtilsInternal; import io.javaoperatorsdk.operator.api.config.ControllerConfiguration; import io.javaoperatorsdk.operator.api.config.Informable; import io.javaoperatorsdk.operator.api.config.informer.InformerConfiguration; +import io.javaoperatorsdk.operator.api.config.informer.InformerEventSourceConfiguration; import io.javaoperatorsdk.operator.health.InformerHealthIndicator; -import io.javaoperatorsdk.operator.processing.LifecycleAware; import io.javaoperatorsdk.operator.processing.event.ResourceID; import io.javaoperatorsdk.operator.processing.event.source.Cache; import io.javaoperatorsdk.operator.processing.event.source.IndexerResourceCache; +import io.javaoperatorsdk.operator.processing.event.source.informer.pool.InformerClassifier; +import io.javaoperatorsdk.operator.processing.event.source.informer.pool.InformerPool; import static io.javaoperatorsdk.operator.api.reconciler.Constants.WATCH_ALL_NAMESPACES; class InformerManager> - implements LifecycleAware, IndexerResourceCache { + implements IndexerResourceCache { private static final Logger log = LoggerFactory.getLogger(InformerManager.class); private final Map> sources = new ConcurrentHashMap<>(); private final C configuration; - private final MixedOperation, Resource> client; private final ResourceEventHandler eventHandler; private final Map>> indexers = new HashMap<>(); private ControllerConfiguration controllerConfiguration; + private InformerPool informerPool; - InformerManager( - MixedOperation, Resource> client, - C configuration, - ResourceEventHandler eventHandler) { - this.client = client; + InformerManager(C configuration, ResourceEventHandler eventHandler) { this.configuration = configuration; this.eventHandler = eventHandler; } void setControllerConfiguration(ControllerConfiguration controllerConfiguration) { this.controllerConfiguration = controllerConfiguration; + this.informerPool = controllerConfiguration.getConfigurationService().informerPool(); } - @Override public void start() throws OperatorException { initSources(); // make sure informers are all started before proceeding further @@ -78,8 +73,8 @@ public void start() throws OperatorException { .getExecutorServiceManager() .boundedExecuteAndWaitForAllToComplete( sources.values().stream(), - iw -> { - iw.start(); + wrapper -> { + start(wrapper); return null; }, iw -> @@ -96,12 +91,12 @@ private void initSources() { final var targetNamespaces = configuration.getInformerConfig().getEffectiveNamespaces(controllerConfiguration); if (InformerConfiguration.allNamespacesWatched(targetNamespaces)) { - var source = createEventSourceForNamespace(WATCH_ALL_NAMESPACES); + var source = getEventSourceForNamespace(WATCH_ALL_NAMESPACES); log.debug("Registered {} -> {} for any namespace", this, source); } else { targetNamespaces.forEach( ns -> { - final var source = createEventSourceForNamespace(ns); + final var source = getEventSourceForNamespace(ns); log.debug("Registered {} -> {} for namespace: {}", this, source, ns); }); } @@ -109,12 +104,22 @@ private void initSources() { public void changeNamespaces(Set namespaces) { var sourcesToRemove = - sources.keySet().stream().filter(k -> !namespaces.contains(k)).collect(Collectors.toSet()); + sources.entrySet().stream() + .filter(e -> !namespaces.contains(e.getKey())) + .collect(Collectors.toSet()); log.debug("Stopped informer {} for namespaces: {}", this, sourcesToRemove); - sourcesToRemove.forEach(k -> sources.remove(k).stop()); + sourcesToRemove.forEach( + e -> { + var informer = + informerPool.releaseInformer( + controllerConfiguration.getName(), + configuration.getInformerConfig().getName(), + e.getValue().getClassifier()); + sources.remove(e.getKey()); + informer.ifPresent(i -> i.removeEventHandler(eventHandler)); + }); - var newNamespaces = - namespaces.stream().filter(ns -> !sources.containsKey(ns)).collect(Collectors.toList()); + var newNamespaces = namespaces.stream().filter(ns -> !sources.containsKey(ns)).toList(); if (newNamespaces.isEmpty()) { return; } @@ -125,76 +130,76 @@ public void changeNamespaces(Set namespaces) { .boundedExecuteAndWaitForAllToComplete( newNamespaces.stream(), ns -> { - final var source = createEventSourceForNamespace(ns); - source.start(); + final var source = getEventSourceForNamespace(ns); + // block until the informer's cache is synced (or the sync timeout elapses) + start(source); log.debug("Registered new {} -> {} for namespace: {}", this, source, ns); return null; }, ns -> "InformerStarter-" + ns + "-" + configuration.getResourceClass().getSimpleName()); } - private InformerWrapper createEventSourceForNamespace(String namespace) { + private void start(InformerWrapper informerWrapper) { + informerPool.start(informerWrapper.getInformer(), informerWrapper.getClassifier()); + } + + private InformerWrapper getEventSourceForNamespace(String namespaceIdentifier) { final InformerWrapper source; - final var labelSelector = configuration.getInformerConfig().getLabelSelector(); - final var shardSelector = configuration.getInformerConfig().getShardSelector(); - if (namespace.equals(WATCH_ALL_NAMESPACES)) { - final var filteredBySelectorClient = - client.inAnyNamespace().withLabelSelector(labelSelector).withShardSelector(shardSelector); - source = createEventSource(filteredBySelectorClient, eventHandler, WATCH_ALL_NAMESPACES); - } else { - source = - createEventSource( - client - .inNamespace(namespace) - .withLabelSelector(labelSelector) - .withShardSelector(shardSelector), - eventHandler, - namespace); - } + InformerClassifier classifier = getClassifier(namespaceIdentifier); + var informer = + informerPool.getInformer( + controllerConfiguration.getName(), + configuration.getInformerConfig().getName(), + classifier, + getTargetClient()); + source = new InformerWrapper<>(informer, namespaceIdentifier, classifier); + source.addEventHandler(eventHandler); + sources.put(namespaceIdentifier, source); source.addIndexers(indexers); return source; } - private InformerWrapper createEventSource( - FilterWatchListDeletable, Resource> filteredBySelectorClient, - ResourceEventHandler eventHandler, - String namespaceIdentifier) { - final var informerConfig = configuration.getInformerConfig(); + private InformerClassifier getClassifier(String namespaceIdentifier) { + KubernetesClient targetClient = getTargetClient(); + + return new InformerClassifier<>( + targetClient.getConfiguration().getMasterUrl(), + configuration.getInformerConfig().getLabelSelector(), + configuration.getInformerConfig().getShardSelector(), + namespaceIdentifier, + configuration.getResourceClass(), + configuration.getInformerConfig().getResourceGroupVersionKind(), + configuration.getInformerConfig().getFieldSelector(), + configuration.getInformerConfig().getInformerListLimit(), + configuration.getInformerConfig().getItemStore()); + } - if (informerConfig.getFieldSelector() != null - && !informerConfig.getFieldSelector().getFields().isEmpty()) { - for (var f : informerConfig.getFieldSelector().getFields()) { - if (f.negated()) { - filteredBySelectorClient = filteredBySelectorClient.withoutField(f.path(), f.value()); - } else { - filteredBySelectorClient = filteredBySelectorClient.withField(f.path(), f.value()); - } + private KubernetesClient getTargetClient() { + KubernetesClient targetClient = + controllerConfiguration.getConfigurationService().getKubernetesClient(); + if (configuration instanceof InformerEventSourceConfiguration iesc) { + var remoteClient = iesc.getKubernetesClient().orElse(null); + if (remoteClient != null) { + targetClient = remoteClient; } } - - var informer = - Optional.ofNullable(informerConfig.getInformerListLimit()) - .map(filteredBySelectorClient::withLimit) - .orElse(filteredBySelectorClient) - .runnableInformer(0); - Optional.ofNullable(informerConfig.getItemStore()).ifPresent(informer::itemStore); - var source = - new InformerWrapper<>( - informer, controllerConfiguration.getConfigurationService(), namespaceIdentifier); - source.addEventHandler(eventHandler); - sources.put(namespaceIdentifier, source); - return source; + return targetClient; } - @Override public void stop() { sources.forEach( - (ns, source) -> { + (ns, wrapper) -> { try { - log.debug("Stopping informer for namespace: {} -> {}", ns, source); - source.stop(); + var informer = + informerPool.releaseInformer( + controllerConfiguration.getName(), + configuration.getInformerConfig().getName(), + wrapper.getClassifier()); + informer.ifPresent(i -> i.removeEventHandler(eventHandler)); + + log.debug("Stopping informer for namespace: {} -> {}", ns, wrapper); } catch (Exception e) { - log.warn("Error stopping informer for namespace: {} -> {}", ns, source, e); + log.warn("Error stopping informer for namespace: {} -> {}", ns, wrapper, e); } }); sources.clear(); diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerWrapper.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerWrapper.java index 541068aa93..3183576e4a 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerWrapper.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerWrapper.java @@ -18,9 +18,6 @@ import java.util.List; import java.util.Map; import java.util.Optional; -import java.util.concurrent.ExecutionException; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.TimeoutException; import java.util.function.Function; import java.util.function.Predicate; import java.util.stream.Stream; @@ -30,125 +27,34 @@ import io.fabric8.kubernetes.api.model.GenericKubernetesResource; import io.fabric8.kubernetes.api.model.HasMetadata; -import io.fabric8.kubernetes.client.informers.ExceptionHandler; import io.fabric8.kubernetes.client.informers.ResourceEventHandler; import io.fabric8.kubernetes.client.informers.SharedIndexInformer; import io.fabric8.kubernetes.client.informers.cache.Cache; -import io.javaoperatorsdk.operator.OperatorException; import io.javaoperatorsdk.operator.ReconcilerUtilsInternal; -import io.javaoperatorsdk.operator.api.config.ConfigurationService; import io.javaoperatorsdk.operator.health.InformerHealthIndicator; import io.javaoperatorsdk.operator.health.Status; -import io.javaoperatorsdk.operator.processing.LifecycleAware; import io.javaoperatorsdk.operator.processing.event.ResourceID; import io.javaoperatorsdk.operator.processing.event.source.IndexerResourceCache; +import io.javaoperatorsdk.operator.processing.event.source.informer.pool.InformerClassifier; class InformerWrapper - implements LifecycleAware, IndexerResourceCache, InformerHealthIndicator { + implements IndexerResourceCache, InformerHealthIndicator { private static final Logger log = LoggerFactory.getLogger(InformerWrapper.class); private final SharedIndexInformer informer; private final Cache cache; private final String namespaceIdentifier; - private final ConfigurationService configurationService; + private final InformerClassifier informerClassifier; public InformerWrapper( SharedIndexInformer informer, - ConfigurationService configurationService, - String namespaceIdentifier) { + String namespaceIdentifier, + InformerClassifier classifier) { this.informer = informer; this.namespaceIdentifier = namespaceIdentifier; this.cache = (Cache) informer.getStore(); - this.configurationService = configurationService; - } - - @Override - public void start() throws OperatorException { - try { - - // register stopped handler if we have one defined - configurationService - .getInformerStoppedHandler() - .ifPresent( - ish -> { - final var stopped = informer.stopped(); - if (stopped != null) { - stopped.handle( - (res, ex) -> { - ish.onStop(informer, ex); - return null; - }); - } else { - final var apiTypeClass = informer.getApiTypeClass(); - final var fullResourceName = HasMetadata.getFullResourceName(apiTypeClass); - final var version = HasMetadata.getVersion(apiTypeClass); - throw new IllegalStateException( - "Cannot retrieve 'stopped' callback to listen to informer stopping for" - + " informer for " - + fullResourceName - + "/" - + version); - } - }); - if (!configurationService.stopOnInformerErrorDuringStartup()) { - informer.exceptionHandler((b, t) -> !ExceptionHandler.isDeserializationException(t)); - } - // change thread name for easier debugging - final var thread = Thread.currentThread(); - final var name = thread.getName(); - try { - thread.setName(informerInfo() + " " + thread.getId()); - final var resourceName = informer.getApiTypeClass().getSimpleName(); - log.debug( - "Starting informer for namespace: {} resource: {}", namespaceIdentifier, resourceName); - var start = informer.start(); - // note that in case we don't put here timeout and stopOnInformerErrorDuringStartup is - // false, and there is a rbac issue the get never returns; therefore operator never really - // starts - log.trace( - "Waiting informer to start namespace: {} resource: {}", - namespaceIdentifier, - resourceName); - start - .toCompletableFuture() - .get(configurationService.cacheSyncTimeout().toMillis(), TimeUnit.MILLISECONDS); - log.debug( - "Started informer for namespace: {} resource: {}", namespaceIdentifier, resourceName); - } catch (TimeoutException | ExecutionException e) { - if (configurationService.stopOnInformerErrorDuringStartup()) { - log.error("Informer startup error. Operator will be stopped. Informer: {}", informer, e); - throw new OperatorException(e); - } else { - log.warn("Informer startup error. Will periodically retry. Informer: {}", informer, e); - } - } catch (InterruptedException e) { - thread.interrupt(); - throw new IllegalStateException(e); - } finally { - // restore original name - thread.setName(name); - } - - } catch (Exception e) { - ReconcilerUtilsInternal.handleKubernetesClientException( - e, HasMetadata.getFullResourceName(informer.getApiTypeClass())); - throw new OperatorException( - "Couldn't start informer for " + versionedFullResourceName() + " resources", e); - } - } - - private String versionedFullResourceName() { - final var apiTypeClass = informer.getApiTypeClass(); - if (apiTypeClass.isAssignableFrom(GenericKubernetesResource.class)) { - return GenericKubernetesResource.class.getSimpleName(); - } - return ReconcilerUtilsInternal.getResourceTypeNameWithVersion(apiTypeClass); - } - - @Override - public void stop() throws OperatorException { - informer.stop(); + this.informerClassifier = classifier; } @Override @@ -201,7 +107,15 @@ public String toString() { } private String informerInfo() { - return "InformerWrapper [" + versionedFullResourceName() + "]"; + return "InformerWrapper [ " + versionedFullResourceName() + " ]"; + } + + private String versionedFullResourceName() { + final var apiTypeClass = informer.getApiTypeClass(); + if (apiTypeClass.isAssignableFrom(GenericKubernetesResource.class)) { + return GenericKubernetesResource.class.getSimpleName(); + } + return ReconcilerUtilsInternal.getResourceTypeNameWithVersion(apiTypeClass); } @Override @@ -237,4 +151,12 @@ public Status getStatus() { public String getTargetNamespace() { return namespaceIdentifier; } + + public InformerClassifier getClassifier() { + return informerClassifier; + } + + public SharedIndexInformer getInformer() { + return informer; + } } diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/ManagedInformerEventSource.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/ManagedInformerEventSource.java index 8352bef665..ec08f93d93 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/ManagedInformerEventSource.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/ManagedInformerEventSource.java @@ -32,7 +32,6 @@ import org.slf4j.LoggerFactory; import io.fabric8.kubernetes.api.model.HasMetadata; -import io.fabric8.kubernetes.client.dsl.MixedOperation; import io.fabric8.kubernetes.client.informers.ResourceEventHandler; import io.javaoperatorsdk.operator.OperatorException; import io.javaoperatorsdk.operator.ReconcilerUtilsInternal; @@ -51,7 +50,6 @@ import static io.javaoperatorsdk.operator.api.reconciler.Experimental.API_MIGHT_CHANGE; -@SuppressWarnings("rawtypes") public abstract class ManagedInformerEventSource< R extends HasMetadata, P extends HasMetadata, C extends Informable> extends AbstractEventSource @@ -70,13 +68,11 @@ public abstract class ManagedInformerEventSource< private final C configuration; private final Map>> indexers = new HashMap<>(); protected TemporaryResourceCache temporaryResourceCache; - protected MixedOperation client; - protected ManagedInformerEventSource(String name, MixedOperation client, C configuration) { + protected ManagedInformerEventSource(String name, C configuration) { super(configuration.getResourceClass(), name); this.comparableResourceVersions = configuration.getInformerConfig().isComparableResourceVersions(); - this.client = client; this.configuration = configuration; } @@ -159,16 +155,20 @@ protected abstract void handleEvent( Boolean deletedFinalStateUnknown, Set relatedPrimaryIDs); - @SuppressWarnings("unchecked") @Override public synchronized void start() { if (isRunning()) { return; } temporaryResourceCache = new TemporaryResourceCache<>(comparableResourceVersions, this); - this.cache = new InformerManager<>(client, configuration, this); + this.cache = new InformerManager<>(configuration, this); cache.setControllerConfiguration(controllerConfiguration); cache.addIndexers(indexers); + // A dynamically registered event source may join an already-running shared informer whose cache + // is already populated. Those pre-existing resources are still delivered to this newly added + // handler: the underlying Fabric8 informer replays the current cache contents to every handler + // at registration time (see SharedProcessor#addProcessorListener). Replaying them here as well + // would deliver every pre-existing resource twice. manager().start(); super.start(); } diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/pool/AbstractInformerPool.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/pool/AbstractInformerPool.java new file mode 100644 index 0000000000..bc656709c2 --- /dev/null +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/pool/AbstractInformerPool.java @@ -0,0 +1,183 @@ +/* + * Copyright Java Operator SDK Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.javaoperatorsdk.operator.processing.event.source.informer.pool; + +import java.util.Optional; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import io.fabric8.kubernetes.api.model.GenericKubernetesResource; +import io.fabric8.kubernetes.api.model.HasMetadata; +import io.fabric8.kubernetes.client.KubernetesClient; +import io.fabric8.kubernetes.client.dsl.FilterWatchListDeletable; +import io.fabric8.kubernetes.client.dsl.MixedOperation; +import io.fabric8.kubernetes.client.informers.ExceptionHandler; +import io.fabric8.kubernetes.client.informers.SharedIndexInformer; +import io.javaoperatorsdk.operator.OperatorException; +import io.javaoperatorsdk.operator.ReconcilerUtilsInternal; +import io.javaoperatorsdk.operator.api.config.ConfigurationService; + +import static io.javaoperatorsdk.operator.api.reconciler.Constants.WATCH_ALL_NAMESPACES; + +public abstract class AbstractInformerPool implements InformerPool { + + private static final Logger log = LoggerFactory.getLogger(AbstractInformerPool.class); + + protected ConfigurationService configurationService; + + public ConfigurationService getConfigurationService() { + return configurationService; + } + + @Override + public void setConfigurationService(ConfigurationService configurationService) { + this.configurationService = configurationService; + } + + @SuppressWarnings({"rawtypes", "unchecked"}) + protected SharedIndexInformer createInformer( + InformerClassifier classifier, KubernetesClient client) { + + MixedOperation clientWithResource; + if (classifier.groupVersionKind() != null) { + clientWithResource = + client.genericKubernetesResources( + classifier.groupVersionKind().getApiVersion(), + classifier.groupVersionKind().getKind()); + } else { + clientWithResource = client.resources(classifier.resourceClass()); + } + + FilterWatchListDeletable filteredClient; + if (WATCH_ALL_NAMESPACES.equals(classifier.namespaceIdentifier())) { + filteredClient = clientWithResource.inAnyNamespace(); + } else { + filteredClient = clientWithResource.inNamespace(classifier.namespaceIdentifier()); + } + filteredClient = + (FilterWatchListDeletable) filteredClient.withLabelSelector(classifier.labelSelector()); + filteredClient = + (FilterWatchListDeletable) filteredClient.withShardSelector(classifier.shardSelector()); + + if (classifier.fieldSelector() != null && !classifier.fieldSelector().getFields().isEmpty()) { + for (var f : classifier.fieldSelector().getFields()) { + if (f.negated()) { + filteredClient = + (FilterWatchListDeletable) filteredClient.withoutField(f.path(), f.value()); + } else { + filteredClient = (FilterWatchListDeletable) filteredClient.withField(f.path(), f.value()); + } + } + } + + if (classifier.informerListLimit() != null) { + filteredClient = + (FilterWatchListDeletable) filteredClient.withLimit(classifier.informerListLimit()); + } + + var informer = filteredClient.runnableInformer(0); + + Optional.ofNullable(classifier.itemStore()).ifPresent(informer::itemStore); + + configurationService + .getInformerStoppedHandler() + .ifPresent( + ish -> { + final var stopped = informer.stopped(); + if (stopped != null) { + stopped.handle( + (res, ex) -> { + ish.onStop(informer, (Throwable) ex); + return null; + }); + } else { + final var apiTypeClass = informer.getApiTypeClass(); + final var fullResourceName = HasMetadata.getFullResourceName(apiTypeClass); + final var version = HasMetadata.getVersion(apiTypeClass); + throw new IllegalStateException( + "Cannot retrieve 'stopped' callback to listen to informer stopping for" + + " informer for " + + fullResourceName + + "/" + + version); + } + }); + if (!configurationService.stopOnInformerErrorDuringStartup()) { + informer.exceptionHandler((b, t) -> !ExceptionHandler.isDeserializationException(t)); + } + return informer; + } + + @Override + public void start( + SharedIndexInformer informer, InformerClassifier informerClassifier) { + // change thread name for easier debugging + final var thread = Thread.currentThread(); + final var name = thread.getName(); + try { + thread.setName( + "InformerInfo[" + informer.getApiTypeClass().getSimpleName() + "] " + thread.getId()); + final var resourceName = informer.getApiTypeClass().getSimpleName(); + // idempotent: if the informer was already started (e.g. by the pool when it was + // created/reused), this just returns the existing start future without restarting it + var start = informer.start(); + // note that in case we don't put here timeout and stopOnInformerErrorDuringStartup is + // false, and there is a rbac issue the get never returns; therefore operator never really + // starts + log.trace( + "Waiting informer to start namespace: {} resource: {}", + informerClassifier.namespaceIdentifier(), + resourceName); + start + .toCompletableFuture() + .get(configurationService.cacheSyncTimeout().toMillis(), TimeUnit.MILLISECONDS); + log.debug( + "Started informer for namespace: {} resource: {}", + informerClassifier.namespaceIdentifier(), + resourceName); + } catch (TimeoutException | ExecutionException e) { + if (configurationService.stopOnInformerErrorDuringStartup()) { + log.error("Informer startup error. Operator will be stopped. Informer: {}", informer, e); + throw new OperatorException(e); + } else { + log.warn("Informer startup error. Will periodically retry. Informer: {}", informer, e); + } + } catch (InterruptedException e) { + thread.interrupt(); + throw new IllegalStateException(e); + } catch (Exception e) { + ReconcilerUtilsInternal.handleKubernetesClientException( + e, HasMetadata.getFullResourceName(informer.getApiTypeClass())); + throw new OperatorException( + "Couldn't start informer for " + versionedFullResourceName(informer) + " resources", e); + } finally { + // restore original name + thread.setName(name); + } + } + + private String versionedFullResourceName(SharedIndexInformer informer) { + final var apiTypeClass = informer.getApiTypeClass(); + if (GenericKubernetesResource.class.isAssignableFrom(apiTypeClass)) { + return GenericKubernetesResource.class.getSimpleName(); + } + return ReconcilerUtilsInternal.getResourceTypeNameWithVersion(apiTypeClass); + } +} diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/pool/AlwaysNewInformerPool.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/pool/AlwaysNewInformerPool.java new file mode 100644 index 0000000000..6e808e40d2 --- /dev/null +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/pool/AlwaysNewInformerPool.java @@ -0,0 +1,92 @@ +/* + * Copyright Java Operator SDK Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.javaoperatorsdk.operator.processing.event.source.informer.pool; + +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.ConcurrentHashMap; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import io.fabric8.kubernetes.api.model.HasMetadata; +import io.fabric8.kubernetes.client.KubernetesClient; +import io.fabric8.kubernetes.client.informers.SharedIndexInformer; +import io.javaoperatorsdk.operator.OperatorException; + +@SuppressWarnings({"unchecked", "rawtypes"}) +public class AlwaysNewInformerPool extends AbstractInformerPool { + + private static final Logger log = LoggerFactory.getLogger(AlwaysNewInformerPool.class); + + private final Map informers = new ConcurrentHashMap(); + + @Override + public SharedIndexInformer getInformer( + String controllerName, + String name, + InformerClassifier classifier, + KubernetesClient client) { + var key = new ClassifierWithName(controllerName, name, classifier); + if (informers.containsKey(key)) { + throw new OperatorException( + "Informer already registered for controller: " + + controllerName + + ", event source: " + + name + + ", classifier: " + + classifier + + ". This pool creates a dedicated informer per controller/event source and never" + + " shares them, so requesting one twice for the same combination without releasing" + + " the previous one first would leak the earlier informer."); + } + var informer = createInformer(classifier, client); + informers.put(key, informer); + return informer; + } + + @Override + public Optional> releaseInformer( + String controllerName, String name, InformerClassifier classifier) { + var informer = informers.remove(new ClassifierWithName(controllerName, name, classifier)); + if (informer != null) { + informer.stop(); + } else { + log.warn("Informer was not found for classifier: {}", classifier); + } + return Optional.ofNullable(informer); + } + + /** Number of informers currently tracked (i.e. created but not yet released). */ + public int size() { + return informers.size(); + } + + /** + * Number of distinct informers currently held for the given resource type. Since this pool never + * shares informers, this equals the number of registered users (controller + event source name) + * watching that resource type. + */ + @Override + public long numberOfInformersForResource(Class resourceClass) { + return informers.keySet().stream() + .filter(key -> resourceClass.equals(key.classifier().resourceClass())) + .count(); + } + + public record ClassifierWithName( + String controllerName, String name, InformerClassifier classifier) {} +} diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/pool/DefaultInformerPool.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/pool/DefaultInformerPool.java new file mode 100644 index 0000000000..66b341cce1 --- /dev/null +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/pool/DefaultInformerPool.java @@ -0,0 +1,136 @@ +/* + * Copyright Java Operator SDK Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.javaoperatorsdk.operator.processing.event.source.informer.pool; + +import java.util.HashMap; +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.atomic.AtomicInteger; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import io.fabric8.kubernetes.api.model.HasMetadata; +import io.fabric8.kubernetes.client.KubernetesClient; +import io.fabric8.kubernetes.client.informers.SharedIndexInformer; + +public class DefaultInformerPool extends AbstractInformerPool { + + private static final Logger log = LoggerFactory.getLogger(DefaultInformerPool.class); + + private final Map, SharedIndexInformer> informers = new HashMap<>(); + private final Map, AtomicInteger> counters = new HashMap<>(); + + @SuppressWarnings("unchecked") + @Override + public SharedIndexInformer getInformer( + String controllerName, + String name, + InformerClassifier classifier, + KubernetesClient client) { + SharedIndexInformer informer; + synchronized (this) { + informer = (SharedIndexInformer) informers.get(classifier); + if (informer == null) { + informer = createInformer(classifier, client); + informers.put(classifier, informer); + counters.put(classifier, new AtomicInteger(1)); + log.debug( + "Created new pooled informer for classifier: {}. Requested by controller: {}, event" + + " source: {}", + classifier, + controllerName, + name); + } else { + informers.keySet().stream() + .filter(existing -> existing.differsOnlyByInformerListLimit(classifier)) + .findFirst() + .ifPresent( + existing -> + log.warn( + "Reusing informer for classifier {} that differs only by informerListLimit" + + " (existing: {}, requested: {}). The existing informerListLimit is" + + " kept.", + classifier, + existing.informerListLimit(), + classifier.informerListLimit())); + var referenceCount = counters.get(classifier).incrementAndGet(); + log.debug( + "Reusing pooled informer for classifier: {}. Reference count now: {}. Requested by" + + " controller: {}, event source: {}", + classifier, + referenceCount, + controllerName, + name); + } + } + return informer; + } + + @SuppressWarnings("unchecked") + @Override + public synchronized Optional> releaseInformer( + String controllerName, String name, InformerClassifier classifier) { + var counter = counters.get(classifier); + var informer = (SharedIndexInformer) informers.get(classifier); + if (counter == null || informer == null) { + log.warn("No informer found in the pool for classifier: {}", classifier); + return Optional.empty(); + } + // Only the last controller sharing the informer stops it; the informer is still returned to the + // caller in every case so it can remove its own event handler from the (possibly still running) + // shared informer. + var referenceCount = counter.decrementAndGet(); + if (referenceCount == 0) { + counters.remove(classifier); + informers.remove(classifier); + informer.stop(); + log.debug( + "Released and stopped last-referenced pooled informer for classifier: {}. Released by" + + " controller: {}, event source: {}", + classifier, + controllerName, + name); + } else { + log.debug( + "Released pooled informer for classifier: {}, kept running. Reference count now: {}." + + " Released by controller: {}, event source: {}", + classifier, + referenceCount, + controllerName, + name); + } + return Optional.of(informer); + } + + /** Total number of distinct informers currently held in the pool. */ + public synchronized int size() { + return informers.size(); + } + + /** + * Number of distinct informers currently held in the pool for the given resource type. When + * multiple controllers share a single informer for a resource, this returns {@code 1} for that + * resource type regardless of how many controllers use it. + */ + @Override + public synchronized long numberOfInformersForResource( + Class resourceClass) { + return informers.keySet().stream() + .filter(classifier -> resourceClass.equals(classifier.resourceClass())) + .count(); + } +} diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/pool/InformerClassifier.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/pool/InformerClassifier.java new file mode 100644 index 0000000000..92c265f1a7 --- /dev/null +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/pool/InformerClassifier.java @@ -0,0 +1,78 @@ +/* + * Copyright Java Operator SDK Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.javaoperatorsdk.operator.processing.event.source.informer.pool; + +import java.util.Objects; + +import io.fabric8.kubernetes.api.model.HasMetadata; +import io.fabric8.kubernetes.client.informers.cache.ItemStore; +import io.javaoperatorsdk.operator.api.config.informer.FieldSelector; +import io.javaoperatorsdk.operator.processing.GroupVersionKind; + +/** + * Indexers are not part of classifier since those can be added dynamically to an informer, also can + * live side by side from different controllers, just users have to avoid naming collision. + */ +public record InformerClassifier( + String apiServerUrl, + String labelSelector, + String shardSelector, + String namespaceIdentifier, + Class resourceClass, + GroupVersionKind groupVersionKind, + FieldSelector fieldSelector, + Long informerListLimit, + ItemStore itemStore) { + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (!(o instanceof InformerClassifier that)) { + return false; + } + return Objects.equals(apiServerUrl, that.apiServerUrl) + && Objects.equals(labelSelector, that.labelSelector) + && Objects.equals(shardSelector, that.shardSelector) + && Objects.equals(namespaceIdentifier, that.namespaceIdentifier) + && Objects.equals(resourceClass, that.resourceClass) + && Objects.equals(groupVersionKind, that.groupVersionKind) + && Objects.equals(fieldSelector, that.fieldSelector) + && Objects.equals(itemStore, that.itemStore); + } + + @Override + public int hashCode() { + return Objects.hash( + apiServerUrl, + labelSelector, + shardSelector, + namespaceIdentifier, + resourceClass, + groupVersionKind, + fieldSelector, + itemStore); + } + + /** + * Checks whether this classifier and the other are equal in every attribute except for the {@link + * #informerListLimit()}, which differs between them. + */ + public boolean differsOnlyByInformerListLimit(InformerClassifier other) { + return equals(other) && !Objects.equals(informerListLimit, other.informerListLimit); + } +} diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/pool/InformerPool.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/pool/InformerPool.java new file mode 100644 index 0000000000..5b8d864210 --- /dev/null +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/pool/InformerPool.java @@ -0,0 +1,59 @@ +/* + * Copyright Java Operator SDK Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.javaoperatorsdk.operator.processing.event.source.informer.pool; + +import java.util.Optional; + +import io.fabric8.kubernetes.api.model.HasMetadata; +import io.fabric8.kubernetes.client.KubernetesClient; +import io.fabric8.kubernetes.client.informers.SharedIndexInformer; +import io.javaoperatorsdk.operator.api.config.ConfigurationService; +import io.javaoperatorsdk.operator.api.reconciler.Experimental; + +@Experimental( + "This is experimental only in the sense that the API could be improved in a" + + " non-backwards-compatible way. The feature we provide otherwise is prod ready.") +public interface InformerPool { + + SharedIndexInformer getInformer( + String controllerName, + String name, + InformerClassifier classifier, + KubernetesClient client); + + /** + * Starts the informer (if not already started) and blocks until its cache has synced, or the + * configured {@link ConfigurationService#cacheSyncTimeout()} elapses. Callers are expected to + * invoke this after {@link #getInformer(String,String, InformerClassifier, KubernetesClient)} + * returns; the pool itself only registers/reference-counts the informer and does not block on + * cache sync internally. + */ + void start( + SharedIndexInformer informer, InformerClassifier classifier); + + Optional> releaseInformer( + String controllerName, String name, InformerClassifier classifier); + + /** + * Number of distinct informers currently held in the pool for the given resource type. With a + * sharing pool multiple controllers watching the same resource are backed by a single informer + * (so this returns {@code 1}), whereas a non-sharing pool creates one informer per user. + */ + long numberOfInformersForResource(Class resourceClass); + + /** Injecting the configuration service, so users have a cleaner API creating the pool. */ + void setConfigurationService(ConfigurationService configurationService); +} diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/MockKubernetesClient.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/MockKubernetesClient.java index 61b434c0c4..3e5b872ba2 100644 --- a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/MockKubernetesClient.java +++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/MockKubernetesClient.java @@ -26,12 +26,12 @@ import io.fabric8.kubernetes.api.model.authorization.v1.ResourceRule; import io.fabric8.kubernetes.api.model.authorization.v1.SelfSubjectRulesReview; import io.fabric8.kubernetes.api.model.authorization.v1.SubjectRulesReviewStatus; +import io.fabric8.kubernetes.client.Config; import io.fabric8.kubernetes.client.KubernetesClient; import io.fabric8.kubernetes.client.V1ApiextensionAPIGroupDSL; import io.fabric8.kubernetes.client.dsl.AnyNamespaceOperation; import io.fabric8.kubernetes.client.dsl.ApiextensionsAPIGroupDSL; import io.fabric8.kubernetes.client.dsl.FilterWatchListDeletable; -import io.fabric8.kubernetes.client.dsl.Informable; import io.fabric8.kubernetes.client.dsl.MixedOperation; import io.fabric8.kubernetes.client.dsl.NamespaceableResource; import io.fabric8.kubernetes.client.dsl.NonNamespaceOperation; @@ -112,9 +112,9 @@ public static KubernetesClient client( when(filterable.runnableInformer(anyLong())).thenReturn(informer); - Informable informable = mock(Informable.class); - when(filterable.withLimit(anyLong())).thenReturn(informable); - when(informable.runnableInformer(anyLong())).thenReturn(informer); + // The informer pool casts the result of withLimit() back to FilterWatchListDeletable, so it has + // to return the filterable mock (which is one) rather than a plain Informable mock. + when(filterable.withLimit(anyLong())).thenReturn(filterable); when(client.resources(clazz)).thenReturn(resources); when(client.leaderElector()) @@ -138,6 +138,10 @@ public static KubernetesClient client( final var serialization = new KubernetesSerialization(); when(client.getKubernetesSerialization()).thenReturn(serialization); + final var config = mock(Config.class); + when(config.getMasterUrl()).thenReturn("https://localhost:8443/"); + when(client.getConfiguration()).thenReturn(config); + return client; } diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/ControllerTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/ControllerTest.java index 91d60f7aa7..b725f49132 100644 --- a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/ControllerTest.java +++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/ControllerTest.java @@ -62,6 +62,9 @@ class ControllerTest { @Test void crdShouldNotBeCheckedForNativeResources() { final var client = MockKubernetesClient.client(Secret.class); + final var configurationService = + ConfigurationService.newOverriddenConfigurationService( + this.configurationService, o -> o.withKubernetesClient(client)); final var configuration = MockControllerConfiguration.forResource(Secret.class, configurationService); final var controller = new Controller(reconciler, configuration, client); @@ -75,7 +78,8 @@ void notifiesMetricsWhenEventProcessorStarts() { final var metrics = mock(Metrics.class); final var configurationService = ConfigurationService.newOverriddenConfigurationService( - new BaseConfigurationService(), o -> o.withMetrics(metrics)); + new BaseConfigurationService(), + o -> o.withMetrics(metrics).withKubernetesClient(client)); final var configuration = MockControllerConfiguration.forResource(Secret.class, configurationService); final var controller = new Controller(reconciler, configuration, client); @@ -95,7 +99,8 @@ void doesNotNotifyMetricsWhenEventProcessorNotStarted() { final var metrics = mock(Metrics.class); final var configurationService = ConfigurationService.newOverriddenConfigurationService( - new BaseConfigurationService(), o -> o.withMetrics(metrics)); + new BaseConfigurationService(), + o -> o.withMetrics(metrics).withKubernetesClient(client)); final var configuration = MockControllerConfiguration.forResource(Secret.class, configurationService); final var controller = new Controller(reconciler, configuration, client); @@ -110,7 +115,8 @@ void crdShouldNotBeCheckedForCustomResourcesIfDisabled() { final var client = MockKubernetesClient.client(TestCustomResource.class); ConfigurationService configurationService = ConfigurationService.newOverriddenConfigurationService( - new BaseConfigurationService(), o -> o.checkingCRDAndValidateLocalModel(false)); + new BaseConfigurationService(), + o -> o.checkingCRDAndValidateLocalModel(false).withKubernetesClient(client)); final var configuration = MockControllerConfiguration.forResource(TestCustomResource.class, configurationService); 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 251a0e47ae..f47b72820f 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 @@ -24,6 +24,7 @@ import io.javaoperatorsdk.operator.MockKubernetesClient; import io.javaoperatorsdk.operator.OperatorException; import io.javaoperatorsdk.operator.api.config.BaseConfigurationService; +import io.javaoperatorsdk.operator.api.config.ConfigurationService; import io.javaoperatorsdk.operator.api.config.MockControllerConfiguration; import io.javaoperatorsdk.operator.api.config.informer.InformerEventSourceConfiguration; import io.javaoperatorsdk.operator.api.reconciler.Reconciler; @@ -202,12 +203,14 @@ void changesNamespacesOnControllerAndInformerEventSources() { private EventSourceManager initManager() { final var configuration = MockControllerConfiguration.forResource(ConfigMap.class); - final var configService = new BaseConfigurationService(); + final var mockClient = MockKubernetesClient.client(ConfigMap.class); + final var configService = + ConfigurationService.newOverriddenConfigurationService( + new BaseConfigurationService(), + overrider -> overrider.withKubernetesClient(mockClient)); when(configuration.getConfigurationService()).thenReturn(configService); - final Controller controller = - new Controller( - mock(Reconciler.class), configuration, MockKubernetesClient.client(ConfigMap.class)); + final Controller controller = new Controller(mock(Reconciler.class), configuration, mockClient); return new EventSourceManager(controller); } } diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/ReconciliationDispatcherTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/ReconciliationDispatcherTest.java index ac24375242..191836a0dc 100644 --- a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/ReconciliationDispatcherTest.java +++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/ReconciliationDispatcherTest.java @@ -122,6 +122,21 @@ private ReconciliationDispatcher init( boolean useFinalizer) { final Class resourceClass = (Class) customResource.getClass(); + final var kubernetesClient = MockKubernetesClient.client(resourceClass); + // The informer pool obtains its client from the ConfigurationService, so the mock client has to + // be set there as well (not only on the Controller); otherwise starting the informers would hit + // a real cluster. Cloner and SSA settings are re-supplied so the re-wrapping does not drop + // them. + configurationService = + ConfigurationService.newOverriddenConfigurationService( + configurationService, + overrider -> + overrider + .withKubernetesClient(kubernetesClient) + .withResourceCloner(configurationService.getResourceCloner()) + .withUseSSAToPatchPrimaryResource( + configurationService.useSSAToPatchPrimaryResource())); + configuration = configuration == null ? MockControllerConfiguration.forResource(resourceClass, configurationService) @@ -139,7 +154,7 @@ private ReconciliationDispatcher init( .thenReturn(Optional.of(Duration.ofHours(RECONCILIATION_MAX_INTERVAL))); Controller controller = - new Controller<>(reconciler, configuration, MockKubernetesClient.client(resourceClass)) { + new Controller<>(reconciler, configuration, kubernetesClient) { @Override public boolean useFinalizer() { return useFinalizer; diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/controller/ControllerEventSourceTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/controller/ControllerEventSourceTest.java index 38190a96dc..185b626161 100644 --- a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/controller/ControllerEventSourceTest.java +++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/controller/ControllerEventSourceTest.java @@ -27,6 +27,7 @@ import io.javaoperatorsdk.operator.ReconcilerUtilsInternal; import io.javaoperatorsdk.operator.TestUtils; import io.javaoperatorsdk.operator.api.config.BaseConfigurationService; +import io.javaoperatorsdk.operator.api.config.ConfigurationService; import io.javaoperatorsdk.operator.api.config.ControllerConfiguration; import io.javaoperatorsdk.operator.api.config.ResolvedControllerConfiguration; import io.javaoperatorsdk.operator.api.config.informer.InformerConfiguration; @@ -59,7 +60,11 @@ class ControllerEventSourceTest @BeforeEach public void setup() { - when(controllerConfig.getConfigurationService()).thenReturn(new BaseConfigurationService()); + var clientMock = MockKubernetesClient.client(TestCustomResource.class); + when(controllerConfig.getConfigurationService()) + .thenReturn( + ConfigurationService.newOverriddenConfigurationService( + new BaseConfigurationService(), o -> o.withKubernetesClient(clientMock))); var ic = mock(InformerConfiguration.class); when(controllerConfig.getInformerConfig()).thenReturn(ic); diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerEventSourceTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerEventSourceTest.java index 210ce52fcc..01ee25e44c 100644 --- a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerEventSourceTest.java +++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerEventSourceTest.java @@ -94,6 +94,10 @@ void setup() { when(informerEventSourceConfiguration.getResourceClass()).thenReturn(Deployment.class); when(informerConfig.isComparableResourceVersions()).thenReturn(true); when(informerConfig.getEffectiveNamespaces(any())).thenReturn(DEFAULT_NAMESPACES_SET); + // a plain Long-returning Mockito mock yields 0 here, but a real unconfigured informer has no + // list limit; without this the pool would take the withLimit(...) branch when creating the + // informer + when(informerConfig.getInformerListLimit()).thenReturn(null); informerEventSource = buildInformerEventSource(); } @@ -101,7 +105,7 @@ void setup() { private InformerEventSource buildInformerEventSource() { InformerEventSource eventSource = spy( - new InformerEventSource<>(informerEventSourceConfiguration, clientMock) { + new InformerEventSource<>(informerEventSourceConfiguration) { // mocking start @Override public synchronized void start() {} @@ -259,22 +263,25 @@ void ownUpdateEventIsDeferredDuringActiveFilter() { void informerStoppedHandlerShouldBeCalledWhenInformerStops() { final var exception = new RuntimeException("Informer stopped exceptionally!"); final var informerStoppedHandler = mock(InformerStoppedHandler.class); + // the informer is created by the pool, which uses the client from the configuration service, so + // the mock client has to be set there for its informer-start behavior to take effect + final var mockClient = + MockKubernetesClient.client( + Deployment.class, + unused -> { + throw exception; + }); var configuration = ConfigurationService.newOverriddenConfigurationService( new BaseConfigurationService(), - o -> o.withInformerStoppedHandler(informerStoppedHandler)); + o -> + o.withInformerStoppedHandler(informerStoppedHandler) + .withKubernetesClient(mockClient)); var mockControllerConfig = mock(ControllerConfiguration.class); when(mockControllerConfig.getConfigurationService()).thenReturn(configuration); - informerEventSource = - new InformerEventSource<>( - informerEventSourceConfiguration, - MockKubernetesClient.client( - Deployment.class, - unused -> { - throw exception; - })); + informerEventSource = new InformerEventSource<>(informerEventSourceConfiguration); informerEventSource.setControllerConfiguration(mockControllerConfig); // by default informer fails to start if there is an exception in the client on start. diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/informer/pool/AlwaysNewInformerPoolTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/informer/pool/AlwaysNewInformerPoolTest.java new file mode 100644 index 0000000000..25196c5074 --- /dev/null +++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/informer/pool/AlwaysNewInformerPoolTest.java @@ -0,0 +1,151 @@ +/* + * Copyright Java Operator SDK Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.javaoperatorsdk.operator.processing.event.source.informer.pool; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import io.fabric8.kubernetes.client.KubernetesClient; +import io.javaoperatorsdk.operator.MockKubernetesClient; +import io.javaoperatorsdk.operator.OperatorException; +import io.javaoperatorsdk.operator.api.config.BaseConfigurationService; +import io.javaoperatorsdk.operator.sample.simple.TestCustomResource; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; + +/** + * Unit tests for {@link AlwaysNewInformerPool}: unlike {@link DefaultInformerPool} it never shares + * informers. A distinct informer is created for every {@code controllerName}+{@code name}+{@code + * classifier} combination, and release is scoped to that same combination. + */ +class AlwaysNewInformerPoolTest { + + private static final String CONTROLLER = "controller"; + private static final String ES_NAME = "event-source"; + + private final KubernetesClient client = MockKubernetesClient.client(TestCustomResource.class); + private final AlwaysNewInformerPool pool = new AlwaysNewInformerPool(); + + @BeforeEach + void setUp() { + pool.setConfigurationService(new BaseConfigurationService()); + } + + @Test + void createsSeparateInformerForSameClassifierFromDifferentControllers() { + var classifier = classifier("default"); + + pool.getInformer(CONTROLLER, ES_NAME, classifier, client); + pool.getInformer("other-controller", ES_NAME, classifier, client); + + // no sharing: one informer created per user even for an identical classifier + assertThat(pool.size()).isEqualTo(2); + verify(client, times(2)).resources(TestCustomResource.class); + } + + @Test + void createsSeparateInformerForDifferentEventSourceNames() { + var classifier = classifier("default"); + + pool.getInformer(CONTROLLER, "event-source-1", classifier, client); + pool.getInformer(CONTROLLER, "event-source-2", classifier, client); + + assertThat(pool.size()).isEqualTo(2); + verify(client, times(2)).resources(TestCustomResource.class); + } + + @Test + void throwsWhenRequestingAnInformerForAnAlreadyRegisteredKey() { + var classifier = classifier("default"); + + pool.getInformer(CONTROLLER, ES_NAME, classifier, client); + + // requesting the same controller+event source+classifier combination again without releasing + // first would otherwise silently overwrite the map entry and leak the earlier informer + assertThatThrownBy(() -> pool.getInformer(CONTROLLER, ES_NAME, classifier, client)) + .isInstanceOf(OperatorException.class) + .hasMessageContaining(CONTROLLER) + .hasMessageContaining(ES_NAME) + .hasMessageContaining(classifier.toString()); + + // the earlier informer is left untouched, still registered exactly once + assertThat(pool.size()).isEqualTo(1); + verify(client, times(1)).resources(TestCustomResource.class); + } + + @Test + void allowsReRequestingAnInformerAfterItWasReleased() { + var classifier = classifier("default"); + pool.getInformer(CONTROLLER, ES_NAME, classifier, client); + pool.releaseInformer(CONTROLLER, ES_NAME, classifier); + + // must not throw: releasing frees up the key for reuse + pool.getInformer(CONTROLLER, ES_NAME, classifier, client); + + assertThat(pool.size()).isEqualTo(1); + verify(client, times(2)).resources(TestCustomResource.class); + } + + @Test + void releaseStopsAndRemovesTheInformer() { + var classifier = classifier("default"); + var informer = pool.getInformer(CONTROLLER, ES_NAME, classifier, client); + + var released = pool.releaseInformer(CONTROLLER, ES_NAME, classifier); + + assertThat(released).containsSame(informer); + verify(informer, times(1)).stop(); + assertThat(pool.size()).isZero(); + } + + @Test + void releaseIsScopedToControllerAndName() { + var classifier = classifier("default"); + var informer = pool.getInformer(CONTROLLER, ES_NAME, classifier, client); + + // same classifier but a different controller: nothing is released or stopped + var released = pool.releaseInformer("other-controller", ES_NAME, classifier); + + assertThat(released).isEmpty(); + verify(informer, never()).stop(); + assertThat(pool.size()).isEqualTo(1); + } + + @Test + void releaseOfUnknownInformerReturnsEmptyAndDoesNotThrow() { + var released = pool.releaseInformer(CONTROLLER, ES_NAME, classifier("never-registered")); + + assertThat(released).isEmpty(); + assertThat(pool.size()).isZero(); + } + + private InformerClassifier classifier(String namespace) { + return new InformerClassifier<>( + client.getConfiguration().getMasterUrl(), + null, + null, + namespace, + TestCustomResource.class, + null, + null, + null, + null); + } +} diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/informer/pool/DefaultInformerPoolTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/informer/pool/DefaultInformerPoolTest.java new file mode 100644 index 0000000000..983494fd03 --- /dev/null +++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/informer/pool/DefaultInformerPoolTest.java @@ -0,0 +1,134 @@ +/* + * Copyright Java Operator SDK Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.javaoperatorsdk.operator.processing.event.source.informer.pool; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import io.fabric8.kubernetes.client.KubernetesClient; +import io.javaoperatorsdk.operator.MockKubernetesClient; +import io.javaoperatorsdk.operator.api.config.BaseConfigurationService; +import io.javaoperatorsdk.operator.sample.simple.TestCustomResource; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; + +/** + * Unit tests for the reference-counting / sharing behavior of {@link DefaultInformerPool}: a single + * informer is created and shared per classifier, and it is stopped only when the last user releases + * it. + */ +class DefaultInformerPoolTest { + + private static final String CONTROLLER = "controller"; + private static final String ES_NAME = "event-source"; + + private final KubernetesClient client = MockKubernetesClient.client(TestCustomResource.class); + private final DefaultInformerPool pool = new DefaultInformerPool(); + + @BeforeEach + void setup() { + pool.setConfigurationService(new BaseConfigurationService()); + } + + @Test + void sharesSingleInformerForTheSameClassifier() { + var classifier = classifier("default"); + + var first = pool.getInformer(CONTROLLER, ES_NAME, classifier, client); + var second = pool.getInformer("other-controller", "other-es", classifier, client); + + assertThat(first).isSameAs(second); + assertThat(pool.size()).isEqualTo(1); + assertThat(pool.numberOfInformersForResource(TestCustomResource.class)).isEqualTo(1); + // the underlying informer must be created exactly once, not once per user + verify(client, times(1)).resources(TestCustomResource.class); + } + + @Test + void createsSeparateInformersForDifferentClassifiers() { + pool.getInformer(CONTROLLER, ES_NAME, classifier("ns1"), client); + pool.getInformer(CONTROLLER, ES_NAME, classifier("ns2"), client); + + assertThat(pool.size()).isEqualTo(2); + assertThat(pool.numberOfInformersForResource(TestCustomResource.class)).isEqualTo(2); + verify(client, times(2)).resources(TestCustomResource.class); + } + + @Test + void sharesInformerWhenClassifiersDifferOnlyByListLimit() { + var withLimit100 = + new InformerClassifier<>( + masterUrl(), null, null, "default", TestCustomResource.class, null, null, 100L, null); + var withLimit200 = + new InformerClassifier<>( + masterUrl(), null, null, "default", TestCustomResource.class, null, null, 200L, null); + + var first = pool.getInformer(CONTROLLER, ES_NAME, withLimit100, client); + var second = pool.getInformer("other-controller", "other-es", withLimit200, client); + + assertThat(first).isSameAs(second); + assertThat(pool.size()).isEqualTo(1); + verify(client, times(1)).resources(TestCustomResource.class); + } + + @Test + void doesNotStopSharedInformerUntilLastRelease() { + var classifier = classifier("default"); + var informer = pool.getInformer(CONTROLLER, ES_NAME, classifier, client); + pool.getInformer("other-controller", "other-es", classifier, client); + + pool.releaseInformer(CONTROLLER, ES_NAME, classifier); + verify(informer, never()).stop(); + assertThat(pool.size()).isEqualTo(1); + + pool.releaseInformer("other-controller", "other-es", classifier); + verify(informer, times(1)).stop(); + assertThat(pool.size()).isZero(); + } + + @Test + void releaseReturnsInformerEvenWhileStillShared() { + var classifier = classifier("default"); + var informer = pool.getInformer(CONTROLLER, ES_NAME, classifier, client); + pool.getInformer("other-controller", "other-es", classifier, client); + + // the caller needs the (still-running) informer back so it can remove its own event handler + var released = pool.releaseInformer(CONTROLLER, ES_NAME, classifier); + + assertThat(released).containsSame(informer); + verify(informer, never()).stop(); + } + + @Test + void releaseOfUnknownClassifierReturnsEmptyAndDoesNotThrow() { + var released = pool.releaseInformer(CONTROLLER, ES_NAME, classifier("never-registered")); + + assertThat(released).isEmpty(); + assertThat(pool.size()).isZero(); + } + + private String masterUrl() { + return client.getConfiguration().getMasterUrl(); + } + + private InformerClassifier classifier(String namespace) { + return new InformerClassifier<>( + masterUrl(), null, null, namespace, TestCustomResource.class, null, null, null, null); + } +} diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/informer/pool/InformerClassifierTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/informer/pool/InformerClassifierTest.java new file mode 100644 index 0000000000..c251a31f3f --- /dev/null +++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/informer/pool/InformerClassifierTest.java @@ -0,0 +1,265 @@ +/* + * Copyright Java Operator SDK Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.javaoperatorsdk.operator.processing.event.source.informer.pool; + +import org.junit.jupiter.api.Test; + +import io.fabric8.kubernetes.client.informers.cache.ItemStore; +import io.javaoperatorsdk.operator.api.config.informer.FieldSelector; +import io.javaoperatorsdk.operator.processing.GroupVersionKind; +import io.javaoperatorsdk.operator.sample.simple.TestCustomResource; +import io.javaoperatorsdk.operator.sample.simple.TestCustomResourceOtherV1; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.Mockito.mock; + +/** + * Unit tests for the identity semantics of {@link InformerClassifier}. The classifier is used as + * the key that decides whether two controllers share a single pooled informer, so its equality + * contract (in particular that {@code informerListLimit} is intentionally excluded) is + * load-bearing. + */ +class InformerClassifierTest { + + private static final String URL = "https://localhost:8443/"; + private static final String LABEL = "app=foo"; + private static final String SHARD = "shard-1"; + private static final String NAMESPACE = "default"; + private static final GroupVersionKind GVK = new GroupVersionKind("sample.io/v1", "Foo"); + private static final FieldSelector FIELD_SELECTOR = + new FieldSelector(new FieldSelector.Field("status.phase", "Running")); + private static final Long LIMIT = 100L; + private static final ItemStore ITEM_STORE = mock(ItemStore.class); + + private static InformerClassifier base() { + return new InformerClassifier<>( + URL, + LABEL, + SHARD, + NAMESPACE, + TestCustomResource.class, + GVK, + FIELD_SELECTOR, + LIMIT, + ITEM_STORE); + } + + @Test + void classifiersWithIdenticalFieldsAreEqual() { + assertThat(base()).isEqualTo(base()); + assertThat(base()).hasSameHashCodeAs(base()); + } + + @Test + void informerListLimitIsExcludedFromEqualityAndHashCode() { + var withOtherLimit = + new InformerClassifier<>( + URL, + LABEL, + SHARD, + NAMESPACE, + TestCustomResource.class, + GVK, + FIELD_SELECTOR, + 999L, + ITEM_STORE); + + assertThat(base()).isEqualTo(withOtherLimit); + assertThat(base()).hasSameHashCodeAs(withOtherLimit); + } + + @Test + void differsWhenApiServerUrlDiffers() { + assertThat(base()) + .isNotEqualTo( + new InformerClassifier<>( + "https://other:8443/", + LABEL, + SHARD, + NAMESPACE, + TestCustomResource.class, + GVK, + FIELD_SELECTOR, + LIMIT, + ITEM_STORE)); + } + + @Test + void differsWhenLabelSelectorDiffers() { + assertThat(base()) + .isNotEqualTo( + new InformerClassifier<>( + URL, + "app=bar", + SHARD, + NAMESPACE, + TestCustomResource.class, + GVK, + FIELD_SELECTOR, + LIMIT, + ITEM_STORE)); + } + + @Test + void differsWhenShardSelectorDiffers() { + assertThat(base()) + .isNotEqualTo( + new InformerClassifier<>( + URL, + LABEL, + "shard-2", + NAMESPACE, + TestCustomResource.class, + GVK, + FIELD_SELECTOR, + LIMIT, + ITEM_STORE)); + } + + @Test + void differsWhenNamespaceDiffers() { + assertThat(base()) + .isNotEqualTo( + new InformerClassifier<>( + URL, + LABEL, + SHARD, + "other-ns", + TestCustomResource.class, + GVK, + FIELD_SELECTOR, + LIMIT, + ITEM_STORE)); + } + + @Test + void differsWhenResourceClassDiffers() { + // item stores are null here because their generic type is tied to the resource class, which is + // exactly the field under test; this keeps the resource class the only difference. + var forTestResource = + new InformerClassifier<>( + URL, + LABEL, + SHARD, + NAMESPACE, + TestCustomResource.class, + GVK, + FIELD_SELECTOR, + LIMIT, + null); + var forOtherResource = + new InformerClassifier<>( + URL, + LABEL, + SHARD, + NAMESPACE, + TestCustomResourceOtherV1.class, + GVK, + FIELD_SELECTOR, + LIMIT, + null); + + assertThat(forTestResource).isNotEqualTo(forOtherResource); + } + + @Test + void differsWhenGroupVersionKindDiffers() { + assertThat(base()) + .isNotEqualTo( + new InformerClassifier<>( + URL, + LABEL, + SHARD, + NAMESPACE, + TestCustomResource.class, + new GroupVersionKind("sample.io/v1", "Bar"), + FIELD_SELECTOR, + LIMIT, + ITEM_STORE)); + } + + @Test + void differsWhenFieldSelectorDiffers() { + assertThat(base()) + .isNotEqualTo( + new InformerClassifier<>( + URL, + LABEL, + SHARD, + NAMESPACE, + TestCustomResource.class, + GVK, + new FieldSelector(new FieldSelector.Field("status.phase", "Pending")), + LIMIT, + ITEM_STORE)); + } + + @Test + void differsWhenItemStoreDiffers() { + assertThat(base()) + .isNotEqualTo( + new InformerClassifier<>( + URL, + LABEL, + SHARD, + NAMESPACE, + TestCustomResource.class, + GVK, + FIELD_SELECTOR, + LIMIT, + mock(ItemStore.class))); + } + + @Test + void differsOnlyByInformerListLimitIsTrueWhenOnlyLimitDiffers() { + var withOtherLimit = + new InformerClassifier<>( + URL, + LABEL, + SHARD, + NAMESPACE, + TestCustomResource.class, + GVK, + FIELD_SELECTOR, + 999L, + ITEM_STORE); + + assertThat(base().differsOnlyByInformerListLimit(withOtherLimit)).isTrue(); + } + + @Test + void differsOnlyByInformerListLimitIsFalseWhenFullyEqual() { + assertThat(base().differsOnlyByInformerListLimit(base())).isFalse(); + } + + @Test + void differsOnlyByInformerListLimitIsFalseWhenAnotherFieldDiffers() { + // different namespace AND different limit: not "only by limit" + var differentNamespaceAndLimit = + new InformerClassifier<>( + URL, + LABEL, + SHARD, + "other-ns", + TestCustomResource.class, + GVK, + FIELD_SELECTOR, + 999L, + ITEM_STORE); + + assertThat(base().differsOnlyByInformerListLimit(differentNamespaceAndLimit)).isFalse(); + } +} diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/informerpool/basic/AbstractSharedInformerIT.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/informerpool/basic/AbstractSharedInformerIT.java new file mode 100644 index 0000000000..ff50ea7f43 --- /dev/null +++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/informerpool/basic/AbstractSharedInformerIT.java @@ -0,0 +1,105 @@ +/* + * Copyright Java Operator SDK Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.javaoperatorsdk.operator.baseapi.informerpool.basic; + +import java.util.function.Consumer; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.RegisterExtension; + +import io.fabric8.kubernetes.api.model.ConfigMap; +import io.fabric8.kubernetes.api.model.ObjectMetaBuilder; +import io.javaoperatorsdk.operator.api.config.ConfigurationServiceOverrider; +import io.javaoperatorsdk.operator.junit.LocallyRunOperatorExtension; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.awaitility.Awaitility.await; + +/** + * Registers two controllers, each backed by its own primary custom resource, that both watch {@link + * ConfigMap} as a secondary resource using an identical {@code InformerEventSource} configuration. + * + *

The functional behavior (both controllers reconcile) is identical regardless of the informer + * pool strategy; only the number of underlying {@code ConfigMap} informers differs, which is why + * the expected count is left abstract. Concrete subclasses pick the pool strategy via {@link + * #configurationServiceOverrider()}. + */ +public abstract class AbstractSharedInformerIT { + + public static final String TEST_RESOURCE_1 = "test1"; + public static final String TEST_RESOURCE_2 = "test2"; + + @RegisterExtension + LocallyRunOperatorExtension extension = + LocallyRunOperatorExtension.builder() + .withReconciler(new SharedInformerReconciler1()) + .withReconciler(new SharedInformerReconciler2()) + .withConfigurationService(configurationServiceOverrider()) + .build(); + + /** The informer pool strategy under test. */ + protected abstract Consumer configurationServiceOverrider(); + + /** + * Expected number of {@code ConfigMap} informers: {@code 1} when the two controllers share a + * single informer, {@code 2} when each controller gets its own. + */ + protected abstract long expectedConfigMapInformerCount(); + + @Test + void bothControllersReconcileWatchingConfigMap() { + extension.create(customResource1(TEST_RESOURCE_1)); + extension.create(customResource2(TEST_RESOURCE_2)); + + // both controllers reconcile, which guarantees their event sources (and thus their informers) + // have been started + await() + .untilAsserted( + () -> { + assertThat( + extension + .getReconcilerOfType(SharedInformerReconciler1.class) + .getNumberOfExecutions()) + .isPositive(); + assertThat( + extension + .getReconcilerOfType(SharedInformerReconciler2.class) + .getNumberOfExecutions()) + .isPositive(); + }); + + var pool = extension.getOperator().getConfigurationService().informerPool(); + + // the ConfigMap informer count depends on the pool strategy (shared vs. one-per-controller) + assertThat(pool.numberOfInformersForResource(ConfigMap.class)) + .isEqualTo(expectedConfigMapInformerCount()); + // the two distinct primary resources are always backed by their own informers + assertThat(pool.numberOfInformersForResource(SharedInformerCustomResource1.class)).isEqualTo(1); + assertThat(pool.numberOfInformersForResource(SharedInformerCustomResource2.class)).isEqualTo(1); + } + + SharedInformerCustomResource1 customResource1(String name) { + var res = new SharedInformerCustomResource1(); + res.setMetadata(new ObjectMetaBuilder().withName(name).build()); + return res; + } + + SharedInformerCustomResource2 customResource2(String name) { + var res = new SharedInformerCustomResource2(); + res.setMetadata(new ObjectMetaBuilder().withName(name).build()); + return res; + } +} diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/informerpool/basic/AlwaysNewInformerPoolSharedInformerIT.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/informerpool/basic/AlwaysNewInformerPoolSharedInformerIT.java new file mode 100644 index 0000000000..a3ecd7c5e4 --- /dev/null +++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/informerpool/basic/AlwaysNewInformerPoolSharedInformerIT.java @@ -0,0 +1,39 @@ +/* + * Copyright Java Operator SDK Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.javaoperatorsdk.operator.baseapi.informerpool.basic; + +import java.util.function.Consumer; + +import io.javaoperatorsdk.operator.api.config.ConfigurationServiceOverrider; +import io.javaoperatorsdk.operator.processing.event.source.informer.pool.AlwaysNewInformerPool; + +/** + * Runs {@link AbstractSharedInformerIT} with the {@link AlwaysNewInformerPool}: informers are never + * shared, so each of the two controllers watching {@code ConfigMap} gets its own informer. + */ +public class AlwaysNewInformerPoolSharedInformerIT extends AbstractSharedInformerIT { + + @Override + protected Consumer configurationServiceOverrider() { + return overrider -> overrider.withInformerPool(new AlwaysNewInformerPool()); + } + + @Override + protected long expectedConfigMapInformerCount() { + // no sharing: one ConfigMap informer per controller + return 2; + } +} diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/informerpool/basic/SharedInformerCustomResource1.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/informerpool/basic/SharedInformerCustomResource1.java new file mode 100644 index 0000000000..af73cb15ff --- /dev/null +++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/informerpool/basic/SharedInformerCustomResource1.java @@ -0,0 +1,28 @@ +/* + * Copyright Java Operator SDK Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.javaoperatorsdk.operator.baseapi.informerpool.basic; + +import io.fabric8.kubernetes.api.model.Namespaced; +import io.fabric8.kubernetes.client.CustomResource; +import io.fabric8.kubernetes.model.annotation.Group; +import io.fabric8.kubernetes.model.annotation.ShortNames; +import io.fabric8.kubernetes.model.annotation.Version; + +@Group("sample.javaoperatorsdk") +@Version("v1") +@ShortNames("si1") +public class SharedInformerCustomResource1 extends CustomResource + implements Namespaced {} diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/informerpool/basic/SharedInformerCustomResource2.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/informerpool/basic/SharedInformerCustomResource2.java new file mode 100644 index 0000000000..e4dbf667c2 --- /dev/null +++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/informerpool/basic/SharedInformerCustomResource2.java @@ -0,0 +1,28 @@ +/* + * Copyright Java Operator SDK Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.javaoperatorsdk.operator.baseapi.informerpool.basic; + +import io.fabric8.kubernetes.api.model.Namespaced; +import io.fabric8.kubernetes.client.CustomResource; +import io.fabric8.kubernetes.model.annotation.Group; +import io.fabric8.kubernetes.model.annotation.ShortNames; +import io.fabric8.kubernetes.model.annotation.Version; + +@Group("sample.javaoperatorsdk") +@Version("v1") +@ShortNames("si2") +public class SharedInformerCustomResource2 extends CustomResource + implements Namespaced {} diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/informerpool/basic/SharedInformerIT.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/informerpool/basic/SharedInformerIT.java new file mode 100644 index 0000000000..6e51bcb44a --- /dev/null +++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/informerpool/basic/SharedInformerIT.java @@ -0,0 +1,38 @@ +/* + * Copyright Java Operator SDK Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.javaoperatorsdk.operator.baseapi.informerpool.basic; + +import java.util.function.Consumer; + +import io.javaoperatorsdk.operator.api.config.ConfigurationServiceOverrider; + +/** + * Runs {@link AbstractSharedInformerIT} with the default (sharing) informer pool, so both + * controllers watching {@code ConfigMap} are backed by a single shared informer. + */ +public class SharedInformerIT extends AbstractSharedInformerIT { + + @Override + protected Consumer configurationServiceOverrider() { + // no override: use the default DefaultInformerPool + return overrider -> {}; + } + + @Override + protected long expectedConfigMapInformerCount() { + return 1; + } +} diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/informerpool/basic/SharedInformerReconciler1.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/informerpool/basic/SharedInformerReconciler1.java new file mode 100644 index 0000000000..d526ff7713 --- /dev/null +++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/informerpool/basic/SharedInformerReconciler1.java @@ -0,0 +1,63 @@ +/* + * Copyright Java Operator SDK Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.javaoperatorsdk.operator.baseapi.informerpool.basic; + +import java.util.List; +import java.util.concurrent.atomic.AtomicInteger; + +import io.fabric8.kubernetes.api.model.ConfigMap; +import io.javaoperatorsdk.operator.api.config.informer.InformerEventSourceConfiguration; +import io.javaoperatorsdk.operator.api.reconciler.Context; +import io.javaoperatorsdk.operator.api.reconciler.ControllerConfiguration; +import io.javaoperatorsdk.operator.api.reconciler.EventSourceContext; +import io.javaoperatorsdk.operator.api.reconciler.Reconciler; +import io.javaoperatorsdk.operator.api.reconciler.UpdateControl; +import io.javaoperatorsdk.operator.processing.event.source.EventSource; +import io.javaoperatorsdk.operator.processing.event.source.informer.InformerEventSource; + +/** + * Watches {@link ConfigMap} as a secondary resource. Together with {@link + * SharedInformerReconciler2}, which watches the same secondary resource type with an identical + * configuration, this is used to verify that both controllers share a single underlying informer + * from the informer pool. + */ +@ControllerConfiguration +public class SharedInformerReconciler1 implements Reconciler { + + private final AtomicInteger numberOfExecutions = new AtomicInteger(0); + + @Override + public List> prepareEventSources( + EventSourceContext context) { + var config = + InformerEventSourceConfiguration.from(ConfigMap.class, SharedInformerCustomResource1.class) + .build(); + return List.of(new InformerEventSource<>(config, context)); + } + + @Override + public UpdateControl reconcile( + SharedInformerCustomResource1 resource, Context context) { + numberOfExecutions.incrementAndGet(); + resource.setStatus(new SharedInformerStatus()); + resource.getStatus().setReconciledBy(getClass().getSimpleName()); + return UpdateControl.patchStatus(resource); + } + + public int getNumberOfExecutions() { + return numberOfExecutions.get(); + } +} diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/informerpool/basic/SharedInformerReconciler2.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/informerpool/basic/SharedInformerReconciler2.java new file mode 100644 index 0000000000..84f9718d52 --- /dev/null +++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/informerpool/basic/SharedInformerReconciler2.java @@ -0,0 +1,62 @@ +/* + * Copyright Java Operator SDK Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.javaoperatorsdk.operator.baseapi.informerpool.basic; + +import java.util.List; +import java.util.concurrent.atomic.AtomicInteger; + +import io.fabric8.kubernetes.api.model.ConfigMap; +import io.javaoperatorsdk.operator.api.config.informer.InformerEventSourceConfiguration; +import io.javaoperatorsdk.operator.api.reconciler.Context; +import io.javaoperatorsdk.operator.api.reconciler.ControllerConfiguration; +import io.javaoperatorsdk.operator.api.reconciler.EventSourceContext; +import io.javaoperatorsdk.operator.api.reconciler.Reconciler; +import io.javaoperatorsdk.operator.api.reconciler.UpdateControl; +import io.javaoperatorsdk.operator.processing.event.source.EventSource; +import io.javaoperatorsdk.operator.processing.event.source.informer.InformerEventSource; + +/** + * Watches {@link ConfigMap} as a secondary resource with the same configuration as {@link + * SharedInformerReconciler1} so that both controllers share a single underlying informer from the + * informer pool. + */ +@ControllerConfiguration +public class SharedInformerReconciler2 implements Reconciler { + + private final AtomicInteger numberOfExecutions = new AtomicInteger(0); + + @Override + public List> prepareEventSources( + EventSourceContext context) { + var config = + InformerEventSourceConfiguration.from(ConfigMap.class, SharedInformerCustomResource2.class) + .build(); + return List.of(new InformerEventSource<>(config, context)); + } + + @Override + public UpdateControl reconcile( + SharedInformerCustomResource2 resource, Context context) { + numberOfExecutions.incrementAndGet(); + resource.setStatus(new SharedInformerStatus()); + resource.getStatus().setReconciledBy(getClass().getSimpleName()); + return UpdateControl.patchStatus(resource); + } + + public int getNumberOfExecutions() { + return numberOfExecutions.get(); + } +} diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/informerpool/basic/SharedInformerStatus.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/informerpool/basic/SharedInformerStatus.java new file mode 100644 index 0000000000..446fee6e8b --- /dev/null +++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/informerpool/basic/SharedInformerStatus.java @@ -0,0 +1,29 @@ +/* + * Copyright Java Operator SDK Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.javaoperatorsdk.operator.baseapi.informerpool.basic; + +public class SharedInformerStatus { + + private String reconciledBy; + + public String getReconciledBy() { + return reconciledBy; + } + + public void setReconciledBy(String reconciledBy) { + this.reconciledBy = reconciledBy; + } +} diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/informerpool/deregister/AbstractDeregisterSharedInformerIT.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/informerpool/deregister/AbstractDeregisterSharedInformerIT.java new file mode 100644 index 0000000000..2ca7b864f8 --- /dev/null +++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/informerpool/deregister/AbstractDeregisterSharedInformerIT.java @@ -0,0 +1,97 @@ +/* + * Copyright Java Operator SDK Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.javaoperatorsdk.operator.baseapi.informerpool.deregister; + +import java.util.function.Consumer; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.RegisterExtension; + +import io.fabric8.kubernetes.api.model.ObjectMetaBuilder; +import io.javaoperatorsdk.operator.api.config.ConfigurationServiceOverrider; +import io.javaoperatorsdk.operator.junit.LocallyRunOperatorExtension; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.awaitility.Awaitility.await; + +/** + * Verifies the lifecycle of a dynamically registered event source in the informer pool: registering + * it creates a (single) informer for the watched resource, and de-registering it releases that + * informer so the pool no longer holds one for that resource type. + * + *

A single controller registers a single event source, so no informer sharing is involved: the + * behavior and assertions are identical for every pool strategy. Concrete subclasses only pick the + * strategy via {@link #configurationServiceOverrider()}. + */ +public abstract class AbstractDeregisterSharedInformerIT { + + private static final String PRIMARY_NAME = "primary1"; + + @RegisterExtension + LocallyRunOperatorExtension extension = + LocallyRunOperatorExtension.builder() + .withAdditionalCustomResourceDefinition(DeregisterWatchedCustomResource.class) + .withReconciler(new DeregisterReconciler()) + .withConfigurationService(configurationServiceOverrider()) + .build(); + + /** The informer pool strategy under test. */ + protected abstract Consumer configurationServiceOverrider(); + + @Test + void deregisteringDynamicEventSourceRemovesInformerFromPool() { + var reconciler = extension.getReconcilerOfType(DeregisterReconciler.class); + var pool = extension.getOperator().getConfigurationService().informerPool(); + + // Create the primary with registration enabled: the reconciler dynamically registers the event + // source for the watched resource. + extension.create(primary(true)); + + // The dynamically registered event source establishes exactly one informer for the watched + // resource in the pool. + await() + .untilAsserted( + () -> { + assertThat(reconciler.getNumberOfExecutions()).isPositive(); + assertThat(pool.numberOfInformersForResource(DeregisterWatchedCustomResource.class)) + .isEqualTo(1); + }); + + var executionsBeforeDeregister = reconciler.getNumberOfExecutions(); + + // Flip the spec so the next reconciliation de-registers the event source. + var toUpdate = extension.get(DeregisterPrimaryCustomResource.class, PRIMARY_NAME); + toUpdate.getSpec().setRegisterEventSource(false); + extension.replace(toUpdate); + + // After the de-registration reconciliation runs, the informer is released from the pool. + await() + .untilAsserted( + () -> { + assertThat(reconciler.getNumberOfExecutions()) + .isGreaterThan(executionsBeforeDeregister); + assertThat(pool.numberOfInformersForResource(DeregisterWatchedCustomResource.class)) + .isZero(); + }); + } + + DeregisterPrimaryCustomResource primary(boolean registerEventSource) { + var res = new DeregisterPrimaryCustomResource(); + res.setMetadata(new ObjectMetaBuilder().withName(PRIMARY_NAME).build()); + res.setSpec(new DeregisterSpec().setRegisterEventSource(registerEventSource)); + return res; + } +} diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/informerpool/deregister/AlwaysNewInformerPoolDeregisterSharedInformerIT.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/informerpool/deregister/AlwaysNewInformerPoolDeregisterSharedInformerIT.java new file mode 100644 index 0000000000..cfb7e9c83e --- /dev/null +++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/informerpool/deregister/AlwaysNewInformerPoolDeregisterSharedInformerIT.java @@ -0,0 +1,34 @@ +/* + * Copyright Java Operator SDK Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.javaoperatorsdk.operator.baseapi.informerpool.deregister; + +import java.util.function.Consumer; + +import io.javaoperatorsdk.operator.api.config.ConfigurationServiceOverrider; +import io.javaoperatorsdk.operator.processing.event.source.informer.pool.AlwaysNewInformerPool; + +/** + * Runs {@link AbstractDeregisterSharedInformerIT} with the {@link AlwaysNewInformerPool}. Since + * only one controller registers/de-registers a single event source, the register-then-release + * lifecycle (and thus the assertions) is identical to the sharing pool. + */ +class AlwaysNewInformerPoolDeregisterSharedInformerIT extends AbstractDeregisterSharedInformerIT { + + @Override + protected Consumer configurationServiceOverrider() { + return overrider -> overrider.withInformerPool(new AlwaysNewInformerPool()); + } +} diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/informerpool/deregister/DeregisterPrimaryCustomResource.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/informerpool/deregister/DeregisterPrimaryCustomResource.java new file mode 100644 index 0000000000..7e03081b1a --- /dev/null +++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/informerpool/deregister/DeregisterPrimaryCustomResource.java @@ -0,0 +1,29 @@ +/* + * Copyright Java Operator SDK Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.javaoperatorsdk.operator.baseapi.informerpool.deregister; + +import io.fabric8.kubernetes.api.model.Namespaced; +import io.fabric8.kubernetes.client.CustomResource; +import io.fabric8.kubernetes.model.annotation.Group; +import io.fabric8.kubernetes.model.annotation.ShortNames; +import io.fabric8.kubernetes.model.annotation.Version; + +/** Primary resource of {@link DeregisterReconciler}. */ +@Group("sample.javaoperatorsdk") +@Version("v1") +@ShortNames("dreg-p") +public class DeregisterPrimaryCustomResource extends CustomResource + implements Namespaced {} diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/informerpool/deregister/DeregisterReconciler.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/informerpool/deregister/DeregisterReconciler.java new file mode 100644 index 0000000000..680ce87077 --- /dev/null +++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/informerpool/deregister/DeregisterReconciler.java @@ -0,0 +1,73 @@ +/* + * Copyright Java Operator SDK Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.javaoperatorsdk.operator.baseapi.informerpool.deregister; + +import java.util.Set; +import java.util.concurrent.atomic.AtomicInteger; + +import io.javaoperatorsdk.operator.api.config.informer.InformerEventSourceConfiguration; +import io.javaoperatorsdk.operator.api.reconciler.Context; +import io.javaoperatorsdk.operator.api.reconciler.ControllerConfiguration; +import io.javaoperatorsdk.operator.api.reconciler.Reconciler; +import io.javaoperatorsdk.operator.api.reconciler.UpdateControl; +import io.javaoperatorsdk.operator.processing.event.ResourceID; +import io.javaoperatorsdk.operator.processing.event.source.informer.InformerEventSource; + +/** + * Dynamically registers an event source for {@link DeregisterWatchedCustomResource} while the + * primary's spec requests it, and dynamically de-registers it otherwise. Used to verify that + * de-registering a dynamically registered event source releases the underlying informer from the + * pool. + */ +@ControllerConfiguration +public class DeregisterReconciler implements Reconciler { + + public static final String WATCHED_EVENT_SOURCE_NAME = "deregister-watched-informer"; + + private final AtomicInteger numberOfExecutions = new AtomicInteger(0); + + @Override + public UpdateControl reconcile( + DeregisterPrimaryCustomResource primary, Context context) { + numberOfExecutions.incrementAndGet(); + + if (primary.getSpec() != null && primary.getSpec().isRegisterEventSource()) { + context.eventSourceRetriever().dynamicallyRegisterEventSource(watchedEventSource(context)); + } else { + context.eventSourceRetriever().dynamicallyDeRegisterEventSource(WATCHED_EVENT_SOURCE_NAME); + } + + return UpdateControl.noUpdate(); + } + + private InformerEventSource + watchedEventSource(Context context) { + var config = + InformerEventSourceConfiguration.from( + DeregisterWatchedCustomResource.class, DeregisterPrimaryCustomResource.class) + .withName(WATCHED_EVENT_SOURCE_NAME) + .withSecondaryToPrimaryMapper( + (DeregisterWatchedCustomResource watched) -> + Set.of(new ResourceID("ignored", watched.getMetadata().getNamespace()))) + .build(); + return new InformerEventSource<>( + config, context.eventSourceRetriever().eventSourceContextForDynamicRegistration()); + } + + public int getNumberOfExecutions() { + return numberOfExecutions.get(); + } +} diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/informerpool/deregister/DeregisterSharedInformerIT.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/informerpool/deregister/DeregisterSharedInformerIT.java new file mode 100644 index 0000000000..8c9a209553 --- /dev/null +++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/informerpool/deregister/DeregisterSharedInformerIT.java @@ -0,0 +1,29 @@ +/* + * Copyright Java Operator SDK Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.javaoperatorsdk.operator.baseapi.informerpool.deregister; + +import java.util.function.Consumer; + +import io.javaoperatorsdk.operator.api.config.ConfigurationServiceOverrider; + +/** Runs {@link AbstractDeregisterSharedInformerIT} with the default (sharing) informer pool. */ +class DeregisterSharedInformerIT extends AbstractDeregisterSharedInformerIT { + + @Override + protected Consumer configurationServiceOverrider() { + return overrider -> {}; + } +} diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/informerpool/deregister/DeregisterSpec.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/informerpool/deregister/DeregisterSpec.java new file mode 100644 index 0000000000..c88b0cde9d --- /dev/null +++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/informerpool/deregister/DeregisterSpec.java @@ -0,0 +1,37 @@ +/* + * Copyright Java Operator SDK Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.javaoperatorsdk.operator.baseapi.informerpool.deregister; + +/** + * Spec of {@link DeregisterPrimaryCustomResource}; toggles the dynamic event source registration. + */ +public class DeregisterSpec { + + /** + * When {@code true} the reconciler dynamically registers the event source for {@link + * DeregisterWatchedCustomResource}; when {@code false} it de-registers it. + */ + private boolean registerEventSource = true; + + public boolean isRegisterEventSource() { + return registerEventSource; + } + + public DeregisterSpec setRegisterEventSource(boolean registerEventSource) { + this.registerEventSource = registerEventSource; + return this; + } +} diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/informerpool/deregister/DeregisterWatchedCustomResource.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/informerpool/deregister/DeregisterWatchedCustomResource.java new file mode 100644 index 0000000000..9593bbd146 --- /dev/null +++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/informerpool/deregister/DeregisterWatchedCustomResource.java @@ -0,0 +1,32 @@ +/* + * Copyright Java Operator SDK Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.javaoperatorsdk.operator.baseapi.informerpool.deregister; + +import io.fabric8.kubernetes.api.model.Namespaced; +import io.fabric8.kubernetes.client.CustomResource; +import io.fabric8.kubernetes.model.annotation.Group; +import io.fabric8.kubernetes.model.annotation.ShortNames; +import io.fabric8.kubernetes.model.annotation.Version; + +/** + * The custom resource watched via a dynamically registered (and later de-registered) event source + * by {@link DeregisterReconciler}. It has no reconciler of its own. + */ +@Group("sample.javaoperatorsdk") +@Version("v1") +@ShortNames("dreg-w") +public class DeregisterWatchedCustomResource extends CustomResource + implements Namespaced {} diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/informerpool/dynamic/AbstractDynamicSharedInformerIT.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/informerpool/dynamic/AbstractDynamicSharedInformerIT.java new file mode 100644 index 0000000000..befc1dd5f2 --- /dev/null +++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/informerpool/dynamic/AbstractDynamicSharedInformerIT.java @@ -0,0 +1,138 @@ +/* + * Copyright Java Operator SDK Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.javaoperatorsdk.operator.baseapi.informerpool.dynamic; + +import java.time.Duration; +import java.util.function.Consumer; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.RegisterExtension; + +import io.fabric8.kubernetes.api.model.ObjectMetaBuilder; +import io.javaoperatorsdk.operator.api.config.ConfigurationServiceOverrider; +import io.javaoperatorsdk.operator.junit.LocallyRunOperatorExtension; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.awaitility.Awaitility.await; + +/** + * Two reconcilers watch the same "third" custom resource ({@link + * DynamicSharedInformerThirdCustomResource}) as a secondary resource, but register their informer + * event sources differently: + * + *

+ * + *

Regardless of the pool strategy, the dynamically registered event source must be triggered for + * the pre-existing third resource: on registration the framework replays the resources already in + * the running informer's cache to the newly added handler, which maps the third resource back to + * the dynamic reconciler's primary. + * + *

What differs by strategy is the number of underlying informers for the third resource: a + * sharing pool establishes a single informer used by both reconcilers, whereas a non-sharing pool + * creates one per reconciler. That expected count is left abstract; concrete subclasses pick the + * strategy via {@link #configurationServiceOverrider()}. + */ +public abstract class AbstractDynamicSharedInformerIT { + + private static final String THIRD_RESOURCE_NAME = "third1"; + + @RegisterExtension + LocallyRunOperatorExtension extension = + LocallyRunOperatorExtension.builder() + .withAdditionalCustomResourceDefinition(DynamicSharedInformerThirdCustomResource.class) + .withReconciler(new StaticSharedInformerReconciler()) + .withReconciler(new DynamicSharedInformerReconciler()) + .withConfigurationService(configurationServiceOverrider()) + .build(); + + /** The informer pool strategy under test. */ + protected abstract Consumer configurationServiceOverrider(); + + /** + * Expected number of informers for the third resource once both reconcilers watch it: {@code 1} + * when the informer is shared, {@code 2} when each reconciler gets its own. + */ + protected abstract long expectedThirdResourceInformerCount(); + + @Test + void dynamicallyRegisteredEventSourceReceivesInitialEvent() { + var staticReconciler = extension.getReconcilerOfType(StaticSharedInformerReconciler.class); + var dynamicReconciler = extension.getReconcilerOfType(DynamicSharedInformerReconciler.class); + + // The static reconciler's primary must exist so that third-resource events (which map to it) + // actually result in a reconciliation. + extension.create(primary1()); + // The third resource is created before the dynamic event source is registered, so it is a + // pre-existing resource from the perspective of the dynamically added handler. + extension.create(thirdResource()); + + // Sanity check that the informer machinery works at all: the static reconciler, whose handler + // was present from startup, is triggered by the (live) creation of the third resource. + await().untilAsserted(() -> assertThat(staticReconciler.getNumberOfExecutions()).isPositive()); + + // Creating the second primary triggers the dynamic reconciler, which registers its own event + // source for the third resource against an already-running informer. + extension.create(primary2()); + await().untilAsserted(() -> assertThat(dynamicReconciler.getNumberOfExecutions()).isPositive()); + + // (1) Informer count for the third resource, which depends on the pool strategy. + var pool = extension.getOperator().getConfigurationService().informerPool(); + await() + .untilAsserted( + () -> + assertThat( + pool.numberOfInformersForResource( + DynamicSharedInformerThirdCustomResource.class)) + .isEqualTo(expectedThirdResourceInformerCount())); + + // (2) The dynamically registered event source now watches the pre-existing third resource. On + // registration the framework replays the resources already in the running informer's cache to + // the newly added handler, which maps the third resource back to the dynamic reconciler's + // primary. This triggers a second reconciliation, in addition to the first one (the primary2 + // creation) that performed the registration. Without the replay the dynamic reconciler would + // only ever run once. + await() + .atMost(Duration.ofSeconds(15)) + .untilAsserted( + () -> assertThat(dynamicReconciler.getNumberOfExecutions()).isGreaterThanOrEqualTo(2)); + } + + DynamicSharedInformerPrimaryCustomResource1 primary1() { + var res = new DynamicSharedInformerPrimaryCustomResource1(); + res.setMetadata( + new ObjectMetaBuilder().withName(StaticSharedInformerReconciler.PRIMARY_NAME).build()); + return res; + } + + DynamicSharedInformerPrimaryCustomResource2 primary2() { + var res = new DynamicSharedInformerPrimaryCustomResource2(); + res.setMetadata( + new ObjectMetaBuilder().withName(DynamicSharedInformerReconciler.PRIMARY_NAME).build()); + return res; + } + + DynamicSharedInformerThirdCustomResource thirdResource() { + var res = new DynamicSharedInformerThirdCustomResource(); + res.setMetadata(new ObjectMetaBuilder().withName(THIRD_RESOURCE_NAME).build()); + return res; + } +} diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/informerpool/dynamic/AlwaysNewInformerPoolDynamicSharedInformerIT.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/informerpool/dynamic/AlwaysNewInformerPoolDynamicSharedInformerIT.java new file mode 100644 index 0000000000..f5802bbb4f --- /dev/null +++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/informerpool/dynamic/AlwaysNewInformerPoolDynamicSharedInformerIT.java @@ -0,0 +1,41 @@ +/* + * Copyright Java Operator SDK Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.javaoperatorsdk.operator.baseapi.informerpool.dynamic; + +import java.util.function.Consumer; + +import io.javaoperatorsdk.operator.api.config.ConfigurationServiceOverrider; +import io.javaoperatorsdk.operator.processing.event.source.informer.pool.AlwaysNewInformerPool; + +/** + * Runs {@link AbstractDynamicSharedInformerIT} with the {@link AlwaysNewInformerPool}: the static + * and dynamic reconcilers each get their own informer for the third resource. The dynamic + * reconciler is still triggered for the pre-existing third resource, because its own (newly + * created) informer replays the cache to the handler once it syncs. + */ +class AlwaysNewInformerPoolDynamicSharedInformerIT extends AbstractDynamicSharedInformerIT { + + @Override + protected Consumer configurationServiceOverrider() { + return overrider -> overrider.withInformerPool(new AlwaysNewInformerPool()); + } + + @Override + protected long expectedThirdResourceInformerCount() { + // no sharing: one informer for the static reconciler and one for the dynamic reconciler + return 2; + } +} diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/informerpool/dynamic/DynamicSharedInformerIT.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/informerpool/dynamic/DynamicSharedInformerIT.java new file mode 100644 index 0000000000..b39664a0c2 --- /dev/null +++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/informerpool/dynamic/DynamicSharedInformerIT.java @@ -0,0 +1,37 @@ +/* + * Copyright Java Operator SDK Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.javaoperatorsdk.operator.baseapi.informerpool.dynamic; + +import java.util.function.Consumer; + +import io.javaoperatorsdk.operator.api.config.ConfigurationServiceOverrider; + +/** + * Runs {@link AbstractDynamicSharedInformerIT} with the default (sharing) informer pool: the static + * and dynamic reconcilers share a single informer for the third resource. + */ +class DynamicSharedInformerIT extends AbstractDynamicSharedInformerIT { + + @Override + protected Consumer configurationServiceOverrider() { + return overrider -> {}; + } + + @Override + protected long expectedThirdResourceInformerCount() { + return 1; + } +} diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/informerpool/dynamic/DynamicSharedInformerPrimaryCustomResource1.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/informerpool/dynamic/DynamicSharedInformerPrimaryCustomResource1.java new file mode 100644 index 0000000000..7515f3b971 --- /dev/null +++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/informerpool/dynamic/DynamicSharedInformerPrimaryCustomResource1.java @@ -0,0 +1,29 @@ +/* + * Copyright Java Operator SDK Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.javaoperatorsdk.operator.baseapi.informerpool.dynamic; + +import io.fabric8.kubernetes.api.model.Namespaced; +import io.fabric8.kubernetes.client.CustomResource; +import io.fabric8.kubernetes.model.annotation.Group; +import io.fabric8.kubernetes.model.annotation.ShortNames; +import io.fabric8.kubernetes.model.annotation.Version; + +/** Primary resource of {@link StaticSharedInformerReconciler}. */ +@Group("sample.javaoperatorsdk") +@Version("v1") +@ShortNames("dsi1") +public class DynamicSharedInformerPrimaryCustomResource1 extends CustomResource + implements Namespaced {} diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/informerpool/dynamic/DynamicSharedInformerPrimaryCustomResource2.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/informerpool/dynamic/DynamicSharedInformerPrimaryCustomResource2.java new file mode 100644 index 0000000000..55b16d3ab8 --- /dev/null +++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/informerpool/dynamic/DynamicSharedInformerPrimaryCustomResource2.java @@ -0,0 +1,29 @@ +/* + * Copyright Java Operator SDK Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.javaoperatorsdk.operator.baseapi.informerpool.dynamic; + +import io.fabric8.kubernetes.api.model.Namespaced; +import io.fabric8.kubernetes.client.CustomResource; +import io.fabric8.kubernetes.model.annotation.Group; +import io.fabric8.kubernetes.model.annotation.ShortNames; +import io.fabric8.kubernetes.model.annotation.Version; + +/** Primary resource of {@link DynamicSharedInformerReconciler}. */ +@Group("sample.javaoperatorsdk") +@Version("v1") +@ShortNames("dsi2") +public class DynamicSharedInformerPrimaryCustomResource2 extends CustomResource + implements Namespaced {} diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/informerpool/dynamic/DynamicSharedInformerReconciler.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/informerpool/dynamic/DynamicSharedInformerReconciler.java new file mode 100644 index 0000000000..1779a1b43e --- /dev/null +++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/informerpool/dynamic/DynamicSharedInformerReconciler.java @@ -0,0 +1,81 @@ +/* + * Copyright Java Operator SDK Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.javaoperatorsdk.operator.baseapi.informerpool.dynamic; + +import java.util.Set; +import java.util.concurrent.atomic.AtomicInteger; + +import io.javaoperatorsdk.operator.api.config.informer.InformerEventSourceConfiguration; +import io.javaoperatorsdk.operator.api.reconciler.Context; +import io.javaoperatorsdk.operator.api.reconciler.ControllerConfiguration; +import io.javaoperatorsdk.operator.api.reconciler.Reconciler; +import io.javaoperatorsdk.operator.api.reconciler.UpdateControl; +import io.javaoperatorsdk.operator.processing.event.ResourceID; +import io.javaoperatorsdk.operator.processing.event.source.informer.InformerEventSource; + +/** + * Watches the same {@link DynamicSharedInformerThirdCustomResource} ("third" custom resource) as + * {@link StaticSharedInformerReconciler}, but registers its informer event source + * dynamically from within {@link #reconcile} instead of statically at startup. Since the + * configuration matches the static reconciler's, the pool is expected to hand out the same, + * already-running informer, so the two reconcilers share a single informer for the third resource. + * + *

The event source is registered while the shared informer is already running and has the + * pre-existing third resource in its cache. Because the handler is added to an already-running + * informer, the framework replays the resources already in that informer's cache to this newly + * added handler on registration, so this reconciler is triggered for the pre-existing third + * resource. This is asserted by the integration test. + */ +@ControllerConfiguration +public class DynamicSharedInformerReconciler + implements Reconciler { + + public static final String PRIMARY_NAME = "dynamic-primary"; + public static final String THIRD_EVENT_SOURCE_NAME = "dynamic-third-informer"; + + private final AtomicInteger numberOfExecutions = new AtomicInteger(0); + + @Override + public UpdateControl reconcile( + DynamicSharedInformerPrimaryCustomResource2 resource, + Context context) { + numberOfExecutions.incrementAndGet(); + context + .eventSourceRetriever() + .dynamicallyRegisterEventSource(thirdResourceEventSource(context)); + return UpdateControl.noUpdate(); + } + + private InformerEventSource< + DynamicSharedInformerThirdCustomResource, DynamicSharedInformerPrimaryCustomResource2> + thirdResourceEventSource(Context context) { + var config = + InformerEventSourceConfiguration.from( + DynamicSharedInformerThirdCustomResource.class, + DynamicSharedInformerPrimaryCustomResource2.class) + .withName(THIRD_EVENT_SOURCE_NAME) + .withSecondaryToPrimaryMapper( + (DynamicSharedInformerThirdCustomResource third) -> + Set.of(new ResourceID(PRIMARY_NAME, third.getMetadata().getNamespace()))) + .build(); + return new InformerEventSource<>( + config, context.eventSourceRetriever().eventSourceContextForDynamicRegistration()); + } + + public int getNumberOfExecutions() { + return numberOfExecutions.get(); + } +} diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/informerpool/dynamic/DynamicSharedInformerThirdCustomResource.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/informerpool/dynamic/DynamicSharedInformerThirdCustomResource.java new file mode 100644 index 0000000000..fb4ad62586 --- /dev/null +++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/informerpool/dynamic/DynamicSharedInformerThirdCustomResource.java @@ -0,0 +1,34 @@ +/* + * Copyright Java Operator SDK Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.javaoperatorsdk.operator.baseapi.informerpool.dynamic; + +import io.fabric8.kubernetes.api.model.Namespaced; +import io.fabric8.kubernetes.client.CustomResource; +import io.fabric8.kubernetes.model.annotation.Group; +import io.fabric8.kubernetes.model.annotation.ShortNames; +import io.fabric8.kubernetes.model.annotation.Version; + +/** + * The "third" custom resource that is watched as a secondary resource by two different reconcilers. + * It has no reconciler of its own. One reconciler watches it via a statically registered event + * source, the other via a dynamically registered one; both are expected to share a single informer + * for this type from the informer pool. + */ +@Group("sample.javaoperatorsdk") +@Version("v1") +@ShortNames("dsi3") +public class DynamicSharedInformerThirdCustomResource extends CustomResource + implements Namespaced {} diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/informerpool/dynamic/StaticSharedInformerReconciler.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/informerpool/dynamic/StaticSharedInformerReconciler.java new file mode 100644 index 0000000000..be7bce6ce6 --- /dev/null +++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/informerpool/dynamic/StaticSharedInformerReconciler.java @@ -0,0 +1,73 @@ +/* + * Copyright Java Operator SDK Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.javaoperatorsdk.operator.baseapi.informerpool.dynamic; + +import java.util.List; +import java.util.Set; +import java.util.concurrent.atomic.AtomicInteger; + +import io.javaoperatorsdk.operator.api.config.informer.InformerEventSourceConfiguration; +import io.javaoperatorsdk.operator.api.reconciler.Context; +import io.javaoperatorsdk.operator.api.reconciler.ControllerConfiguration; +import io.javaoperatorsdk.operator.api.reconciler.EventSourceContext; +import io.javaoperatorsdk.operator.api.reconciler.Reconciler; +import io.javaoperatorsdk.operator.api.reconciler.UpdateControl; +import io.javaoperatorsdk.operator.processing.event.ResourceID; +import io.javaoperatorsdk.operator.processing.event.source.EventSource; +import io.javaoperatorsdk.operator.processing.event.source.informer.InformerEventSource; + +/** + * Watches {@link DynamicSharedInformerThirdCustomResource} (the "third" custom resource) as a + * secondary resource using a statically registered informer event source. Because this + * event source is registered at startup, its handler is present on the underlying informer before + * any third resource exists, so this reconciler is triggered by third-resource events. It is the + * counterpart to {@link DynamicSharedInformerReconciler}, which watches the same third resource but + * registers its event source dynamically. + */ +@ControllerConfiguration +public class StaticSharedInformerReconciler + implements Reconciler { + + public static final String PRIMARY_NAME = "static-primary"; + + private final AtomicInteger numberOfExecutions = new AtomicInteger(0); + + @Override + public List> prepareEventSources( + EventSourceContext context) { + var config = + InformerEventSourceConfiguration.from( + DynamicSharedInformerThirdCustomResource.class, + DynamicSharedInformerPrimaryCustomResource1.class) + .withSecondaryToPrimaryMapper( + (DynamicSharedInformerThirdCustomResource third) -> + Set.of(new ResourceID(PRIMARY_NAME, third.getMetadata().getNamespace()))) + .build(); + return List.of(new InformerEventSource<>(config, context)); + } + + @Override + public UpdateControl reconcile( + DynamicSharedInformerPrimaryCustomResource1 resource, + Context context) { + numberOfExecutions.incrementAndGet(); + return UpdateControl.noUpdate(); + } + + public int getNumberOfExecutions() { + return numberOfExecutions.get(); + } +} diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/dependent/standalonedependent/StandaloneDependentResourceIT.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/dependent/standalonedependent/StandaloneDependentResourceIT.java index a44b3e13f3..c61ff90140 100644 --- a/operator-framework/src/test/java/io/javaoperatorsdk/operator/dependent/standalonedependent/StandaloneDependentResourceIT.java +++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/dependent/standalonedependent/StandaloneDependentResourceIT.java @@ -28,6 +28,7 @@ import io.javaoperatorsdk.operator.api.config.*; import io.javaoperatorsdk.operator.api.reconciler.Reconciler; import io.javaoperatorsdk.operator.junit.LocallyRunOperatorExtension; +import io.javaoperatorsdk.operator.processing.event.source.informer.pool.InformerPool; import static org.assertj.core.api.Assertions.assertThat; import static org.awaitility.Awaitility.await; @@ -127,6 +128,12 @@ public Set getKnownReconcilerNames() { public Version getVersion() { return null; } + + // only used here to obtain the resource cloner, so the pool is never accessed + @Override + public InformerPool informerPool() { + return null; + } }.getResourceCloner(); } }