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/dependent-resource-and-workflows/dependent-resources.md b/docs/content/en/docs/documentation/dependent-resource-and-workflows/dependent-resources.md
index 2b3e317baa..8974c41f2a 100644
--- a/docs/content/en/docs/documentation/dependent-resource-and-workflows/dependent-resources.md
+++ b/docs/content/en/docs/documentation/dependent-resource-and-workflows/dependent-resources.md
@@ -289,6 +289,41 @@ If you encounter this issue on an older Kubernetes version, consider changing yo
that resource, or even upgrading your Kubernetes version. If you encounter it on a newer Kubernetes version, please log
an issue with the JOSDK and with upstream Kubernetes.
+### Detecting dependent resource API version changes (experimental)
+
+When a dependent resource's CRD gains a new API version and the operator is upgraded to target it,
+comparing `actualResource.getApiVersion()` with the desired resource's API version is not a
+reliable way to detect resources that still need to be updated: the Kubernetes API server serves a
+resource using the requested, served API version regardless of which version it is actually stored
+as, so this comparison would always trivially match.
+
+`KubernetesDependentResource` therefore ignores `apiVersion` when matching. To still force a
+one-time update of dependent resources after such an upgrade, without triggering an update on every
+reconciliation, `KubernetesDependent` provides the opt-in, experimental
+`detectApiVersionChange` flag:
+
+```java
+@KubernetesDependent(detectApiVersionChange = true)
+public class MyDependentResource extends CRUDKubernetesDependentResource {
+ // ...
+}
+```
+
+When enabled, JOSDK records the API version it applies in the `javaoperatorsdk.io/last-applied-api-version`
+annotation. On subsequent reconciliations, the resource is considered mismatched (and thus updated)
+if that recorded marker differs from the API version the operator currently uses - this also
+covers resources that predate this feature and therefore have no marker at all. Once the resource
+has been updated, the marker matches the current API version again, so no further update is
+requested until the API version changes again.
+
+This is disabled by default: existing behavior, including for resources created before this
+feature existed, is unaffected unless you opt in. It does not read or infer the actual storage
+version of the resource from the Kubernetes API, since that information is not reliably exposed;
+it only tracks what the operator itself last applied. It is also not a replacement for
+Kubernetes' [StorageVersionMigration](https://kubernetes.io/docs/tasks/manage-kubernetes-objects/storage-version-migration/),
+which addresses migrating the stored representation of resources, a concern orthogonal to this
+feature.
+
## Telling JOSDK how to find which secondary resources are associated with a given primary resource
[`KubernetesDependentResource`](https://github.com/java-operator-sdk/java-operator-sdk/blob/main/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/kubernetes/KubernetesDependentResource.java)
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/Informable.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/Informable.java
index 12c6b4fe06..5175efb898 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/Informable.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/Informable.java
@@ -15,7 +15,10 @@
*/
package io.javaoperatorsdk.operator.api.config;
+import java.util.Optional;
+
import io.fabric8.kubernetes.api.model.HasMetadata;
+import io.fabric8.kubernetes.client.KubernetesClient;
import io.javaoperatorsdk.operator.api.config.informer.InformerConfiguration;
public interface Informable {
@@ -29,4 +32,12 @@ default String getResourceTypeName() {
default Class getResourceClass() {
return getInformerConfig().getResourceClass();
}
+
+ /**
+ * Optional, specific kubernetes client, typically to connect to a different cluster than the rest
+ * of the operator. Note that this is solely for multi cluster support.
+ */
+ default Optional getKubernetesClient() {
+ return Optional.empty();
+ }
}
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..ae2b12fe16 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,20 +76,13 @@ default boolean followControllerNamespaceChanges() {
PrimaryToSecondaryMapper
getPrimaryToSecondaryMapper();
+ // todo deprecate
Optional getGroupVersionKind();
default String name() {
return getInformerConfig().getName();
}
- /**
- * Optional, specific kubernetes client, typically to connect to a different cluster than the rest
- * of the operator. Note that this is solely for multi cluster support.
- */
- default Optional getKubernetesClient() {
- return Optional.empty();
- }
-
class DefaultInformerEventSourceConfiguration
implements InformerEventSourceConfiguration {
private final PrimaryToSecondaryMapper> primaryToSecondaryMapper;
@@ -167,7 +160,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/PrimaryUpdateAndCacheUtils.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/PrimaryUpdateAndCacheUtils.java
index f74cd49ee7..1be8e3f09e 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/PrimaryUpdateAndCacheUtils.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/PrimaryUpdateAndCacheUtils.java
@@ -25,12 +25,12 @@
import org.slf4j.LoggerFactory;
import io.fabric8.kubernetes.api.model.HasMetadata;
-import io.fabric8.kubernetes.api.model.ObjectMeta;
import io.fabric8.kubernetes.client.KubernetesClient;
import io.fabric8.kubernetes.client.KubernetesClientException;
import io.fabric8.kubernetes.client.dsl.base.PatchContext;
import io.fabric8.kubernetes.client.dsl.base.PatchType;
import io.javaoperatorsdk.operator.OperatorException;
+import io.javaoperatorsdk.operator.ReconcilerUtilsInternal;
import io.javaoperatorsdk.operator.processing.event.ResourceID;
import static io.javaoperatorsdk.operator.processing.KubernetesResourceUtils.getUID;
@@ -430,10 +430,7 @@ public static
P addFinalizerWithSSA(
}
try {
P resource = (P) originalResource.getClass().getConstructor().newInstance();
- ObjectMeta objectMeta = new ObjectMeta();
- objectMeta.setName(originalResource.getMetadata().getName());
- objectMeta.setNamespace(originalResource.getMetadata().getNamespace());
- resource.setMetadata(objectMeta);
+ resource.initNameAndNamespaceFrom(originalResource);
resource.addFinalizer(finalizerName);
return client
.resource(resource)
@@ -456,43 +453,10 @@ public static
P addFinalizerWithSSA(
}
public static int compareResourceVersions(HasMetadata h1, HasMetadata h2) {
- return compareResourceVersions(
- h1.getMetadata().getResourceVersion(), h2.getMetadata().getResourceVersion());
+ return ReconcilerUtilsInternal.validateAndCompareResourceVersions(h1, h2);
}
public static int compareResourceVersions(String v1, String v2) {
- int v1Length = validateResourceVersion(v1);
- int v2Length = validateResourceVersion(v2);
- int comparison = v1Length - v2Length;
- if (comparison != 0) {
- return comparison;
- }
- for (int i = 0; i < v2Length; i++) {
- int comp = v1.charAt(i) - v2.charAt(i);
- if (comp != 0) {
- return comp;
- }
- }
- return 0;
- }
-
- private static int validateResourceVersion(String v1) {
- int v1Length = v1.length();
- if (v1Length == 0) {
- throw new NonComparableResourceVersionException("Resource version is empty");
- }
- for (int i = 0; i < v1Length; i++) {
- char char1 = v1.charAt(i);
- if (char1 == '0') {
- if (i == 0) {
- throw new NonComparableResourceVersionException(
- "Resource version cannot begin with 0: " + v1);
- }
- } else if (char1 < '0' || char1 > '9') {
- throw new NonComparableResourceVersionException(
- "Non numeric characters in resource version: " + v1);
- }
- }
- return v1Length;
+ return ReconcilerUtilsInternal.validateAndCompareResourceVersions(v1, v2);
}
}
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..4c8ff8de9c 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
@@ -556,7 +556,7 @@ public R jsonPatch(R actualResource, UnaryOperator un
*/
public R jsonPatch(
R actualResource, UnaryOperator unaryOperator, Options options) {
- R desired = desiredForJsonPatch(actualResource, unaryOperator, options);
+ R desired = desiredForJsonPatch(actualResource, unaryOperator);
return resourcePatch(
desired,
actualResource,
@@ -580,7 +580,7 @@ public R jsonPatch(
UnaryOperator unaryOperator,
InformerEventSource informerEventSource,
Options options) {
- R desired = desiredForJsonPatch(actualResource, unaryOperator, options);
+ R desired = desiredForJsonPatch(actualResource, unaryOperator);
return resourcePatch(
desired,
actualResource,
@@ -620,7 +620,7 @@ public R jsonPatchStatus(
*/
public R jsonPatchStatus(
R actualResource, UnaryOperator unaryOperator, Options options) {
- R desired = desiredForJsonPatch(actualResource, unaryOperator, options);
+ R desired = desiredForJsonPatch(actualResource, unaryOperator);
return resourcePatch(
desired,
actualResource,
@@ -645,7 +645,7 @@ public R jsonPatchStatus(
UnaryOperator unaryOperator,
InformerEventSource informerEventSource,
Options options) {
- R desired = desiredForJsonPatch(actualResource, unaryOperator, options);
+ R desired = desiredForJsonPatch(actualResource, unaryOperator);
return resourcePatch(
desired,
actualResource,
@@ -680,7 +680,7 @@ public P jsonPatchPrimary(P actualResource, UnaryOperator
unaryOperator) {
* @return the patched resource as returned by the API server
*/
public P jsonPatchPrimary(P actualResource, UnaryOperator
unaryOperator, Options options) {
- P desired = desiredForJsonPatch(actualResource, unaryOperator, options);
+ P desired = desiredForJsonPatch(actualResource, unaryOperator);
return resourcePatch(
desired,
actualResource,
@@ -717,7 +717,7 @@ public P jsonPatchPrimaryStatus(P actualResource, UnaryOperator
unaryOperator
*/
public P jsonPatchPrimaryStatus(
P actualResource, UnaryOperator
unaryOperator, Options options) {
- P desired = desiredForJsonPatch(actualResource, unaryOperator, options);
+ P desired = desiredForJsonPatch(actualResource, unaryOperator);
return resourcePatch(
desired,
actualResource,
@@ -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);
}
@@ -1435,7 +1441,7 @@ public enum Mode {
}
private T desiredForJsonPatch(
- T actualResource, UnaryOperator unaryOperator, Options options) {
+ T actualResource, UnaryOperator unaryOperator) {
var cloned =
context
.getControllerConfiguration()
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/GenericKubernetesResourceMatcher.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/kubernetes/GenericKubernetesResourceMatcher.java
index b5a0728e16..23fb29151f 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/kubernetes/GenericKubernetesResourceMatcher.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/kubernetes/GenericKubernetesResourceMatcher.java
@@ -38,6 +38,12 @@ public class GenericKubernetesResourceMatcher SPEC_PREFIX = List.of(SPEC);
+ private static final List STATUS_PREFIX = List.of(STATUS);
+ private static final List METADATA_PREFIX = List.of(METADATA);
+ private static final List LABELS_AND_ANNOTATIONS_PREFIX =
+ List.of(METADATA_LABELS, METADATA_ANNOTATIONS);
+
private static final String PATH = "path";
private static final String[] EMPTY_ARRAY = {};
@@ -182,11 +188,11 @@ public static Matcher.Result m
boolean matched = true;
for (int i = 0; i < wholeDiffJsonPatch.size() && matched; i++) {
var node = wholeDiffJsonPatch.get(i);
- if (nodeIsChildOf(node, List.of(SPEC))) {
+ if (nodeIsChildOf(node, SPEC_PREFIX)) {
matched = match(valuesEquality, node, ignoreList);
- } else if (nodeIsChildOf(node, List.of(METADATA))) {
+ } else if (nodeIsChildOf(node, METADATA_PREFIX)) {
// conditionally consider labels and annotations
- if (nodeIsChildOf(node, List.of(METADATA_LABELS, METADATA_ANNOTATIONS))) {
+ if (nodeIsChildOf(node, LABELS_AND_ANNOTATIONS_PREFIX)) {
matched = match(labelsAndAnnotationsEquality, node, Collections.emptyList());
}
} else if (!nodeIsChildOf(node, IGNORED_FIELDS)) {
@@ -241,7 +247,7 @@ public static Matcher.Result m
boolean matched = true;
for (int i = 0; i < wholeDiffJsonPatch.size() && matched; i++) {
var node = wholeDiffJsonPatch.get(i);
- if (nodeIsChildOf(node, List.of(STATUS))) {
+ if (nodeIsChildOf(node, STATUS_PREFIX)) {
matched = match(valuesEquality, node, Collections.emptyList());
}
}
@@ -261,7 +267,12 @@ private static boolean match(boolean equality, JsonNode diff, final List
static boolean nodeIsChildOf(JsonNode n, List prefixes) {
var path = getPath(n);
- return prefixes.stream().anyMatch(path::startsWith);
+ for (int i = 0; i < prefixes.size(); i++) {
+ if (path.startsWith(prefixes.get(i))) {
+ return true;
+ }
+ }
+ return false;
}
static String getPath(JsonNode n) {
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/dependent/kubernetes/KubernetesDependent.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/kubernetes/KubernetesDependent.java
index 35bcde9052..a23d2b3aa8 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/kubernetes/KubernetesDependent.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/kubernetes/KubernetesDependent.java
@@ -21,6 +21,9 @@
import java.lang.annotation.Target;
import io.javaoperatorsdk.operator.api.config.informer.Informer;
+import io.javaoperatorsdk.operator.api.reconciler.Experimental;
+
+import static io.javaoperatorsdk.operator.api.reconciler.Experimental.API_MIGHT_CHANGE;
@Retention(RetentionPolicy.RUNTIME)
@Target({ElementType.TYPE})
@@ -62,4 +65,32 @@ boolean createResourceOnlyIfNotExistingWithSSA() default
*/
Class extends SSABasedGenericKubernetesResourceMatcher> matcher() default
SSABasedGenericKubernetesResourceMatcher.class;
+
+ /**
+ * Whether JOSDK should detect that the API version of this dependent resource's desired state has
+ * changed since it was last applied by the operator (for example after the operator was upgraded
+ * to target a new CRD version) and, in that case, request a one-time update of the actual
+ * resource.
+ *
+ *
When enabled, JOSDK records the API version it applies in the {@value
+ * KubernetesDependentResource#LAST_APPLIED_API_VERSION_ANNOTATION_KEY} annotation. On subsequent
+ * reconciliations, the resource is considered mismatched (and thus updated) if that recorded
+ * marker differs from the API version the operator currently uses, including when the marker is
+ * missing entirely (for example on resources created before this feature was enabled). Once the
+ * resource has been updated, the marker matches the current API version again, so no further
+ * update is requested until the API version changes again.
+ *
+ *
This is opt-in and disabled by default: when disabled, no marker annotation is ever added or
+ * read, and matching behavior is unchanged. It does not read or infer the actual storage version
+ * of the resource in Kubernetes, since that information is not reliably exposed by the API
+ * server; it only tracks what the operator itself last applied. It is not a replacement for
+ * Kubernetes' StorageVersionMigration.
+ *
+ * @return {@code true} if API version change detection is enabled, {@code false} otherwise
+ * @since 5.6
+ */
+ @Experimental(API_MIGHT_CHANGE)
+ boolean detectApiVersionChange() default
+ KubernetesDependentResourceConfig.DEFAULT_DETECT_API_VERSION_CHANGE;
}
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/kubernetes/KubernetesDependentConverter.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/kubernetes/KubernetesDependentConverter.java
index d39066e5d9..00c802867c 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/kubernetes/KubernetesDependentConverter.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/kubernetes/KubernetesDependentConverter.java
@@ -35,6 +35,8 @@ public KubernetesDependentResourceConfig configFrom(
ControllerConfiguration> controllerConfig) {
var createResourceOnlyIfNotExistingWithSSA =
DEFAULT_CREATE_RESOURCE_ONLY_IF_NOT_EXISTING_WITH_SSA;
+ var detectApiVersionChange =
+ KubernetesDependentResourceConfig.DEFAULT_DETECT_API_VERSION_CHANGE;
Boolean useSSA = null;
SSABasedGenericKubernetesResourceMatcher matcher =
@@ -43,6 +45,7 @@ public KubernetesDependentResourceConfig configFrom(
createResourceOnlyIfNotExistingWithSSA =
configAnnotation.createResourceOnlyIfNotExistingWithSSA();
useSSA = configAnnotation.useSSA().asBoolean();
+ detectApiVersionChange = configAnnotation.detectApiVersionChange();
// check if we have a specific matcher
Class extends KubernetesDependentResource, ?>> dependentResourceClass =
@@ -62,7 +65,11 @@ public KubernetesDependentResourceConfig configFrom(
controllerConfig);
return new KubernetesDependentResourceConfig<>(
- useSSA, createResourceOnlyIfNotExistingWithSSA, informerConfiguration, matcher);
+ useSSA,
+ createResourceOnlyIfNotExistingWithSSA,
+ informerConfiguration,
+ matcher,
+ detectApiVersionChange);
}
@SuppressWarnings({"unchecked"})
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/kubernetes/KubernetesDependentResource.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/kubernetes/KubernetesDependentResource.java
index bb59d6eed6..654b71fc5c 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/kubernetes/KubernetesDependentResource.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/kubernetes/KubernetesDependentResource.java
@@ -15,6 +15,7 @@
*/
package io.javaoperatorsdk.operator.processing.dependent.kubernetes;
+import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
@@ -52,6 +53,15 @@ public abstract class KubernetesDependentResource kubernetesDependentResourceConfig;
private volatile Boolean useSSA;
@@ -160,6 +170,12 @@ public Result match(R actualResource, R desired, P primary, Context
contex
protected void addMetadata(
boolean forMatch, R actualResource, final R target, P primary, Context
context) {
+ if (kubernetesDependentResourceConfig != null
+ && kubernetesDependentResourceConfig.detectApiVersionChange()) {
+ // desired resources might expose a null or immutable annotations map (e.g. Map.of(...));
+ // make sure it's a mutable one before this method or its callees write to it
+ ensureMutableAnnotations(target);
+ }
if (forMatch) { // keep the current previous annotation
String actual =
actualResource
@@ -173,9 +189,36 @@ protected void addMetadata(
annotations.remove(InformerEventSource.PREVIOUS_ANNOTATION_KEY);
}
}
+ addLastAppliedApiVersion(target);
addReferenceHandlingMetadata(target, primary);
}
+ private static void ensureMutableAnnotations(HasMetadata target) {
+ var metadata = target.getMetadata();
+ metadata.setAnnotations(
+ new LinkedHashMap<>(Optional.ofNullable(metadata.getAnnotations()).orElseGet(Map::of)));
+ }
+
+ /**
+ * When {@link KubernetesDependentResourceConfig#detectApiVersionChange()} is enabled, marks the
+ * target resource with the API version the operator is currently applying. Comparing this marker
+ * with the one recorded on the actual resource lets the regular matching logic detect a mismatch,
+ * without ever inspecting the actual, potentially unreliable, stored API version.
+ */
+ private void addLastAppliedApiVersion(R target) {
+ if (kubernetesDependentResourceConfig == null
+ || !kubernetesDependentResourceConfig.detectApiVersionChange()) {
+ return;
+ }
+ var apiVersion = target.getApiVersion();
+ if (apiVersion != null) {
+ target
+ .getMetadata()
+ .getAnnotations()
+ .put(LAST_APPLIED_API_VERSION_ANNOTATION_KEY, apiVersion);
+ }
+ }
+
protected boolean useSSA(Context
context) {
if (useSSA == null) {
useSSA =
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/kubernetes/KubernetesDependentResourceConfig.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/kubernetes/KubernetesDependentResourceConfig.java
index 05ff71335c..b7f8db5439 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/kubernetes/KubernetesDependentResourceConfig.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/kubernetes/KubernetesDependentResourceConfig.java
@@ -21,11 +21,13 @@
public class KubernetesDependentResourceConfig {
public static final boolean DEFAULT_CREATE_RESOURCE_ONLY_IF_NOT_EXISTING_WITH_SSA = true;
+ public static final boolean DEFAULT_DETECT_API_VERSION_CHANGE = false;
private final Boolean useSSA;
private final boolean createResourceOnlyIfNotExistingWithSSA;
private final InformerConfiguration informerConfig;
private final SSABasedGenericKubernetesResourceMatcher matcher;
+ private final boolean detectApiVersionChange;
public KubernetesDependentResourceConfig(
Boolean useSSA,
@@ -39,11 +41,26 @@ public KubernetesDependentResourceConfig(
boolean createResourceOnlyIfNotExistingWithSSA,
InformerConfiguration informerConfig,
SSABasedGenericKubernetesResourceMatcher matcher) {
+ this(
+ useSSA,
+ createResourceOnlyIfNotExistingWithSSA,
+ informerConfig,
+ matcher,
+ DEFAULT_DETECT_API_VERSION_CHANGE);
+ }
+
+ public KubernetesDependentResourceConfig(
+ Boolean useSSA,
+ boolean createResourceOnlyIfNotExistingWithSSA,
+ InformerConfiguration informerConfig,
+ SSABasedGenericKubernetesResourceMatcher matcher,
+ boolean detectApiVersionChange) {
this.useSSA = useSSA;
this.createResourceOnlyIfNotExistingWithSSA = createResourceOnlyIfNotExistingWithSSA;
this.informerConfig = informerConfig;
this.matcher =
matcher != null ? matcher : SSABasedGenericKubernetesResourceMatcher.getInstance();
+ this.detectApiVersionChange = detectApiVersionChange;
}
public boolean createResourceOnlyIfNotExistingWithSSA() {
@@ -61,4 +78,16 @@ public InformerConfiguration informerConfig() {
public SSABasedGenericKubernetesResourceMatcher matcher() {
return matcher;
}
+
+ /**
+ * Whether JOSDK should detect when the API version of this dependent resource's desired state has
+ * changed since it was last applied by the operator and, in that case, request a one-time update
+ * of the actual resource.
+ *
+ * @return {@code true} if API version change detection is enabled, {@code false} otherwise
+ * @since 5.6
+ */
+ public boolean detectApiVersionChange() {
+ return detectApiVersionChange;
+ }
}
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/kubernetes/KubernetesDependentResourceConfigBuilder.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/kubernetes/KubernetesDependentResourceConfigBuilder.java
index bdd6b068b3..3463eea7f1 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/kubernetes/KubernetesDependentResourceConfigBuilder.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/kubernetes/KubernetesDependentResourceConfigBuilder.java
@@ -24,6 +24,8 @@ public final class KubernetesDependentResourceConfigBuilder informerConfiguration;
private SSABasedGenericKubernetesResourceMatcher matcher;
+ private boolean detectApiVersionChange =
+ KubernetesDependentResourceConfig.DEFAULT_DETECT_API_VERSION_CHANGE;
public KubernetesDependentResourceConfigBuilder() {}
@@ -51,8 +53,18 @@ public KubernetesDependentResourceConfigBuilder withSSAMatcher(
return this;
}
+ public KubernetesDependentResourceConfigBuilder withDetectApiVersionChange(
+ boolean detectApiVersionChange) {
+ this.detectApiVersionChange = detectApiVersionChange;
+ return this;
+ }
+
public KubernetesDependentResourceConfig build() {
return new KubernetesDependentResourceConfig<>(
- useSSA, createResourceOnlyIfNotExistingWithSSA, informerConfiguration, matcher);
+ useSSA,
+ createResourceOnlyIfNotExistingWithSSA,
+ informerConfiguration,
+ matcher,
+ detectApiVersionChange);
}
}
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/kubernetes/SSABasedGenericKubernetesResourceMatcher.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/kubernetes/SSABasedGenericKubernetesResourceMatcher.java
index d3e5b6dbc5..abec13290d 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/kubernetes/SSABasedGenericKubernetesResourceMatcher.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/kubernetes/SSABasedGenericKubernetesResourceMatcher.java
@@ -38,6 +38,7 @@
import io.fabric8.kubernetes.api.model.apps.Deployment;
import io.fabric8.kubernetes.api.model.apps.ReplicaSet;
import io.fabric8.kubernetes.api.model.apps.StatefulSet;
+import io.fabric8.kubernetes.api.model.apps.StatefulSetSpec;
import io.fabric8.kubernetes.client.utils.KubernetesSerialization;
import io.javaoperatorsdk.operator.OperatorException;
import io.javaoperatorsdk.operator.api.reconciler.Context;
@@ -199,25 +200,7 @@ protected void sanitizeState(R actual, R desired, Map actualMap)
&& desired instanceof StatefulSet desiredStatefulSet) {
var actualSpec = actualStatefulSet.getSpec();
var desiredSpec = desiredStatefulSet.getSpec();
- int claims = desiredSpec.getVolumeClaimTemplates().size();
- if (claims == actualSpec.getVolumeClaimTemplates().size()) {
- for (int i = 0; i < claims; i++) {
- var claim = desiredSpec.getVolumeClaimTemplates().get(i);
- if (claim.getSpec().getVolumeMode() == null) {
- Optional.ofNullable(
- GenericKubernetesResource.get(
- actualMap, "spec", "volumeClaimTemplates", i, "spec"))
- .map(Map.class::cast)
- .ifPresent(m -> m.remove("volumeMode"));
- }
- if (claim.getStatus() == null) {
- Optional.ofNullable(
- GenericKubernetesResource.get(actualMap, "spec", "volumeClaimTemplates", i))
- .map(Map.class::cast)
- .ifPresent(m -> m.remove("status"));
- }
- }
- }
+ sanitizeVolumeClaimTemplates(actualMap, actualSpec, desiredSpec);
sanitizePodTemplateSpec(actualMap, actualSpec.getTemplate(), desiredSpec.getTemplate());
} else if (actual instanceof Deployment actualDeployment
&& desired instanceof Deployment desiredDeployment) {
@@ -240,6 +223,29 @@ protected void sanitizeState(R actual, R desired, Map actualMap)
}
}
+ private static void sanitizeVolumeClaimTemplates(
+ Map actualMap, StatefulSetSpec actualSpec, StatefulSetSpec desiredSpec) {
+ int claims = desiredSpec.getVolumeClaimTemplates().size();
+ if (claims != actualSpec.getVolumeClaimTemplates().size()) {
+ return;
+ }
+ for (int i = 0; i < claims; i++) {
+ var claim = desiredSpec.getVolumeClaimTemplates().get(i);
+ if (claim.getSpec().getVolumeMode() == null) {
+ Optional.ofNullable(
+ GenericKubernetesResource.get(actualMap, "spec", "volumeClaimTemplates", i, "spec"))
+ .map(Map.class::cast)
+ .ifPresent(m -> m.remove("volumeMode"));
+ }
+ if (claim.getStatus() == null) {
+ Optional.ofNullable(
+ GenericKubernetesResource.get(actualMap, "spec", "volumeClaimTemplates", i))
+ .map(Map.class::cast)
+ .ifPresent(m -> m.remove("status"));
+ }
+ }
+ }
+
@SuppressWarnings("unchecked")
static void keepOnlyManagedFields(
Map result,
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/AbstractWorkflowExecutor.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/AbstractWorkflowExecutor.java
index d3907b657a..665d80063b 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/AbstractWorkflowExecutor.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/AbstractWorkflowExecutor.java
@@ -52,7 +52,7 @@ protected AbstractWorkflowExecutor(DefaultWorkflow
workflow, P primary, Conte
this.context = context;
this.primaryID = ResourceID.fromResource(primary);
executorService = context.getWorkflowExecutorService();
- results = new HashMap<>(workflow.getDependentResourcesByName().size());
+ results = new HashMap<>(workflow.size());
}
protected abstract Logger logger();
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/EventFilterWindow.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/EventFilterWindow.java
index 826551656e..c63261c0b1 100644
--- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/EventFilterWindow.java
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/EventFilterWindow.java
@@ -239,8 +239,7 @@ public synchronized void addRelatedEvent(ExtendedResourceEvent event) {
event.setPartOfReList(true);
}
- relatedEvents.put(
- Long.valueOf(event.getResource().orElseThrow().getMetadata().getResourceVersion()), event);
+ relatedEvents.put(event.getResourceVersion(), event);
}
public synchronized void setReListStarted() {
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..6caf39ccd9 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,10 +26,7 @@
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;
@@ -37,39 +34,42 @@
import io.javaoperatorsdk.operator.api.config.Informable;
import io.javaoperatorsdk.operator.api.config.informer.InformerConfiguration;
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 +78,8 @@ public void start() throws OperatorException {
.getExecutorServiceManager()
.boundedExecuteAndWaitForAllToComplete(
sources.values().stream(),
- iw -> {
- iw.start();
+ wrapper -> {
+ start(wrapper);
return null;
},
iw ->
@@ -96,25 +96,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 +126,100 @@ 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 =
+ configuration
+ .getKubernetesClient()
+ .orElseGet(
+ () -> controllerConfiguration.getConfigurationService().getKubernetesClient());
}
-
- 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..2c960a7b6f
--- /dev/null
+++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/pool/AbstractInformerPool.java
@@ -0,0 +1,197 @@
+/*
+ * 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 {
+ throw new IllegalStateException(
+ "Cannot retrieve 'stopped' callback to listen to informer stopping for"
+ + " informer for "
+ + ReconcilerUtilsInternal.getResourceTypeNameWithVersion(
+ informer.getApiTypeClass()));
+ }
+ });
+ 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