diff --git a/.github/workflows/e2e-test.yml b/.github/workflows/e2e-test.yml
index 5b09cf1037..6a86b16f69 100644
--- a/.github/workflows/e2e-test.yml
+++ b/.github/workflows/e2e-test.yml
@@ -25,6 +25,7 @@ jobs:
- "sample-operators/webpage"
- "sample-operators/leader-election"
- "sample-operators/operations"
+ - "sample-operators/kotlin-operator"
runs-on: ubuntu-latest
steps:
- name: Checkout
diff --git a/README.md b/README.md
index 987750324a..be946e9090 100644
--- a/README.md
+++ b/README.md
@@ -65,19 +65,21 @@ projects want to advertise that fact here. For this reason, we ask that if you'd
to be featured in this section, please open a PR, adding a link to and short description of your
project, as shown below:
-- [kroxylicious](https://github.com/kroxylicious/kroxylicious/tree/main/kroxylicious-kubernetes/kroxylicious-operator) Kafka proxy operator
-- [ExposedApp operator](https://github.com/halkyonio/exposedapp-rhdblog): a sample operator
- written to illustrate JOSDK concepts and its Quarkus extension in the ["Write Kubernetes
- Operators in Java with the Java Operator SDK" blog series](https://developers.redhat.com/articles/2022/02/15/write-kubernetes-java-java-operator-sdk#).
-- [Keycloak operator](https://github.com/keycloak/keycloak/tree/main/operator): the official
- Keycloak operator, built with Quarkus and JOSDK.
-- [Apache Flink Kubernetes operator](https://github.com/apache/flink-kubernetes-operator) is the market leader among Flink operators.
+- [Apache Flink Kubernetes Operator](https://github.com/apache/flink-kubernetes-operator) is the market leader among Flink operators.
- [Apache Spark Kubernetes Operator](https://github.com/apache/spark-kubernetes-operator) emerging operator for Spark.
+- [kroxylicious](https://github.com/kroxylicious/kroxylicious/tree/main/kroxylicious-kubernetes/kroxylicious-operator) Kafka proxy operator
+- [Debezium Operator](https://github.com/debezium/debezium-operator): Debezium Operator adds Change-Data-Capture capabilities to your Kubernetes or OpenShift cluster by providing an easy way to run and manage [Debezium Server](https://debezium.io/documentation/reference/stable/operations/debezium-server.html) instances.
+- [Keycloak operator](https://github.com/keycloak/keycloak/tree/main/operator): the official Keycloak operator, built with Quarkus and JOSDK.
+- [Apache Hive Kubernetes Operator](https://github.com/apache/hive/tree/master/packaging/src/kubernetes/src/java/org/apache/hive/kubernetes/operator)
+- [OpenMetadata Operator](https://github.com/open-metadata/OpenMetadata/tree/main/openmetadata-k8s-operator) for AI context, data cataloging, and metadata management service
- [Strimzi Access operator](https://github.com/strimzi/kafka-access-operator). While the core Strimzi operator development predates
JOSDK, but new components like the Access operator is using the framework.
+- [Wanaku](https://github.com/wanaku-ai/wanaku/tree/main/apps/wanaku-operator) MCP Router Operator
+- [Apicurio Registry Operator](https://github.com/Apicurio/apicurio-registry/tree/main/operator)
+- [Kaap](https://github.com/datastax/kaap) autoscaling Operator for Apache Pulsar
- [EureKubeOperator](https://medium.com/@heesuk.dev/implementing-kubernetes-operator-for-eureka-service-discovery-integration-by-java-operator-sdk-d21d8087c38e): integrates service discovery of Eureka and Kubernetes using the framework - developed by 11street. It is not released as an open source yet but is very interesting to read about this problem and how it is solved by an operator written with JOSDK.
- [Locust k8s operator](https://github.com/AbdelrhmanHamouda/locust-k8s-operator): Cloud native solution to run performance tests on any Kubernetes cluster.
-- [Strimzi Schema Registry Operator](https://github.com/shangyuantech/strimzi-registry-ksql-operator): A Schema Registry Operator based on JOSDK for running the Confluent Schema Registry with a Strimzi-based Kafka cluster.
-- [Airflow Dag Operator](https://github.com/cdmikechen/airflow-dag-operator): Use JOSDK(Quarkus Extension) to replace Airflow Git Sync strategy. The main idea of the project is to start a synchronization container on each airflow pod to synchronize the DAG/files into the DAG folder.
+- [ExposedApp operator](https://github.com/halkyonio/exposedapp-rhdblog): a sample operator
+ written to illustrate JOSDK concepts and its Quarkus extension in the ["Write Kubernetes
+ Operators in Java with the Java Operator SDK" blog series](https://developers.redhat.com/articles/2022/02/15/write-kubernetes-java-java-operator-sdk#).
- [Glasskube Operator](https://github.com/glasskube/operator): simplifies the deployment, maintenance and upgrade of popular open source business tools. It is written in Kotlin and uses the JOSDK and fabric8 Kubernetes client with Kotlin-based DSL.
-- [Debezium Operator](https://github.com/debezium/debezium-operator): Debezium Operator adds Change-Data-Capture capabilities to your Kubernetes or Openshift cluster by providing an easy way to run and manage [Debezium Server](https://debezium.io/documentation/reference/stable/operations/debezium-server.html) instances.
diff --git a/bootstrapper-maven-plugin/pom.xml b/bootstrapper-maven-plugin/pom.xml
index b0c3cae39a..bcd6c2e743 100644
--- a/bootstrapper-maven-plugin/pom.xml
+++ b/bootstrapper-maven-plugin/pom.xml
@@ -22,7 +22,7 @@
io.javaoperatorsdkjava-operator-sdk
- 5.5.1-SNAPSHOT
+ 999-SNAPSHOTbootstrapper
diff --git a/caffeine-bounded-cache-support/pom.xml b/caffeine-bounded-cache-support/pom.xml
index 97ea417a47..be70ab9a2e 100644
--- a/caffeine-bounded-cache-support/pom.xml
+++ b/caffeine-bounded-cache-support/pom.xml
@@ -21,7 +21,7 @@
io.javaoperatorsdkjava-operator-sdk
- 5.5.1-SNAPSHOT
+ 999-SNAPSHOTcaffeine-bounded-cache-support
diff --git a/docs/content/en/docs/documentation/eventing.md b/docs/content/en/docs/documentation/eventing.md
index d2a104737b..340946d393 100644
--- a/docs/content/en/docs/documentation/eventing.md
+++ b/docs/content/en/docs/documentation/eventing.md
@@ -346,4 +346,70 @@ 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 `KubernetesClient` they watch through, compared by instance: normally every event source
+ resolves the operator's own client, but an event source watching another cluster brings its own
+ (see [multi-cluster](#informereventsource-multi-cluster-support)). Two separate client instances
+ never share an informer, not even when they connect to the same API server — they may differ in
+ credentials, impersonation or TLS material, and the informer keeps using the client it was created
+ from,
+- 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: they are
+registered on the shared informer under a name qualified with the controller and event source that
+added them, so index names are private to an event source and cannot collide with those of another
+one. You keep looking indexes up by the name you registered, and the indexers of an event source are
+removed from the shared informer when it stops using it.
+
+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.
+- [`NonSharingInformerPool`](https://github.com/operator-framework/java-operator-sdk/blob/main/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/pool/NonSharingInformerPool.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 NonSharingInformerPool()));
+```
+
+A custom strategy has to extend
+[`AbstractInformerPool`](https://github.com/operator-framework/java-operator-sdk/blob/main/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/pool/AbstractInformerPool.java),
+which is what `withInformerPool` accepts: it already creates the informers from an
+`InformerClassifier` and starts them, leaving the subclass to decide only whether and how they are
+shared. `InformerPool` itself is just the narrower contract that the event sources consume.
diff --git a/micrometer-support/pom.xml b/micrometer-support/pom.xml
index 733bc4528c..ae3c4d0be1 100644
--- a/micrometer-support/pom.xml
+++ b/micrometer-support/pom.xml
@@ -21,7 +21,7 @@
io.javaoperatorsdkjava-operator-sdk
- 5.5.1-SNAPSHOT
+ 999-SNAPSHOTmicrometer-support
diff --git a/migration/pom.xml b/migration/pom.xml
index d0be1d99b4..ed27afe269 100644
--- a/migration/pom.xml
+++ b/migration/pom.xml
@@ -21,7 +21,7 @@
io.javaoperatorsdkjava-operator-sdk
- 5.5.1-SNAPSHOT
+ 999-SNAPSHOTmigration
@@ -29,7 +29,7 @@
OpenRewrite migration recipes for Java Operator SDK
- 8.87.5
+ 8.88.0
diff --git a/operator-framework-bom/pom.xml b/operator-framework-bom/pom.xml
index ef9e408aec..0f974400b1 100644
--- a/operator-framework-bom/pom.xml
+++ b/operator-framework-bom/pom.xml
@@ -21,7 +21,7 @@
io.javaoperatorsdkoperator-framework-bom
- 5.5.1-SNAPSHOT
+ 999-SNAPSHOTpomOperator SDK - Bill of MaterialsJava SDK for implementing Kubernetes operators
@@ -54,7 +54,7 @@
3.2.83.4.03.12.0
- 3.8.0
+ 3.9.00.11.0
diff --git a/operator-framework-core/pom.xml b/operator-framework-core/pom.xml
index 5763c5490a..a7d06ebdc1 100644
--- a/operator-framework-core/pom.xml
+++ b/operator-framework-core/pom.xml
@@ -21,7 +21,7 @@
io.javaoperatorsdkjava-operator-sdk
- 5.5.1-SNAPSHOT
+ 999-SNAPSHOT../pom.xml
@@ -30,6 +30,11 @@
Operator SDK - Framework - CoreCore framework for implementing Kubernetes operators
+
+
+ 2.4.10
+
+
io.github.java-diff-utils
@@ -101,6 +106,13 @@
kube-api-test-client-injecttest
+
+
+ org.jetbrains.kotlin
+ kotlin-stdlib
+ ${kotlin.version}
+ test
+
@@ -147,6 +159,37 @@
+
+
+ org.jetbrains.kotlin
+ kotlin-maven-plugin
+ ${kotlin.version}
+
+ ${java.version}
+
+
+
+ kotlin-test-compile
+
+ test-compile
+
+ process-test-sources
+
+
+ ${project.basedir}/src/test/kotlin
+
+ ${project.basedir}/src/test/java
+
+
+
+
+
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..46be5c59c9 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,9 @@
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.AbstractInformerPool;
+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 +38,7 @@ public class AbstractConfigurationService implements ConfigurationService {
private KubernetesClient client;
private Cloner cloner;
private ExecutorServiceManager executorServiceManager;
+ private AbstractInformerPool informerPool;
protected AbstractConfigurationService(Version version) {
this(version, null);
@@ -190,4 +194,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..2af5604ae1 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
@@ -36,6 +36,7 @@
import io.fabric8.kubernetes.client.utils.KubernetesSerialization;
import io.javaoperatorsdk.operator.api.monitoring.Metrics;
import io.javaoperatorsdk.operator.api.reconciler.Context;
+import io.javaoperatorsdk.operator.api.reconciler.Experimental;
import io.javaoperatorsdk.operator.api.reconciler.Reconciler;
import io.javaoperatorsdk.operator.api.reconciler.dependent.DependentResourceFactory;
import io.javaoperatorsdk.operator.processing.dependent.kubernetes.KubernetesDependent;
@@ -43,6 +44,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 +479,27 @@ default boolean useSSAToPatchPrimaryResource() {
default boolean cloneSecondaryResourcesWhenGettingFromCache() {
return false;
}
+
+ /**
+ * The informer pool 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
+ */
+ @Experimental(
+ "Only the configuration API around informer pooling could still change in a"
+ + " non-backwards-compatible way, the pooling itself is prod ready.")
+ default InformerPool informerPool() {
+ var pool = new DefaultInformerPool();
+ pool.setConfigurationService(this);
+ return pool;
+ }
}
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..e07cd336f1 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
@@ -28,7 +28,10 @@
import io.fabric8.kubernetes.client.KubernetesClient;
import io.javaoperatorsdk.operator.Operator;
import io.javaoperatorsdk.operator.api.monitoring.Metrics;
+import io.javaoperatorsdk.operator.api.reconciler.Experimental;
import io.javaoperatorsdk.operator.api.reconciler.dependent.DependentResourceFactory;
+import io.javaoperatorsdk.operator.processing.event.source.informer.pool.AbstractInformerPool;
+import io.javaoperatorsdk.operator.processing.event.source.informer.pool.InformerPool;
@SuppressWarnings({"unused", "UnusedReturnValue"})
public class ConfigurationServiceOverrider {
@@ -53,6 +56,7 @@ public class ConfigurationServiceOverrider {
private Set> defaultNonSSAResource;
private Boolean useSSAToPatchPrimaryResource;
private Boolean cloneSecondaryResourcesWhenGettingFromCache;
+ private InformerPool informerPool;
@SuppressWarnings("rawtypes")
private DependentResourceFactory dependentResourceFactory;
@@ -176,6 +180,21 @@ public ConfigurationServiceOverrider withCloneSecondaryResourcesWhenGettingFromC
return this;
}
+ /**
+ * Overrides the informer pool strategy used to create/share the informers backing the event
+ * sources. When not set, the default (informer-sharing) pool is used.
+ *
+ *
Custom strategies extend {@link AbstractInformerPool}, which already takes care of creating
+ * and starting the informers.
+ */
+ @Experimental(
+ "Only the configuration API around informer pooling could still change in a"
+ + " non-backwards-compatible way, the pooling itself is prod ready.")
+ public ConfigurationServiceOverrider withInformerPool(AbstractInformerPool informerPool) {
+ this.informerPool = informerPool;
+ return this;
+ }
+
public ConfigurationService build() {
return new BaseConfigurationService(original.getVersion(), cloner, client) {
@Override
@@ -309,6 +328,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..1ee1e4e4a7 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,21 @@ 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);
+ }
+
+ @Override
+ public String toString() {
+ return "FieldSelector{" + "fields=" + 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..9fe25c999d 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,9 +93,14 @@ protected InformerConfiguration(
this.comparableResourceVersions = comparableResourceVersions;
}
- private InformerConfiguration(Class resourceClass) {
+ private InformerConfiguration(Class resourceClass, GroupVersionKind resourceGroupVersionKind) {
this.resourceClass = resourceClass;
+ this.resourceGroupVersionKind = resourceGroupVersionKind;
this.resourceTypeName =
+ // note the direction: this is true for GenericKubernetesResource, but also when the
+ // resource
+ // class is a supertype of it - i.e. a plain HasMetadata, for which no type name can be
+ // resolved from @Group/@Version annotations
resourceClass.isAssignableFrom(GenericKubernetesResource.class)
// in general this is irrelevant now for secondary resources it is used just by
// controller
@@ -101,10 +109,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 +126,7 @@ public static InformerConfiguration.Builder builder(
InformerConfiguration original) {
return new InformerConfiguration(
original.resourceClass,
+ original.resourceGroupVersionKind,
original.name,
original.namespaces,
original.followControllerNamespaceChanges,
@@ -305,6 +320,10 @@ public Long getInformerListLimit() {
return informerListLimit;
}
+ public GroupVersionKind getResourceGroupVersionKind() {
+ return resourceGroupVersionKind;
+ }
+
public FieldSelector getFieldSelector() {
return fieldSelector;
}
@@ -500,10 +519,20 @@ public Builder withInformerListLimit(Long informerListLimit) {
}
public Builder withFieldSelector(FieldSelector fieldSelector) {
- InformerConfiguration.this.fieldSelector = fieldSelector;
+ // an empty selector filters nothing, so it must not be distinguishable from having none at
+ // all: the informer pool keys on the field selector, and the annotation path always builds
+ // one (@Informer#fieldSelector defaults to {}) where the programmatic path leaves it null,
+ // which would otherwise stop the two from sharing an informer
+ InformerConfiguration.this.fieldSelector = isEmpty(fieldSelector) ? null : fieldSelector;
return this;
}
+ private static boolean isEmpty(FieldSelector fieldSelector) {
+ return fieldSelector == null
+ || fieldSelector.getFields() == null
+ || fieldSelector.getFields().isEmpty();
+ }
+
public Builder withComparableResourceVersions(boolean comparableResourceVersions) {
InformerConfiguration.this.comparableResourceVersions = comparableResourceVersions;
return this;
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/api/reconciler/ResourceOperations.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/ResourceOperations.java
index c4532aa284..603548e33d 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/ResourceOperations.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/ResourceOperations.java
@@ -751,7 +751,10 @@ public R jsonMergePatch(R desired) {
* @return the patched resource as returned by the API server
*/
public R jsonMergePatch(R desired, Options options) {
- return resourcePatch(desired, r -> context.getClient().resource(r).patch(), options);
+ return resourcePatch(
+ desired,
+ r -> context.getClient().resource(r).patch(PatchContext.of(PatchType.JSON_MERGE)),
+ options);
}
/**
@@ -767,7 +770,10 @@ public R jsonMergePatch(R desired, Options options) {
public R jsonMergePatch(
R desired, InformerEventSource informerEventSource, Options options) {
return resourcePatch(
- desired, r -> context.getClient().resource(r).patch(), informerEventSource, options);
+ desired,
+ r -> context.getClient().resource(r).patch(PatchContext.of(PatchType.JSON_MERGE)),
+ informerEventSource,
+ options);
}
/**
@@ -838,7 +844,7 @@ public P jsonMergePatchPrimary(P resource) {
public P jsonMergePatchPrimary(P resource, Options options) {
return resourcePatch(
resource,
- r -> context.getClient().resource(r).patch(),
+ r -> context.getClient().resource(r).patch(PatchContext.of(PatchType.JSON_MERGE)),
context.eventSourceRetriever().getControllerEventSource(),
options);
}
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/dependent/kubernetes/GroupVersionKindPlural.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/kubernetes/GroupVersionKindPlural.java
index a3ed4d2d97..728673ad25 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/kubernetes/GroupVersionKindPlural.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/kubernetes/GroupVersionKindPlural.java
@@ -53,10 +53,11 @@ protected GroupVersionKindPlural(GroupVersionKind gvk, String plural) {
@Override
protected boolean specificEquals(GroupVersionKind that) {
- if (plural == null) {
- return true;
- }
- return that instanceof GroupVersionKindPlural gvkp && gvkp.plural.equals(plural);
+ // a GroupVersionKind that is not plural-aware carries no plural form, which is the same as an
+ // unspecified one: that keeps this consistent with hashCode(), which only mixes the plural in
+ // when it is present
+ final var thatPlural = that instanceof GroupVersionKindPlural gvkp ? gvkp.plural : null;
+ return Objects.equals(plural, thatPlural);
}
@Override
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..b271a385e0 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");
@@ -250,7 +250,9 @@ public EventSource dynamicallyRegisterEventSource(EventSource ev
}
}
// 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.
+ // actual event source. Think of this as a form of lock striping. Note that two event sources
+ // backed by the same pooled informer may reach this concurrently; starting an already started
+ // informer is a no-op.
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..3908bbcf09 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,51 @@
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;
+ // the identity of the event source these informers are managed for, towards the pool and towards
+ // the index names on a shared informer. Deliberately the event source's own name rather than
+ // InformerConfiguration#getName, which is null unless the event source was explicitly named
+ private final String eventSourceName;
private final Map>> indexers = new HashMap<>();
private ControllerConfiguration controllerConfiguration;
+ private InformerPool informerPool;
+ private KubernetesClient targetClient;
- InformerManager(
- MixedOperation, Resource> client,
- C configuration,
- ResourceEventHandler eventHandler) {
- this.client = client;
+ InformerManager(C configuration, ResourceEventHandler eventHandler, String eventSourceName) {
this.configuration = configuration;
this.eventHandler = eventHandler;
+ this.eventSourceName = eventSourceName;
}
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 +79,8 @@ public void start() throws OperatorException {
.getExecutorServiceManager()
.boundedExecuteAndWaitForAllToComplete(
sources.values().stream(),
- iw -> {
- iw.start();
+ wrapper -> {
+ start(wrapper);
return null;
},
iw ->
@@ -96,25 +97,26 @@ 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);
});
}
}
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());
-
- var newNamespaces =
- namespaces.stream().filter(ns -> !sources.containsKey(ns)).collect(Collectors.toList());
+ var namespacesToRemove =
+ sources.keySet().stream()
+ .filter(ns -> !namespaces.contains(ns))
+ .collect(Collectors.toSet());
+ log.debug("Stopped informer {} for namespaces: {}", this, namespacesToRemove);
+ namespacesToRemove.forEach(this::releaseSource);
+
+ var newNamespaces = namespaces.stream().filter(ns -> !sources.containsKey(ns)).toList();
if (newNamespaces.isEmpty()) {
return;
}
@@ -125,79 +127,102 @@ 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(), eventSourceName, classifier);
+ source =
+ new InformerWrapper<>(
+ informer,
+ namespaceIdentifier,
+ classifier,
+ controllerConfiguration.getName(),
+ eventSourceName);
+ sources.put(namespaceIdentifier, source);
source.addIndexers(indexers);
+ source.addEventHandler(eventHandler);
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,
+ 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() {
+ // resolved once: the client is part of the informer classifier's identity, so every classifier
+ // this manager builds (one per watched namespace, and more when namespaces change later on) has
+ // to see the very same instance. ConfigurationService#getKubernetesClient is expected to return
+ // a stable instance, but its default implementation does create a new client on every call.
+ if (targetClient == null) {
+ 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) -> {
- try {
- log.debug("Stopping informer for namespace: {} -> {}", ns, source);
- source.stop();
- } catch (Exception e) {
- log.warn("Error stopping informer for namespace: {} -> {}", ns, source, e);
- }
- });
- sources.clear();
+ sources
+ .keySet()
+ .forEach(
+ ns -> {
+ try {
+ log.debug("Stopping informer for namespace: {}", ns);
+ releaseSource(ns);
+ } catch (Exception e) {
+ log.warn("Error stopping informer for namespace: {}", ns, e);
+ }
+ });
+ }
+
+ /**
+ * Gives the informer backing the given namespace back to the pool, but only if this manager still
+ * holds it: removing it from {@link #sources} is what claims the right to release it. {@link
+ * #stop()} and {@link #changeNamespaces(Set)} can run concurrently, and since the pool
+ * reference-counts its informers, releasing the same namespace twice would consume a reference
+ * another controller still holds and make the pool stop an informer that is still in use.
+ */
+ private void releaseSource(String namespaceIdentifier) {
+ var wrapper = sources.remove(namespaceIdentifier);
+ if (wrapper == null) {
+ return;
+ }
+ // the informer may be shared, in which case it keeps running and would otherwise hold on to
+ // this event source's indexers
+ wrapper.removeIndexers();
+ informerPool
+ .releaseInformer(
+ controllerConfiguration.getName(), eventSourceName, wrapper.getClassifier())
+ .ifPresent(i -> i.removeEventHandler(eventHandler));
}
@Override
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..9548e8c540 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
@@ -15,12 +15,12 @@
*/
package io.javaoperatorsdk.operator.processing.event.source.informer;
+import java.util.HashMap;
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.Set;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Function;
import java.util.function.Predicate;
import java.util.stream.Stream;
@@ -30,125 +30,39 @@
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;
+ private final String indexNamePrefix;
+ private final Set registeredIndexNames = ConcurrentHashMap.newKeySet();
public InformerWrapper(
SharedIndexInformer informer,
- ConfigurationService configurationService,
- String namespaceIdentifier) {
+ String namespaceIdentifier,
+ InformerClassifier classifier,
+ String controllerName,
+ String eventSourceName) {
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;
+ this.indexNamePrefix = "josdk/" + controllerName + "/" + eventSourceName + "/";
}
@Override
@@ -187,12 +101,42 @@ public void addEventHandler(ResourceEventHandler eventHandler) {
@Override
public void addIndexers(Map>> indexers) {
- informer.getIndexer().addIndexers(indexers);
+ Map>> qualified = new HashMap<>();
+ indexers.forEach((name, indexer) -> qualified.put(qualify(name), indexer));
+ informer.getIndexer().addIndexers(qualified);
+ registeredIndexNames.addAll(qualified.keySet());
+ }
+
+ /**
+ * Removes the indexers this event source added, to be called when its informer is released. A
+ * shared informer outlives the event sources that stop using it, so without this its indexer
+ * would keep both the index and the (possibly capturing) index function of every event source
+ * that ever used it, and re-registering the same event source later would be rejected as a name
+ * conflict.
+ */
+ void removeIndexers() {
+ registeredIndexNames.forEach(name -> informer.getIndexer().removeIndexer(name));
+ registeredIndexNames.clear();
}
@Override
public List byIndex(String indexName, String indexKey) {
- return informer.getIndexer().byIndex(indexName, indexKey);
+ return informer.getIndexer().byIndex(qualify(indexName), indexKey);
+ }
+
+ /**
+ * The informer can be shared by event sources of several controllers, while its indexer is a
+ * single namespace of index names: two event sources registering the same index name on it would
+ * be rejected by the client, and one could read the other's index. Names are therefore qualified
+ * with the event source that registered them.
+ *
+ *
This stays invisible to callers, who keep using their own names, but only for as long as
+ * this class remains the only place that talks to {@link SharedIndexInformer#getIndexer()}:
+ * adding, reading and removing all have to go through here so that the qualification stays
+ * symmetric.
+ */
+ private String qualify(String indexName) {
+ return indexNamePrefix + indexName;
}
@Override
@@ -201,7 +145,15 @@ public String toString() {
}
private String informerInfo() {
- return "InformerWrapper [" + versionedFullResourceName() + "]";
+ return "InformerWrapper [ " + versionedFullResourceName() + " ]";
+ }
+
+ private String versionedFullResourceName() {
+ final var apiTypeClass = informer.getApiTypeClass();
+ if (GenericKubernetesResource.class.isAssignableFrom(apiTypeClass)) {
+ return GenericKubernetesResource.class.getSimpleName();
+ }
+ return ReconcilerUtilsInternal.getResourceTypeNameWithVersion(apiTypeClass);
}
@Override
@@ -237,4 +189,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..5aecf34eb0 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;
}
@@ -85,10 +81,14 @@ protected InformerManager manager() {
}
@Override
- public void changeNamespaces(Set namespaces) {
- if (allowsNamespaceChanges()) {
- manager().changeNamespaces(namespaces);
+ public synchronized void changeNamespaces(Set namespaces) {
+ // a stopped event source has released its informers and its manager holds no sources, so every
+ // requested namespace would look new: it would acquire and start pooled informers that nothing
+ // can ever release, since stop() short-circuits on a non-running event source
+ if (!isRunning() || !allowsNamespaceChanges()) {
+ return;
}
+ manager().changeNamespaces(namespaces);
}
/**
@@ -159,17 +159,31 @@ 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, name());
cache.setControllerConfiguration(controllerConfiguration);
cache.addIndexers(indexers);
- manager().start();
+ // 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.
+ try {
+ manager().start();
+ } catch (RuntimeException e) {
+ // The manager acquires a pooled informer for every watched namespace before any of them is
+ // started, so a startup failure has to hand those references back here: super.start() is not
+ // reached, which leaves isRunning() false and makes stop() skip the release entirely. The
+ // pooled informer would then be referenced forever (never stopped, even on a clean shutdown)
+ // and a retried start() would acquire it a second time.
+ manager().stop();
+ throw e;
+ }
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..81f8f979c1
--- /dev/null
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/pool/AbstractInformerPool.java
@@ -0,0 +1,201 @@
+/*
+ * 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.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 io.javaoperatorsdk.operator.api.reconciler.Experimental;
+
+import static io.javaoperatorsdk.operator.api.reconciler.Constants.WATCH_ALL_NAMESPACES;
+import static io.javaoperatorsdk.operator.api.reconciler.Experimental.API_MIGHT_CHANGE;
+
+/**
+ * Base class for the informer pool strategies, and the type the configuration API accepts (see
+ * {@link io.javaoperatorsdk.operator.api.config.ConfigurationServiceOverrider#withInformerPool}),
+ * so custom strategies are expected to extend this rather than to implement {@link InformerPool}
+ * directly.
+ *
+ *
Creating an informer from an {@link InformerClassifier}, starting it and waiting for its cache
+ * to sync, and holding on to the injected {@link ConfigurationService} are handled here. Subclasses
+ * are left with the actual strategy: whether an informer is handed out to more than one event
+ * source and, consequently, when it is stopped.
+ */
+@Experimental(API_MIGHT_CHANGE)
+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;
+ }
+
+ /**
+ * 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.
+ */
+ public abstract long numberOfInformersForResource(Class extends HasMetadata> resourceClass);
+
+ @SuppressWarnings({"rawtypes", "unchecked"})
+ protected SharedIndexInformer createInformer(InformerClassifier> classifier) {
+ var client = classifier.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();
+ 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 (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/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..4440cd17f9
--- /dev/null
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/pool/DefaultInformerPool.java
@@ -0,0 +1,133 @@
+/*
+ * 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.informers.SharedIndexInformer;
+
+public class DefaultInformerPool extends AbstractInformerPool {
+
+ private static final Logger log = LoggerFactory.getLogger(DefaultInformerPool.class);
+
+ /** A pooled informer together with the number of event sources currently sharing it. */
+ private record PooledInformer(SharedIndexInformer> informer, AtomicInteger referenceCount) {}
+
+ private final Map, PooledInformer> informers = new HashMap<>();
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public SharedIndexInformer getInformer(
+ String controllerName, String name, InformerClassifier classifier) {
+ SharedIndexInformer informer;
+ synchronized (this) {
+ var pooled = informers.get(classifier);
+ if (pooled == null) {
+ informer = createInformer(classifier);
+ informers.put(classifier, new PooledInformer(informer, new AtomicInteger(1)));
+ log.debug(
+ "Created new pooled informer for classifier: {}. Requested by controller: {}, event"
+ + " source: {}",
+ classifier,
+ controllerName,
+ name);
+ } else {
+ informer = (SharedIndexInformer) pooled.informer();
+ 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 = pooled.referenceCount().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 pooled = informers.get(classifier);
+ if (pooled == null) {
+ log.warn("No informer found in the pool for classifier: {}", classifier);
+ return Optional.empty();
+ }
+ var informer = (SharedIndexInformer) pooled.informer();
+ // 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 = pooled.referenceCount().decrementAndGet();
+ if (referenceCount == 0) {
+ 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. */
+ 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 extends HasMetadata> 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..e4023a93e9
--- /dev/null
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/pool/InformerClassifier.java
@@ -0,0 +1,143 @@
+/*
+ * 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.KubernetesClient;
+import io.fabric8.kubernetes.client.informers.cache.ItemStore;
+import io.javaoperatorsdk.operator.api.config.informer.FieldSelector;
+import io.javaoperatorsdk.operator.processing.GroupVersionKind;
+
+/**
+ * Identifies the informer that backs an event source: two event sources whose classifiers are equal
+ * can be served by one shared informer. It also carries everything needed to create that informer,
+ * including the {@link KubernetesClient} to create it from.
+ *
+ *
Note that {@link #equals(Object)} and {@link #hashCode()} deliberately do not
+ * cover every record component:
+ *
+ *
+ *
{@link #informerListLimit()} is excluded, so event sources that only disagree on the list
+ * limit still share an informer; the limit of whichever classifier created the informer is
+ * kept (a pool is expected to warn about this, see {@link
+ * #differsOnlyByInformerListLimit(InformerClassifier)}).
+ *
Indexers are not part of the classifier at all: they are registered on the informer under a
+ * name qualified with the event source that added them, so those of different event sources
+ * can live side by side on a shared informer without colliding.
+ *
+ *
+ *
The {@link #client()} takes part in equality by identity: event sources
+ * sharing an informer must be watching through the very same client, since the informer is created
+ * from (and keeps using) the client of whichever event source established it. Two separate clients
+ * are therefore never assumed to be interchangeable, not even when they connect to the same API
+ * server — they may well differ in credentials, impersonation or TLS material, and the pool cannot
+ * tell.
+ *
+ *
Note that this is also why nothing security relevant from the client's configuration is part
+ * of the classifier: instances end up in log messages and exception messages, so a credential held
+ * here would leak into those.
+ */
+public record InformerClassifier(
+ KubernetesClient client,
+ 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 client == that.client
+ && 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(
+ System.identityHashCode(client),
+ labelSelector,
+ shardSelector,
+ namespaceIdentifier,
+ resourceClass,
+ groupVersionKind,
+ fieldSelector,
+ itemStore);
+ }
+
+ /**
+ * Hand written instead of using the one generated for the record, so that the API server URL is
+ * part of it: classifiers show up in log and exception messages, where the client on its own
+ * identifies the instance but not the cluster it connects to. The URL is derived from the {@link
+ * #client()} rather than held as a component of its own, since it would be redundant for the
+ * identity and could only ever contradict the client.
+ */
+ @Override
+ public String toString() {
+ return "InformerClassifier[client="
+ + client
+ + " ("
+ + masterUrl()
+ + "), labelSelector="
+ + labelSelector
+ + ", shardSelector="
+ + shardSelector
+ + ", namespaceIdentifier="
+ + namespaceIdentifier
+ + ", resourceClass="
+ + (resourceClass != null ? resourceClass.getName() : null)
+ + ", groupVersionKind="
+ + groupVersionKind
+ + ", fieldSelector="
+ + fieldSelector
+ + ", informerListLimit="
+ + informerListLimit
+ + ", itemStore="
+ + itemStore
+ + "]";
+ }
+
+ private String masterUrl() {
+ if (client == null || client.getConfiguration() == null) {
+ return null;
+ }
+ return client.getConfiguration().getMasterUrl();
+ }
+
+ /**
+ * 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..404f14b5f4
--- /dev/null
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/pool/InformerPool.java
@@ -0,0 +1,98 @@
+/*
+ * 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.informers.SharedIndexInformer;
+import io.javaoperatorsdk.operator.api.config.ConfigurationService;
+import io.javaoperatorsdk.operator.api.reconciler.Experimental;
+
+/**
+ * The contract consumed by the event sources. Implementations must extend {@link
+ * AbstractInformerPool} — that is the type the configuration API accepts — which additionally
+ * handles informer creation, startup and the {@link ConfigurationService} injection.
+ */
+@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 {
+
+ /**
+ * The informer backing the event source identified by {@code controllerName} and {@code name}: a
+ * sharing pool returns the existing informer for an equal {@link InformerClassifier} if there is
+ * one and creates it otherwise, a non-sharing pool always creates a dedicated one. A newly
+ * created informer is created from the classifier's {@link InformerClassifier#client()}, which is
+ * part of the classifier's identity precisely so that a shared informer is only ever handed to
+ * event sources watching through that same client.
+ *
+ *
The returned informer is not started, callers are expected to call {@link
+ * #start(SharedIndexInformer, InformerClassifier)} afterwards. When joining an already running
+ * shared informer it may however be started and hold a populated cache already; handlers
+ * registered on it still receive the cache contents, so callers must not replay those themselves.
+ *
+ *
This registers the caller as a user of the informer and must therefore be paired with
+ * exactly one {@link #releaseInformer(String, String, InformerClassifier)} for the same
+ * controller name, event source name and classifier. Requesting an informer twice for the same
+ * combination without releasing it in between is a programming error: a sharing pool would count
+ * the caller twice and consequently never stop the informer, which is why {@link
+ * NonSharingInformerPool} rejects it outright.
+ */
+ SharedIndexInformer getInformer(
+ String controllerName, String name, InformerClassifier classifier);
+
+ /**
+ * 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)} returns; the pool
+ * itself only registers/reference-counts the informer and does not block on cache sync
+ * internally.
+ */
+ void start(
+ SharedIndexInformer informer, InformerClassifier classifier);
+
+ /**
+ * Signals that the identified user (controller + event source name) no longer needs the informer
+ * for the given classifier. A sharing pool only stops the informer once its last user has
+ * released it, a non-sharing pool stops it right away.
+ *
+ *
The informer is returned in either case, even when it is left running for the remaining
+ * users, since the caller still has to remove its own event handler from it. Callers must not
+ * assume the returned informer is stopped, and must not stop it themselves.
+ *
+ * @return the released informer, or empty if the pool holds none for this user and classifier
+ */
+ Optional> releaseInformer(
+ String controllerName, String name, InformerClassifier classifier);
+
+ /**
+ * Binds this pool to the {@link ConfigurationService} it belongs to. Called by the framework when
+ * the pool is resolved from that configuration service, before the pool is used; users are not
+ * expected to call it themselves.
+ *
+ *
The pool needs the configuration service to create and start informers: the {@link
+ * ConfigurationService#cacheSyncTimeout()} to wait for, whether to {@link
+ * ConfigurationService#stopOnInformerErrorDuringStartup()}, and the {@link
+ * ConfigurationService#getInformerStoppedHandler()} to hook up.
+ *
+ *
Injecting it here, rather than requiring it as a constructor argument, is what keeps
+ * creating a pool a plain {@code new NonSharingInformerPool()} for users configuring one through
+ * {@link io.javaoperatorsdk.operator.api.config.ConfigurationServiceOverrider#withInformerPool}.
+ * A pool instance therefore belongs to exactly one configuration service.
+ */
+ void setConfigurationService(ConfigurationService configurationService);
+}
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/pool/NonSharingInformerPool.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/pool/NonSharingInformerPool.java
new file mode 100644
index 0000000000..da7ac06ab6
--- /dev/null
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/pool/NonSharingInformerPool.java
@@ -0,0 +1,88 @@
+/*
+ * 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.informers.SharedIndexInformer;
+import io.javaoperatorsdk.operator.OperatorException;
+
+@SuppressWarnings({"unchecked", "rawtypes"})
+public class NonSharingInformerPool extends AbstractInformerPool {
+
+ private static final Logger log = LoggerFactory.getLogger(NonSharingInformerPool.class);
+
+ private final Map informers = new ConcurrentHashMap();
+
+ @Override
+ public synchronized SharedIndexInformer getInformer(
+ String controllerName, String name, InformerClassifier classifier) {
+ 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);
+ 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). */
+ 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 extends HasMetadata> 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/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/api/config/InformerConfigurationTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/InformerConfigurationTest.java
index 95b8465706..16e5ab578b 100644
--- a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/InformerConfigurationTest.java
+++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/InformerConfigurationTest.java
@@ -16,11 +16,13 @@
package io.javaoperatorsdk.operator.api.config;
import java.util.Collections;
+import java.util.List;
import java.util.Set;
import org.junit.jupiter.api.Test;
import io.fabric8.kubernetes.api.model.ConfigMap;
+import io.javaoperatorsdk.operator.api.config.informer.FieldSelector;
import io.javaoperatorsdk.operator.api.config.informer.InformerConfiguration;
import io.javaoperatorsdk.operator.api.reconciler.Constants;
@@ -79,6 +81,37 @@ void nullShardSelectorByDefault() {
assertNull(informerConfig.getShardSelector());
}
+ @Test
+ void nullFieldSelectorByDefault() {
+ final var informerConfig = InformerConfiguration.builder(ConfigMap.class).build();
+ assertNull(informerConfig.getFieldSelector());
+ }
+
+ @Test
+ void emptyFieldSelectorIsNormalizedToNoFieldSelector() {
+ // the annotation path always builds a FieldSelector (@Informer#fieldSelector defaults to {})
+ // while the programmatic path leaves it null. An empty selector filters nothing, so the two
+ // must not get classifiers that disagree and therefore refuse to share an informer
+ assertNull(
+ InformerConfiguration.builder(ConfigMap.class)
+ .withFieldSelector(new FieldSelector(List.of()))
+ .build()
+ .getFieldSelector());
+ assertNull(
+ InformerConfiguration.builder(ConfigMap.class)
+ .withFieldSelector(new FieldSelector())
+ .build()
+ .getFieldSelector());
+ }
+
+ @Test
+ void fieldSelectorIsSetOnBuilderWhenNotEmpty() {
+ final var fieldSelector = new FieldSelector(new FieldSelector.Field("metadata.name", "foo"));
+ final var informerConfig =
+ InformerConfiguration.builder(ConfigMap.class).withFieldSelector(fieldSelector).build();
+ assertEquals(fieldSelector, informerConfig.getFieldSelector());
+ }
+
@Test
void shardSelectorIsSetOnBuilder() {
final var informerConfig =
diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/reconciler/ResourceOperationsTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/reconciler/ResourceOperationsTest.java
index f5b010d21c..95c1b1815e 100644
--- a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/reconciler/ResourceOperationsTest.java
+++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/reconciler/ResourceOperationsTest.java
@@ -22,6 +22,7 @@
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
import org.mockito.stubbing.Answer;
import io.fabric8.kubernetes.api.model.HasMetadata;
@@ -31,6 +32,7 @@
import io.fabric8.kubernetes.client.dsl.NamespaceableResource;
import io.fabric8.kubernetes.client.dsl.Resource;
import io.fabric8.kubernetes.client.dsl.base.PatchContext;
+import io.fabric8.kubernetes.client.dsl.base.PatchType;
import io.fabric8.kubernetes.client.utils.KubernetesSerialization;
import io.javaoperatorsdk.operator.TestUtils;
import io.javaoperatorsdk.operator.api.config.Cloner;
@@ -40,6 +42,7 @@
import io.javaoperatorsdk.operator.processing.event.EventSourceRetriever;
import io.javaoperatorsdk.operator.processing.event.source.EventSource;
import io.javaoperatorsdk.operator.processing.event.source.controller.ControllerEventSource;
+import io.javaoperatorsdk.operator.processing.event.source.informer.InformerEventSource;
import io.javaoperatorsdk.operator.processing.event.source.informer.ManagedInformerEventSource;
import io.javaoperatorsdk.operator.sample.simple.TestCustomResource;
@@ -471,15 +474,20 @@ private TestCustomResource wireVerbMocks() {
when(verbClientResource.patch(any(PatchContext.class))).thenReturn(updated);
// both cache paths execute the update operation so the underlying client verb runs
- Answer