diff --git a/bootstrapper-maven-plugin/pom.xml b/bootstrapper-maven-plugin/pom.xml index c847096839..bcd6c2e743 100644 --- a/bootstrapper-maven-plugin/pom.xml +++ b/bootstrapper-maven-plugin/pom.xml @@ -22,7 +22,7 @@ io.javaoperatorsdk java-operator-sdk - 5.4.1-SNAPSHOT + 999-SNAPSHOT bootstrapper diff --git a/caffeine-bounded-cache-support/pom.xml b/caffeine-bounded-cache-support/pom.xml index 394fa4aa3b..be70ab9a2e 100644 --- a/caffeine-bounded-cache-support/pom.xml +++ b/caffeine-bounded-cache-support/pom.xml @@ -21,7 +21,7 @@ io.javaoperatorsdk java-operator-sdk - 5.4.1-SNAPSHOT + 999-SNAPSHOT caffeine-bounded-cache-support diff --git a/micrometer-support/pom.xml b/micrometer-support/pom.xml index a691af2324..ae3c4d0be1 100644 --- a/micrometer-support/pom.xml +++ b/micrometer-support/pom.xml @@ -21,7 +21,7 @@ io.javaoperatorsdk java-operator-sdk - 5.4.1-SNAPSHOT + 999-SNAPSHOT micrometer-support diff --git a/migration/pom.xml b/migration/pom.xml index 54af000ff1..504a6a3280 100644 --- a/migration/pom.xml +++ b/migration/pom.xml @@ -21,7 +21,7 @@ io.javaoperatorsdk java-operator-sdk - 5.4.1-SNAPSHOT + 999-SNAPSHOT migration diff --git a/operator-framework-bom/pom.xml b/operator-framework-bom/pom.xml index 62b2dead1e..9b874fbbcc 100644 --- a/operator-framework-bom/pom.xml +++ b/operator-framework-bom/pom.xml @@ -21,7 +21,7 @@ io.javaoperatorsdk operator-framework-bom - 5.4.1-SNAPSHOT + 999-SNAPSHOT pom Operator SDK - Bill of Materials Java SDK for implementing Kubernetes operators diff --git a/operator-framework-core/pom.xml b/operator-framework-core/pom.xml index d446877a9a..2356433ca9 100644 --- a/operator-framework-core/pom.xml +++ b/operator-framework-core/pom.xml @@ -21,7 +21,7 @@ io.javaoperatorsdk java-operator-sdk - 5.4.1-SNAPSHOT + 999-SNAPSHOT ../pom.xml 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..975db86787 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,8 @@ 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.DefaultInformerPool; +import io.javaoperatorsdk.operator.processing.event.source.informer.pool.InformerPool; /** An interface from which to retrieve configuration information. */ public interface ConfigurationService { @@ -476,4 +478,8 @@ default boolean useSSAToPatchPrimaryResource() { default boolean cloneSecondaryResourcesWhenGettingFromCache() { return false; } + + default InformerPool informerPool() { + return new DefaultInformerPool(getKubernetesClient(), this); + } } 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..67aa0f9966 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 @@ -27,7 +27,6 @@ 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.informers.ResourceEventHandler; @@ -41,6 +40,8 @@ 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; @@ -55,6 +56,7 @@ class InformerManager> private final ResourceEventHandler eventHandler; private final Map>> indexers = new HashMap<>(); private ControllerConfiguration controllerConfiguration; + private InformerPool informerPool; InformerManager( MixedOperation, Resource> client, @@ -67,6 +69,7 @@ class InformerManager> void setControllerConfiguration(ControllerConfiguration controllerConfiguration) { this.controllerConfiguration = controllerConfiguration; + this.informerPool = controllerConfiguration.getConfigurationService().informerPool(); } @Override @@ -97,11 +100,14 @@ private void initSources() { configuration.getInformerConfig().getEffectiveNamespaces(controllerConfiguration); if (InformerConfiguration.allNamespacesWatched(targetNamespaces)) { var source = createEventSourceForNamespace(WATCH_ALL_NAMESPACES); + source.start(); log.debug("Registered {} -> {} for any namespace", this, source); } else { targetNamespaces.forEach( ns -> { + // todo should we start parallel final var source = createEventSourceForNamespace(ns); + source.start(); log.debug("Registered {} -> {} for namespace: {}", this, source, ns); }); } @@ -111,7 +117,14 @@ public void changeNamespaces(Set namespaces) { var sourcesToRemove = sources.keySet().stream().filter(k -> !namespaces.contains(k)).collect(Collectors.toSet()); log.debug("Stopped informer {} for namespaces: {}", this, sourcesToRemove); - sourcesToRemove.forEach(k -> sources.remove(k).stop()); + // todo remove event handler + // remove from pool + sourcesToRemove.forEach( + k -> { + informerPool.releaseInformer( + configuration.getInformerConfig().getName(), getClassifier(k)); + sources.remove(k); + }); var newNamespaces = namespaces.stream().filter(ns -> !sources.containsKey(ns)).collect(Collectors.toList()); @@ -126,66 +139,37 @@ public void changeNamespaces(Set namespaces) { newNamespaces.stream(), ns -> { final var source = createEventSourceForNamespace(ns); - source.start(); log.debug("Registered new {} -> {} for namespace: {}", this, source, ns); return null; }, ns -> "InformerStarter-" + ns + "-" + configuration.getResourceClass().getSimpleName()); } - private InformerWrapper createEventSourceForNamespace(String namespace) { + private InformerWrapper createEventSourceForNamespace(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); - } - source.addIndexers(indexers); - return source; - } - - private InformerWrapper createEventSource( - FilterWatchListDeletable, Resource> filteredBySelectorClient, - ResourceEventHandler eventHandler, - String namespaceIdentifier) { - final var informerConfig = configuration.getInformerConfig(); - - 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()); - } - } - } - + InformerClassifier classifier = getClassifier(namespaceIdentifier); var informer = - Optional.ofNullable(informerConfig.getInformerListLimit()) - .map(filteredBySelectorClient::withLimit) - .orElse(filteredBySelectorClient) - .runnableInformer(0); - Optional.ofNullable(informerConfig.getItemStore()).ifPresent(informer::itemStore); - var source = + informerPool.getInformer(configuration.getInformerConfig().getName(), classifier); + source = new InformerWrapper<>( - informer, controllerConfiguration.getConfigurationService(), namespaceIdentifier); + informer, namespaceIdentifier, controllerConfiguration.getConfigurationService()); source.addEventHandler(eventHandler); sources.put(namespaceIdentifier, source); + source.addIndexers(indexers); return source; } + private InformerClassifier getClassifier(String namespaceIdentifier) { + return new InformerClassifier<>( + configuration.getInformerConfig().getLabelSelector(), + configuration.getInformerConfig().getShardSelector(), + namespaceIdentifier, + configuration.getResourceClass(), + configuration.getInformerConfig().getFieldSelector(), + configuration.getInformerConfig().getInformerListLimit(), + configuration.getInformerConfig().getItemStore()); + } + @Override public void stop() { sources.forEach( 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..97b7da743e 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; @@ -28,14 +25,11 @@ 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.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; @@ -55,96 +49,17 @@ class InformerWrapper public InformerWrapper( SharedIndexInformer informer, - ConfigurationService configurationService, - String namespaceIdentifier) { + String namespaceIdentifier, + ConfigurationService configurationService) { this.informer = informer; this.namespaceIdentifier = namespaceIdentifier; this.cache = (Cache) informer.getStore(); this.configurationService = configurationService; } + // todo sync running multiple informers @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); - } + public void start() {} @Override public void stop() throws OperatorException { @@ -201,7 +116,7 @@ public String toString() { } private String informerInfo() { - return "InformerWrapper [" + versionedFullResourceName() + "]"; + return "InformerWrapper [informerInfo" + informer.getApiTypeClass().getSimpleName() + "]"; } @Override 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..3abec24fef --- /dev/null +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/pool/AbstractInformerPool.java @@ -0,0 +1,196 @@ +/* + * 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.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 KubernetesClient client; + protected ConfigurationService configurationService; + + public KubernetesClient getClient() { + return client; + } + + public void setClient(KubernetesClient client) { + this.client = client; + } + + public ConfigurationService getConfigurationService() { + return configurationService; + } + + public void setConfigurationService(ConfigurationService configurationService) { + this.configurationService = configurationService; + } + + @SuppressWarnings({"rawtypes", "unchecked"}) + protected SharedIndexInformer createInformer(InformerClassifier classifier) { + FilterWatchListDeletable filteredClient; + + if (WATCH_ALL_NAMESPACES.equals(classifier.namespaceIdentifier())) { + filteredClient = client.resources(classifier.resourceClass()).inAnyNamespace(); + + } else { + filteredClient = + client + .resources(classifier.resourceClass()) + .inNamespace(classifier.namespaceIdentifier()) + .withLabelSelector(classifier.labelSelector()) + .withShardSelector(classifier.shardSelector()); + } + + if (classifier.labelSelector() != null) { + filteredClient = + (FilterWatchListDeletable) filteredClient.withLabelSelector(classifier.labelSelector()); + } + if (classifier.shardSelector() != null) { + 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; + } + + protected 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[informerInfo" + + informer.getApiTypeClass().getSimpleName() + + "]" + + " " + + thread.getId()); + final var resourceName = informer.getApiTypeClass().getSimpleName(); + log.debug( + "Starting informer for namespace: {} resource: {}", + informerClassifier.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: {}", + 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 (apiTypeClass.isAssignableFrom(GenericKubernetesResource.class)) { + 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/AlwaysCreateInformerPool.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/pool/AlwaysCreateInformerPool.java new file mode 100644 index 0000000000..22dc48202f --- /dev/null +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/pool/AlwaysCreateInformerPool.java @@ -0,0 +1,48 @@ +/* + * 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.concurrent.ConcurrentHashMap; + +import io.fabric8.kubernetes.api.model.HasMetadata; +import io.fabric8.kubernetes.client.informers.SharedIndexInformer; + +@SuppressWarnings({"unchecked", "rawtypes"}) +public class AlwaysCreateInformerPool extends AbstractInformerPool { + + private Map informers = new ConcurrentHashMap(); + + @Override + public SharedIndexInformer getInformer( + String name, InformerClassifier classifier) { + var informer = createInformer(classifier); + start(informer, classifier); + informers.put(new ClassifierWithName(name, classifier), informer); + return informer; + } + + // todo controller name should be added here? for better naming => referencing and have guaranteed + // uniqueness + @Override + public void releaseInformer( + String name, InformerClassifier classifier) { + informers.get(new ClassifierWithName(name, classifier)).stop(); + } + + public record ClassifierWithName( + 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..fced967697 --- /dev/null +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/pool/DefaultInformerPool.java @@ -0,0 +1,72 @@ +/* + * 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.concurrent.atomic.AtomicInteger; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import io.fabric8.kubernetes.api.model.HasMetadata; +import io.fabric8.kubernetes.client.informers.SharedIndexInformer; + +public class DefaultInformerPool extends AbstractInformerPool { + + private static final Logger log = LoggerFactory.getLogger(DefaultInformerPool.class); + + // todo logging + private final Map, SharedIndexInformer> informers = new HashMap<>(); + private final Map, AtomicInteger> counters = new HashMap<>(); + + @SuppressWarnings("unchecked") + public SharedIndexInformer getInformer( + String name, InformerClassifier classifier) { + SharedIndexInformer informer = null; + synchronized (this) { + informer = (SharedIndexInformer) informers.get(classifier); + if (informer == null) { + informer = createInformer(classifier); + informers.put(classifier, informer); + counters.put(classifier, new AtomicInteger(1)); + } else { + counters.get(classifier).incrementAndGet(); + } + } + start(informer, classifier); + return informer; + } + + @SuppressWarnings("rawtypes") + public synchronized void releaseInformer( + String name, InformerClassifier classifier) { + SharedIndexInformer informer = null; + synchronized (this) { + var counter = counters.get(classifier); + if (counter != null && counter.decrementAndGet() == 0) { + informer = informers.get(classifier); + counters.remove(classifier); + informers.remove(classifier); + } else { + log.warn("No informer found in the pool."); + } + } + if (informer != null) { + informer.stop(); + } + } +} 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..c8fe98275c --- /dev/null +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/pool/InformerClassifier.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.processing.event.source.informer.pool; + +import io.fabric8.kubernetes.api.model.HasMetadata; +import io.fabric8.kubernetes.client.informers.cache.ItemStore; +import io.javaoperatorsdk.operator.api.config.informer.FieldSelector; + +// todo add this to docs +/** + * 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 labelSelector, + String shardSelector, + String namespaceIdentifier, + Class resourceClass, + FieldSelector fieldSelector, + Long informerListLimit, + ItemStore itemStore) {} 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..5a939f42b0 --- /dev/null +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/pool/InformerPool.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.processing.event.source.informer.pool; + +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 name, InformerClassifier classifier); + + void releaseInformer(String name, InformerClassifier classifier); + + void setClient(KubernetesClient client); + + void setConfigurationService(ConfigurationService configurationService); +} diff --git a/operator-framework-junit/pom.xml b/operator-framework-junit/pom.xml index 0282b515d5..aa18d5c778 100644 --- a/operator-framework-junit/pom.xml +++ b/operator-framework-junit/pom.xml @@ -21,7 +21,7 @@ io.javaoperatorsdk java-operator-sdk - 5.4.1-SNAPSHOT + 999-SNAPSHOT operator-framework-junit diff --git a/operator-framework/pom.xml b/operator-framework/pom.xml index a57c8e7c18..f94dfa757d 100644 --- a/operator-framework/pom.xml +++ b/operator-framework/pom.xml @@ -21,7 +21,7 @@ io.javaoperatorsdk java-operator-sdk - 5.4.1-SNAPSHOT + 999-SNAPSHOT operator-framework diff --git a/pom.xml b/pom.xml index 57431dde1c..ae2bb24ae7 100644 --- a/pom.xml +++ b/pom.xml @@ -21,7 +21,7 @@ io.javaoperatorsdk java-operator-sdk - 5.4.1-SNAPSHOT + 999-SNAPSHOT pom Operator SDK for Java Java SDK for implementing Kubernetes operators diff --git a/sample-operators/controller-namespace-deletion/pom.xml b/sample-operators/controller-namespace-deletion/pom.xml index 3453174aad..af4be01972 100644 --- a/sample-operators/controller-namespace-deletion/pom.xml +++ b/sample-operators/controller-namespace-deletion/pom.xml @@ -22,7 +22,7 @@ io.javaoperatorsdk sample-operators - 5.4.1-SNAPSHOT + 999-SNAPSHOT sample-controller-namespace-deletion diff --git a/sample-operators/leader-election/pom.xml b/sample-operators/leader-election/pom.xml index 42fd41b3f3..4f896485d1 100644 --- a/sample-operators/leader-election/pom.xml +++ b/sample-operators/leader-election/pom.xml @@ -22,7 +22,7 @@ io.javaoperatorsdk sample-operators - 5.4.1-SNAPSHOT + 999-SNAPSHOT sample-leader-election diff --git a/sample-operators/mysql-schema/pom.xml b/sample-operators/mysql-schema/pom.xml index 6e62ebfde8..d2872c921a 100644 --- a/sample-operators/mysql-schema/pom.xml +++ b/sample-operators/mysql-schema/pom.xml @@ -22,7 +22,7 @@ io.javaoperatorsdk sample-operators - 5.4.1-SNAPSHOT + 999-SNAPSHOT sample-mysql-schema-operator diff --git a/sample-operators/operations/pom.xml b/sample-operators/operations/pom.xml index 7940f54aaa..1786cf39d0 100644 --- a/sample-operators/operations/pom.xml +++ b/sample-operators/operations/pom.xml @@ -22,7 +22,7 @@ io.javaoperatorsdk sample-operators - 5.4.1-SNAPSHOT + 999-SNAPSHOT sample-operations diff --git a/sample-operators/pom.xml b/sample-operators/pom.xml index 2236f39543..9313095584 100644 --- a/sample-operators/pom.xml +++ b/sample-operators/pom.xml @@ -22,7 +22,7 @@ io.javaoperatorsdk java-operator-sdk - 5.4.1-SNAPSHOT + 999-SNAPSHOT sample-operators diff --git a/sample-operators/tomcat-operator/pom.xml b/sample-operators/tomcat-operator/pom.xml index bf096330f6..ea964a2b07 100644 --- a/sample-operators/tomcat-operator/pom.xml +++ b/sample-operators/tomcat-operator/pom.xml @@ -22,7 +22,7 @@ io.javaoperatorsdk sample-operators - 5.4.1-SNAPSHOT + 999-SNAPSHOT sample-tomcat-operator diff --git a/sample-operators/webpage/pom.xml b/sample-operators/webpage/pom.xml index 707fd33d8d..d50e5ef03c 100644 --- a/sample-operators/webpage/pom.xml +++ b/sample-operators/webpage/pom.xml @@ -22,7 +22,7 @@ io.javaoperatorsdk sample-operators - 5.4.1-SNAPSHOT + 999-SNAPSHOT sample-webpage-operator diff --git a/test-index-processor/pom.xml b/test-index-processor/pom.xml index c2c8e380e8..2ae7c5f454 100644 --- a/test-index-processor/pom.xml +++ b/test-index-processor/pom.xml @@ -22,7 +22,7 @@ io.javaoperatorsdk java-operator-sdk - 5.4.1-SNAPSHOT + 999-SNAPSHOT test-index-processor