feat: schedule polling and retries on managed executors instead of Timers

bootstrapper-maven-plugin/pom.xml

@@ -22,7 +22,7 @@
   <parent>
     <groupId>io.javaoperatorsdk</groupId>
     <artifactId>java-operator-sdk</artifactId>
-    <version>5.6.1-SNAPSHOT</version>
+    <version>999-SNAPSHOT</version>
   </parent>
 
   <artifactId>bootstrapper</artifactId>

caffeine-bounded-cache-support/pom.xml

@@ -21,7 +21,7 @@
   <parent>
     <groupId>io.javaoperatorsdk</groupId>
     <artifactId>java-operator-sdk</artifactId>
-    <version>5.6.1-SNAPSHOT</version>
+    <version>999-SNAPSHOT</version>
   </parent>
 
   <artifactId>caffeine-bounded-cache-support</artifactId>

docs/content/en/docs/documentation/dependent-resource-and-workflows/dependent-resources.md

@@ -531,6 +531,34 @@ samples [here](https://github.com/java-operator-sdk/java-operator-sdk/tree/main/
    in [related integration test](https://github.com/operator-framework/java-operator-sdk/blob/main/operator-framework/src/test/java/io/javaoperatorsdk/operator/workflow/orderedmanageddependent/ConfigMapDependentResource2.java)
    .
 
+### Adding Common Metadata to All Managed Resources (Desired State Aspects)
+
+Operators often need to mark every resource they manage in a uniform way, for example with a
+`app.kubernetes.io/managed-by` label, so that these resources can easily be identified, selected or
+garbage-collected later on. Instead of repeating that logic in every `desired()` implementation, a
+`DesiredStateAspect` can be registered once, at the operator level, and is then applied to the
+desired state of every Kubernetes dependent resource managed by the operator:
+
+```java
+Operator operator = new Operator(overrider -> overrider
+    .withDesiredStateAspects(List.of(
+        (desired, dependentResource, context) -> desired.getMetadata().getLabels()
+            .put("app.kubernetes.io/managed-by", "my-operator"))));
+```
+
+Aspects are applied, in registration order, right after the desired state has been computed and
+before the desired state is matched against the actual resource, created or updated. As a
+consequence, the metadata added by an aspect is part of the desired state proper: if it is removed
+from the actual resource, or if the aspect itself changes, the associated secondary resources are
+updated accordingly on the next reconciliation.
+
+Since the desired state is computed at most once per reconciliation and cached in the `Context`,
+aspects are called at most once per dependent resource and reconciliation. They are only called for
+dependent resources whose desired state is a `HasMetadata`, meaning that external (non-Kubernetes)
+dependent resources are left untouched. Implementations are expected to modify the provided desired
+state in place and need to be thread-safe as they can be called concurrently for different primary
+resources.
+
 ## "Read-only" Dependent Resources vs. Event Source
 
 See Integration test for a read-only

docs/content/en/docs/documentation/eventing.md

@@ -222,6 +222,31 @@ is similar to `PerResourcePollingEventSource` except that, contrary to that even
 doesn't poll a specific API separately per resource, but periodically and independently of
 actually observed primary resources.
 
+#### Threading of the polling event sources
+
+Both polling event sources schedule their polls on an executor the operator shares between all of
+its polling event sources. A poll therefore only starts once one of that executor's threads is
+free, so if your operator registers many polling event sources, or if fetching your external
+resources is slow, size the pool accordingly with
+`ConfigurationServiceOverrider.withConcurrentScheduledTaskThreads` (4 threads by default):
+
+```java
+Operator operator = new Operator(overrider -> overrider.withConcurrentScheduledTaskThreads(20));
+```
+
+Retried and rescheduled reconciliations are triggered on a separate executor, so a slow poll can
+never delay them. It is sized with `withConcurrentRetryAndRescheduleThreads` (2 threads by
+default); few threads are needed there since triggering a reconciliation only enqueues an event
+for one of the reconciliation threads to pick up.
+
+Use `ConfigurationServiceOverrider.withScheduledExecutorService` to replace the polling executor
+altogether. The operator still manages the lifecycle of that executor: it shuts it down when the
+operator is stopped, so don't shut it down yourself.
+
+Use the `withExecutorService` method of the event source's own configuration builder to poll a
+single event source on an executor of its own. An executor provided that way is not managed by the
+operator: it is your responsibility to shut it down.
+
 #### Inbound event sources
 
 [SimpleInboundEventSource](https://github.com/operator-framework/java-operator-sdk/blob/main/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/inbound/SimpleInboundEventSource.java)
@@ -348,6 +373,30 @@ 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.
 
+### Removing the Namespace Index
+
+Informers keep an index from namespace to the resources cached for it. JOSDK never reads that index,
+only the ones registered explicitly through `IndexerResourceCache.addIndexers(..)`, so it can be
+removed to save an entry per cached resource:
+
+```java
+@ControllerConfiguration(informer = @Informer(withoutNamespaceIndex = true))
+public class MyReconciler implements Reconciler<MyCustomResource> { }
+```
+
+The same option is available on `InformerEventSourceConfiguration.Builder` for event sources, and on
+`InformerConfiguration.Builder`.
+
+This matters most for informers that cache a large number of resources, and in particular together
+with a custom [item store](#bounded-caches-for-informers) that keeps only a reduced form of each
+resource: the index is keyed independently of what the store does with the resource itself, so
+shrinking what is cached does not shrink the index.
+
+Note that this takes part in the informer pool identity described below: an event source that
+removes the index does not share an informer with one that keeps it. If two event sources watch the
+same resource type and only one of them sets the option, they end up with two informers, and two
+caches of that resource type, which can cost far more memory than the index ever did.
+
 ### Sharing Informers Between Controllers (Informer Pool)
 
 {{% alert title="Experimental" color="warning" %}}
@@ -374,7 +423,10 @@ Two event sources share an informer when their effective informer configuration
 - 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 configured [item store](#bounded-caches-for-informers),
+- whether the [namespace index is removed](#removing-the-namespace-index): it cannot be present for
+  one event source and absent for another on one shared informer, so event sources that disagree on
+  it are backed by separate 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

docs/content/en/docs/documentation/operations/configuration.md

@@ -281,6 +281,13 @@ All operator-level keys are prefixed with `josdk.`.
 |---|---|---|
 | `josdk.workflow.executor-threads` | `Integer` | Thread pool size for workflow execution |
 
+#### Scheduled Tasks
+
+| Key | Type | Description |
+|---|---|---|
+| `josdk.scheduled-tasks.concurrent-threads` | `Integer` | Thread pool size for the operator's scheduled tasks, i.e. the polling event sources |
+| `josdk.retry-and-reschedule.concurrent-threads` | `Integer` | Thread pool size for triggering retried and rescheduled reconciliations |
+
 #### Informer
 
 | Key | Type | Description |
@@ -356,6 +363,7 @@ All controller-level keys are prefixed with `josdk.controller.<controller-name>.
 | `josdk.controller.<name>.informer.label-selector` | `String` | Label selector for the primary resource informer (alias for `label-selector`) |
 | `josdk.controller.<name>.informer.shard-selector` | `String` | Shard selector for the primary resource informer (alias for `shard-selector`) |
 | `josdk.controller.<name>.informer.list-limit` | `Long` | Page size for paginated informer list requests; omit for no pagination |
+| `josdk.controller.<name>.informer.without-namespace-index` | `Boolean` | Removes the namespace index the informer maintains; defaults to `false`. See [Removing the Namespace Index](../eventing#removing-the-namespace-index) |
 
 #### Retry
 

micrometer-support/pom.xml

@@ -21,7 +21,7 @@
   <parent>
     <groupId>io.javaoperatorsdk</groupId>
     <artifactId>java-operator-sdk</artifactId>
-    <version>5.6.1-SNAPSHOT</version>
+    <version>999-SNAPSHOT</version>
   </parent>
 
   <artifactId>micrometer-support</artifactId>

migration/pom.xml

@@ -21,7 +21,7 @@
   <parent>
     <groupId>io.javaoperatorsdk</groupId>
     <artifactId>java-operator-sdk</artifactId>
-    <version>5.6.1-SNAPSHOT</version>
+    <version>999-SNAPSHOT</version>
   </parent>
 
   <artifactId>migration</artifactId>

operator-framework-bom/pom.xml

@@ -21,7 +21,7 @@
 
   <groupId>io.javaoperatorsdk</groupId>
   <artifactId>operator-framework-bom</artifactId>
-  <version>5.6.1-SNAPSHOT</version>
+  <version>999-SNAPSHOT</version>
   <packaging>pom</packaging>
   <name>Operator SDK - Bill of Materials</name>
   <description>Java SDK for implementing Kubernetes operators</description>

operator-framework-core/pom.xml

@@ -21,7 +21,7 @@
   <parent>
     <groupId>io.javaoperatorsdk</groupId>
     <artifactId>java-operator-sdk</artifactId>
-    <version>5.6.1-SNAPSHOT</version>
+    <version>999-SNAPSHOT</version>
     <relativePath>../pom.xml</relativePath>
   </parent>
 
@@ -128,7 +128,7 @@
         <version>${git-commit-id-maven-plugin.version}</version>
         <configuration>
           <generateGitPropertiesFile>true</generateGitPropertiesFile>
-          <generateGitPropertiesFilename>${project.build.outputDirectory}/version.properties</generateGitPropertiesFilename>
+          <generateGitPropertiesFilename>${project.build.outputDirectory}/operator-sdk-version.properties</generateGitPropertiesFilename>
           <includeOnlyProperties>
             <includeOnlyProperty>^git.build.time$</includeOnlyProperty>
             <includeOnlyProperty>^git.commit.id.(abbrev|full)$</includeOnlyProperty>

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ConfigurationService.java

@@ -16,10 +16,12 @@
 package io.javaoperatorsdk.operator.api.config;
 
 import java.time.Duration;
+import java.util.List;
 import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
 import java.util.function.Consumer;
 
 import org.slf4j.Logger;
@@ -41,6 +43,7 @@
 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.api.reconciler.dependent.DesiredStateAspect;
 import io.javaoperatorsdk.operator.processing.dependent.kubernetes.KubernetesDependent;
 import io.javaoperatorsdk.operator.processing.dependent.kubernetes.KubernetesDependentResource;
 import io.javaoperatorsdk.operator.processing.dependent.kubernetes.KubernetesDependentResourceConfig;
@@ -62,6 +65,20 @@ public interface ConfigurationService {
   /** The default number of threads used to process dependent workflows */
   int DEFAULT_WORKFLOW_EXECUTOR_THREAD_NUMBER = DEFAULT_RECONCILIATION_THREADS_NUMBER;
 
+  /**
+   * The default number of threads used to run the operator's scheduled tasks, i.e. the periodic
+   * polls of {@link io.javaoperatorsdk.operator.processing.event.source.polling.PollingEventSource}
+   * and {@link
+   * io.javaoperatorsdk.operator.processing.event.source.polling.PerResourcePollingEventSource}
+   */
+  int DEFAULT_SCHEDULED_TASK_THREADS_NUMBER = 4;
+
+  /**
+   * The default number of threads used to trigger the operator's retried and rescheduled
+   * reconciliations
+   */
+  int DEFAULT_RETRY_AND_RESCHEDULE_THREADS_NUMBER = 2;
+
   /**
    * Creates a new {@link ConfigurationService} instance used to configure an {@link
    * io.javaoperatorsdk.operator.Operator} instance, starting from the specified base configuration
@@ -217,6 +234,33 @@ default int concurrentWorkflowExecutorThreads() {
     return DEFAULT_WORKFLOW_EXECUTOR_THREAD_NUMBER;
   }
 
+  /**
+   * Number of threads the operator can spin out to run its scheduled (i.e. periodic or delayed)
+   * tasks with the default executor. These threads are shared by all the polling event sources of
+   * the operator, so this number should be raised when many, or slow, polling event sources are
+   * registered: a poll only starts once a thread is available, and a slow poll therefore delays the
+   * polls of the other event sources.
+   *
+   * @return the maximum number of concurrent scheduled task threads
+   * @since 5.6.0
+   */
+  default int concurrentScheduledTaskThreads() {
+    return DEFAULT_SCHEDULED_TASK_THREADS_NUMBER;
+  }
+
+  /**
+   * Number of threads the operator can spin out to trigger its retried and rescheduled
+   * reconciliations with the default executor. These threads are shared by all the controllers of
+   * the operator, but the tasks they run only enqueue an event for the reconciliation to happen on
+   * a reconciliation thread, so few of them are needed.
+   *
+   * @return the maximum number of concurrent retry and reschedule threads
+   * @since 5.6.0
+   */
+  default int concurrentRetryAndRescheduleThreads() {
+    return DEFAULT_RETRY_AND_RESCHEDULE_THREADS_NUMBER;
+  }
+
   /**
    * Override to provide a custom {@link Metrics} implementation
    *
@@ -247,6 +291,44 @@ default ExecutorService getWorkflowExecutorService() {
     return Executors.newFixedThreadPool(concurrentWorkflowExecutorThreads());
   }
 
+  /**
+   * Override to provide a custom {@link ScheduledExecutorService} implementation to change how the
+   * operator's scheduled (i.e. periodic or delayed) tasks are run. This executor is shared by all
+   * the polling event sources of the operator. The retried and rescheduled reconciliations run on
+   * an executor of their own, see {@link #getRetryAndRescheduleExecutorService()}, so that a slow
+   * poll can't delay them.
+   *
+   * <p>Note that the default implementation lets the executor discard the tasks that were scheduled
+   * for later when it is shut down, so that they don't delay the termination of the operator, and
+   * that it creates daemon threads so that a never stopped operator doesn't keep the JVM alive.
+   * Custom implementations are advised to do the same.
+   *
+   * @return the {@link ScheduledExecutorService} implementation to use to run scheduled tasks
+   * @since 5.6.0
+   */
+  default ScheduledExecutorService getScheduledExecutorService() {
+    return Utils.daemonScheduledThreadPool(
+        concurrentScheduledTaskThreads(), "josdk-scheduled-task");
+  }
+
+  /**
+   * Override to provide a custom {@link ScheduledExecutorService} implementation to change how the
+   * operator's retried and rescheduled reconciliations are triggered. This executor is kept
+   * separate from the one the polling event sources use, see {@link
+   * #getScheduledExecutorService()}, so that a slow poll can't delay a retry.
+   *
+   * <p>The same notes as for {@link #getScheduledExecutorService()} apply to custom
+   * implementations.
+   *
+   * @return the {@link ScheduledExecutorService} implementation to use to trigger retried and
+   *     rescheduled reconciliations
+   * @since 5.6.0
+   */
+  default ScheduledExecutorService getRetryAndRescheduleExecutorService() {
+    return Utils.daemonScheduledThreadPool(
+        concurrentRetryAndRescheduleThreads(), "josdk-retry-reschedule");
+  }
+
   /**
    * Determines whether the associated Kubernetes client should be closed when the associated {@link
    * io.javaoperatorsdk.operator.Operator} is stopped.
@@ -539,4 +621,18 @@ default InformerPool informerPool() {
     pool.setConfigurationService(this);
     return pool;
   }
+
+  /**
+   * Retrieves the {@link DesiredStateAspect}s applied to the desired state of all the Kubernetes
+   * dependent resources managed by the operator. Aspects are applied in the order in which they are
+   * returned, right after the desired state has been computed, and are typically used to add common
+   * metadata (such as a label identifying the operator managing the resource) to all the resources
+   * the operator creates or updates.
+   *
+   * @return the list of aspects to apply to computed desired states, empty by default
+   * @since 5.6.0
+   */
+  default List<DesiredStateAspect> desiredStateAspects() {
+    return List.of();
+  }
 }