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 extends HasMetadata> 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 extends HasMetadata> 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